amadan

carlos / platform Public

Merge pull request #280 from carlosframework/ses-event-path

Count deliveries, and say when a topic is refused. Closes the first two of #270.

Merged with main brought in — including #325's Server fields, which land beside this branch's refusedTopics; both kept. Gate reproduced locally on the merge result: 37 packages, vet and gofmt clean, GOOS=darwin build.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 0489037c56f2e011cd0332fdcae9dd17f37ba94d parent 039c064parent 043436d
7 files changed, +210 −6
  • internal/console/apimail.go +1 −0
  • internal/console/mailcounts.go +9 −2
  • internal/console/server.go +6 −0
  • internal/console/sesevents.go +72 −0
  • internal/console/sesevents_test.go +108 −0
  • internal/console/store.go +9 −0
  • mail_cli.go +5 −4

CI passed — run details

diff --git a/internal/console/apimail.go b/internal/console/apimail.go
index 136f882e7..59359f671 100644
--- a/internal/console/apimail.go
+++ b/internal/console/apimail.go
@@ -424,6 +424,7 @@ func (s *Server) emailReply(ctx context.Context, acct *Account, app string, rec
"counts": map[string]any{
"sends": counts.Sends, "bounces": counts.Bounces,
"complaints": counts.Complaints, "rejects": counts.Rejects,
+ "deliveries": counts.Deliveries,
},
}, nil
}
diff --git a/internal/console/mailcounts.go b/internal/console/mailcounts.go
index 60e4bd265..4fbccca63 100644
--- a/internal/console/mailcounts.go
+++ b/internal/console/mailcounts.go
@@ -25,6 +25,11 @@ type MailCounts struct {
Bounces int
Complaints int
Rejects int
+ // Deliveries is the denominator that makes a bounce count readable —
+ // "12 bounces" means something different against 50 delivered than
+ // against 50 sent and 12 delivered. Display-only, deliberately: see
+ // BounceRate below and the migration's own note.
+ Deliveries int
}
// BounceRate and ComplaintRate are zero on a day with no sends — not NaN.
@@ -86,6 +91,8 @@ func mailCountColumn(evType string) (string, bool) {
return "complaints", true
case mail.EventReject:
return "rejects", true
+ case mail.EventDelivery:
+ return "deliveries", true
default:
return "", false
}
@@ -112,9 +119,9 @@ ON CONFLICT(account, app, day) DO UPDATE SET %s = %s + 1`, col, col, col)
// all is a legitimate zero, not an error.
func (s *Store) MailCounts(ctx context.Context, acct, app, day string) (MailCounts, error) {
var c MailCounts
- row := s.db.QueryRowContext(ctx, `SELECT sends, bounces, complaints, rejects
+ row := s.db.QueryRowContext(ctx, `SELECT sends, bounces, complaints, rejects, deliveries
FROM mail_counts WHERE account = ? AND app = ? AND day = ?`, acct, app, day)
- err := row.Scan(&c.Sends, &c.Bounces, &c.Complaints, &c.Rejects)
+ err := row.Scan(&c.Sends, &c.Bounces, &c.Complaints, &c.Rejects, &c.Deliveries)
if err == sql.ErrNoRows {
return MailCounts{}, nil
}
diff --git a/internal/console/server.go b/internal/console/server.go
index 781b9cb81..90ea0f831 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -693,6 +693,12 @@ type Server struct {
// TopicArn, which is not a secret, triggers a fresh outbound HTTPS fetch
// and full chain verification. Zero value is ready to use.
snsCerts snsCertCache
+ // refusedTopics bounds the SES ingest door's refusal log (sesevents.go's
+ // logRefusedTopic): the door is unauthenticated and internet-reachable,
+ // so the diagnostic has to be rate-limited rather than omitted. Zero
+ // value is ready to use.
+ refusedTopicsMu sync.Mutex
+ refusedTopics map[string]time.Time
// edgeFold caches the deployment's folded live edge addresses — every
// box's control/edges/<box>.json joined against its status snapshot —
// so the guidance strings can name every armed edge rather than only
diff --git a/internal/console/sesevents.go b/internal/console/sesevents.go
index 0aaa0b2c7..05fabef6e 100644
--- a/internal/console/sesevents.go
+++ b/internal/console/sesevents.go
@@ -328,6 +328,7 @@ func (s *Server) handleSESEvents(w http.ResponseWriter, r *http.Request) {
// accepts nothing, rather than accepting whatever a caller who also
// sends "" happens to send.
if s.cfg.MailDeployment.SNSTopicARN == "" || env.TopicArn != s.cfg.MailDeployment.SNSTopicARN {
+ s.logRefusedTopic(env.TopicArn)
jsonError(w, http.StatusForbidden, "unrecognised SNS topic")
return
}
@@ -692,3 +693,74 @@ func verifySNSSignatureBytes(pub *rsa.PublicKey, canon []byte, sigVersion, sigB6
}
return nil
}
+
+// refusedTopicWindow is how long one distinct refused ARN stays quiet after
+// being logged, and refusedTopicMax how many distinct ARNs are remembered.
+const (
+ refusedTopicWindow = time.Hour
+ refusedTopicMax = 8
+)
+
+// logRefusedTopic names a rejected TopicArn in the log, at most once per
+// hour per distinct ARN.
+//
+// WHY LOG AT ALL. The TopicArn allowlist is this door's first real gate, and
+// a stale or mistyped CARLOS_MAIL_SNS_TOPIC_ARN is the single most likely
+// misconfiguration in the whole event path. Before this it left NO trace:
+// 403s at a door nobody watches, and counts that never move. "Events stopped
+// arriving and nothing said so" is precisely the silence this is for. The
+// ARN is not a secret — it is published in terraform and read off the box —
+// so naming it costs nothing.
+//
+// WHY BOUNDED. This door is UNAUTHENTICATED and internet-reachable by
+// design (SNS has no credential to present), so an unbounded log line here
+// is a stranger's write access to the disk. Bounding answers that without
+// giving up the diagnostic: the misconfiguration this exists to catch
+// produces exactly ONE distinct ARN, so one line an hour says everything,
+// while a flood of fabricated ARNs costs at most refusedTopicMax lines an
+// hour and a map of the same size.
+//
+// The 403 body is deliberately unchanged. The external answer stays
+// constant and uninformative; the operator's answer goes to the log. They
+// are different readers and only one of them is trusted.
+func (s *Server) logRefusedTopic(got string) {
+ // %q escapes control bytes, so a caller cannot forge log lines; the
+ // truncation bounds the damage a very long value can do to the line. A
+ // real ARN is about 100 bytes.
+ if len(got) > 256 {
+ got = got[:256]
+ }
+ want := s.cfg.MailDeployment.SNSTopicARN
+ wantStr := fmt.Sprintf("%q", want)
+ if want == "" {
+ // Its own misconfiguration signal, and a different one: the
+ // deployment believes email events are off while SNS is actively
+ // delivering to it.
+ wantStr = "(none configured)"
+ }
+
+ s.refusedTopicsMu.Lock()
+ now := time.Now()
+ if s.refusedTopics == nil {
+ s.refusedTopics = map[string]time.Time{}
+ }
+ if last, seen := s.refusedTopics[got]; seen && now.Sub(last) < refusedTopicWindow {
+ s.refusedTopicsMu.Unlock()
+ return
+ }
+ if _, seen := s.refusedTopics[got]; !seen && len(s.refusedTopics) >= refusedTopicMax {
+ // Evict the oldest rather than refusing to record: a genuine new
+ // misconfiguration arriving after a flood must still be logged.
+ oldestK, oldestT := "", now
+ for k, t := range s.refusedTopics {
+ if t.Before(oldestT) || oldestK == "" {
+ oldestK, oldestT = k, t
+ }
+ }
+ delete(s.refusedTopics, oldestK)
+ }
+ s.refusedTopics[got] = now
+ s.refusedTopicsMu.Unlock()
+
+ log.Printf("console: ses events: refused topic %q — this deployment expects %s", got, wantStr)
+}
diff --git a/internal/console/sesevents_test.go b/internal/console/sesevents_test.go
index e25b2c0a4..d02c32a5f 100644
--- a/internal/console/sesevents_test.go
+++ b/internal/console/sesevents_test.go
@@ -12,6 +12,7 @@ package console
// ordering that runs all of it before any network call or trusted field.
import (
+ "bytes"
"context"
"crypto"
"crypto/rand"
@@ -24,10 +25,12 @@ import (
"encoding/base64"
"encoding/pem"
"errors"
+ "log"
"math/big"
"net"
"net/http"
"net/http/httptest"
+ "os"
"strings"
"testing"
"time"
@@ -196,6 +199,111 @@ func TestSESEventsRefusesForeignTopicEvenWithAnOtherwiseValidRequest(t *testing.
}
}
+// TestSESEventsLogsARefusedTopicOnceAnHour is ruling 2c. A stale or mistyped
+// CARLOS_MAIL_SNS_TOPIC_ARN is the likeliest misconfiguration in the whole
+// event path, and before this it left NO trace: 403s at a door nobody
+// watches and counts that never move.
+//
+// The bound is the other half of the ruling and is tested with it, because
+// this door is unauthenticated and internet-reachable — an unbounded log
+// line here is a stranger's write access to the operator's disk.
+func TestSESEventsLogsARefusedTopicOnceAnHour(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.MailDeployment.SNSTopicARN = "arn:aws:sns:eu-west-1:1:carlos-mail"
+ var buf bytes.Buffer
+ log.SetOutput(&buf)
+ t.Cleanup(func() { log.SetOutput(os.Stderr) })
+
+ body := `{"Type":"Notification","TopicArn":"arn:aws:sns:eu-west-1:9:someone-else",` +
+ `"Message":"{}","SigningCertURL":"https://sns.eu-west-1.amazonaws.com/x.pem","Signature":"x"}`
+ for i := 0; i < 100; i++ {
+ resp := sesPost(t, fx, body)
+ resp.Body.Close()
+ }
+
+ got := buf.String()
+ if n := strings.Count(got, "refused topic"); n != 1 {
+ t.Errorf("logged %d refusal lines for 100 identical requests, want exactly 1 — "+
+ "an unauthenticated door must not let a stranger fill the log\n%s", n, got)
+ }
+ // Both ARNs must appear: the received one to identify the sender, the
+ // expected one because "refused" is useless without what it wanted.
+ if !strings.Contains(got, "someone-else") || !strings.Contains(got, "carlos-mail") {
+ t.Errorf("the line does not name both the received and expected ARNs:\n%s", got)
+ }
+}
+
+// TestSESEventsRefusalLogNamesAnUnconfiguredDeploymentDifferently: a console
+// with no topic configured that is nonetheless receiving SNS deliveries is
+// its own misconfiguration — the deployment believes email events are off
+// while SNS is actively delivering to it — and %q of the empty string would
+// read as a bug rather than as that.
+func TestSESEventsRefusalLogNamesAnUnconfiguredDeploymentDifferently(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.MailDeployment.SNSTopicARN = ""
+ var buf bytes.Buffer
+ log.SetOutput(&buf)
+ t.Cleanup(func() { log.SetOutput(os.Stderr) })
+
+ resp := sesPost(t, fx, `{"Type":"Notification","TopicArn":"arn:aws:sns:eu-west-1:9:whoever",`+
+ `"Message":"{}","SigningCertURL":"https://sns.eu-west-1.amazonaws.com/x.pem","Signature":"x"}`)
+ resp.Body.Close()
+
+ if got := buf.String(); !strings.Contains(got, "(none configured)") {
+ t.Errorf("an unconfigured deployment should say so, not print an empty quoted string:\n%s", got)
+ }
+}
+
+// TestSESEventsCountsADelivery is ruling 2a, through the DOOR rather than
+// against mailCountColumn — the standing lesson of this subsystem is that a
+// mapping can be right while nothing calls it.
+func TestSESEventsCountsADelivery(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.MailDeployment.SNSTopicARN = "arn:aws:sns:eu-west-1:1:carlos-mail"
+ fx.srv.snsVerify = func(_ snsEnvelope) error { return nil }
+ ctx := context.Background()
+ now := time.Now()
+ if _, err := fx.st.CreateAppIn(OpsAccountID, "seapointish", "paul@keymail.dev", now); err != nil {
+ t.Fatal(err)
+ }
+ rec := &mail.DomainRecord{
+ Domain: "seapointish.bag.oncarlos.com",
+ Kind: mail.KindPlatform,
+ Apps: []string{"seapointish"},
+ Regions: map[string]*mail.RegionState{
+ "eu-west-1": {
+ IdentityARN: "arn:aws:ses:eu-west-1:1:identity/seapointish.bag.oncarlos.com",
+ MailFrom: "mail-eu-west-1.seapointish.bag.oncarlos.com",
+ Status: mail.StatusVerified,
+ },
+ },
+ CreatedAt: now,
+ }
+ if _, err := mail.PutDomain(ctx, fx.rel, release.OpsAccount, rec, ""); err != nil {
+ t.Fatal(err)
+ }
+
+ resp := sesPost(t, fx, `{"Type":"Notification","TopicArn":"arn:aws:sns:eu-west-1:1:carlos-mail",`+
+ `"Message":"{\"eventType\":\"Delivery\",\"mail\":{\"source\":\"noreply@seapointish.bag.oncarlos.com\"}}",`+
+ `"SigningCertURL":"https://sns.eu-west-1.amazonaws.com/x.pem","Signature":"x"}`)
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("status = %d, want 200", resp.StatusCode)
+ }
+
+ day := now.UTC().Format("2006-01-02")
+ c, err := fx.st.MailCounts(ctx, release.OpsAccount, "seapointish", day)
+ if err != nil {
+ t.Fatalf("MailCounts: %v", err)
+ }
+ if c.Deliveries != 1 {
+ t.Errorf("deliveries = %d, want 1 — the ingest already verifies and parses these; dropping them wastes the work", c.Deliveries)
+ }
+ if c.Sends != 0 || c.Bounces != 0 || c.Complaints != 0 || c.Rejects != 0 {
+ t.Errorf("a Delivery moved another column: %+v — the cap and thresholds must be untouched by it", c)
+ }
+}
+
// An unknown event type is a 200. SNS retries non-2xx, so answering 500
// turns one unrecognised field into an unbounded retry flood against the
// console.
diff --git a/internal/console/store.go b/internal/console/store.go
index 96c5f2eb6..6fdeb890a 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -206,6 +206,15 @@ func (s *Store) migrate() error {
// re-running is always safe.
for _, stmt := range []string{
`ALTER TABLE users ADD COLUMN operator INTEGER NOT NULL DEFAULT 0;`,
+ // deliveries (tenant-email, ruling 2a): SES already publishes
+ // Delivery to the event destination and the ingest already
+ // verifies and parses it — this stops throwing the answer away.
+ // Display-only: the cap and the bounce/complaint thresholds are
+ // deliberately NOT rebased on it, because SES's own account-review
+ // math is bounces over SENDS, and a limiter that measures something
+ // else from what SES reviews on is a limiter that trips at the
+ // wrong time.
+ `ALTER TABLE mail_counts ADD COLUMN deliveries INTEGER NOT NULL DEFAULT 0;`,
// Where a returning user lands: the account they were last in.
`ALTER TABLE users ADD COLUMN last_account INTEGER NOT NULL DEFAULT 0;`,
// Apps belong to an account. The column lands with default 1 (ops), so
diff --git a/mail_cli.go b/mail_cli.go
index e00d7dfba..891d80c11 100644
--- a/mail_cli.go
+++ b/mail_cli.go
@@ -224,6 +224,7 @@ type apiEmailCounts struct {
Bounces int `json:"bounces"`
Complaints int `json:"complaints"`
Rejects int `json:"rejects"`
+ Deliveries int `json:"deliveries"`
}
// apiEmailReply is the shape every declare/status/enable/rotate/pause door
@@ -877,8 +878,8 @@ func printEmailStatus(w io.Writer, res apiEmailReply) {
// honest thing to print is the evidence a member CAN see
// (today's counts against the cap), not a diagnosis this
// reply cannot actually back up.
- mutedf(w, " today: %d sends, %d bounces, %d complaints, %d rejects against a cap of %d",
- res.Counts.Sends, res.Counts.Bounces, res.Counts.Complaints, res.Counts.Rejects, res.DailyCap)
+ mutedf(w, " today: %d sends, %d delivered, %d bounces, %d complaints, %d rejects against a cap of %d",
+ res.Counts.Sends, res.Counts.Deliveries, res.Counts.Bounces, res.Counts.Complaints, res.Counts.Rejects, res.DailyCap)
// I3 (review follow-up): `carlos email resume` is OPERATOR-only
// and 404s for an ordinary member — pointing every paused
// member at it, regardless of why, was actively wrong advice.
@@ -908,8 +909,8 @@ func printEmailStatus(w io.Writer, res apiEmailReply) {
mutedf(w, " not enabled yet — carlos email enable --app %s", res.App)
}
- fmt.Fprintf(w, " today: %d/%d sends, %d bounces, %d complaints, %d rejects\n",
- res.Counts.Sends, res.DailyCap, res.Counts.Bounces, res.Counts.Complaints, res.Counts.Rejects)
+ fmt.Fprintf(w, " today: %d/%d sends, %d delivered, %d bounces, %d complaints, %d rejects\n",
+ res.Counts.Sends, res.DailyCap, res.Counts.Deliveries, res.Counts.Bounces, res.Counts.Complaints, res.Counts.Rejects)
if len(res.MissingVars) > 0 {
warnf(w, " drift: %s not yet delivered to this app's config", strings.Join(res.MissingVars, ", "))