OSS/openmail
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix 13 findings from adversarial milestone-1 review

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
karti-ai
2026-06-21 13:22:35 -07:00
parent 319a2fa689
commit 428040d964
6 changed files with 93 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/handlers.go
+27 -21
View File
@@ -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 {
internal/api/server.go
+2 -1
View File
@@ -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
}
internal/core/core.go
+1 -1
View File
@@ -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.
internal/core/message.go
+54 -14
View File
@@ -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
internal/core/thread.go
+5 -2
View File
@@ -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
}
internal/store/migrations/0001_init.sql
+4 -1
View File
@@ -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);