Scope refusal-time consumption records to the presenting agent's jkt

Per Dick Hardt's design note: the budget_consumed list in a refusal's
resource token is limited to tokens bound to the presenter's cnf key, so
one agent never learns about a sibling's spending (and gains no extra
figures to infer the ceiling from). The meter records each jti's jkt at
observe_grant; consumed_records(key, jkt=...) filters; the middleware
passes the verified signature's keyid. Unscoped access remains for the
PS-side/audit view. Consequence documented: an abandoned agent's records
are never carried home by siblings — the conservative PS rule (unreported
expired allocation = fully consumed) is the normative backstop.

Author: abay94

src/regent_httpsig/budget.py

@@ -104,6 +104,7 @@ class _Pool:
     decimals: int
     grants: dict[str, tuple[int, float]] = field(default_factory=dict)  # jti -> (amount, exp)
     consumed: dict[str, int] = field(default_factory=dict)  # jti -> total committed
+    jkt_of: dict[str, str] = field(default_factory=dict)  # jti -> presenting key thumbprint
     reservations: dict[int, tuple[str, int, float]] = field(default_factory=dict)
     last_activity: float = 0.0
 
@@ -157,10 +158,13 @@ def _remaining(pool: _Pool) -> int:
     # ── public interface (the BudgetMeter contract) ──────────────────────────
 
     async def observe_grant(self, key: MeterKey, jti: str, claim: BudgetClaim,
-                            exp: float) -> None:
+                            exp: float, jkt: str = "") -> None:
         """Register a token's envelope in the principal's pool (idempotent per
-        ``jti``). Raises :class:`UnitMismatch` if the pool already runs in a
-        different unit — one envelope, one unit, no FX at the meter."""
+        ``jti``). ``jkt`` is the RFC 7638 thumbprint of the token's ``cnf`` key —
+        recorded so consumption records can be scoped to the presenting agent
+        (one agent must not learn about its siblings). Raises
+        :class:`UnitMismatch` if the pool already runs in a different unit —
+        one envelope, one unit, no FX at the meter."""
         async with self._lock:
             now = time.monotonic()
             wall_delta = exp - time.time()
@@ -173,6 +177,8 @@ async def observe_grant(self, key: MeterKey, jti: str, claim: BudgetClaim,
                     f"pool runs in {pool.unit}/{pool.decimals}, "
                     f"grant is {claim.unit}/{claim.decimals}")
             pool.last_activity = now
+            if jkt:
+                pool.jkt_of.setdefault(jti, jkt)
             if jti not in pool.grants and wall_delta > 0:
                 pool.grants[jti] = (claim.amount, now + wall_delta)
 
@@ -219,16 +225,24 @@ async def remaining(self, key: MeterKey) -> int:
             pool = self._purge(key, time.monotonic())
             return 0 if pool is None else self._remaining(pool)
 
-    async def consumed_records(self, key: MeterKey) -> list[dict[str, Any]]:
+    async def consumed_records(self, key: MeterKey,
+                               jkt: str | None = None) -> list[dict[str, Any]]:
         """Per-token consumption for the resource token's ``budget_consumed``
         claim: ``[{"jti": ..., "consumed": ...}, ...]``. Non-destructive — the
-        PS deduplicates by ``jti``, so reporting the same record twice is safe."""
+        PS deduplicates by ``jti``, so reporting the same record twice is safe.
+
+        When ``jkt`` is given, records are scoped to tokens bound to that key:
+        the agent carrying the resource token sees only its OWN spending, never
+        its siblings' (privacy between a principal's agents, and no extra
+        figures to infer the ceiling from). Consequence: an abandoned agent's
+        records are never carried home by siblings — the PS-side conservative
+        rule (unreported expired allocation = fully consumed) is the backstop."""
         async with self._lock:
             pool = self._purge(key, time.monotonic())
             if pool is None:
                 return []
             return [
                 {"jti": jti, "consumed": total}
                 for jti, total in sorted(pool.consumed.items())
-                if total > 0
+                if total > 0 and (jkt is None or pool.jkt_of.get(jti) == jkt)
             ]

src/regent_httpsig/fastapi.py

@@ -207,8 +207,12 @@ async def dispatch(
             str(sig.claims.get("aud", "")),
         )
         try:
+            # sig.keyid is the RFC 7638 thumbprint of the token's cnf key (jkt) —
+            # recorded so refusal-time consumption records are scoped to the
+            # presenting agent and never disclose its siblings' spending.
             await self._meter.observe_grant(key, jti, envelope,
-                                            float(sig.claims.get("exp", 0)))
+                                            float(sig.claims.get("exp", 0)),
+                                            jkt=sig.keyid)
         except UnitMismatch as exc:
             logger.warning("budget unit mismatch for %s: %s", key, exc)
             return self._refusal(reason="insufficient-budget", envelope=envelope,
@@ -218,7 +222,8 @@ async def dispatch(
         if isinstance(outcome, InsufficientBudget):
             reason = "budget-exhausted" if outcome.exhausted else "insufficient-budget"
             return await self._refusal_with_token(
-                reason=reason, envelope=envelope, remaining=outcome.remaining, key=key
+                reason=reason, envelope=envelope, remaining=outcome.remaining,
+                key=key, jkt=sig.keyid,
             )
 
         reservation: Reservation = outcome
@@ -258,12 +263,12 @@ async def _verified(self, request: Request) -> VerifiedSignature | None:
 
     async def _refusal_with_token(
         self, *, reason: str, envelope: BudgetClaim,
-        remaining: int, key: MeterKey,
+        remaining: int, key: MeterKey, jkt: str | None = None,
     ) -> Response:
         token: str | None = None
         if self._resource_token is not None:
             try:
-                records = await self._meter.consumed_records(key)
+                records = await self._meter.consumed_records(key, jkt=jkt)
                 token = await _maybe_await(self._resource_token(key, records))
             except Exception:  # noqa: BLE001 — refusal must not fail on the extras
                 logger.warning("resource_token_provider failed", exc_info=True)

tests/test_budget.py

@@ -175,6 +175,58 @@ async def test_meter_commit_clamps_to_reservation() -> None:
     assert await meter.commit(res, 999999) == 700  # clamped to the 300 hold
 
 
+async def test_consumed_records_scoped_to_presenting_jkt() -> None:
+    """Two agents of one principal share the (iss, sub, aud) pool, but each
+    sees only ITS OWN consumption records — never its siblings' (privacy +
+    no extra figures to infer the ceiling from)."""
+    meter = InMemoryMeter()
+    now = time.time()
+    await meter.observe_grant(KEY, "jti-a", _claim(500), now + 60, jkt="jkt-agent-A")
+    await meter.observe_grant(KEY, "jti-b", _claim(500), now + 60, jkt="jkt-agent-B")
+    for jti, cost in (("jti-a", 100), ("jti-b", 250)):
+        res = await meter.reserve(KEY, jti, cost)
+        assert isinstance(res, Reservation)
+        await meter.commit(res, cost)
+
+    assert await meter.consumed_records(KEY, jkt="jkt-agent-A") == [
+        {"jti": "jti-a", "consumed": 100}
+    ]
+    assert await meter.consumed_records(KEY, jkt="jkt-agent-B") == [
+        {"jti": "jti-b", "consumed": 250}
+    ]
+    # Unscoped (PS-side / audit view) still returns the whole pool.
+    assert len(await meter.consumed_records(KEY)) == 2
+
+
+async def test_refusal_records_scoped_to_presenter(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The resource token embedded in a refusal carries only the presenting
+    agent's records: sibling B's spend must not ride home with agent A."""
+    ps_priv, ps_jwk = _ps_pair()
+    agent_a = EgressSigner(seed=generate_seed(), signature_agent=PS_ISS)
+    agent_b = EgressSigner(seed=generate_seed(), signature_agent=PS_ISS)
+    token_a = _auth_token(ps_priv, agent_a, amount=400, jti="at-A")
+    token_b = _auth_token(ps_priv, agent_b, amount=400, jti="at-B")
+    provider_records: list[Any] = []
+
+    def provider(key: Any, records: Any) -> str:
+        provider_records.append(records)
+        return "resource.token"
+
+    app = _app(_verifier(ps_jwk, monkeypatch), resource_token_provider=provider)
+
+    # A spends 300 (price of /v1/search), B spends 300 — pool now at 200.
+    assert (await _post(app, "/v1/search",
+                        _signed_headers(agent_a, token_a, "/v1/search"))).status_code == 200
+    assert (await _post(app, "/v1/search",
+                        _signed_headers(agent_b, token_b, "/v1/search"))).status_code == 200
+    # B asks again: 300 > 200 remaining → refusal with records — B's only.
+    r = await _post(app, "/v1/search", _signed_headers(agent_b, token_b, "/v1/search"))
+    assert r.status_code == 401
+    assert provider_records == [[{"jti": "at-B", "consumed": 300}]]
+
+
 async def test_meter_concurrent_reserves_never_oversell() -> None:
     meter = InMemoryMeter()
     await meter.observe_grant(KEY, "jti-1", _claim(1000), time.time() + 60)