user-engine/tests/test_evidence.py

import unittest
from dataclasses import fields
from datetime import timedelta

from user_engine.adapters.local import InMemoryUserEngineStore, LocalAuthorizationCheckPort
from user_engine.domain import (
    AccessControlFact,
    AccessMembershipRequirement,
    AccessProfile,
    ActiveAccessContext,
    AccountStatus,
    AuthorizationDecision,
    AuthorizationEffect,
)
from user_engine.evidence import (
    HEARTBEAT_EVENT_TYPE,
    classify_audit,
    classify_outbox,
    load_bearing_counts,
)
from user_engine.service import AccessControlFactExport, AccessProfileSelection, UserEngineService
from user_engine.testing.fixtures import (
    FixtureIdentityClaimsAdapter,
    human_actor_claims,
    sample_application,
    sample_application_binding,
    sample_catalog,
)


_FORBIDDEN_DECISION_FIELDS = frozenset(
    {"effect", "allowed", "decision", "decision_id", "deny", "allow"}
)


class EvidenceClassificationTests(unittest.TestCase):
    def test_denials_and_revocations_are_load_bearing(self):
        self.assertEqual(classify_audit("authorization denied"), "load-bearing")
        self.assertEqual(
            classify_audit("authorization denied (stance fail_closed)"),
            "load-bearing",
        )
        self.assertEqual(classify_outbox("account.status_changed"), "load-bearing")
        self.assertEqual(classify_outbox("prepared_account.revoked"), "load-bearing")
        self.assertEqual(classify_outbox("user.created"), "attributive")
        self.assertEqual(classify_outbox(HEARTBEAT_EVENT_TYPE), "heartbeat")

    def test_heartbeat_counts_load_bearing_events_and_does_not_claim_completeness(self):
        service, _, _ = _service()
        session = service.me(_claims(), correlation_id="corr-me")
        service.set_account_status(
            session.actor,
            session.user.user_id,
            AccountStatus.DISABLED,
            correlation_id="corr-disable",
        )
        deny = LocalAuthorizationCheckPort(default_effect=AuthorizationEffect.DENY)
        denied = UserEngineService(
            store=service.store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=deny,
        )
        from user_engine.errors import AuthorizationDenied

        with self.assertRaises(AuthorizationDenied):
            denied.create_user(
                session.actor,
                display_name="x",
                primary_email=None,
                correlation_id="corr-denied",
            )

        event = service.record_evidence_heartbeat(correlation_id="corr-heartbeat")
        self.assertEqual(event.event_type, HEARTBEAT_EVENT_TYPE)
        self.assertFalse(event.payload["completeness_claimed"])
        self.assertEqual(event.payload["form"], "heartbeat")
        self.assertGreaterEqual(event.payload["load_bearing"]["account.status_changed"], 1)
        self.assertGreaterEqual(event.payload["load_bearing"]["authorization_denied"], 1)
        self.assertIn("were not altered or truncated after arrival", event.payload["bound"])
        self.assertNotIn("complete", event.payload["bound"])

        counts = load_bearing_counts(service.store.audit_log(), service.store.pending_outbox())
        self.assertEqual(counts["account.status_changed"], 1)

    def test_allow_without_lifetime_is_denied(self):
        from user_engine.errors import AuthorizationDenied

        class _StandingGrant:
            def check(self, request):
                return AuthorizationDecision(effect=AuthorizationEffect.ALLOW, reason="standing")

        store = InMemoryUserEngineStore()
        service = UserEngineService(
            store=store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=_StandingGrant(),
        )
        with self.assertRaisesRegex(AuthorizationDenied, "no lifetime"):
            service.me(_claims(), correlation_id="corr-standing")
        self.assertEqual(store.audit_log()[-1].summary, "authorization denied (allow has no lifetime)")
        self.assertIsNone(store.audit_log()[-1].decision_id)

    def test_expired_allow_is_denied(self):
        from user_engine.errors import AuthorizationDenied

        class _StaleAllow:
            def check(self, request):
                decision = AuthorizationDecision.for_standalone(
                    AuthorizationEffect.ALLOW, reason="stale"
                )
                return AuthorizationDecision(
                    effect=decision.effect,
                    reason=decision.reason,
                    binding=decision.binding,
                    lifetime=timedelta(seconds=1),
                    issued_at=decision.issued_at - timedelta(seconds=5),
                )

        store = InMemoryUserEngineStore()
        service = UserEngineService(
            store=store,
            identity_adapter=FixtureIdentityClaimsAdapter(),
            authorization=_StaleAllow(),
        )
        with self.assertRaisesRegex(AuthorizationDenied, "expired"):
            service.me(_claims(), correlation_id="corr-stale")


class AccessControlFactsAreClaimsTests(unittest.TestCase):
    def test_claim_shapes_have_no_authorization_effect(self):
        for model in (AccessControlFact, AccessProfile, ActiveAccessContext):
            names = {item.name for item in fields(model)}
            overlap = names & _FORBIDDEN_DECISION_FIELDS
            self.assertFalse(
                overlap,
                f"{model.__name__} carries decision fields {overlap}",
            )

    def test_hat_selection_and_export_never_return_allow_or_deny(self):
        service, _, _ = _service()
        session = _bootstrap(service)
        service.add_membership(
            session.actor,
            session.user.user_id,
            tenant="tenant:coulomb",
            scope_type="realm",
            scope_id="realm:citadel",
            kind="operator",
            correlation_id="corr-realm-membership",
        )
        profile = service.register_access_profile(
            session.actor,
            AccessProfile(
                tenant="tenant:coulomb",
                display_name="Operator",
                hat="operator",
                realm_id="realm:citadel",
                membership_requirements=(
                    AccessMembershipRequirement(
                        scope_type="realm",
                        scope_id="realm:citadel",
                        kind="operator",
                    ),
                ),
            ),
            correlation_id="corr-profile-register",
        )
        selection = service.select_active_hat(
            session.actor,
            session.user.user_id,
            profile.access_profile_id,
            correlation_id="corr-select-hat",
        )
        export = service.export_access_control_facts(
            session.actor,
            tenant="tenant:coulomb",
            user_id=session.user.user_id,
            correlation_id="corr-export-facts",
        )
        self.assertIsInstance(selection, AccessProfileSelection)
        self.assertIsInstance(export, AccessControlFactExport)
        self._assert_no_decision_payload(selection)
        self._assert_no_decision_payload(export)
        for fact in export.facts:
            self._assert_no_decision_payload(fact)

    def _assert_no_decision_payload(self, value) -> None:
        blob = repr(value).lower()
        self.assertNotIn("authorizationeffect", blob)
        self.assertNotIn("effect=allow", blob)
        self.assertNotIn("effect=deny", blob)
        if hasattr(value, "__dict__") or hasattr(value, "__dataclass_fields__"):
            names = {item.name for item in fields(type(value))}
            self.assertFalse(names & _FORBIDDEN_DECISION_FIELDS)


def _service():
    store = InMemoryUserEngineStore()
    service = UserEngineService(
        store=store,
        identity_adapter=FixtureIdentityClaimsAdapter(),
        authorization=LocalAuthorizationCheckPort(),
    )
    return service, store, None


def _bootstrap(service: UserEngineService):
    session = service.me(_claims(), correlation_id="corr-me")
    service.register_application(
        session.actor,
        sample_application(),
        binding=sample_application_binding(),
        correlation_id="corr-app",
    )
    service.publish_catalog(
        session.actor,
        sample_catalog(),
        correlation_id="corr-catalog",
    )
    return session


def _claims():
    claims = human_actor_claims(subject="ada", tenant="tenant:coulomb")
    claims["roles"] = ["tenant-admin"]
    return claims


if __name__ == "__main__":
    unittest.main()