QONTO-WP-0004-T04: live flex-auth + tenant-engine authorization gate

Replaces the config-only QONTO_ASSISTANT_ENFORCE_SCOPE cached-claim
check with two live-checked facts, per docs/SecurityPractice.md #4:

1. flex-auth POST /v1/check on finance.qonto.read for the calling
   actor/tenant (FlexAuthCheckClient, modeled on tenant-engine's own
   client for the same API). Registration lives in the flex-auth repo
   (examples/qonto-assistant/) -- rules + embedded tests verified with
   flex-auth test-policy/load-registry/check, and a live flex-auth
   serve hit by this exact client over real HTTP (not a mock).
2. tenant-engine's live capability-role lookup
   (GET /tenants/{id}/roles/live), denying unless the tenant currently
   holds one of QONTO_TENANT_ENGINE_REQUIRED_ROLES (default VEN,CUS) --
   optional and additive to the flex-auth check.

Both clients fail closed by construction (unreachable/malformed/non-2xx
all deny, never grant), matching FlexAuthCheckClient's existing
fail-closed philosophy elsewhere in the fleet. LiveAuthorizationGate
combines both and is wired into CapabilityService._execute ahead of
the internal policy kernel; off by default (no QONTO_FLEX_AUTH_URL
set) so existing deployments are unaffected until configured.

Verified beyond mocked unit tests: ran a real `flex-auth serve` loaded
with the registered policy, and a real tenant-engine instance seeded
with a VEN grant for tenant:friendly:binky, and exercised this repo's
actual FlexAuthCheckClient/TenantEngineClient/LiveAuthorizationGate
against both live processes over real HTTP -- allow for the correct
tenant, live_authz_denied for a mismatched tenant.

28 new unit tests (flex_auth_client, tenant_engine_client,
live_authorization_gate + CapabilityService integration). Full suite
-> 80 passed; REST/MCP smokes and compileall still clean.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

tests/test_api.py

@@ -44,6 +44,11 @@ def _settings() -> Settings:
         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",

tests/test_flex_auth_client.py

@@ -0,0 +1,92 @@
+import json
+
+import httpx
+import pytest
+
+from qonto_assistant.flex_auth_client import CheckRequest, FlexAuthCheckClient, new_request_id
+
+
+def _request() -> CheckRequest:
+    return CheckRequest(
+        request_id=new_request_id(),
+        tenant="tenant:friendly:binky",
+        subject_id="agent-harness-binky",
+        subject_type="agent",
+    )
+
+
+def _client(handler) -> FlexAuthCheckClient:
+    return FlexAuthCheckClient(
+        base_url="https://flex-auth.example.test",
+        timeout_seconds=1,
+        transport=httpx.MockTransport(handler),
+    )
+
+
+def test_allow_effect_authorizes() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}})
+
+    assert _client(handler).is_allowed(_request()) is True
+
+
+@pytest.mark.parametrize("effect", ["deny", "redact", "audit_only", "not_applicable"])
+def test_non_allow_effects_deny(effect: str) -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}})
+
+    assert _client(handler).is_allowed(_request()) is False
+
+
+def test_non_200_status_denies() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(500, json={"error": "internal"})
+
+    assert _client(handler).is_allowed(_request()) is False
+
+
+def test_malformed_json_body_denies() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, content=b"not json")
+
+    assert _client(handler).is_allowed(_request()) is False
+
+
+def test_non_object_json_body_denies() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=["not", "an", "object"])
+
+    assert _client(handler).is_allowed(_request()) is False
+
+
+def test_connection_failure_denies() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("connection refused", request=request)
+
+    assert _client(handler).is_allowed(_request()) is False
+
+
+def test_timeout_denies() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.TimeoutException("timed out", request=request)
+
+    assert _client(handler).is_allowed(_request()) is False
+
+
+def test_request_body_matches_schema_shape() -> None:
+    seen: dict[str, object] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.update(json.loads(request.content))
+        return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}})
+
+    _client(handler).is_allowed(_request())
+
+    assert seen["tenant"] == "tenant:friendly:binky"
+    assert seen["action"] == "finance.qonto.read"
+    assert seen["subject"] == {"id": "agent-harness-binky", "type": "agent"}
+    assert seen["resource"] == {
+        "id": "finance-snapshot",
+        "type": "finance-snapshot",
+        "system": "qonto-assistant",
+    }

tests/test_key_cape_auth.py

@@ -200,6 +200,11 @@ def _settings() -> Settings:
         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",

tests/test_live_authorization_gate.py

@@ -0,0 +1,162 @@
+from pathlib import Path
+
+import httpx
+import pytest
+
+from qonto_assistant.audit import AuditLogger
+from qonto_assistant.contracts import ActorClaims
+from qonto_assistant.errors import PolicyDeniedError
+from qonto_assistant.flex_auth_client import FlexAuthCheckClient
+from qonto_assistant.live_authorization import DENY_LIVE_AUTHZ, DENY_TENANT_ROLE, LiveAuthorizationGate
+from qonto_assistant.policy import PolicyEngine
+from qonto_assistant.qonto_client import FixtureQontoClient
+from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter
+from qonto_assistant.service import CapabilityService
+from qonto_assistant.tenant_engine_client import TenantEngineClient
+
+POLICY_FILE = Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml"
+FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "qonto"
+
+
+def _claims(tenant_id: str = "tenant:friendly:binky") -> ActorClaims:
+    return ActorClaims(actor_id="agent-harness-binky", tenant_id=tenant_id, lane="green")
+
+
+def _flex_auth_client(effect: str) -> FlexAuthCheckClient:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"id": "d-1", "effect": effect, "resource": {}, "subject": {}, "provenance": {}})
+
+    return FlexAuthCheckClient(base_url="https://flex-auth.test", transport=httpx.MockTransport(handler))
+
+
+def _tenant_engine_client(roles: list[str]) -> TenantEngineClient:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": roles})
+
+    return TenantEngineClient(base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler))
+
+
+# --- Gate unit tests -------------------------------------------------------------------
+
+
+def test_gate_allows_when_flex_auth_allows_and_role_matches() -> None:
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("allow"),
+        tenant_engine_client=_tenant_engine_client(["VEN"]),
+        required_tenant_roles=frozenset({"VEN", "CUS"}),
+    )
+
+    assert gate.check(_claims()) is None
+
+
+def test_gate_denies_when_flex_auth_denies() -> None:
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("deny"),
+        tenant_engine_client=_tenant_engine_client(["VEN"]),
+        required_tenant_roles=frozenset({"VEN", "CUS"}),
+    )
+
+    assert gate.check(_claims()) == DENY_LIVE_AUTHZ
+
+
+def test_gate_denies_when_tenant_lacks_required_role() -> None:
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("allow"),
+        tenant_engine_client=_tenant_engine_client(["PLTF"]),  # not VEN/CUS
+        required_tenant_roles=frozenset({"VEN", "CUS"}),
+    )
+
+    assert gate.check(_claims()) == DENY_TENANT_ROLE
+
+
+def test_gate_skips_tenant_role_check_when_no_roles_required() -> None:
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("allow"),
+        tenant_engine_client=_tenant_engine_client([]),
+        required_tenant_roles=frozenset(),
+    )
+
+    assert gate.check(_claims()) is None
+
+
+def test_gate_skips_tenant_role_check_when_no_tenant_engine_configured() -> None:
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("allow"),
+        tenant_engine_client=None,
+        required_tenant_roles=frozenset({"VEN"}),
+    )
+
+    assert gate.check(_claims()) is None
+
+
+def test_gate_denies_when_tenant_engine_unreachable() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("connection refused", request=request)
+
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("allow"),
+        tenant_engine_client=TenantEngineClient(base_url="https://tenant-engine.test", transport=httpx.MockTransport(handler)),
+        required_tenant_roles=frozenset({"VEN"}),
+    )
+
+    assert gate.check(_claims()) == DENY_TENANT_ROLE
+
+
+# --- CapabilityService integration -----------------------------------------------------
+
+
+def _service(gate: LiveAuthorizationGate, events: list[dict[str, object]]) -> CapabilityService:
+    return CapabilityService(
+        client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
+        policy=PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False),
+        audit_logger=AuditLogger(sink=events.append),
+        rate_limiter=RateLimiter(limit=20, window_seconds=60),
+        concurrency_limiter=ConcurrencyLimiter(limit=4),
+        live_authorization_gate=gate,
+    )
+
+
+def test_service_denies_capability_call_when_gate_denies() -> None:
+    events: list[dict[str, object]] = []
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("deny"),
+        tenant_engine_client=None,
+        required_tenant_roles=frozenset(),
+    )
+    service = _service(gate, events)
+
+    with pytest.raises(PolicyDeniedError) as exc_info:
+        service.get_accounts(claims=_claims(), request_id="req-1", protocol="rest")
+
+    assert exc_info.value.decision.reason == DENY_LIVE_AUTHZ
+    deny_events = [e for e in events if e.get("deny_reason") == DENY_LIVE_AUTHZ]
+    assert len(deny_events) == 1
+
+
+def test_service_allows_capability_call_when_gate_allows() -> None:
+    events: list[dict[str, object]] = []
+    gate = LiveAuthorizationGate(
+        flex_auth_client=_flex_auth_client("allow"),
+        tenant_engine_client=_tenant_engine_client(["VEN"]),
+        required_tenant_roles=frozenset({"VEN", "CUS"}),
+    )
+    service = _service(gate, events)
+
+    payload = service.get_accounts(claims=_claims(), request_id="req-2", protocol="rest")
+
+    assert "organization" in payload
+
+
+def test_service_without_gate_configured_behaves_as_before() -> None:
+    events: list[dict[str, object]] = []
+    service = CapabilityService(
+        client=FixtureQontoClient(fixture_dir=FIXTURE_DIR),
+        policy=PolicyEngine.from_file(POLICY_FILE, required_scope="finance.qonto.read", enforce_scope=False),
+        audit_logger=AuditLogger(sink=events.append),
+        rate_limiter=RateLimiter(limit=20, window_seconds=60),
+        concurrency_limiter=ConcurrencyLimiter(limit=4),
+    )
+
+    payload = service.get_accounts(claims=_claims(), request_id="req-3", protocol="rest")
+
+    assert "organization" in payload

tests/test_tenant_engine_client.py

@@ -0,0 +1,70 @@
+import httpx
+
+from qonto_assistant.tenant_engine_client import TenantEngineClient
+
+
+def _client(handler) -> TenantEngineClient:
+    return TenantEngineClient(
+        base_url="https://tenant-engine.example.test",
+        timeout_seconds=1,
+        transport=httpx.MockTransport(handler),
+    )
+
+
+def test_active_roles_returns_roles_on_200() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.url.path == "/tenants/tenant:friendly:binky/roles/live"
+        return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": ["VEN", "CUS"]})
+
+    roles = _client(handler).active_roles("tenant:friendly:binky")
+
+    assert roles == frozenset({"VEN", "CUS"})
+
+
+def test_active_roles_empty_when_no_roles() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": []})
+
+    assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
+
+
+def test_active_roles_fails_closed_on_404_tenant_not_found() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(404, json={"detail": "tenant_not_found"})
+
+    assert _client(handler).active_roles("tenant:friendly:nobody") == frozenset()
+
+
+def test_active_roles_fails_closed_on_503_store_unavailable() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(503, json={"detail": "tenant_roles_unavailable"})
+
+    assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
+
+
+def test_active_roles_fails_closed_on_malformed_body() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, content=b"not json")
+
+    assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
+
+
+def test_active_roles_fails_closed_on_non_list_roles_field() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"tenant_id": "tenant:friendly:binky", "roles": "VEN"})
+
+    assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
+
+
+def test_active_roles_fails_closed_on_connection_error() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("connection refused", request=request)
+
+    assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()
+
+
+def test_active_roles_fails_closed_on_timeout() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.TimeoutException("timed out", request=request)
+
+    assert _client(handler).active_roles("tenant:friendly:binky") == frozenset()