From 3787cf4b885ed0e450c1eefe4b7c6157d13b0b46 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 07:25:20 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20harden=20the=20capability-token=20l?= =?UTF-8?q?ifecycle=20=E2=80=94=20rotation,=20TTL,=20invoke-time=20enforce?= =?UTF-8?q?ment?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Groups six issues that share one implementation path (a CapabilityToken from issuance through invoke-time enforcement) and one code area. - #185 Signing-key rotation: HMACTokenProvider(secrets={key_id: secret}, active_key_id=...) verifies previous-key tokens during an overlap window; key_id is signed into the payload; unknown key id fails closed. New WEAVER_KERNEL_SECRETS / WEAVER_KERNEL_ACTIVE_KEY env resolution. - #200 CapabilityToken.from_dict raises typed TokenInvalid on malformed input instead of leaking KeyError/ValueError. - #203 Per-grant TTL: grant_capability(ttl_s=...) with DefaultPolicyEngine( max_ttl_s=...); non-positive or over-max is denied (not clamped), audited. - #183 Signed argument constraints (constraints["args"]: allowed_keys/pinned/ prefix) enforced at invoke() and invoke_stream() before the driver runs, raising TokenScopeError with a failure trace; dry-run predicts the same. - #170 Opt-in per-invocation rate limiting (Kernel(invoke_rate_limits=...)), default off, race-free check-then-record; dry-run never consumes. - #224 ADR docs/adr/0001-token-signing-evolution.md (HMAC vs macaroon vs Biscuit, measured); recommends HMAC + re-issuance now. No code/deps change. To stay within the AGENTS.md 300-line module budget, HMACTokenProvider moved to _hmac_provider.py (re-exported; logger name unchanged) and helpers were split into _token_signing.py, policy_ttl.py, kernel/_grant.py, kernel/_constraints.py. Removed the private policy._DEFAULT_RATE_LIMITS/_SERVICE_RATE_MULTIPLIER aliases (part of #196). Breaking: the signed payload now includes key_id, so pre-upgrade tokens no longer verify (accepted pre-1.0; legacy config unchanged). make ci: fmt-check, lint, mypy --strict (67 files), 856 passed / 1 skipped, 94% branch coverage, all 13 examples. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_019WRxQL8t2Uusa6845jWtVV --- CHANGELOG.md | 47 +++ docs/adr/0001-token-signing-evolution.md | 111 +++++++ docs/agent-context/invariants.md | 10 + docs/security.md | 84 ++++- src/weaver_kernel/_hmac_provider.py | 260 ++++++++++++++++ src/weaver_kernel/_secrets.py | 102 +++++++ src/weaver_kernel/_token_signing.py | 146 +++++++++ src/weaver_kernel/errors.py | 12 +- src/weaver_kernel/kernel/__init__.py | 115 ++++--- src/weaver_kernel/kernel/_constraints.py | 200 ++++++++++++ src/weaver_kernel/kernel/_grant.py | 144 +++++++++ src/weaver_kernel/otel.py | 3 +- src/weaver_kernel/policy.py | 14 +- src/weaver_kernel/policy_reasons.py | 6 + src/weaver_kernel/policy_ttl.py | 59 ++++ src/weaver_kernel/tokens.py | 242 ++------------- tests/test_architecture.py | 5 +- tests/test_kernel.py | 374 +++++++++++++++++++++++ tests/test_policy.py | 35 +++ tests/test_tokens.py | 200 ++++++++++++ 20 files changed, 1887 insertions(+), 282 deletions(-) create mode 100644 docs/adr/0001-token-signing-evolution.md create mode 100644 src/weaver_kernel/_hmac_provider.py create mode 100644 src/weaver_kernel/_token_signing.py create mode 100644 src/weaver_kernel/kernel/_constraints.py create mode 100644 src/weaver_kernel/kernel/_grant.py create mode 100644 src/weaver_kernel/policy_ttl.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c89903f..fe07370 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -113,8 +113,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 deny→allow, and reason-code flips deterministically. Rate-limit-dependent flips are surfaced separately (`DecisionDiff.rate_limited`). Companion: `examples/trace_replay_demo.py`. +- **Capability-token lifecycle hardening.** A grouped pass over token issuance, + rotation, and invoke-time enforcement: + - **Signing-key rotation (#185).** `HMACTokenProvider(secrets={key_id: secret}, + active_key_id=...)` signs new tokens under one key while verifying tokens + signed under others during an overlap window, so `WEAVER_KERNEL_SECRET` can + rotate without invalidating every outstanding token at once. The signing + `key_id` is inside the signed payload (tamper-evident); an unknown key id + fails closed as `TokenInvalid`. A new `WEAVER_KERNEL_SECRETS` (JSON + `{key_id: secret}`) / `WEAVER_KERNEL_ACTIVE_KEY` env pair configures it, and a + non-active-key verification logs `token_verified_non_active_key` (key id only, + never the secret) so operators can tell when a key is safe to retire. + - **Per-grant TTL (#203).** `Kernel.grant_capability(..., ttl_s=...)` sets a + token's lifetime per grant; `DefaultPolicyEngine(max_ttl_s=...)` (a single cap + or per-safety-class map) bounds it. A non-positive or over-maximum request is + **denied** with a stable reason code (`invalid_constraint` / `ttl_exceeded`), + never silently clamped. + - **Signed argument-level constraints (#183).** A token's `constraints["args"]` + (`allowed_keys`, `pinned`, `prefix`) is enforced by `invoke()` and + `invoke_stream()` **before** the driver runs and budget is reserved, raising + `TokenScopeError` (`arg_constraint_violation`) with an audited failure trace. + Dry-run predicts the identical outcome. The declarative policy engine needed + no changes — `constraints` already flows into the issued token. + - **Opt-in per-invocation rate limiting (#170).** `Kernel(invoke_rate_limits= + {SafetyClass: (limit, window_s)})` adds an invoke-time sliding-window limit, + independent of and additional to the grant-time limit. **Default off.** The + check-then-record pair runs with no `await` between them, so concurrent + invokes cannot over-admit; dry-run never consumes it. + - **Typed `CapabilityToken.from_dict` errors (#200).** A malformed serialized + token (missing field, wrong type, invalid timestamp, non-object + `constraints`) now raises `TokenInvalid` with a descriptive message instead + of a bare `KeyError`/`ValueError`. Valid round-trips are unchanged. + - **Token-format evolution ADR (#224).** `docs/adr/0001-token-signing-evolution.md` + evaluates HMAC (status quo), macaroon-style caveat chaining, and Biscuit + against the kernel's invariants with measured numbers, and recommends staying + HMAC + re-issuance now (macaroon-chaining as the documented future path, + Biscuit deferred). No code or dependency change. ### Changed +- **Capability-token signed payload now includes `key_id` (#185).** A token + issued by pre-upgrade code fails verification after this deploys — the signed + payload shape differs even under the same secret. Accepted as a break given the + pre-1.0 alpha status and the default 1-hour token TTL; legacy single-secret + *configuration* (`secret=` / `WEAVER_KERNEL_SECRET`) keeps working unchanged. + The `HMACTokenProvider` implementation moved to `weaver_kernel._hmac_provider` + (re-exported from `weaver_kernel.tokens`; the logger name is unchanged). +- **Removed the private `policy._DEFAULT_RATE_LIMITS` / `_SERVICE_RATE_MULTIPLIER` + aliases (part of #196).** Internal call sites use the canonical + `rate_limit.DEFAULT_RATE_LIMITS` / `SERVICE_RATE_MULTIPLIER`. Private surface; + no deprecation cycle. - **CI aligned with `make ci` and hardened (#209, #210, #232).** The `ci.yml` test job now invokes the Makefile targets (`fmt-check`/`lint`/`type`/`test`/ `example`) instead of re-implementing them, so the local gate and CI cannot diff --git a/docs/adr/0001-token-signing-evolution.md b/docs/adr/0001-token-signing-evolution.md new file mode 100644 index 0000000..3726a92 --- /dev/null +++ b/docs/adr/0001-token-signing-evolution.md @@ -0,0 +1,111 @@ +# ADR 0001 — Capability-token signing format evolution + +- **Status:** Accepted (investigation; no production code change) +- **Tracks:** #224 · **Feeds:** #129 (delegated/attenuated grants), #103 (production-hardening roadmap) +- **Related code:** `tokens.py`, `_token_signing.py`, `_hmac_provider.py`, `federation_discovery.py` + +## Context + +agent-kernel authorizes tool calls with HMAC-SHA256 capability tokens. Today a +token binds `principal + capability + constraints` under a single shared secret +(now a rotatable key-ring, #185). Two roadmap directions push on that format: + +- **Delegated, attenuated grants (#129):** an agent that holds a grant wants to + hand a *narrower* grant to a sub-agent. HMAC cannot do this offline — narrowing + requires re-issuance by the holder of the signing secret. +- **Cross-boundary verification (federation, manifest signing):** a peer that + should *verify* a token must currently hold the *signing* secret, which is the + wrong trust boundary for public verification. + +This ADR evaluates whether to evolve the token format, and records a +recommendation. It changes no production code. + +## Decision drivers (from the kernel's invariants) + +1. **I-06** — a token must keep binding `principal + capability + constraints`; + any format must preserve tamper-evidence of those fields. +2. **Minimal-dependency policy** (AGENTS.md, `invariants.md` #6) — runtime deps + are `httpx` + `pydantic` only; a mandatory crypto dependency is a high bar. +3. **Determinism** — no randomness on the verification path. +4. **Revocation model** — the kernel relies on a server-side revocation store + checked before signature (`_hmac_provider.verify` step 0). Offline-attenuable + formats weaken the default "revoke by id" posture unless paired with it. +5. **`explain()` transparency** — denials must stay human- and agent-legible; + a Datalog policy layer is powerful but less transparent than the current + first-match rule chain. +6. **0.x migration cost** — pre-1.0, a format break is acceptable (tokens live + ≤1h by default) but should not be gratuitous. + +## Options considered + +Measurements below are from this repo (`python 3.11`, `_token_signing.sign`), +recorded so the tradeoff is concrete rather than asserted. + +### A. Status quo — shared-secret HMAC + re-issuance-based attenuation + +- **Binding / determinism / deps:** all satisfied; stdlib `hmac` only. +- **Attenuation:** via re-issuance — a "delegation" is just `grant_capability` + with narrower `constraints`. In the kernel's current single-process, in-process + deployment this is a function call, not a network round-trip, so offline + attenuation buys little. +- **Measured:** token ≈ **378 bytes** JSON (299-byte signable payload); + **≈12.0 µs/verify**, ≈20.4 µs/issue. +- **Revocation:** native (server-side store, checked before crypto). + +### B. Macaroon-style HMAC caveat chaining (in-tree, stdlib only) + +- **Binding / determinism / deps:** all satisfied — caveats chain with stdlib + `hmac` (no new dependency). Verified with a micro-prototype in this ADR's + investigation. +- **Attenuation:** *offline* — a holder appends a caveat and re-chains the + signature **without** the root secret. This is the capability HMAC lacks. +- **Measured (3-caveat prototype):** token ≈ **234 bytes**; **≈8.0 µs/verify**. + Offline attenuation (narrowing `args.path.prefix` from `/safe/` to + `/safe/reports/` with no root secret) confirmed working. +- **Costs:** first-party-caveat predicates become a small language to design and + keep deterministic; revocation of a *delegated* leaf needs an identifier + scheme layered on top of the existing store. + +### C. Biscuit (public-key + Datalog attenuation, behind an extra) + +- **Binding:** satisfied, plus public-key verification (verify without the + signing secret) — the one thing neither A nor B offers. +- **Deps:** a **mandatory third-party library** with native crypto — directly + against the minimal-dependency invariant; only viable behind an optional extra. +- **Determinism / transparency:** Datalog is expressive but reduces + `explain()`-style transparency and adds a non-trivial evaluation surface. +- **Revocation:** offline-verifiable tokens are the *hardest* to revoke; needs a + parallel revocation channel. + +## Decision + +**Stay on HMAC (Option A) for now**, strengthened by the key-ring rotation +shipped in #185 (which closes the "can't rotate the secret" gap that most +motivated looking elsewhere). + +- **Defer Biscuit (C).** Its unique win — public-key verification — has no + current consumer, and its mandatory dependency + weakened default revocation + posture conflict with two invariants. Revisit only if cross-trust-boundary + *offline* verification becomes a real requirement. +- **Keep macaroon-style chaining (B) as the documented evolution path** if an + **offline** delegation requirement actually materializes (e.g. once a remote + kernel mode, #227, makes delegation a network hop rather than a function call). + The `constraints["args"]` vocabulary added in #183 (`allowed_keys` / `pinned` / + `prefix`) is deliberately shaped to be reusable as a first-party caveat + predicate language if that day comes. +- **Implement delegation (#129) via re-issuance** in the meantime: a delegation + request is `grant_capability` with narrower `constraints`, which stays inside + the existing policy → token → revocation pipeline. + +## Consequences + +- No dependency change; no token-format change beyond #185's `key_id`. +- #129 proceeds on re-issuance semantics; #103 records rotation as its first + shipped hardening slice. +- The `TokenProvider` Protocol remains the seam: any future B/C provider slots in + behind it with a dual-verification window, without kernel-wide changes. + +## Revisit triggers + +- A concrete need for **offline** attenuation or **secret-less** verification. +- A remote/sidecar kernel mode (#227) that turns delegation into a network hop. diff --git a/docs/agent-context/invariants.md b/docs/agent-context/invariants.md index cb4435c..edcfc24 100644 --- a/docs/agent-context/invariants.md +++ b/docs/agent-context/invariants.md @@ -65,6 +65,16 @@ tag is **silently ignored** — capabilities tagged with it pass policy without **Rule:** When adding a `SensitivityTag`, always add a matching policy rule and test. +### Invoke-time enforcement placement +Argument-constraint enforcement (#183) and the optional per-invocation rate limit +(#170) run in `Kernel.invoke`/`Kernel.invoke_stream` **after** `verify()` and +**before** the driver runs or budget is reserved (`kernel/_constraints.py`). A +violation on the real path records a failure `ActionTrace` before raising, so I-02 +holds for denied executions; `dry_run=True` evaluates the same checks for parity +but records no rate-limit usage and writes no trace. Both single-shot and +streaming entry points must call `run_pre_invoke_checks` — a new execution entry +point must call it too, or it silently bypasses argument scoping and rate limits. + ### Dry-run response-mode parity `Kernel.invoke(dry_run=True)` reports the response mode the caller would actually get at real-invoke time. The Firewall downgrades `raw` to `summary` for non-admin diff --git a/docs/security.md b/docs/security.md index 3e136f1..4dc98a8 100644 --- a/docs/security.md +++ b/docs/security.md @@ -24,11 +24,93 @@ A `CapabilityToken` binds: - `capability_id` — which capability is authorized - `principal_id` — who the token was issued to -- `constraints` — max_rows, allowed_fields, etc. (signed into the token) +- `constraints` — max_rows, allowed_fields, `args`, etc. (signed into the token) - `expires_at` — validity window +- `key_id` — which signing key produced the signature (for rotation, below) Any change to these fields invalidates the HMAC signature. +### Per-grant TTL (#203) + +`Kernel.grant_capability(request, principal, justification=..., ttl_s=...)` sets a +token's lifetime per grant instead of the fixed provider default (3600 s). Least +privilege is temporal as well as scoped — a one-shot lookup need not yield an +hour-long credential. Configure a ceiling on the policy engine: + +```python +from weaver_kernel import DefaultPolicyEngine, SafetyClass + +# One cap for every safety class, or a per-class map: +DefaultPolicyEngine(max_ttl_s=300) +DefaultPolicyEngine(max_ttl_s={SafetyClass.READ: 60, SafetyClass.DESTRUCTIVE: 30}) +``` + +A non-positive `ttl_s`, or one above the policy maximum, is **denied** (reason +codes `invalid_constraint` / `ttl_exceeded`) and audited as a `"deny"` trace — +never silently clamped, so a caller never receives less privilege than it can see. + +### Signed argument constraints (#183) + +Beyond authorizing a *capability*, a token can pin the *arguments* an invocation +may pass, signed into `constraints["args"]` and enforced at `invoke()` / +`invoke_stream()` time (before the driver runs and budget is reserved). The v1 +vocabulary is deliberately tiny and deterministic (top-level keys only): + +| Rule | Meaning | +|------|---------| +| `allowed_keys` | Every argument key must appear in this list. | +| `pinned` | Each named key must be present and exactly equal to the given value. | +| `prefix` | Each named key must be a string starting with the given prefix. | + +A violation raises `TokenScopeError` (`reason_code = arg_constraint_violation`) +with an audited failure trace and never reaches the driver; `dry_run=True` +predicts the identical outcome. This is the difference between "may call the +refund tool" and "may refund order #123". + +### Signing-key rotation (#185) + +`HMACTokenProvider` verifies against a small key-ring so `WEAVER_KERNEL_SECRET` +can rotate without invalidating every outstanding token at once: + +```python +# Sign new tokens under k2; keep k1 for the overlap window so tokens signed +# under it still verify. Retire k1 once max TTL has elapsed. +HMACTokenProvider(secrets={"k1": old, "k2": new}, active_key_id="k2") +``` + +The signing `key_id` is part of the signed payload (tamper-evident); a token +declaring a key id not in the ring fails closed as `TokenInvalid`. Verifying a +non-active-key token logs `token_verified_non_active_key` (key id only, never the +secret) so operators can see when the previous key is safe to retire. + +**Secret resolution precedence** (first match wins): the `secrets=` /`secret=` +constructor argument → `WEAVER_KERNEL_SECRETS` (JSON `{key_id: secret}`, with +`WEAVER_KERNEL_ACTIVE_KEY` naming the active key) → the legacy single +`WEAVER_KERNEL_SECRET` → a random development secret (with a one-time warning). +The resolved secret is never logged. + +**Rotation runbook:** add the new key alongside the old (`secrets={old, new}`) → +set `active_key_id`/`WEAVER_KERNEL_ACTIVE_KEY` to the new key → wait for the +maximum token TTL so no live token is still signed under the old key → drop the +old key from the ring. + +### Per-invocation rate limiting (#170) + +The policy engine rate-limits at *grant* time; a multi-use token can then drive +many `invoke()` calls until it expires. For runaway-loop and abuse protection, +`Kernel(invoke_rate_limits={SafetyClass.READ: (limit, window_s)})` adds an +independent sliding-window limit on the *execution* path (default off). It is +enforced identically for `invoke()` and `invoke_stream()`; an exhausted limit +raises `PolicyDenied` (`reason_code = rate_limited`) with an audited failure +trace, and `dry_run=True` never consumes the window. + +### Token deserialization (#200) + +`CapabilityToken.from_dict` validates untrusted input and raises the typed +`TokenInvalid` (never a bare `KeyError`/`ValueError`) on a missing field, wrong +type, malformed timestamp, or non-object `constraints` — tokens cross process +boundaries, so a malformed blob is an expected input class, not a crash. + ## Confused deputy prevention Consider an agent that obtains a token for `billing.list_invoices` then passes it to a different agent. The second agent cannot use it because `verify()` checks that `token.principal_id == expected_principal_id`. diff --git a/src/weaver_kernel/_hmac_provider.py b/src/weaver_kernel/_hmac_provider.py new file mode 100644 index 0000000..8b1d512 --- /dev/null +++ b/src/weaver_kernel/_hmac_provider.py @@ -0,0 +1,260 @@ +"""The :class:`HMACTokenProvider` implementation. + +Extracted from :mod:`weaver_kernel.tokens` to keep that module within the +AGENTS.md 300-line budget. :class:`~weaver_kernel.tokens.CapabilityToken` and +the :class:`~weaver_kernel.tokens.TokenProvider` Protocol remain in +:mod:`weaver_kernel.tokens`, which re-exports this class so +``from weaver_kernel.tokens import HMACTokenProvider`` keeps working. +""" + +from __future__ import annotations + +import datetime +import hmac +import logging +import uuid +from typing import Any + +from ._secrets import resolve_keyring +from ._token_signing import KeyRing, sign +from .errors import AgentKernelError, TokenExpired, TokenInvalid, TokenRevoked, TokenScopeError +from .stores import InMemoryRevocationStore, RevocationStoreProtocol +from .tokens import CapabilityToken + +# Keep the logger name stable across the tokens.py → _hmac_provider.py split so +# operators (and tests) filtering on "weaver_kernel.tokens" still see these records. +logger = logging.getLogger("weaver_kernel.tokens") + + +class HMACTokenProvider: + """Issues and verifies HMAC-SHA256 capability tokens. + + Supports signing-key rotation (#185): pass a ``secrets`` map of + ``{key_id: secret}`` plus an ``active_key_id`` to sign new tokens under one + key while still verifying tokens signed under others during an overlap + window. A single ``secret`` (or the ``WEAVER_KERNEL_SECRET`` env var) is + filed under the ``"default"`` key id. When nothing is configured the + ``WEAVER_KERNEL_SECRETS`` / ``WEAVER_KERNEL_SECRET`` env vars are consulted, + falling back to a random development secret with a one-time warning. + + Args: + secret: A single signing secret. Mutually exclusive with *secrets*. + secrets: A ``{key_id: secret}`` key-ring for rotation. + active_key_id: Which *secrets* key to sign new tokens with. Required when + *secrets* holds more than one key; inferred when it holds exactly one. + revocation_store: Backing store for revocation state; defaults to an + in-memory store. + + Raises: + AgentKernelError: If both *secret* and *secrets* are given, or an + explicit key-ring is empty, malformed, or names an unknown + *active_key_id*. + """ + + def __init__( + self, + secret: str | None = None, + *, + secrets: dict[str, str] | None = None, + active_key_id: str | None = None, + revocation_store: RevocationStoreProtocol | None = None, + ) -> None: + if secret is not None and secrets is not None: + raise AgentKernelError( + "HMACTokenProvider: pass either 'secret' or 'secrets', not both." + ) + self._secret = secret + self._secrets = secrets + self._active_key_id_arg = active_key_id + # Explicit config is validated eagerly; the env/dev-fallback path stays + # lazy so the dev-secret warning fires only on first use, not import. + self._keyring: KeyRing | None = None + self._active_key_id: str = "" + if secret is not None or secrets is not None: + self._keyring, self._active_key_id = resolve_keyring(secret, secrets, active_key_id) + # Revocation state lives behind a protocol so it can be made durable + # (e.g. SQLiteRevocationStore) without weakening verify-before-invoke. + self._revocation: RevocationStoreProtocol = revocation_store or InMemoryRevocationStore() + + @staticmethod + def _log_verify_failure(token_id: str, reason: str, **extra: Any) -> None: + """Log a token verification failure at WARNING.""" + logger.warning( + "token_verify_failed", + extra={"token_id": token_id, "reason": reason, **extra}, + ) + + def _resolve_keyring(self) -> tuple[KeyRing, str]: + """Return the ``(keyring, active_key_id)`` pair, resolving env/dev lazily.""" + if self._keyring is None: + self._keyring, self._active_key_id = resolve_keyring( + self._secret, self._secrets, self._active_key_id_arg + ) + return self._keyring, self._active_key_id + + def issue( + self, + capability_id: str, + principal_id: str, + *, + constraints: dict[str, Any] | None = None, + ttl_seconds: int = 3600, + audit_id: str = "", + ) -> CapabilityToken: + """Issue a new signed token. + + Args: + capability_id: The capability this token authorises. + principal_id: The principal this token is issued to. + constraints: Optional execution constraints. + ttl_seconds: How long the token is valid (default 1 hour). + audit_id: Audit trail ID to embed in the token. + + Returns: + A freshly signed :class:`CapabilityToken`. + """ + keyring, active_key_id = self._resolve_keyring() + now = datetime.datetime.now(tz=datetime.timezone.utc) + token = CapabilityToken( + token_id=str(uuid.uuid4()), + capability_id=capability_id, + principal_id=principal_id, + issued_at=now, + expires_at=now + datetime.timedelta(seconds=ttl_seconds), + constraints=constraints or {}, + audit_id=audit_id, + key_id=active_key_id, + ) + token.signature = sign(keyring[active_key_id], token._signable_payload()) + self._revocation.track(principal_id, token.token_id, token.expires_at) + logger.debug( + "token_issued", + extra={ + "token_id": token.token_id, + "capability_id": capability_id, + "principal_id": principal_id, + "audit_id": audit_id, + "expires_at": token.expires_at.isoformat(), + }, + ) + return token + + def revoke(self, token_id: str) -> None: + """Revoke a single token by ID. + + Idempotent — revoking an already-revoked or unknown token is a no-op. + + Args: + token_id: The ID of the token to revoke. + """ + self._revocation.revoke(token_id) + + def revoke_all(self, principal_id: str) -> int: + """Revoke all tokens issued to a principal. + + Args: + principal_id: The principal whose tokens should be revoked. + + Returns: + The number of tokens newly revoked by this call (excluding tokens + that were already revoked). + """ + return self._revocation.revoke_principal(principal_id) + + def sweep_revocations(self, now: datetime.datetime | None = None) -> int: + """Drop revocation bookkeeping for tokens that have already expired. + + Bounds revocation-state growth in long-lived processes (#182). Safe to + call at any time: an expired token fails the verifier's expiry check + regardless, so sweeping its entry never un-revokes a live token. The + in-memory store also sweeps itself lazily; durable backends expose this + for an operator to call on a schedule. + + Args: + now: Reference time; defaults to the current UTC time. + + Returns: + The number of tracked tokens whose state was removed. + """ + when = now or datetime.datetime.now(tz=datetime.timezone.utc) + return self._revocation.sweep_expired(when) + + def verify( + self, + token: CapabilityToken, + *, + expected_principal_id: str, + expected_capability_id: str, + ) -> None: + """Verify a token's signature, expiry, and scope bindings. + + Args: + token: The token to verify. + expected_principal_id: The principal that should own this token. + expected_capability_id: The capability this token should authorize. + + Raises: + TokenRevoked: If the token has been revoked. + TokenExpired: If ``token.expires_at`` is in the past. + TokenInvalid: If the HMAC signature does not verify, or the token + declares an unknown signing key id. + TokenScopeError: If principal or capability do not match. + """ + # 0. Revocation (fast lookup before any crypto) + if self._revocation.is_revoked(token.token_id): + self._log_verify_failure(token.token_id, "revoked") + raise TokenRevoked(f"Token '{token.token_id}' has been revoked.") + + # 1. Expiry + now = datetime.datetime.now(tz=datetime.timezone.utc) + if token.expires_at <= now: + self._log_verify_failure( + token.token_id, "expired", expires_at=token.expires_at.isoformat() + ) + raise TokenExpired( + f"Token '{token.token_id}' expired at {token.expires_at.isoformat()}." + ) + + # 2. Signature (rotation-aware): select the secret for the token's + # declared key id. An unknown key id fails closed — never fall through + # to another key. + keyring, active_key_id = self._resolve_keyring() + secret = keyring.get(token.key_id) + if secret is None: + self._log_verify_failure(token.token_id, "unknown_key_id", key_id=token.key_id) + raise TokenInvalid( + f"Token '{token.token_id}' was signed with unknown key id '{token.key_id}'." + ) + expected_sig = sign(secret, token._signable_payload()) + if not hmac.compare_digest(expected_sig, token.signature): + self._log_verify_failure(token.token_id, "invalid_signature") + raise TokenInvalid( + f"Token '{token.token_id}' has an invalid signature. " + "The token may have been tampered with." + ) + if token.key_id != active_key_id: + # Never logs the secret — only the key id — so operators can tell + # when the previous key is safe to retire (#185). + logger.info( + "token_verified_non_active_key", + extra={"token_id": token.token_id, "key_id": token.key_id}, + ) + + # 3. Principal binding (confused-deputy prevention) + if token.principal_id != expected_principal_id: + self._log_verify_failure(token.token_id, "principal_mismatch") + raise TokenScopeError( + f"Token '{token.token_id}' was issued for principal " + f"'{token.principal_id}', not '{expected_principal_id}'." + ) + + # 4. Capability binding + if token.capability_id != expected_capability_id: + self._log_verify_failure(token.token_id, "capability_mismatch") + raise TokenScopeError( + f"Token '{token.token_id}' was issued for capability " + f"'{token.capability_id}', not '{expected_capability_id}'." + ) + + +__all__ = ["HMACTokenProvider"] diff --git a/src/weaver_kernel/_secrets.py b/src/weaver_kernel/_secrets.py index 3a81ca7..4ff9254 100644 --- a/src/weaver_kernel/_secrets.py +++ b/src/weaver_kernel/_secrets.py @@ -11,16 +11,28 @@ from __future__ import annotations +import json import logging import os import secrets import threading +from .errors import AgentKernelError + logger = logging.getLogger(__name__) SECRET_ENV_VAR = "WEAVER_KERNEL_SECRET" """Environment variable holding the HMAC secret used for tokens and audit chains.""" +SECRETS_ENV_VAR = "WEAVER_KERNEL_SECRETS" +"""Environment variable holding a JSON ``{key_id: secret}`` map for key rotation (#185).""" + +ACTIVE_KEY_ENV_VAR = "WEAVER_KERNEL_ACTIVE_KEY" +"""Environment variable naming which :data:`SECRETS_ENV_VAR` key to sign new tokens with.""" + +LEGACY_KEY_ID = "default" +"""Key id assigned to a single-secret configuration (``secret=`` or ``WEAVER_KERNEL_SECRET``).""" + _DEV_SECRET: str | None = None _DEV_SECRET_LOCK = threading.Lock() @@ -62,3 +74,93 @@ def resolve_hmac_secret(explicit: str | None = None) -> str: if explicit: return explicit return _get_secret() + + +def _assemble_keyring( + secrets_map: dict[str, str], + active_key_id: str | None, + *, + source: str, +) -> tuple[dict[str, str], str]: + """Validate a key id → secret map and resolve which key is active. + + Args: + secrets_map: Candidate ``{key_id: secret}`` mapping. + active_key_id: The key id new tokens should be signed with, or ``None`` + to infer it (only possible when the map holds exactly one key). + source: Human-readable origin used in error messages. + + Returns: + A ``(keyring, active_key_id)`` pair with a validated, non-empty keyring. + + Raises: + AgentKernelError: If the map is empty, holds non-string keys/values, has + multiple keys without an explicit active key, or names an active key + id absent from the map. + """ + if not secrets_map: + raise AgentKernelError(f"{source} keyring is empty; at least one key is required.") + if any(not isinstance(k, str) or not isinstance(v, str) for k, v in secrets_map.items()): + raise AgentKernelError(f"{source} keyring must map string key ids to string secrets.") + keyring = dict(secrets_map) + if active_key_id is None: + if len(keyring) == 1: + active_key_id = next(iter(keyring)) + else: + raise AgentKernelError( + f"{source} has multiple keys; an active key id must be specified " + f"(via active_key_id or {ACTIVE_KEY_ENV_VAR})." + ) + if active_key_id not in keyring: + raise AgentKernelError( + f"active key id {active_key_id!r} is not present in the {source} keyring." + ) + return keyring, active_key_id + + +def resolve_keyring( + explicit_secret: str | None, + explicit_secrets: dict[str, str] | None, + explicit_active_key_id: str | None, +) -> tuple[dict[str, str], str]: + """Resolve the signing key-ring and active key id for token rotation (#185). + + Precedence: an explicit ``secrets`` map, then an explicit single ``secret``, + then :data:`SECRETS_ENV_VAR` (JSON ``{key_id: secret}``), then the legacy + single :data:`SECRET_ENV_VAR`, then a generated dev secret (with a one-time + warning). A single secret is filed under :data:`LEGACY_KEY_ID`. + + Args: + explicit_secret: A single secret passed to the provider, or ``None``. + explicit_secrets: A ``{key_id: secret}`` map passed to the provider, or + ``None``. + explicit_active_key_id: The active key id passed to the provider, or + ``None`` to infer it. + + Returns: + A ``(keyring, active_key_id)`` pair. + + Raises: + AgentKernelError: If the resolved configuration is empty, malformed, or + names an unknown active key id. + """ + if explicit_secrets is not None: + return _assemble_keyring(explicit_secrets, explicit_active_key_id, source="secrets=") + if explicit_secret is not None: + return {LEGACY_KEY_ID: explicit_secret}, LEGACY_KEY_ID + env_secrets = os.environ.get(SECRETS_ENV_VAR) + if env_secrets: + try: + parsed = json.loads(env_secrets) + except json.JSONDecodeError as exc: + raise AgentKernelError(f"{SECRETS_ENV_VAR} is not valid JSON: {exc}.") from exc + if not isinstance(parsed, dict): + raise AgentKernelError( + f"{SECRETS_ENV_VAR} must be a JSON object of {{key_id: secret}} strings." + ) + active = explicit_active_key_id or os.environ.get(ACTIVE_KEY_ENV_VAR) + return _assemble_keyring(parsed, active, source=SECRETS_ENV_VAR) + single = os.environ.get(SECRET_ENV_VAR) + if single: + return {LEGACY_KEY_ID: single}, LEGACY_KEY_ID + return {LEGACY_KEY_ID: _get_secret()}, LEGACY_KEY_ID diff --git a/src/weaver_kernel/_token_signing.py b/src/weaver_kernel/_token_signing.py new file mode 100644 index 0000000..2a5718e --- /dev/null +++ b/src/weaver_kernel/_token_signing.py @@ -0,0 +1,146 @@ +"""HMAC signing and serialization helpers for capability tokens. + +Extracted from :mod:`weaver_kernel.tokens` to keep that module within the +AGENTS.md 300-line budget and to isolate the crypto/serialization concern: +building the canonical signable payload, HMAC signing, and parsing an untrusted +serialized token into validated fields — raising :class:`TokenInvalid` rather +than leaking a bare ``KeyError``/``ValueError`` (#200). + +The signed payload includes the ``key_id`` so signing-key rotation (#185) is +tamper-evident: a token cannot be re-labelled to verify against a different key. +""" + +from __future__ import annotations + +import datetime +import hashlib +import hmac +import json +from typing import TYPE_CHECKING, Any + +from .errors import TokenInvalid + +if TYPE_CHECKING: # pragma: no cover + from .tokens import CapabilityToken + +KeyRing = dict[str, str] +"""Mapping of key id → HMAC secret used for signing-key rotation (#185).""" + + +def build_signable_payload(token: CapabilityToken) -> str: + """Return the canonical JSON string used as the HMAC message. + + Args: + token: The token whose bound fields (principal, capability, constraints, + expiry, and signing ``key_id``) form the signature input. + + Returns: + A deterministic, key-sorted JSON string. + """ + payload = { + "token_id": token.token_id, + "capability_id": token.capability_id, + "principal_id": token.principal_id, + "issued_at": token.issued_at.isoformat(), + "expires_at": token.expires_at.isoformat(), + "constraints": token.constraints, + "audit_id": token.audit_id, + "key_id": token.key_id, + } + return json.dumps(payload, sort_keys=True, separators=(",", ":")) + + +def sign(secret: str, payload: str) -> str: + """Return the hex HMAC-SHA256 of *payload* under *secret*. + + Args: + secret: The signing secret (never logged). + payload: The canonical signable payload string. + + Returns: + The hex-encoded signature. + """ + return hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest() + + +def _require_str(data: dict[str, Any], field: str) -> str: + """Return a required string field, raising :class:`TokenInvalid` otherwise.""" + if field not in data: + raise TokenInvalid(f"malformed token payload: missing field '{field}'.") + value = data[field] + if not isinstance(value, str): + raise TokenInvalid( + f"malformed token payload: field '{field}' must be a string, " + f"got {type(value).__name__}." + ) + return value + + +def _optional_str(data: dict[str, Any], field: str) -> str: + """Return an optional string field (default ``""``), validating its type.""" + value = data.get(field, "") + if not isinstance(value, str): + raise TokenInvalid( + f"malformed token payload: field '{field}' must be a string, " + f"got {type(value).__name__}." + ) + return value + + +def _require_timestamp(data: dict[str, Any], field: str) -> datetime.datetime: + """Return a required ISO-8601 timestamp field, raising :class:`TokenInvalid`.""" + if field not in data: + raise TokenInvalid(f"malformed token payload: missing field '{field}'.") + raw = data[field] + if not isinstance(raw, str): + raise TokenInvalid( + f"malformed token payload: field '{field}' must be an ISO-8601 string, " + f"got {type(raw).__name__}." + ) + try: + return datetime.datetime.fromisoformat(raw) + except ValueError as exc: + raise TokenInvalid( + f"malformed token payload: invalid timestamp in field '{field}': {raw!r}." + ) from exc + + +def parse_token_fields(data: dict[str, Any]) -> dict[str, Any]: + """Validate a serialized token dict into constructor kwargs (#200). + + Tokens cross process boundaries by design, so a malformed dict is an + expected input class, not a programming error. Every failure raises + :class:`TokenInvalid` with a stable, descriptive message. Unknown extra + keys are tolerated; ``constraints`` defaults to ``{}`` and must be an object. + + Args: + data: The plain dict produced by :meth:`CapabilityToken.to_dict` (or an + untrusted equivalent). + + Returns: + Keyword arguments suitable for the :class:`CapabilityToken` constructor. + + Raises: + TokenInvalid: If any field is missing, of the wrong type, or a + malformed timestamp. + """ + constraints = data.get("constraints", {}) + if not isinstance(constraints, dict): + raise TokenInvalid( + f"malformed token payload: field 'constraints' must be an object, " + f"got {type(constraints).__name__}." + ) + return { + "token_id": _require_str(data, "token_id"), + "capability_id": _require_str(data, "capability_id"), + "principal_id": _require_str(data, "principal_id"), + "issued_at": _require_timestamp(data, "issued_at"), + "expires_at": _require_timestamp(data, "expires_at"), + "constraints": constraints, + "audit_id": _optional_str(data, "audit_id"), + "signature": _optional_str(data, "signature"), + "key_id": _optional_str(data, "key_id"), + } + + +__all__ = ["KeyRing", "build_signable_payload", "sign", "parse_token_fields"] diff --git a/src/weaver_kernel/errors.py b/src/weaver_kernel/errors.py index 245299e..fa0e1b1 100644 --- a/src/weaver_kernel/errors.py +++ b/src/weaver_kernel/errors.py @@ -17,7 +17,17 @@ class TokenInvalid(AgentKernelError): class TokenScopeError(AgentKernelError): - """Raised when a token is used by the wrong principal or for the wrong capability.""" + """Raised when a token is used outside its bound scope. + + Covers a token presented by the wrong principal or for the wrong capability, + and invocation arguments that violate a signed ``constraints["args"]`` rule + (#183). Carries an optional stable ``reason_code`` (the same vocabulary as + :class:`PolicyDenied`) so metrics and UI mapping use one denial taxonomy. + """ + + def __init__(self, message: str, *, reason_code: str | None = None) -> None: + super().__init__(message) + self.reason_code: str | None = reason_code class TokenRevoked(AgentKernelError): diff --git a/src/weaver_kernel/kernel/__init__.py b/src/weaver_kernel/kernel/__init__.py index 2a20899..247e9e2 100644 --- a/src/weaver_kernel/kernel/__init__.py +++ b/src/weaver_kernel/kernel/__init__.py @@ -11,11 +11,12 @@ import logging import uuid -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable from typing import Any, Literal, overload from ..drivers.base import Driver, StreamingDriver -from ..errors import AgentKernelError, PolicyDenied +from ..enums import SafetyClass +from ..errors import AgentKernelError from ..federation import TrustPolicy from ..firewall.budget_manager import BudgetManager from ..firewall.transform import Firewall @@ -35,6 +36,7 @@ RoutePlan, ) from ..policy import DefaultPolicyEngine, PolicyEngine +from ..rate_limit import RateLimiter from ..registry import CapabilityRegistry from ..router import Router, StaticRouter from ..stats import KernelStats, StatsSnapshot @@ -42,13 +44,15 @@ from ..tokens import CapabilityToken, HMACTokenProvider, TokenProvider from ..trace import TraceStore from ..trace_query import TraceQuery -from ._audit import record_denial_trace, record_expansion_trace +from ._audit import record_expansion_trace +from ._constraints import run_pre_invoke_checks, validate_invoke_rate_limits from ._dry_run import build_dry_run_result from ._federation import ( perform_advertise, perform_discover_peers, perform_import_remote, ) +from ._grant import perform_grant from ._invoke import perform_invoke from ._stream import invoke_stream_impl @@ -86,7 +90,11 @@ def __init__( trace_store: TraceStoreProtocol | None = None, budget_manager: BudgetManager | None = None, kernel_id: str = "agent-kernel", + invoke_rate_limits: dict[SafetyClass, tuple[int, float]] | None = None, + invoke_rate_clock: Callable[[], float] | None = None, ) -> None: + # invoke_rate_limits (#170): optional invoke-time limits, default off. + validate_invoke_rate_limits(invoke_rate_limits) self._registry = registry self._policy: PolicyEngine = policy or DefaultPolicyEngine() self._token_provider: TokenProvider = token_provider or HMACTokenProvider() @@ -98,6 +106,8 @@ def __init__( self._drivers: dict[str, Driver] = {} self._kernel_id = kernel_id self._stats = KernelStats() + self._invoke_rate_limits: dict[SafetyClass, tuple[int, float]] = invoke_rate_limits or {} + self._invoke_limiter = RateLimiter(clock=invoke_rate_clock) @property def kernel_id(self) -> str: @@ -137,6 +147,7 @@ def grant_capability( principal: Principal, *, justification: str, + ttl_s: int | None = None, ) -> CapabilityGrant: """Evaluate the policy and, if approved, issue a signed token. @@ -146,61 +157,17 @@ def grant_capability( "who was refused what, and why" (#175). A trace-store write failure is logged but never masks the denial. Denials are also counted in :attr:`stats`. + + Args: + request: The capability request being granted. + principal: The principal the grant is issued to. + justification: Free-text justification forwarded to the policy engine. + ttl_s: Optional per-grant token time-to-live in seconds (#203). + ``None`` uses the token provider's default. A non-positive value, + or one exceeding the policy's ``max_ttl_s``, is denied (never + silently clamped). """ - capability = self._registry.get(request.capability_id) - try: - decision = self._policy.evaluate( - request, capability, principal, justification=justification - ) - except PolicyDenied as exc: - self._stats.on_denial(exc.reason_code) - # The denial is authoritative and already fails closed (no token is - # issued). Recording its audit trace is best-effort: a trace-store - # write failure must never mask the PolicyDenied the caller expects. - try: - record_denial_trace( - capability_id=request.capability_id, - principal_id=principal.principal_id, - reason_code=exc.reason_code, - message=str(exc), - trace_store=self._trace_store, - ) - except Exception: - logger.warning( - "deny_trace_record_failed", - extra={ - "capability_id": request.capability_id, - "principal_id": principal.principal_id, - "reason_code": exc.reason_code, - }, - exc_info=True, - ) - raise - audit_id = str(uuid.uuid4()) - token = self._token_provider.issue( - capability.capability_id, - principal.principal_id, - constraints=decision.constraints, - audit_id=audit_id, - ) - logger.info( - "grant_capability", - extra={ - "principal_id": principal.principal_id, - "capability_id": capability.capability_id, - "safety_class": capability.safety_class.value, - "audit_id": audit_id, - "token_id": token.token_id, - }, - ) - self._stats.on_grant() - return CapabilityGrant( - request=request, - principal=principal, - decision=decision, - token=token, - audit_id=audit_id, - ) + return perform_grant(self, request, principal, justification=justification, ttl_s=ttl_s) def get_token( self, @@ -208,9 +175,19 @@ def get_token( principal: Principal, *, justification: str, + ttl_s: int | None = None, ) -> CapabilityToken: - """Like :meth:`grant_capability` but returns the token directly.""" - return self.grant_capability(request, principal, justification=justification).token + """Like :meth:`grant_capability` but returns the token directly. + + Args: + request: The capability request being granted. + principal: The principal the grant is issued to. + justification: Free-text justification forwarded to the policy engine. + ttl_s: Optional per-grant token time-to-live in seconds (#203). + """ + return self.grant_capability( + request, principal, justification=justification, ttl_s=ttl_s + ).token @overload async def invoke( @@ -256,6 +233,16 @@ async def invoke( ) capability = self._registry.get(token.capability_id) plan: RoutePlan = self._router.route(token.capability_id) + # Invoke-time enforcement (#183 arg constraints, #170 rate limits). + run_pre_invoke_checks( + self, + token=token, + capability=capability, + principal=principal, + args=args, + response_mode=response_mode, + dry_run=dry_run, + ) if dry_run: return build_dry_run_result( token=token, @@ -314,6 +301,16 @@ async def invoke_stream( ) capability = self._registry.get(token.capability_id) plan: RoutePlan = self._router.route(token.capability_id) + # Same invoke-time enforcement as the single-shot path (#183, #170). + run_pre_invoke_checks( + self, + token=token, + capability=capability, + principal=principal, + args=args, + response_mode=response_mode, + dry_run=False, + ) async for frame in invoke_stream_impl( kernel=self, token=token, diff --git a/src/weaver_kernel/kernel/_constraints.py b/src/weaver_kernel/kernel/_constraints.py new file mode 100644 index 0000000..7d22e0f --- /dev/null +++ b/src/weaver_kernel/kernel/_constraints.py @@ -0,0 +1,200 @@ +"""Invoke-time enforcement: signed argument constraints (#183) and per-invocation +rate limiting (#170). + +Both checks run on the execution path — *after* token verification, *before* the +driver runs and *before* budget is reserved — so a violation costs nothing +downstream. They are invoked from :meth:`Kernel.invoke` and +:meth:`Kernel.invoke_stream` alike (via :func:`run_pre_invoke_checks`), so the +streaming path is covered too. Dry-run evaluates the identical checks for parity +but never records rate-limit usage. On a real (non-dry-run) violation a failure +:class:`~weaver_kernel.models.ActionTrace` is recorded before the error +propagates, so I-02 (auditability) holds. +""" + +from __future__ import annotations + +import uuid +from collections.abc import Callable +from typing import TYPE_CHECKING, Any + +from ..enums import SafetyClass +from ..errors import AgentKernelError, PolicyDenied, TokenScopeError +from ..models import Capability, Principal, ResponseMode +from ..policy_reasons import DenialReason +from ..tokens import CapabilityToken +from ._invoke import record_failure_trace + +if TYPE_CHECKING: # pragma: no cover + from . import Kernel + +ARG_CONSTRAINTS_KEY = "args" +"""Token-constraint key carrying the argument-level rules enforced at invoke (#183).""" + + +def validate_invoke_rate_limits( + limits: dict[SafetyClass, tuple[int, float]] | None, +) -> None: + """Reject a malformed ``invoke_rate_limits`` configuration at construction (#170). + + Args: + limits: The per-safety-class ``(max_invocations, window_seconds)`` map, or + ``None``. + + Raises: + AgentKernelError: If any limit is < 1 or any window is <= 0. + """ + if not limits: + return + for safety_class, (limit, window) in limits.items(): + if limit < 1 or window <= 0: + raise AgentKernelError( + f"Invalid invoke_rate_limits for {safety_class.value}: limit must be " + f">= 1 and window must be > 0, got limit={limit}, window={window}." + ) + + +def enforce_arg_constraints(args: dict[str, Any], constraints: dict[str, Any]) -> None: + """Enforce a token's signed ``constraints["args"]`` against invocation *args* (#183). + + The v1 vocabulary is deliberately tiny and deterministic — no expression + language (see ``docs/agent-context/invariants.md`` on determinism): + + * ``allowed_keys``: every argument key must be in this list. + * ``pinned``: each named key must be present and exactly equal to the value. + * ``prefix``: each named key must be a string starting with the given prefix. + + Only top-level argument keys are inspected in v1. A ``pinned``/``prefix`` rule + on an omitted argument fails closed. + + Args: + args: The invocation arguments. + constraints: The verified token's ``constraints`` mapping. + + Raises: + TokenScopeError: If any argument violates the spec, carrying + :attr:`~weaver_kernel.policy_reasons.DenialReason.ARG_CONSTRAINT_VIOLATION`. + """ + spec = constraints.get(ARG_CONSTRAINTS_KEY) + if not spec: + return + if not isinstance(spec, dict): + raise _violation(f"malformed 'args' constraint: expected an object, got {spec!r}.") + + allowed = spec.get("allowed_keys") + if allowed is not None: + extra = sorted(set(args) - set(allowed)) + if extra: + raise _violation(f"arguments {extra} are not permitted by the token's allowed_keys.") + + pinned = spec.get("pinned") + if isinstance(pinned, dict): + for key, expected in pinned.items(): + if key not in args or args[key] != expected: + raise _violation(f"argument '{key}' must equal the token's pinned value.") + + prefix = spec.get("prefix") + if isinstance(prefix, dict): + for key, required_prefix in prefix.items(): + value = args.get(key) + if not isinstance(value, str) or not value.startswith(required_prefix): + raise _violation( + f"argument '{key}' must be a string starting with '{required_prefix}'." + ) + + +def _violation(message: str) -> TokenScopeError: + """Build a :class:`TokenScopeError` for an argument-constraint violation.""" + return TokenScopeError(message, reason_code=DenialReason.ARG_CONSTRAINT_VIOLATION) + + +def _check_invoke_rate( + kernel: Kernel, + token: CapabilityToken, + capability: Capability, + principal: Principal, + *, + dry_run: bool, +) -> None: + """Apply the optional invoke-time sliding-window rate limit (#170). + + Independent of and additional to the grant-time limit. The check-then-record + pair runs with no ``await`` between them, so concurrent invokes cannot + over-admit. Dry-run checks but never records. + """ + limits = kernel._invoke_rate_limits + limit_window = limits.get(capability.safety_class) + if limit_window is None: + return + limit, window = limit_window + key = f"{principal.principal_id}:{token.capability_id}" + if not kernel._invoke_limiter.check(key, limit, window): + raise PolicyDenied( + f"Invoke-time rate limit exceeded: {limit} {capability.safety_class.value} " + f"invocations per {window}s for principal '{principal.principal_id}'.", + reason_code=DenialReason.RATE_LIMITED, + ) + if not dry_run: + kernel._invoke_limiter.record(key) + + +def run_pre_invoke_checks( + kernel: Kernel, + *, + token: CapabilityToken, + capability: Capability, + principal: Principal, + args: dict[str, Any], + response_mode: ResponseMode, + dry_run: bool, +) -> None: + """Run invoke-time argument-constraint and rate-limit checks (#183, #170). + + Args: + kernel: The orchestrating :class:`Kernel`. + token: The already-verified capability token. + capability: The resolved capability (its sensitivity tags the audit trace). + principal: The invoking principal. + args: The invocation arguments. + response_mode: The caller-requested response mode (recorded on a denial trace). + dry_run: When ``True``, evaluate the checks for parity but record no + rate-limit usage and write no audit trace. + + Raises: + TokenScopeError: If *args* violate a signed argument constraint (#183). + PolicyDenied: If the invoke-time rate limit is exceeded (#170). + """ + try: + enforce_arg_constraints(args, token.constraints) + _check_invoke_rate(kernel, token, capability, principal, dry_run=dry_run) + except (TokenScopeError, PolicyDenied) as exc: + if not dry_run: + # I-02: a denied execution attempt is still auditable. No driver ran + # and no budget was reserved, so record a failure trace directly. + record_failure_trace( + action_id=str(uuid.uuid4()), + capability_id=token.capability_id, + principal_id=principal.principal_id, + token_id=token.token_id, + args=args, + response_mode=response_mode, + error_message=str(exc), + trace_store=kernel._traces, + sensitivity=capability.sensitivity, + driver_id="", + ) + kernel._stats.on_invocation( + failed=True, fallback=False, redacted=False, downgraded=False + ) + raise + + +# Re-exported for the kernel constructor's clock injection typing. +InvokeRateClock = Callable[[], float] + +__all__ = [ + "ARG_CONSTRAINTS_KEY", + "enforce_arg_constraints", + "run_pre_invoke_checks", + "validate_invoke_rate_limits", + "InvokeRateClock", +] diff --git a/src/weaver_kernel/kernel/_grant.py b/src/weaver_kernel/kernel/_grant.py new file mode 100644 index 0000000..7e60ffe --- /dev/null +++ b/src/weaver_kernel/kernel/_grant.py @@ -0,0 +1,144 @@ +"""Grant-capability orchestration and per-grant TTL enforcement (#203). + +Extracted from :mod:`weaver_kernel.kernel` to keep the public API module within +the AGENTS.md 300-line budget (mirrors the ``_invoke``/``_dry_run`` split). The +per-grant TTL is validated *before* policy evaluation so a doomed grant never +consumes rate-limit quota, then threaded into token issuance. +""" + +from __future__ import annotations + +import logging +import uuid +from typing import TYPE_CHECKING + +from ..errors import PolicyDenied +from ..models import Capability, CapabilityGrant, CapabilityRequest, Principal +from ..policy import PolicyEngine +from ..policy_reasons import DenialReason +from ..policy_ttl import resolve_max_ttl_s +from ._audit import record_denial_trace + +if TYPE_CHECKING: # pragma: no cover + from . import Kernel + +logger = logging.getLogger("weaver_kernel.kernel") + + +def _validate_ttl(policy: PolicyEngine, capability: Capability, ttl_s: int | None) -> None: + """Deny a per-grant TTL that is non-positive or over the policy maximum (#203). + + Args: + policy: The active policy engine; consulted (duck-typed) for + ``max_ttl_s`` so third-party engines without it impose no maximum. + capability: The capability being granted. + ttl_s: The requested TTL in seconds, or ``None`` for the provider default. + + Raises: + PolicyDenied: If *ttl_s* is non-positive (``INVALID_CONSTRAINT``) or + exceeds the policy maximum (``TTL_EXCEEDED``). Never clamps silently. + """ + if ttl_s is None: + return + if ttl_s <= 0: + raise PolicyDenied( + f"Requested ttl_s must be a positive number of seconds, got {ttl_s}.", + reason_code=DenialReason.INVALID_CONSTRAINT, + ) + max_ttl = resolve_max_ttl_s(getattr(policy, "max_ttl_s", None), capability) + if max_ttl is not None and ttl_s > max_ttl: + raise PolicyDenied( + f"Requested ttl_s={ttl_s} exceeds the maximum of {max_ttl}s for " + f"{capability.safety_class.value} capabilities.", + reason_code=DenialReason.TTL_EXCEEDED, + ) + + +def perform_grant( + kernel: Kernel, + request: CapabilityRequest, + principal: Principal, + *, + justification: str, + ttl_s: int | None, +) -> CapabilityGrant: + """Evaluate the policy and, if approved, issue a signed token. + + On a :class:`~weaver_kernel.PolicyDenied` rejection — including a TTL denial + raised before evaluation — a ``"deny"`` audit record (carrying the stable + reason code) is written to the trace store (best-effort) before the exception + propagates. A trace-store write failure is logged but never masks the denial. + + Args: + kernel: The orchestrating :class:`Kernel` (private accessors used for the + registry, policy, token provider, trace store, and stats). + request: The capability request being granted. + principal: The principal the grant is issued to. + justification: Free-text justification forwarded to the policy engine. + ttl_s: Optional per-grant token TTL in seconds; ``None`` uses the token + provider's default. + + Returns: + The issued :class:`~weaver_kernel.models.CapabilityGrant`. + """ + capability = kernel._registry.get(request.capability_id) + try: + _validate_ttl(kernel._policy, capability, ttl_s) + decision = kernel._policy.evaluate( + request, capability, principal, justification=justification + ) + except PolicyDenied as exc: + kernel._stats.on_denial(exc.reason_code) + # The denial is authoritative and already fails closed (no token is + # issued). Recording its audit trace is best-effort: a trace-store write + # failure must never mask the PolicyDenied the caller expects. + try: + record_denial_trace( + capability_id=request.capability_id, + principal_id=principal.principal_id, + reason_code=exc.reason_code, + message=str(exc), + trace_store=kernel._traces, + ) + except Exception: + logger.warning( + "deny_trace_record_failed", + extra={ + "capability_id": request.capability_id, + "principal_id": principal.principal_id, + "reason_code": exc.reason_code, + }, + exc_info=True, + ) + raise + audit_id = str(uuid.uuid4()) + issue_kwargs = {} if ttl_s is None else {"ttl_seconds": ttl_s} + token = kernel._token_provider.issue( + capability.capability_id, + principal.principal_id, + constraints=decision.constraints, + audit_id=audit_id, + **issue_kwargs, + ) + logger.info( + "grant_capability", + extra={ + "principal_id": principal.principal_id, + "capability_id": capability.capability_id, + "safety_class": capability.safety_class.value, + "audit_id": audit_id, + "token_id": token.token_id, + "ttl_s": ttl_s, + }, + ) + kernel._stats.on_grant() + return CapabilityGrant( + request=request, + principal=principal, + decision=decision, + token=token, + audit_id=audit_id, + ) + + +__all__ = ["perform_grant"] diff --git a/src/weaver_kernel/otel.py b/src/weaver_kernel/otel.py index e21a688..4a01947 100644 --- a/src/weaver_kernel/otel.py +++ b/src/weaver_kernel/otel.py @@ -204,6 +204,7 @@ def instrumented_grant( principal: Any, *, justification: str, + ttl_s: int | None = None, ) -> Any: attributes: dict[str, Any] = { ATTR_PRINCIPAL: principal.principal_id, @@ -211,7 +212,7 @@ def instrumented_grant( } with tracer.start_as_current_span("weaver_kernel.grant", attributes=attributes) as span: try: - return original_grant(request, principal, justification=justification) + return original_grant(request, principal, justification=justification, ttl_s=ttl_s) except Exception as exc: reason_code = getattr(exc, "reason_code", "") or "" denials.add( diff --git a/src/weaver_kernel/policy.py b/src/weaver_kernel/policy.py index 2d7dde3..9d4a963 100644 --- a/src/weaver_kernel/policy.py +++ b/src/weaver_kernel/policy.py @@ -19,6 +19,7 @@ Principal, ) from .policy_reasons import AllowReason, DenialReason +from .policy_ttl import validate_max_ttl_s from .rate_limit import DEFAULT_RATE_LIMITS, SERVICE_RATE_MULTIPLIER, RateLimiter logger = logging.getLogger(__name__) @@ -30,11 +31,6 @@ _MAX_ROWS_USER = 50 _MAX_ROWS_SERVICE = 500 -# Backwards-compatible aliases — these used to be defined here. New code -# should import the names without the leading underscore from ``rate_limit``. -_DEFAULT_RATE_LIMITS = DEFAULT_RATE_LIMITS -_SERVICE_RATE_MULTIPLIER = SERVICE_RATE_MULTIPLIER - class PolicyEngine(Protocol): """Interface for a policy engine. @@ -133,6 +129,7 @@ def __init__( *, rate_limits: dict[SafetyClass, tuple[int, float]] | None = None, clock: Callable[[], float] | None = None, + max_ttl_s: int | dict[SafetyClass, int] | None = None, ) -> None: """Initialise the policy engine. @@ -143,8 +140,9 @@ def __init__( unspecified safety classes retain their default limits. clock: Monotonic clock callable for rate-limiter. Defaults to :func:`time.monotonic`. + max_ttl_s: Maximum per-grant token TTL in seconds — one cap or a per-safety-class map; ``None`` = uncapped. A longer request is denied, not clamped (#203). """ - limits = dict(_DEFAULT_RATE_LIMITS) + limits = dict(DEFAULT_RATE_LIMITS) if rate_limits is not None: limits.update(rate_limits) for sc, (count, window) in limits.items(): @@ -154,8 +152,10 @@ def __init__( f"limit must be >= 1 and window must be > 0, " f"got limit={count}, window={window}." ) + validate_max_ttl_s(max_ttl_s) self._rate_limits = limits self._limiter = RateLimiter(clock=clock) + self.max_ttl_s = max_ttl_s @staticmethod def _deny( @@ -420,7 +420,7 @@ def _record_deny(detail: str, code: str) -> None: if capability.safety_class in self._rate_limits: limit, window = self._rate_limits[capability.safety_class] if "service" in roles: - limit *= _SERVICE_RATE_MULTIPLIER + limit *= SERVICE_RATE_MULTIPLIER if not self._limiter.check(rate_key, limit, window): detail = ( f"Rate limit exceeded: {limit} {capability.safety_class.value} " diff --git a/src/weaver_kernel/policy_reasons.py b/src/weaver_kernel/policy_reasons.py index 8c00f35..a73ebc2 100644 --- a/src/weaver_kernel/policy_reasons.py +++ b/src/weaver_kernel/policy_reasons.py @@ -54,6 +54,12 @@ class DenialReason(_StrEnumCompat): INVALID_CONSTRAINT = "invalid_constraint" """A constraint value (e.g. ``max_rows``) is not parseable or in range.""" + TTL_EXCEEDED = "ttl_exceeded" + """A requested per-grant token TTL exceeds the policy maximum (#203).""" + + ARG_CONSTRAINT_VIOLATION = "arg_constraint_violation" + """Invocation arguments violated a signed ``constraints["args"]`` rule (#183).""" + # Rate limiting RATE_LIMITED = "rate_limited" """The sliding-window rate limit for this principal/capability was exceeded.""" diff --git a/src/weaver_kernel/policy_ttl.py b/src/weaver_kernel/policy_ttl.py new file mode 100644 index 0000000..f24f9cd --- /dev/null +++ b/src/weaver_kernel/policy_ttl.py @@ -0,0 +1,59 @@ +"""Per-grant TTL validation and resolution (#203). + +Extracted from :mod:`weaver_kernel.policy` to keep that module within the +AGENTS.md 300-line budget (it is already at its ratchet ceiling). The maximum +per-grant token TTL is policy configuration; :class:`DefaultPolicyEngine` stores +the raw value and delegates validation and per-capability resolution here. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .enums import SafetyClass +from .errors import AgentKernelError + +if TYPE_CHECKING: # pragma: no cover + from .models import Capability + +MaxTTLConfig = int | dict[SafetyClass, int] | None +"""A single TTL cap, a per-safety-class mapping, or ``None`` for no maximum.""" + + +def validate_max_ttl_s(max_ttl_s: MaxTTLConfig) -> None: + """Reject a non-positive ``max_ttl_s`` configuration. + + Args: + max_ttl_s: The configured maximum TTL (scalar, per-class map, or ``None``). + + Raises: + AgentKernelError: If any configured value is not a positive integer. + """ + if max_ttl_s is None: + return + values = max_ttl_s.values() if isinstance(max_ttl_s, dict) else [max_ttl_s] + for value in values: + if not isinstance(value, int) or isinstance(value, bool) or value <= 0: + raise AgentKernelError( + f"Invalid max_ttl_s: values must be positive integers, got {value!r}." + ) + + +def resolve_max_ttl_s(max_ttl_s: MaxTTLConfig, capability: Capability) -> int | None: + """Return the maximum allowed TTL for *capability*, or ``None`` if uncapped. + + Args: + max_ttl_s: The configured maximum TTL (scalar, per-class map, or ``None``). + capability: The capability whose grant TTL is being bounded. + + Returns: + The maximum TTL in seconds, or ``None`` when no cap applies. + """ + if max_ttl_s is None: + return None + if isinstance(max_ttl_s, dict): + return max_ttl_s.get(capability.safety_class) + return max_ttl_s + + +__all__ = ["MaxTTLConfig", "validate_max_ttl_s", "resolve_max_ttl_s"] diff --git a/src/weaver_kernel/tokens.py b/src/weaver_kernel/tokens.py index 8b3a450..d4389a7 100644 --- a/src/weaver_kernel/tokens.py +++ b/src/weaver_kernel/tokens.py @@ -1,22 +1,19 @@ -"""HMAC-SHA256 token provider for capability authorization.""" +"""Capability tokens: the :class:`CapabilityToken` dataclass and the +:class:`TokenProvider` Protocol. + +The concrete :class:`HMACTokenProvider` lives in +:mod:`weaver_kernel._hmac_provider` (extracted to honour the AGENTS.md +300-line module budget) and is re-exported here so +``from weaver_kernel.tokens import HMACTokenProvider`` keeps working. +""" from __future__ import annotations import datetime -import hashlib -import hmac -import json -import logging -import uuid from dataclasses import dataclass, field from typing import Any, Protocol -from ._secrets import _get_secret -from .errors import TokenExpired, TokenInvalid, TokenRevoked, TokenScopeError -from .stores import InMemoryRevocationStore, RevocationStoreProtocol - -logger = logging.getLogger(__name__) - +from ._token_signing import build_signable_payload, parse_token_fields # ── Token dataclass ─────────────────────────────────────────────────────────── @@ -38,21 +35,13 @@ class CapabilityToken: constraints: dict[str, Any] = field(default_factory=dict) audit_id: str = "" signature: str = "" + key_id: str = "" # ── Serialization ───────────────────────────────────────────────────────── def _signable_payload(self) -> str: """Return the canonical JSON string used as the HMAC message.""" - payload = { - "token_id": self.token_id, - "capability_id": self.capability_id, - "principal_id": self.principal_id, - "issued_at": self.issued_at.isoformat(), - "expires_at": self.expires_at.isoformat(), - "constraints": self.constraints, - "audit_id": self.audit_id, - } - return json.dumps(payload, sort_keys=True, separators=(",", ":")) + return build_signable_payload(self) def to_dict(self) -> dict[str, Any]: """Serialise the token to a plain dict (suitable for JSON transport).""" @@ -65,21 +54,25 @@ def to_dict(self) -> dict[str, Any]: "constraints": self.constraints, "audit_id": self.audit_id, "signature": self.signature, + "key_id": self.key_id, } @classmethod def from_dict(cls, data: dict[str, Any]) -> CapabilityToken: - """Reconstruct a token from a plain dict.""" - return cls( - token_id=data["token_id"], - capability_id=data["capability_id"], - principal_id=data["principal_id"], - issued_at=datetime.datetime.fromisoformat(data["issued_at"]), - expires_at=datetime.datetime.fromisoformat(data["expires_at"]), - constraints=data.get("constraints", {}), - audit_id=data.get("audit_id", ""), - signature=data.get("signature", ""), - ) + """Reconstruct a token from a plain dict. + + Args: + data: A serialized token, e.g. from :meth:`to_dict` or an untrusted + transport source. + + Returns: + The reconstructed :class:`CapabilityToken`. + + Raises: + TokenInvalid: If *data* is missing a required field, has a field of + the wrong type, or carries a malformed timestamp (#200). + """ + return cls(**parse_token_fields(data)) # ── Protocol ────────────────────────────────────────────────────────────────── @@ -154,183 +147,10 @@ def revoke_all(self, principal_id: str) -> int: ... -# ── Implementation ──────────────────────────────────────────────────────────── - - -class HMACTokenProvider: - """Issues and verifies HMAC-SHA256 capability tokens. - - The signing secret is read from the ``WEAVER_KERNEL_SECRET`` environment - variable. If the variable is absent a random development secret is - generated and a warning is logged. - """ - - def __init__( - self, - secret: str | None = None, - *, - revocation_store: RevocationStoreProtocol | None = None, - ) -> None: - self._secret = secret # None → use env / dev fallback at call time - # Revocation state lives behind a protocol so it can be made durable - # (e.g. SQLiteRevocationStore) without weakening verify-before-invoke. - self._revocation: RevocationStoreProtocol = revocation_store or InMemoryRevocationStore() - - @staticmethod - def _log_verify_failure(token_id: str, reason: str, **extra: Any) -> None: - """Log a token verification failure at WARNING.""" - logger.warning( - "token_verify_failed", - extra={"token_id": token_id, "reason": reason, **extra}, - ) - - def _secret_bytes(self) -> bytes: - return (self._secret or _get_secret()).encode() +# ── Implementation (re-exported) ────────────────────────────────────────────── - def _sign(self, payload: str) -> str: - return hmac.new(self._secret_bytes(), payload.encode(), hashlib.sha256).hexdigest() +# HMACTokenProvider lives in a sibling module to keep this file within the +# AGENTS.md 300-line budget; re-exported so its public import path is unchanged. +from ._hmac_provider import HMACTokenProvider # noqa: E402 - def issue( - self, - capability_id: str, - principal_id: str, - *, - constraints: dict[str, Any] | None = None, - ttl_seconds: int = 3600, - audit_id: str = "", - ) -> CapabilityToken: - """Issue a new signed token. - - Args: - capability_id: The capability this token authorises. - principal_id: The principal this token is issued to. - constraints: Optional execution constraints. - ttl_seconds: How long the token is valid (default 1 hour). - audit_id: Audit trail ID to embed in the token. - - Returns: - A freshly signed :class:`CapabilityToken`. - """ - now = datetime.datetime.now(tz=datetime.timezone.utc) - token = CapabilityToken( - token_id=str(uuid.uuid4()), - capability_id=capability_id, - principal_id=principal_id, - issued_at=now, - expires_at=now + datetime.timedelta(seconds=ttl_seconds), - constraints=constraints or {}, - audit_id=audit_id, - ) - token.signature = self._sign(token._signable_payload()) - self._revocation.track(principal_id, token.token_id, token.expires_at) - logger.debug( - "token_issued", - extra={ - "token_id": token.token_id, - "capability_id": capability_id, - "principal_id": principal_id, - "audit_id": audit_id, - "expires_at": token.expires_at.isoformat(), - }, - ) - return token - - def revoke(self, token_id: str) -> None: - """Revoke a single token by ID. - - Idempotent — revoking an already-revoked or unknown token is a no-op. - - Args: - token_id: The ID of the token to revoke. - """ - self._revocation.revoke(token_id) - - def revoke_all(self, principal_id: str) -> int: - """Revoke all tokens issued to a principal. - - Args: - principal_id: The principal whose tokens should be revoked. - - Returns: - The number of tokens newly revoked by this call (excluding tokens - that were already revoked). - """ - return self._revocation.revoke_principal(principal_id) - - def sweep_revocations(self, now: datetime.datetime | None = None) -> int: - """Drop revocation bookkeeping for tokens that have already expired. - - Bounds revocation-state growth in long-lived processes (#182). Safe to - call at any time: an expired token fails the verifier's expiry check - regardless, so sweeping its entry never un-revokes a live token. The - in-memory store also sweeps itself lazily; durable backends expose this - for an operator to call on a schedule. - - Args: - now: Reference time; defaults to the current UTC time. - - Returns: - The number of tracked tokens whose state was removed. - """ - when = now or datetime.datetime.now(tz=datetime.timezone.utc) - return self._revocation.sweep_expired(when) - - def verify( - self, - token: CapabilityToken, - *, - expected_principal_id: str, - expected_capability_id: str, - ) -> None: - """Verify a token's signature, expiry, and scope bindings. - - Args: - token: The token to verify. - expected_principal_id: The principal that should own this token. - expected_capability_id: The capability this token should authorize. - - Raises: - TokenRevoked: If the token has been revoked. - TokenExpired: If ``token.expires_at`` is in the past. - TokenInvalid: If the HMAC signature does not verify. - TokenScopeError: If principal or capability do not match. - """ - # 0. Revocation (fast lookup before any crypto) - if self._revocation.is_revoked(token.token_id): - self._log_verify_failure(token.token_id, "revoked") - raise TokenRevoked(f"Token '{token.token_id}' has been revoked.") - - # 1. Expiry - now = datetime.datetime.now(tz=datetime.timezone.utc) - if token.expires_at <= now: - self._log_verify_failure( - token.token_id, "expired", expires_at=token.expires_at.isoformat() - ) - raise TokenExpired( - f"Token '{token.token_id}' expired at {token.expires_at.isoformat()}." - ) - - # 2. Signature - expected_sig = self._sign(token._signable_payload()) - if not hmac.compare_digest(expected_sig, token.signature): - self._log_verify_failure(token.token_id, "invalid_signature") - raise TokenInvalid( - f"Token '{token.token_id}' has an invalid signature. " - "The token may have been tampered with." - ) - - # 3. Principal binding (confused-deputy prevention) - if token.principal_id != expected_principal_id: - self._log_verify_failure(token.token_id, "principal_mismatch") - raise TokenScopeError( - f"Token '{token.token_id}' was issued for principal " - f"'{token.principal_id}', not '{expected_principal_id}'." - ) - - # 4. Capability binding - if token.capability_id != expected_capability_id: - self._log_verify_failure(token.token_id, "capability_mismatch") - raise TokenScopeError( - f"Token '{token.token_id}' was issued for capability " - f"'{token.capability_id}', not '{expected_capability_id}'." - ) +__all__ = ["CapabilityToken", "TokenProvider", "HMACTokenProvider"] diff --git a/tests/test_architecture.py b/tests/test_architecture.py index afd428a..3bd86d9 100644 --- a/tests/test_architecture.py +++ b/tests/test_architecture.py @@ -50,14 +50,15 @@ "__init__.py": 341, "models.py": 753, "policy.py": 652, - "kernel/__init__.py": 541, + "kernel/__init__.py": 540, "adapters/_base.py": 459, "kernel/_invoke.py": 390, "firewall/transform.py": 377, "adapters/openai.py": 358, "stores/sqlite.py": 350, - "tokens.py": 336, "federation_discovery.py": 306, + # tokens.py was split into _token_signing.py + _hmac_provider.py (#185) and + # is now well under the 300-line budget, so it is no longer ratcheted. } _LINE_BUDGET = 300 diff --git a/tests/test_kernel.py b/tests/test_kernel.py index b07b49f..38c1fdb 100644 --- a/tests/test_kernel.py +++ b/tests/test_kernel.py @@ -19,6 +19,7 @@ SafetyClass, StaticRouter, TokenExpired, + TokenScopeError, ) from weaver_kernel.drivers.base import ExecutionContext from weaver_kernel.errors import FirewallError @@ -1233,3 +1234,376 @@ def test_kernel_query_traces(kernel: Kernel, reader_principal: Principal) -> Non assert denied[0].capability_id == "billing.delete_invoice" # Filtering by a principal who did nothing yields nothing. assert kernel.query_traces(TraceQuery(principal_id="ghost")) == [] + + +# ── Per-grant TTL (#203) ─────────────────────────────────────────────────────── + + +def _read_req() -> CapabilityRequest: + return CapabilityRequest(capability_id="billing.list_invoices", goal="lookup") + + +def test_grant_ttl_s_sets_token_expiry(kernel: Kernel, reader_principal: Principal) -> None: + grant = kernel.grant_capability(_read_req(), reader_principal, justification="", ttl_s=60) + delta = (grant.token.expires_at - grant.token.issued_at).total_seconds() + assert delta == 60 + + +def test_grant_default_ttl_unchanged(kernel: Kernel, reader_principal: Principal) -> None: + grant = kernel.grant_capability(_read_req(), reader_principal, justification="") + delta = (grant.token.expires_at - grant.token.issued_at).total_seconds() + assert delta == 3600 + + +def test_grant_get_token_threads_ttl(kernel: Kernel, reader_principal: Principal) -> None: + token = kernel.get_token(_read_req(), reader_principal, justification="", ttl_s=120) + assert (token.expires_at - token.issued_at).total_seconds() == 120 + + +@pytest.mark.parametrize("bad_ttl", [0, -5]) +def test_grant_non_positive_ttl_denied( + kernel: Kernel, reader_principal: Principal, bad_ttl: int +) -> None: + with pytest.raises(PolicyDenied) as exc_info: + kernel.grant_capability(_read_req(), reader_principal, justification="", ttl_s=bad_ttl) + assert exc_info.value.reason_code == "invalid_constraint" + + +def _capped_kernel(registry: CapabilityRegistry, memory_driver: InMemoryDriver) -> Kernel: + from weaver_kernel import DefaultPolicyEngine + + router = StaticRouter(routes={"billing.list_invoices": ["memory"]}) + k = Kernel( + registry=registry, + policy=DefaultPolicyEngine(max_ttl_s={SafetyClass.READ: 30}), + token_provider=HMACTokenProvider(secret="test-secret-do-not-use-in-prod"), + router=router, + ) + k.register_driver(memory_driver) + return k + + +def test_grant_ttl_over_max_denied( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + capped = _capped_kernel(registry, memory_driver) + with pytest.raises(PolicyDenied) as exc_info: + capped.grant_capability(_read_req(), reader_principal, justification="", ttl_s=120) + assert exc_info.value.reason_code == "ttl_exceeded" + + +def test_grant_ttl_within_max_allowed( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + capped = _capped_kernel(registry, memory_driver) + grant = capped.grant_capability(_read_req(), reader_principal, justification="", ttl_s=30) + assert (grant.token.expires_at - grant.token.issued_at).total_seconds() == 30 + + +def test_grant_ttl_denial_is_audited( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + capped = _capped_kernel(registry, memory_driver) + with pytest.raises(PolicyDenied): + capped.grant_capability(_read_req(), reader_principal, justification="", ttl_s=999) + traces = capped.list_traces() + assert any(t.event_type == "deny" and t.reason_code == "ttl_exceeded" for t in traces), ( + "TTL denial should be recorded as a deny audit trace" + ) + + +def test_grant_ttl_graceful_when_engine_has_no_max( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + """A policy engine without ``max_ttl_s`` imposes no cap; ttl_s is still honored.""" + from weaver_kernel.models import PolicyDecision + + class _StubPolicy: + def evaluate(self, request, capability, principal, *, justification): # type: ignore[no-untyped-def] + return PolicyDecision(allowed=True, reason="ok", constraints={}) + + router = StaticRouter(routes={"billing.list_invoices": ["memory"]}) + k = Kernel( + registry=registry, + policy=_StubPolicy(), # type: ignore[arg-type] + token_provider=HMACTokenProvider(secret="test-secret-do-not-use-in-prod"), + router=router, + ) + k.register_driver(memory_driver) + grant = k.grant_capability(_read_req(), reader_principal, justification="", ttl_s=99999) + assert (grant.token.expires_at - grant.token.issued_at).total_seconds() == 99999 + # A non-positive TTL is still rejected regardless of engine capabilities. + with pytest.raises(PolicyDenied): + k.grant_capability(_read_req(), reader_principal, justification="", ttl_s=-1) + + +# ── Signed argument-level constraints at invoke time (#183) ──────────────────── + + +def _req_with_args(spec: dict[str, object]) -> CapabilityRequest: + return CapabilityRequest( + capability_id="billing.list_invoices", goal="lookup", constraints={"args": spec} + ) + + +@pytest.mark.asyncio +async def test_arg_constraint_allowed_keys_violation_denied( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args({"allowed_keys": ["operation"]}), reader_principal, justification="" + ) + with pytest.raises(TokenScopeError) as exc_info: + await kernel.invoke( + token, + principal=reader_principal, + args={"operation": "billing.list_invoices", "leak": "x"}, + ) + assert exc_info.value.reason_code == "arg_constraint_violation" + + +@pytest.mark.asyncio +async def test_arg_constraint_allowed_keys_compliant_passes( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args({"allowed_keys": ["operation"]}), reader_principal, justification="" + ) + frame = await kernel.invoke( + token, principal=reader_principal, args={"operation": "billing.list_invoices"} + ) + assert frame.action_id != "" + + +@pytest.mark.asyncio +async def test_arg_constraint_pinned_mismatch_denied( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args({"pinned": {"customer_id": "c123"}}), reader_principal, justification="" + ) + with pytest.raises(TokenScopeError, match="pinned"): + await kernel.invoke( + token, + principal=reader_principal, + args={"operation": "billing.list_invoices", "customer_id": "c999"}, + ) + + +@pytest.mark.asyncio +async def test_arg_constraint_pinned_missing_key_fails_closed( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args({"pinned": {"customer_id": "c123"}}), reader_principal, justification="" + ) + with pytest.raises(TokenScopeError): + await kernel.invoke( + token, principal=reader_principal, args={"operation": "billing.list_invoices"} + ) + + +@pytest.mark.asyncio +async def test_arg_constraint_prefix_violation_denied( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args({"prefix": {"path": "/safe/"}}), reader_principal, justification="" + ) + with pytest.raises(TokenScopeError, match="starting with"): + await kernel.invoke( + token, + principal=reader_principal, + args={"operation": "billing.list_invoices", "path": "/etc/passwd"}, + ) + + +@pytest.mark.asyncio +async def test_arg_constraint_prefix_compliant_passes( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args({"prefix": {"path": "/safe/"}}), reader_principal, justification="" + ) + frame = await kernel.invoke( + token, + principal=reader_principal, + args={"operation": "billing.list_invoices", "path": "/safe/report.csv"}, + ) + assert frame.action_id != "" + + +@pytest.mark.asyncio +async def test_arg_constraint_dry_run_parity(kernel: Kernel, reader_principal: Principal) -> None: + """Dry-run raises the same TokenScopeError a real invoke would (#183).""" + token = kernel.get_token( + _req_with_args({"allowed_keys": ["operation"]}), reader_principal, justification="" + ) + with pytest.raises(TokenScopeError): + await kernel.invoke( + token, + principal=reader_principal, + args={"operation": "billing.list_invoices", "leak": "x"}, + dry_run=True, + ) + + +@pytest.mark.asyncio +async def test_arg_constraint_violation_records_failure_trace( + kernel: Kernel, reader_principal: Principal +) -> None: + """A denied invoke is audited (I-02) with no driver reached; dry-run is not.""" + token = kernel.get_token( + _req_with_args({"allowed_keys": ["operation"]}), reader_principal, justification="" + ) + with pytest.raises(TokenScopeError): + await kernel.invoke( + token, + principal=reader_principal, + args={"operation": "billing.list_invoices", "leak": "x"}, + ) + failures = [t for t in kernel.list_traces() if t.error and t.driver_id == ""] + assert failures and "allowed_keys" in failures[-1].error + + +# ── Per-invocation rate limiting (#170) ──────────────────────────────────────── + + +def _rate_limited_kernel( + registry: CapabilityRegistry, + memory_driver: InMemoryDriver, + limits: dict[SafetyClass, tuple[int, float]], + clock, # type: ignore[no-untyped-def] +) -> Kernel: + router = StaticRouter(routes={"billing.list_invoices": ["memory"]}) + k = Kernel( + registry=registry, + token_provider=HMACTokenProvider(secret="test-secret-do-not-use-in-prod"), + router=router, + invoke_rate_limits=limits, + invoke_rate_clock=clock, + ) + k.register_driver(memory_driver) + return k + + +@pytest.mark.asyncio +async def test_invoke_rate_limit_blocks_over_limit( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + now = [1000.0] + k = _rate_limited_kernel( + registry, memory_driver, {SafetyClass.READ: (2, 60.0)}, lambda: now[0] + ) + token = k.get_token(_read_req(), reader_principal, justification="") + args = {"operation": "billing.list_invoices"} + await k.invoke(token, principal=reader_principal, args=args) + await k.invoke(token, principal=reader_principal, args=args) + with pytest.raises(PolicyDenied) as exc_info: + await k.invoke(token, principal=reader_principal, args=args) + assert exc_info.value.reason_code == "rate_limited" + + +@pytest.mark.asyncio +async def test_invoke_rate_limit_default_off(kernel: Kernel, reader_principal: Principal) -> None: + token = kernel.get_token(_read_req(), reader_principal, justification="") + args = {"operation": "billing.list_invoices"} + for _ in range(20): + await kernel.invoke(token, principal=reader_principal, args=args) + + +@pytest.mark.asyncio +async def test_invoke_rate_limit_window_reset( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + now = [1000.0] + k = _rate_limited_kernel( + registry, memory_driver, {SafetyClass.READ: (1, 60.0)}, lambda: now[0] + ) + token = k.get_token(_read_req(), reader_principal, justification="") + args = {"operation": "billing.list_invoices"} + await k.invoke(token, principal=reader_principal, args=args) + with pytest.raises(PolicyDenied): + await k.invoke(token, principal=reader_principal, args=args) + now[0] += 61.0 # slide past the window + await k.invoke(token, principal=reader_principal, args=args) + + +@pytest.mark.asyncio +async def test_invoke_rate_limit_is_per_principal( + registry: CapabilityRegistry, + memory_driver: InMemoryDriver, + reader_principal: Principal, + service_principal: Principal, +) -> None: + now = [1000.0] + k = _rate_limited_kernel( + registry, memory_driver, {SafetyClass.READ: (1, 60.0)}, lambda: now[0] + ) + args = {"operation": "billing.list_invoices"} + t1 = k.get_token(_read_req(), reader_principal, justification="") + t2 = k.get_token(_read_req(), service_principal, justification="") + await k.invoke(t1, principal=reader_principal, args=args) + # A different principal has an independent window. + await k.invoke(t2, principal=service_principal, args=args) + + +@pytest.mark.asyncio +async def test_invoke_rate_limit_dry_run_does_not_consume( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + now = [1000.0] + k = _rate_limited_kernel( + registry, memory_driver, {SafetyClass.READ: (1, 60.0)}, lambda: now[0] + ) + token = k.get_token(_read_req(), reader_principal, justification="") + args = {"operation": "billing.list_invoices"} + # Many dry-runs consume nothing... + for _ in range(5): + await k.invoke(token, principal=reader_principal, args=args, dry_run=True) + # ...so a real invoke still fits within the limit of 1. + await k.invoke(token, principal=reader_principal, args=args) + + +@pytest.mark.asyncio +async def test_invoke_rate_limit_concurrent_calls_do_not_exceed_limit( + registry: CapabilityRegistry, memory_driver: InMemoryDriver, reader_principal: Principal +) -> None: + now = [1000.0] + k = _rate_limited_kernel( + registry, memory_driver, {SafetyClass.READ: (3, 60.0)}, lambda: now[0] + ) + token = k.get_token(_read_req(), reader_principal, justification="") + args = {"operation": "billing.list_invoices"} + results = await asyncio.gather( + *(k.invoke(token, principal=reader_principal, args=args) for _ in range(10)), + return_exceptions=True, + ) + admitted = sum(1 for r in results if not isinstance(r, Exception)) + denied = sum(1 for r in results if isinstance(r, PolicyDenied)) + assert admitted == 3 + assert denied == 7 + + +def test_invalid_invoke_rate_limits_rejected_at_construction( + registry: CapabilityRegistry, +) -> None: + from weaver_kernel import AgentKernelError + + with pytest.raises(AgentKernelError, match="invoke_rate_limits"): + Kernel(registry=registry, invoke_rate_limits={SafetyClass.READ: (0, 60.0)}) + + +@pytest.mark.asyncio +async def test_malformed_args_constraint_denied( + kernel: Kernel, reader_principal: Principal +) -> None: + token = kernel.get_token( + _req_with_args("not-a-dict"), # type: ignore[arg-type] + reader_principal, + justification="", + ) + with pytest.raises(TokenScopeError, match="malformed 'args' constraint"): + await kernel.invoke( + token, principal=reader_principal, args={"operation": "billing.list_invoices"} + ) diff --git a/tests/test_policy.py b/tests/test_policy.py index 8cdaed6..171ee86 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -1835,3 +1835,38 @@ def test_policy_denied_default_reason_code_is_none() -> None: def test_policy_denied_carries_reason_code() -> None: err = PolicyDenied("msg", reason_code=DenialReason.MISSING_ROLE) assert err.reason_code == DenialReason.MISSING_ROLE + + +# ── Per-grant TTL configuration (#203) ───────────────────────────────────────── + + +def test_max_ttl_s_scalar_resolves_for_all_classes() -> None: + from weaver_kernel.policy_ttl import resolve_max_ttl_s + + engine = DefaultPolicyEngine(max_ttl_s=300) + assert engine.max_ttl_s == 300 + assert resolve_max_ttl_s(engine.max_ttl_s, _cap("c", SafetyClass.READ)) == 300 + assert resolve_max_ttl_s(engine.max_ttl_s, _cap("c", SafetyClass.DESTRUCTIVE)) == 300 + + +def test_max_ttl_s_per_class_map_resolves_and_defaults_to_none() -> None: + from weaver_kernel.policy_ttl import resolve_max_ttl_s + + engine = DefaultPolicyEngine(max_ttl_s={SafetyClass.READ: 60}) + assert resolve_max_ttl_s(engine.max_ttl_s, _cap("c", SafetyClass.READ)) == 60 + # A class absent from the map is uncapped. + assert resolve_max_ttl_s(engine.max_ttl_s, _cap("c", SafetyClass.WRITE)) is None + + +def test_max_ttl_s_none_is_uncapped() -> None: + from weaver_kernel.policy_ttl import resolve_max_ttl_s + + engine = DefaultPolicyEngine() + assert engine.max_ttl_s is None + assert resolve_max_ttl_s(engine.max_ttl_s, _cap("c", SafetyClass.READ)) is None + + +@pytest.mark.parametrize("bad", [0, -1, {SafetyClass.READ: 0}, {SafetyClass.WRITE: -5}, True]) +def test_max_ttl_s_non_positive_rejected_at_construction(bad: object) -> None: + with pytest.raises(AgentKernelError, match="max_ttl_s"): + DefaultPolicyEngine(max_ttl_s=bad) # type: ignore[arg-type] diff --git a/tests/test_tokens.py b/tests/test_tokens.py index d02600e..629c879 100644 --- a/tests/test_tokens.py +++ b/tests/test_tokens.py @@ -3,10 +3,13 @@ from __future__ import annotations import datetime +from dataclasses import replace import pytest from weaver_kernel import ( + AgentKernelError, + CapabilityToken, HMACTokenProvider, TokenExpired, TokenInvalid, @@ -241,3 +244,200 @@ def test_track_and_sweep_accept_naive_datetimes() -> None: # Naive 'now' after expiry removes it. assert store.sweep_expired(datetime.datetime(2099, 1, 2)) == 1 assert not store.is_revoked("t1") + + +# ── Signing-key rotation (#185) ──────────────────────────────────────────────── + + +def test_single_secret_uses_default_key_id() -> None: + """Legacy single-secret config files the key under the 'default' key id.""" + provider = HMACTokenProvider(secret="s1") + token = provider.issue("cap.x", "user-1") + assert token.key_id == "default" + provider.verify(token, expected_principal_id="user-1", expected_capability_id="cap.x") + + +def test_rotation_overlap_window_verifies_previous_key() -> None: + """A token signed under a retired key still verifies while both keys are present.""" + old = HMACTokenProvider(secrets={"k1": "s1"}, active_key_id="k1") + token = old.issue("cap.x", "user-1") + assert token.key_id == "k1" + # Operator rotates: new active key k2, but k1 kept for the overlap window. + rotated = HMACTokenProvider(secrets={"k1": "s1", "k2": "s2"}, active_key_id="k2") + rotated.verify(token, expected_principal_id="user-1", expected_capability_id="cap.x") + # New tokens are signed under the active key. + assert rotated.issue("cap.x", "user-1").key_id == "k2" + + +def test_unknown_key_id_fails_closed() -> None: + """A token whose key id is not in the verifier's keyring fails as TokenInvalid.""" + issuer = HMACTokenProvider(secrets={"k1": "s1"}, active_key_id="k1") + token = issuer.issue("cap.x", "user-1") + # k1 was retired entirely — only k2 remains. + verifier = HMACTokenProvider(secrets={"k2": "s2"}, active_key_id="k2") + with pytest.raises(TokenInvalid, match="unknown key id"): + verifier.verify(token, expected_principal_id="user-1", expected_capability_id="cap.x") + + +def test_key_id_is_signed_tamper_evident() -> None: + """Re-labelling a token's key_id to a present key breaks the signature.""" + provider = HMACTokenProvider(secrets={"k1": "s1", "k2": "s2"}, active_key_id="k1") + token = provider.issue("cap.x", "user-1") + relabelled = replace(token, key_id="k2") + with pytest.raises(TokenInvalid, match="invalid signature"): + provider.verify(relabelled, expected_principal_id="user-1", expected_capability_id="cap.x") + + +def test_non_active_key_verification_logs_key_id_not_secret(caplog) -> None: # type: ignore[no-untyped-def] + """Verifying a non-active-key token logs the key id (never the secret).""" + provider = HMACTokenProvider(secrets={"k1": "s1", "k2": "s2"}, active_key_id="k2") + token = replace( + HMACTokenProvider(secrets={"k1": "s1"}, active_key_id="k1").issue("cap.x", "u1") + ) + with caplog.at_level("INFO"): + provider.verify(token, expected_principal_id="u1", expected_capability_id="cap.x") + records = [r for r in caplog.records if r.message == "token_verified_non_active_key"] + assert records and getattr(records[0], "key_id", None) == "k1" + assert "s1" not in caplog.text and "s2" not in caplog.text + + +def test_secret_and_secrets_together_is_rejected() -> None: + with pytest.raises(AgentKernelError, match="either 'secret' or 'secrets'"): + HMACTokenProvider(secret="s", secrets={"k1": "s1"}) + + +def test_empty_keyring_is_rejected() -> None: + with pytest.raises(AgentKernelError, match="empty"): + HMACTokenProvider(secrets={}) + + +def test_multi_key_without_active_key_id_is_rejected() -> None: + with pytest.raises(AgentKernelError, match="active key id must be specified"): + HMACTokenProvider(secrets={"k1": "s1", "k2": "s2"}) + + +def test_active_key_id_absent_from_keyring_is_rejected() -> None: + with pytest.raises(AgentKernelError, match="not present"): + HMACTokenProvider(secrets={"k1": "s1"}, active_key_id="k9") + + +def test_env_secrets_json_used_when_no_explicit_config(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("WEAVER_KERNEL_SECRETS", '{"k1": "s1", "k2": "s2"}') + monkeypatch.setenv("WEAVER_KERNEL_ACTIVE_KEY", "k2") + provider = HMACTokenProvider() + token = provider.issue("cap.x", "user-1") + assert token.key_id == "k2" + provider.verify(token, expected_principal_id="user-1", expected_capability_id="cap.x") + + +def test_env_secrets_precedes_legacy_single_secret(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("WEAVER_KERNEL_SECRETS", '{"k1": "s1"}') + monkeypatch.setenv("WEAVER_KERNEL_SECRET", "legacy") + provider = HMACTokenProvider() + assert provider.issue("cap.x", "u1").key_id == "k1" + + +def test_env_secrets_malformed_json_fails_closed(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("WEAVER_KERNEL_SECRETS", "{not json") + provider = HMACTokenProvider() + with pytest.raises(AgentKernelError, match="not valid JSON"): + provider.issue("cap.x", "u1") + + +# ── Typed from_dict errors (#200) ────────────────────────────────────────────── + + +def _valid_token_dict() -> dict: # type: ignore[type-arg] + provider = HMACTokenProvider(secret="s1") + return provider.issue("cap.x", "user-1").to_dict() + + +def test_from_dict_valid_roundtrip_unchanged() -> None: + provider = HMACTokenProvider(secret="s1") + token = provider.issue("cap.x", "user-1", constraints={"max_rows": 5}) + restored = CapabilityToken.from_dict(token.to_dict()) + assert restored == token + provider.verify(restored, expected_principal_id="user-1", expected_capability_id="cap.x") + + +@pytest.mark.parametrize("field", ["token_id", "capability_id", "principal_id"]) +def test_from_dict_missing_required_field_raises_token_invalid(field: str) -> None: + data = _valid_token_dict() + del data[field] + with pytest.raises(TokenInvalid, match=f"missing field '{field}'"): + CapabilityToken.from_dict(data) + + +@pytest.mark.parametrize("field", ["issued_at", "expires_at"]) +def test_from_dict_missing_timestamp_raises_token_invalid(field: str) -> None: + data = _valid_token_dict() + del data[field] + with pytest.raises(TokenInvalid, match=f"missing field '{field}'"): + CapabilityToken.from_dict(data) + + +@pytest.mark.parametrize("field", ["issued_at", "expires_at"]) +def test_from_dict_bad_timestamp_raises_token_invalid(field: str) -> None: + data = _valid_token_dict() + data[field] = "not-a-timestamp" + with pytest.raises(TokenInvalid, match=f"invalid timestamp in field '{field}'"): + CapabilityToken.from_dict(data) + + +def test_from_dict_wrong_type_field_raises_token_invalid() -> None: + data = _valid_token_dict() + data["token_id"] = 123 + with pytest.raises(TokenInvalid, match="must be a string"): + CapabilityToken.from_dict(data) + + +def test_from_dict_non_object_constraints_raises_token_invalid() -> None: + data = _valid_token_dict() + data["constraints"] = ["not", "a", "dict"] + with pytest.raises(TokenInvalid, match="'constraints' must be an object"): + CapabilityToken.from_dict(data) + + +def test_from_dict_tolerates_unknown_extra_keys() -> None: + data = _valid_token_dict() + data["future_field"] = "ignored" + restored = CapabilityToken.from_dict(data) + assert restored.token_id == data["token_id"] + + +# ── Additional fail-closed coverage (#185 / #200) ────────────────────────────── + + +def test_keyring_non_string_values_rejected() -> None: + with pytest.raises(AgentKernelError, match="string key ids to string secrets"): + HMACTokenProvider(secrets={"k1": 123}) # type: ignore[dict-item] + + +def test_env_secrets_non_object_json_rejected(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("WEAVER_KERNEL_SECRETS", '"just-a-string"') + provider = HMACTokenProvider() + with pytest.raises(AgentKernelError, match="JSON object"): + provider.issue("cap.x", "u1") + + +def test_env_active_key_absent_from_map_rejected(monkeypatch) -> None: # type: ignore[no-untyped-def] + monkeypatch.setenv("WEAVER_KERNEL_SECRETS", '{"k1": "s1", "k2": "s2"}') + monkeypatch.setenv("WEAVER_KERNEL_ACTIVE_KEY", "k9") + provider = HMACTokenProvider() + with pytest.raises(AgentKernelError, match="not present"): + provider.issue("cap.x", "u1") + + +@pytest.mark.parametrize("field", ["audit_id", "signature", "key_id"]) +def test_from_dict_non_string_optional_field_rejected(field: str) -> None: + data = _valid_token_dict() + data[field] = 123 + with pytest.raises(TokenInvalid, match="must be a string"): + CapabilityToken.from_dict(data) + + +def test_from_dict_non_string_timestamp_rejected() -> None: + data = _valid_token_dict() + data["issued_at"] = 123 + with pytest.raises(TokenInvalid, match="must be an ISO-8601 string"): + CapabilityToken.from_dict(data) From 679a1117322324e5ff24c5f9b3e7a97d9497bcf1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 07:50:49 +0000 Subject: [PATCH 2/2] fix: break token-module import cycle (CodeQL) and harden fail-closed inputs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses PR #259 review: - CodeQL cyclic-import errors: the tokens <-> _hmac_provider re-export and the tokens <-> _token_signing TYPE_CHECKING import formed cycles. Made the graph acyclic — _token_signing no longer references tokens (build_signable_payload inlined into CapabilityToken._signable_payload), and tokens no longer re-exports HMACTokenProvider. Importers (__init__, kernel, cli/_doctor) now import HMACTokenProvider from _hmac_provider; the public `from weaver_kernel import HMACTokenProvider` is unchanged. - Copilot: from_dict now coerces naive ISO timestamps to UTC, so a malformed token can't cause a naive-vs-aware TypeError at verify() time. - Copilot: DefaultPolicyEngine(max_ttl_s=...) now rejects non-SafetyClass dict keys at construction (fail closed instead of silently ignoring the cap). - Copilot: enforce_arg_constraints fails closed (TokenScopeError, arg_constraint_violation) on malformed args.allowed_keys/pinned/prefix types instead of raising an untyped TypeError or silently ignoring them. Tests added for each. make ci green: 861 passed, 1 skipped, 94% branch coverage. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_019WRxQL8t2Uusa6845jWtVV --- src/weaver_kernel/__init__.py | 6 +-- src/weaver_kernel/_hmac_provider.py | 5 +- src/weaver_kernel/_token_signing.py | 61 +++++++++--------------- src/weaver_kernel/cli/_doctor.py | 3 +- src/weaver_kernel/kernel/__init__.py | 3 +- src/weaver_kernel/kernel/_constraints.py | 15 ++++-- src/weaver_kernel/policy_ttl.py | 13 ++++- src/weaver_kernel/tokens.py | 35 +++++++++----- tests/test_kernel.py | 24 ++++++++++ tests/test_policy.py | 5 ++ tests/test_tokens.py | 10 ++++ 11 files changed, 118 insertions(+), 62 deletions(-) diff --git a/src/weaver_kernel/__init__.py b/src/weaver_kernel/__init__.py index 94a3b39..a86ed72 100644 --- a/src/weaver_kernel/__init__.py +++ b/src/weaver_kernel/__init__.py @@ -57,8 +57,7 @@ AgentKernelError, TokenExpired, TokenInvalid, TokenScopeError, TokenRevoked, PolicyDenied, PolicyConfigError, - DriverError, FirewallError, AdapterParseError, - BudgetExhausted, BudgetConfigError, + DriverError, FirewallError, AdapterParseError, BudgetExhausted, BudgetConfigError, CapabilityNotFound, CapabilityAlreadyRegistered, HandleNotFound, HandleExpired, HandleTooLarge, HandleConstraintViolation, NamespaceNotFound, FederationError, ManifestError, ManifestSignatureError, @@ -69,6 +68,7 @@ from importlib.metadata import PackageNotFoundError from importlib.metadata import version as _pkg_version +from ._hmac_provider import HMACTokenProvider from .adapters import AnthropicMiddleware, OpenAIMiddleware from .drivers.base import Driver, ExecutionContext from .drivers.http import HTTPDriver @@ -177,7 +177,7 @@ TraceStoreProtocol, verify_chain, ) -from .tokens import CapabilityToken, HMACTokenProvider +from .tokens import CapabilityToken from .trace import ( TRACE_EXPORT_SCHEMA, TRACE_EXPORT_VERSION, diff --git a/src/weaver_kernel/_hmac_provider.py b/src/weaver_kernel/_hmac_provider.py index 8b1d512..d21daf8 100644 --- a/src/weaver_kernel/_hmac_provider.py +++ b/src/weaver_kernel/_hmac_provider.py @@ -3,8 +3,9 @@ Extracted from :mod:`weaver_kernel.tokens` to keep that module within the AGENTS.md 300-line budget. :class:`~weaver_kernel.tokens.CapabilityToken` and the :class:`~weaver_kernel.tokens.TokenProvider` Protocol remain in -:mod:`weaver_kernel.tokens`, which re-exports this class so -``from weaver_kernel.tokens import HMACTokenProvider`` keeps working. +:mod:`weaver_kernel.tokens`. This module imports from ``tokens`` (a one-way +dependency), so ``tokens`` does *not* re-export this class — that would form an +import cycle. Import it from :mod:`weaver_kernel` (public) instead. """ from __future__ import annotations diff --git a/src/weaver_kernel/_token_signing.py b/src/weaver_kernel/_token_signing.py index 2a5718e..2caf5b3 100644 --- a/src/weaver_kernel/_token_signing.py +++ b/src/weaver_kernel/_token_signing.py @@ -1,13 +1,15 @@ -"""HMAC signing and serialization helpers for capability tokens. +"""HMAC signing and token-parsing helpers. Extracted from :mod:`weaver_kernel.tokens` to keep that module within the -AGENTS.md 300-line budget and to isolate the crypto/serialization concern: -building the canonical signable payload, HMAC signing, and parsing an untrusted -serialized token into validated fields — raising :class:`TokenInvalid` rather -than leaking a bare ``KeyError``/``ValueError`` (#200). - -The signed payload includes the ``key_id`` so signing-key rotation (#185) is -tamper-evident: a token cannot be re-labelled to verify against a different key. +AGENTS.md 300-line budget and to isolate the crypto/deserialization concern: +HMAC signing, and parsing an untrusted serialized token into validated +constructor kwargs — raising :class:`TokenInvalid` rather than leaking a bare +``KeyError``/``ValueError`` (#200). + +This module is a leaf: it imports nothing from :mod:`weaver_kernel.tokens` (not +even under ``TYPE_CHECKING``), so the ``tokens`` → ``_token_signing`` dependency +stays acyclic. Canonical payload building lives on +:meth:`CapabilityToken._signable_payload` in ``tokens.py``. """ from __future__ import annotations @@ -15,41 +17,14 @@ import datetime import hashlib import hmac -import json -from typing import TYPE_CHECKING, Any +from typing import Any from .errors import TokenInvalid -if TYPE_CHECKING: # pragma: no cover - from .tokens import CapabilityToken - KeyRing = dict[str, str] """Mapping of key id → HMAC secret used for signing-key rotation (#185).""" -def build_signable_payload(token: CapabilityToken) -> str: - """Return the canonical JSON string used as the HMAC message. - - Args: - token: The token whose bound fields (principal, capability, constraints, - expiry, and signing ``key_id``) form the signature input. - - Returns: - A deterministic, key-sorted JSON string. - """ - payload = { - "token_id": token.token_id, - "capability_id": token.capability_id, - "principal_id": token.principal_id, - "issued_at": token.issued_at.isoformat(), - "expires_at": token.expires_at.isoformat(), - "constraints": token.constraints, - "audit_id": token.audit_id, - "key_id": token.key_id, - } - return json.dumps(payload, sort_keys=True, separators=(",", ":")) - - def sign(secret: str, payload: str) -> str: """Return the hex HMAC-SHA256 of *payload* under *secret*. @@ -88,7 +63,12 @@ def _optional_str(data: dict[str, Any], field: str) -> str: def _require_timestamp(data: dict[str, Any], field: str) -> datetime.datetime: - """Return a required ISO-8601 timestamp field, raising :class:`TokenInvalid`.""" + """Return a required ISO-8601 timestamp field, raising :class:`TokenInvalid`. + + A naive (timezone-less) timestamp is treated as UTC — matching how the + revocation stores handle naive datetimes — so a malformed/untrusted token can + never turn into a naive-vs-aware ``TypeError`` at ``verify()`` time. + """ if field not in data: raise TokenInvalid(f"malformed token payload: missing field '{field}'.") raw = data[field] @@ -98,11 +78,14 @@ def _require_timestamp(data: dict[str, Any], field: str) -> datetime.datetime: f"got {type(raw).__name__}." ) try: - return datetime.datetime.fromisoformat(raw) + parsed = datetime.datetime.fromisoformat(raw) except ValueError as exc: raise TokenInvalid( f"malformed token payload: invalid timestamp in field '{field}': {raw!r}." ) from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=datetime.timezone.utc) + return parsed def parse_token_fields(data: dict[str, Any]) -> dict[str, Any]: @@ -143,4 +126,4 @@ def parse_token_fields(data: dict[str, Any]) -> dict[str, Any]: } -__all__ = ["KeyRing", "build_signable_payload", "sign", "parse_token_fields"] +__all__ = ["KeyRing", "sign", "parse_token_fields"] diff --git a/src/weaver_kernel/cli/_doctor.py b/src/weaver_kernel/cli/_doctor.py index 2d02999..3e94417 100644 --- a/src/weaver_kernel/cli/_doctor.py +++ b/src/weaver_kernel/cli/_doctor.py @@ -19,10 +19,11 @@ import sys from dataclasses import dataclass +from .._hmac_provider import HMACTokenProvider from .._secrets import SECRET_ENV_VAR from ..models import ActionTrace from ..stores.audit_chain import build_record, verify_chain -from ..tokens import CapabilityToken, HMACTokenProvider +from ..tokens import CapabilityToken OK = "ok" WARN = "warn" diff --git a/src/weaver_kernel/kernel/__init__.py b/src/weaver_kernel/kernel/__init__.py index 247e9e2..6d045c4 100644 --- a/src/weaver_kernel/kernel/__init__.py +++ b/src/weaver_kernel/kernel/__init__.py @@ -14,6 +14,7 @@ from collections.abc import AsyncIterator, Callable from typing import Any, Literal, overload +from .._hmac_provider import HMACTokenProvider from ..drivers.base import Driver, StreamingDriver from ..enums import SafetyClass from ..errors import AgentKernelError @@ -41,7 +42,7 @@ from ..router import Router, StaticRouter from ..stats import KernelStats, StatsSnapshot from ..stores import TraceStoreProtocol -from ..tokens import CapabilityToken, HMACTokenProvider, TokenProvider +from ..tokens import CapabilityToken, TokenProvider from ..trace import TraceStore from ..trace_query import TraceQuery from ._audit import record_expansion_trace diff --git a/src/weaver_kernel/kernel/_constraints.py b/src/weaver_kernel/kernel/_constraints.py index 7d22e0f..5d8c639 100644 --- a/src/weaver_kernel/kernel/_constraints.py +++ b/src/weaver_kernel/kernel/_constraints.py @@ -80,20 +80,29 @@ def enforce_arg_constraints(args: dict[str, Any], constraints: dict[str, Any]) - if not isinstance(spec, dict): raise _violation(f"malformed 'args' constraint: expected an object, got {spec!r}.") + # A malformed rule value is a security-scoping misconfiguration — fail closed + # (deny) rather than silently ignoring it (fail open) or crashing with an + # untyped TypeError that would escape the audited denial path. allowed = spec.get("allowed_keys") if allowed is not None: - extra = sorted(set(args) - set(allowed)) + if not isinstance(allowed, (list, tuple, set)): + raise _violation("malformed 'args.allowed_keys' constraint: expected a list.") + extra = sorted(k for k in args if k not in set(allowed)) if extra: raise _violation(f"arguments {extra} are not permitted by the token's allowed_keys.") pinned = spec.get("pinned") - if isinstance(pinned, dict): + if pinned is not None: + if not isinstance(pinned, dict): + raise _violation("malformed 'args.pinned' constraint: expected an object.") for key, expected in pinned.items(): if key not in args or args[key] != expected: raise _violation(f"argument '{key}' must equal the token's pinned value.") prefix = spec.get("prefix") - if isinstance(prefix, dict): + if prefix is not None: + if not isinstance(prefix, dict): + raise _violation("malformed 'args.prefix' constraint: expected an object.") for key, required_prefix in prefix.items(): value = args.get(key) if not isinstance(value, str) or not value.startswith(required_prefix): diff --git a/src/weaver_kernel/policy_ttl.py b/src/weaver_kernel/policy_ttl.py index f24f9cd..4f770e8 100644 --- a/src/weaver_kernel/policy_ttl.py +++ b/src/weaver_kernel/policy_ttl.py @@ -31,7 +31,18 @@ def validate_max_ttl_s(max_ttl_s: MaxTTLConfig) -> None: """ if max_ttl_s is None: return - values = max_ttl_s.values() if isinstance(max_ttl_s, dict) else [max_ttl_s] + if isinstance(max_ttl_s, dict): + # Fail closed on non-SafetyClass keys: resolve_max_ttl_s() looks up by + # SafetyClass, so a stray string key (e.g. from config parsing) would be + # silently ignored and the cap never applied. + for key in max_ttl_s: + if not isinstance(key, SafetyClass): + raise AgentKernelError( + f"Invalid max_ttl_s: keys must be SafetyClass members, got {key!r}." + ) + values = list(max_ttl_s.values()) + else: + values = [max_ttl_s] for value in values: if not isinstance(value, int) or isinstance(value, bool) or value <= 0: raise AgentKernelError( diff --git a/src/weaver_kernel/tokens.py b/src/weaver_kernel/tokens.py index d4389a7..7665b91 100644 --- a/src/weaver_kernel/tokens.py +++ b/src/weaver_kernel/tokens.py @@ -3,17 +3,20 @@ The concrete :class:`HMACTokenProvider` lives in :mod:`weaver_kernel._hmac_provider` (extracted to honour the AGENTS.md -300-line module budget) and is re-exported here so -``from weaver_kernel.tokens import HMACTokenProvider`` keeps working. +300-line module budget). Import it from :mod:`weaver_kernel` (public) or +:mod:`weaver_kernel._hmac_provider`. It is intentionally *not* re-exported here: +``_hmac_provider`` imports this module, so re-exporting would create an import +cycle (flagged by CodeQL). """ from __future__ import annotations import datetime +import json from dataclasses import dataclass, field from typing import Any, Protocol -from ._token_signing import build_signable_payload, parse_token_fields +from ._token_signing import parse_token_fields # ── Token dataclass ─────────────────────────────────────────────────────────── @@ -40,8 +43,22 @@ class CapabilityToken: # ── Serialization ───────────────────────────────────────────────────────── def _signable_payload(self) -> str: - """Return the canonical JSON string used as the HMAC message.""" - return build_signable_payload(self) + """Return the canonical JSON string used as the HMAC message. + + The signing ``key_id`` is included so a token cannot be re-labelled to + verify against a different rotation key (#185). + """ + payload = { + "token_id": self.token_id, + "capability_id": self.capability_id, + "principal_id": self.principal_id, + "issued_at": self.issued_at.isoformat(), + "expires_at": self.expires_at.isoformat(), + "constraints": self.constraints, + "audit_id": self.audit_id, + "key_id": self.key_id, + } + return json.dumps(payload, sort_keys=True, separators=(",", ":")) def to_dict(self) -> dict[str, Any]: """Serialise the token to a plain dict (suitable for JSON transport).""" @@ -147,10 +164,4 @@ def revoke_all(self, principal_id: str) -> int: ... -# ── Implementation (re-exported) ────────────────────────────────────────────── - -# HMACTokenProvider lives in a sibling module to keep this file within the -# AGENTS.md 300-line budget; re-exported so its public import path is unchanged. -from ._hmac_provider import HMACTokenProvider # noqa: E402 - -__all__ = ["CapabilityToken", "TokenProvider", "HMACTokenProvider"] +__all__ = ["CapabilityToken", "TokenProvider"] diff --git a/tests/test_kernel.py b/tests/test_kernel.py index 38c1fdb..b41bdfb 100644 --- a/tests/test_kernel.py +++ b/tests/test_kernel.py @@ -1607,3 +1607,27 @@ async def test_malformed_args_constraint_denied( await kernel.invoke( token, principal=reader_principal, args={"operation": "billing.list_invoices"} ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "spec,needle", + [ + ({"allowed_keys": 5}, "allowed_keys"), + ({"pinned": ["not", "a", "dict"]}, "pinned"), + ({"prefix": ["not", "a", "dict"]}, "prefix"), + ], +) +async def test_malformed_arg_constraint_nested_type_fails_closed( + kernel: Kernel, reader_principal: Principal, spec: dict, needle: str +) -> None: + """A malformed nested arg-constraint denies (fail closed), never an untyped crash.""" + token = kernel.get_token(_req_with_args(spec), reader_principal, justification="") + with pytest.raises(TokenScopeError) as exc_info: + await kernel.invoke( + token, principal=reader_principal, args={"operation": "billing.list_invoices"} + ) + assert exc_info.value.reason_code == "arg_constraint_violation" + assert needle in str(exc_info.value) + # And it is audited (I-02), reached no driver. + assert any(t.error and t.driver_id == "" for t in kernel.list_traces()) diff --git a/tests/test_policy.py b/tests/test_policy.py index 171ee86..218d85e 100644 --- a/tests/test_policy.py +++ b/tests/test_policy.py @@ -1870,3 +1870,8 @@ def test_max_ttl_s_none_is_uncapped() -> None: def test_max_ttl_s_non_positive_rejected_at_construction(bad: object) -> None: with pytest.raises(AgentKernelError, match="max_ttl_s"): DefaultPolicyEngine(max_ttl_s=bad) # type: ignore[arg-type] + + +def test_max_ttl_s_non_safetyclass_key_rejected() -> None: + with pytest.raises(AgentKernelError, match="keys must be SafetyClass"): + DefaultPolicyEngine(max_ttl_s={"read": 60}) # type: ignore[dict-item] diff --git a/tests/test_tokens.py b/tests/test_tokens.py index 629c879..988ae22 100644 --- a/tests/test_tokens.py +++ b/tests/test_tokens.py @@ -441,3 +441,13 @@ def test_from_dict_non_string_timestamp_rejected() -> None: data["issued_at"] = 123 with pytest.raises(TokenInvalid, match="must be an ISO-8601 string"): CapabilityToken.from_dict(data) + + +def test_from_dict_naive_timestamp_coerced_to_utc() -> None: + """A naive timestamp is treated as UTC so verify() never hits a naive/aware TypeError.""" + data = _valid_token_dict() + data["issued_at"] = "2026-01-01T00:00:00" # no timezone + data["expires_at"] = "2099-01-01T00:00:00" + token = CapabilityToken.from_dict(data) + assert token.issued_at.tzinfo is datetime.timezone.utc + assert token.expires_at.tzinfo is datetime.timezone.utc