49 lines
1.4 KiB
Python
49 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)
|