diff --git a/internal/mailgun/webhook.go b/internal/mailgun/webhook.go index 43931d7..dc2ece0 100644 --- a/internal/mailgun/webhook.go +++ b/internal/mailgun/webhook.go @@ -4,11 +4,14 @@ import ( "encoding/json" "errors" "fmt" + "io" "mime/multipart" "net/http" + "net/mail" "sort" "strconv" "strings" + "time" "filemill/internal/alert" "filemill/internal/app" @@ -21,7 +24,8 @@ const requestOverhead = 1 << 20 // 1 MiB // handle is the webhook entry point. It delegates the decision to receive and // then maps the single (status, reason) result onto the HTTP response: reasons // are logged (useful for a public endpoint), and 4xx/5xx additionally send an -// error body so Mailgun can distinguish "don't retry" (4xx) from "retry" (5xx). +// error body. Mailgun itself only distinguishes 406 ("don't retry") from every +// other non-200 code ("retry"). func (s *Service) handle(w http.ResponseWriter, r *http.Request) { // Unconditional arrival log: any request that reaches this handler is // recorded before any check, so "did the POST arrive?" is never ambiguous. @@ -43,10 +47,15 @@ func (s *Service) handle(w http.ResponseWriter, r *http.Request) { // taxonomy is deliberate: // // 200 - accepted, or "received, nothing to do" (Mailgun should stop) -// 400 - the client's fault; do not retry (malformed body, oversize) // 401 - bad, missing, or stale signature // 405 - not a POST -// 500 - our fault; please retry (storage, filesystem, submit failures) +// 406 - can never succeed; do not retry (malformed body, oversize) +// 500 - our fault or the network's; please retry (cut-off upload, storage, +// filesystem, submit failures) +// +// Mailgun retries every non-200 response except 406, on a fixed schedule over +// 8 hours, so a permanent refusal must be 406 - a 400 is retried just like a +// 500. // // A rejected attachment (wrong file type) is the sender's fault and can never // succeed on retry, so it maps to 200, not 500 — returning 500 made Mailgun @@ -56,8 +65,10 @@ func (s *Service) receive(r *http.Request) (status int, reason string) { if r.Method != http.MethodPost { return http.StatusMethodNotAllowed, "method not allowed" } - if err := parseForm(r); err != nil { - return http.StatusBadRequest, "malformed request body" + body := &countingBody{ReadCloser: r.Body} + r.Body = body + if status, reason, failed := bodyFailure(parseForm(r), body, r.ContentLength); failed { + return status, reason } if !authenticSignature(s.signKey, r.FormValue("timestamp"), r.FormValue("token"), r.FormValue("signature")) { return http.StatusUnauthorized, "invalid signature" @@ -110,7 +121,7 @@ func (s *Service) receive(r *http.Request) (status int, reason string) { } if !s.withinSizeLimit(files) { - return http.StatusBadRequest, fmt.Sprintf("attachment exceeds size limit (to %s from %s)", recipient, sender) + return http.StatusNotAcceptable, fmt.Sprintf("attachment exceeds size limit (to %s from %s)", recipient, sender) } if err := s.intake(r, operation, files); err != nil { @@ -127,7 +138,85 @@ func (s *Service) receive(r *http.Request) (status int, reason string) { }) return http.StatusInternalServerError, fmt.Sprintf("intake failed from %s: %v", sender, err) } - return http.StatusOK, fmt.Sprintf("accepted: sender=%q recipient=%q operation=%s", sender, r.FormValue("recipient"), operation) + return http.StatusOK, fmt.Sprintf("accepted: sender=%q recipient=%q operation=%s %s", sender, r.FormValue("recipient"), operation, s.arrival(r)) +} + +// countingBody counts the bytes read from a request body and keeps the first +// read error, so a body that would not parse can be told apart: cut off in +// transit, or complete but malformed. +type countingBody struct { + io.ReadCloser + n int64 + err error +} + +func (c *countingBody) Read(p []byte) (int, error) { + n, err := c.ReadCloser.Read(p) + c.n += int64(n) + if err != nil && err != io.EOF && c.err == nil { + c.err = err + } + return n, err +} + +// bodyFailure judges the request body after parsing, reporting failed when the +// request must not go on. The rest of the body is drained first: the parser +// stops at its first error, often long before the end, and it also takes a +// clean end-of-stream between parts as the end of the form - so a body cut +// short there parses without error, and only the byte count shows it is +// incomplete. A cut-off upload (Mailgun gives up after 10 seconds) is not the +// sender's fault and is retried (500); a body that is too large, or arrived +// whole and still would not parse, can never succeed and is refused (406). +func bodyFailure(parseErr error, body *countingBody, declared int64) (status int, reason string, failed bool) { + _, _ = io.Copy(io.Discard, body) + var tooLarge *http.MaxBytesError + switch { + case errors.As(parseErr, &tooLarge) || errors.As(body.err, &tooLarge): + return http.StatusNotAcceptable, fmt.Sprintf("request body exceeds %d bytes", tooLarge.Limit), true + case body.err != nil: + return http.StatusInternalServerError, fmt.Sprintf("request body incomplete: got %d bytes: %v", body.n, body.err), true + case declared > 0 && body.n < declared: + return http.StatusInternalServerError, fmt.Sprintf("request body incomplete: got %d of %d bytes", body.n, declared), true + case parseErr != nil: + return http.StatusNotAcceptable, fmt.Sprintf("malformed request body: %v", parseErr), true + } + return 0, "", false +} + +// arrival describes when an email was sent (its Date header), when Mailgun +// posted it (the signed timestamp), and the lag from sending to arrival here. +// Without these, a slow reply could not be placed: before the sender's email +// reached Mailgun, between Mailgun and here, or after. +func (s *Service) arrival(r *http.Request) string { + posted := "unknown" + if unix, err := strconv.ParseInt(r.FormValue("timestamp"), 10, 64); err == nil { + posted = time.Unix(unix, 0).UTC().Format(time.RFC3339) + } + sent, ok := sentAt(r) + if !ok { + return "sent=unknown posted=" + posted + } + lag := s.clock().Sub(sent).Round(time.Second) + return fmt.Sprintf("sent=%s posted=%s lag=%s", sent.UTC().Format(time.RFC3339), posted, lag) +} + +// sentAt reads the email's Date header. Mailgun posts headers as top-level +// fields (Message-Id is read the same way), and also as the message-headers +// JSON list of [name, value] pairs, which is the fallback. +func sentAt(r *http.Request) (time.Time, bool) { + raw := r.FormValue("Date") + if raw == "" { + var headers [][]string + _ = json.Unmarshal([]byte(r.FormValue("message-headers")), &headers) + for _, h := range headers { + if len(h) == 2 && strings.EqualFold(h[0], "Date") { + raw = h[1] + break + } + } + } + sent, err := mail.ParseDate(raw) + return sent, err == nil } // parseForm decodes the request body. Mailgun sends no-attachment notifications diff --git a/internal/mailgun/webhook_test.go b/internal/mailgun/webhook_test.go index cdd0717..dbdac9a 100644 --- a/internal/mailgun/webhook_test.go +++ b/internal/mailgun/webhook_test.go @@ -5,7 +5,9 @@ import ( "crypto/hmac" "crypto/sha256" "encoding/hex" + "errors" "fmt" + "io" "log" "mime/multipart" "net/http" @@ -70,8 +72,186 @@ func TestWebhookRejectsOversizeAttachment(t *testing.T) { map[string][]byte{"attachment-1": []byte("too large")}) w := httptest.NewRecorder() s.handle(w, r) - if w.Code != http.StatusBadRequest { - t.Fatalf("got %d, want 400", w.Code) + if w.Code != http.StatusNotAcceptable { + t.Fatalf("got %d, want 406 (an oversize attachment can never succeed on retry)", w.Code) + } +} + +// Mailgun retries every non-200 response except 406, so a body that will never +// parse must get 406 or Mailgun resends it for 8 hours. The parser's error is +// logged, because "malformed" alone could not be diagnosed after the fact. +func TestWebhookRefusesMalformedBody(t *testing.T) { + logger, buf := captureLogger() + s := &Service{signKey: "key", maxBytes: 1024, routes: map[string]string{}, allowed: map[string]bool{}, log: logger} + r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", strings.NewReader("this is not multipart")) + r.Header.Set("Content-Type", "multipart/form-data; boundary=xyz") + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusNotAcceptable { + t.Fatalf("got %d, want 406", w.Code) + } + if !strings.Contains(buf.String(), "malformed request body: ") { + t.Errorf("the parser's error must be logged; got %q", buf.String()) + } +} + +func TestWebhookRefusesMultipartWithoutBoundary(t *testing.T) { + s := &Service{signKey: "key", maxBytes: 1024, routes: map[string]string{}, allowed: map[string]bool{}, log: discardLogger()} + r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", strings.NewReader("--xyz--")) + r.Header.Set("Content-Type", "multipart/form-data") + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusNotAcceptable { + t.Fatalf("got %d, want 406", w.Code) + } +} + +func TestWebhookRefusesBodyOverTheRequestLimit(t *testing.T) { + logger, buf := captureLogger() + s := &Service{signKey: "key", maxBytes: 2, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger} + r := signedMultipart(t, s.signKey, + map[string]string{"recipient": "workerlist@mill.example.com"}, + map[string][]byte{"attachment-1": bytes.Repeat([]byte("x"), requestOverhead+1024)}) + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusNotAcceptable { + t.Fatalf("got %d, want 406", w.Code) + } + if !strings.Contains(buf.String(), "request body exceeds") { + t.Errorf("the size refusal must say so; got %q", buf.String()) + } +} + +// A body that stops short of its Content-Length was cut off in transit - on +// 2026-10-28 Mailgun hung up after 10 seconds mid-upload. That is not the +// sender's fault and the retry will likely succeed, so it must stay retryable. +// Every cut point is tried: the multipart parser takes a clean end-of-stream +// between parts as the end of the form, so some cuts parse without error and +// only the byte count catches them. +func TestWebhookRetriesBodyCutOffBeforeContentLength(t *testing.T) { + full := signedMultipart(t, "key", + map[string]string{"recipient": "workerlist@mill.example.com"}, + map[string][]byte{"attachment-1": []byte("pdf")}) + raw, err := io.ReadAll(full.Body) + if err != nil { + t.Fatal(err) + } + for cut := 1; cut < len(raw); cut++ { + logger, buf := captureLogger() + s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger} + r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", bytes.NewReader(raw[:cut])) + r.Header.Set("Content-Type", full.Header.Get("Content-Type")) + r.ContentLength = int64(len(raw)) + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusInternalServerError { + t.Fatalf("cut at %d of %d: got %d, want 500 (a cut-off upload must be retried)", cut, len(raw), w.Code) + } + want := fmt.Sprintf("request body incomplete: got %d of %d bytes", cut, len(raw)) + if !strings.Contains(buf.String(), want) { + t.Fatalf("cut at %d: log must contain %q; got %q", cut, want, buf.String()) + } + } +} + +// failingReader yields its data and then a read error, like a connection reset +// partway through an upload. +type failingReader struct { + data []byte + err error +} + +func (f *failingReader) Read(p []byte) (int, error) { + if len(f.data) == 0 { + return 0, f.err + } + n := copy(p, f.data) + f.data = f.data[n:] + return n, nil +} + +func TestWebhookRetriesBodyWhoseReadFails(t *testing.T) { + logger, buf := captureLogger() + s := &Service{signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger} + full := signedMultipart(t, s.signKey, + map[string]string{"recipient": "workerlist@mill.example.com"}, + map[string][]byte{"attachment-1": []byte("pdf")}) + raw, err := io.ReadAll(full.Body) + if err != nil { + t.Fatal(err) + } + r := httptest.NewRequest(http.MethodPost, "/mailgun-webhook", &failingReader{data: raw[:len(raw)/2], err: errors.New("connection reset by peer")}) + r.Header.Set("Content-Type", full.Header.Get("Content-Type")) + r.ContentLength = -1 // chunked: no length to compare against, only the read error + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusInternalServerError { + t.Fatalf("got %d, want 500 (a failed read must be retried)", w.Code) + } + if !strings.Contains(buf.String(), "connection reset by peer") { + t.Errorf("the read error must be logged; got %q", buf.String()) + } +} + +// The accept line records when the email was sent (its Date header, in UTC), +// when Mailgun posted it (the signed timestamp), and the lag from sending to +// arrival here - so a "my reply was slow" report can be measured from the log. +func TestWebhookLogsArrivalTimesOnAccept(t *testing.T) { + logger, buf := captureLogger() + sent := time.Date(2026, 10, 28, 4, 6, 44, 0, time.UTC) + s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger, + now: func() time.Time { return sent.Add(18*time.Minute + 19*time.Second + 400*time.Millisecond) }} + r := signedMultipart(t, s.signKey, + map[string]string{"recipient": "workerlist@mill.example.com", "sender": "sender@example.com", "Date": "Tue, 27 Oct 2026 22:06:44 -0600"}, + map[string][]byte{"attachment-1": []byte("pdf")}) + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusOK { + t.Fatalf("got %d, want 200", w.Code) + } + unix, err := strconv.ParseInt(r.FormValue("timestamp"), 10, 64) + if err != nil { + t.Fatal(err) + } + posted := time.Unix(unix, 0).UTC().Format(time.RFC3339) + for _, want := range []string{"sent=2026-10-28T04:06:44Z", "posted=" + posted, "lag=18m19s"} { + if !strings.Contains(buf.String(), want) { + t.Errorf("accept line must contain %q; got %q", want, buf.String()) + } + } +} + +func TestWebhookReadsDateFromMessageHeadersWhenFieldIsAbsent(t *testing.T) { + logger, buf := captureLogger() + s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger} + headers := `[["Subject", "Sister INI"], ["Date", "Wed, 28 Oct 2026 04:14:49 +0000"]]` + r := signedMultipart(t, s.signKey, + map[string]string{"recipient": "workerlist@mill.example.com", "sender": "sender@example.com", "message-headers": headers}, + map[string][]byte{"attachment-1": []byte("pdf")}) + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusOK { + t.Fatalf("got %d, want 200", w.Code) + } + if !strings.Contains(buf.String(), "sent=2026-10-28T04:14:49Z") { + t.Errorf("Date must be read from message-headers; got %q", buf.String()) + } +} + +// A missing or garbled Date is logged as unknown and never blocks the email. +func TestWebhookAcceptsEmailWithUnparseableDate(t *testing.T) { + logger, buf := captureLogger() + s := &Service{engine: newFakeEngine(), signKey: "key", maxBytes: 1024, routes: map[string]string{"workerlist@mill.example.com": "workerlist"}, allowed: map[string]bool{}, log: logger} + r := signedMultipart(t, s.signKey, + map[string]string{"recipient": "workerlist@mill.example.com", "sender": "sender@example.com", "Date": "sometime last week"}, + map[string][]byte{"attachment-1": []byte("pdf")}) + w := httptest.NewRecorder() + s.handle(w, r) + if w.Code != http.StatusOK { + t.Fatalf("got %d, want 200", w.Code) + } + if !strings.Contains(buf.String(), "sent=unknown") || strings.Contains(buf.String(), "lag=") { + t.Errorf("an unparseable Date must log sent=unknown and no lag; got %q", buf.String()) } }