feat(shield)!: GCRA rate limiting with pluggable keys, limits, and async fleet reconciliation

.github/workflows/ci.yml

@@ -18,6 +18,17 @@ jobs:
   quality-checks:
     name: Quality Checks (${{ matrix.backend.name }})
     runs-on: ubuntu-latest
+    services:
+      # Shared store for the rate-limit cross-instance reconciliation test.
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
     strategy:
       fail-fast: false
       matrix:
@@ -53,6 +64,9 @@ jobs:
         run: cargo build --release ${{ matrix.backend.flags }}
 
       - name: Test
+        env:
+          # Exercises the rate-limit reconciliation against the Redis service.
+          SHIELD_REDIS_TEST_URL: redis://127.0.0.1:6379/
         run: cargo test ${{ matrix.backend.flags }}
 
       - name: Publish dry-run
@@ -69,6 +83,6 @@ jobs:
       # Scans Cargo.lock for RustSec advisories; honors the ignore list in
       # deny.toml. Third-party action: pinned to a commit SHA (dependabot
       # keeps it fresh).
-      - uses: EmbarkStudios/cargo-deny-action@bb137d7af7e4fb67e5f82a49c4fce4fad40782fe # v2.0.20
+      - uses: EmbarkStudios/cargo-deny-action@3c6349835b2b7b196a839186cb8b78e02f7b5f25 # v2.1.1
         with:
           command: check advisories

Cargo.toml

@@ -73,7 +73,7 @@ dashmap = "6"
 async-trait = "0.1"
 ipnet = "2"
 # Optional shared rate-limit store for multi-instance deployments.
-redis = { version = "1.2", features = ["tokio-comp"], optional = true }
+redis = { version = "1.2", features = ["tokio-comp", "connection-manager"], optional = true }
 
 # Auth (JWT validation, route policies).
 # Crypto backend is selected via this crate's `rust_crypto` / `aws_lc_rs`

README.md

@@ -25,7 +25,7 @@ Works with **any** gRPC service via proto descriptor files. No code generation,
 - **Health endpoints** `/health/live`, `/health/ready` (upstream gRPC health probe), `/health/startup`
 - **Prometheus metrics** at `/metrics`
 - **CORS** with a configurable origin allow-list
-- **Rate limiting (Shield)**: per-client endpoint classes + per-identifier limits, in-process by default or Redis-backed (feature `redis`) for multi-instance
+- **Rate limiting (Shield)**: local GCRA shaper (no blocking latency) keyed by client IP, header, or validated JWT claim; named limit tiers as config data; optional async cross-instance reconciliation for an approximate fleet-wide limit (requires both the `redis` feature and a configured `sync` block)
 - **JWT auth**: validate `Bearer` tokens via an Ed25519 PEM key or JWKS auto-discovery, enforce per-route `require_auth` / `required_roles`, and forward claims as headers
 - **OIDC discovery**: serve `/.well-known/openid-configuration` and a JWKS endpoint (Ed25519) built from config, to front an identity provider
 - **Forward-auth**: a verification endpoint (`/auth/verify`) for a fronting proxy (nginx `auth_request`, Traefik `forwardAuth`) to delegate auth, returning the verified identity as headers
@@ -108,25 +108,38 @@ streaming:
   sse_keep_alive_secs: 15
 
 # Rate limiting (Shield)
+#
+# Every decision is made locally with a GCRA shaper (no blocking latency).
+# Named profiles define tiers as data; rules bind a path glob to a key and a
+# limit. Rules keyed by a JWT claim run after auth (so the claim is verified);
+# the rest run before auth so anonymous floods are shed cheaply.
 shield:
   enabled: true
-  window_secs: 60 # default window for bare counts like "20"
-  # Optional: shared counters across replicas (needs the `redis` build feature).
-  # Omit for an in-process per-replica store.
-  # redis_url: "redis://127.0.0.1/"
   # CIDR ranges of trusted proxies/LBs. X-Forwarded-For is honored only from
   # these peers; set this behind a load balancer for correct per-client limits.
   trusted_proxies: ["10.0.0.0/8"]
-  # Classify endpoints by glob pattern → class → rate (limited per client IP)
-  endpoint_classes:
+  # Limit tiers: sustained rate ("N/unit" or a bare count = per minute) + burst
+  # (max back-to-back requests; defaults to one window of the rate).
+  profiles:
+    anon: { rate: "60/min", burst: 20 }
+    premium: { rate: "1000/min", burst: 100 }
+  # Applied when a matched rule resolves no other limit.
+  default_profile: "anon"
+  rules:
+    # Anonymous heavy endpoints, keyed by client IP (runs before auth).
     - pattern: "/api/v1/heavy-*"
-      class: "heavy"
-      rate: "10/min"
-  # Per-identifier limits keyed by a request body field
-  identifier_endpoints:
-    - path: "/api/v1/login"
-      body_field: "email"
-      rate: "5/min"
+      key: { type: ip }
+      profile: "anon"
+    # Per-principal limit keyed by a validated JWT claim (runs after auth).
+    - pattern: "/api/v1/**"
+      key: { type: jwt_claim, claim: "sub" }
+  # Optional: resolve a key's limit from the JWT itself (tier name → a profile,
+  # or explicit ratelimit_rpm / ratelimit_burst claims).
+  # jwt_limits: { tier_claim: "ratelimit_tier" }
+  # Optional: async cross-instance reconciliation via a shared store for an
+  # approximate fleet-wide limit (needs the `redis` build feature). The request
+  # path never blocks on it; a store outage degrades to per-instance limits.
+  # sync: { redis_url: "redis://127.0.0.1/", interval_ms: 500 }
 
 # JWT auth
 auth:
@@ -165,6 +178,73 @@ buf build -o my-service.descriptor.bin
 protoc --descriptor_set_out=my-service.descriptor.bin --include_imports *.proto
 ```
 
+## Rate limiting
+
+Shield is an embedded, config-driven limiter designed for a data plane: every
+decision is made in-process by a GCRA shaper, so it adds no blocking latency to
+the request path. GCRA (a token-bucket equivalent storing one timestamp per key)
+lets legitimate bursts through up to a configured `burst` while throttling
+sustained abuse to the `rate`, with no fixed-window boundary burst.
+
+**Keying and phases.** A rule keys on the client IP, a header value (API key),
+or a validated JWT claim. The phase is derived from the key, not configured: an
+IP/header rule needs no verified identity so it runs *before* auth (a fast,
+purely local check that sheds anonymous floods before any signature verification,
+and short-circuits so blocked clients never reach the auth layer); a `jwt_claim`
+rule needs the verified principal so it runs *after* auth. A key falls back to
+the client IP when its value is absent within its own phase, so a limit can't be
+dodged by omitting a header. Note the fallback is phase-local: an anonymous
+request under a `jwt_claim` rule keys by IP in the post-auth phase, but is *not*
+shed pre-auth. For anonymous flood protection, add a separate pre-auth IP (or
+header) rule covering the same paths; a path may match one rule per phase and
+each is enforced independently (defense in depth).
+
+**Limit sources.** A key's `{rate, burst}` resolves in order: the JWT itself
+(a `ratelimit_tier` claim naming a profile, or explicit `ratelimit_rpm` /
+`ratelimit_burst`), then an external service (cached and refreshed in the
+background, never blocking), then the rule's pinned profile, then the default.
+JWT-based resolution only applies to `jwt_claim` rules, since only they run with
+verified claims available; setting `jwt_limits` has no effect on an IP/header
+rule, which runs pre-auth (use a `jwt_claim` key if you want the token's tier to
+drive the limit). Tier-name indirection lets you retune the numbers in config
+without re-issuing tokens or changing the service.
+
+**Response headers.** Every metered response carries the
+[draft-ietf-httpapi-ratelimit-headers](https://datatracker.ietf.org/doc/draft-ietf-httpapi-ratelimit-headers/)
+fields: `RateLimit-Limit` (the tier's per-window quota), `RateLimit-Remaining`
+(requests still admissible now), and `RateLimit-Reset` (whole seconds until the
+limiter drains toward full). A rejected request returns `429` with `Retry-After`
+(whole seconds until a retry would conform). Clients should back off for
+`Retry-After` seconds on a `429`, and may pace themselves using `RateLimit-*` on
+allowed responses. Behind a browser, these are exposed via CORS.
+
+**Deployment modes.**
+
+- **Local (default).** No shared store. Each instance enforces the limit
+  independently, so the fleet-wide effect is roughly `N × rate` for `N`
+  instances. Zero dependencies, lowest latency. Set per-instance limits with
+  that multiplier in mind.
+- **Reconciled (`sync` + `redis` feature).** Instances asynchronously push their
+  deltas to a shared store and pull the aggregate on an interval, converging on
+  an approximate fleet-wide limit. The configured `rate` is then the *fleet*
+  budget. The request path still never blocks on the store; if the store is
+  unreachable, instances degrade to local limiting rather than failing requests.
+
+**Sizing the overshoot.** In reconciled mode the aggregate lags by up to one
+`sync.interval_ms`. Within that lag each of the other instances can admit its
+local `burst` plus a `rate` fraction of the window before the estimate catches
+up (the fleet gate caps sustained fleet volume at `rate`, but `burst` is a
+per-instance allowance the gate does not pre-reserve). With the interval in the
+same time unit as the window, the worst-case fleet overshoot is about
+`(N - 1) × (burst + rate × (interval / window))` requests. For example,
+`burst = 100`, `rate = 1000/min`, `interval = 500 ms`, `N = 4` gives
+`3 × (100 + 1000 × (0.5 / 60)) ≈ 325` extra requests. Smaller `burst` and shorter
+intervals tighten the bound. The global view uses a sliding-window counter, so
+there is no boundary burst on top of this lag.
+
+See the `shield:` block under [Configuration](#configuration) for the full
+schema.
+
 ## Library Usage
 
 ```rust

src/auth/forward.rs

@@ -67,7 +67,9 @@ impl ForwardAuth {
         match self.auth.decide(headers, &path, &method).await {
             // 200 carries the verified claim headers for the fronting proxy to
             // copy upstream.
-            AuthDecision::Allow(claim_headers) => (StatusCode::OK, claim_headers).into_response(),
+            AuthDecision::Allow(claim_headers, _) => {
+                (StatusCode::OK, claim_headers).into_response()
+            }
             AuthDecision::Unauthenticated(msg) => self.deny(msg),
             AuthDecision::Forbidden(msg) => forbidden(msg),
         }

src/auth/jwks.rs

@@ -38,7 +38,7 @@ const JWKS_HTTP_TIMEOUT: Duration = Duration::from_secs(5);
 /// Uses the pure-Rust `ring` provider (installed per-config, not process-global)
 /// and bundles Mozilla's root store via `webpki-roots`, so the binary needs no
 /// system CA bundle and works in musl / scratch / distroless images.
-fn build_tls_config() -> rustls::ClientConfig {
+pub(crate) fn build_tls_config() -> rustls::ClientConfig {
     let mut roots = rustls::RootCertStore::empty();
     roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
     rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))

src/auth/mod.rs

@@ -120,14 +120,23 @@ impl Auth {
 
 /// The outcome of an auth check for a request.
 pub(crate) enum AuthDecision {
-    /// Allowed; forward these (verified) claim headers to the upstream.
-    Allow(HeaderMap),
+    /// Allowed; forward these (verified) claim headers to the upstream, and the
+    /// verified claims themselves (`None` for anonymous access) for downstream
+    /// consumers such as per-principal rate limiting.
+    Allow(HeaderMap, Option<Value>),
     /// Rejected: no/invalid credentials (HTTP 401).
     Unauthenticated(&'static str),
     /// Rejected: authenticated but lacking a required role (HTTP 403).
     Forbidden(&'static str),
 }
 
+/// Verified JWT claims attached to the request by the auth middleware. Present
+/// only when a valid token was supplied, and set exclusively from a verified
+/// token (never from client input), so downstream consumers may safely key
+/// security decisions (e.g. rate limits) on it.
+#[derive(Clone)]
+pub(crate) struct ValidatedClaims(pub(crate) std::sync::Arc<Value>);
+
 impl Auth {
     /// Evaluate auth for a request: validate the bearer token, apply the route
     /// policy, and render the claim headers to forward. This is the single
@@ -168,7 +177,7 @@ impl Auth {
         if let Some(claims) = &claims {
             inject_claim_headers(&mut claim_headers, claims, &self.claims_headers);
         }
-        AuthDecision::Allow(claim_headers)
+        AuthDecision::Allow(claim_headers, claims)
     }
 }
 
@@ -189,11 +198,18 @@ pub async fn middleware(
     match auth.decide(request.headers(), &path, &method).await {
         AuthDecision::Unauthenticated(msg) => unauthorized(msg),
         AuthDecision::Forbidden(msg) => forbidden(msg),
-        AuthDecision::Allow(claim_headers) => {
+        AuthDecision::Allow(claim_headers, claims) => {
             let dst = request.headers_mut();
             for (name, value) in &claim_headers {
                 dst.insert(name.clone(), value.clone());
             }
+            // Expose the verified claims to inner layers (e.g. per-principal rate
+            // limiting) as a typed extension a client cannot forge.
+            if let Some(claims) = claims {
+                request
+                    .extensions_mut()
+                    .insert(ValidatedClaims(std::sync::Arc::new(claims)));
+            }
             next.run(request).await
         }
     }