Components / policy

policy

Authorization port for permission checks

v0.1.0 stable queries events hooks

Policy is a platform primitive that abstracts authorization. Components inject a Policy port to check if actors can perform actions on resources. Implementations include AllowAllPolicy (dev/single-user), OwnerOnlyPolicy (user owns their data), and can be extended for roles, groups, or custom authorization logic.

0
Ports
0
Schemas
3
Hooks
3
Events
01

Ports

Required

No required ports

Optional
  • EventBus optional
    Publish permission check events for auditing
  • ActivityLogRepo optional
    Audit permission checks for compliance
Adapters Provided
  • AllowAllPolicy
    implements Policy
    Permit all actions (development/single-user)
  • OwnerOnlyPolicy
    implements Policy
    Users can only access their own resources
02

Schemas

Defines

No schemas defined

Uses

No external schemas used

03

Hooks

  • before_check
    Before evaluating a permission check
    outputs: Patch
  • on_grant
    After permission is granted
    outputs: SideEffectRequest
  • on_deny
    After permission is denied
    outputs: SideEffectRequest
04

Events

  • PermissionGranted v1
    After a permission check passes
    payload: {actor_id, action, owner_id, resource_type}
  • PermissionDenied v1
    After a permission check fails
    payload: {actor_id, action, owner_id, resource_type, reason}
  • PermissionRequired v1
    When require() raises PermissionDeniedError
    payload: {actor_id, action, owner_id, resource_type}
05

Examples

Check permission in a use case 
class CompleteTaskUseCase:
    def __init__(self, repo: TaskRepository, policy: Policy) -> None:
        self._repo = repo
        self._policy = policy

    async def execute(self, task_id: UUID, actor_id: str) -> None:
        task = await self._repo.get(task_id)

        if not self._policy.can(actor_id, Action.COMPLETE, task.owner_id, "Task"):
            raise NotAuthorizedError("Cannot complete this task")

        task.complete()
        await self._repo.update(task)

Use policy.can() to check before performing actions.

Use require() for cleaner code 
async def execute(self, task_id: UUID, actor_id: str) -> None:
    task = await self._repo.get(task_id)

    # Raises PermissionDeniedError if not allowed
    self._policy.require(actor_id, Action.DELETE, task.owner_id, "Task")

    await self._repo.delete(task_id)

Raises PermissionDeniedError automatically.

Test with AllowAllPolicy 
from psp.platform.policy import AllowAllPolicy

def test_complete_task():
    policy = AllowAllPolicy()  # Allows everything
    use_case = CompleteTaskUseCase(repo=InMemoryRepo(), policy=policy)

    # No authorization error
    use_case.execute(task_id=task.id, actor_id="any-user")

Bypass authorization in tests.

OwnerOnlyPolicy for user isolation 
from psp.platform.policy import OwnerOnlyPolicy, Action

policy = OwnerOnlyPolicy()

# Owner can access
policy.can("user-123", Action.READ, owner_id="user-123", "Task")  # True

# Others cannot
policy.can("user-456", Action.READ, owner_id="user-123", "Task")  # False

# System can access everything
policy.can("system", Action.READ, owner_id="user-123", "Task")  # True

Users can only access their own resources.

API Reference

This component mounts routes under /v1/policy. View OpenAPI specification