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, flex_auth_base_url=None, flex_auth_timeout_seconds=3, tenant_engine_base_url=None, tenant_engine_timeout_seconds=3, tenant_engine_required_roles=frozenset({"VEN", "CUS"}), 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 )