From 428040d964b6ffa3a88c47e4307d4baafab9ece7 Mon Sep 17 00:00:00 2001 From: karti-ai Date: Sun, 21 Jun 2026 13:22:35 -0700 Subject: [PATCH] Fix 13 findings from adversarial milestone-1 review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Blockers in the ingest path (all funnel real mail in milestone 2): - H1: coerce body bytes to valid UTF-8 (ToValidUTF8) — non-UTF-8/mid-rune cuts no longer abort the ingest tx and drop the message. - H2: EnsurePart's recoverable parse error is non-fatal — proceed with the guaranteed-usable Part so messy real-world mail is stored, not rejected. - H3: extractBodies descends into message/rfc822 (Part.Message via SetMessageReaderAt) — forwarded/bounce bodies no longer lost. - H4: GetMessage/GetThread/GetThreadMessages scoped to inbox_id — no cross-inbox access; reply no longer a confused deputy. Hardening: - M1: /healthz no longer leaks DB error to unauthenticated callers. - M2: all DB errors funnel through handleErr; malformed UUID -> 404, dup -> 409, internal errors no longer echo the driver string. - M3: index messages(inbox_id, message_id_hdr) for thread resolution. - M4: pods UNIQUE(name) + ON CONFLICT (name) — no duplicate default pods. - L1: skip empty-User/Host addresses (no literal "@"). - L2: skip attachment-disposition parts when picking the body. - L3: case-insensitive, trimmed 'Re:' detection. Verified e2e vs Postgres 16: latin1 body stored valid UTF-8; rfc822-only body extracted; cross-inbox 404; malformed UUID 404; dup 409; threading regression OK. Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/api/handlers.go | 48 +++++++++-------- internal/api/server.go | 3 +- internal/core/core.go | 2 +- internal/core/message.go | 68 ++++++++++++++++++++----- internal/core/thread.go | 7 ++- internal/store/migrations/0001_init.sql | 5 +- 6 files changed, 93 insertions(+), 40 deletions(-) diff --git a/internal/api/handlers.go b/internal/api/handlers.go index e7012f5..6e8c69a 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -6,8 +6,10 @@ import ( "io" "net/http" "strconv" + "strings" "github.com/go-chi/chi/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/karti-ai/openmail/internal/core" "github.com/karti-ai/openmail/internal/mail" @@ -28,8 +30,7 @@ func (s *Server) createInbox(w http.ResponseWriter, r *http.Request) { return } ib, err := s.core.CreateInbox(r.Context(), s.podID, body.Address, body.DisplayName) - if err != nil { - writeErr(w, http.StatusInternalServerError, "create_failed", err.Error()) + if handleErr(w, err) { return } writeJSON(w, http.StatusCreated, ib) @@ -40,8 +41,7 @@ func (s *Server) listInboxes(w http.ResponseWriter, r *http.Request) { return } list, err := s.core.ListInboxes(r.Context(), s.podID) - if err != nil { - writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + if handleErr(w, err) { return } writeJSON(w, http.StatusOK, map[string]any{"inboxes": list}) @@ -74,8 +74,7 @@ func (s *Server) ingest(w http.ResponseWriter, r *http.Request) { return } msg, err := s.core.IngestRaw(r.Context(), inboxID, raw) - if err != nil { - writeErr(w, http.StatusInternalServerError, "ingest_failed", err.Error()) + if handleErr(w, err) { return } writeJSON(w, http.StatusCreated, msg) @@ -87,8 +86,7 @@ func (s *Server) listMessages(w http.ResponseWriter, r *http.Request) { } limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) list, err := s.core.ListMessages(r.Context(), chi.URLParam(r, "id"), limit) - if err != nil { - writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + if handleErr(w, err) { return } writeJSON(w, http.StatusOK, map[string]any{"messages": list}) @@ -98,7 +96,7 @@ func (s *Server) getMessage(w http.ResponseWriter, r *http.Request) { if !s.requireCore(w) { return } - m, err := s.core.GetMessage(r.Context(), chi.URLParam(r, "msgID")) + m, err := s.core.GetMessage(r.Context(), chi.URLParam(r, "id"), chi.URLParam(r, "msgID")) if handleErr(w, err) { return } @@ -110,8 +108,7 @@ func (s *Server) listThreads(w http.ResponseWriter, r *http.Request) { return } list, err := s.core.ListThreads(r.Context(), chi.URLParam(r, "id")) - if err != nil { - writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + if handleErr(w, err) { return } writeJSON(w, http.StatusOK, map[string]any{"threads": list}) @@ -121,14 +118,14 @@ func (s *Server) getThread(w http.ResponseWriter, r *http.Request) { if !s.requireCore(w) { return } + inboxID := chi.URLParam(r, "id") threadID := chi.URLParam(r, "threadID") - t, err := s.core.GetThread(r.Context(), threadID) + t, err := s.core.GetThread(r.Context(), inboxID, threadID) if handleErr(w, err) { return } - msgs, err := s.core.GetThreadMessages(r.Context(), threadID) - if err != nil { - writeErr(w, http.StatusInternalServerError, "list_failed", err.Error()) + msgs, err := s.core.GetThreadMessages(r.Context(), inboxID, threadID) + if handleErr(w, err) { return } writeJSON(w, http.StatusOK, map[string]any{"thread": t, "messages": msgs}) @@ -171,7 +168,7 @@ func (s *Server) replyMessage(w http.ResponseWriter, r *http.Request) { if handleErr(w, err) { return } - orig, err := s.core.GetMessage(r.Context(), chi.URLParam(r, "msgID")) + orig, err := s.core.GetMessage(r.Context(), inboxID, chi.URLParam(r, "msgID")) if handleErr(w, err) { return } @@ -215,25 +212,34 @@ func (s *Server) dispatch(w http.ResponseWriter, r *http.Request, out *mail.Outg } func handleErr(w http.ResponseWriter, err error) bool { + var pgErr *pgconn.PgError switch { case err == nil: return false case errors.Is(err, core.ErrNotFound): writeErr(w, http.StatusNotFound, "not_found", "resource not found") + case errors.As(err, &pgErr) && pgErr.Code == "22P02": + // invalid_text_representation, e.g. a malformed UUID in the path — treat + // as not found rather than a 500 that echoes the driver error. + writeErr(w, http.StatusNotFound, "not_found", "resource not found") + case errors.As(err, &pgErr) && pgErr.Code == "23505": + // unique_violation, e.g. an inbox address that already exists. + writeErr(w, http.StatusConflict, "conflict", "resource already exists") default: - writeErr(w, http.StatusInternalServerError, "internal_error", err.Error()) + writeErr(w, http.StatusInternalServerError, "internal_error", "internal error") } return true } func replySubject(s *string) string { - if s == nil || *s == "" { + if s == nil || strings.TrimSpace(*s) == "" { return "Re:" } - if len(*s) >= 3 && (*s)[:3] == "Re:" { - return *s + trimmed := strings.TrimSpace(*s) + if strings.HasPrefix(strings.ToLower(trimmed), "re:") { + return trimmed } - return "Re: " + *s + return "Re: " + trimmed } func strPtr(s *string) []string { diff --git a/internal/api/server.go b/internal/api/server.go index 6904310..8e89cf5 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -57,8 +57,9 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { status := map[string]any{"status": "ok", "backend": backendName(s.backend)} if s.core != nil { if err := s.core.Ping(r.Context()); err != nil { + // /healthz is unauthenticated — don't leak DSN/host/internal details. status["status"] = "degraded" - status["db"] = err.Error() + status["db"] = "error" writeJSON(w, http.StatusServiceUnavailable, status) return } diff --git a/internal/core/core.go b/internal/core/core.go index ae015ae..3cbc2fb 100644 --- a/internal/core/core.go +++ b/internal/core/core.go @@ -40,7 +40,7 @@ func (s *Service) EnsureDefaultPod(ctx context.Context) (string, error) { } err = s.pool.QueryRow(ctx, `INSERT INTO pods (name) VALUES ('default') - ON CONFLICT DO NOTHING + ON CONFLICT (name) DO NOTHING RETURNING id::text`).Scan(&id) if errors.Is(err, pgx.ErrNoRows) { // Lost a race; read the row the other writer created. diff --git a/internal/core/message.go b/internal/core/message.go index 2833e8d..2abdfb9 100644 --- a/internal/core/message.go +++ b/internal/core/message.go @@ -48,8 +48,11 @@ func scanMessage(row pgx.Row) (Message, error) { return m, err } -func (s *Service) GetMessage(ctx context.Context, id string) (Message, error) { - m, err := scanMessage(s.pool.QueryRow(ctx, `SELECT `+messageCols+` FROM messages WHERE id = $1`, id)) +// GetMessage is scoped to the inbox: a message id that belongs to another inbox +// returns ErrNotFound, preventing cross-inbox access. +func (s *Service) GetMessage(ctx context.Context, inboxID, id string) (Message, error) { + m, err := scanMessage(s.pool.QueryRow(ctx, + `SELECT `+messageCols+` FROM messages WHERE id = $1 AND inbox_id = $2`, id, inboxID)) if errors.Is(err, pgx.ErrNoRows) { return Message{}, ErrNotFound } @@ -70,9 +73,10 @@ func (s *Service) ListMessages(ctx context.Context, inboxID string, limit int) ( return collectMessages(rows) } -func (s *Service) GetThreadMessages(ctx context.Context, threadID string) ([]Message, error) { +func (s *Service) GetThreadMessages(ctx context.Context, inboxID, threadID string) ([]Message, error) { rows, err := s.pool.Query(ctx, - `SELECT `+messageCols+` FROM messages WHERE thread_id = $1 ORDER BY created_at ASC`, threadID) + `SELECT `+messageCols+` FROM messages WHERE thread_id = $1 AND inbox_id = $2 ORDER BY created_at ASC`, + threadID, inboxID) if err != nil { return nil, err } @@ -164,7 +168,7 @@ func (s *Service) IngestRaw(ctx context.Context, inboxID string, raw []byte) (Me if err := tx.Commit(ctx); err != nil { return Message{}, err } - return s.GetMessage(ctx, msgID) + return s.GetMessage(ctx, inboxID, msgID) } // resolveThreadTx finds the thread for a message via In-Reply-To/References @@ -220,10 +224,12 @@ var discardLog = slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{L func parseRaw(raw []byte) (parsed, error) { var pr parsed - p, err := message.EnsurePart(discardLog, false, bytes.NewReader(raw), int64(len(raw))) - if err != nil { - return pr, err - } + // EnsurePart always returns a usable Part — building an octet-stream fallback + // even when parsing hits a recoverable defect (bare CR/LF, bad Content-Type, + // missing boundary, truncated DSN). That tolerance for messy real-world mail + // is precisely why mox was chosen, so we proceed with the returned part and + // do NOT treat the recoverable error as fatal. + p, _ := message.EnsurePart(discardLog, false, bytes.NewReader(raw), int64(len(raw))) if p.Envelope != nil { e := p.Envelope pr.subject = e.Subject @@ -246,9 +252,22 @@ func parseRaw(raw []byte) (parsed, error) { } // extractBodies walks the MIME tree and returns the first text/plain and -// text/html leaf bodies (decoded UTF-8). +// text/html leaf bodies (coerced to valid UTF-8). It descends into embedded +// messages and skips attachment parts. func extractBodies(p *message.Part) (text, html string) { + // Embedded message (message/rfc822 or message/global): the sub-message lives + // under p.Message, not p.Parts. Wire its reader, then recurse — otherwise + // forwarded mail and DSN/bounce bodies are lost. + if p.Message != nil { + if err := p.SetMessageReaderAt(); err == nil { + return extractBodies(p.Message) + } + return "", "" + } if len(p.Parts) == 0 { + if isAttachment(p) { + return "", "" // an attachment is not the message body + } body := readBody(p) switch { case p.MediaType == "TEXT" && p.MediaSubType == "HTML": @@ -271,6 +290,16 @@ func extractBodies(p *message.Part) (text, html string) { return text, html } +// isAttachment reports whether a part is declared as an attachment (so it is not +// treated as the message body). Content-Disposition carries params, so we match +// the leading token. +func isAttachment(p *message.Part) bool { + if p.ContentDisposition == nil { + return false + } + return strings.HasPrefix(strings.ToLower(strings.TrimSpace(*p.ContentDisposition)), "attachment") +} + const maxBodyBytes = 2 << 20 // 2 MiB cap per body part for milestone 1 func readBody(p *message.Part) string { @@ -280,19 +309,30 @@ func readBody(p *message.Part) string { } var b strings.Builder _, _ = io.Copy(&b, io.LimitReader(rd, maxBodyBytes)) - return b.String() + // Bodies may be non-UTF-8 (mox returns raw bytes for unknown/empty charsets) + // and LimitReader can cut mid-rune; Postgres text/tsvector reject invalid + // UTF-8 and would roll back the whole ingest. Coerce to valid UTF-8. + return strings.ToValidUTF8(b.String(), "�") } +// firstAddr returns the first address that has both a localpart and a host. mox +// appends empty-User/Host entries for addresses it cannot parse; emitting "@" +// for those would be wrong, so we skip them. func firstAddr(as []message.Address) string { - if len(as) == 0 { - return "" + for _, a := range as { + if a.User != "" && a.Host != "" { + return a.User + "@" + a.Host + } } - return as[0].User + "@" + as[0].Host + return "" } func addrList(as []message.Address) []string { out := make([]string, 0, len(as)) for _, a := range as { + if a.User == "" || a.Host == "" { + continue + } out = append(out, a.User+"@"+a.Host) } return out diff --git a/internal/core/thread.go b/internal/core/thread.go index 82b703a..1efefc9 100644 --- a/internal/core/thread.go +++ b/internal/core/thread.go @@ -45,8 +45,11 @@ func (s *Service) ListThreads(ctx context.Context, inboxID string) ([]Thread, er return out, rows.Err() } -func (s *Service) GetThread(ctx context.Context, id string) (Thread, error) { - t, err := scanThread(s.pool.QueryRow(ctx, `SELECT `+threadCols+` FROM threads WHERE id = $1`, id)) +// GetThread is scoped to the inbox: a thread id belonging to another inbox +// returns ErrNotFound. +func (s *Service) GetThread(ctx context.Context, inboxID, id string) (Thread, error) { + t, err := scanThread(s.pool.QueryRow(ctx, + `SELECT `+threadCols+` FROM threads WHERE id = $1 AND inbox_id = $2`, id, inboxID)) if errors.Is(err, pgx.ErrNoRows) { return Thread{}, ErrNotFound } diff --git a/internal/store/migrations/0001_init.sql b/internal/store/migrations/0001_init.sql index d6625a4..2139a77 100644 --- a/internal/store/migrations/0001_init.sql +++ b/internal/store/migrations/0001_init.sql @@ -7,7 +7,8 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto; -- gen_random_uuid() CREATE TABLE IF NOT EXISTS pods ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), name text NOT NULL, - created_at timestamptz NOT NULL DEFAULT now() + created_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (name) ); -- Per-domain DKIM keys + DNS verification state. @@ -142,6 +143,8 @@ CREATE INDEX IF NOT EXISTS idx_inboxes_pod ON inboxes(pod_id); CREATE INDEX IF NOT EXISTS idx_threads_inbox ON threads(inbox_id); CREATE INDEX IF NOT EXISTS idx_messages_thread ON messages(thread_id); CREATE INDEX IF NOT EXISTS idx_messages_inbox_time ON messages(inbox_id, created_at DESC); +-- Supports per-delivery thread resolution (message_id_hdr = ANY(...) per inbox). +CREATE INDEX IF NOT EXISTS idx_messages_msgid ON messages(inbox_id, message_id_hdr) WHERE message_id_hdr IS NOT NULL; CREATE INDEX IF NOT EXISTS idx_messages_ts ON messages USING gin(ts); CREATE INDEX IF NOT EXISTS idx_messages_labels ON messages USING gin(labels); CREATE INDEX IF NOT EXISTS idx_threads_labels ON threads USING gin(labels);