From fe4f473053fb3a55f9306faa740cbe9dfd02d923 Mon Sep 17 00:00:00 2001 From: Koishore Roy Date: Thu, 11 Jun 2026 03:09:25 +0530 Subject: [PATCH] =?UTF-8?q?Implement=20the=20=C2=A79=20authorization=20tok?= =?UTF-8?q?en=20(optional=20profile)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A portable, signed PDP->PEP decision artifact (spec §9): a compact EdDSA JWS minted on 'allow'/released approval, verified by a separated broker before it injects a credential. Built on cryptography (no new dependency). - delego/token.py: TokenIssuer.mint (allow/release only), verify_token (§9.1 1-4: pin EdDSA, reject none/alg-confusion with the key from verifier config not the header, exact aud, exp+skew, single-use jti/cns), require_fingerprint (§9.1 step 5 -- recompute the fingerprint of the request the broker will send; the enforcement-side closure of the declared-vs-sent gap). - Firewall(token_issuer=, token_audience=) / build_firewall(mint_tokens=True); token rides on Decision.token. Separate token_key.pem (distinct from the audit key, spec §9). Brokers accept optional token=; HTTPProxyBroker forwards it. - Off by default: protocol unchanged (0.3), existing behaviour byte-for-byte identical. Package 0.3.3. 14 new tests (mint gating + every §9.1 failure mode + step 5); 65 total green. --- CHANGELOG.md | 34 ++++- README.md | 17 ++- ROADMAP.md | 12 +- SECURITY.md | 7 +- delego/__init__.py | 7 +- delego/brokers.py | 45 ++++-- delego/config.py | 41 +++++- delego/engine.py | 53 ++++++- delego/models.py | 4 + delego/token.py | 327 +++++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- server.json | 4 +- tests/test_broker.py | 27 ++++ tests/test_token.py | 182 ++++++++++++++++++++++++ 14 files changed, 725 insertions(+), 37 deletions(-) create mode 100644 delego/token.py create mode 100644 tests/test_token.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ccb325..31bd9be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,37 @@ adhere to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +## [0.3.3] — 2026-06-11 + +Implements the **§9 authorization token** (optional profile). Protocol unchanged +(still 0.3) — the token is additive and changes no hashed or signed bytes; it's +off by default, so existing deployments are byte-for-byte unchanged. + +### Added +- **`delego.token`** — a portable, signed PDP→PEP decision artifact (spec §9): + a compact JWS / JWT with `alg = EdDSA`. Built on `cryptography` (already a + dependency) — **no new dependency**, no PyJWT. + - **`TokenIssuer`** mints a short-lived token (TTL ≤ 300s, default 45s) for an + `allow` outcome or a released approval — and *only* those; `deny`, + `needs_approval`, `denied`, and `consumed` mint nothing. + - **`verify_token`** performs §9.1 checks 1–4: pins `EdDSA` and rejects + `none`/algorithm-confusion (the key comes from verifier config, never the + token header), exact `aud` match, expiry with bounded clock skew, and + single-use `jti` (replay) and `cns` (consumption nonce). + - **`require_fingerprint`** is §9.1 step 5 — the crux: a broker recomputes the + fingerprint of the request it's about to send and requires it equals the + token's `fpr`. This is the enforcement-side closure of the "declared vs. + sent" gap (a token minted for action A cannot release action B). +- **`Firewall(token_issuer=..., token_audience=...)`** and + **`build_firewall(..., mint_tokens=True)`** wire the profile; the minted token + rides on `Decision.token`. A **separate** token signing key + (`token_key.pem`, distinct from the audit key per spec §9) is generated on + first use. +- Brokers accept an optional `token=` keyword (non-breaking — a bare + `execute(action)` adapter still works); `HTTPProxyBroker` forwards the token to + the gateway as `authorization_token` so the separated PEP can verify it. +- Exports: `TokenIssuer`, `verify_token`, `require_fingerprint`, `TokenError`. + ## [0.3.2] — 2026-06-10 Packaging only; protocol unchanged (still 0.3). No functional or API changes. @@ -276,7 +307,8 @@ published.) Implements wire-protocol **0.2**; see a FastMCP server exposing propose / resolve / audit_tail / show_policy. - `NullBroker` (default; holds no credentials) and an `HTTPProxyBroker` sketch. -[Unreleased]: https://github.com/Delego-Dev/delego/compare/v0.3.2...HEAD +[Unreleased]: https://github.com/Delego-Dev/delego/compare/v0.3.3...HEAD +[0.3.3]: https://github.com/Delego-Dev/delego/compare/v0.3.2...v0.3.3 [0.3.2]: https://github.com/Delego-Dev/delego/compare/v0.3.1...v0.3.2 [0.3.1]: https://github.com/Delego-Dev/delego/compare/v0.3.0...v0.3.1 [0.3.0]: https://github.com/Delego-Dev/delego/compare/v0.2.4...v0.3.0 diff --git a/README.md b/README.md index 24ebb46..5e41524 100644 --- a/README.md +++ b/README.md @@ -206,16 +206,19 @@ See **[ROADMAP.md](ROADMAP.md)** for where delego is going and where to help. - **Implemented (protocol 0.3):** the policy engine, intent hashing, action fingerprinting **with the URL query folded into the fingerprint** (spec §4.2 — `/orders?to=me` and `/orders?to=attacker` are different actions), the - confused-deputy guard, intent-bound + single-use human approvals, and the + confused-deputy guard, intent-bound + single-use human approvals, the signed, hash-chained audit ledger with verification and an external - head-anchor check (`delego verify --expected-head`). + head-anchor check (`delego verify --expected-head`), and the **§9 authorization + token** (optional profile) — a short-lived, EdDSA-signed JWS a separated broker + verifies before injecting a credential (`build_firewall(..., mint_tokens=True)`; + `verify_token` / `require_fingerprint`). - **Brokers:** the default `NullBroker` holds no credentials and makes no real request — it records what *would* be sent (for demos and tests). `HTTPProxyBroker` - forwards an authorised action to an external credential gateway; or write your - own against the `BrokerAdapter` protocol in `delego/brokers.py`. -- **Not yet:** the authorization token (spec §9, an optional profile), an - always-on daemon (state is file-backed and shared by the CLI and MCP server), - and a non-MCP HTTP surface. + forwards an authorised action — and its authorization token — to an external + credential gateway; or write your own against the `BrokerAdapter` protocol in + `delego/brokers.py`. +- **Not yet:** an always-on daemon (state is file-backed and shared by the CLI + and MCP server), and a non-MCP HTTP surface. - **Known limitations:** concurrent writes to the file-backed ledger and approval store are serialised with an OS file lock (corruption-safe). Rate limits are **exact on a single host**: a policy carrying a `rate_limit` runs each propose diff --git a/ROADMAP.md b/ROADMAP.md index baf5176..2f8a7f5 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -26,6 +26,11 @@ implements and `y` the iteration. Normative changes land in the spec first. the `action_fingerprint`, so decision-relevant data can't ride it (0.3.0, breaking; regenerated CTK vectors). Rate-limited proposes are serialized under the ledger lock, making the cap exact on a single host (0.3.0). +- **Signed authorization token** (spec §9) — on `allow`/release the authorizer + mints a short-lived EdDSA JWS bound to the action fingerprint + intent; a + separated broker verifies it (pin EdDSA, exact `aud`, single-use `jti`/`cns`) + and re-checks the fingerprint of the request it's about to send before + injecting a credential (0.3.3). No new dependency. ## Now — make it usable in production (protocol 0.3) @@ -46,12 +51,7 @@ implements and `y` the iteration. Normative changes land in the spec first. ## Next — differentiate and harden (protocol 0.3, spec-first) -4. **Signed authorization token** (spec §9, already drafted). On allow/approve, - mint a short-lived JWS bound to the action fingerprint + intent; a broker - verifies it before injecting a credential. Closes the gap where a broker would - inject for *any* in-scope request, and makes delego composable across vendors. - *This is the moat.* -5. **Single-writer daemon.** A long-running process so non-MCP clients work, the +4. **Single-writer daemon.** A long-running process so non-MCP clients work, the CLI + MCP share live state over a socket, and rate-limit counting is exact across hosts without holding a file lock through broker calls (0.2.1 made writes corruption-safe; 0.3.0 made the cap exact on one host; this makes it diff --git a/SECURITY.md b/SECURITY.md index 1243562..0d4f1c9 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -42,7 +42,12 @@ Be precise about what delego does and does not protect against: The agent supplies `params`, `url`, and `instruction`; delego trusts them. Its guarantees hold only if your broker executes exactly the fingerprinted action. delego is a control only when the credential is reachable **solely** through the - broker — an agent with raw network/exec can bypass it entirely. + broker — an agent with raw network/exec can bypass it entirely. The + enforcement-side closure is the §9 token's step-5 re-check + (`require_fingerprint`): a token-requiring broker recomputes the fingerprint of + the request it is about to send and refuses if it doesn't equal the authorized + `fpr`, so an agent that declared one action but tries to send another is + refused at the PEP. - **The recorded intent is the agent's claim.** `intent_hash` binds the instruction the agent *reported*; it is not verified against the real human request, so under a fully compromised agent the audited intent is diff --git a/delego/__init__.py b/delego/__init__.py index 3af0e04..85199d3 100644 --- a/delego/__init__.py +++ b/delego/__init__.py @@ -24,8 +24,9 @@ ProposedAction, ) from .policy import Policy, PolicyError +from .token import TokenError, TokenIssuer, require_fingerprint, verify_token -__version__ = "0.3.2" # PyPI package version, 0.x.y (x = protocol, y = iteration) +__version__ = "0.3.3" # PyPI package version, 0.x.y (x = protocol, y = iteration) # Highest delego *protocol* version (see the wire spec's "Protocol versions") # this reference implements. Protocol/spec versions are 0.x (two-component); the @@ -44,6 +45,10 @@ "ensure_keys", "Paths", "build_firewall", + "TokenIssuer", + "verify_token", + "require_fingerprint", + "TokenError", "OUTCOME_ALLOW", "OUTCOME_DENY", "OUTCOME_APPROVAL", diff --git a/delego/brokers.py b/delego/brokers.py index bcbb571..8f1109e 100644 --- a/delego/brokers.py +++ b/delego/brokers.py @@ -46,11 +46,19 @@ class BrokerAdapter(Protocol): and MUST refuse — raise :class:`BrokerRefusal`, never silently strip — when ``action.url`` carries a fragment (:attr:`ProposedAction.has_fragment`) that the fingerprint does not represent. + + **Authorization token (spec §9, optional).** When a token issuer is + configured the firewall passes the minted token as the ``token`` keyword. A + *separated* broker (one across a process/network boundary from the + authorizer) SHOULD verify it (:func:`delego.verify_token` + + :func:`delego.require_fingerprint`) before injecting a credential. A broker + that does not implement the token may keep a bare ``execute(action)`` + signature — the firewall falls back to it. """ name: str - def execute(self, action: ProposedAction) -> dict[str, Any]: + def execute(self, action: ProposedAction, token: str | None = None) -> dict[str, Any]: ... @@ -86,7 +94,7 @@ class NullBroker: def __init__(self) -> None: self.sent: list[dict] = [] - def execute(self, action: ProposedAction) -> dict[str, Any]: + def execute(self, action: ProposedAction, token: str | None = None) -> dict[str, Any]: # Honour the execution contract even when simulating: a stray fragment # is unauthorised data regardless of whether a real request is made. _require_no_unauthorised_fragment(action) @@ -94,6 +102,9 @@ def execute(self, action: ProposedAction) -> dict[str, Any]: "broker": self.name, "would_send": action.summary(), "note": "stub: no credential injected, no upstream request made", + # Surfaced so the decision -> execution loop (including a minted §9 + # token) is observable end to end; NullBroker verifies nothing. + "token": token, } self.sent.append(record) return {"status": "simulated", "detail": record} @@ -138,7 +149,7 @@ def __init__( self.timeout = timeout self._headers = {"content-type": "application/json", **(gateway_headers or {})} - def execute(self, action: ProposedAction) -> dict[str, Any]: + def execute(self, action: ProposedAction, token: str | None = None) -> dict[str, Any]: import json import urllib.error import urllib.request @@ -147,18 +158,22 @@ def execute(self, action: ProposedAction) -> dict[str, Any]: # data outside the fingerprint preimage (spec §4.2). _require_no_unauthorised_fragment(action) - payload = json.dumps( - { - "method": action.method.upper(), - # Forward only the fingerprinted URL (scheme+host+path+query); - # the fragment is never represented in the fingerprint, so it is - # never sent. - "url": action.fingerprinted_url, - "params": action.params, - "intent_hash": action.intent_hash, - "action_fingerprint": action.fingerprint, - } - ).encode("utf-8") + body: dict[str, Any] = { + "method": action.method.upper(), + # Forward only the fingerprinted URL (scheme+host+path+query); the + # fragment is never represented in the fingerprint, so it is never + # sent. + "url": action.fingerprinted_url, + "params": action.params, + "intent_hash": action.intent_hash, + "action_fingerprint": action.fingerprint, + } + # When a §9 token was minted, forward it so the gateway can verify the + # authorization (and re-check the fingerprint of what it's about to send) + # before injecting the credential. The gateway is the separated PEP. + if token is not None: + body["authorization_token"] = token + payload = json.dumps(body).encode("utf-8") req = urllib.request.Request( self.gateway_url, data=payload, method="POST", headers=self._headers ) diff --git a/delego/config.py b/delego/config.py index 73e277a..ce5ed43 100644 --- a/delego/config.py +++ b/delego/config.py @@ -26,6 +26,7 @@ _STATE_GITIGNORE = """\ # delego runtime state — do not commit secrets or the audit ledger. signing_key.pem +token_key.pem audit.log.jsonl approvals.jsonl *.lock @@ -84,6 +85,16 @@ def approvals(self) -> Path: def policy(self) -> Path: return self.home / "policy.yaml" + @property + def token_private_key(self) -> Path: + # Distinct from the audit signing key (spec §9): a token-minting + # compromise must not be able to forge the audit chain. + return self.home / "token_key.pem" + + @property + def token_public_key(self) -> Path: + return self.home / "token_key.pub" + def ensure_home_gitignore(home: str | os.PathLike) -> None: """Drop a ``.gitignore`` in the home so keys/ledger aren't committed when the @@ -95,7 +106,13 @@ def ensure_home_gitignore(home: str | os.PathLike) -> None: gitignore.write_text(_STATE_GITIGNORE, encoding="utf-8") -def build_firewall(paths: Paths, broker: BrokerAdapter | None = None) -> Firewall: +def build_firewall( + paths: Paths, + broker: BrokerAdapter | None = None, + *, + mint_tokens: bool = False, + token_audience: str = "broker:default", +) -> Firewall: # Load policy first: a missing/invalid policy fails closed with a clear # message before any state (keys, ledger) is created. policy = Policy.load(paths.policy) @@ -103,4 +120,24 @@ def build_firewall(paths: Paths, broker: BrokerAdapter | None = None) -> Firewal ensure_keys(paths.private_key, paths.public_key) audit = AuditLog(paths.audit_log, paths.private_key, paths.public_key) approvals = ApprovalStore(paths.approvals) - return Firewall(policy, audit, approvals, broker=broker) + + token_issuer = None + if mint_tokens: + # The §9 profile is opt-in: only when requested do we generate the + # (separate) token key and wire an issuer. Off by default, so existing + # deployments are byte-for-byte unchanged. + from .token import TokenIssuer + + token_issuer = TokenIssuer.from_files( + paths.token_private_key, + paths.token_public_key, + issuer="delego:local", + ) + return Firewall( + policy, + audit, + approvals, + broker=broker, + token_issuer=token_issuer, + token_audience=token_audience, + ) diff --git a/delego/engine.py b/delego/engine.py index 8311c1d..caaeda0 100644 --- a/delego/engine.py +++ b/delego/engine.py @@ -15,6 +15,8 @@ from __future__ import annotations +from typing import TYPE_CHECKING + from .approval import ( STATUS_APPROVED, STATUS_CONSUMED, @@ -33,6 +35,9 @@ ) from .policy import Policy +if TYPE_CHECKING: + from .token import TokenIssuer + class Firewall: def __init__( @@ -41,11 +46,20 @@ def __init__( audit: AuditLog, approvals: ApprovalStore, broker: BrokerAdapter | None = None, + token_issuer: "TokenIssuer | None" = None, + token_audience: str = "broker:default", ) -> None: self.policy = policy self.audit = audit self.approvals = approvals self.broker = broker or NullBroker() + # Optional §9 profile: when wired, the firewall mints a short-lived + # authorization token for `allow` outcomes (and released approvals) and + # attaches it to the Decision for a separated broker to verify. Default + # off — the in-process broker already trusts the decision, so behaviour + # and receipts are unchanged when this is None. + self.token_issuer = token_issuer + self.token_audience = token_audience # ------------------------------------------------------------------ # def propose(self, action: ProposedAction) -> Decision: @@ -202,9 +216,14 @@ def _execute( crash between authorisation and execution leaves no trace the action was ever authorised. The failure is recorded as an ``execution``/deny receipt and the exception re-raised for the caller. + + If a token issuer is configured, a §9 authorization token is minted for + this `allow` and handed to the broker (which MAY verify it before + injecting) and returned on the Decision. """ + token = self._mint_token(rule, intent_hash, fingerprint, approval_id) try: - result = self.broker.execute(action) + result = self._broker_execute(action, token) except Exception as e: audit.append( phase="execution", @@ -236,4 +255,36 @@ def _execute( approval_id=approval_id, executed=True, result=result, + token=token, + ) + + # ------------------------------------------------------------------ # + def _mint_token(self, rule, intent_hash, fingerprint, approval_id): + """Mint a §9 token for an `allow`, or None if no issuer is configured. + + Only ever called from :meth:`_execute`, i.e. for `allow` and released + approvals — never for `deny`/`needs_approval`/`denied`/`consumed`, which + per spec §9 MUST NOT mint.""" + if self.token_issuer is None: + return None + return self.token_issuer.mint( + action_fingerprint=fingerprint, + intent_hash=intent_hash, + audience=self.token_audience, + approval_id=approval_id, + policy_version=self.policy.version, + rule=rule, ) + + def _broker_execute(self, action, token): + """Call the broker, passing the token when the broker accepts one. + + Brokers MAY take an optional ``token`` keyword (the shipped adapters do); + a 0.2/0.3-era adapter with a bare ``execute(action)`` keeps working.""" + if token is None: + return self.broker.execute(action) + try: + return self.broker.execute(action, token=token) + except TypeError: + # Broker predates the token kwarg — fall back, unchanged behaviour. + return self.broker.execute(action) diff --git a/delego/models.py b/delego/models.py index 2a2a0b7..fa83824 100644 --- a/delego/models.py +++ b/delego/models.py @@ -145,6 +145,10 @@ class Decision: approval_id: Optional[str] = None executed: bool = False result: Optional[dict] = None + # The §9 authorization token, when a token issuer is configured and the + # outcome is `allow` (or a released approval). A short-lived JWS a separated + # broker can verify before injecting a credential. None otherwise. + token: Optional[str] = None @property def allowed(self) -> bool: diff --git a/delego/token.py b/delego/token.py new file mode 100644 index 0000000..c8a58ef --- /dev/null +++ b/delego/token.py @@ -0,0 +1,327 @@ +"""The §9 authorization token — a portable, signed PDP→PEP decision artifact. + +The token is **not** delego's load-bearing control (that is the deterministic +decision, the fingerprint/intent binding, single-use approval, the audit chain, +and the PDP/PEP split). It is an *optional profile*: a way to carry a `allow` +verdict across a process or network boundary so a **separated** broker (PEP) can +verify "this exact action is authorized right now" without re-consulting the +authorizer (PDP). When the broker is in-process and already trusts the decision, +you don't need it. + +It is a compact **JWS / JWT** with `alg = EdDSA` (Ed25519). We build the compact +serialization here on `cryptography` (already a dependency) rather than pulling in +a JWT library: a security tool's token path should be short and auditable, and it +sidesteps the system-PyJWT conflicts the optional `mcp` extra already has to dodge. + +Two halves: + +* :class:`TokenIssuer` mints a token when the Authorizer renders `allow` (or + releases a human-approved action). Minting for any other outcome is a bug — + the issuer only mints for the two it is asked to. +* :func:`verify_token` is the broker side, and the crux. It performs the §9.1 + checks 1–4 (pin EdDSA / reject ``none`` and algorithm confusion; exact ``aud``; + ``exp``; single-use ``jti`` and ``cns``). Check 5 — recomputing the + ``action_fingerprint`` of the request the broker is *about to send* and + requiring it equals ``fpr`` — is :func:`require_fingerprint`, called by the + broker against the concrete action. That binding is what makes the token bind a + credential injection to *that exact action*, not merely "something in policy". + +**Key separation (spec §9 SHOULD).** The token signing key is distinct from the +audit-chain key: the two have different lifetimes and blast radii, and a +token-minting compromise must not also forge audit history. +""" + +from __future__ import annotations + +import base64 +import json +import os +import secrets +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Optional + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +from .util import canonical_json, sha256_hex + +# Spec §9: TTL SHOULD be <= 60s and MUST NOT exceed 300s. +DEFAULT_TTL_SECONDS = 45 +MAX_TTL_SECONDS = 300 + +# The claims a conformant token MUST carry (schema/authorization-token.json). +_REQUIRED_CLAIMS = ("iss", "aud", "iat", "exp", "jti", "cns", "fpr", "iht") + + +class TokenError(ValueError): + """A token could not be minted, or failed verification. + + Subclasses ``ValueError`` so existing ``except ValueError`` handlers treat a + bad token as a bad request. A broker that catches this **MUST NOT** inject a + credential — verification failing is a fail-closed deny, never a soft pass. + """ + + +# --------------------------------------------------------------------------- # +# base64url (no padding) — the JOSE encoding +# --------------------------------------------------------------------------- # +def _b64url(raw: bytes) -> str: + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _b64url_decode(s: str) -> bytes: + pad = "=" * (-len(s) % 4) + return base64.urlsafe_b64decode(s + pad) + + +def _now_epoch() -> int: + return int(datetime.now(timezone.utc).timestamp()) + + +def _kid_for(pub: Ed25519PublicKey) -> str: + """A stable key id: the first 16 hex of the SHA-256 of the public key's raw + bytes. Lets a verifier select among configured keys and supports rotation.""" + raw = pub.public_bytes( + encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw + ) + return sha256_hex(raw.hex())[:16] + + +# --------------------------------------------------------------------------- # +# keys +# --------------------------------------------------------------------------- # +def ensure_token_keys(priv_path: Path, pub_path: Path) -> None: + """Generate a token signing keypair if one doesn't exist. + + Deliberately a *separate* file from the audit signing key (spec §9): a + token-minting compromise must not be able to forge the audit chain. + """ + priv_path = Path(priv_path) + pub_path = Path(pub_path) + if priv_path.exists(): + return + priv_path.parent.mkdir(parents=True, exist_ok=True) + priv = Ed25519PrivateKey.generate() + priv_path.write_bytes( + priv.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + os.chmod(priv_path, 0o600) + pub_path.write_bytes( + priv.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + ) + + +# --------------------------------------------------------------------------- # +# minting +# --------------------------------------------------------------------------- # +@dataclass +class TokenIssuer: + """Mints short-lived authorization tokens for `allow` decisions. + + Construct with the issuer identity and the *token* signing key (not the audit + key). ``mint`` is called by the Firewall only when it renders `allow` or + releases an approved action; it is never called for any other outcome. + """ + + issuer: str + private_key: Ed25519PrivateKey + ttl_seconds: int = DEFAULT_TTL_SECONDS + + @classmethod + def from_files(cls, priv_path, pub_path, *, issuer: str, ttl_seconds: int = DEFAULT_TTL_SECONDS): + ensure_token_keys(Path(priv_path), Path(pub_path)) + priv = serialization.load_pem_private_key(Path(priv_path).read_bytes(), password=None) + return cls(issuer=issuer, private_key=priv, ttl_seconds=ttl_seconds) + + @property + def public_key(self) -> Ed25519PublicKey: + return self.private_key.public_key() + + @property + def kid(self) -> str: + return _kid_for(self.public_key) + + def mint( + self, + *, + action_fingerprint: str, + intent_hash: str, + audience: str, + approval_id: Optional[str] = None, + subject: Optional[str] = None, + policy_version: Optional[int] = None, + rule: Optional[str] = None, + # Test/vector hooks: pin the otherwise time/random fields so a CTK + # vector is byte-reproducible. Never set these in production. + _iat: Optional[int] = None, + _jti: Optional[str] = None, + _cns: Optional[str] = None, + ) -> str: + """Mint a compact JWS for an authorized action. Returns the token string. + + TTL is clamped to ``MAX_TTL_SECONDS`` (spec: MUST NOT exceed 300s). + ``jti`` (replay id) and ``cns`` (consumption nonce) are unique per call. + """ + ttl = min(self.ttl_seconds, MAX_TTL_SECONDS) + iat = _iat if _iat is not None else _now_epoch() + claims: dict[str, Any] = { + "iss": self.issuer, + "aud": audience, + "iat": iat, + "exp": iat + ttl, + "jti": _jti or secrets.token_hex(16), + "cns": _cns or secrets.token_hex(16), + "fpr": action_fingerprint, + "iht": intent_hash, + } + if approval_id is not None: + claims["apr"] = approval_id + if subject is not None: + claims["sub"] = subject + if policy_version is not None or rule is not None: + claims["pol"] = {"version": policy_version, "rule": rule} + + header = {"alg": "EdDSA", "typ": "JWT", "kid": self.kid} + signing_input = f"{_b64url(canonical_json(header).encode())}.{_b64url(canonical_json(claims).encode())}" + sig = self.private_key.sign(signing_input.encode("ascii")) + return f"{signing_input}.{_b64url(sig)}" + + +# --------------------------------------------------------------------------- # +# verification (the broker side, §9.1) +# --------------------------------------------------------------------------- # +def verify_token( + token: str, + *, + public_key: Ed25519PublicKey | None = None, + key_resolver: Callable[[str], Ed25519PublicKey] | None = None, + audience: str, + now: int | None = None, + leeway: int = 60, + seen_jti: "set[str] | None" = None, + consumed_cns: "set[str] | None" = None, + max_ttl: int = MAX_TTL_SECONDS, +) -> dict[str, Any]: + """Verify a token per spec §9.1 steps 1–4 and return its validated claims. + + Raises :class:`TokenError` on any failure — a broker that catches it **MUST + NOT** inject a credential. This does **not** perform step 5 (the fingerprint + re-check); call :func:`require_fingerprint` with the concrete action the + broker is about to send. + + Key selection is by **verifier configuration**, never the token: pass a + single ``public_key``, or a ``key_resolver`` mapping the header ``kid`` to a + key *from your own key set*. The header's ``alg`` is never trusted to decide + how to verify — EdDSA is pinned and ``none`` (and anything else) is rejected. + That is the JWT algorithm-confusion defense (spec §9.1 step 1, §11). + + ``seen_jti`` / ``consumed_cns``, if given, are checked for replay and, on + success, the token's ``jti``/``cns`` are added to them (so a second + verification of the same token is refused). Pass shared, persisted sets in a + real deployment; each is retained until at least the token's ``exp``. + """ + if public_key is None and key_resolver is None: + raise TokenError("verify_token requires public_key or key_resolver") + + parts = token.split(".") + if len(parts) != 3: + raise TokenError("malformed token: expected three dot-separated segments") + header_b64, payload_b64, sig_b64 = parts + + # --- 1. header: pin EdDSA, reject `none`/confusion; key from OUR config --- # + try: + header = json.loads(_b64url_decode(header_b64)) + except Exception as e: + raise TokenError(f"unparseable token header: {e}") from e + if not isinstance(header, dict) or header.get("alg") != "EdDSA": + # Covers alg="none", alg="HS256" (key-confusion), or a missing alg. + raise TokenError(f"token alg must be EdDSA, got {header.get('alg')!r} (algorithm-confusion guard)") + if key_resolver is not None: + kid = header.get("kid") + if not isinstance(kid, str): + raise TokenError("token header missing 'kid' but a key_resolver was given") + try: + key = key_resolver(kid) + except Exception as e: + raise TokenError(f"no configured key for kid {kid!r}: {e}") from e + if key is None: + raise TokenError(f"no configured key for kid {kid!r}") + else: + key = public_key + + # --- signature over the exact signing input --- # + try: + key.verify(_b64url_decode(sig_b64), f"{header_b64}.{payload_b64}".encode("ascii")) + except InvalidSignature as e: + raise TokenError("bad token signature") from e + except Exception as e: + raise TokenError(f"signature verification error: {e}") from e + + # --- claims --- # + try: + claims = json.loads(_b64url_decode(payload_b64)) + except Exception as e: + raise TokenError(f"unparseable token claims: {e}") from e + if not isinstance(claims, dict): + raise TokenError("token claims must be a JSON object") + missing = [c for c in _REQUIRED_CLAIMS if c not in claims] + if missing: + raise TokenError(f"token missing required claim(s): {missing}") + + # --- 2. exp / aud --- # + now = now if now is not None else _now_epoch() + iat, exp = claims["iat"], claims["exp"] + if not isinstance(iat, int) or not isinstance(exp, int): + raise TokenError("token iat/exp must be integers (epoch seconds)") + if exp - iat > max_ttl: + raise TokenError(f"token TTL {exp - iat}s exceeds the {max_ttl}s maximum") + if now >= exp + leeway: + raise TokenError("token has expired") + if now < iat - leeway: + raise TokenError("token used before its issued-at time (clock skew beyond leeway)") + if claims["aud"] != audience: + # Exact match only — no wildcard, no prefix (spec §9.1 step 2). + raise TokenError(f"token aud {claims['aud']!r} != this broker's identifier {audience!r}") + + # --- 3. jti replay / 4. cns single-use --- # + if seen_jti is not None and claims["jti"] in seen_jti: + raise TokenError("token jti already seen (replay)") + if consumed_cns is not None and claims["cns"] in consumed_cns: + raise TokenError("token cns already consumed (single-use)") + + if seen_jti is not None: + seen_jti.add(claims["jti"]) + if consumed_cns is not None: + consumed_cns.add(claims["cns"]) + return claims + + +def require_fingerprint(claims: dict[str, Any], action) -> None: + """Spec §9.1 step 5 — the point of the token. + + Recompute the ``action_fingerprint`` of the request the broker is about to + send and require it equals the token's ``fpr``. A token minted for action A + cannot release action B, even if both are in policy scope — and an agent that + declared one set of params but tries to send another is refused here, the + enforcement-side closure of the "declared vs. sent" gap. + + Raises :class:`TokenError` on mismatch; the broker **MUST NOT** inject. + """ + if action.fingerprint != claims.get("fpr"): + raise TokenError( + "action fingerprint does not match the token's fpr: the request the " + "broker is about to send is not the action that was authorized" + ) diff --git a/pyproject.toml b/pyproject.toml index d448485..5e58926 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "delego" -version = "0.3.2" +version = "0.3.3" description = "Intent-bound action authorization for AI agents — authorise before a credential is used, require human approval for sensitive actions, and prove every action traces back to the instruction that authorised it." readme = "README.md" requires-python = ">=3.10" diff --git a/server.json b/server.json index c923da6..fce46a3 100644 --- a/server.json +++ b/server.json @@ -3,7 +3,7 @@ "name": "io.github.Delego-Dev/delego", "description": "Intent-bound action authorization for AI agents: policy, human approval, and a signed audit trail.", "websiteUrl": "https://delegohq.com", - "version": "0.3.2", + "version": "0.3.3", "repository": { "url": "https://github.com/Delego-Dev/delego", "source": "github" @@ -13,7 +13,7 @@ "registryType": "pypi", "registryBaseUrl": "https://pypi.org", "identifier": "delego", - "version": "0.3.2", + "version": "0.3.3", "runtimeHint": "uvx", "runtimeArguments": [ { diff --git a/tests/test_broker.py b/tests/test_broker.py index f843951..9bd6586 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -94,3 +94,30 @@ def test_firewall_allow_executes_through_the_gateway(tmp_path, gateway): assert d.outcome == "allow" and d.executed is True assert d.result["response"]["injected"] is True assert fw.audit.verify()[0] is True + + +def test_firewall_forwards_a_verifiable_token_to_the_gateway(tmp_path, gateway): + # With the §9 profile on, the broker forwards the minted token; a separated + # gateway can verify it and re-check the fingerprint of what it will send. + from delego import require_fingerprint, verify_token + + home = tmp_path / "home" + home.mkdir() + (home / "policy.yaml").write_text(_POLICY, encoding="utf-8") + fw = build_firewall( + Paths.resolve(home), + broker=HTTPProxyBroker(gateway), + mint_tokens=True, + token_audience="broker:default", + ) + + action = ProposedAction("read my data", "GET", "https://api.example.com/x", {}) + d = fw.propose(action) + + assert d.outcome == "allow" and d.token is not None + forwarded = d.result["response"]["forwarded"] + assert forwarded["authorization_token"] == d.token # reached the gateway + + # The gateway verifies the token and binds it to the exact action (step 5). + claims = verify_token(d.token, public_key=fw.token_issuer.public_key, audience="broker:default") + require_fingerprint(claims, action) diff --git a/tests/test_token.py b/tests/test_token.py new file mode 100644 index 0000000..ffd8287 --- /dev/null +++ b/tests/test_token.py @@ -0,0 +1,182 @@ +"""The §9 authorization token: minting only for `allow`, and the §9.1 verifier. + +The verifier is security-critical — JWT algorithm confusion is the classic +footgun — so these pin every failure mode the spec names: EdDSA-only (reject +`none` and HS-style confusion), exact `aud`, expiry, single-use `jti`/`cns`, and +the fingerprint re-check (step 5) that binds a credential release to one action. +""" + +from __future__ import annotations + +import base64 +import json +from pathlib import Path + +import pytest + +_POLICY_EXAMPLE = Path(__file__).resolve().parent.parent / "policy.example.yaml" + +from delego import ( + ProposedAction, + TokenError, + TokenIssuer, + build_firewall, + require_fingerprint, + verify_token, +) +from delego.config import Paths +from delego.token import ensure_token_keys +from cryptography.hazmat.primitives import serialization + +AUD = "broker:onecli" +FPR = "c70d4ee57957202087887cb5e9d32222977b728bd06947b7761c283b6d4ed394" +IHT = "76f8eef1b97e1213a59eec28cedf15bb999fdb00a3fd17f8343bc4676fdbb4f3" + + +@pytest.fixture +def issuer(tmp_path): + return TokenIssuer.from_files( + tmp_path / "token_key.pem", tmp_path / "token_key.pub", issuer="delego:test" + ) + + +def _b64(obj) -> str: + return base64.urlsafe_b64encode(json.dumps(obj, separators=(",", ":")).encode()).rstrip(b"=").decode() + + +# --- minting is gated to `allow` / released approvals -------------------------- + +def _fw(tmp_path): + home = tmp_path / "home" + home.mkdir() + (home / "policy.yaml").write_text(_POLICY_EXAMPLE.read_text(encoding="utf-8"), encoding="utf-8") + return build_firewall(Paths.resolve(home), mint_tokens=True, token_audience=AUD) + + +def test_allow_mints_a_token_and_it_verifies(tmp_path): + fw = _fw(tmp_path) + d = fw.propose(ProposedAction("read my account details", "GET", "https://api.example.com/accounts/me", {})) + assert d.outcome == "allow" and d.token is not None + claims = verify_token(d.token, public_key=fw.token_issuer.public_key, audience=AUD) + assert claims["fpr"] == d.action_fingerprint + assert claims["iht"] == d.intent_hash + assert claims["pol"] == {"version": 1, "rule": "read-accounts"} + + +def test_deny_and_needs_approval_mint_no_token(tmp_path): + fw = _fw(tmp_path) + deny = fw.propose(ProposedAction("share", "POST", "https://api.example.com/accounts/me/permissions", {"grant": "x"})) + assert deny.outcome == "deny" and deny.token is None + parked = fw.propose(ProposedAction("order", "POST", "https://api.example.com/orders", + {"amount": 100, "currency": "USD", "destination": "internal"})) + assert parked.outcome == "needs_approval" and parked.token is None + + +def test_released_approval_mints_a_token(tmp_path): + fw = _fw(tmp_path) + order = ProposedAction("place a small order", "POST", "https://api.example.com/orders", + {"amount": 2400, "currency": "USD", "destination": "internal"}) + parked = fw.propose(order) + fw.approvals.decide(parked.approval_id, approved=True, approver="human") + released = fw.resolve(parked.approval_id, order) + assert released.outcome == "allow" and released.token is not None + claims = verify_token(released.token, public_key=fw.token_issuer.public_key, audience=AUD) + assert claims["apr"] == parked.approval_id + + +# --- the §9.1 verifier -------------------------------------------------------- + +def _mint(issuer, **over): + kw = dict(action_fingerprint=FPR, intent_hash=IHT, audience=AUD, rule="place-order", policy_version=1) + kw.update(over) + return issuer.mint(**kw) + + +def test_valid_token_round_trips(issuer): + claims = verify_token(_mint(issuer), public_key=issuer.public_key, audience=AUD) + assert claims["fpr"] == FPR and claims["aud"] == AUD + + +def test_alg_none_is_rejected(issuer): + # Strip to an unsigned alg=none token — the classic JWT bypass. + _, payload, _ = _mint(issuer).split(".") + forged = f"{_b64({'alg': 'none', 'typ': 'JWT'})}.{payload}." + with pytest.raises(TokenError, match="EdDSA"): + verify_token(forged, public_key=issuer.public_key, audience=AUD) + + +def test_tampered_payload_fails_signature(issuer): + header, _, sig = _mint(issuer).split(".") + swapped = _b64({"iss": "delego:test", "aud": AUD, "iat": 1, "exp": 10**12, + "jti": "x", "cns": "y", "fpr": "0" * 64, "iht": IHT}) + with pytest.raises(TokenError, match="signature"): + verify_token(f"{header}.{swapped}.{sig}", public_key=issuer.public_key, audience=AUD) + + +def test_wrong_audience_is_rejected(issuer): + with pytest.raises(TokenError, match="aud"): + verify_token(_mint(issuer), public_key=issuer.public_key, audience="broker:evil") + + +def test_expired_token_is_rejected(issuer): + tok = _mint(issuer, _iat=1_000_000) # long past; exp = iat + ttl + with pytest.raises(TokenError, match="expired"): + verify_token(tok, public_key=issuer.public_key, audience=AUD, now=2_000_000) + + +def test_ttl_over_max_is_rejected(issuer): + # Hand-craft a token whose exp - iat exceeds 300s, signed correctly. + iat = 1_000_000 + header = {"alg": "EdDSA", "typ": "JWT", "kid": issuer.kid} + claims = {"iss": "delego:test", "aud": AUD, "iat": iat, "exp": iat + 3600, + "jti": "j", "cns": "c", "fpr": FPR, "iht": IHT} + si = f"{_b64(header)}.{_b64(claims)}" + sig = issuer.private_key.sign(si.encode()) + tok = f"{si}.{base64.urlsafe_b64encode(sig).rstrip(b'=').decode()}" + with pytest.raises(TokenError, match="TTL"): + verify_token(tok, public_key=issuer.public_key, audience=AUD, now=iat + 1) + + +def test_jti_replay_and_cns_single_use(issuer): + seen, consumed = set(), set() + tok = _mint(issuer) + verify_token(tok, public_key=issuer.public_key, audience=AUD, seen_jti=seen, consumed_cns=consumed) + # Same token again: jti is now seen. + with pytest.raises(TokenError, match="replay"): + verify_token(tok, public_key=issuer.public_key, audience=AUD, seen_jti=seen, consumed_cns=consumed) + # A fresh token shares no jti but, if it reused a cns, would be refused. + tok2 = _mint(issuer, _cns=list(consumed)[0]) + with pytest.raises(TokenError, match="cns"): + verify_token(tok2, public_key=issuer.public_key, audience=AUD, seen_jti=set(), consumed_cns=consumed) + + +def test_wrong_key_is_rejected(issuer, tmp_path): + ensure_token_keys(tmp_path / "other.pem", tmp_path / "other.pub") + other_pub = serialization.load_pem_public_key((tmp_path / "other.pub").read_bytes()) + with pytest.raises(TokenError, match="signature"): + verify_token(_mint(issuer), public_key=other_pub, audience=AUD) + + +def test_key_resolver_selects_by_kid(issuer): + # kid is taken from the header only to *select among configured keys* — the + # key itself never comes from the token. + keys = {issuer.kid: issuer.public_key} + claims = verify_token(_mint(issuer), key_resolver=keys.get, audience=AUD) + assert claims["fpr"] == FPR + with pytest.raises(TokenError, match="no configured key"): + verify_token(_mint(issuer), key_resolver={}.get, audience=AUD) + + +# --- step 5: the fingerprint re-check (the #7 closure) ------------------------ + +def test_require_fingerprint_binds_to_one_exact_action(issuer): + authorized = ProposedAction("place a small order", "POST", "https://api.example.com/orders", + {"amount": 2400, "currency": "USD", "destination": "internal"}) + claims = verify_token(_mint(issuer, action_fingerprint=authorized.fingerprint), + public_key=issuer.public_key, audience=AUD) + require_fingerprint(claims, authorized) # ok + # Agent declared one set of params, tries to send another → refused at the PEP. + substituted = ProposedAction("place a small order", "POST", "https://api.example.com/orders", + {"amount": 2400, "currency": "USD", "destination": "internal", "recipient": "attacker"}) + with pytest.raises(TokenError, match="not the action that was authorized"): + require_fingerprint(claims, substituted)