tegwick
5d57c7d488
flex_auth.py: CheckRequest + FlexAuthCheckClient against flex-auth's real POST /v1/check contract (schemas/check_request.schema.json, decision_envelope.schema.json, read directly from the flex-auth repo, not guessed). Fail-closed by construction: only effect=="allow" authorizes; every other effect, non-200, malformed body, or transport failure resolves to deny, nothing raises past is_allowed(). authz.FlexAuthWriteAuthorizer implements the existing WriteAuthorizer Protocol. Action -> resource-type mapping coordinated with FLEX-WP-0008's planned vocabulary (both repos reference the same table). DefaultDenyWriteAuthorizer stays the fallback when no flex-auth URL is configured. config.py: Settings.from_env(), mirroring qonto-assistant's pattern. docs/flex-auth-integration.md documents the contract, fail-closed rule, and current real state (denies everything until FLEX-WP-0008 lands). 60 tests passing. Verified live twice over real HTTP between separate processes (not just MockTransport): a deny-returning flex-auth double produces 403 from POST /tenants, an allow-returning one produces 201. Also registered (not implemented) the two workplans this depends on for a complete picture: flex-auth/FLEX-WP-0008 (protected-system registration -- what makes allow reachable) and key-cape/KEY-WP-0005 (discovered key-cape emits none of iam-profile_v0.3.md's core claims yet, not just missing tenant_roles -- a bigger, security-sensitive gap flagged rather than quietly worked around). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from tenant_engine.app import create_app
|
|
from tenant_engine.authz import (
|
|
DefaultDenyWriteAuthorizer,
|
|
FlexAuthWriteAuthorizer,
|
|
WriteAuthorizationDeniedError,
|
|
)
|
|
from tenant_engine.config import Settings
|
|
from tenant_engine.flex_auth import FlexAuthCheckClient
|
|
|
|
|
|
def _settings(*, flex_auth_url: str | None) -> Settings:
|
|
return Settings(
|
|
flex_auth_base_url=flex_auth_url,
|
|
flex_auth_timeout_seconds=1,
|
|
host="127.0.0.1",
|
|
port=8090,
|
|
)
|
|
|
|
|
|
def test_create_app_defaults_to_default_deny_without_flex_auth_url() -> None:
|
|
app = create_app(settings=_settings(flex_auth_url=None))
|
|
assert isinstance(app.state.authorizer, DefaultDenyWriteAuthorizer)
|
|
|
|
|
|
def test_create_app_uses_flex_auth_authorizer_when_url_configured() -> None:
|
|
app = create_app(settings=_settings(flex_auth_url="https://flex-auth.example.test"))
|
|
assert isinstance(app.state.authorizer, FlexAuthWriteAuthorizer)
|
|
|
|
|
|
def test_flex_auth_authorizer_denies_on_deny_effect() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json={"id": "d-1", "effect": "deny", "resource": {}, "subject": {}, "provenance": {}})
|
|
|
|
client = FlexAuthCheckClient(
|
|
base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
|
|
)
|
|
authorizer = FlexAuthWriteAuthorizer(client=client)
|
|
|
|
with pytest.raises(WriteAuthorizationDeniedError):
|
|
authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="ops")
|
|
|
|
|
|
def test_flex_auth_authorizer_denies_on_not_applicable_effect() -> None:
|
|
"""The realistic state until FLEX-WP-0008's policy package exists."""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
200, json={"id": "d-1", "effect": "not_applicable", "resource": {}, "subject": {}, "provenance": {}}
|
|
)
|
|
|
|
client = FlexAuthCheckClient(
|
|
base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
|
|
)
|
|
authorizer = FlexAuthWriteAuthorizer(client=client)
|
|
|
|
with pytest.raises(WriteAuthorizationDeniedError):
|
|
authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="ops")
|
|
|
|
|
|
def test_flex_auth_authorizer_allows_on_allow_effect() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}})
|
|
|
|
client = FlexAuthCheckClient(
|
|
base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
|
|
)
|
|
authorizer = FlexAuthWriteAuthorizer(client=client)
|
|
|
|
authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="ops") # does not raise
|
|
|
|
|
|
def test_full_write_lifecycle_succeeds_when_flex_auth_allows() -> None:
|
|
"""End-to-end: create_app() wired to a flex-auth double that allows
|
|
|
|
everything -- proves the seam actually gates through create_app's own
|
|
authorizer selection, not just when constructed directly.
|
|
"""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json={"id": "d-1", "effect": "allow", "resource": {}, "subject": {}, "provenance": {}})
|
|
|
|
client = FlexAuthCheckClient(
|
|
base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
|
|
)
|
|
app = create_app(authorizer=FlexAuthWriteAuthorizer(client=client))
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
test_client = TestClient(app)
|
|
created = test_client.post(
|
|
"/tenants", json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"}
|
|
)
|
|
assert created.status_code == 201
|