From aa28ef353acf3d3e41c3e3bfef115b8475650858 Mon Sep 17 00:00:00 2001 From: tegwick Date: Fri, 24 Jul 2026 00:10:41 +0200 Subject: [PATCH] QONTO-WP-0004-T03: verify key-cape IAM Profile tokens Replaces the interim shared-secret bearer token's role as the identity boundary with real key-cape JWKS-based verification, closing the gap docs/mcp-integration.md called out explicitly ("no OIDC issuer exists in this fleet yet") -- key-cape's /jwks is a standard RS256 endpoint and needed no key-cape-side work to consume. KeyCapeTokenVerifier fetches and caches signing keys over httpx (matching FlexAuthCheckClient's pattern elsewhere in this codebase), validates iss/aud/exp and the IAM Profile v0.3 required claims, and derives ActorClaims from the token (tenant, scopes, and a lane inferred from the roles claim). Wired into auth.actor_claims_from_headers, the single seam both REST and MCP already used -- a verified bearer token now takes precedence over self-asserted X-Actor-* headers, and can be made mandatory via QONTO_KEY_CAPE_REQUIRED once real tokens are issued to callers. Off by default (no QONTO_KEY_CAPE_JWKS_URL set) so existing deployments are unaffected until configured. The QONTO_ASSISTANT_MCP_TOKEN shared secret remains as a documented local-dev/legacy fallback, not the auth boundary going forward. Verified: 13 new tests (test_key_cape_auth.py) covering valid/expired/ wrong-audience/wrong-issuer/missing-claim/unknown-key/rotated-key tokens plus the auth.py precedence and required-vs-optional integration paths, using a real generated RSA keypair and JWKS served over httpx.MockTransport. Full suite -> 52 passed; REST and MCP smokes both still pass against fixtures; compileall clean. Co-Authored-By: Claude Sonnet 5 --- docs/mcp-integration.md | 44 +++-- pyproject.toml | 1 + src/qonto_assistant/app.py | 20 ++- src/qonto_assistant/auth.py | 33 +++- src/qonto_assistant/config.py | 12 ++ src/qonto_assistant/key_cape_auth.py | 151 ++++++++++++++++ src/qonto_assistant/mcp_server.py | 24 ++- tests/test_api.py | 6 + tests/test_key_cape_auth.py | 256 +++++++++++++++++++++++++++ 9 files changed, 514 insertions(+), 33 deletions(-) create mode 100644 src/qonto_assistant/key_cape_auth.py create mode 100644 tests/test_key_cape_auth.py diff --git a/docs/mcp-integration.md b/docs/mcp-integration.md index 4421307..8498c86 100644 --- a/docs/mcp-integration.md +++ b/docs/mcp-integration.md @@ -22,18 +22,26 @@ Two independent layers, same as the blueprint's identity separation (`specs/ArchitectureBlueprint.md` §4.7): 1. **Workload auth (is this caller allowed to reach the service at all?)** - Today: a static shared-secret bearer token - (`QONTO_ASSISTANT_MCP_TOKEN`), checked by `BearerTokenAuthMiddleware` - in front of `/mcp`. If the env var is unset, the middleware is not - installed — used only for local fixture-backed smoke work, never for a - deployment holding real credentials. + Two mechanisms now coexist: - This is **not** the OIDC/workload-identity target the blueprint - describes — there is no OIDC issuer in this fleet yet to federate - against. A shared-secret bearer token is the deployable "or similar" - primitive for now; upgrading to real OIDC/workload identity is - fleet-level follow-on work (tracked as a Phase 2 gap, not closed by this - task). + - **key-cape IAM Profile tokens (QONTO-WP-0004-T03, current target).** + Set `QONTO_KEY_CAPE_JWKS_URL` (and, if not using the defaults, + `QONTO_KEY_CAPE_ISSUER`/`QONTO_KEY_CAPE_AUDIENCE`) and a + `KeyCapeTokenVerifier` is built and wired into both REST and MCP + (`auth.actor_claims_from_headers`). A verified `Authorization: Bearer + ` takes precedence over self-asserted headers; actor identity + (tenant, scopes, and a lane derived from the profile's `roles` claim) + comes from the verified token, not the caller's assertion. Set + `QONTO_KEY_CAPE_REQUIRED=true` to reject any request without a valid + bearer token outright, once real key-cape tokens are actually being + issued to callers — until then, leave it `false` so unconfigured + deployments keep working during rollout. + - **Shared-secret bearer token (legacy/local-dev).** + `QONTO_ASSISTANT_MCP_TOKEN`, checked by `BearerTokenAuthMiddleware` in + front of `/mcp` only (REST has no equivalent — see below). If unset, + the middleware is not installed. This predates the key-cape + integration and should be treated as fixture/local-dev only, never the + auth boundary for a deployment holding real credentials. This token is a **service credential**, not a bank credential — it never reaches Qonto and is never logged. Treat it like any other shared @@ -41,11 +49,15 @@ Two independent layers, same as the blueprint's identity separation plaintext config committed to a repo. 2. **Actor identity (who is calling, for policy and audit purposes?)** - Same `X-Actor-ID` / `X-Tenant-ID` / `X-Actor-Lane` / `X-Actor-Scopes` - header convention REST already uses (`auth.actor_claims_from_headers`). - These are self-asserted today, not cryptographically bound to the bearer - token — tightening that binding is exactly what Phase 3's flex-auth - resource (`finance.qonto.read`) is for. + When a key-cape verifier is configured, actor identity comes from the + verified token's claims (`sub` → actor id, `tenant` → tenant id, `roles` + → lane, `scope`/`scp` → scopes) — see + `src/qonto_assistant/key_cape_auth.py`. Otherwise, the same `X-Actor-ID` + / `X-Tenant-ID` / `X-Actor-Lane` / `X-Actor-Scopes` header convention + REST and MCP both use (`auth.actor_claims_from_headers`) applies, + self-asserted and not cryptographically bound to anything — the gap + that configuring key-cape closes. Tightening this further with a live + `flex-auth` decision on `finance.qonto.read` is QONTO-WP-0004-T04. ## One shared client config snippet diff --git a/pyproject.toml b/pyproject.toml index 4b2b4ac..34e7598 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "fastapi>=0.115,<1.0", "httpx>=0.27,<1.0", "mcp>=1.9,<2.0", + "PyJWT[crypto]>=2.9,<3.0", "PyYAML>=6.0,<7.0", "uvicorn[standard]>=0.30,<1.0", ] diff --git a/src/qonto_assistant/app.py b/src/qonto_assistant/app.py index ce9b46f..28d907a 100644 --- a/src/qonto_assistant/app.py +++ b/src/qonto_assistant/app.py @@ -16,6 +16,7 @@ from qonto_assistant.auth import actor_claims_from_request from qonto_assistant.config import Settings from qonto_assistant.credentials import build_credential_provider from qonto_assistant.errors import QontoAssistantError, UpstreamError +from qonto_assistant.key_cape_auth import KeyCapeTokenVerifier from qonto_assistant.mcp_auth import BearerTokenAuthMiddleware from qonto_assistant.mcp_server import create_mcp_server from qonto_assistant.policy import PolicyEngine @@ -32,6 +33,7 @@ def create_app( audit_logger: AuditLogger | None = None, rate_limiter: RateLimiter | None = None, concurrency_limiter: ConcurrencyLimiter | None = None, + key_cape_verifier: KeyCapeTokenVerifier | None = None, ) -> FastAPI: settings = settings or Settings.from_env() audit_logger = audit_logger or AuditLogger() @@ -41,8 +43,18 @@ def create_app( rate_limiter=rate_limiter, concurrency_limiter=concurrency_limiter, ) + if key_cape_verifier is None and settings.key_cape_jwks_url: + key_cape_verifier = KeyCapeTokenVerifier( + jwks_url=settings.key_cape_jwks_url, + issuer=settings.key_cape_issuer, + audience=settings.key_cape_audience, + required=settings.key_cape_required, + default_lane=settings.default_actor_lane, + timeout_seconds=settings.key_cape_timeout_seconds, + cache_seconds=settings.key_cape_cache_seconds, + ) - mcp_server = create_mcp_server(settings=settings, service=service) + mcp_server = create_mcp_server(settings=settings, service=service, key_cape_verifier=key_cape_verifier) mcp_app = mcp_server.streamable_http_app() if settings.mcp_auth_token: mcp_app.add_middleware(BearerTokenAuthMiddleware, token=settings.mcp_auth_token) @@ -78,7 +90,7 @@ def create_app( @app.get("/v1/accounts") async def get_accounts(request: Request) -> JSONResponse: - claims = actor_claims_from_request(request, settings) + claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) payload = await run_in_threadpool(service.get_accounts, claims=claims, request_id=request_id) return JSONResponse(content=payload, headers={"X-Request-ID": request_id}) @@ -93,7 +105,7 @@ def create_app( status: str | None = "completed", side: str | None = None, ) -> JSONResponse: - claims = actor_claims_from_request(request, settings) + claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) payload = await run_in_threadpool( service.list_transactions, @@ -114,7 +126,7 @@ def create_app( window_days: int = 31, page_size: int = 50, ) -> JSONResponse: - claims = actor_claims_from_request(request, settings) + claims = actor_claims_from_request(request, settings, key_cape_verifier=key_cape_verifier) request_id = _request_id(request) payload = await run_in_threadpool( service.get_snapshot, diff --git a/src/qonto_assistant/auth.py b/src/qonto_assistant/auth.py index c1d0c02..3acbdcd 100644 --- a/src/qonto_assistant/auth.py +++ b/src/qonto_assistant/auth.py @@ -6,14 +6,32 @@ from fastapi import Request from qonto_assistant.config import Settings from qonto_assistant.contracts import ActorClaims +from qonto_assistant.key_cape_auth import KeyCapeAuthError, KeyCapeTokenVerifier -def actor_claims_from_headers(headers: Mapping[str, str], settings: Settings) -> ActorClaims: +def actor_claims_from_headers( + headers: Mapping[str, str], + settings: Settings, + *, + key_cape_verifier: KeyCapeTokenVerifier | None = None, +) -> ActorClaims: """Shared REST/MCP claims parsing so both transports enforce identical actor identity. - Real workload/OIDC auth for the MCP transport is QONTO-WP-0003-T03; until - then MCP callers use the same X-Actor-* header convention as REST. + When a `KeyCapeTokenVerifier` is configured (QONTO-WP-0004-T03), a + verified `Authorization: Bearer ` takes precedence over the + self-asserted `X-Actor-*` header convention -- closing the gap + `docs/mcp-integration.md` called out explicitly. If the verifier is + marked `required`, a missing or invalid bearer token is rejected + outright rather than silently falling back to self-asserted headers. """ + authorization = headers.get("authorization", "") + if authorization.lower().startswith("bearer ") and key_cape_verifier is not None: + token = authorization.split(" ", 1)[1].strip() + return key_cape_verifier.verify(token) + + if key_cape_verifier is not None and key_cape_verifier.required: + raise KeyCapeAuthError("bearer_token_required") + actor_id = headers.get("x-actor-id", "anonymous") tenant_id = headers.get("x-tenant-id", settings.default_tenant_id) lane = headers.get("x-actor-lane", settings.default_actor_lane) @@ -22,5 +40,10 @@ def actor_claims_from_headers(headers: Mapping[str, str], settings: Settings) -> return ActorClaims(actor_id=actor_id, tenant_id=tenant_id, lane=lane, scopes=scopes) -def actor_claims_from_request(request: Request, settings: Settings) -> ActorClaims: - return actor_claims_from_headers(request.headers, settings) +def actor_claims_from_request( + request: Request, + settings: Settings, + *, + key_cape_verifier: KeyCapeTokenVerifier | None = None, +) -> ActorClaims: + return actor_claims_from_headers(request.headers, settings, key_cape_verifier=key_cape_verifier) diff --git a/src/qonto_assistant/config.py b/src/qonto_assistant/config.py index 1f551ac..23b752c 100644 --- a/src/qonto_assistant/config.py +++ b/src/qonto_assistant/config.py @@ -32,6 +32,12 @@ class Settings: deny_escalation_threshold: int deny_escalation_window_seconds: int deny_escalation_lockout_seconds: int + key_cape_jwks_url: str | None + key_cape_issuer: str + key_cape_audience: str + key_cape_required: bool + key_cape_cache_seconds: float + key_cape_timeout_seconds: float credential_source: str openbao_path: str openbao_command: str @@ -74,6 +80,12 @@ class Settings: deny_escalation_threshold=int(os.getenv("QONTO_DENY_ESCALATION_THRESHOLD", "3")), deny_escalation_window_seconds=int(os.getenv("QONTO_DENY_ESCALATION_WINDOW_SECONDS", "60")), deny_escalation_lockout_seconds=int(os.getenv("QONTO_DENY_ESCALATION_LOCKOUT_SECONDS", "300")), + key_cape_jwks_url=os.getenv("QONTO_KEY_CAPE_JWKS_URL") or None, + key_cape_issuer=os.getenv("QONTO_KEY_CAPE_ISSUER", "https://key-cape.netkingdom"), + key_cape_audience=os.getenv("QONTO_KEY_CAPE_AUDIENCE", "qonto-assistant"), + key_cape_required=os.getenv("QONTO_KEY_CAPE_REQUIRED", "false").lower() == "true", + key_cape_cache_seconds=float(os.getenv("QONTO_KEY_CAPE_CACHE_SECONDS", "300")), + key_cape_timeout_seconds=float(os.getenv("QONTO_KEY_CAPE_TIMEOUT_SECONDS", "5")), credential_source=os.getenv("QONTO_CREDENTIAL_SOURCE", "env"), openbao_path=os.getenv("QONTO_OPENBAO_PATH", "tenants/binky/qonto-api"), openbao_command=os.getenv("QONTO_OPENBAO_COMMAND", "bao"), diff --git a/src/qonto_assistant/key_cape_auth.py b/src/qonto_assistant/key_cape_auth.py new file mode 100644 index 0000000..1bb81d1 --- /dev/null +++ b/src/qonto_assistant/key_cape_auth.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import time +from collections.abc import Sequence +from typing import Any + +import httpx +import jwt +from jwt.algorithms import RSAAlgorithm + +from qonto_assistant.contracts import ActorClaims +from qonto_assistant.errors import QontoAssistantError + +# net-kingdom/canon/standards/iam-profile_v0.3.md "Core Claims" table -- +# the subset this verifier requires to be present on every token. +REQUIRED_CLAIMS: tuple[str, ...] = ("iss", "sub", "aud", "exp", "iat", "tenant", "principal_type") + +# AutonomyPolicy lanes (binky-control/AutonomyPolicy.md), not part of the +# IAM Profile itself. Until a canonical claim-to-lane mapping exists, this +# verifier looks for a lane name inside the profile's `roles` claim and +# falls back to the service's configured default -- documented here rather +# than silently defaulting for every token. +_KNOWN_LANES: tuple[str, ...] = ("green", "blue", "yellow", "orange", "red") + + +class KeyCapeAuthError(QontoAssistantError): + error_code = "key_cape_auth_failed" + status_code = 401 + + def __init__(self, message: str) -> None: + super().__init__(message) + self.error_code = "key_cape_auth_failed" + + +class KeyCapeTokenVerifier: + """Verifies key-cape-issued IAM Profile tokens against its JWKS endpoint. + + Fetches and caches signing keys directly over httpx (matching this + codebase's existing FlexAuthCheckClient/QontoClient pattern) rather than + PyJWT's built-in network-fetching PyJWKClient, so tests can inject a + MockTransport the same way the rest of the codebase does. + """ + + def __init__( + self, + *, + jwks_url: str, + issuer: str, + audience: str, + required: bool, + default_lane: str = "green", + timeout_seconds: float = 5.0, + cache_seconds: float = 300.0, + transport: httpx.BaseTransport | None = None, + clock: Any = time.monotonic, + ) -> None: + self.jwks_url = jwks_url + self.issuer = issuer + self.audience = audience + self.required = required + self.default_lane = default_lane + self.cache_seconds = cache_seconds + self.clock = clock + self._client = httpx.Client(timeout=httpx.Timeout(timeout_seconds), transport=transport) + self._cached_keys: dict[str, Any] = {} + self._cached_at: float | None = None + + def close(self) -> None: + self._client.close() + + def verify(self, token: str) -> ActorClaims: + self._ensure_keys() + + try: + header = jwt.get_unverified_header(token) + except jwt.InvalidTokenError as exc: + raise KeyCapeAuthError("malformed_token") from exc + + kid = header.get("kid") + key = self._cached_keys.get(kid) if kid else None + if key is None: + # Key rotation: force one refresh before giving up. + self._refresh_keys() + key = self._cached_keys.get(kid) if kid else None + if key is None: + raise KeyCapeAuthError("unknown_signing_key") + + try: + claims = jwt.decode( + token, + key=key, + algorithms=["RS256"], + audience=self.audience, + issuer=self.issuer, + options={"require": list(REQUIRED_CLAIMS)}, + ) + except jwt.InvalidTokenError as exc: + raise KeyCapeAuthError(f"invalid_token:{exc}") from exc + + return ActorClaims( + actor_id=str(claims["sub"]), + tenant_id=str(claims["tenant"]), + lane=_lane_from_roles(claims.get("roles"), default=self.default_lane), + scopes=frozenset(_coerce_scopes(claims.get("scope") or claims.get("scp"))), + ) + + def _ensure_keys(self) -> None: + if self._cached_keys and self._cached_at is not None: + if (self.clock() - self._cached_at) < self.cache_seconds: + return + self._refresh_keys() + + def _refresh_keys(self) -> None: + try: + response = self._client.get(self.jwks_url) + response.raise_for_status() + payload = response.json() + except httpx.HTTPError as exc: + raise KeyCapeAuthError("jwks_unavailable") from exc + + keys: dict[str, Any] = {} + for jwk in payload.get("keys", []): + kid = jwk.get("kid") + if not kid: + continue + try: + keys[kid] = RSAAlgorithm.from_jwk(jwk) + except (ValueError, TypeError): + continue + + if keys: + self._cached_keys = keys + self._cached_at = self.clock() + + +def _coerce_scopes(value: Any) -> Sequence[str]: + if isinstance(value, str): + return [item for item in value.split() if item] + if isinstance(value, list): + return [str(item) for item in value] + return [] + + +def _lane_from_roles(roles: Any, *, default: str) -> str: + if not isinstance(roles, list): + return default + lowered = {str(role).lower() for role in roles} + for lane in _KNOWN_LANES: + if lane in lowered: + return lane + return default diff --git a/src/qonto_assistant/mcp_server.py b/src/qonto_assistant/mcp_server.py index b580686..b7ed9e1 100644 --- a/src/qonto_assistant/mcp_server.py +++ b/src/qonto_assistant/mcp_server.py @@ -10,17 +10,25 @@ from starlette.applications import Starlette from qonto_assistant import __version__ from qonto_assistant.auth import actor_claims_from_headers from qonto_assistant.config import Settings +from qonto_assistant.key_cape_auth import KeyCapeTokenVerifier from qonto_assistant.service import CapabilityService -def create_mcp_server(*, settings: Settings, service: CapabilityService | None = None) -> FastMCP: +def create_mcp_server( + *, + settings: Settings, + service: CapabilityService | None = None, + key_cape_verifier: KeyCapeTokenVerifier | None = None, +) -> FastMCP: """Build the MCP adapter on the same capability core as the REST surface. Every tool here maps 1:1 to a REST-allowed capability id and goes through `CapabilityService`, which in turn calls the shared `PolicyEngine.decide()` -- no separate MCP policy path. Client auth is - still the simple X-Actor-* header convention (QONTO-WP-0003-T03 upgrades - this to real OIDC/workload auth); no bank secrets are ever exposed here. + the same seam REST uses (`auth.actor_claims_from_headers`): a verified + key-cape bearer token when `key_cape_verifier` is configured + (QONTO-WP-0004-T03), else the X-Actor-* header convention. No bank + secrets are ever exposed here. """ server = FastMCP( name=settings.service_name, @@ -45,7 +53,7 @@ def create_mcp_server(*, settings: Settings, service: CapabilityService | None = @server.tool() def qonto_org_summary(ctx: Context) -> dict[str, Any]: """Organization name and per-account balances (org_summary capability).""" - claims = _claims(ctx, settings) + claims = _claims(ctx, settings, key_cape_verifier) return service.get_accounts(claims=claims, request_id=_request_id(ctx), protocol="mcp") @server.tool() @@ -59,7 +67,7 @@ def create_mcp_server(*, settings: Settings, service: CapabilityService | None = side: str | None = None, ) -> dict[str, Any]: """Capped, filtered transaction history (list_transactions capability).""" - claims = _claims(ctx, settings) + claims = _claims(ctx, settings, key_cape_verifier) return service.list_transactions( claims=claims, request_id=_request_id(ctx), @@ -79,7 +87,7 @@ def create_mcp_server(*, settings: Settings, service: CapabilityService | None = page_size: int = 50, ) -> dict[str, Any]: """Normalized recurring-cost hints, not a raw export (cost_run_rate_hints capability).""" - claims = _claims(ctx, settings) + claims = _claims(ctx, settings, key_cape_verifier) return service.get_cost_run_rate_hints( claims=claims, request_id=_request_id(ctx), @@ -95,9 +103,9 @@ def mcp_asgi_app(*, settings: Settings, service: CapabilityService | None = None return create_mcp_server(settings=settings, service=service).streamable_http_app() -def _claims(ctx: Context, settings: Settings): +def _claims(ctx: Context, settings: Settings, key_cape_verifier: KeyCapeTokenVerifier | None = None): headers = _headers_from_context(ctx) - return actor_claims_from_headers(headers, settings) + return actor_claims_from_headers(headers, settings, key_cape_verifier=key_cape_verifier) def _headers_from_context(ctx: Context) -> Mapping[str, str]: diff --git a/tests/test_api.py b/tests/test_api.py index 530b60b..adfbcb8 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -38,6 +38,12 @@ def _settings() -> Settings: deny_escalation_threshold=3, deny_escalation_window_seconds=60, deny_escalation_lockout_seconds=300, + key_cape_jwks_url=None, + key_cape_issuer="https://key-cape.netkingdom", + key_cape_audience="qonto-assistant", + key_cape_required=False, + key_cape_cache_seconds=300, + key_cape_timeout_seconds=5, credential_source="env", openbao_path="tenants/binky/qonto-api", openbao_command="bao", diff --git a/tests/test_key_cape_auth.py b/tests/test_key_cape_auth.py new file mode 100644 index 0000000..161bf06 --- /dev/null +++ b/tests/test_key_cape_auth.py @@ -0,0 +1,256 @@ +import json +import time +from datetime import UTC, datetime, timedelta + +import httpx +import jwt +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa +from jwt.algorithms import RSAAlgorithm + +from qonto_assistant.auth import actor_claims_from_headers +from qonto_assistant.config import Settings +from qonto_assistant.key_cape_auth import KeyCapeAuthError, KeyCapeTokenVerifier + +ISSUER = "https://key-cape.netkingdom.test" +AUDIENCE = "qonto-assistant" +KID = "test-key-1" + + +def _keypair() -> tuple[rsa.RSAPrivateKey, dict]: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + jwk = json.loads(RSAAlgorithm.to_jwk(private_key.public_key())) + jwk["kid"] = KID + jwk["use"] = "sig" + jwk["alg"] = "RS256" + return private_key, jwk + + +def _jwks_transport(jwk: dict, *, calls: list[int] | None = None) -> httpx.MockTransport: + def handler(request: httpx.Request) -> httpx.Response: + if calls is not None: + calls.append(1) + return httpx.Response(200, json={"keys": [jwk]}) + + return httpx.MockTransport(handler) + + +def _token(private_key: rsa.RSAPrivateKey, *, kid: str = KID, **claim_overrides) -> str: + now = datetime.now(UTC) + claims = { + "iss": ISSUER, + "sub": "agent-harness-binky", + "aud": AUDIENCE, + "exp": int((now + timedelta(minutes=5)).timestamp()), + "iat": int(now.timestamp()), + "tenant": "tenant:friendly:binky", + "principal_type": "agent", + "groups": ["group:qonto-assistant-readers"], + "roles": ["blue"], + "scope": "finance.qonto.read", + } + claims.update(claim_overrides) + return jwt.encode(claims, private_key, algorithm="RS256", headers={"kid": kid}) + + +def _verifier(jwk: dict, *, required: bool = False, calls: list[int] | None = None) -> KeyCapeTokenVerifier: + return KeyCapeTokenVerifier( + jwks_url="https://key-cape.netkingdom.test/jwks", + issuer=ISSUER, + audience=AUDIENCE, + required=required, + transport=_jwks_transport(jwk, calls=calls), + ) + + +def test_verify_accepts_valid_token() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + + claims = verifier.verify(_token(private_key)) + + assert claims.actor_id == "agent-harness-binky" + assert claims.tenant_id == "tenant:friendly:binky" + assert claims.lane == "blue" + assert "finance.qonto.read" in claims.scopes + + +def test_verify_rejects_expired_token() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + expired = _token(private_key, exp=int(time.time()) - 60) + + with pytest.raises(KeyCapeAuthError): + verifier.verify(expired) + + +def test_verify_rejects_wrong_audience() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + wrong_aud = _token(private_key, aud="some-other-service") + + with pytest.raises(KeyCapeAuthError): + verifier.verify(wrong_aud) + + +def test_verify_rejects_wrong_issuer() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + wrong_iss = _token(private_key, iss="https://not-key-cape.test") + + with pytest.raises(KeyCapeAuthError): + verifier.verify(wrong_iss) + + +def test_verify_rejects_missing_required_claim() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + now = datetime.now(UTC) + token = jwt.encode( + { + "iss": ISSUER, + "sub": "agent-harness-binky", + "aud": AUDIENCE, + "exp": int((now + timedelta(minutes=5)).timestamp()), + "iat": int(now.timestamp()), + # tenant and principal_type deliberately omitted + }, + private_key, + algorithm="RS256", + headers={"kid": KID}, + ) + + with pytest.raises(KeyCapeAuthError): + verifier.verify(token) + + +def test_verify_rejects_unknown_signing_key() -> None: + private_key, jwk = _keypair() + other_key, _ = _keypair() + verifier = _verifier(jwk) + token_from_other_key = _token(other_key, kid="a-different-kid") + + with pytest.raises(KeyCapeAuthError): + verifier.verify(token_from_other_key) + + +def test_verify_falls_back_to_default_lane_when_roles_dont_match() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + token = _token(private_key, roles=["some-other-role"]) + + claims = verifier.verify(token) + + assert claims.lane == "green" + + +def test_jwks_cached_across_calls() -> None: + private_key, jwk = _keypair() + calls: list[int] = [] + verifier = _verifier(jwk, calls=calls) + + verifier.verify(_token(private_key)) + verifier.verify(_token(private_key)) + + assert len(calls) == 1 + + +def test_jwks_refetched_on_unknown_kid_before_failing() -> None: + private_key, jwk = _keypair() + calls: list[int] = [] + verifier = _verifier(jwk, calls=calls) + verifier.verify(_token(private_key)) # primes the cache with one fetch + + with pytest.raises(KeyCapeAuthError): + verifier.verify(_token(private_key, kid="rotated-kid-not-in-jwks")) + + # one fetch to prime the cache, one forced refresh attempt on the miss + assert len(calls) == 2 + + +# --- auth.py integration --------------------------------------------------------------- + + +def _settings() -> Settings: + return Settings( + service_name="qonto-assistant", + default_tenant_id="binky", + default_actor_lane="green", + required_scope="finance.qonto.read", + enforce_scope=False, + policy_file=None, # unused by these tests + qonto_base_url="https://example.test", + qonto_fixture_dir=None, + qonto_auth_mode="legacy_api_key", + qonto_organization_path="/v2/organization", + qonto_transactions_path="/v2/transactions", + qonto_timeout_seconds=1, + qonto_max_retries=0, + qonto_secret_ttl_seconds=60, + rate_limit_requests=20, + rate_limit_window_seconds=60, + max_concurrency=4, + deny_escalation_enabled=True, + deny_escalation_threshold=3, + deny_escalation_window_seconds=60, + deny_escalation_lockout_seconds=300, + key_cape_jwks_url=None, + key_cape_issuer=ISSUER, + key_cape_audience=AUDIENCE, + key_cape_required=False, + key_cape_cache_seconds=300, + key_cape_timeout_seconds=5, + credential_source="env", + openbao_path="tenants/binky/qonto-api", + openbao_command="bao", + openbao_timeout_seconds=5, + mcp_auth_token=None, + host="127.0.0.1", + port=8080, + ) + + +def test_bearer_token_takes_precedence_over_self_asserted_headers() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk) + token = _token(private_key) + headers = { + "authorization": f"Bearer {token}", + "x-actor-id": "someone-self-asserted", + "x-tenant-id": "not-binky", + } + + claims = actor_claims_from_headers(headers, _settings(), key_cape_verifier=verifier) + + assert claims.actor_id == "agent-harness-binky" + assert claims.tenant_id == "tenant:friendly:binky" + + +def test_required_verifier_rejects_missing_bearer_token() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk, required=True) + + with pytest.raises(KeyCapeAuthError): + actor_claims_from_headers({"x-actor-id": "someone"}, _settings(), key_cape_verifier=verifier) + + +def test_optional_verifier_falls_back_to_self_asserted_headers_when_absent() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk, required=False) + + claims = actor_claims_from_headers( + {"x-actor-id": "someone", "x-tenant-id": "binky"}, _settings(), key_cape_verifier=verifier + ) + + assert claims.actor_id == "someone" + assert claims.tenant_id == "binky" + + +def test_invalid_bearer_token_is_rejected_even_when_not_required() -> None: + private_key, jwk = _keypair() + verifier = _verifier(jwk, required=False) + + with pytest.raises(KeyCapeAuthError): + actor_claims_from_headers( + {"authorization": "Bearer not-a-real-jwt"}, _settings(), key_cape_verifier=verifier + )