feat(store): expire, cap and rate-limit the anonymous registration path

Cargo.toml

@@ -63,6 +63,16 @@ url = "2"
 # the tree via rustls) so the vulnerable `rsa` crate (Marvin Attack,
 # RUSTSEC-2023-0071) never enters — same "no rsa" stance as VAPID/APNs above.
 aws-lc-rs = "1"
+# Rate limiting in two layers. `governor` (GCRA) is the limiter itself, used
+# directly inside the transport-agnostic dispatch core so that DIDComm — which
+# never passes through axum middleware — is covered too. `tower_governor` is the
+# HTTP-middleware wrapper around it, keyed by peer IP on the public router; it
+# already targets axum 0.8.
+governor = "0.10"
+# `default-features = false` matters: the default feature set enables `tonic`,
+# which drags a gRPC stack (and its own HTTP/2 and prost trees) into a crate that
+# speaks neither. Only the axum integration is wanted.
+tower_governor = { version = "0.8", default-features = false, features = ["axum"] }
 # The store snapshot holds raw push tokens, so it is written through a private
 # temp file: `NamedTempFile::new_in` gives an O_EXCL, mode-0600, unpredictably
 # named file in the target directory, which a `with_extension("json.tmp")` path

README.md

@@ -162,6 +162,31 @@ cargo run
 # GATEWAY_METRICS_TOKEN=<secret>   require `Authorization: Bearer <secret>` on
 #                       the management listener. Unset = no auth (fine on
 #                       loopback).
+# Registry bounds (push/register is anonymous, so these cap what an
+# unauthenticated caller can make the gateway hold; all optional):
+# GATEWAY_UNPROVISIONED_TTL_SECS=86400   drop a handle whose VTA never
+#                       provisioned a trigger after this long. A provisioned
+#                       handle is never swept. This is the main bound on
+#                       anonymous growth; the sweeper runs every 60s.
+# GATEWAY_MAX_HANDLES=100000   total live handles before register is refused
+#                       with "gateway at capacity".
+# GATEWAY_MAX_HANDLES_PER_TOKEN=4   live handles sharing one device token /
+#                       Web Push endpoint, so one device (or one stolen token)
+#                       cannot occupy the registry.
+# GATEWAY_SNAPSHOT_FLUSH_MS=1000   minimum gap between snapshot writes.
+#                       Mutations set a dirty flag; a background flusher writes
+#                       at most this often instead of reserialising the whole
+#                       map per request.
+# Rate limits (all optional; two layers, because DIDComm bypasses HTTP
+# middleware — see the Security notes):
+# GATEWAY_REGISTER_PER_SEC=5 / GATEWAY_REGISTER_BURST=20
+#                       global budget for anonymous push/register.
+# GATEWAY_PER_DID_PER_SEC=20 / GATEWAY_PER_DID_BURST=60
+#                       budget per authenticated caller DID, for
+#                       push/provision and push/wake.
+# GATEWAY_HTTP_PER_SEC=10 / GATEWAY_HTTP_BURST=40
+#                       per-peer-IP budget on POST /trust-tasks (429 when
+#                       exceeded). HTTP transport only.
 # Egress / endpoint policy (all optional):
 # GATEWAY_WEBPUSH_ALLOWED_HOSTS=@default,push.example.org,*.up.example.net
 #                       Web Push services a registration may target. Unset = the
@@ -337,3 +362,28 @@ Trust Task is pulled from the mediator.
 - A `push/*` payload that fails to deserialise gets one fixed reason
   (`payload does not match the push/* 0.2 schema`); the serde detail goes to a
   debug log rather than back to the caller.
+- **The anonymous registration path is bounded.** `push/register` needs no
+  credentials, so it is rate-limited, capped, and expiring:
+  - **Expiry is the root-cause fix.** A freshly registered handle is inert until
+    its VTA provisions a trigger, so a handle still unprovisioned after
+    `GATEWAY_UNPROVISIONED_TTL_SECS` (default 24 h) is swept. Anonymous growth
+    becomes bounded churn instead of a monotonic leak. A provisioned handle is
+    never swept, however old.
+  - **Caps:** `GATEWAY_MAX_HANDLES` in total, and
+    `GATEWAY_MAX_HANDLES_PER_TOKEN` live handles per device token / Web Push
+    endpoint, so one token cannot occupy the registry.
+  - **Rate limits in two layers**, because the DIDComm transport — the preferred
+    one — never passes through HTTP middleware. A `tower_governor` layer limits
+    `POST /trust-tasks` per peer IP (429), and the transport-agnostic dispatch
+    core limits `register` against a global budget and `provision`/`wake` against
+    a budget keyed by the authenticated caller DID. The keyed buckets are
+    themselves reclaimed on a timer, since they are keyed by caller-chosen input.
+  - **Snapshot writes are debounced.** Mutations set a dirty flag and a
+    background flusher writes at most once per `GATEWAY_SNAPSHOT_FLUSH_MS`,
+    rather than reserialising the whole registry on every anonymous request.
+    Writes keep the temp-file + fsync + rename sequence, and a clean shutdown
+    flushes.
+- A caller that exceeds a budget gets a `trust-task-error` with
+  `rate limit exceeded; retry later`; one that hits a registry cap gets
+  `gateway at capacity` or `too many handles for this push token`. Neither
+  reveals anything about other tenants.

src/api.rs

@@ -37,6 +37,7 @@ use uuid::Uuid;
 
 use crate::auth::{self, HEADER_DID, HEADER_SIG};
 use crate::egress::EgressPolicy;
+use crate::limits::Limits;
 use crate::metrics::Metrics;
 use crate::sender::{self, PushSender, SendOutcome};
 use crate::store::{ProvisionOutcome, Store, WakeAuthz};
@@ -75,6 +76,9 @@ pub struct AppState {
     pub metrics: Arc<Metrics>,
     /// What registrations may point push delivery at.
     pub egress: Arc<EgressPolicy>,
+    /// Per-operation rate limits. Consulted in [`dispatch_push`] rather than in
+    /// HTTP middleware, so the DIDComm transport is covered too.
+    pub limits: Arc<Limits>,
 }
 
 /// The **public** router: the `push/*` Trust-Task endpoint and a liveness probe.
@@ -181,6 +185,23 @@ fn reject_value(doc: &TrustTask<Value>, reason: RejectReason) -> Value {
     serde_json::to_value(doc.reject_with(new_id(), reason)).unwrap_or(Value::Null)
 }
 
+/// A `trust-task-error` for a caller that has exceeded its budget.
+///
+/// `TaskFailed` rather than a transport status because the dispatch core is
+/// transport-agnostic: a DIDComm caller has no HTTP status to receive, and the
+/// in-band envelope is the contract both transports share. The HTTP layer
+/// separately answers 429 for requests it sheds before parsing.
+fn rate_limited(doc: &TrustTask<Value>, operation: &str) -> Value {
+    tracing::warn!(operation, "rate limit exceeded; refusing the request");
+    reject_value(
+        doc,
+        RejectReason::TaskFailed {
+            reason: "rate limit exceeded; retry later".into(),
+            details: None,
+        },
+    )
+}
+
 /// A `malformed_request` error document with a fixed, caller-safe reason.
 fn malformed(doc: &TrustTask<Value>, reason: &str) -> Value {
     reject_value(
@@ -220,9 +241,34 @@ pub(crate) async fn dispatch_push(
         // field-identical; `push/wake`'s response `status` enum became
         // `tokenUnregistered`. `respond_with` mirrors the request version
         // into the `#response`.
-        ("push/register", 0, 2) => handle_register(state, doc).await,
-        ("push/provision", 0, 2) => handle_provision(state, sender, doc).await,
-        ("push/wake", 0, 2) => handle_wake(state, sender, doc).await,
+        //
+        // Rate limits are applied here, before any handler work. `register` is
+        // anonymous so it draws on one global budget; `provision`/`wake` are
+        // authenticated so they are keyed by the caller DID. An unauthenticated
+        // provision/wake is not charged to anyone — it is refused by the handler
+        // for lack of proof, which costs nothing.
+        ("push/register", 0, 2) => {
+            if !state.limits.allow_register() {
+                return rate_limited(doc, "push/register");
+            }
+            handle_register(state, doc).await
+        }
+        ("push/provision", 0, 2) => {
+            if let Some(caller) = sender.as_deref() {
+                if !state.limits.allow_did(caller) {
+                    return rate_limited(doc, "push/provision");
+                }
+            }
+            handle_provision(state, sender, doc).await
+        }
+        ("push/wake", 0, 2) => {
+            if let Some(caller) = sender.as_deref() {
+                if !state.limits.allow_did(caller) {
+                    return rate_limited(doc, "push/wake");
+                }
+            }
+            handle_wake(state, sender, doc).await
+        }
         _ => reject_value(
             doc,
             RejectReason::UnsupportedType {
@@ -254,9 +300,21 @@ async fn handle_register(state: &AppState, doc: &TrustTask<Value>) -> Value {
         );
     }
     let handle = new_handle();
-    state
+    // The store enforces the registry caps; a refusal is reported in-band with a
+    // reason that names the limit but nothing about other tenants.
+    if let Err(e) = state
         .store
-        .insert(handle.clone(), req.registration, req.controller_vta_did);
+        .insert(handle.clone(), req.registration, req.controller_vta_did)
+    {
+        tracing::warn!(reason = e.reason(), "refusing registration");
+        return reject_value(
+            doc,
+            RejectReason::TaskFailed {
+                reason: e.reason().into(),
+                details: None,
+            },
+        );
+    }
     state.metrics.inc_register();
     success_value(
         doc,

src/lib.rs

@@ -7,6 +7,7 @@ pub mod auth;
 pub mod didcomm;
 pub mod egress;
 pub mod identity;
+pub mod limits;
 pub mod metrics;
 pub mod resolver;
 pub mod secretfile;