diff --git a/appview/config.go b/appview/config.go index 51d576e..ab0a99c 100644 --- a/appview/config.go +++ b/appview/config.go @@ -25,8 +25,17 @@ type Config struct { AdminDIDs []string CORSOrigins []string RateLimitEnabled bool - RateLimitRPS float64 - RateLimitBurst int + // RateLimitRPS / RateLimitBurst govern the IP tier (shared across every + // caller behind the same source IP). Keep these conservative — one bad + // actor behind a NAT can otherwise exhaust the bucket for everyone. + RateLimitRPS float64 + RateLimitBurst int + // RateLimitSubRPS / RateLimitSubBurst govern the authenticated tier + // (one bucket per DID). When zero, the rate limiter defaults them to + // 2× the IP values so callers that only set the single pair still get + // meaningful tiering. + RateLimitSubRPS float64 + RateLimitSubBurst int } func (c Config) Validate() error { @@ -65,6 +74,12 @@ func (c Config) Validate() error { if c.RateLimitEnabled && c.RateLimitBurst <= 0 { return fmt.Errorf("rate limit burst must be positive when rate limiting is enabled") } + if c.RateLimitEnabled && c.RateLimitSubRPS < 0 { + return fmt.Errorf("rate limit sub rps must be >= 0 (0 defaults to 2× ip rps)") + } + if c.RateLimitEnabled && c.RateLimitSubBurst < 0 { + return fmt.Errorf("rate limit sub burst must be >= 0 (0 defaults to 2× ip burst)") + } for _, did := range c.AdminDIDs { if !strings.HasPrefix(did, "did:") { return fmt.Errorf("admin DID %q must start with \"did:\"", did) diff --git a/appview/httpmw/ratelimit.go b/appview/httpmw/ratelimit.go index 998e34a..c74133c 100644 --- a/appview/httpmw/ratelimit.go +++ b/appview/httpmw/ratelimit.go @@ -2,6 +2,7 @@ package httpmw import ( "fmt" + "log/slog" "net/http" "strings" "sync" @@ -12,11 +13,32 @@ import ( "tangled.org/sparrowtek.com/effem-AppView/appview/metrics" ) +// RateLimiterConfig declares two independent bucket tiers: +// +// - Sub tier (authenticated callers, identifiers prefixed "sub:"): one +// bucket per DID. Higher limits because the principal is known, +// bannable, and already passed auth. +// - IP tier (unauthenticated or IP-keyed callers, "ip:"): one +// bucket per source IP. Shared across every anonymous caller behind +// the same IP, so the default must be conservative. +// +// SubRPS / SubBurst default to 2× the IP values when left at zero. That +// keeps single-knob deployments (only IP values set) working with a +// reasonable sub tier, while still letting operators override both. type RateLimiterConfig struct { Enabled bool - RPS float64 - Burst int BucketTTL time.Duration + + SubRPS float64 + SubBurst int + + IPRPS float64 + IPBurst int +} + +type tier struct { + rps rate.Limit + burst int } type limiterEntry struct { @@ -26,9 +48,10 @@ type limiterEntry struct { type PrincipalRateLimiter struct { enabled bool - rps rate.Limit - burst int + subTier tier + ipTier tier bucketTTL time.Duration + logger *slog.Logger mu sync.Mutex entries map[string]*limiterEntry @@ -36,24 +59,37 @@ type PrincipalRateLimiter struct { sweepAfter time.Duration } -func NewPrincipalRateLimiter(cfg RateLimiterConfig) (*PrincipalRateLimiter, error) { +// NewPrincipalRateLimiter returns a limiter configured with the given +// tiers. A nil logger is replaced with slog.Default so callers in tests do +// not have to wire one up. +func NewPrincipalRateLimiter(cfg RateLimiterConfig, logger *slog.Logger) (*PrincipalRateLimiter, error) { + if logger == nil { + logger = slog.Default() + } if cfg.BucketTTL <= 0 { cfg.BucketTTL = 5 * time.Minute } if cfg.Enabled { - if cfg.RPS <= 0 { - return nil, fmt.Errorf("rate limit rps must be positive") + if cfg.IPRPS <= 0 { + return nil, fmt.Errorf("ip tier rps must be positive") + } + if cfg.IPBurst <= 0 { + return nil, fmt.Errorf("ip tier burst must be positive") } - if cfg.Burst <= 0 { - return nil, fmt.Errorf("rate limit burst must be positive") + if cfg.SubRPS <= 0 { + cfg.SubRPS = cfg.IPRPS * 2 + } + if cfg.SubBurst <= 0 { + cfg.SubBurst = cfg.IPBurst * 2 } } return &PrincipalRateLimiter{ enabled: cfg.Enabled, - rps: rate.Limit(cfg.RPS), - burst: cfg.Burst, + subTier: tier{rps: rate.Limit(cfg.SubRPS), burst: cfg.SubBurst}, + ipTier: tier{rps: rate.Limit(cfg.IPRPS), burst: cfg.IPBurst}, bucketTTL: cfg.BucketTTL, + logger: logger, entries: map[string]*limiterEntry{}, lastSweep: time.Now(), sweepAfter: time.Minute, @@ -72,6 +108,18 @@ func (rl *PrincipalRateLimiter) Middleware(next echo.HandlerFunc) echo.HandlerFu } key := rl.identifier(c) + if key == "" { + // No principal and no detectable IP. With a properly-configured + // reverse proxy (see Phase 0 Step 3) this branch is unreachable; + // firing it means we'd otherwise lump every such caller into one + // shared bucket. Reject the request outright instead. + metrics.IncRateLimitRejection("unknown") + return c.JSON(http.StatusBadRequest, map[string]string{ + "error": "InvalidRequest", + "message": "could not identify caller; reverse proxy may be misconfigured", + }) + } + now := time.Now() if !rl.allow(key, now) { metrics.IncRateLimitRejection(identifierKind(key)) @@ -90,10 +138,10 @@ func (rl *PrincipalRateLimiter) Middleware(next echo.HandlerFunc) echo.HandlerFu // Keep the returned set small — it ends up as a Prometheus label value. func identifierKind(id string) string { switch { + case id == "": + return "unknown" case strings.HasPrefix(id, "sub:"): return "sub" - case id == "ip:unknown": - return "unknown" case strings.HasPrefix(id, "ip:"): return "ip" default: @@ -110,7 +158,11 @@ func (rl *PrincipalRateLimiter) identifier(c echo.Context) string { if ip := c.RealIP(); ip != "" { return "ip:" + ip } - return "ip:unknown" + rl.logger.Warn("rate limiter: no principal and no IP detected — rejecting", + "path", c.Path(), + "method", c.Request().Method, + ) + return "" } func (rl *PrincipalRateLimiter) allow(key string, now time.Time) bool { @@ -124,13 +176,25 @@ func (rl *PrincipalRateLimiter) allow(key string, now time.Time) bool { entry, ok := rl.entries[key] if !ok { - entry = &limiterEntry{limiter: rate.NewLimiter(rl.rps, rl.burst)} + t := rl.tierFor(key) + entry = &limiterEntry{limiter: rate.NewLimiter(t.rps, t.burst)} rl.entries[key] = entry } entry.lastSeen = now return entry.limiter.Allow() } +// tierFor selects the bucket config for a given identifier. Anything that +// isn't explicitly "sub:" falls back to the IP tier — a well-formed +// identifier always has a known prefix, so the fallback exists only to +// keep a malformed key from panicking at runtime. +func (rl *PrincipalRateLimiter) tierFor(id string) tier { + if strings.HasPrefix(id, "sub:") { + return rl.subTier + } + return rl.ipTier +} + func (rl *PrincipalRateLimiter) sweepStaleLocked(now time.Time) { for key, entry := range rl.entries { if now.Sub(entry.lastSeen) > rl.bucketTTL { diff --git a/appview/httpmw/ratelimit_test.go b/appview/httpmw/ratelimit_test.go index 0c83897..87e5902 100644 --- a/appview/httpmw/ratelimit_test.go +++ b/appview/httpmw/ratelimit_test.go @@ -1,8 +1,11 @@ package httpmw import ( + "bytes" + "log/slog" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -21,10 +24,12 @@ func TestPrincipalRateLimiter(t *testing.T) { ) rl, err := NewPrincipalRateLimiter(RateLimiterConfig{ Enabled: true, - RPS: 1, - Burst: 1, + IPRPS: 1, + IPBurst: 1, + SubRPS: 1, + SubBurst: 1, BucketTTL: time.Minute, - }) + }, nil) if err != nil { t.Fatalf("failed to create rate limiter: %v", err) } @@ -51,6 +56,126 @@ func TestPrincipalRateLimiter(t *testing.T) { } } +// TestSubTierIsolatedFromIPTier verifies the two tiers are independent +// buckets. An exhausted IP bucket must not lock an authenticated caller +// out; an exhausted sub bucket must not lock IP callers out. +func TestSubTierIsolatedFromIPTier(t *testing.T) { + rl, err := NewPrincipalRateLimiter(RateLimiterConfig{ + Enabled: true, + IPRPS: 1, + IPBurst: 1, + SubRPS: 1, + SubBurst: 1, + BucketTTL: time.Minute, + }, nil) + if err != nil { + t.Fatalf("create: %v", err) + } + + now := time.Now() + if !rl.allow("ip:1.2.3.4", now) { + t.Fatal("ip first request should pass") + } + if rl.allow("ip:1.2.3.4", now) { + t.Fatal("ip second request should be rejected") + } + // The sub bucket must still have capacity. + if !rl.allow("sub:did:plc:alice", now) { + t.Fatal("sub bucket should not be affected by exhausted ip bucket") + } +} + +// TestSubBurstDefaultsToDoubleIP confirms the "leave sub fields zero" path +// produces a 2× IP default. This is the shape most programmatic callers +// get when they forget to wire the sub-tier knobs. +func TestSubBurstDefaultsToDoubleIP(t *testing.T) { + rl, err := NewPrincipalRateLimiter(RateLimiterConfig{ + Enabled: true, + IPRPS: 1, + IPBurst: 3, + }, nil) + if err != nil { + t.Fatalf("create: %v", err) + } + + now := time.Now() + // Sub burst should be 6 (2×3). Confirm six successive allows, then a reject. + for i := 0; i < 6; i++ { + if !rl.allow("sub:did:plc:alice", now) { + t.Fatalf("sub allow %d should pass (expected 2×IPBurst=6)", i+1) + } + } + if rl.allow("sub:did:plc:alice", now) { + t.Fatal("7th sub allow should be rejected") + } +} + +// TestMiddlewareRejectsUnidentifiedCaller: if the identifier function +// returns empty (no principal, no RealIP), middleware must emit 400 with +// InvalidRequest, not 429. This is the misconfigured-proxy canary. +func TestMiddlewareRejectsUnidentifiedCaller(t *testing.T) { + var logBuf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelWarn})) + rl, err := NewPrincipalRateLimiter(RateLimiterConfig{ + Enabled: true, + IPRPS: 5, + IPBurst: 20, + }, logger) + if err != nil { + t.Fatalf("create: %v", err) + } + + e := echo.New() + // Force c.RealIP() to return empty by trusting nothing (no loopback, + // no private) and not setting X-Forwarded-For or a remote addr we'd + // normally trust. We also skip Authentication middleware so no + // principal is set in context. + e.IPExtractor = func(r *http.Request) string { return "" } + e.Use(rl.Middleware) + e.GET("/xrpc/anon", func(c echo.Context) error { + return c.NoContent(http.StatusOK) + }) + + req := httptest.NewRequest(http.MethodGet, "/xrpc/anon", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400 InvalidRequest, got %d: %s", rec.Code, rec.Body.String()) + } + if !strings.Contains(logBuf.String(), "no principal and no IP detected") { + t.Fatalf("expected warning log, got:\n%s", logBuf.String()) + } +} + +// TestIdentifierKindLabeling locks the metric-label contract — adding a +// new kind requires either updating the ALERTS / dashboards or knowing +// that existing tooling sees the new value. +func TestIdentifierKindLabeling(t *testing.T) { + cases := map[string]string{ + "": "unknown", + "sub:did:plc:alice": "sub", + "sub:*": "sub", + "ip:1.2.3.4": "ip", + "ip:unknown": "ip", + "garbage-without-sep": "other", + } + for id, want := range cases { + if got := identifierKind(id); got != want { + t.Errorf("identifierKind(%q): want %q, got %q", id, want, got) + } + } +} + +func TestRateLimiterRequiresPositiveIPTier(t *testing.T) { + if _, err := NewPrincipalRateLimiter(RateLimiterConfig{Enabled: true}, nil); err == nil { + t.Fatal("expected error when ip tier unset") + } + if _, err := NewPrincipalRateLimiter(RateLimiterConfig{Enabled: true, IPRPS: 1}, nil); err == nil { + t.Fatal("expected error when ip burst unset") + } +} + func doRequest(t *testing.T, e *echo.Echo, token string) int { t.Helper() req := httptest.NewRequest(http.MethodGet, "/xrpc/test", nil) diff --git a/appview/server.go b/appview/server.go index ac7ad64..54a18d8 100644 --- a/appview/server.go +++ b/appview/server.go @@ -69,10 +69,12 @@ func NewServer(cfg Config) (*Server, error) { WithAdminDIDs(cfg.AdminDIDs) rateLimiter, err := httpmw.NewPrincipalRateLimiter(httpmw.RateLimiterConfig{ Enabled: cfg.RateLimitEnabled, - RPS: cfg.RateLimitRPS, - Burst: cfg.RateLimitBurst, + IPRPS: cfg.RateLimitRPS, + IPBurst: cfg.RateLimitBurst, + SubRPS: cfg.RateLimitSubRPS, + SubBurst: cfg.RateLimitSubBurst, BucketTTL: 5 * time.Minute, - }) + }, logger) if err != nil { return nil, fmt.Errorf("invalid rate limiter configuration: %w", err) } diff --git a/cmd/effem-appview/main.go b/cmd/effem-appview/main.go index 75db3f7..677d315 100644 --- a/cmd/effem-appview/main.go +++ b/cmd/effem-appview/main.go @@ -100,13 +100,25 @@ func main() { Name: "rate-limit-rps", Value: 5, EnvVars: []string{"EFFEM_RATE_LIMIT_RPS"}, - Usage: "Per-principal steady-state request rate per second", + Usage: "Per-IP steady-state request rate per second (ip tier)", }, &cli.IntFlag{ Name: "rate-limit-burst", Value: 20, EnvVars: []string{"EFFEM_RATE_LIMIT_BURST"}, - Usage: "Per-principal burst request capacity", + Usage: "Per-IP burst request capacity (ip tier)", + }, + &cli.Float64Flag{ + Name: "rate-limit-sub-rps", + Value: 10, + EnvVars: []string{"EFFEM_RATE_LIMIT_SUB_RPS"}, + Usage: "Per-DID steady-state request rate per second (sub tier)", + }, + &cli.IntFlag{ + Name: "rate-limit-sub-burst", + Value: 40, + EnvVars: []string{"EFFEM_RATE_LIMIT_SUB_BURST"}, + Usage: "Per-DID burst request capacity (sub tier)", }, }, Action: run, @@ -148,6 +160,8 @@ func run(cctx *cli.Context) error { RateLimitEnabled: cctx.Bool("rate-limit-enabled"), RateLimitRPS: cctx.Float64("rate-limit-rps"), RateLimitBurst: cctx.Int("rate-limit-burst"), + RateLimitSubRPS: cctx.Float64("rate-limit-sub-rps"), + RateLimitSubBurst: cctx.Int("rate-limit-sub-burst"), } srv, err := appview.NewServer(cfg)