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>
This commit is contained in:
@@ -0,0 +1,214 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Rate-limit scopes (spec 7.4). A level-2 limit is attached either to a domain
|
||||
// (counted across all its applications and IPs) or to a single application.
|
||||
const (
|
||||
RateLimitScopeDomain = "domain"
|
||||
RateLimitScopeApp = "application"
|
||||
)
|
||||
|
||||
// RateLimit is a differentiated level-2 rate limit (spec 7.4): an optional set
|
||||
// of expected client IPs plus a message ceiling over a sliding window, attached
|
||||
// to a domain or an application. It is enforced in the journal-milter; level 1
|
||||
// (Postfix anvil, spec 5) is the IP backstop that always applies even when this
|
||||
// is absent or the milter is down.
|
||||
//
|
||||
// Both the IP binding and the ceiling are optional in the schema, but a limit is
|
||||
// only enforced when it is Active(): the design deliberately allows an admin to
|
||||
// leave the IP binding empty for apps that send from changing IPs, in which case
|
||||
// only level 1 protects them (spec 7.4's caveat).
|
||||
type RateLimit struct {
|
||||
Scope string
|
||||
RefID int64
|
||||
AllowedIPs []string // canonical client IPs this limit applies to
|
||||
MaxMessages int
|
||||
WindowSeconds int
|
||||
}
|
||||
|
||||
// Active reports whether the limit is fully configured and should be enforced.
|
||||
// A missing IP binding, ceiling or window leaves the differentiated limit inert
|
||||
// (spec 7.4): the IP binding is what scopes the limit to a known sender.
|
||||
func (r RateLimit) Active() bool {
|
||||
return len(r.AllowedIPs) > 0 && r.MaxMessages > 0 && r.WindowSeconds > 0
|
||||
}
|
||||
|
||||
// AllowsIP reports whether ip is one of the limit's registered client IPs. The
|
||||
// comparison parses both sides so equivalent textual forms of the same address
|
||||
// match; a client IP outside the list means the differentiated limit does not
|
||||
// apply to it (level 1 still does).
|
||||
func (r RateLimit) AllowsIP(ip string) bool {
|
||||
c := net.ParseIP(ip)
|
||||
if c == nil {
|
||||
return false
|
||||
}
|
||||
for _, a := range r.AllowedIPs {
|
||||
if p := net.ParseIP(a); p != nil && p.Equal(c) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// GetRateLimit loads the level-2 limit configured for a domain or application by
|
||||
// its id, for the panel's edit form. ok is false when none is configured.
|
||||
func (s *Store) GetRateLimit(scope string, refID int64) (RateLimit, bool, error) {
|
||||
row := s.db.QueryRow(
|
||||
`SELECT allowed_ips, max_messages, window_seconds
|
||||
FROM rate_limits WHERE scope = ? AND ref_id = ?`, scope, refID)
|
||||
rl, err := scanRateLimit(row)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return RateLimit{Scope: scope, RefID: refID}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return RateLimit{}, false, fmt.Errorf("get rate limit: %w", err)
|
||||
}
|
||||
rl.Scope, rl.RefID = scope, refID
|
||||
return rl, true, nil
|
||||
}
|
||||
|
||||
// SetRateLimit upserts the level-2 limit for a domain or application. The caller
|
||||
// (panel) has already validated the IPs and numbers (spec 7.6.2); values are
|
||||
// stored via bound parameters and read back live by the milter.
|
||||
func (s *Store) SetRateLimit(rl RateLimit) error {
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO rate_limits (scope, ref_id, allowed_ips, max_messages, window_seconds)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
ON CONFLICT(scope, ref_id) DO UPDATE SET
|
||||
allowed_ips = excluded.allowed_ips,
|
||||
max_messages = excluded.max_messages,
|
||||
window_seconds = excluded.window_seconds`,
|
||||
rl.Scope, rl.RefID, strings.Join(rl.AllowedIPs, ","), rl.MaxMessages, rl.WindowSeconds,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("set rate limit: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteRateLimit removes the level-2 limit for a domain or application, so the
|
||||
// admin can clear it and fall back to level 1 only.
|
||||
func (s *Store) DeleteRateLimit(scope string, refID int64) error {
|
||||
if _, err := s.db.Exec(`DELETE FROM rate_limits WHERE scope = ? AND ref_id = ?`, scope, refID); err != nil {
|
||||
return fmt.Errorf("delete rate limit: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteRateLimitsForDomain removes the domain's own limit and the limits of all
|
||||
// its applications in one statement. It is called on domain deletion, before the
|
||||
// application rows are cascade-deleted, to avoid leaving orphan limit rows.
|
||||
func (s *Store) DeleteRateLimitsForDomain(domainID int64) error {
|
||||
_, err := s.db.Exec(
|
||||
`DELETE FROM rate_limits
|
||||
WHERE (scope = 'domain' AND ref_id = ?)
|
||||
OR (scope = 'application' AND ref_id IN (SELECT id FROM applications WHERE domain_id = ?))`,
|
||||
domainID, domainID,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete domain rate limits: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RateLimit resolves the level-2 limit that applies to a message, keyed by the
|
||||
// human-readable reference the milter has on the receive path: the sending
|
||||
// domain name (scope "domain") or the SASL login (scope "application"). ok is
|
||||
// false when no limit is configured for that reference.
|
||||
func (s *Store) RateLimit(scope, ref string) (RateLimit, bool, error) {
|
||||
var query string
|
||||
switch scope {
|
||||
case RateLimitScopeDomain:
|
||||
query = `SELECT rl.allowed_ips, rl.max_messages, rl.window_seconds
|
||||
FROM rate_limits rl JOIN domains d ON d.id = rl.ref_id
|
||||
WHERE rl.scope = 'domain' AND d.name = ?`
|
||||
case RateLimitScopeApp:
|
||||
query = `SELECT rl.allowed_ips, rl.max_messages, rl.window_seconds
|
||||
FROM rate_limits rl JOIN applications a ON a.id = rl.ref_id
|
||||
WHERE rl.scope = 'application' AND a.login = ?`
|
||||
default:
|
||||
return RateLimit{}, false, fmt.Errorf("unknown rate-limit scope %q", scope)
|
||||
}
|
||||
rl, err := scanRateLimit(s.db.QueryRow(query, ref))
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return RateLimit{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return RateLimit{}, false, fmt.Errorf("rate limit for %s %q: %w", scope, ref, err)
|
||||
}
|
||||
rl.Scope = scope
|
||||
return rl, true, nil
|
||||
}
|
||||
|
||||
// CountMessages returns how many distinct messages the reference (a domain name
|
||||
// or an application login) has queued since t, for the level-2 sliding window
|
||||
// (spec 7.4). It counts distinct queue-ids — one message with many recipients is
|
||||
// one message, matching level 1's per-message semantics — and excludes rows that
|
||||
// were themselves rejected by a limit (they were never sent). It reuses the send
|
||||
// log the journal already writes (spec 7.4: "переиспользует данные журнала").
|
||||
func (s *Store) CountMessages(scope, ref string, since time.Time) (int64, error) {
|
||||
var column string
|
||||
switch scope {
|
||||
case RateLimitScopeDomain:
|
||||
column = "domain"
|
||||
case RateLimitScopeApp:
|
||||
column = "app_login"
|
||||
default:
|
||||
return 0, fmt.Errorf("unknown rate-limit scope %q", scope)
|
||||
}
|
||||
var n int64
|
||||
// created_at is stored as RFC3339 UTC, so a lexical comparison against the
|
||||
// same format is chronologically correct (as in DeleteSendLogBefore).
|
||||
err := s.db.QueryRow(
|
||||
`SELECT COUNT(DISTINCT queue_id) FROM send_log
|
||||
WHERE `+column+` = ? AND status != ? AND created_at >= ?`,
|
||||
ref, StatusRejected, since.UTC().Format(time.RFC3339),
|
||||
).Scan(&n)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("count messages for %s %q: %w", scope, ref, err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// scanRateLimit reads the three stored columns, tolerating NULL numeric columns
|
||||
// (an IP-only draft) by leaving the corresponding field zero, which makes the
|
||||
// limit inert via Active().
|
||||
func scanRateLimit(r scanRow) (RateLimit, error) {
|
||||
var (
|
||||
ips sql.NullString
|
||||
maxMsgs sql.NullInt64
|
||||
windowSecs sql.NullInt64
|
||||
)
|
||||
if err := r.Scan(&ips, &maxMsgs, &windowSecs); err != nil {
|
||||
return RateLimit{}, err
|
||||
}
|
||||
return RateLimit{
|
||||
AllowedIPs: splitIPs(ips.String),
|
||||
MaxMessages: int(maxMsgs.Int64),
|
||||
WindowSeconds: int(windowSecs.Int64),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// splitIPs parses the comma-separated storage form back into a slice, dropping
|
||||
// empties so an empty column yields nil (an inactive limit).
|
||||
func splitIPs(s string) []string {
|
||||
if strings.TrimSpace(s) == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(s, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, p := range parts {
|
||||
if p = strings.TrimSpace(p); p != "" {
|
||||
out = append(out, p)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,178 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRateLimitSetGetDelete(t *testing.T) {
|
||||
st := openTestStore(t)
|
||||
d, err := st.AddDomain("example.com", "selfpost")
|
||||
if err != nil {
|
||||
t.Fatalf("AddDomain: %v", err)
|
||||
}
|
||||
|
||||
if _, ok, err := st.GetRateLimit(RateLimitScopeDomain, d.ID); err != nil || ok {
|
||||
t.Fatalf("GetRateLimit on empty: ok=%v err=%v", ok, err)
|
||||
}
|
||||
|
||||
want := RateLimit{
|
||||
Scope: RateLimitScopeDomain,
|
||||
RefID: d.ID,
|
||||
AllowedIPs: []string{"203.0.113.1", "203.0.113.2"},
|
||||
MaxMessages: 100,
|
||||
WindowSeconds: 3600,
|
||||
}
|
||||
if err := st.SetRateLimit(want); err != nil {
|
||||
t.Fatalf("SetRateLimit: %v", err)
|
||||
}
|
||||
|
||||
got, ok, err := st.GetRateLimit(RateLimitScopeDomain, d.ID)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("GetRateLimit after set: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if got.MaxMessages != 100 || got.WindowSeconds != 3600 || len(got.AllowedIPs) != 2 ||
|
||||
got.AllowedIPs[0] != "203.0.113.1" || got.AllowedIPs[1] != "203.0.113.2" {
|
||||
t.Fatalf("roundtrip mismatch: %+v", got)
|
||||
}
|
||||
|
||||
// Upsert replaces in place (UNIQUE(scope, ref_id)).
|
||||
want.MaxMessages = 5
|
||||
want.AllowedIPs = []string{"198.51.100.9"}
|
||||
if err := st.SetRateLimit(want); err != nil {
|
||||
t.Fatalf("SetRateLimit upsert: %v", err)
|
||||
}
|
||||
got, _, _ = st.GetRateLimit(RateLimitScopeDomain, d.ID)
|
||||
if got.MaxMessages != 5 || len(got.AllowedIPs) != 1 || got.AllowedIPs[0] != "198.51.100.9" {
|
||||
t.Fatalf("upsert did not replace: %+v", got)
|
||||
}
|
||||
|
||||
if err := st.DeleteRateLimit(RateLimitScopeDomain, d.ID); err != nil {
|
||||
t.Fatalf("DeleteRateLimit: %v", err)
|
||||
}
|
||||
if _, ok, _ := st.GetRateLimit(RateLimitScopeDomain, d.ID); ok {
|
||||
t.Fatalf("limit still present after delete")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRateLimitByNameAndLogin(t *testing.T) {
|
||||
st := openTestStore(t)
|
||||
d, _ := st.AddDomain("example.com", "selfpost")
|
||||
a, err := st.AddApplication(d.ID, "app1", AddressModeWildcard, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("AddApplication: %v", err)
|
||||
}
|
||||
|
||||
if err := st.SetRateLimit(RateLimit{Scope: RateLimitScopeDomain, RefID: d.ID, AllowedIPs: []string{"203.0.113.1"}, MaxMessages: 10, WindowSeconds: 60}); err != nil {
|
||||
t.Fatalf("set domain limit: %v", err)
|
||||
}
|
||||
if err := st.SetRateLimit(RateLimit{Scope: RateLimitScopeApp, RefID: a.ID, AllowedIPs: []string{"203.0.113.2"}, MaxMessages: 3, WindowSeconds: 60}); err != nil {
|
||||
t.Fatalf("set app limit: %v", err)
|
||||
}
|
||||
|
||||
// The milter resolves limits by domain name and by SASL login.
|
||||
rl, ok, err := st.RateLimit(RateLimitScopeDomain, "example.com")
|
||||
if err != nil || !ok || rl.MaxMessages != 10 {
|
||||
t.Fatalf("RateLimit domain: ok=%v err=%v rl=%+v", ok, err, rl)
|
||||
}
|
||||
rl, ok, err = st.RateLimit(RateLimitScopeApp, "app1")
|
||||
if err != nil || !ok || rl.MaxMessages != 3 {
|
||||
t.Fatalf("RateLimit app: ok=%v err=%v rl=%+v", ok, err, rl)
|
||||
}
|
||||
if _, ok, _ := st.RateLimit(RateLimitScopeDomain, "unknown.example"); ok {
|
||||
t.Fatalf("RateLimit for unknown domain should be not-ok")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCountMessagesDistinctAndWindowed(t *testing.T) {
|
||||
st := openTestStore(t)
|
||||
|
||||
// Two recipients share a queue-id → one message. A second message → two.
|
||||
for _, to := range []string{"a@x.net", "b@x.net"} {
|
||||
if err := st.InsertQueued(SendLogEntry{QueueID: "Q1", Domain: "example.com", AppLogin: "app1", To: to}); err != nil {
|
||||
t.Fatalf("InsertQueued: %v", err)
|
||||
}
|
||||
}
|
||||
if err := st.InsertQueued(SendLogEntry{QueueID: "Q2", Domain: "example.com", AppLogin: "app1", To: "c@x.net"}); err != nil {
|
||||
t.Fatalf("InsertQueued: %v", err)
|
||||
}
|
||||
// A rejected row must not count toward the window.
|
||||
if err := st.InsertRejected(SendLogEntry{Domain: "example.com", AppLogin: "app1", From: "s@example.com"}); err != nil {
|
||||
t.Fatalf("InsertRejected: %v", err)
|
||||
}
|
||||
|
||||
n, err := st.CountMessages(RateLimitScopeDomain, "example.com", time.Now().Add(-time.Hour))
|
||||
if err != nil {
|
||||
t.Fatalf("CountMessages: %v", err)
|
||||
}
|
||||
if n != 2 {
|
||||
t.Fatalf("distinct-message count = %d, want 2 (two queue-ids, rejected excluded)", n)
|
||||
}
|
||||
n, _ = st.CountMessages(RateLimitScopeApp, "app1", time.Now().Add(-time.Hour))
|
||||
if n != 2 {
|
||||
t.Fatalf("app count = %d, want 2", n)
|
||||
}
|
||||
|
||||
// Backdate Q1 beyond the window: only Q2 remains inside a 30-minute window.
|
||||
old := time.Now().UTC().Add(-2 * time.Hour).Format(time.RFC3339)
|
||||
if _, err := st.db.Exec(`UPDATE send_log SET created_at = ? WHERE queue_id = 'Q1'`, old); err != nil {
|
||||
t.Fatalf("backdate: %v", err)
|
||||
}
|
||||
n, _ = st.CountMessages(RateLimitScopeDomain, "example.com", time.Now().Add(-30*time.Minute))
|
||||
if n != 1 {
|
||||
t.Fatalf("windowed count = %d, want 1 (Q1 aged out)", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteRateLimitsForDomain(t *testing.T) {
|
||||
st := openTestStore(t)
|
||||
d, _ := st.AddDomain("example.com", "selfpost")
|
||||
a, _ := st.AddApplication(d.ID, "app1", AddressModeWildcard, nil)
|
||||
other, _ := st.AddDomain("other.example", "selfpost")
|
||||
|
||||
_ = st.SetRateLimit(RateLimit{Scope: RateLimitScopeDomain, RefID: d.ID, AllowedIPs: []string{"203.0.113.1"}, MaxMessages: 10, WindowSeconds: 60})
|
||||
_ = st.SetRateLimit(RateLimit{Scope: RateLimitScopeApp, RefID: a.ID, AllowedIPs: []string{"203.0.113.2"}, MaxMessages: 3, WindowSeconds: 60})
|
||||
_ = st.SetRateLimit(RateLimit{Scope: RateLimitScopeDomain, RefID: other.ID, AllowedIPs: []string{"203.0.113.9"}, MaxMessages: 1, WindowSeconds: 60})
|
||||
|
||||
if err := st.DeleteRateLimitsForDomain(d.ID); err != nil {
|
||||
t.Fatalf("DeleteRateLimitsForDomain: %v", err)
|
||||
}
|
||||
if _, ok, _ := st.GetRateLimit(RateLimitScopeDomain, d.ID); ok {
|
||||
t.Fatalf("domain limit survived")
|
||||
}
|
||||
if _, ok, _ := st.GetRateLimit(RateLimitScopeApp, a.ID); ok {
|
||||
t.Fatalf("application limit survived")
|
||||
}
|
||||
// The unrelated domain's limit is untouched.
|
||||
if _, ok, _ := st.GetRateLimit(RateLimitScopeDomain, other.ID); !ok {
|
||||
t.Fatalf("unrelated domain limit was deleted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRateLimitActiveAndAllowsIP(t *testing.T) {
|
||||
inactive := []RateLimit{
|
||||
{},
|
||||
{AllowedIPs: []string{"203.0.113.1"}}, // no ceiling
|
||||
{AllowedIPs: []string{"203.0.113.1"}, MaxMessages: 5}, // no window
|
||||
{MaxMessages: 5, WindowSeconds: 60}, // no IPs
|
||||
}
|
||||
for i, rl := range inactive {
|
||||
if rl.Active() {
|
||||
t.Fatalf("case %d: %+v should be inactive", i, rl)
|
||||
}
|
||||
}
|
||||
active := RateLimit{AllowedIPs: []string{"203.0.113.1", "2001:db8::1"}, MaxMessages: 5, WindowSeconds: 60}
|
||||
if !active.Active() {
|
||||
t.Fatalf("should be active: %+v", active)
|
||||
}
|
||||
if !active.AllowsIP("203.0.113.1") || !active.AllowsIP("2001:db8::1") {
|
||||
t.Fatalf("registered IPs should match")
|
||||
}
|
||||
// Equivalent textual form of the IPv6 address must still match.
|
||||
if !active.AllowsIP("2001:0db8:0000:0000:0000:0000:0000:0001") {
|
||||
t.Fatalf("expanded IPv6 form should match")
|
||||
}
|
||||
if active.AllowsIP("198.51.100.7") || active.AllowsIP("not-an-ip") || active.AllowsIP("") {
|
||||
t.Fatalf("unregistered/invalid IPs must not match")
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,10 @@ const (
|
||||
StatusSent = "sent"
|
||||
StatusDeferred = "deferred"
|
||||
StatusBounced = "bounced"
|
||||
// StatusRejected marks a message the journal-milter refused with a 4xx under
|
||||
// a level-2 rate limit (spec 7.4). Such a row never gets a queue-id and is
|
||||
// excluded from the level-2 message count (it was never sent).
|
||||
StatusRejected = "rejected"
|
||||
)
|
||||
|
||||
// SendLogEntry is a single queued send-log row. The journal-milter creates one
|
||||
@@ -46,6 +50,24 @@ func (s *Store) InsertQueued(e SendLogEntry) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// InsertRejected records a message the journal-milter refused under a level-2
|
||||
// rate limit (spec 7.4), so the rejection is visible in the send-log UI. Only
|
||||
// the fields known at MAIL FROM are set (domain, sender, app login); there is no
|
||||
// queue-id or recipient because the message was rejected before it was queued.
|
||||
func (s *Store) InsertRejected(e SendLogEntry) error {
|
||||
now := time.Now().UTC().Format(time.RFC3339)
|
||||
_, err := s.db.Exec(
|
||||
`INSERT INTO send_log
|
||||
(queue_id, domain, app_login, from_addr, to_addr, subject, status, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
e.QueueID, e.Domain, e.AppLogin, e.From, e.To, e.Subject, StatusRejected, now, now,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("insert rejected send_log: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateStatus advances the delivery status of the send-log rows matching a
|
||||
// (queue-id, recipient) pair, which the log-tailer parses out of mail.log.
|
||||
// Recipient matching is case-insensitive because Postfix may normalise address
|
||||
|
||||
Reference in New Issue
Block a user