diff --git a/cmd/archivmail/cmd_import.go b/cmd/archivmail/cmd_import.go index 34696ea..e387ef5 100644 --- a/cmd/archivmail/cmd_import.go +++ b/cmd/archivmail/cmd_import.go @@ -183,7 +183,11 @@ func runImport(args []string) { var messages [][]byte if fe.isMbox { - messages = mailparser.SplitMbox(raw) + var splitErr error + messages, splitErr = mailparser.SplitMboxErr(raw) + if splitErr != nil { + fmt.Fprintf(os.Stderr, "warning: %s: %v\n", fe.path, splitErr) + } if len(messages) == 0 { continue } diff --git a/cmd/archivmail/main.go b/cmd/archivmail/main.go index 3d96784..e119d1d 100644 --- a/cmd/archivmail/main.go +++ b/cmd/archivmail/main.go @@ -32,6 +32,7 @@ import ( "archivmail/internal/ocr" pop3store "archivmail/internal/pop3" "archivmail/internal/reconciliation" + "archivmail/internal/safego" "archivmail/internal/smtpoutconfig" "archivmail/internal/smtpd" "archivmail/internal/storage" @@ -246,7 +247,7 @@ func main() { // queries only return genuinely outstanding jobs. // PROJ-58: skipped in batch mode — the cron job drains the backlog instead. if !cfg.OCR.BatchMode { - go func() { + safego.Go(logger, "ocr boot-resume", func() { ctx := context.Background() queueCap := 1000 // matches ocr.Options.QueueSize above processed := 0 @@ -281,7 +282,7 @@ func main() { logger.Info("ocr boot-resume: enqueued batch", "batch", len(pending), "total_so_far", processed) } - }() + }) } // User store @@ -519,11 +520,15 @@ func main() { if cfg.OCR.BatchMode { backfillOCR = nil } - go runBackfill(context.Background(), mailStore, idx, tenantWorker, logger, backfillOCR) + safego.Go(logger, "backfill", func() { + runBackfill(context.Background(), mailStore, idx, tenantWorker, logger, backfillOCR) + }) } // Background integrity verification — runs every 5 minutes - go runIntegrityCheck(context.Background(), mailStore, logger) + safego.Go(logger, "integrity check", func() { + runIntegrityCheck(context.Background(), mailStore, logger) + }) // Start HTTP API go func() { @@ -538,11 +543,13 @@ func main() { // (which would otherwise drop SMTP/IMAP connections unnecessarily). reload := make(chan os.Signal, 1) signal.Notify(reload, syscall.SIGHUP) - go func() { + safego.Go(logger, "sighup reload", func() { for range reload { - reloadOCRPauseWindow(*configPath, ocrWorker, logger) + safego.Run(logger, "ocr pause-window reload", func() { + reloadOCRPauseWindow(*configPath, ocrWorker, logger) + }) } - }() + }) // Graceful shutdown quit := make(chan os.Signal, 1) diff --git a/internal/api/imap_handlers.go b/internal/api/imap_handlers.go index 76dfcfc..27a39f1 100644 --- a/internal/api/imap_handlers.go +++ b/internal/api/imap_handlers.go @@ -9,6 +9,7 @@ import ( "archivmail/internal/auth" imapstore "archivmail/internal/imap" + "archivmail/internal/safego" "archivmail/internal/userstore" ) @@ -211,7 +212,9 @@ func (s *Server) handleStartImport(w http.ResponseWriter, r *http.Request) { return } - go s.imapImporter.Run(context.Background(), id) + safego.Go(s.logger, "imap import", func() { + s.imapImporter.Run(context.Background(), id) + }) // Return current account state (status will switch to "running" shortly) acc.Status = "running" diff --git a/internal/api/invite_handlers.go b/internal/api/invite_handlers.go index e3b9a1d..41b6b18 100644 --- a/internal/api/invite_handlers.go +++ b/internal/api/invite_handlers.go @@ -8,6 +8,7 @@ import ( "archivmail/internal/audit" "archivmail/internal/mailer" + "archivmail/internal/safego" "archivmail/internal/tokenstore" "archivmail/internal/userstore" ) @@ -55,11 +56,12 @@ func (s *Server) handleCreateInvite(w http.ResponseWriter, r *http.Request) { tenantName = t.Name } } - go func() { + email := body.Email + safego.Go(s.logger, "send invite mail", func() { html := mailer.InviteHTML(s.fqdn, token, tenantName) txt := mailer.InviteText(s.fqdn, token, tenantName) - _ = s.mailer.Send(body.Email, "Einladung zu archivmail", html, txt) - }() + _ = s.mailer.Send(email, "Einladung zu archivmail", html, txt) + }) } if s.audlog != nil { diff --git a/internal/api/onboarding_handlers.go b/internal/api/onboarding_handlers.go index eb322a7..ef5d8d3 100644 --- a/internal/api/onboarding_handlers.go +++ b/internal/api/onboarding_handlers.go @@ -7,6 +7,7 @@ import ( "archivmail/internal/audit" "archivmail/internal/mailer" + "archivmail/internal/safego" "archivmail/internal/tokenstore" "archivmail/internal/userstore" ) @@ -79,11 +80,12 @@ func (s *Server) handleSignup(w http.ResponseWriter, r *http.Request) { if err != nil { // SEC: token already consumed — response is identical to success to prevent // enumeration of whether the email address already existed. - go func() { + email := body.Email + safego.Go(s.logger, "send already-registered mail", func() { html := mailer.AlreadyRegisteredHTML(s.fqdn) txt := mailer.AlreadyRegisteredText(s.fqdn) - _ = s.mailer.Send(body.Email, "archivmail – Registrierungsversuch", html, txt) - }() + _ = s.mailer.Send(email, "archivmail – Registrierungsversuch", html, txt) + }) writeJSON(w, http.StatusOK, map[string]string{"message": signupMsg}) return } @@ -100,11 +102,11 @@ func (s *Server) handleSignup(w http.ResponseWriter, r *http.Request) { } // Send verification email. - go func() { + safego.Go(s.logger, "send verify mail", func() { html := mailer.VerifyEmailHTML(s.fqdn, token, u.Username) txt := mailer.VerifyEmailText(s.fqdn, token, u.Username) _ = s.mailer.Send(u.Email, "archivmail – E-Mail bestätigen", html, txt) - }() + }) if s.audlog != nil { s.audlog.Log(audit.Entry{ @@ -193,11 +195,11 @@ func (s *Server) handleForgotPassword(w http.ResponseWriter, r *http.Request) { return } - go func() { + safego.Go(s.logger, "send password-reset mail", func() { html := mailer.ResetPasswordHTML(s.fqdn, token, u.Username) txt := mailer.ResetPasswordText(s.fqdn, token, u.Username) _ = s.mailer.Send(u.Email, "archivmail – Passwort zurücksetzen", html, txt) - }() + }) if s.audlog != nil { s.audlog.Log(audit.Entry{ diff --git a/internal/api/pop3_handlers.go b/internal/api/pop3_handlers.go index 654df68..375f722 100644 --- a/internal/api/pop3_handlers.go +++ b/internal/api/pop3_handlers.go @@ -9,6 +9,7 @@ import ( "archivmail/internal/auth" pop3store "archivmail/internal/pop3" + "archivmail/internal/safego" "archivmail/internal/userstore" ) @@ -212,7 +213,9 @@ func (s *Server) handleStartPop3Import(w http.ResponseWriter, r *http.Request) { return } - go s.pop3Importer.Run(context.Background(), id) + safego.Go(s.logger, "pop3 import", func() { + s.pop3Importer.Run(context.Background(), id) + }) // Return current account state (status will switch to "running" shortly) acc.Status = "running" diff --git a/internal/api/upload.go b/internal/api/upload.go index af3d4c4..19c958d 100644 --- a/internal/api/upload.go +++ b/internal/api/upload.go @@ -9,6 +9,7 @@ import ( "sync" "archivmail/internal/index" + "archivmail/internal/safego" "archivmail/pkg/mailparser" ) @@ -94,7 +95,12 @@ func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) { var allMessages [][]byte for _, e := range entries { if e.isMbox { - msgs := mailparser.SplitMbox(e.data) + msgs, err := mailparser.SplitMboxErr(e.data) + if err != nil { + // Partial result: import what was parsed, but make the + // truncation visible instead of losing mails silently. + s.logger.Warn("upload: mbox scan incomplete", "messages", len(msgs), "err", err) + } allMessages = append(allMessages, msgs...) } else { allMessages = append(allMessages, e.data) @@ -113,7 +119,9 @@ func (s *Server) handleUpload(w http.ResponseWriter, r *http.Request) { tenantID := tenantFromCtx(r.Context()) // Run import in background - go s.runUploadJob(job, allMessages, tenantID) + safego.Go(s.logger, "upload import job", func() { + s.runUploadJob(job, allMessages, tenantID) + }) writeJSON(w, http.StatusAccepted, map[string]string{"job_id": jobID}) } @@ -126,7 +134,11 @@ func (s *Server) handleUploadProgress(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusNotFound, "job not found") return } - job := val.(*UploadJob) + job, ok := val.(*UploadJob) + if !ok || job == nil { + writeError(w, http.StatusNotFound, "job not found") + return + } writeJSON(w, http.StatusOK, job.snapshot()) } diff --git a/internal/auth/apikey_middleware.go b/internal/auth/apikey_middleware.go index 168b4be..522b438 100644 --- a/internal/auth/apikey_middleware.go +++ b/internal/auth/apikey_middleware.go @@ -52,6 +52,7 @@ type APIKeyRow struct { // tokenBucket implements a simple per-key token-bucket rate limiter. type tokenBucket struct { + mu sync.Mutex // guards tokens/lastCheck — concurrent requests share one bucket tokens float64 limit float64 lastCheck time.Time @@ -128,7 +129,14 @@ func (m *APIKeyMiddleware) allow(keyID int64, limitPerMin int) bool { limit: limit, lastCheck: now, }) - bucket := val.(*tokenBucket) + bucket, ok := val.(*tokenBucket) + if !ok || bucket == nil { + // Should never happen — fail closed rather than panic. + return false + } + + bucket.mu.Lock() + defer bucket.mu.Unlock() elapsed := now.Sub(bucket.lastCheck).Seconds() bucket.lastCheck = now diff --git a/internal/imap/scheduler.go b/internal/imap/scheduler.go index 1caac55..a406303 100644 --- a/internal/imap/scheduler.go +++ b/internal/imap/scheduler.go @@ -10,6 +10,7 @@ import ( "time" "archivmail/internal/audit" + "archivmail/internal/safego" imapv2 "github.com/emersion/go-imap/v2" "github.com/emersion/go-imap/v2/imapclient" @@ -107,7 +108,9 @@ func (s *Scheduler) TriggerSync(ctx context.Context, accountID int64) error { s.running[accountID] = true s.mu.Unlock() - go s.runSyncWithRetry(context.Background(), accountID) + safego.Go(s.logger, "imap manual sync", func() { + s.runSyncWithRetry(context.Background(), accountID) + }) return nil } @@ -121,7 +124,10 @@ func (s *Scheduler) loop(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - s.checkAccounts(ctx) + // A panic in one tick must not stop the scheduler permanently. + safego.Run(s.logger, "imap scheduler tick", func() { + s.checkAccounts(ctx) + }) } } } @@ -166,7 +172,10 @@ func (s *Scheduler) checkAccounts(ctx context.Context) { s.logger.Info("imap scheduler: starting scheduled sync", "account_id", acc.ID, "name", acc.Name) - go s.runSyncWithRetry(context.Background(), acc.ID) + accID := acc.ID + safego.Go(s.logger, "imap scheduled sync", func() { + s.runSyncWithRetry(context.Background(), accID) + }) } } } diff --git a/internal/imapserver/server.go b/internal/imapserver/server.go index c497304..efc14c1 100644 --- a/internal/imapserver/server.go +++ b/internal/imapserver/server.go @@ -10,6 +10,7 @@ import ( "bufio" "context" "crypto/tls" + "errors" "fmt" "log/slog" "net" @@ -31,6 +32,9 @@ const ( maxConnsPerUser = 5 readBufferSize = 8192 maxLineLength = 65536 + // maxFetchResults caps how many messages a single sequence set may expand + // to, bounding per-request memory for "FETCH 1:*" on huge mailboxes. + maxFetchResults = 50000 ) // tenantIMAPModeGetter is satisfied by tenantstore.Store. @@ -199,6 +203,7 @@ func (s *Server) handleConnection(conn net.Conn) { defer conn.Close() remoteAddr := conn.RemoteAddr().String() + s.logger.Debug("imapserver: new connection", "remote", remoteAddr) sess := &session{ @@ -209,6 +214,19 @@ func (s *Server) handleConnection(conn net.Conn) { state: stateNotAuth, } + // A panic while parsing a malformed/hostile client command must only kill + // this connection, never the whole archivmail process. The per-user + // connection counter is released here as well so it cannot leak. + defer func() { + if r := recover(); r != nil { + if sess.username != "" { + s.releaseConn(sess.username) + } + s.logger.Error("imapserver: recovered from panic in connection handler", + "remote", remoteAddr, "panic", fmt.Sprintf("%v", r)) + } + }() + // Send greeting — use FQDN if configured (RFC 3501 §7.1) fqdn := s.cfg.FQDN if fqdn == "" { @@ -282,12 +300,26 @@ type mailEntry struct { HasAttach bool } +// readLine reads a single CRLF-terminated command line. The length is capped at +// maxLineLength: a client that never sends a newline would otherwise make the +// buffer grow without bound until the process is killed by the OOM killer. func (sess *session) readLine() (string, error) { - line, err := sess.reader.ReadString('\n') - if err != nil { + var buf []byte + for { + chunk, err := sess.reader.ReadSlice('\n') + if len(buf)+len(chunk) > maxLineLength { + return "", fmt.Errorf("imapserver: command line exceeds %d bytes", maxLineLength) + } + buf = append(buf, chunk...) + if err == nil { + break + } + if errors.Is(err, bufio.ErrBufferFull) { + continue + } return "", err } - return strings.TrimRight(line, "\r\n"), nil + return strings.TrimRight(string(buf), "\r\n"), nil } func (sess *session) writeResponse(line string) { @@ -949,26 +981,29 @@ func parseFetchItems(items string) []string { items = strings.TrimSuffix(items, ")") var result []string - current := "" + // strings.Builder instead of `current += string(ch)`: the latter reallocates + // and copies on every rune, i.e. O(n²) for a command line that may be up to + // maxLineLength bytes long. + var current strings.Builder bracketDepth := 0 for _, ch := range items { if ch == '[' { bracketDepth++ - current += string(ch) + current.WriteRune(ch) } else if ch == ']' { bracketDepth-- - current += string(ch) + current.WriteRune(ch) } else if ch == ' ' && bracketDepth == 0 { - if current != "" { - result = append(result, current) - current = "" + if current.Len() > 0 { + result = append(result, current.String()) + current.Reset() } } else { - current += string(ch) + current.WriteRune(ch) } } - if current != "" { - result = append(result, current) + if current.Len() > 0 { + result = append(result, current.String()) } return result @@ -976,15 +1011,22 @@ func parseFetchItems(items string) []string { // parseSequenceSet parses an IMAP sequence set (e.g. "1:*", "1,3:5", "1") // and returns the expanded list of numbers. +// +// The expansion is capped at maxFetchResults: "FETCH 1:*" on a mailbox with +// millions of mails would otherwise allocate a huge slice plus dedup map per +// request — several concurrent clients could exhaust the machine's memory. func parseSequenceSet(set string, maxVal uint32) []uint32 { if maxVal == 0 { return nil } - var result []uint32 + result := make([]uint32, 0, 64) seen := make(map[uint32]bool) for _, part := range strings.Split(set, ",") { + if len(result) >= maxFetchResults { + break + } part = strings.TrimSpace(part) if part == "" { continue @@ -998,6 +1040,9 @@ func parseSequenceSet(set string, maxVal uint32) []uint32 { start, end = end, start } for i := start; i <= end; i++ { + if len(result) >= maxFetchResults { + break + } if !seen[i] { result = append(result, i) seen[i] = true @@ -1151,23 +1196,24 @@ func stripQuotes(s string) string { func splitIMAPArgs(s string) []string { var parts []string - current := "" + // See parseFetchItems: strings.Builder avoids O(n²) rune-wise concatenation. + var current strings.Builder inQuote := false for _, ch := range s { if ch == '"' { inQuote = !inQuote - current += string(ch) + current.WriteRune(ch) } else if ch == ' ' && !inQuote { - if current != "" { - parts = append(parts, current) - current = "" + if current.Len() > 0 { + parts = append(parts, current.String()) + current.Reset() } } else { - current += string(ch) + current.WriteRune(ch) } } - if current != "" { - parts = append(parts, current) + if current.Len() > 0 { + parts = append(parts, current.String()) } return parts } diff --git a/internal/index/tenant_worker.go b/internal/index/tenant_worker.go index 8705d3e..f9a5047 100644 --- a/internal/index/tenant_worker.go +++ b/internal/index/tenant_worker.go @@ -1,6 +1,7 @@ package index import ( + "fmt" "log/slog" "sync" ) @@ -83,7 +84,20 @@ func (w *TenantIndexWorker) QueueLen() int { } func (w *TenantIndexWorker) indexDoc(doc MailDocument) { + // A panic while indexing a single document must not kill the long-running + // worker goroutine (and with it the whole process). + defer func() { + if r := recover(); r != nil { + w.logger.Error("tenant index worker: recovered from panic", + "id", doc.ID, "tenant_id", doc.TenantID, "panic", fmt.Sprintf("%v", r)) + } + }() + idx := w.mgr.ForTenant(doc.TenantID) + if idx == nil { + w.logger.Error("tenant index worker: no indexer for tenant", "id", doc.ID, "tenant_id", doc.TenantID) + return + } if err := idx.IndexSync(doc); err != nil { w.logger.Error("tenant index worker: index failed", "id", doc.ID, "tenant_id", doc.TenantID, "err", err) } diff --git a/internal/index/worker.go b/internal/index/worker.go index cc1efbb..1eede8a 100644 --- a/internal/index/worker.go +++ b/internal/index/worker.go @@ -1,6 +1,7 @@ package index import ( + "fmt" "log/slog" "sync" ) @@ -53,9 +54,7 @@ func (w *IndexWorker) Start() { // Channel closed, drain complete return } - if err := w.idx.IndexSync(doc); err != nil { - w.logger.Error("index worker: index failed", "id", doc.ID, "err", err) - } + w.indexDoc(doc, "") case <-w.done: // Drain remaining items in the queue before exiting for { @@ -64,9 +63,7 @@ func (w *IndexWorker) Start() { if !ok { return } - if err := w.idx.IndexSync(doc); err != nil { - w.logger.Error("index worker: index failed (drain)", "id", doc.ID, "err", err) - } + w.indexDoc(doc, " (drain)") default: return } @@ -76,6 +73,21 @@ func (w *IndexWorker) Start() { }() } +// indexDoc indexes a single document. A panic in the backend must not kill the +// long-running worker goroutine (and with it the whole process). +func (w *IndexWorker) indexDoc(doc MailDocument, phase string) { + defer func() { + if r := recover(); r != nil { + w.logger.Error("index worker: recovered from panic"+phase, + "id", doc.ID, "panic", fmt.Sprintf("%v", r)) + } + }() + + if err := w.idx.IndexSync(doc); err != nil { + w.logger.Error("index worker: index failed"+phase, "id", doc.ID, "err", err) + } +} + // Stop signals the worker to drain remaining items and stop. It blocks until // the worker goroutine has exited. func (w *IndexWorker) Stop() { diff --git a/internal/ocr/worker.go b/internal/ocr/worker.go index e1daf86..1609767 100644 --- a/internal/ocr/worker.go +++ b/internal/ocr/worker.go @@ -3,6 +3,7 @@ package ocr import ( "context" "errors" + "fmt" "log/slog" "strings" "sync" @@ -199,6 +200,19 @@ func (w *Worker) run(ctx context.Context, id int) { } func (w *Worker) process(ctx context.Context, job Job) { + // A panic while processing a single mail (malformed attachment, external + // tool output, index backend) must not kill the worker goroutine — the + // whole process would go down with it. + defer func() { + if r := recover(); r != nil { + w.logger.Error("ocr worker: recovered from panic", + "mail_id", job.MailID, "panic", fmt.Sprintf("%v", r)) + if w.store != nil { + _ = w.store.SetOCRResult(ctx, job.MailID, "failed", 0) + } + } + }() + // PROJ-44: The canonical source of truth for a mail's tenant assignment // is emails.tenant_id in PostgreSQL — never the submitter's context. // Re-imports via IMAP/POP3 scheduler may submit the same mail with a diff --git a/internal/safego/safego.go b/internal/safego/safego.go new file mode 100644 index 0000000..55effa0 --- /dev/null +++ b/internal/safego/safego.go @@ -0,0 +1,32 @@ +// Package safego starts background goroutines that cannot take the process +// down. A panic inside a detached goroutine is fatal for the whole binary in +// Go — there is no per-goroutine boundary like net/http provides for request +// handlers. Every "fire and forget" goroutine in archivmail (mail sending, +// importer runs, boot-resume jobs) therefore goes through Go/Run. +package safego + +import ( + "fmt" + "log/slog" +) + +// Run executes fn synchronously and recovers from a panic, logging it with the +// given name. It returns true when fn completed without panicking. +func Run(logger *slog.Logger, name string, fn func()) (ok bool) { + defer func() { + if r := recover(); r != nil { + ok = false + if logger != nil { + logger.Error("recovered from panic in background task", + "task", name, "panic", fmt.Sprintf("%v", r)) + } + } + }() + fn() + return true +} + +// Go runs fn in a new goroutine, recovering and logging any panic. +func Go(logger *slog.Logger, name string, fn func()) { + go Run(logger, name, fn) //nolint:errcheck // result is irrelevant for detached tasks +} diff --git a/internal/smtpd/smtpd.go b/internal/smtpd/smtpd.go index 9c438d8..7402230 100644 --- a/internal/smtpd/smtpd.go +++ b/internal/smtpd/smtpd.go @@ -365,7 +365,23 @@ func (s *session) Rcpt(to string, _ *smtp.RcptOptions) error { return nil } -func (s *session) Data(r io.Reader) error { +func (s *session) Data(r io.Reader) (err error) { + // go-smtp does not guard session handlers: a panic while processing a + // malformed mail would take the whole archivmail process down. Convert it + // into a temporary failure so the sending MTA retries. + defer func() { + if rec := recover(); rec != nil { + s.daemon.stats.Rejected.Add(1) + s.daemon.logger.Error("SMTP: recovered from panic while processing message", + "from", s.from, "ip", s.remoteIP, "panic", fmt.Sprintf("%v", rec)) + err = &smtp.SMTPError{ + Code: 451, + EnhancedCode: smtp.EnhancedCode{4, 3, 0}, + Message: "Temporary processing failure, please retry", + } + } + }() + var buf bytes.Buffer if _, err := io.Copy(&buf, r); err != nil { s.daemon.stats.Rejected.Add(1) diff --git a/pkg/mailparser/mbox.go b/pkg/mailparser/mbox.go index 0fb4f72..d01c334 100644 --- a/pkg/mailparser/mbox.go +++ b/pkg/mailparser/mbox.go @@ -3,12 +3,30 @@ package mailparser import ( "bufio" "bytes" - "strings" + "fmt" +) + +var ( + mboxSeparator = []byte("From ") + mboxHeaderTag = []byte("From: ") + mboxQuoted = []byte(">From ") ) // SplitMbox splits a raw mbox file into individual RFC 2822 message bytes. // Each message starts with a "From " separator line which is skipped. +// +// Scan errors (e.g. a single line larger than the 10 MB scanner limit) are +// ignored here for backwards compatibility — use SplitMboxErr to detect a +// truncated result instead of silently importing partial data. func SplitMbox(data []byte) [][]byte { + messages, _ := SplitMboxErr(data) + return messages +} + +// SplitMboxErr behaves like SplitMbox but also reports scanner failures. +// On error the already-collected messages are still returned so the caller can +// decide between aborting and importing a partial file. +func SplitMboxErr(data []byte) ([][]byte, error) { var messages [][]byte var current bytes.Buffer @@ -17,11 +35,16 @@ func SplitMbox(data []byte) [][]byte { inMessage := false for scanner.Scan() { - line := scanner.Text() + // Bytes() avoids a string allocation per line — mbox files routinely + // have millions of lines. + line := scanner.Bytes() // mbox separator: line starts with "From " but not "From:" header - if strings.HasPrefix(line, "From ") && !strings.HasPrefix(line, "From: ") { + if bytes.HasPrefix(line, mboxSeparator) && !bytes.HasPrefix(line, mboxHeaderTag) { if inMessage && current.Len() > 0 { - messages = append(messages, bytes.TrimSpace(current.Bytes())) + // Copy: current.Bytes() aliases the buffer, which is reused + // after Reset() and would corrupt already-collected messages. + msg := bytes.TrimSpace(current.Bytes()) + messages = append(messages, append([]byte(nil), msg...)) current.Reset() } inMessage = true @@ -29,15 +52,20 @@ func SplitMbox(data []byte) [][]byte { } if inMessage { // unescape ">From " lines (mbox quoting) - if strings.HasPrefix(line, ">From ") { + if bytes.HasPrefix(line, mboxQuoted) { line = line[1:] } - current.WriteString(line) + current.Write(line) current.WriteByte('\n') } } if inMessage && current.Len() > 0 { - messages = append(messages, bytes.TrimSpace(current.Bytes())) + msg := bytes.TrimSpace(current.Bytes()) + messages = append(messages, append([]byte(nil), msg...)) } - return messages + + if err := scanner.Err(); err != nil { + return messages, fmt.Errorf("mailparser: mbox scan aborted after %d messages: %w", len(messages), err) + } + return messages, nil } diff --git a/pkg/mailparser/parser.go b/pkg/mailparser/parser.go index 32141df..f40e84a 100644 --- a/pkg/mailparser/parser.go +++ b/pkg/mailparser/parser.go @@ -37,10 +37,33 @@ type ParsedMail struct { Date time.Time Attachments []Attachment Raw []byte + + // Accumulators used while walking MIME parts. Appending to the TextBody / + // HTMLBody strings directly is O(n²) for mails with many parts, so parts + // are collected here and joined once at the end of Parse. + textBuf strings.Builder + htmlBuf strings.Builder } +// maxMultipartDepth caps MIME nesting. Malformed or malicious mails can nest +// multipart parts arbitrarily deep; without a limit the recursion in +// parseMultipart can exhaust the goroutine stack, which is an unrecoverable +// fatal error for the whole process. +const maxMultipartDepth = 20 + // Parse parses a raw RFC 2822 / MIME email and returns a ParsedMail. -func Parse(raw []byte) (*ParsedMail, error) { +// +// Input is untrusted (IMAP/POP3/SMTP/upload). A panic inside the MIME/charset +// decoding stack must never take down the importer or the whole daemon, so it +// is converted into a normal parse error here. +func Parse(raw []byte) (pmOut *ParsedMail, errOut error) { + defer func() { + if r := recover(); r != nil { + pmOut = nil + errOut = fmt.Errorf("mailparser: panic while parsing message: %v", r) + } + }() + msg, err := mail.ReadMessage(bytes.NewReader(raw)) if err != nil { return nil, fmt.Errorf("mailparser: read message: %w", err) @@ -182,7 +205,9 @@ func Parse(raw []byte) (*ParsedMail, error) { boundary := params["boundary"] // Ignore parse errors — return partial content instead of failing completely. // Malformed/truncated multipart emails still get metadata + whatever parts parsed. - parseMultipart(pm, msg.Body, boundary) //nolint:errcheck + parseMultipart(pm, msg.Body, boundary, 0) //nolint:errcheck + pm.TextBody = pm.textBuf.String() + pm.HTMLBody = pm.htmlBuf.String() } else { body, _ := io.ReadAll(msg.Body) decoded := decodeBody(body, msg.Header.Get("Content-Transfer-Encoding")) @@ -199,7 +224,11 @@ func Parse(raw []byte) (*ParsedMail, error) { // parseMultipart walks MIME parts and fills text, html, and attachments. // Truncated or malformed parts are skipped — partial content is better than nothing. -func parseMultipart(pm *ParsedMail, body io.Reader, boundary string) { +// depth guards against unbounded MIME nesting (stack overflow). +func parseMultipart(pm *ParsedMail, body io.Reader, boundary string, depth int) { + if depth >= maxMultipartDepth || boundary == "" { + return + } mr := multipart.NewReader(body, boundary) for { part, err := mr.NextPart() @@ -243,15 +272,15 @@ func parseMultipart(pm *ParsedMail, body io.Reader, boundary string) { // Nested multipart if strings.HasPrefix(mediaType, "multipart/") { - parseMultipart(pm, bytes.NewReader(decoded), params["boundary"]) + parseMultipart(pm, bytes.NewReader(decoded), params["boundary"], depth+1) continue } switch { case strings.Contains(mediaType, "text/plain"): - pm.TextBody += string(decoded) + pm.textBuf.Write(decoded) case strings.Contains(mediaType, "text/html"): - pm.HTMLBody += string(decoded) + pm.htmlBuf.Write(decoded) } } } diff --git a/src/app/admin/upload/page.tsx b/src/app/admin/upload/page.tsx index abd062d..81dd816 100644 --- a/src/app/admin/upload/page.tsx +++ b/src/app/admin/upload/page.tsx @@ -199,7 +199,9 @@ export default function UploadPage() {