tenant-engine/tests/test_pep_write_path.py

"""TEN-WP-0011-T02: decision records, stance, no verdict cache."""

from __future__ import annotations

import httpx
from fastapi.testclient import TestClient

from tenant_engine.app import create_app
from tenant_engine.authz import FlexAuthWriteAuthorizer
from tenant_engine.config import Settings
from tenant_engine.domain import Tenant
from tenant_engine.flex_auth import FlexAuthCheckClient
from tenant_engine.stance import FAIL_CLOSED
from tenant_engine.store import InMemoryTenantStore


def _allowing_client(handler) -> FlexAuthCheckClient:
    return FlexAuthCheckClient(
        base_url="https://flex-auth.example.test", transport=httpx.MockTransport(handler)
    )


def test_granted_role_persists_decision_id_on_the_event():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "id": "decision:grant-1",
                "effect": "allow",
                "resource": {},
                "subject": {},
                "provenance": {},
            },
        )

    store = InMemoryTenantStore()
    app = create_app(
        store=store,
        authorizer=FlexAuthWriteAuthorizer(client=_allowing_client(handler)),
    )
    client = TestClient(app)
    assert (
        client.post(
            "/tenants",
            json={
                "tenant_id": "t-1",
                "identifier": "tenant:friendly:binky",
                "actor": "tenant-engine",
            },
        ).status_code
        == 201
    )
    granted = client.post(
        "/tenants/t-1/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "CUS",
            "grant_reason": "manual_grant",
            "granted_by": "ops",
            "correlation_id": "c-1",
            "actor": "tenant-engine",
        },
    )
    assert granted.status_code == 201
    event = [e for e in store.events_for("t-1") if e.event_type == "role_granted"][-1]
    assert event.payload["authorization_decision_id"] == "decision:grant-1"
    assert event.payload["authorization_source"] == "decision"
    records = store.authorization_records("t-1")
    assert any(r.decision_id == "decision:grant-1" and r.allowed for r in records)


def test_denied_grant_leaves_a_reconstructable_record():
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "id": "decision:deny-1",
                "effect": "deny",
                "resource": {},
                "subject": {},
                "provenance": {},
            },
        )

    store = InMemoryTenantStore()
    store.create_tenant(Tenant.create(tenant_id="t-1", identifier="tenant:friendly:binky"))
    app = create_app(
        store=store,
        authorizer=FlexAuthWriteAuthorizer(client=_allowing_client(handler)),
    )
    response = TestClient(app).post(
        "/tenants/t-1/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "CUS",
            "grant_reason": "manual_grant",
            "granted_by": "ops",
            "correlation_id": "c-1",
            "actor": "tenant-engine",
        },
    )
    assert response.status_code == 403
    records = store.authorization_records("t-1")
    assert records
    assert records[-1].allowed is False
    assert records[-1].decision_id == "decision:deny-1"
    assert records[-1].source == "decision"
    assert store.events_for("t-1")  # tenant_created only; no role_granted
    assert not any(e.event_type == "role_granted" for e in store.events_for("t-1"))


def test_unreachable_engine_records_fail_closed_stance():
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused", request=request)

    store = InMemoryTenantStore()
    app = create_app(
        store=store,
        authorizer=FlexAuthWriteAuthorizer(client=_allowing_client(handler)),
        settings=Settings(
            flex_auth_base_url="https://flex-auth.example.test",
            flex_auth_timeout_seconds=1,
            host="127.0.0.1",
            port=8090,
        ),
    )
    response = TestClient(app).post(
        "/tenants",
        json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "tenant-engine"},
    )
    assert response.status_code == 403
    records = store.authorization_records("t-1")
    assert records[-1].allowed is False
    assert records[-1].source == "stance"
    assert records[-1].stance == FAIL_CLOSED


def test_unset_authorizer_records_fail_closed_stance():
    store = InMemoryTenantStore()
    app = create_app(store=store)  # DefaultDeny
    response = TestClient(app).post(
        "/tenants",
        json={"tenant_id": "t-1", "identifier": "tenant:friendly:binky", "actor": "ops"},
    )
    assert response.status_code == 403
    records = store.authorization_records("t-1")
    assert records[-1].source == "stance"
    assert records[-1].stance == FAIL_CLOSED


def test_verdict_is_not_cached_across_requests():
    calls = {"n": 0}

    def handler(request: httpx.Request) -> httpx.Response:
        calls["n"] += 1
        return httpx.Response(
            200,
            json={
                "id": f"d-{calls['n']}",
                "effect": "allow",
                "resource": {},
                "subject": {},
                "provenance": {},
            },
        )

    authorizer = FlexAuthWriteAuthorizer(client=_allowing_client(handler))
    authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="tenant-engine")
    authorizer.authorize(action="tenant.create", tenant_id="t-1", actor="tenant-engine")
    assert calls["n"] == 2


def test_a_previous_allow_cannot_authorize_a_different_request():
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        import json

        body = json.loads(request.content)
        seen.append(body["action"])
        effect = "allow" if body["action"] == "tenant.create" else "deny"
        return httpx.Response(
            200,
            json={
                "id": f"d-{body['action']}",
                "effect": effect,
                "resource": {},
                "subject": {},
                "provenance": {},
            },
        )

    store = InMemoryTenantStore()
    app = create_app(
        store=store,
        authorizer=FlexAuthWriteAuthorizer(client=_allowing_client(handler)),
    )
    client = TestClient(app)
    assert (
        client.post(
            "/tenants",
            json={
                "tenant_id": "t-1",
                "identifier": "tenant:friendly:binky",
                "actor": "tenant-engine",
            },
        ).status_code
        == 201
    )
    denied = client.post(
        "/tenants/t-1/roles/grant",
        json={
            "grant_id": "g-1",
            "role": "CUS",
            "grant_reason": "manual_grant",
            "granted_by": "ops",
            "correlation_id": "c-1",
            "actor": "tenant-engine",
        },
    )
    assert denied.status_code == 403
    assert seen == ["tenant.create", "tenant.role.grant"]