1package imapserver
2
3import (
4 "errors"
5 "io"
6 "os"
7 "time"
8
9 "github.com/mjl-/bstore"
10
11 "github.com/mjl-/mox/config"
12 "github.com/mjl-/mox/message"
13 "github.com/mjl-/mox/mlog"
14 "github.com/mjl-/mox/store"
15)
16
17// Replace relaces a message for another, atomically, possibly in another mailbox,
18// without needing a sequence of: append message, store \deleted flag, expunge.
19//
20// State: Selected
21func (c *conn) cmdxReplace(isUID bool, tag, cmd string, p *parser) {
22 // Command: ../rfc/8508:158 ../rfc/8508:198
23
24 // Request syntax: ../rfc/8508:471
25 p.xspace()
26 star := p.take("*")
27 var num uint32
28 if !star {
29 num = p.xnznumber()
30 }
31 p.xspace()
32 name := p.xmailbox()
33
34 // ../rfc/4466:473
35 p.xspace()
36 var storeFlags store.Flags
37 var keywords []string
38 if p.hasPrefix("(") {
39 // Error must be a syntax error, to properly abort the connection due to literal.
40 var err error
41 storeFlags, keywords, err = store.ParseFlagsKeywords(p.xflagList())
42 if err != nil {
43 xsyntaxErrorf("parsing flags: %v", err)
44 }
45 p.xspace()
46 }
47
48 var tm time.Time
49 if p.hasPrefix(`"`) {
50 tm = p.xdateTime()
51 p.xspace()
52 } else {
53 tm = time.Now()
54 }
55
56 // todo: only with utf8 should we we accept message headers with utf-8. we currently always accept them.
57 // todo: this is only relevant if we also support the CATENATE extension?
58 // ../rfc/6855:204
59 utf8 := p.take("UTF8 (")
60 if utf8 {
61 p.xtake("~")
62 }
63 // Always allow literal8, for binary extension. ../rfc/4466:486
64 // For utf8, we already consumed the required ~ above.
65 size, synclit := p.xliteralSize(!utf8, false)
66
67 // Check the request, including old message in database, whether the message fits
68 // in quota. If a non-nil func is returned, an error was found. Calling the
69 // function aborts handling this command.
70 var uidOld store.UID
71 checkMessage := func(tx *bstore.Tx) func() {
72 if c.readonly {
73 return func() { xuserErrorf("mailbox open in read-only mode") }
74 }
75
76 mb, err := c.account.MailboxFind(tx, name)
77 if err != nil {
78 return func() { xserverErrorf("finding mailbox: %v", err) }
79 }
80 if mb == nil {
81 return func() { xusercodeErrorf("TRYCREATE", "%w", store.ErrUnknownMailbox) }
82 }
83
84 // Resolve "*" for UID or message sequence.
85 if star {
86 if c.uidonly {
87 q := bstore.QueryTx[store.Message](tx)
88 q.FilterNonzero(store.Message{MailboxID: c.mailboxID})
89 q.FilterEqual("Expunged", false)
90 q.FilterLess("UID", c.uidnext)
91 q.SortDesc("UID")
92 q.Limit(1)
93 m, err := q.Get()
94 if err == bstore.ErrAbsent {
95 return func() { xsyntaxErrorf("cannot use * on empty mailbox") }
96 }
97 xcheckf(err, "get last message in mailbox")
98 num = uint32(m.UID)
99 } else if c.exists == 0 {
100 return func() { xsyntaxErrorf("cannot use * on empty mailbox") }
101 } else if isUID {
102 num = uint32(c.uids[c.exists-1])
103 } else {
104 num = uint32(c.exists)
105 }
106 star = false
107 }
108
109 // Find or verify UID of message to replace.
110 if isUID {
111 uidOld = store.UID(num)
112 } else if num > c.exists {
113 return func() { xuserErrorf("invalid msgseq") }
114 } else {
115 uidOld = c.uids[int(num)-1]
116 }
117
118 // Check the message still exists in the database. If it doesn't, it may have been
119 // deleted just now and we won't check the quota. We'll raise an error later on,
120 // when we are not possibly reading a sync literal and can respond with unsolicited
121 // expunges.
122 q := bstore.QueryTx[store.Message](tx)
123 q.FilterNonzero(store.Message{MailboxID: c.mailboxID, UID: uidOld})
124 q.FilterEqual("Expunged", false)
125 q.FilterLess("UID", c.uidnext)
126 _, err = q.Get()
127 if err == bstore.ErrAbsent {
128 return nil
129 }
130 if err != nil {
131 return func() { xserverErrorf("get message to replace: %v", err) }
132 }
133
134 // Check if we can add size bytes. We can't necessarily remove the current message yet.
135 ok, maxSize, err := c.account.CanAddMessageSize(tx, size)
136 if err != nil {
137 return func() { xserverErrorf("check quota: %v", err) }
138 }
139 if !ok {
140 // ../rfc/9208:472
141 return func() { xusercodeErrorf("OVERQUOTA", "account over maximum total message size %d", maxSize) }
142 }
143 return nil
144 }
145
146 var errfn func()
147 if synclit {
148 // Check request, if it cannot succeed, fail it now before client is sending the data.
149
150 name = xcheckmailboxname(name, true)
151
152 c.account.WithRLock(func() {
153 c.xdbread(func(tx *bstore.Tx) {
154 errfn = checkMessage(tx)
155 if errfn != nil {
156 errfn()
157 }
158 })
159 })
160
161 c.xwritelinef("+ ")
162 } else {
163 var err error
164 name, _, err = config.CheckMailboxName(name, true)
165 if err != nil {
166 errfn = func() { xusercodeErrorf("CANNOT", "%s", err) }
167 } else {
168 c.account.WithRLock(func() {
169 c.xdbread(func(tx *bstore.Tx) {
170 errfn = checkMessage(tx)
171 })
172 })
173 }
174 }
175
176 var file *os.File
177 var newID int64 // Delivered message ID, file removed on error.
178 var f io.Writer
179 var commit bool
180
181 if errfn != nil {
182 // We got a non-sync literal, we will consume some data, but abort if there's too
183 // much. We draw the line at 1mb. Client should have used synchronizing literal.
184 if size > 1000*1000 {
185 // ../rfc/9051:357 ../rfc/3501:347
186 err := errors.New("error condition and non-synchronizing literal too big")
187 bye := "* BYE [ALERT] " + err.Error()
188 panic(syntaxError{bye, "TOOBIG", err.Error(), err})
189 }
190 // Message will not be accepted.
191 f = io.Discard
192 } else {
193 // Read the message into a temporary file.
194 var err error
195 file, err = store.CreateMessageTemp(c.log, "imap-replace")
196 xcheckf(err, "creating temp file for message")
197 defer store.CloseRemoveTempFile(c.log, file, "temporary message file")
198 f = file
199
200 defer func() {
201 if !commit && newID != 0 {
202 p := c.account.MessagePath(newID)
203 err := os.Remove(p)
204 c.xsanity(err, "remove message file for replace after error")
205 }
206 }()
207 }
208
209 // Read the message data.
210 defer c.xtraceread(mlog.LevelTracedata)()
211 mw := message.NewWriter(f)
212 msize, err := io.Copy(mw, io.LimitReader(c.br, size))
213 c.xtraceread(mlog.LevelTrace) // Restore.
214 if err != nil {
215 // Cannot use xcheckf due to %w handling of errIO.
216 c.xbrokenf("reading literal message: %s (%w)", err, errIO)
217 }
218 if msize != size {
219 c.xbrokenf("read %d bytes for message, expected %d (%w)", msize, size, errIO)
220 }
221
222 // Finish reading the command.
223 line := c.xreadline(false)
224 p = newParser(line, c)
225 if utf8 {
226 p.xtake(")")
227 }
228 p.xempty()
229
230 // If an error was found earlier, abort the command now that we've read the message.
231 if errfn != nil {
232 errfn()
233 }
234
235 var oldMsgExpunged bool
236
237 var om, nm store.Message
238 var mbSrc, mbDst store.Mailbox // Src and dst mailboxes can be different. ../rfc/8508:263
239 var overflow bool
240 var pendingChanges []store.Change
241 defer func() {
242 // In case of panic.
243 c.flushChanges(pendingChanges)
244 }()
245
246 c.account.WithWLock(func() {
247 var changes []store.Change
248
249 c.xdbwrite(func(tx *bstore.Tx) {
250 mbSrc = c.xmailboxID(tx, c.mailboxID)
251
252 // Get old message. If it has been expunged, we should have a pending change for
253 // it. We'll send untagged responses and fail the command.
254 var err error
255 qom := bstore.QueryTx[store.Message](tx)
256 qom.FilterNonzero(store.Message{MailboxID: mbSrc.ID, UID: uidOld})
257 om, err = qom.Get()
258 xcheckf(err, "get old message to replace from database")
259 if om.Expunged {
260 oldMsgExpunged = true
261 return
262 }
263
264 // Check quota for addition of new message. We can't necessarily yet remove the old message.
265 ok, maxSize, err := c.account.CanAddMessageSize(tx, mw.Size)
266 xcheckf(err, "checking quota")
267 if !ok {
268 // ../rfc/9208:472
269 xusercodeErrorf("OVERQUOTA", "account over maximum total message size %d", maxSize)
270 }
271
272 modseq, err := c.account.NextModSeq(tx)
273 xcheckf(err, "get next mod seq")
274
275 chremuids, _, err := c.account.MessageRemove(c.log, tx, modseq, &mbSrc, store.RemoveOpts{}, om)
276 xcheckf(err, "expunge old message")
277 changes = append(changes, chremuids)
278 // Note: we only add a mbSrc counts change later on, if it is not equal to mbDst.
279
280 err = tx.Update(&mbSrc)
281 xcheckf(err, "updating source mailbox counts")
282
283 mbDst = c.xmailbox(tx, name, "TRYCREATE")
284 mbDst.ModSeq = modseq
285
286 nkeywords := len(mbDst.Keywords)
287
288 // Make new message to deliver.
289 nm = store.Message{
290 MailboxID: mbDst.ID,
291 MailboxOrigID: mbDst.ID,
292 Received: tm,
293 Flags: storeFlags,
294 Keywords: keywords,
295 Size: mw.Size,
296 ModSeq: modseq,
297 CreateSeq: modseq,
298 }
299
300 err = c.account.MessageAdd(c.log, tx, &mbDst, &nm, file, store.AddOpts{})
301 xcheckf(err, "delivering message")
302 newID = nm.ID
303
304 changes = append(changes, nm.ChangeAddUID(mbDst), mbDst.ChangeCounts())
305 if nkeywords != len(mbDst.Keywords) {
306 changes = append(changes, mbDst.ChangeKeywords())
307 }
308
309 err = tx.Update(&mbDst)
310 xcheckf(err, "updating destination mailbox")
311 })
312
313 // Fetch pending changes, possibly with new UIDs, so we can apply them before adding our own new UID.
314 overflow, pendingChanges = c.comm.Get()
315
316 if oldMsgExpunged {
317 return
318 }
319
320 // Success, make sure messages aren't cleaned up anymore.
321 commit = true
322
323 // Broadcast the change to other connections.
324 if mbSrc.ID != mbDst.ID {
325 changes = append(changes, mbSrc.ChangeCounts())
326 }
327 c.broadcast(changes)
328 })
329
330 // Must update our msgseq/uids tracking with latest pending changes.
331 l := pendingChanges
332 pendingChanges = nil
333 c.xapplyChanges(overflow, l, false)
334
335 // If we couldn't find the message, send a NO response. We've just applied pending
336 // changes, which should have expunged the absent message.
337 if oldMsgExpunged {
338 xuserErrorf("message to be replaced has been expunged")
339 }
340
341 // If the destination mailbox is our currently selected mailbox, we register and
342 // announce the new message.
343 if mbDst.ID == c.mailboxID {
344 c.uidAppend(nm.UID)
345 // We send an untagged OK with APPENDUID, for sane bookkeeping in clients. ../rfc/8508:401
346 c.xbwritelinef("* OK [APPENDUID %d %d] ", mbDst.UIDValidity, nm.UID)
347 c.xbwritelinef("* %d EXISTS", c.exists)
348 }
349
350 // We must return vanished instead of expunge, and also highestmodseq, when qresync
351 // was enabled. ../rfc/8508:422 ../rfc/7162:1883
352 qresync := c.enabled[capQresync]
353
354 // Now that we are in sync with msgseq, we can find our old msgseq and say it is
355 // expunged or vanished. ../rfc/7162:1900
356 var oseq msgseq
357 if c.uidonly {
358 c.exists--
359 } else {
360 oseq = c.xsequence(om.UID)
361 c.sequenceRemove(oseq, om.UID)
362 }
363 if qresync || c.uidonly {
364 c.xbwritelinef("* VANISHED %d", om.UID)
365 // ../rfc/7162:1916
366 } else {
367 c.xbwritelinef("* %d EXPUNGE", oseq)
368 }
369 c.xwriteresultf("%s OK [HIGHESTMODSEQ %d] replaced", tag, nm.ModSeq.Client())
370}
371