All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 37s
Engine/PIP declaration is now checkable (layer.yaml plus a Tooling-client scan). Writes persist a decision record or the published fail-closed stance, live-lookup freshness is published, events_for is tenant-scoped, and mutation evidence drains to audit-core from a local outbox without blocking the mutation. Sender registration is requested as AUDIT-IN-0002. Boundary-contract amendment is requested as NET-IN-0002. Assistant: grok Assistant-Session: 01a04cea-e5e8-7081-a0fc-808ebbc35fa9
48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
from tenant_engine.authz import (
|
|
AuthorizationOutcome,
|
|
WriteAuthorizationDeniedError,
|
|
)
|
|
|
|
|
|
class AllowAllAuthorizer:
|
|
"""Test double: allow every action and leave a reconstructable decision record."""
|
|
|
|
def authorize(self, *, action: str, tenant_id: str, actor: str) -> AuthorizationOutcome:
|
|
return AuthorizationOutcome(
|
|
action=action,
|
|
tenant_id=tenant_id,
|
|
actor=actor,
|
|
allowed=True,
|
|
source="decision",
|
|
reason="test_allow_all",
|
|
decision_id="test:allow",
|
|
request_digest="test",
|
|
effect="allow",
|
|
)
|
|
|
|
|
|
def deny(
|
|
action: str, tenant_id: str, actor: str, reason: str = "not permitted"
|
|
) -> WriteAuthorizationDeniedError:
|
|
return WriteAuthorizationDeniedError(
|
|
AuthorizationOutcome(
|
|
action=action,
|
|
tenant_id=tenant_id,
|
|
actor=actor,
|
|
allowed=False,
|
|
source="decision",
|
|
reason=reason,
|
|
decision_id="test:deny",
|
|
effect="deny",
|
|
)
|
|
)
|
|
|
|
|
|
class ScopedAuthorizer:
|
|
def __init__(self, *allowed: str) -> None:
|
|
self._allowed = set(allowed)
|
|
|
|
def authorize(self, *, action: str, tenant_id: str, actor: str) -> AuthorizationOutcome:
|
|
if action not in self._allowed:
|
|
raise deny(action, tenant_id, actor)
|
|
return AllowAllAuthorizer().authorize(action=action, tenant_id=tenant_id, actor=actor)
|