serv/mox
1
0
Fork 0
mirror of https://github.com/mjl-/mox.git synced 2024-12-26 08:23:48 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

add funtionality to import zip/tgz with maildirs/mboxes to account page

Browse source 
so users can easily take their email out of somewhere else, and import it into mox.

this goes a little way to give feedback as the import progresses: upload
progress is shown (surprisingly, browsers aren't doing this...), imported
mailboxes/messages are counted (batched) and import issues/warnings are
displayed, all sent over an SSE connection. an import token is stored in
sessionstorage. if you reload the page (e.g. after a connection error), the
browser will reconnect to the running import and show its progress again. and
you can just abort the import before it is finished and committed, and nothing
will have changed.

this also imports flags/keywords from mbox files.
This commit is contained in:
Mechiel Lukkien 2023-02-16 09:57:27 +01:00
parent 23b530ae36
commit 5336032088
No known key found for this signature in database
32 changed files with 1968 additions and 518 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
README.md
View file

@ -140,12 +140,13 @@ documentation.
## How do I import/export email?
Use the "mox import maildir" or "mox import mbox" subcommands. You could also
use your IMAP email client, add your mox account, and copy or move messages
from one account to the other.
Use the import functionality on the accounts web page to import a zip/tgz with
maildirs/mbox files, or use the "mox import maildir" or "mox import mbox"
subcommands. You could also use your IMAP email client, add your mox account,
and copy or move messages from one account to the other.
Similarly, see the "mox export maildir" and "mox export mbox" subcommands to
export email.
Similarly, see the export functionality on the accounts web page and the "mox
export maildir" and "mox export mbox" subcommands to export email.
## How can I help?
@ -168,8 +169,8 @@ work.
## How do I change my password?
Regular users (doing IMAP/SMTP with authentication) can change their password
at the account page, e.g. http://127.0.0.1/account/. Or you can set a password
with "mox setaccountpassword".
at the account page, e.g. http://127.0.0.1/. Or you can set a password with "mox
setaccountpassword".
The admin password can be changed with "mox setadminpassword".

34
doc.go
View file

@ -206,56 +206,44 @@ The message is printed to stdout and is in standard internet mail format.
Import a maildir into an account.
By default, messages will train the junk filter based on their flags and
mailbox naming. If the destination mailbox name starts with "junk" or "spam"
(case insensitive), messages are imported and trained as junk regardless of
pre-existing flags. Use the -train=false flag to prevent training the filter.
By default, messages will train the junk filter based on their flags and, if
"automatic junk flags" configuration is set, based on mailbox naming.
If the destination mailbox is "Sent", the recipients of the messages are added
to the message metadata, causing later incoming messages from these recipients
to be accepted, unless other reputation signals prevent that.
The message "read"/"seen" flag can be overridden during import with the
-markread flag.
Users can also import mailboxes/messages through the account web page by
uploading a zip or tgz file with mbox and/or maildirs.
Mailbox flags, like "seen", "answered", "forwarded", will be imported. An
attempt is made to parse dovecot keyword files.
Mailbox flags, like "seen", "answered", will be imported. An optional
dovecot-keywords file can specify additional flags, like Forwarded/Junk/NotJunk.
The maildir files/directories are read by the mox process, so make sure it has
access to the maildir directories/files.
usage: mox import maildir accountname mailboxname maildir
-markread
mark all imported messages as read
-train
train junkfilter with messages (default true)
# mox import mbox
Import an mbox into an account.
Using mbox is not recommended, maildir is a better format.
Using mbox is not recommended, maildir is a better defined format.
By default, messages will train the junk filter based on their flags and
mailbox naming. If the destination mailbox name starts with "junk" or "spam"
(case insensitive), messages are imported and trained as junk regardless of
pre-existing flags. Use the -train=false flag to prevent training the filter.
By default, messages will train the junk filter based on their flags and, if
"automatic junk flags" configuration is set, based on mailbox naming.
If the destination mailbox is "Sent", the recipients of the messages are added
to the message metadata, causing later incoming messages from these recipients
to be accepted, unless other reputation signals prevent that.
The message "read"/"seen" flag can be overridden during import with the
-markread flag.
Users can also import mailboxes/messages through the account web page by
uploading a zip or tgz file with mbox and/or maildirs.
The mailbox is read by the mox process, so make sure it has access to the
maildir directories/files.
usage: mox import mbox accountname mailboxname mbox
-markread
mark all imported messages as read
-train
train junkfilter with messages (default true)
# mox export maildir

114
http/account.go
View file

@ -6,6 +6,7 @@ import (
"compress/gzip"
"context"
"encoding/base64"
"encoding/json"
"errors"
"io"
"net"
@ -106,6 +107,63 @@ func accountHandle(w http.ResponseWriter, r *http.Request) {
ctx := context.WithValue(r.Context(), mlog.CidKey, mox.Cid())
log := xlog.WithContext(ctx).Fields(mlog.Field("userauth", ""))
// Without authentication. The token is unguessable.
if r.URL.Path == "/importprogress" {
if r.Method != "GET" {
http.Error(w, "405 - method not allowed - get required", http.StatusMethodNotAllowed)
return
}
q := r.URL.Query()
token := q.Get("token")
if token == "" {
http.Error(w, "400 - bad request - missing token", http.StatusBadRequest)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
log.Error("internal error: ResponseWriter not a http.Flusher")
http.Error(w, "500 - internal error - cannot sync to http connection", 500)
return
}
l := importListener{token, make(chan importEvent, 100), make(chan bool, 1)}
importers.Register <- &l
ok = <-l.Register
if !ok {
http.Error(w, "400 - bad request - unknown token, import may have finished more than a minute ago", http.StatusBadRequest)
return
}
defer func() {
importers.Unregister <- &l
}()
h := w.Header()
h.Set("Content-Type", "text/event-stream")
h.Set("Cache-Control", "no-cache")
_, err := w.Write([]byte(": keepalive\n\n"))
if err != nil {
return
}
flusher.Flush()
ctx := r.Context()
for {
select {
case e := <-l.Events:
_, err := w.Write(e.SSEMsg)
flusher.Flush()
if err != nil {
return
}
case <-ctx.Done():
return
}
}
}
accName := checkAccountAuth(ctx, log, w, r)
if accName == "" {
// Response already sent.
@ -165,6 +223,54 @@ func accountHandle(w http.ResponseWriter, r *http.Request) {
log.Errorx("exporting mail", err)
}
case "/import":
if r.Method != "POST" {
http.Error(w, "405 - method not allowed - post required", http.StatusMethodNotAllowed)
return
}
f, _, err := r.FormFile("file")
if err != nil {
if errors.Is(err, http.ErrMissingFile) {
http.Error(w, "400 - bad request - missing file", http.StatusBadRequest)
} else {
http.Error(w, "500 - internal server error - "+err.Error(), http.StatusInternalServerError)
}
return
}
defer f.Close()
skipMailboxPrefix := r.FormValue("skipMailboxPrefix")
tmpf, err := os.CreateTemp("", "mox-import")
if err != nil {
http.Error(w, "500 - internal server error - "+err.Error(), http.StatusInternalServerError)
return
}
defer func() {
if tmpf != nil {
tmpf.Close()
}
}()
if err := os.Remove(tmpf.Name()); err != nil {
log.Errorx("removing temporary file", err)
http.Error(w, "500 - internal server error - "+err.Error(), http.StatusInternalServerError)
return
}
if _, err := io.Copy(tmpf, f); err != nil {
log.Errorx("copying import to temporary file", err)
http.Error(w, "500 - internal server error - "+err.Error(), http.StatusInternalServerError)
return
}
token, err := importStart(log, accName, tmpf, skipMailboxPrefix)
if err != nil {
log.Errorx("starting import", err)
http.Error(w, "500 - internal server error - "+err.Error(), http.StatusInternalServerError)
return
}
tmpf = nil // importStart is now responsible for closing.
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{"ImportToken": token})
default:
if strings.HasPrefix(r.URL.Path, "/api/") {
accountSherpaHandler.ServeHTTP(w, r.WithContext(context.WithValue(ctx, authCtxKey, accName)))
@ -230,3 +336,11 @@ func (Account) DestinationSave(ctx context.Context, destName string, oldDest, ne
err := mox.DestinationSave(ctx, accountName, destName, newDest)
xcheckf(ctx, err, "saving destination")
}
// ImportAbort aborts an import that is in progress. If the import exists and isn't
// finished, no changes will have been made by the import.
func (Account) ImportAbort(ctx context.Context, importToken string) error {
req := importAbortRequest{importToken, make(chan error)}
importers.Abort <- req
return <-req.Response
}

264
http/account.html
View file

@ -130,10 +130,132 @@ const domainString = d => {
return d.ASCII
}
const box = (color, ...l) => [
dom.div(
style({
display: 'inline-block',
padding: '.25em .5em',
backgroundColor: color,
borderRadius: '3px',
margin: '.5ex 0',
}),
l,
),
dom.br(),
]
const green = '#1dea20'
const yellow = '#ffe400'
const red = '#ff7443'
const blue = '#8bc8ff'
const index = async () => {
const [domain, destinations] = await api.Destinations()
let form, fieldset, password1, password2, passwordHint
let passwordForm, passwordFieldset, password1, password2, passwordHint
let importForm, importFieldset, mailboxFile, mailboxFileHint, mailboxPrefix, mailboxPrefixHint, importProgress, importAbortBox, importAbort
const importTrack = async (token) => {
const importConnection = dom.div('Waiting for updates...')
importProgress.appendChild(importConnection)
let countsTbody
let counts = {} // mailbox -> elem
let problems // element
await new Promise((resolve, reject) => {
const eventSource = new window.EventSource('importprogress?token=' + encodeURIComponent(token))
eventSource.addEventListener('open', function(e) {
console.log('eventsource open', {e})
dom._kids(importConnection, dom.div('Waiting for updates, connected...'))
dom._kids(importAbortBox,
importAbort=dom.button('Abort import', attr({title: 'If the import is not yet finished, it can be aborted and no messages will have been imported.'}), async function click(e) {
try {
await api.ImportAbort(token)
} catch (err) {
console.log({err})
window.alert('Error: ' + err.message)
}
// On success, the event source will get an aborted notification and shutdown the connection.
})
)
})
eventSource.addEventListener('error', function(e) {
console.log('eventsource error', {e})
dom._kids(importConnection, box(red, 'Connection error'))
reject({message: 'Connection error'})
})
eventSource.addEventListener('count', (e) => {
const data = JSON.parse(e.data) // {Mailbox: ..., Count: ...}
console.log('import count event', {e, data})
if (!countsTbody) {
importProgress.appendChild(
dom.div(
dom.br(),
dom.h3('Importing mailboxes and messages...'),
dom.table(
dom.thead(
dom.tr(dom.th('Mailbox'), dom.th('Messages')),
),
countsTbody=dom.tbody(),
),
)
)
}
let elem = counts[data.Mailbox]
if (!elem) {
countsTbody.appendChild(
dom.tr(
dom.td(data.Mailbox),
elem=dom.td(style({textAlign: 'right'}), ''+data.Count),
),
)
counts[data.Mailbox] = elem
}
dom._kids(elem, ''+data.Count)
})
eventSource.addEventListener('problem', (e) => {
const data = JSON.parse(e.data) // {Message: ...}
console.log('import problem event', {e, data})
if (!problems) {
importProgress.appendChild(
dom.div(
dom.br(),
dom.h3('Problems during import'),
problems=dom.div(),
),
)
}
problems.appendChild(dom.div(box(yellow, data.Message)))
})
eventSource.addEventListener('done', (e) => {
console.log('import done event', {e})
importProgress.appendChild(dom.div(dom.br(), box(blue, 'Import finished')))
eventSource.close()
dom._kids(importConnection)
dom._kids(importAbortBox)
window.sessionStorage.removeItem('ImportToken')
resolve()
})
eventSource.addEventListener('aborted', function(e) {
console.log('import aborted event', {e})
importProgress.appendChild(dom.div(dom.br(), box(red, 'Import aborted, no message imported')))
eventSource.close()
dom._kids(importConnection)
dom._kids(importAbortBox)
window.sessionStorage.removeItem('ImportToken')
reject({message: 'Import aborted'})
})
})
}
const page = document.getElementById('page')
dom._kids(page,
@ -154,8 +276,8 @@ const index = async () => {
),
dom.br(),
dom.h2('Change password'),
form=dom.form(
fieldset=dom.fieldset(
passwordForm=dom.form(
passwordFieldset=dom.fieldset(
dom.label(
style({display: 'inline-block'}),
'New password',
@ -182,16 +304,16 @@ const index = async () => {
window.alert('Passwords do not match.')
return
}
fieldset.disabled = true
passwordFieldset.disabled = true
try {
await api.SetPassword(password1.value)
window.alert('Password has been changed.')
form.reset()
passwordForm.reset()
} catch (err) {
console.log({err})
window.alert('Error: ' + err.message)
} finally {
fieldset.disabled = false
passwordFieldset.disabled = false
}
},
),
@ -204,8 +326,138 @@ const index = async () => {
dom.li(dom.a('mail-export-mbox.tgz', attr({href: 'mail-export-mbox.tgz'}))),
dom.li(dom.a('mail-export-mbox.zip', attr({href: 'mail-export-mbox.zip'}))),
),
dom.br(),
dom.h2('Import'),
dom.p('Import messages from a .zip or .tgz file with maildirs and/or mbox files.'),
importForm=dom.form(
async function submit(e) {
e.preventDefault()
e.stopPropagation()
const request = () => {
return new Promise((resolve, reject) => {
// Browsers can do everything. Except show a progress bar while uploading...
let progressBox, progressPercentage, progressBar
dom._kids(importProgress,
progressBox=dom.div(
dom.div('Uploading... ', progressPercentage=dom.span()),
),
)
importProgress.style.display = ''
const xhr = new window.XMLHttpRequest()
xhr.open('POST', 'import', true)
xhr.upload.addEventListener('progress', (e) => {
if (!e.lengthComputable) {
return
}
const pct = Math.floor(100*e.loaded/e.total)
dom._kids(progressPercentage, pct+'%')
})
xhr.addEventListener('load', () => {
console.log('upload done', {xhr: xhr, status: xhr.status})
if (xhr.status !== 200) {
reject({message: 'status '+xhr.status})
return
}
let resp
try {
resp = JSON.parse(xhr.responseText)
} catch (err) {
reject({message: 'parsing resonse json: '+err.message})
return
}
resolve(resp)
})
xhr.addEventListener('error', (e) => reject({message: 'upload error', event: e}))
xhr.addEventListener('abort', (e) => reject({message: 'upload aborted', event: e}))
xhr.send(new window.FormData(importForm))
})
}
try {
const p = request()
importFieldset.disabled = true
const result = await p
try {
window.sessionStorage.setItem('ImportToken', result.ImportToken)
} catch (err) {
console.log('storing import token in session storage', {err})
// Ignore error, could be some browser security thing like private browsing.
}
await importTrack(result.ImportToken)
} catch (err) {
console.log({err})
window.alert('Error: '+err.message)
} finally {
importFieldset.disabled = false
}
},
importFieldset=dom.fieldset(
dom.div(
style({marginBottom: '1ex'}),
dom.label(
dom.div(style({marginBottom: '.5ex'}), 'File'),
mailboxFile=dom.input(attr({type: 'file', required: '', name: 'file'}), function focus() {
mailboxFileHint.style.display = ''
}),
),
mailboxFileHint=dom.p(style({display: 'none', fontStyle: 'italic', marginTop: '.5ex'}), 'This file must either be a zip file or a gzipped tar file with mbox and/or maildir mailboxes. For maildirs, an optional file "dovecot-keywords" is read additional keywords, like Forwarded/Junk/NotJunk. If an imported mailbox already exists by name, messages are added to the existing mailbox. If a mailbox does not yet exist it will be created.'),
),
dom.div(
style({marginBottom: '1ex'}),
dom.label(
dom.div(style({marginBottom: '.5ex'}), 'Skip mailbox prefix (optional)'),
mailboxPrefix=dom.input(attr({name: 'skipMailboxPrefix'}), function focus() {
mailboxPrefixHint.style.display = ''
}),
),
mailboxPrefixHint=dom.p(style({display: 'none', fontStyle: 'italic', marginTop: '.5ex'}), 'If set, any mbox/maildir path with this prefix will have it stripped before importing. For example, if all mailboxes are in a directory "Takeout", specify that path in the field above so mailboxes like "Takeout/Inbox.mbox" are imported into a mailbox called "Inbox" instead of "Takeout/Inbox".'),
),
dom.div(
dom.button('Upload and import'),
dom.p(style({fontStyle: 'italic', marginTop: '.5ex'}), 'The file is uploaded first, then its messages are imported. Importing is done in a transaction, you can abort the entire import before it is finished.'),
),
),
),
importAbortBox=dom.div(), // Outside fieldset because it gets disabled, above progress because may be scrolling it down quickly with problems.
importProgress=dom.div(
style({display: 'none'}),
),
footer,
)
// Try to show the progress of an earlier import session. The user may have just
// refreshed the browser.
let importToken
try {
importToken = window.sessionStorage.getItem('ImportToken')
} catch (err) {
console.log('looking up ImportToken in session storage', {err})
return
}
if (!importToken) {
return
}
importFieldset.disabled = true
dom._kids(importProgress,
dom.div(
dom.div('Reconnecting to import...'),
),
)
importProgress.style.display = ''
importTrack(importToken)
.catch((err) => {
if (window.confirm('Error reconnecting to import. Remove this import session?')) {
window.sessionStorage.removeItem('ImportToken')
dom._kids(importProgress)
importProgress.style.display = 'none'
}
})
.finally(() => {
importFieldset.disabled = false
})
}
const destination = async (name) => {

178
http/account_test.go
View file

@ -1,3 +1,179 @@
package http
// todo: write test for account api calls, at least for authentation and SetPassword.
import (
"archive/tar"
"archive/zip"
"bytes"
"compress/gzip"
"context"
"encoding/json"
"io"
"mime/multipart"
"net/http"
"net/http/httptest"
"os"
"path"
"path/filepath"
"strings"
"testing"
"github.com/mjl-/mox/mlog"
"github.com/mjl-/mox/mox-"
"github.com/mjl-/mox/store"
)
func tcheck(t *testing.T, err error, msg string) {
t.Helper()
if err != nil {
t.Fatalf("%s: %s", msg, err)
}
}
func TestAccount(t *testing.T) {
os.RemoveAll("../testdata/httpaccount/data")
mox.ConfigStaticPath = "../testdata/httpaccount/mox.conf"
mox.ConfigDynamicPath = filepath.Join(filepath.Dir(mox.ConfigStaticPath), "domains.conf")
mox.MustLoadConfig()
acc, err := store.OpenAccount("mjl")
tcheck(t, err, "open account")
defer acc.Close()
switchDone := store.Switchboard()
defer close(switchDone)
log := mlog.New("store")
test := func(authHdr string, expect string) {
t.Helper()
w := httptest.NewRecorder()
r := httptest.NewRequest("GET", "/ignored", nil)
if authHdr != "" {
r.Header.Add("Authorization", authHdr)
}
ok := checkAccountAuth(context.Background(), log, w, r)
if ok != expect {
t.Fatalf("got %v, expected %v", ok, expect)
}
}
const authOK = "Basic bWpsQG1veC5leGFtcGxlOnRlc3QxMjM0" // mjl@mox.example:test1234
const authBad = "Basic bWpsQG1veC5leGFtcGxlOmJhZHBhc3N3b3Jk" // mjl@mox.example:badpassword
authCtx := context.WithValue(context.Background(), authCtxKey, "mjl")
test(authOK, "") // No password set yet.
Account{}.SetPassword(authCtx, "test1234")
test(authOK, "mjl")
test(authBad, "")
_, dests := Account{}.Destinations(authCtx)
Account{}.DestinationSave(authCtx, "mjl", dests["mjl"], dests["mjl"]) // todo: save modified value and compare it afterwards
// Import mbox/maildir tgz/zip.
testImport := func(filename string, expect int) {
t.Helper()
var reqBody bytes.Buffer
mpw := multipart.NewWriter(&reqBody)
part, err := mpw.CreateFormFile("file", path.Base(filename))
tcheck(t, err, "creating form file")
buf, err := os.ReadFile(filename)
tcheck(t, err, "reading file")
_, err = part.Write(buf)
tcheck(t, err, "write part")
err = mpw.Close()
tcheck(t, err, "close multipart writer")
r := httptest.NewRequest("POST", "/import", &reqBody)
r.Header.Add("Content-Type", mpw.FormDataContentType())
r.Header.Add("Authorization", authOK)
w := httptest.NewRecorder()
accountHandle(w, r)
if w.Code != http.StatusOK {
t.Fatalf("import, got status code %d, expected 200: %s", w.Code, w.Body.Bytes())
}
m := map[string]string{}
if err := json.Unmarshal(w.Body.Bytes(), &m); err != nil {
t.Fatalf("parsing import response: %v", err)
}
token := m["ImportToken"]
l := importListener{token, make(chan importEvent, 100), make(chan bool)}
importers.Register <- &l
if !<-l.Register {
t.Fatalf("register failed")
}
defer func() {
importers.Unregister <- &l
}()
count := 0
loop:
for {
e := <-l.Events
switch x := e.Event.(type) {
case importCount:
count += x.Count
case importProblem:
t.Fatalf("unexpected problem: %q", x.Message)
case importDone:
break loop
case importAborted:
t.Fatalf("unexpected aborted import")
default:
panic("missing case")
}
}
if count != expect {
t.Fatalf("imported %d messages, expected %d", count, expect)
}
}
testImport("../testdata/importtest.mbox.zip", 2)
testImport("../testdata/importtest.maildir.tgz", 2)
testExport := func(httppath string, iszip bool, expectFiles int) {
t.Helper()
r := httptest.NewRequest("GET", httppath, nil)
r.Header.Add("Authorization", authOK)
w := httptest.NewRecorder()
accountHandle(w, r)
if w.Code != http.StatusOK {
t.Fatalf("export, got status code %d, expected 200: %s", w.Code, w.Body.Bytes())
}
var count int
if iszip {
buf := w.Body.Bytes()
zr, err := zip.NewReader(bytes.NewReader(buf), int64(len(buf)))
tcheck(t, err, "reading zip")
for _, f := range zr.File {
if !strings.HasSuffix(f.Name, "/") {
count++
}
}
} else {
gzr, err := gzip.NewReader(w.Body)
tcheck(t, err, "gzip reader")
tr := tar.NewReader(gzr)
for {
h, err := tr.Next()
if err == io.EOF {
break
}
tcheck(t, err, "next file in tar")
if !strings.HasSuffix(h.Name, "/") {
count++
}
_, err = io.Copy(io.Discard, tr)
tcheck(t, err, "reading from tar")
}
}
if count != expectFiles {
t.Fatalf("export, has %d files, expected %d", count, expectFiles)
}
}
testExport("/mail-export-maildir.tgz", false, 6) // 2 mailboxes, each with 2 messages and a dovecot-keyword file
testExport("/mail-export-maildir.zip", true, 6)
testExport("/mail-export-mbox.tgz", false, 2)
testExport("/mail-export-mbox.zip", true, 2)
}

13
http/accountapi.json
View file

@ -59,6 +59,19 @@
}
],
"Returns": []
},
{
"Name": "ImportAbort",
"Docs": "ImportAbort aborts an import that is in progress. If the import exists and isn't\nfinished, no changes will have been made by the import.",
"Params": [
{
"Name": "importToken",
"Typewords": [
"string"
]
}
],
"Returns": []
}
],
"Sections": [],

1
http/admin.html
View file

@ -110,6 +110,7 @@ return [dom, style, attr, prop]
const green = '#1dea20'
const yellow = '#ffe400'
const red = '#ff7443'
const blue = '#8bc8ff'
const link = (href, anchorOpt) => dom.a(attr({href: href, rel: 'noopener noreferrer'}), anchorOpt || href)

788
http/import.go Normal file
View file

@ -0,0 +1,788 @@
package http
import (
"archive/tar"
"archive/zip"
"bufio"
"bytes"
"compress/gzip"
"context"
cryptrand "crypto/rand"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"path"
"runtime/debug"
"strconv"
"strings"
"time"
"golang.org/x/text/unicode/norm"
"github.com/mjl-/bstore"
"github.com/mjl-/mox/message"
"github.com/mjl-/mox/mlog"
"github.com/mjl-/mox/mox-"
"github.com/mjl-/mox/store"
)
type importListener struct {
Token string
Events chan importEvent
Register chan bool // Whether register is successful.
}
type importEvent struct {
Token string
SSEMsg []byte // Full SSE message, including event: ... and data: ... \n\n
Event any // nil, importCount, importProblem, importDone, importAborted
Cancel func() // For cancelling the context causing abort of the import. Set in first, import-registering, event.
}
type importAbortRequest struct {
Token string
Response chan error
}
var importers = struct {
Register chan *importListener
Unregister chan *importListener
Events chan importEvent
Abort chan importAbortRequest
}{
make(chan *importListener, 1),
make(chan *importListener, 1),
make(chan importEvent),
make(chan importAbortRequest),
}
func init() {
go importManage()
}
func importManage() {
log := mlog.New("httpimport")
defer func() {
if x := recover(); x != nil {
log.Error("import manage panic", mlog.Field("err", x))
debug.PrintStack()
}
}()
type state struct {
MailboxCounts map[string]int
Problems []string
Done *time.Time
Aborted *time.Time
Listeners map[*importListener]struct{}
Cancel func()
}
imports := map[string]state{} // Token to state.
for {
select {
case l := <-importers.Register:
// If we have state, send it so the client is up to date.
if s, ok := imports[l.Token]; ok {
l.Register <- true
s.Listeners[l] = struct{}{}
sendEvent := func(kind string, v any) {
buf, err := json.Marshal(v)
if err != nil {
log.Errorx("marshal event", err, mlog.Field("kind", kind), mlog.Field("event", v))
return
}
ssemsg := fmt.Sprintf("event: %s\ndata: %s\n\n", kind, buf)
select {
case l.Events <- importEvent{kind, []byte(ssemsg), nil, nil}:
default:
log.Debug("dropped initial import event to slow consumer")
}
}
for m, c := range s.MailboxCounts {
sendEvent("count", importCount{m, c})
}
for _, p := range s.Problems {
sendEvent("problem", importProblem{p})
}
if s.Done != nil {
sendEvent("done", importDone{})
} else if s.Aborted != nil {
sendEvent("aborted", importAborted{})
}
} else {
l.Register <- false
}
case l := <-importers.Unregister:
delete(imports[l.Token].Listeners, l)
case e := <-importers.Events:
s, ok := imports[e.Token]
if !ok {
s := state{
MailboxCounts: map[string]int{},
Listeners: map[*importListener]struct{}{},
Cancel: e.Cancel,
}
imports[e.Token] = s
}
for l := range s.Listeners {
select {
case l.Events <- e:
default:
log.Debug("dropped import event to slow consumer")
}
}
if e.Event != nil {
s := imports[e.Token]
switch x := e.Event.(type) {
case importCount:
s.MailboxCounts[x.Mailbox] = x.Count
case importProblem:
s.Problems = append(s.Problems, x.Message)
case importDone:
now := time.Now()
s.Done = &now
case importAborted:
now := time.Now()
s.Aborted = &now
}
imports[e.Token] = s
}
case a := <-importers.Abort:
s, ok := imports[a.Token]
if !ok {
a.Response <- errors.New("import not found")
return
}
if s.Done != nil {
a.Response <- errors.New("import already finished")
return
}
s.Cancel()
a.Response <- nil
}
// Cleanup old state.
for t, s := range imports {
if len(s.Listeners) > 0 {
continue
}
if s.Done != nil && time.Since(*s.Done) > time.Minute || s.Aborted != nil && time.Since(*s.Aborted) > time.Minute {
delete(imports, t)
}
}
}
}
type importCount struct {
Mailbox string
Count int
}
type importProblem struct {
Message string
}
type importDone struct{}
type importAborted struct{}
// importStart prepare the import and launches the goroutine to actually import.
// importStart is responsible for closing f.
func importStart(log *mlog.Log, accName string, f *os.File, skipMailboxPrefix string) (string, error) {
defer func() {
if f != nil {
f.Close()
}
}()
buf := make([]byte, 16)
if _, err := cryptrand.Read(buf); err != nil {
return "", err
}
token := fmt.Sprintf("%x", buf)
if _, err := f.Seek(0, 0); err != nil {
return "", fmt.Errorf("seek to start of file: %v", err)
}
// Recognize file format.
var iszip bool
magicZip := []byte{0x50, 0x4b, 0x03, 0x04}
magicGzip := []byte{0x1f, 0x8b}
magic := make([]byte, 4)
if _, err := f.ReadAt(magic, 0); err != nil {
return "", fmt.Errorf("detecting file format: %v", err)
}
if bytes.Equal(magic, magicZip) {
iszip = true
} else if !bytes.Equal(magic[:2], magicGzip) {
return "", fmt.Errorf("file is not a zip or gzip file")
}
var zr *zip.Reader
var tr *tar.Reader
if iszip {
fi, err := f.Stat()
if err != nil {
return "", fmt.Errorf("stat temporary import zip file: %v", err)
}
zr, err = zip.NewReader(f, fi.Size())
if err != nil {
return "", fmt.Errorf("opening zip file: %v", err)
}
} else {
gzr, err := gzip.NewReader(f)
if err != nil {
return "", fmt.Errorf("gunzip: %v", err)
}
tr = tar.NewReader(gzr)
}
acc, err := store.OpenAccount(accName)
if err != nil {
return "", fmt.Errorf("open acount: %v", err)
}
acc.Lock() // Not using WithWLock because importMessage is responsible for unlocking.
tx, err := acc.DB.Begin(true)
if err != nil {
acc.Unlock()
acc.Close()
return "", fmt.Errorf("start transaction: %v", err)
}
// Ensure token is registered before returning, with context that can be canceled.
ctx, cancel := context.WithCancel(mox.Shutdown)
importers.Events <- importEvent{token, []byte(": keepalive\n\n"), nil, cancel}
log.Info("starting import")
go importMessages(ctx, log.WithCid(mox.Cid()), token, acc, tx, zr, tr, f, skipMailboxPrefix)
f = nil // importMessages is now responsible for closing.
return token, nil
}
// importMessages imports the messages from zip/tgz file f.
// importMessages is responsible for unlocking and closing acc, and closing tx and f.
func importMessages(ctx context.Context, log *mlog.Log, token string, acc *store.Account, tx *bstore.Tx, zr *zip.Reader, tr *tar.Reader, f *os.File, skipMailboxPrefix string) {
// If a fatal processing error occurs, we panic with this type.
type importError struct{ Err error }
// During import we collect all changes and broadcast them at the end, when successful.
var changes []store.Change
// ID's of delivered messages. If we have to rollback, we have to remove this files.
var deliveredIDs []int64
ximportcheckf := func(err error, format string, args ...any) {
if err != nil {
panic(importError{fmt.Errorf("%s: %s", fmt.Sprintf(format, args...), err)})
}
}
sendEvent := func(kind string, v any) {
buf, err := json.Marshal(v)
if err != nil {
log.Errorx("marshal event", err, mlog.Field("kind", kind), mlog.Field("event", v))
return
}
ssemsg := fmt.Sprintf("event: %s\ndata: %s\n\n", kind, buf)
importers.Events <- importEvent{token, []byte(ssemsg), v, nil}
}
problemf := func(format string, args ...any) {
msg := fmt.Sprintf(format, args...)
sendEvent("problem", importProblem{Message: msg})
}
canceled := func() bool {
select {
case <-ctx.Done():
sendEvent("aborted", importAborted{})
return true
default:
return false
}
}
defer func() {
if err := f.Close(); err != nil {
log.Errorx("closing uploaded messages file", err)
}
for _, id := range deliveredIDs {
p := acc.MessagePath(id)
if err := os.Remove(p); err != nil {
log.Errorx("closing message file after import error", err, mlog.Field("path", p))
}
}
if tx != nil {
if err := tx.Rollback(); err != nil {
log.Errorx("rolling back transaction", err)
}
}
if acc != nil {
acc.Unlock()
if err := acc.Close(); err != nil {
log.Errorx("closing account", err)
}
}
x := recover()
if x == nil {
return
}
if err, ok := x.(importError); ok {
log.Errorx("import error", err.Err)
problemf("%s (aborting)", err.Err)
sendEvent("aborted", importAborted{})
} else {
log.Error("import panic", mlog.Field("err", x))
debug.PrintStack()
}
}()
conf, _ := acc.Conf()
jf, _, err := acc.OpenJunkFilter(log)
if err != nil && !errors.Is(err, store.ErrNoJunkFilter) {
ximportcheckf(err, "open junk filter")
}
defer func() {
if jf != nil {
err := jf.CloseDiscard()
log.Check(err, "closing junk filter")
}
}()
// Mailboxes we imported, and message counts.
mailboxes := map[string]store.Mailbox{}
messages := map[string]int{}
// For maildirs, we are likely to get a possible dovecot-keywords file after having imported the messages. Once we see the keywords, we use them. But before that time we remember which messages miss a keywords. Once the keywords become available, we'll fix up the flags for the unknown messages
mailboxKeywords := map[string]map[rune]string{} // Mailbox to 'a'-'z' to flag name.
mailboxMissingKeywordMessages := map[string]map[int64]string{} // Mailbox to message id to string consisting of the unrecognized flags.
// Previous mailbox an event was sent for. We send an event for new mailboxes, when
// another 100 messages were added, when adding a message to another mailbox, and
// finally at the end as a closing statement.
var prevMailbox string
trainMessage := func(m *store.Message, p message.Part, pos string) {
words, err := jf.ParseMessage(p)
if err != nil {
problemf("parsing message %s for updating junk filter: %v (continuing)", pos, err)
return
}
err = jf.Train(!m.Junk, words)
if err != nil {
problemf("training junk filter for message %s: %v (continuing)", pos, err)
return
}
m.TrainedJunk = &m.Junk
}
openTrainMessage := func(m *store.Message) {
path := acc.MessagePath(m.ID)
f, err := os.Open(path)
if err != nil {
problemf("opening message again for training junk filter: %v (continuing)", err)
return
}
defer f.Close()
p, err := m.LoadPart(f)
if err != nil {
problemf("loading parsed message again for training junk filter: %v (continuing)", err)
return
}
trainMessage(m, p, fmt.Sprintf("message id %d", m.ID))
}
xensureMailbox := func(name string) store.Mailbox {
name = norm.NFC.String(name)
if strings.ToLower(name) == "inbox" {
name = "Inbox"
}
if mb, ok := mailboxes[name]; ok {
return mb
}
var p string
var mb store.Mailbox
for i, e := range strings.Split(name, "/") {
if i == 0 {
p = e
} else {
p = path.Join(p, e)
}
if _, ok := mailboxes[p]; ok {
continue
}
q := bstore.QueryTx[store.Mailbox](tx)
q.FilterNonzero(store.Mailbox{Name: p})
var err error
mb, err = q.Get()
if err == bstore.ErrAbsent {
uidvalidity, err := acc.NextUIDValidity(tx)
ximportcheckf(err, "finding next uid validity")
mb = store.Mailbox{
Name: p,
UIDValidity: uidvalidity,
UIDNext: 1,
// Do not assign special-use flags. This existing account probably already has such mailboxes.
}
err = tx.Insert(&mb)
ximportcheckf(err, "inserting mailbox in database")
changes = append(changes, store.ChangeAddMailbox{Name: p})
// todo: should we also subscribe to the mailbox?
} else if err != nil {
ximportcheckf(err, "creating mailbox %s (aborting)", p)
}
if prevMailbox != "" && mb.Name != prevMailbox {
sendEvent("count", importCount{prevMailbox, messages[prevMailbox]})
}
mailboxes[mb.Name] = mb
sendEvent("count", importCount{mb.Name, 0})
prevMailbox = mb.Name
}
return mb
}
xdeliver := func(mb store.Mailbox, m *store.Message, f *os.File, pos string) {
defer func() {
if f != nil {
err := os.Remove(f.Name())
log.Check(err, "removing temporary message file for delivery")
err = f.Close()
log.Check(err, "closing temporary message file for delivery")
}
x := recover()
if x != nil {
// todo: get a variant of DeliverX that returns an error instead of panicking.
log.Error("delivery panic", mlog.Field("err", x))
}
}()
m.MailboxID = mb.ID
m.MailboxOrigID = mb.ID
// Parse message and store parsed information for later fast retrieval.
p, err := message.EnsurePart(f, m.Size)
if err != nil {
problemf("parsing message %s: %s (continuing)", pos, err)
}
m.ParsedBuf, err = json.Marshal(p)
ximportcheckf(err, "marshal parsed message structure")
if m.Received.IsZero() {
if p.Envelope != nil && !p.Envelope.Date.IsZero() {
m.Received = p.Envelope.Date
} else {
m.Received = time.Now()
}
}
// We set the flags that Deliver would set now and train ourselves. This prevents
// Deliver from training, which would open the junk filter, change it, and write it
// back to disk, for each message (slow).
m.JunkFlagsForMailbox(mb.Name, conf)
if jf != nil && m.NeedsTraining() {
trainMessage(m, p, pos)
}
const consumeFile = true
const sync = false
const notrain = true
acc.DeliverX(log, tx, m, f, consumeFile, mb.Sent, sync, notrain) // todo: need a deliver that returns an error.
deliveredIDs = append(deliveredIDs, m.ID)
changes = append(changes, store.ChangeAddUID{MailboxID: m.MailboxID, UID: m.UID, Flags: m.Flags})
messages[mb.Name]++
if messages[mb.Name]%100 == 0 || prevMailbox != mb.Name {
prevMailbox = mb.Name
sendEvent("count", importCount{mb.Name, messages[mb.Name]})
}
f = nil
}
ximportMbox := func(mailbox, filename string, r io.Reader) {
if mailbox == "" {
problemf("empty mailbox name for mbox file %s (skipping)", filename)
return
}
mb := xensureMailbox(mailbox)
mr := store.NewMboxReader(store.CreateMessageTemp, filename, r, log)
for {
m, mf, pos, err := mr.Next()
if err == io.EOF {
break
} else if err != nil {
ximportcheckf(err, "next message in mbox file")
}
xdeliver(mb, m, mf, pos)
}
}
ximportMaildir := func(mailbox, filename string, r io.Reader) {
if mailbox == "" {
problemf("empty mailbox name for maildir file %s (skipping)", filename)
return
}
mb := xensureMailbox(mailbox)
f, err := store.CreateMessageTemp("import")
ximportcheckf(err, "creating temp message")
defer func() {
if f != nil {
err := os.Remove(f.Name())
log.Check(err, "removing temporary file for delivery")
err = f.Close()
log.Check(err, "closing temporary file for delivery")
}
}()
// Copy data, changing bare \n into \r\n.
br := bufio.NewReader(r)
w := bufio.NewWriter(f)
var size int64
for {
line, err := br.ReadBytes('\n')
if err != nil && err != io.EOF {
ximportcheckf(err, "reading message")
}
if len(line) > 0 {
if !bytes.HasSuffix(line, []byte("\r\n")) {
line = append(line[:len(line)-1], "\r\n"...)
}
n, err := w.Write(line)
ximportcheckf(err, "writing message")
size += int64(n)
}
if err == io.EOF {
break
}
}
err = w.Flush()
ximportcheckf(err, "writing message")
var received time.Time
t := strings.SplitN(path.Base(filename), ".", 2)
if v, err := strconv.ParseInt(t[0], 10, 64); err == nil {
received = time.Unix(v, 0)
}
// Parse flags. See https://cr.yp.to/proto/maildir.html.
var keepFlags string
flags := store.Flags{}
t = strings.SplitN(path.Base(filename), ":2,", 2)
if len(t) == 2 {
for _, c := range t[1] {
switch c {
case 'P':
// Passed, doesn't map to a common IMAP flag.
case 'R':
flags.Answered = true
case 'S':
flags.Seen = true
case 'T':
flags.Deleted = true
case 'D':
flags.Draft = true
case 'F':
flags.Flagged = true
default:
if c >= 'a' && c <= 'z' {
keywords, ok := mailboxKeywords[mailbox]
if !ok {
// No keywords file seen yet, we'll try later if it comes in.
keepFlags += string(c)
} else if kw, ok := keywords[c]; ok {
flagSet(&flags, strings.ToLower(kw))
}
}
}
}
}
m := store.Message{
Received: received,
Flags: flags,
Size: size,
}
xdeliver(mb, &m, f, filename)
f = nil
if keepFlags != "" {
if _, ok := mailboxMissingKeywordMessages[mailbox]; !ok {
mailboxMissingKeywordMessages[mailbox] = map[int64]string{}
}
mailboxMissingKeywordMessages[mailbox][m.ID] = keepFlags
}
}
importFile := func(name string, r io.Reader) {
origName := name
if strings.HasPrefix(name, skipMailboxPrefix) {
name = strings.TrimPrefix(name[len(skipMailboxPrefix):], "/")
}
if strings.HasSuffix(name, "/") {
name = strings.TrimSuffix(name, "/")
dir := path.Dir(name)
switch path.Base(dir) {
case "new", "cur", "tmp":
// Maildir, ensure it exists.
mailbox := path.Dir(dir)
xensureMailbox(mailbox)
}
// Otherwise, this is just a directory that probably holds mbox files and maildirs.
return
}
if strings.HasSuffix(path.Base(name), ".mbox") {
mailbox := name[:len(name)-len(".mbox")]
ximportMbox(mailbox, origName, r)
return
}
dir := path.Dir(name)
dirbase := path.Base(dir)
switch dirbase {
case "new", "cur", "tmp":
mailbox := path.Dir(dir)
ximportMaildir(mailbox, origName, r)
default:
if path.Base(name) == "dovecot-keywords" {
mailbox := path.Dir(name)
keywords := map[rune]string{}
words, err := store.ParseDovecotKeywords(r, log)
log.Check(err, "parsing dovecot keywords for mailbox", mlog.Field("mailbox", mailbox))
for i, kw := range words {
keywords['a'+rune(i)] = kw
}
mailboxKeywords[mailbox] = keywords
for id, chars := range mailboxMissingKeywordMessages[mailbox] {
var flags, zeroflags store.Flags
for _, c := range chars {
kw, ok := keywords[c]
if !ok {
problemf("unspecified message flag %c for message id %d (continuing)", c, id)
continue
}
flagSet(&flags, strings.ToLower(kw))
}
if flags == zeroflags {
continue
}
m := store.Message{ID: id}
err := tx.Get(&m)
ximportcheckf(err, "get imported message for flag update")
m.Flags = m.Flags.Set(flags, flags)
// We train before updating, training may set m.TrainedJunk.
if jf != nil && m.NeedsTraining() {
openTrainMessage(&m)
}
err = tx.Update(&m)
ximportcheckf(err, "updating message after flag update")
changes = append(changes, store.ChangeFlags{MailboxID: m.MailboxID, UID: m.UID, Mask: flags, Flags: flags})
}
delete(mailboxMissingKeywordMessages, mailbox)
} else {
problemf("unrecognized file %s (skipping)", origName)
}
}
}
if zr != nil {
for _, f := range zr.File {
if canceled() {
return
}
zf, err := f.Open()
if err != nil {
problemf("opening file %s in zip: %v", f.Name, err)
continue
}
importFile(f.Name, zf)
zf.Close()
}
} else {
for {
if canceled() {
return
}
h, err := tr.Next()
if err == io.EOF {
break
} else if err != nil {
problemf("reading next tar header: %v (aborting)", err)
return
}
importFile(h.Name, tr)
}
}
total := 0
for _, count := range messages {
total += count
}
log.Debug("message imported", mlog.Field("total", total))
// Send final update for count of last-imported mailbox.
if prevMailbox != "" {
sendEvent("count", importCount{prevMailbox, messages[prevMailbox]})
}
err = tx.Commit()
tx = nil
ximportcheckf(err, "commit")
deliveredIDs = nil
if jf != nil {
err := jf.Close()
if err != nil {
problemf("saving changes of training junk filter: %v (continuing)", err)
log.Errorx("saving changes of training junk filter", err)
}
jf = nil
}
comm := store.RegisterComm(acc)
defer comm.Unregister()
comm.Broadcast(changes)
acc.Unlock()
if err := acc.Close(); err != nil {
log.Errorx("closing account after import", err)
// Continue
}
acc = nil
sendEvent("done", importDone{})
}
func flagSet(flags *store.Flags, word string) {
// todo: custom labels, e.g. $label1, JunkRecorded?
switch word {
case "forwarded", "$forwarded":
flags.Forwarded = true
case "junk", "$junk":
flags.Junk = true
case "notjunk", "$notjunk", "nonjunk", "$nonjunk":
flags.Notjunk = true
case "phishing", "$phishing":
flags.Phishing = true
case "mdnsent", "$mdnsent":
flags.MDNSent = true
}
}

8
imapserver/server.go
View file

@ -647,7 +647,7 @@ func serve(listenerName string, cid int64, tlsConfig *tls.Config, nc net.Conn, x
}()
select {
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
// ../rfc/9051:5381
c.writelinef("* BYE mox shutting down")
panic(errIO)
@ -777,7 +777,7 @@ func (c *conn) command() {
c.cmdMetric = "(unrecognized)"
select {
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
// ../rfc/9051:5375
c.writelinef("* BYE shutting down")
panic(errIO)
@ -2522,7 +2522,7 @@ func (c *conn) cmdAppend(tag, cmd string, p *parser) {
MsgPrefix: msgPrefix,
}
isSent := name == "Sent"
c.account.DeliverX(c.log, tx, &msg, msgFile, true, isSent, true)
c.account.DeliverX(c.log, tx, &msg, msgFile, true, isSent, true, false)
})
// Fetch pending changes, possibly with new UIDs, so we can apply them before adding our own new UID.
@ -2572,7 +2572,7 @@ wait:
case changes := <-c.comm.Changes:
c.applyChanges(changes, false)
c.xflush()
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
// ../rfc/9051:5375
c.writelinef("* BYE shutting down")
panic(errIO)

417
import.go
View file

@ -1,8 +1,6 @@
package main
import (
"bufio"
"bytes"
"encoding/json"
"errors"
"fmt"
@ -11,31 +9,26 @@ import (
"os"
"path/filepath"
"runtime/debug"
"strconv"
"strings"
"time"
"github.com/mjl-/mox/junk"
"github.com/mjl-/mox/message"
"github.com/mjl-/mox/metrics"
"github.com/mjl-/mox/mlog"
"github.com/mjl-/mox/store"
)
// todo: implement export of all maildirs to a zip file and also import of such a zip file.
// todo: add option to trust imported messages, causing us to look at Authentication-Results and Received-SPF headers and add eg verified spf/dkim/dmarc domains to our store, to jumpstart reputation.
const importCommonHelp = `By default, messages will train the junk filter based on their flags and
mailbox naming. If the destination mailbox name starts with "junk" or "spam"
(case insensitive), messages are imported and trained as junk regardless of
pre-existing flags. Use the -train=false flag to prevent training the filter.
const importCommonHelp = `By default, messages will train the junk filter based on their flags and, if
"automatic junk flags" configuration is set, based on mailbox naming.
If the destination mailbox is "Sent", the recipients of the messages are added
to the message metadata, causing later incoming messages from these recipients
to be accepted, unless other reputation signals prevent that.
The message "read"/"seen" flag can be overridden during import with the
-markread flag.
Users can also import mailboxes/messages through the account web page by
uploading a zip or tgz file with mbox and/or maildirs.
`
func cmdImportMaildir(c *cmd) {
@ -43,27 +36,22 @@ func cmdImportMaildir(c *cmd) {
c.help = `Import a maildir into an account.
` + importCommonHelp + `
Mailbox flags, like "seen", "answered", "forwarded", will be imported. An
attempt is made to parse dovecot keyword files.
Mailbox flags, like "seen", "answered", will be imported. An optional
dovecot-keywords file can specify additional flags, like Forwarded/Junk/NotJunk.
The maildir files/directories are read by the mox process, so make sure it has
access to the maildir directories/files.
`
var train bool
var markRead bool
c.flag.BoolVar(&train, "train", true, "train junkfilter with messages")
c.flag.BoolVar(&markRead, "markread", false, "mark all imported messages as read")
args := c.Parse()
xcmdImport(false, train, markRead, args, c)
xcmdImport(false, args, c)
}
func cmdImportMbox(c *cmd) {
c.params = "accountname mailboxname mbox"
c.help = `Import an mbox into an account.
Using mbox is not recommended, maildir is a better format.
Using mbox is not recommended, maildir is a better defined format.
` + importCommonHelp + `
@ -71,16 +59,11 @@ The mailbox is read by the mox process, so make sure it has access to the
maildir directories/files.
`
var train bool
var markRead bool
c.flag.BoolVar(&train, "train", true, "train junkfilter with messages")
c.flag.BoolVar(&markRead, "markread", false, "mark all imported messages as read")
args := c.Parse()
xcmdImport(true, train, markRead, args, c)
xcmdImport(true, args, c)
}
func xcmdImport(mbox, train, markRead bool, args []string, c *cmd) {
func xcmdImport(mbox bool, args []string, c *cmd) {
if len(args) != 3 {
c.Usage()
}
@ -106,16 +89,6 @@ func xcmdImport(mbox, train, markRead bool, args []string, c *cmd) {
ctl.xwrite(account)
ctl.xwrite(mailbox)
ctl.xwrite(src)
if train {
ctl.xwrite("train")
} else {
ctl.xwrite("notrain")
}
if markRead {
ctl.xwrite("markread")
} else {
ctl.xwrite("nomarkread")
}
ctl.xreadok()
fmt.Fprintln(os.Stderr, "importing...")
for {
@ -140,8 +113,6 @@ func importctl(ctl *ctl, mbox bool) {
> account
> mailbox
> src (mbox file or maildir directory)
> "train" or "notrain"
> "markread" or "nomarkread"
< "ok" or error
< "progress" count (zero or more times, once for every 1000 messages)
< "ok" when done, or error
@ -150,28 +121,6 @@ func importctl(ctl *ctl, mbox bool) {
account := ctl.xread()
mailbox := ctl.xread()
src := ctl.xread()
xtrain := ctl.xread()
xmarkread := ctl.xread()
var train bool
switch xtrain {
case "train":
train = true
case "notrain":
train = false
default:
ctl.xerror("bad value for train: " + xtrain)
}
var markRead bool
switch xmarkread {
case "markread":
markRead = true
case "nomarkread":
markRead = false
default:
ctl.xerror("bad value for markread: " + xmarkread)
}
kind := "maildir"
if mbox {
@ -182,7 +131,7 @@ func importctl(ctl *ctl, mbox bool) {
var err error
var mboxf *os.File
var mdnewf, mdcurf *os.File
var msgreader msgReader
var msgreader store.MsgSource
defer func() {
if mboxf != nil {
@ -217,23 +166,19 @@ func importctl(ctl *ctl, mbox bool) {
// Messages don't always have a junk flag set. We'll assume anything in a mailbox
// starting with junk or spam is junk mail.
var msgJunkFlags store.Message
conf, _ := a.Conf()
msgJunkFlags.JunkFlagsForMailbox(mailbox, conf)
// First check if we can access the mbox/maildir.
// Mox needs to be able to access those files, the user running the import command
// may be a different user who can access the files.
if mbox {
mboxf, err = os.Open(src)
ctl.xcheck(err, "open mbox file")
msgreader = newMboxReader(msgJunkFlags.Junk, msgJunkFlags.Notjunk, store.CreateMessageTemp, mboxf, ctl.log)
msgreader = store.NewMboxReader(store.CreateMessageTemp, src, mboxf, ctl.log)
} else {
mdnewf, err = os.Open(filepath.Join(src, "new"))
ctl.xcheck(err, "open subdir new of maildir")
mdcurf, err = os.Open(filepath.Join(src, "cur"))
ctl.xcheck(err, "open subdir cur of maildir")
msgreader = newMaildirReader(msgJunkFlags.Junk, msgJunkFlags.Notjunk, store.CreateMessageTemp, mdnewf, mdcurf, ctl.log)
msgreader = store.NewMaildirReader(store.CreateMessageTemp, mdnewf, mdcurf, ctl.log)
}
tx, err := a.DB.Begin(true)
@ -279,7 +224,8 @@ func importctl(ctl *ctl, mbox bool) {
const consumeFile = true
isSent := mailbox == "Sent"
const sync = false
a.DeliverX(ctl.log, tx, m, mf, consumeFile, isSent, sync)
const notrain = true
a.DeliverX(ctl.log, tx, m, mf, consumeFile, isSent, sync, notrain)
deliveredIDs = append(deliveredIDs, m.ID)
ctl.log.Debug("delivered message", mlog.Field("id", m.ID))
changes = append(changes, store.ChangeAddUID{MailboxID: m.MailboxID, UID: m.UID, Flags: m.Flags})
@ -292,9 +238,7 @@ func importctl(ctl *ctl, mbox bool) {
var mb store.Mailbox
mb, changes = a.MailboxEnsureX(tx, mailbox, true)
var jf *junk.Filter
if train {
jf, _, err = a.OpenJunkFilter(ctl.log)
jf, _, err := a.OpenJunkFilter(ctl.log)
if err != nil && !errors.Is(err, store.ErrNoJunkFilter) {
ctl.xcheck(err, "open junk filter")
}
@ -304,7 +248,8 @@ func importctl(ctl *ctl, mbox bool) {
ctl.xcheck(err, "close junk filter")
}
}()
}
conf, _ := a.Conf()
process := func(m *store.Message, msgf *os.File, origPath string) {
defer func() {
@ -317,12 +262,6 @@ func importctl(ctl *ctl, mbox bool) {
msgf.Close()
}()
if markRead {
m.Seen = true
}
// todo: if message does not contain a date header, but this was a maildir file, add a Date header based on the time in the filename?
// Parse message and store parsed information for later fast retrieval.
p, err := message.EnsurePart(msgf, m.Size)
if err != nil {
@ -339,6 +278,10 @@ func importctl(ctl *ctl, mbox bool) {
}
}
// We set the flags that Deliver would set now and train ourselves. This prevents
// Deliver from training, which would open the junk filter, change it, and write it
// back to disk, for each message (slow).
m.JunkFlagsForMailbox(mb.Name, conf)
if jf != nil && m.NeedsTraining() {
if words, err := jf.ParseMessage(p); err != nil {
ctl.log.Infox("parsing message for updating junk filter", err, mlog.Field("parse", ""), mlog.Field("path", origPath))
@ -389,319 +332,3 @@ func importctl(ctl *ctl, mbox bool) {
ctl.xwriteok()
ctl.xwrite(fmt.Sprintf("%d", n))
}
type msgReader interface {
// Return next message, or io.EOF when there are no more.
Next() (*store.Message, *os.File, string, error)
}
type mboxReader struct {
createTemp func(pattern string) (*os.File, error)
path string
line int
r *bufio.Reader
prevempty bool
nonfirst bool
log *mlog.Log
eof bool
junk bool
notjunk bool
}
func newMboxReader(isjunk, isnotjunk bool, createTemp func(pattern string) (*os.File, error), f *os.File, log *mlog.Log) *mboxReader {
return &mboxReader{createTemp: createTemp, path: f.Name(), line: 1, r: bufio.NewReader(f), log: log, junk: isjunk, notjunk: isnotjunk}
}
func (mr *mboxReader) position() string {
return fmt.Sprintf("%s:%d", mr.path, mr.line)
}
func (mr *mboxReader) Next() (*store.Message, *os.File, string, error) {
if mr.eof {
return nil, nil, "", io.EOF
}
from := []byte("From ")
if !mr.nonfirst {
// First read, we're at the beginning of the file.
line, err := mr.r.ReadBytes('\n')
if err == io.EOF {
return nil, nil, "", io.EOF
}
mr.line++
if !bytes.HasPrefix(line, from) {
return nil, nil, mr.position(), fmt.Errorf(`first line does not start with "From "`)
}
mr.nonfirst = true
}
f, err := mr.createTemp("mboxreader")
if err != nil {
return nil, nil, mr.position(), err
}
defer func() {
if f != nil {
f.Close()
if err := os.Remove(f.Name()); err != nil {
mr.log.Errorx("removing temporary message file after mbox read error", err, mlog.Field("path", f.Name()))
}
}
}()
bf := bufio.NewWriter(f)
var size int64
for {
line, err := mr.r.ReadBytes('\n')
if err != nil && err != io.EOF {
return nil, nil, mr.position(), fmt.Errorf("reading from mbox: %v", err)
}
if len(line) > 0 {
mr.line++
// We store data with crlf, adjust any imported messages with bare newlines.
if !bytes.HasSuffix(line, []byte("\r\n")) {
line = append(line[:len(line)-1], "\r\n"...)
}
// Next mail message starts at bare From word.
if mr.prevempty && bytes.HasPrefix(line, from) {
break
}
if bytes.HasPrefix(line, []byte(">")) && bytes.HasPrefix(bytes.TrimLeft(line, ">"), []byte("From ")) {
line = line[1:]
}
n, err := bf.Write(line)
if err != nil {
return nil, nil, mr.position(), fmt.Errorf("writing message to file: %v", err)
}
size += int64(n)
mr.prevempty = bytes.Equal(line, []byte("\r\n"))
}
if err == io.EOF {
mr.eof = true
break
}
}
if err := bf.Flush(); err != nil {
return nil, nil, mr.position(), fmt.Errorf("flush: %v", err)
}
// todo: look at Status or X-Status header in message?
// todo: take Received from the "From " line if present?
flags := store.Flags{Seen: true, Junk: mr.junk, Notjunk: mr.notjunk}
m := &store.Message{Flags: flags, Size: size}
// Prevent cleanup by defer.
mf := f
f = nil
return m, mf, mr.position(), nil
}
type maildirReader struct {
createTemp func(pattern string) (*os.File, error)
newf, curf *os.File
f *os.File // File we are currently reading from. We first read newf, then curf.
dir string // Name of directory for f. Can be empty on first call.
entries []os.DirEntry
dovecotKeywords []string
log *mlog.Log
junk bool
notjunk bool
}
func newMaildirReader(isjunk, isnotjunk bool, createTemp func(pattern string) (*os.File, error), newf, curf *os.File, log *mlog.Log) *maildirReader {
mr := &maildirReader{createTemp: createTemp, newf: newf, curf: curf, f: newf, log: log, junk: isjunk}
// Best-effort parsing of dovecot keywords.
kf, err := os.Open(filepath.Join(filepath.Dir(newf.Name()), "dovecot-keywords"))
if err == nil {
mr.dovecotKeywords = tryParseDovecotKeywords(kf, log)
kf.Close()
}
return mr
}
func (mr *maildirReader) Next() (*store.Message, *os.File, string, error) {
if mr.dir == "" {
mr.dir = mr.f.Name()
}
if len(mr.entries) == 0 {
var err error
mr.entries, err = mr.f.ReadDir(100)
if err != nil && err != io.EOF {
return nil, nil, "", err
}
if len(mr.entries) == 0 {
if mr.f == mr.curf {
return nil, nil, "", io.EOF
}
mr.f = mr.curf
mr.dir = ""
return mr.Next()
}
}
p := filepath.Join(mr.dir, mr.entries[0].Name())
mr.entries = mr.entries[1:]
sf, err := os.Open(p)
if err != nil {
return nil, nil, p, fmt.Errorf("open message in maildir: %s", err)
}
defer sf.Close()
f, err := mr.createTemp("maildirreader")
if err != nil {
return nil, nil, p, err
}
defer func() {
if f != nil {
f.Close()
if err := os.Remove(f.Name()); err != nil {
mr.log.Errorx("removing temporary message file after maildir read error", err, mlog.Field("path", f.Name()))
}
}
}()
// Copy data, changing bare \n into \r\n.
r := bufio.NewReader(sf)
w := bufio.NewWriter(f)
var size int64
for {
line, err := r.ReadBytes('\n')
if err != nil && err != io.EOF {
return nil, nil, p, fmt.Errorf("reading message: %v", err)
}
if len(line) > 0 {
if !bytes.HasSuffix(line, []byte("\r\n")) {
line = append(line[:len(line)-1], "\r\n"...)
}
if n, err := w.Write(line); err != nil {
return nil, nil, p, fmt.Errorf("writing message: %v", err)
} else {
size += int64(n)
}
}
if err == io.EOF {
break
}
}
if err := w.Flush(); err != nil {
return nil, nil, p, fmt.Errorf("writing message: %v", err)
}
// Take received time from filename.
var received time.Time
t := strings.SplitN(filepath.Base(sf.Name()), ".", 2)
if v, err := strconv.ParseInt(t[0], 10, 64); err == nil {
received = time.Unix(v, 0)
}
// Parse flags. See https://cr.yp.to/proto/maildir.html.
flags := store.Flags{}
t = strings.SplitN(filepath.Base(sf.Name()), ":2,", 2)
if len(t) == 2 {
for _, c := range t[1] {
switch c {
case 'P':
// Passed, doesn't map to a common IMAP flag.
case 'R':
flags.Answered = true
case 'S':
flags.Seen = true
case 'T':
flags.Deleted = true
case 'D':
flags.Draft = true
case 'F':
flags.Flagged = true
default:
if c >= 'a' && c <= 'z' {
index := int(c - 'a')
if index >= len(mr.dovecotKeywords) {
continue
}
kw := mr.dovecotKeywords[index]
switch kw {
case "$Forwarded", "Forwarded":
flags.Forwarded = true
case "$Junk", "Junk":
flags.Junk = true
case "$NotJunk", "NotJunk", "NonJunk":
flags.Notjunk = true
case "$MDNSent":
flags.MDNSent = true
case "$Phishing", "Phishing":
flags.Phishing = true
}
// todo: custom labels, e.g. $label1, JunkRecorded?
}
}
}
}
if mr.junk {
flags.Junk = true
}
if mr.notjunk {
flags.Notjunk = true
}
m := &store.Message{Received: received, Flags: flags, Size: size}
// Prevent cleanup by defer.
mf := f
f = nil
return m, mf, p, nil
}
func tryParseDovecotKeywords(r io.Reader, log *mlog.Log) []string {
/*
If the dovecot-keywords file is present, we parse its additional flags, see
https://doc.dovecot.org/admin_manual/mailbox_formats/maildir/
0 Old
1 Junk
2 NonJunk
3 $Forwarded
4 $Junk
*/
keywords := make([]string, 26)
end := 0
scanner := bufio.NewScanner(r)
for scanner.Scan() {
s := scanner.Text()
t := strings.SplitN(s, " ", 2)
if len(t) != 2 {
log.Info("unexpected dovecot keyword line", mlog.Field("line", s))
continue
}
v, err := strconv.ParseInt(t[0], 10, 32)
if err != nil {
log.Infox("unexpected dovecot keyword index", err, mlog.Field("line", s))
continue
}
if v < 0 || v >= int64(len(keywords)) {
log.Info("dovecot keyword index too big", mlog.Field("line", s))
continue
}
index := int(v)
if keywords[index] != "" {
log.Info("duplicate dovecot keyword", mlog.Field("line", s))
continue
}
keywords[index] = t[1]
if index >= end {
end = index + 1
}
}
if err := scanner.Err(); err != nil {
log.Infox("reading dovecot keywords file", err)
}
return keywords[:end]
}

10
junk/filter.go
View file

@ -79,6 +79,16 @@ func (f *Filter) ensureBloom() error {
return err
}
// CloseDiscard closes the filter, discarding any changes.
func (f *Filter) CloseDiscard() error {
if f.closed {
return errClosed
}
err := f.db.Close()
*f = Filter{log: f.log, closed: true}
return err
}
// Close first saves the filter if it has modifications, then closes the database
// connection and releases the bloom filter.
func (f *Filter) Close() error {

24
main_test.go
View file

@ -1,24 +0,0 @@
package main
import (
"strings"
"testing"
"github.com/mjl-/mox/mlog"
)
func TestParseDovecotKeywords(t *testing.T) {
const data = `0 Old
1 Junk
2 NonJunk
3 $Forwarded
4 $Junk
`
keywords := tryParseDovecotKeywords(strings.NewReader(data), mlog.New("dovecotkeywords"))
got := strings.Join(keywords, ",")
want := "Old,Junk,NonJunk,$Forwarded,$Junk"
if got != want {
t.Fatalf("parsing dovecot keywords, got %q, want %q", got, want)
}
}

8
mlog/log.go
View file

@ -154,6 +154,14 @@ func (l *Log) MoreFields(fn func() []Pair) *Log {
return &nl
}
// Check logs an error if err is not nil. Intended for logging errors that are good
// to know, but would not influence program flow.
func (l *Log) Check(err error, text string, fields ...Pair) {
if err != nil {
l.Errorx(text, err, fields...)
}
}
func (l *Log) Trace(traceLevel Level, text string) bool {
return l.logx(traceLevel, nil, text)
}

5
mox-/config.go
View file

@ -283,6 +283,9 @@ func writeDynamic(ctx context.Context, c config.Dynamic) error {
// MustLoadConfig loads the config, quitting on errors.
func MustLoadConfig() {
Shutdown, ShutdownCancel = context.WithCancel(context.Background())
Context, ContextCancel = context.WithCancel(context.Background())
errs := LoadConfig(context.Background())
if len(errs) > 1 {
xlog.Error("loading config file: multiple errors")
@ -396,7 +399,7 @@ func PrepareStaticConfig(ctx context.Context, configFile string, config *Config,
}
acmeDir := dataDirPath(configFile, c.DataDir, "acme")
os.MkdirAll(acmeDir, 0770)
manager, err := autotls.Load(name, acmeDir, acme.ContactEmail, acme.DirectoryURL, Shutdown)
manager, err := autotls.Load(name, acmeDir, acme.ContactEmail, acme.DirectoryURL, Shutdown.Done())
if err != nil {
addErrorf("loading ACME identity for %q: %s", name, err)
}

10
mox-/lifecycle.go
View file

@ -11,11 +11,12 @@ import (
"github.com/prometheus/client_golang/prometheus/promauto"
)
// Shutdown is closed when a graceful shutdown is initiated. SMTP, IMAP, periodic
// Shutdown is canceled when a graceful shutdown is initiated. SMTP, IMAP, periodic
// processes should check this before starting a new operation. If true, the
// operation should be aborted, and new connections should receive a message that
// the service is currently not available.
var Shutdown chan struct{}
var Shutdown context.Context
var ShutdownCancel func()
// Context should be used as parent by all operations. It is canceled when mox is
// shutdown, aborting all pending operations.
@ -25,10 +26,9 @@ var Shutdown chan struct{}
// context.WithTimeout based on this context, so those contexts are still canceled
// when shutting down.
//
// Explicit read/write deadlines on connections, typically 30s.
//
// HTTP servers don't get graceful shutdown, their connections are just aborted.
var Context context.Context
var ContextCancel func()
// Connections holds all active protocol sockets (smtp, imap). They will be given
// an immediate read/write deadline shortly after initiating mox shutdown, after
@ -61,7 +61,7 @@ func (c *connections) Register(nc net.Conn, protocol, listener string) {
// This can happen, when a connection was initiated before a shutdown, but it
// doesn't hurt to log it.
select {
case <-Shutdown:
case <-Shutdown.Done():
xlog.Error("new connection added while shutting down")
debug.PrintStack()
default:

2
mox-/lifecycle_test.go
View file

@ -1,6 +1,7 @@
package mox
import (
"context"
"errors"
"net"
"os"
@ -10,6 +11,7 @@ import (
)
func TestLifecycle(t *testing.T) {
Shutdown, ShutdownCancel = context.WithCancel(context.Background())
c := &connections{
conns: map[net.Conn]connKind{},
gauges: map[connKind]prometheus.GaugeFunc{},

2
mtastsdb/refresh.go
View file

@ -38,7 +38,7 @@ func refresh() int {
}
select {
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
return refreshed
case <-ticker.C:
}

6
mtastsdb/refresh_test.go
View file

@ -147,13 +147,13 @@ func TestRefresh(t *testing.T) {
t.Fatalf("delete record that would be refreshed: %v", err)
}
mox.Context = context.Background()
mox.Shutdown = make(chan struct{})
close(mox.Shutdown)
mox.Shutdown, mox.ShutdownCancel = context.WithCancel(context.Background())
mox.ShutdownCancel()
n := refresh()
if n != 0 {
t.Fatalf("refresh found unexpected work, n %d", n)
}
mox.Shutdown = make(chan struct{})
mox.Shutdown, mox.ShutdownCancel = context.WithCancel(context.Background())
}
type pipeListener struct {

2
queue/queue.go
View file

@ -356,7 +356,7 @@ func Start(resolver dns.Resolver, done chan struct{}) error {
for {
select {
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
done <- struct{}{}
return
case <-kick:

10
queue/queue_test.go
View file

@ -39,11 +39,11 @@ func setup(t *testing.T) (*store.Account, func()) {
err = acc.SetPassword("testtest")
tcheck(t, err, "set password")
switchDone := store.Switchboard()
mox.Shutdown = make(chan struct{})
mox.Shutdown, mox.ShutdownCancel = context.WithCancel(context.Background())
return acc, func() {
acc.Close()
close(mox.Shutdown)
mox.Shutdown = make(chan struct{})
mox.ShutdownCancel()
mox.Shutdown, mox.ShutdownCancel = context.WithCancel(context.Background())
Shutdown()
close(switchDone)
}
@ -341,9 +341,9 @@ func TestQueueStart(t *testing.T) {
defer cleanup()
done := make(chan struct{}, 1)
defer func() {
close(mox.Shutdown)
mox.ShutdownCancel()
<-done
mox.Shutdown = make(chan struct{})
mox.Shutdown, mox.ShutdownCancel = context.WithCancel(context.Background())
}()
err := Start(resolver, done)
tcheck(t, err, "queue start")

25
serve.go
View file

@ -134,10 +134,6 @@ requested, other TLS certificates are requested on demand.
}
mox.MustLoadConfig()
mox.Shutdown = make(chan struct{})
servectx, servecancel := context.WithCancel(context.Background())
mox.Context = servectx
mlog.Logfmt = true
log := mlog.New("serve")
@ -163,28 +159,33 @@ requested, other TLS certificates are requested on demand.
log.Print("starting up", mlog.Field("version", moxvar.Version))
shutdown := func() {
// We indicate we are shutting down. Causes new connections and new SMTP commands to be rejected. Should stop active connections pretty quickly.
close(mox.Shutdown)
// We indicate we are shutting down. Causes new connections and new SMTP commands
// to be rejected. Should stop active connections pretty quickly.
mox.ShutdownCancel()
// Now we are going to wait for all connections to be gone, up to a timeout.
done := mox.Connections.Done()
second := time.Tick(time.Second)
select {
case <-done:
log.Print("clean shutdown")
log.Print("connections shutdown, waiting until 1 second passed")
<-second
case <-time.Tick(3 * time.Second):
// We now cancel all pending operations, and set an immediate deadline on sockets. Should get us a clean shutdown relatively quickly.
servecancel()
// We now cancel all pending operations, and set an immediate deadline on sockets.
// Should get us a clean shutdown relatively quickly.
mox.ContextCancel()
mox.Connections.Shutdown()
second := time.Tick(time.Second)
select {
case <-done:
log.Print("no more connections, shutdown is clean")
case <-time.Tick(time.Second):
log.Print("no more connections, shutdown is clean, waiting until 1 second passed")
<-second // Still wait for second, giving processes like imports a chance to clean up.
case <-second:
log.Print("shutting down with pending sockets")
}
}
servecancel() // Keep go vet happy.
if err := os.Remove(mox.DataDirPath("ctl")); err != nil {
log.Errorx("removing ctl unix domain socket during shutdown", err)
}

4
smtpserver/server.go
View file

@ -566,7 +566,7 @@ func serve(listenerName string, cid int64, hostname dns.Domain, tlsConfig *tls.C
}()
select {
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
// ../rfc/5321:2811 ../rfc/5321:1666 ../rfc/3463:420
c.writecodeline(smtp.C421ServiceUnavail, smtp.SeSys3NotAccepting2, "shutting down", nil)
return
@ -679,7 +679,7 @@ func command(c *conn) {
// todo future: should we return an error for lines that are too long? perhaps for submission or in a pedantic mode. we would have to take extensions for MAIL into account. ../rfc/5321:3500 ../rfc/5321:3552
select {
case <-mox.Shutdown:
case <-mox.Shutdown.Done():
// ../rfc/5321:2811 ../rfc/5321:1666 ../rfc/3463:420
c.writecodeline(smtp.C421ServiceUnavail, smtp.SeSys3NotAccepting2, "shutting down", nil)
panic(errIO)

43
store/account.go
View file

@ -574,7 +574,7 @@ func (a *Account) WithRLock(fn func()) {
// Must be called with account rlock or wlock.
//
// Caller must broadcast new message.
func (a *Account) DeliverX(log *mlog.Log, tx *bstore.Tx, m *Message, msgFile *os.File, consumeFile, isSent, sync bool) {
func (a *Account) DeliverX(log *mlog.Log, tx *bstore.Tx, m *Message, msgFile *os.File, consumeFile, isSent, sync, notrain bool) {
mb := Mailbox{ID: m.MailboxID}
err := tx.Get(&mb)
xcheckf(err, "get mailbox")
@ -671,6 +671,10 @@ func (a *Account) DeliverX(log *mlog.Log, tx *bstore.Tx, m *Message, msgFile *os
xcheckf(err, "sync directory")
}
if notrain && m.NeedsTraining() {
// If this ever happens, hopefully we'll get bug reports about it.
log.Error("deliver of message that unexpectedly needs training", mlog.Field("messageid", m.ID), mlog.Field("trainedjunk", m.TrainedJunk), mlog.Field("flags", m.Flags))
}
l := []Message{*m}
err = a.RetrainMessages(log, tx, l, false)
xcheckf(err, "training junkfilter")
@ -787,14 +791,14 @@ func (a *Account) Subjectpass(email string) (key string, err error) {
// If subscribe is true, any mailboxes that were created will also be subscribed to.
// Caller must hold account wlock.
// Caller must propagate changes if any.
func (a *Account) MailboxEnsureX(tx *bstore.Tx, name string, subscribe bool) (mb Mailbox, changes []Change) {
func (a *Account) MailboxEnsure(tx *bstore.Tx, name string, subscribe bool) (mb Mailbox, changes []Change, rerr error) {
if norm.NFC.String(name) != name {
panic("mailbox name not normalized")
return Mailbox{}, nil, fmt.Errorf("mailbox name not normalized")
}
// Quick sanity check.
if strings.EqualFold(name, "inbox") && name != "Inbox" {
panic("bad casing for inbox")
return Mailbox{}, nil, fmt.Errorf("bad casing for inbox")
}
elems := strings.Split(name, "/")
@ -803,7 +807,9 @@ func (a *Account) MailboxEnsureX(tx *bstore.Tx, name string, subscribe bool) (mb
return mb.Name == elems[0] || strings.HasPrefix(mb.Name, elems[0]+"/")
})
l, err := q.List()
xcheckf(err, "list mailboxes")
if err != nil {
return Mailbox{}, nil, fmt.Errorf("list mailboxes: %v", err)
}
mailboxes := map[string]Mailbox{}
for _, xmb := range l {
@ -822,24 +828,39 @@ func (a *Account) MailboxEnsureX(tx *bstore.Tx, name string, subscribe bool) (mb
continue
}
uidval, err := a.NextUIDValidity(tx)
xcheckf(err, "next uid validity")
if err != nil {
return Mailbox{}, nil, fmt.Errorf("next uid validity: %v", err)
}
mb = Mailbox{
Name: p,
UIDValidity: uidval,
UIDNext: 1,
}
err = tx.Insert(&mb)
xcheckf(err, "creating new mailbox")
if err != nil {
return Mailbox{}, nil, fmt.Errorf("creating new mailbox: %v", err)
}
change := ChangeAddMailbox{Name: p}
if subscribe {
err := tx.Insert(&Subscription{p})
if err != nil && !errors.Is(err, bstore.ErrUnique) {
xcheckf(err, "subscribing to mailbox")
return Mailbox{}, nil, fmt.Errorf("subscribing to mailbox: %v", err)
}
change.Flags = []string{`\Subscribed`}
}
changes = append(changes, ChangeAddMailbox{Name: p, Flags: []string{`\Subscribed`}})
changes = append(changes, change)
}
return
return mb, changes, nil
}
// MailboxEnsureX calls MailboxEnsure, panicing with the error if it is not nil.
func (a *Account) MailboxEnsureX(tx *bstore.Tx, name string, subscribe bool) (Mailbox, []Change) {
mb, changes, err := a.MailboxEnsure(tx, name, subscribe)
if err != nil {
panic(err)
}
return mb, changes
}
// Check if mailbox exists.
@ -1008,7 +1029,7 @@ func (a *Account) DeliverMailbox(log *mlog.Log, mailbox string, m *Message, msgF
m.MailboxOrigID = mb.ID
changes = append(changes, chl...)
a.DeliverX(log, tx, m, msgFile, consumeFile, mb.Sent, true)
a.DeliverX(log, tx, m, msgFile, consumeFile, mb.Sent, true, false)
return nil
})
// todo: if rename succeeded but transaction failed, we should remove the file.

4
store/account_test.go
View file

@ -72,13 +72,13 @@ func TestMailbox(t *testing.T) {
tcheck(t, err, "sent mailbox")
msent.MailboxID = mbsent.ID
msent.MailboxOrigID = mbsent.ID
acc.DeliverX(xlog, tx, &msent, msgFile, false, true, true)
acc.DeliverX(xlog, tx, &msent, msgFile, false, true, true, false)
err = tx.Insert(&mbrejects)
tcheck(t, err, "insert rejects mailbox")
mreject.MailboxID = mbrejects.ID
mreject.MailboxOrigID = mbrejects.ID
acc.DeliverX(xlog, tx, &mreject, msgFile, false, false, true)
acc.DeliverX(xlog, tx, &mreject, msgFile, false, false, true, false)
return nil
})

410
store/import.go Normal file
View file

@ -0,0 +1,410 @@
package store
import (
"bufio"
"bytes"
"errors"
"fmt"
"io"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/mjl-/mox/mlog"
)
// MsgSource is implemented by readers for mailbox file formats.
type MsgSource interface {
// Return next message, or io.EOF when there are no more.
Next() (*Message, *os.File, string, error)
}
// MboxReader reads messages from an mbox file, implementing MsgSource.
type MboxReader struct {
createTemp func(pattern string) (*os.File, error)
path string
line int
r *bufio.Reader
prevempty bool
nonfirst bool
log *mlog.Log
eof bool
fromLine string // "From "-line for this message.
header bool // Now in header section.
}
func NewMboxReader(createTemp func(pattern string) (*os.File, error), filename string, r io.Reader, log *mlog.Log) *MboxReader {
return &MboxReader{
createTemp: createTemp,
path: filename,
line: 1,
r: bufio.NewReader(r),
log: log,
}
}
// Position returns "<filename>:<lineno>" for the current position.
func (mr *MboxReader) Position() string {
return fmt.Sprintf("%s:%d", mr.path, mr.line)
}
// Next returns the next message read from the mbox file. The file is a temporary
// file and must be removed/consumed. The third return value is the position in the
// file.
func (mr *MboxReader) Next() (*Message, *os.File, string, error) {
if mr.eof {
return nil, nil, "", io.EOF
}
from := []byte("From ")
if !mr.nonfirst {
mr.header = true
// First read, we're at the beginning of the file.
line, err := mr.r.ReadBytes('\n')
if err == io.EOF {
return nil, nil, "", io.EOF
}
mr.line++
if !bytes.HasPrefix(line, from) {
return nil, nil, mr.Position(), fmt.Errorf(`first line does not start with "From "`)
}
mr.nonfirst = true
mr.fromLine = strings.TrimSpace(string(line))
}
f, err := mr.createTemp("mboxreader")
if err != nil {
return nil, nil, mr.Position(), err
}
defer func() {
if f != nil {
f.Close()
if err := os.Remove(f.Name()); err != nil {
mr.log.Errorx("removing temporary message file after mbox read error", err, mlog.Field("path", f.Name()))
}
}
}()
fromLine := mr.fromLine
bf := bufio.NewWriter(f)
var flags Flags
var size int64
for {
line, err := mr.r.ReadBytes('\n')
if err != nil && err != io.EOF {
return nil, nil, mr.Position(), fmt.Errorf("reading from mbox: %v", err)
}
if len(line) > 0 {
mr.line++
// We store data with crlf, adjust any imported messages with bare newlines.
if !bytes.HasSuffix(line, []byte("\r\n")) {
line = append(line[:len(line)-1], "\r\n"...)
}
if mr.header {
// See https://doc.dovecot.org/admin_manual/mailbox_formats/mbox/
if bytes.HasPrefix(line, []byte("Status:")) {
s := strings.TrimSpace(strings.SplitN(string(line), ":", 2)[1])
for _, c := range s {
switch c {
case 'R':
flags.Seen = true
}
}
} else if bytes.HasPrefix(line, []byte("X-Status:")) {
s := strings.TrimSpace(strings.SplitN(string(line), ":", 2)[1])
for _, c := range s {
switch c {
case 'A':
flags.Answered = true
case 'F':
flags.Flagged = true
case 'T':
flags.Draft = true
case 'D':
flags.Deleted = true
}
}
} else if bytes.HasPrefix(line, []byte("X-Keywords:")) {
s := strings.TrimSpace(strings.SplitN(string(line), ":", 2)[1])
for _, t := range strings.Split(s, ",") {
flagSet(&flags, strings.ToLower(strings.TrimSpace(t)))
}
}
}
if bytes.Equal(line, []byte("\r\n")) {
mr.header = false
}
// Next mail message starts at bare From word.
if mr.prevempty && bytes.HasPrefix(line, from) {
mr.fromLine = strings.TrimSpace(string(line))
mr.header = true
break
}
if bytes.HasPrefix(line, []byte(">")) && bytes.HasPrefix(bytes.TrimLeft(line, ">"), []byte("From ")) {
line = line[1:]
}
n, err := bf.Write(line)
if err != nil {
return nil, nil, mr.Position(), fmt.Errorf("writing message to file: %v", err)
}
size += int64(n)
mr.prevempty = bytes.Equal(line, []byte("\r\n"))
}
if err == io.EOF {
mr.eof = true
break
}
}
if err := bf.Flush(); err != nil {
return nil, nil, mr.Position(), fmt.Errorf("flush: %v", err)
}
m := &Message{Flags: flags, Size: size}
if t := strings.SplitN(fromLine, " ", 3); len(t) == 3 {
layouts := []string{time.ANSIC, time.UnixDate, time.RubyDate}
for _, l := range layouts {
t, err := time.Parse(l, t[2])
if err == nil {
m.Received = t
break
}
}
}
// Prevent cleanup by defer.
mf := f
f = nil
return m, mf, mr.Position(), nil
}
type MaildirReader struct {
createTemp func(pattern string) (*os.File, error)
newf, curf *os.File
f *os.File // File we are currently reading from. We first read newf, then curf.
dir string // Name of directory for f. Can be empty on first call.
entries []os.DirEntry
dovecotKeywords []string
log *mlog.Log
}
func NewMaildirReader(createTemp func(pattern string) (*os.File, error), newf, curf *os.File, log *mlog.Log) *MaildirReader {
mr := &MaildirReader{
createTemp: createTemp,
newf: newf,
curf: curf,
f: newf,
log: log,
}
// Best-effort parsing of dovecot keywords.
kf, err := os.Open(filepath.Join(filepath.Dir(newf.Name()), "dovecot-keywords"))
if err == nil {
mr.dovecotKeywords, err = ParseDovecotKeywords(kf, log)
log.Check(err, "parsing dovecot keywords file")
kf.Close()
}
return mr
}
func (mr *MaildirReader) Next() (*Message, *os.File, string, error) {
if mr.dir == "" {
mr.dir = mr.f.Name()
}
if len(mr.entries) == 0 {
var err error
mr.entries, err = mr.f.ReadDir(100)
if err != nil && err != io.EOF {
return nil, nil, "", err
}
if len(mr.entries) == 0 {
if mr.f == mr.curf {
return nil, nil, "", io.EOF
}
mr.f = mr.curf
mr.dir = ""
return mr.Next()
}
}
p := filepath.Join(mr.dir, mr.entries[0].Name())
mr.entries = mr.entries[1:]
sf, err := os.Open(p)
if err != nil {
return nil, nil, p, fmt.Errorf("open message in maildir: %s", err)
}
defer sf.Close()
f, err := mr.createTemp("maildirreader")
if err != nil {
return nil, nil, p, err
}
defer func() {
if f != nil {
f.Close()
if err := os.Remove(f.Name()); err != nil {
mr.log.Errorx("removing temporary message file after maildir read error", err, mlog.Field("path", f.Name()))
}
}
}()
// Copy data, changing bare \n into \r\n.
r := bufio.NewReader(sf)
w := bufio.NewWriter(f)
var size int64
for {
line, err := r.ReadBytes('\n')
if err != nil && err != io.EOF {
return nil, nil, p, fmt.Errorf("reading message: %v", err)
}
if len(line) > 0 {
if !bytes.HasSuffix(line, []byte("\r\n")) {
line = append(line[:len(line)-1], "\r\n"...)
}
if n, err := w.Write(line); err != nil {
return nil, nil, p, fmt.Errorf("writing message: %v", err)
} else {
size += int64(n)
}
}
if err == io.EOF {
break
}
}
if err := w.Flush(); err != nil {
return nil, nil, p, fmt.Errorf("writing message: %v", err)
}
// Take received time from filename.
var received time.Time
t := strings.SplitN(filepath.Base(sf.Name()), ".", 2)
if v, err := strconv.ParseInt(t[0], 10, 64); err == nil {
received = time.Unix(v, 0)
}
// Parse flags. See https://cr.yp.to/proto/maildir.html.
flags := Flags{}
t = strings.SplitN(filepath.Base(sf.Name()), ":2,", 2)
if len(t) == 2 {
for _, c := range t[1] {
switch c {
case 'P':
// Passed, doesn't map to a common IMAP flag.
case 'R':
flags.Answered = true
case 'S':
flags.Seen = true
case 'T':
flags.Deleted = true
case 'D':
flags.Draft = true
case 'F':
flags.Flagged = true
default:
if c >= 'a' && c <= 'z' {
index := int(c - 'a')
if index >= len(mr.dovecotKeywords) {
continue
}
kw := mr.dovecotKeywords[index]
switch kw {
case "$Forwarded", "Forwarded":
flags.Forwarded = true
case "$Junk", "Junk":
flags.Junk = true
case "$NotJunk", "NotJunk", "NonJunk":
flags.Notjunk = true
case "$MDNSent":
flags.MDNSent = true
case "$Phishing", "Phishing":
flags.Phishing = true
}
// todo: custom labels, e.g. $label1, JunkRecorded?
}
}
}
}
m := &Message{Received: received, Flags: flags, Size: size}
// Prevent cleanup by defer.
mf := f
f = nil
return m, mf, p, nil
}
func ParseDovecotKeywords(r io.Reader, log *mlog.Log) ([]string, error) {
/*
If the dovecot-keywords file is present, we parse its additional flags, see
https://doc.dovecot.org/admin_manual/mailbox_formats/maildir/
0 Old
1 Junk
2 NonJunk
3 $Forwarded
4 $Junk
*/
keywords := make([]string, 26)
end := 0
scanner := bufio.NewScanner(r)
var errs []string
for scanner.Scan() {
s := scanner.Text()
t := strings.SplitN(s, " ", 2)
if len(t) != 2 {
errs = append(errs, fmt.Sprintf("unexpected dovecot keyword line: %q", s))
continue
}
v, err := strconv.ParseInt(t[0], 10, 32)
if err != nil {
errs = append(errs, fmt.Sprintf("unexpected dovecot keyword index: %q", s))
continue
}
if v < 0 || v >= int64(len(keywords)) {
errs = append(errs, fmt.Sprintf("dovecot keyword index too big: %q", s))
continue
}
index := int(v)
if keywords[index] != "" {
errs = append(errs, fmt.Sprintf("duplicate dovecot keyword: %q", s))
continue
}
keywords[index] = t[1]
if index >= end {
end = index + 1
}
}
if err := scanner.Err(); err != nil {
errs = append(errs, fmt.Sprintf("reading dovecot keywords file: %v", err))
}
var err error
if len(errs) > 0 {
err = errors.New(strings.Join(errs, "; "))
}
return keywords[:end], err
}
func flagSet(flags *Flags, word string) {
switch word {
case "forwarded", "$forwarded":
flags.Forwarded = true
case "junk", "$junk":
flags.Junk = true
case "notjunk", "$notjunk", "nonjunk", "$nonjunk":
flags.Notjunk = true
case "phishing", "$phishing":
flags.Phishing = true
case "mdnsent", "$mdnsent":
flags.MDNSent = true
}
}

32
import_test.go → store/import_test.go
View file

@ -1,8 +1,9 @@
package main
package store
import (
"io"
"os"
"strings"
"testing"
"github.com/mjl-/mox/mlog"
@ -12,13 +13,13 @@ func TestMboxReader(t *testing.T) {
createTemp := func(pattern string) (*os.File, error) {
return os.CreateTemp("", pattern)
}
mboxf, err := os.Open("testdata/importtest.mbox")
mboxf, err := os.Open("../testdata/importtest.mbox")
if err != nil {
t.Fatalf("open mbox: %v", err)
}
defer mboxf.Close()
mr := newMboxReader(false, false, createTemp, mboxf, mlog.New("mboxreader"))
mr := NewMboxReader(createTemp, mboxf.Name(), mboxf, mlog.New("mboxreader"))
_, mf0, _, err := mr.Next()
if err != nil {
t.Fatalf("next mbox message: %v", err)
@ -44,19 +45,19 @@ func TestMaildirReader(t *testing.T) {
return os.CreateTemp("", pattern)
}
// todo: rename 1642966915.1.mox to "1642966915.1.mox:2,"? cannot have that name in the git repo because go module (or the proxy) doesn't like it. could also add some flags and test they survive the import.
newf, err := os.Open("testdata/importtest.maildir/new")
newf, err := os.Open("../testdata/importtest.maildir/new")
if err != nil {
t.Fatalf("open maildir new: %v", err)
}
defer newf.Close()
curf, err := os.Open("testdata/importtest.maildir/cur")
curf, err := os.Open("../testdata/importtest.maildir/cur")
if err != nil {
t.Fatalf("open maildir cur: %v", err)
}
defer curf.Close()
mr := newMaildirReader(false, false, createTemp, newf, curf, mlog.New("maildirreader"))
mr := NewMaildirReader(createTemp, newf, curf, mlog.New("maildirreader"))
_, mf0, _, err := mr.Next()
if err != nil {
t.Fatalf("next maildir message: %v", err)
@ -76,3 +77,22 @@ func TestMaildirReader(t *testing.T) {
t.Fatalf("got err %v, expected eof for next maildir message", err)
}
}
func TestParseDovecotKeywords(t *testing.T) {
const data = `0 Old
1 Junk
2 NonJunk
3 $Forwarded
4 $Junk
`
keywords, err := ParseDovecotKeywords(strings.NewReader(data), mlog.New("dovecotkeywords"))
if err != nil {
t.Fatalf("parsing dovecot-keywords: %v", err)
}
got := strings.Join(keywords, ",")
want := "Old,Junk,NonJunk,$Forwarded,$Junk"
if got != want {
t.Fatalf("parsing dovecot keywords, got %q, want %q", got, want)
}
}

26
testdata/httpaccount/domains.conf vendored Normal file
View file

@ -0,0 +1,26 @@
Domains:
mox.example: nil
Accounts:
mjl:
Domain: mox.example
Destinations:
mjl:
Mailbox: Inbox
Rulesets:
-
HeadersRegexp:
subject: test
Mailbox: Test
-
HeadersRegexp:
subject: .*
Mailbox: Catchall
other:
Mailbox: Other
JunkFilter:
Threshold: 0.950000
Params:
Twograms: true
MaxPower: 0.100000
TopWords: 10
IgnoreWords: 0.100000

8
testdata/httpaccount/mox.conf vendored Normal file
View file

@ -0,0 +1,8 @@
DataDir: data
LogLevel: trace
Hostname: mox.example
Postmaster:
Account: mjl
Mailbox: postmaster
Listeners:
local: nil

BIN
testdata/importtest.maildir.tgz vendored Normal file
View file

Binary file not shown.

3
testdata/importtest.maildir/dovecot-keywords vendored Normal file
View file

@ -0,0 +1,3 @@
0 $Junk
1 $Forwarded
2 $NotJunk

4
testdata/importtest.mbox vendored
View file

@ -9,10 +9,12 @@ Date: Wed, 10 Nov 2021 23:47:13 +0100
Message-ID: <12312312-f95c-09ec-97c6-94d124f0932d@mox.test>
MIME-Version: 1.0
Status: RO
X-Status: AFTD
X-Keywords: $NotJunk,$Forwarded
Content-Length: 15
Lines: 3
test
>From testing
test2
end

BIN
testdata/importtest.mbox.zip vendored Normal file
View file

Binary file not shown.