Phase 8: level-2 differentiated rate limits (spec 7.4)

The journal-milter, until now a pure monitor, now refuses a message with a
4xx tempfail (RespTempFail/451) at MAIL FROM when a per-domain or per-
application limit is exceeded. Key is the client IP; the count is
COUNT(DISTINCT queue_id) over a sliding window reusing the send log; the
limit applies only when a non-empty IP binding matches the client (empty
binding => level-1 only, per spec 7.4). Enforcement is fail-open on the
milter's own errors — a limiter malfunction never blocks mail, and Postfix's
level-1 anvil limit stays the independent backstop. Refused messages are
recorded in send_log with status "rejected" for UI visibility.

- store/ratelimits.go: RateLimit type (+Active/AllowsIP), id-keyed get/set/
  delete for the panel, name/login-keyed lookup + windowed distinct-message
  count for the milter, DeleteRateLimitsForDomain. No migration — the
  rate_limits table has existed since Phase 2.
- milter: enforce at MailFrom, fail-open helper overLimit, InsertRejected.
- web: server-side validated IP/ceiling/window forms on the domain page and
  per application; routes POST /domains/{id}/ratelimit and
  /applications/{aid}/ratelimit. Milter reads rows live, so no reload.
- domain/app services clear limits on deletion (rate_limits has no FK cascade).

Unit tests + container e2e (p8) green: refusal on both scopes, unregistered
IP ignored, fail-open with the panel stopped.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

internal/milter/milter.go

@@ -16,16 +16,22 @@ import (
 	"net"
 	"net/textproto"
 	"strings"
+	"time"
 
 	"github.com/emersion/go-milter"
 
 	"codeberg.org/mix/selfpost/internal/store"
 )
 
-// Recorder persists queued send-log entries. *store.Store satisfies it; tests
-// substitute a fake.
-type Recorder interface {
+// Store is the persistence the milter needs on the receive path: recording
+// accepted messages (spec 7.3) and, for level-2 rate limiting (spec 7.4),
+// looking up the configured limits and counting recent messages. *store.Store
+// satisfies it; tests substitute a fake.
+type Store interface {
 	InsertQueued(e store.SendLogEntry) error
+	InsertRejected(e store.SendLogEntry) error
+	RateLimit(scope, ref string) (store.RateLimit, bool, error)
+	CountMessages(scope, ref string, since time.Time) (int64, error)
 }
 
 // session accumulates the fields of one message as the milter callbacks fire.
@@ -37,7 +43,7 @@ type Recorder interface {
 // MailFrom (the start of every transaction).
 type session struct {
 	milter.NoOpMilter
-	rec Recorder
+	rec Store
 
 	clientIP string // captured once per connection
 
@@ -59,12 +65,20 @@ func (s *session) Connect(host, family string, port uint16, addr net.IP, m *milt
 
 // MailFrom starts a new message: reset per-message state, then capture the
 // envelope sender and the SASL login ({auth_authen}, carried by the MAIL-stage
-// macros).
+// macros). This is also the earliest stage where both the sending domain (from
+// the sender) and the application (the login) are known, so the level-2 rate
+// limit is enforced here: over the limit, the message is refused with a 4xx
+// tempfail before recipients are even offered (spec 7.4). Enforcement is
+// fail-open — see overLimit.
 func (s *session) MailFrom(from string, m *milter.Modifier) (milter.Response, error) {
 	s.from = cleanAddress(from)
 	s.login = macro(m, "auth_authen")
 	s.rcpts = nil
 	s.subject = ""
+	if s.overLimit() {
+		s.recordRejected()
+		return milter.RespTempFail, nil
+	}
 	return milter.RespContinue, nil
 }
 
@@ -152,7 +166,7 @@ func domainOf(addr string) string {
 
 // Serve runs the journal-milter on ln until ctx is cancelled. Each connection
 // gets a fresh session bound to rec. It returns nil on a clean shutdown.
-func Serve(ctx context.Context, ln net.Listener, rec Recorder) error {
+func Serve(ctx context.Context, ln net.Listener, rec Store) error {
 	srv := &milter.Server{
 		NewMilter: func() milter.Milter { return &session{rec: rec} },
 		Actions:   0,                // read-only: we make no message modifications

internal/milter/milter_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/emersion/go-milter"
 
@@ -11,10 +12,21 @@ import (
 )
 
 // fakeRecorder captures inserts and can be made to fail, to prove the milter
-// swallows recorder errors and still accepts the message.
+// swallows recorder errors and still accepts the message. By default it reports
+// no configured rate limit, so the level-2 check is inert unless a test sets
+// limits (see fakeRecorder fields).
 type fakeRecorder struct {
-	entries []store.SendLogEntry
-	fail    bool
+	entries  []store.SendLogEntry
+	rejected []store.SendLogEntry
+	fail     bool
+
+	// limits, keyed by "scope|ref", drive the level-2 rate-limit tests. counts
+	// gives the recent-message count returned for a "scope|ref". lookupErr and
+	// countErr force the store errors that must fail open.
+	limits    map[string]store.RateLimit
+	counts    map[string]int64
+	lookupErr error
+	countErr  error
 }
 
 func (f *fakeRecorder) InsertQueued(e store.SendLogEntry) error {
@@ -25,12 +37,32 @@ func (f *fakeRecorder) InsertQueued(e store.SendLogEntry) error {
 	return nil
 }
 
+func (f *fakeRecorder) InsertRejected(e store.SendLogEntry) error {
+	f.rejected = append(f.rejected, e)
+	return nil
+}
+
+func (f *fakeRecorder) RateLimit(scope, ref string) (store.RateLimit, bool, error) {
+	if f.lookupErr != nil {
+		return store.RateLimit{}, false, f.lookupErr
+	}
+	rl, ok := f.limits[scope+"|"+ref]
+	return rl, ok, nil
+}
+
+func (f *fakeRecorder) CountMessages(scope, ref string, _ time.Time) (int64, error) {
+	if f.countErr != nil {
+		return 0, f.countErr
+	}
+	return f.counts[scope+"|"+ref], nil
+}
+
 func mods(kv map[string]string) *milter.Modifier {
 	return &milter.Modifier{Macros: kv}
 }
 
 // drive replays a typical message through one session and returns the recorder.
-func drive(t *testing.T, rec Recorder) *session {
+func drive(t *testing.T, rec Store) *session {
 	t.Helper()
 	s := &session{rec: rec}
 	if _, err := s.Connect("localhost", "tcp4", 0, net.ParseIP("203.0.113.7"), mods(nil)); err != nil {
@@ -141,6 +173,135 @@ func TestBracedMacros(t *testing.T) {
 	}
 }
 
+// limitAt is the client IP the rate-limit tests connect from; the limits below
+// register it so the differentiated check applies.
+const limitIP = "203.0.113.7"
+
+func activeLimit(ips ...string) store.RateLimit {
+	return store.RateLimit{AllowedIPs: ips, MaxMessages: 5, WindowSeconds: 3600}
+}
+
+// mailFrom drives just the connect + MAIL FROM stages and returns the response,
+// which is where the level-2 limit is enforced.
+func mailFrom(t *testing.T, rec Store, ip, from, login string) milter.Response {
+	t.Helper()
+	s := &session{rec: rec}
+	if _, err := s.Connect("h", "tcp4", 0, net.ParseIP(ip), mods(nil)); err != nil {
+		t.Fatalf("Connect: %v", err)
+	}
+	resp, err := s.MailFrom(from, mods(map[string]string{"auth_authen": login}))
+	if err != nil {
+		t.Fatalf("MailFrom: %v", err)
+	}
+	return resp
+}
+
+func TestRateLimitRefusesWhenDomainOverLimit(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit(limitIP),
+		},
+		counts: map[string]int64{store.RateLimitScopeDomain + "|example.com": 5}, // == max
+	}
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespTempFail {
+		t.Fatalf("over-limit MAIL FROM = %v, want TempFail (4xx)", resp)
+	}
+	if len(rec.rejected) != 1 || rec.rejected[0].Domain != "example.com" {
+		t.Fatalf("want one rejected send-log row for example.com, got %+v", rec.rejected)
+	}
+}
+
+func TestRateLimitRefusesWhenAppOverLimit(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeApp + "|app1": activeLimit(limitIP),
+		},
+		counts: map[string]int64{store.RateLimitScopeApp + "|app1": 9}, // over max
+	}
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespTempFail {
+		t.Fatalf("over app limit = %v, want TempFail", resp)
+	}
+}
+
+func TestRateLimitAllowsUnderLimit(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit(limitIP),
+		},
+		counts: map[string]int64{store.RateLimitScopeDomain + "|example.com": 4}, // < max
+	}
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("under limit = %v, want Continue", resp)
+	}
+	if len(rec.rejected) != 0 {
+		t.Fatalf("under limit must not record a rejection: %+v", rec.rejected)
+	}
+}
+
+func TestRateLimitIgnoresUnregisteredIP(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit("198.51.100.1"), // not limitIP
+		},
+		counts: map[string]int64{store.RateLimitScopeDomain + "|example.com": 999},
+	}
+	// The sender's IP is not in the domain's registered set, so level-2 does not
+	// apply even though the count is huge (level-1 anvil would still cover it).
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("unregistered IP = %v, want Continue (level-2 n/a)", resp)
+	}
+}
+
+func TestRateLimitInactiveWithoutCeiling(t *testing.T) {
+	rec := &fakeRecorder{
+		// IP registered but no ceiling/window: an inert draft, must not enforce.
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": {AllowedIPs: []string{limitIP}},
+		},
+		counts: map[string]int64{store.RateLimitScopeDomain + "|example.com": 999},
+	}
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("inactive limit = %v, want Continue", resp)
+	}
+}
+
+func TestRateLimitFailsOpenOnLookupError(t *testing.T) {
+	rec := &fakeRecorder{lookupErr: errors.New("db down")}
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("lookup error = %v, want Continue (fail-open)", resp)
+	}
+}
+
+func TestRateLimitFailsOpenOnCountError(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit(limitIP),
+		},
+		countErr: errors.New("db down"),
+	}
+	if resp := mailFrom(t, rec, limitIP, "a@example.com", "app1"); resp != milter.RespContinue {
+		t.Fatalf("count error = %v, want Continue (fail-open)", resp)
+	}
+}
+
+func TestRateLimitNoIPKeyDoesNotApply(t *testing.T) {
+	rec := &fakeRecorder{
+		limits: map[string]store.RateLimit{
+			store.RateLimitScopeDomain + "|example.com": activeLimit(limitIP),
+		},
+		counts: map[string]int64{store.RateLimitScopeDomain + "|example.com": 999},
+	}
+	// A session with no client IP (e.g. local submission) cannot be keyed.
+	s := &session{rec: rec}
+	resp, err := s.MailFrom("a@example.com", mods(map[string]string{"auth_authen": "app1"}))
+	if err != nil {
+		t.Fatalf("MailFrom: %v", err)
+	}
+	if resp != milter.RespContinue {
+		t.Fatalf("no-IP session = %v, want Continue", resp)
+	}
+}
+
 func TestDomainOf(t *testing.T) {
 	cases := map[string]string{
 		"user@Example.COM": "example.com",

internal/milter/ratelimit.go

@@ -0,0 +1,70 @@
+package milter
+
+import (
+	"log"
+	"time"
+
+	"codeberg.org/mix/selfpost/internal/store"
+)
+
+// overLimit reports whether the message currently being received should be
+// refused under a level-2 differentiated limit (spec 7.4). It checks the
+// domain-level and application-level limits in turn; either being exceeded is
+// enough to refuse.
+//
+// It is deliberately fail-open: any store error, or the absence of a usable
+// limit, is treated as "not over limit" so a malfunction of the level-2 limiter
+// can never block mail — Postfix's level-1 anvil limit (spec 5) remains the
+// backstop, and it does not depend on this milter at all. Only a clean count at
+// or above a configured ceiling returns true.
+func (s *session) overLimit() bool {
+	if s.clientIP == "" {
+		return false // no client IP to key on; level-2 does not apply
+	}
+	checks := []struct{ scope, ref string }{
+		{store.RateLimitScopeDomain, domainOf(s.from)},
+		{store.RateLimitScopeApp, s.login},
+	}
+	for _, c := range checks {
+		if c.ref == "" {
+			continue
+		}
+		rl, ok, err := s.rec.RateLimit(c.scope, c.ref)
+		if err != nil {
+			log.Printf("journal-milter: rate-limit lookup %s %q: %v (fail-open)", c.scope, c.ref, err)
+			continue
+		}
+		// No limit configured, an inert draft, or a client IP outside the
+		// registered set: the differentiated limit does not apply here.
+		if !ok || !rl.Active() || !rl.AllowsIP(s.clientIP) {
+			continue
+		}
+		since := time.Now().Add(-time.Duration(rl.WindowSeconds) * time.Second)
+		n, err := s.rec.CountMessages(c.scope, c.ref, since)
+		if err != nil {
+			log.Printf("journal-milter: rate-limit count %s %q: %v (fail-open)", c.scope, c.ref, err)
+			continue
+		}
+		if n >= int64(rl.MaxMessages) {
+			log.Printf("journal-milter: %s %q over limit: %d/%d in %ds from %s — refusing 4xx",
+				c.scope, c.ref, n, rl.MaxMessages, rl.WindowSeconds, s.clientIP)
+			return true
+		}
+	}
+	return false
+}
+
+// recordRejected writes a send-log row for a message refused by a level-2 limit
+// (spec 7.4, "опционально фиксирует ... для видимости в UI"), so the rejection
+// shows up in the monitoring screen. Only MAIL-stage fields are known; the write
+// is best-effort and never affects the response.
+func (s *session) recordRejected() {
+	err := s.rec.InsertRejected(store.SendLogEntry{
+		Domain:   domainOf(s.from),
+		AppLogin: s.login,
+		From:     s.from,
+	})
+	if err != nil {
+		log.Printf("journal-milter: record rejected %s: %v", s.from, err)
+	}
+}