Context as Code
Documentation

Authorization Strategies

Composite authorization pattern — role-based, payment-based, and webhook strategies chained together.

Show:

Status: Extracted
Purpose: Composite authorization pattern, authorization strategies, and authorization service implementation.

Composite Authorization Pattern

Overview

Pattern: Multiple authorization strategies work together

Benefits:

  • Flexible authorization logic
  • Easy to add new strategies
  • Clear separation of concerns
  • Testable individual strategies

Implementation:

  • CompositeAuthorizationService chains multiple strategies
  • Each strategy implements AuthorizationService interface
  • AND Logic: All strategies must pass (if any raises PermissionError, authorization fails)
  • Extensible: New authorization strategies can be added easily

Authorization Strategies

Role-Based Authorization

Strategy: RoleBasedAuthorizationService

Purpose: Checks roles (admin, etc.)

Pattern:

class RoleBasedAuthorizationService(AuthorizationService):
    """Role-based authorization strategy."""
    
    async def check(self, action: str, resource: str, actor: ActorContext) -> None:
        """Check role-based permissions."""
        # Check if actor has required role
        if "admin" in actor.roles:
            return  # Admin has all permissions
        # ... other role checks

Payment-Based Authorization

Strategy: PaymentBasedAuthorizationService

Purpose: Checks payment status and tier (post-P0)

Pattern:

class PaymentBasedAuthorizationService(AuthorizationService):
    """Payment-based authorization strategy."""
    
    async def check(self, action: str, resource: str, actor: ActorContext) -> None:
        """Check payment-based permissions."""
        if actor.payment_status != "active":
            raise PermissionError("Payment required")
        # ... payment tier checks

Webhook Authorization

Strategy: WebhookAuthorizationService

Purpose: Checks system/webhook actors

Pattern:

class WebhookAuthorizationService(AuthorizationService):
    """Webhook authorization strategy."""
    
    async def check(self, action: str, resource: str, actor: ActorContext) -> None:
        """Check webhook permissions."""
        if actor.actor_type == "system":
            return  # System actors have permissions
        # ... webhook-specific checks

Composite Authorization Service

Implementation

Pattern:

class CompositeAuthorizationService(AuthorizationService):
    """Runs multiple authorization strategies in order."""
    
    def __init__(self, strategies: list[AuthorizationService]) -> None:
        self._strategies = strategies
    
    async def check(self, action: str, resource: str, actor: ActorContext) -> None:
        """Check authorization using all strategies."""
        for strategy in self._strategies:
            await strategy.check(action, resource, actor)
        return None

Usage:

strategies = [
    RoleBasedAuthorizationService(),
    PaymentBasedAuthorizationService(),
    WebhookAuthorizationService(),
]
authorization_service = CompositeAuthorizationService(strategies)

Authorization Service Interface

Port Definition

Location: application/security/authorization.py

Pattern:

class AuthorizationService(ABC):
    """Authorization port."""
    
    @abstractmethod
    async def check(self, action: str, resource: str, actor: ActorContext) -> None:
        """
        Check if actor is authorized to perform action on resource.
        
        Raises:
            PermissionError if not authorized
        """
        raise NotImplementedError

Actor Context

Actor Context Structure

Pattern:

@dataclass(frozen=True)
class ActorContext:
    """Represents the actor context for authorization checks."""
    
    actor_id: Optional[str] = None       # user_id or machine_id
    actor_type: str = "user"             # "user" | "system" | "service"
    account_ids: tuple[str, ...] = ()    # accounts the actor is associated with
    roles: tuple[str, ...] = ()          # roles (e.g., admin)
    payment_status: Optional[str] = None # optional, for payment-based authz
    payment_tier: Optional[str] = None   # optional, for payment-based authz

Related Documents

  • Authentication Patterns - Authentication approaches
  • Permission Models - Permission checks
  • Security Practices - Security best practices

Reference implementation: infrastructure/security/authorization_composite.py

Framework implementations

Abstract reference: 15-security/

Authentication is handled in JWTAuthMiddleware. Authorisation is handled via Depends() in route definitions. Application and domain layers are auth-unaware — they receive an ActorContext and apply business rules.

JWT Service Interface (application/authentication/jwt_service.py)

from abc import ABC, abstractmethod
from uuid import UUID
 
 
class JWTService(ABC):
    @abstractmethod
    def generate_token(self, account_id: UUID, account_type: str, roles: list[str]) -> str: ...
 
    @abstractmethod
    def refresh_token(self, token: str) -> str: ...
 
    @abstractmethod
    def validate_token(self, token: str) -> bool: ...
 
    @abstractmethod
    def extract_claims(self, token: str) -> dict: ...

JWT Implementation (infrastructure/authentication/jwt_token.py)

import jwt
from datetime import datetime, timedelta, timezone
from uuid import UUID
from application.authentication.jwt_service import JWTService
 
 
class PyJWTService(JWTService):
    def __init__(self, secret: str, algorithm: str = "HS256",
                 access_expire_minutes: int = 1440,
                 issuer: str = "",
                 audience: str = ""):
        self._secret = secret
        self._algorithm = algorithm
        self._access_expire_minutes = access_expire_minutes
        self._issuer = issuer      # e.g. "https://api.your-service.com"
        self._audience = audience  # e.g. "your-service-api"
 
    def generate_token(self, account_id: UUID, account_type: str, roles: list[str]) -> str:
        now = datetime.now(timezone.utc)
        payload = {
            # Standard OAuth 2.0 / OIDC claims
            "iss": self._issuer,                                          # Issuer — required
            "aud": self._audience,                                        # Audience — required
            "sub": str(account_id),                                       # Subject (principal)
            "iat": now,                                                   # Issued at
            "exp": now + timedelta(minutes=self._access_expire_minutes),  # Expiry
            # Domain-specific claims
            "account_type": account_type,
            "roles": roles,
        }
        return jwt.encode(payload, self._secret, algorithm=self._algorithm)
 
    def validate_token(self, token: str) -> bool:
        try:
            jwt.decode(
                token,
                self._secret,
                algorithms=[self._algorithm],
                audience=self._audience,   # Validates aud claim
                issuer=self._issuer,       # Validates iss claim
            )
            return True
        except jwt.PyJWTError:
            return False
 
    def extract_claims(self, token: str) -> dict:
        return jwt.decode(
            token,
            self._secret,
            algorithms=[self._algorithm],
            audience=self._audience,
            issuer=self._issuer,
        )

Actor Context (application/security/authorization.py)

The actor context is the auth-safe representation of the authenticated user, available to application handlers:

from dataclasses import dataclass
 
 
@dataclass(frozen=True)
class ActorContext:
    actor_id: str | None
    actor_type: str  # "user" | "admin" | "system" | "anonymous"
    roles: tuple[str, ...]
    account_ids: tuple[str, ...]  # IDs this actor can access
 
    @property
    def is_authenticated(self) -> bool:
        return self.actor_id is not None
 
    @property
    def is_admin(self) -> bool:
        return "admin" in self.roles
 
    def can_access_account(self, account_id: str) -> bool:
        return account_id in self.account_ids or self.is_admin

The actor context is populated by JWTAuthMiddleware into request.state.ctx["actor"]. See 07-request-and-actor-context.md for the full propagation flow.

Route-Level Authorisation (api/security.py)

from fastapi import Depends, Request, HTTPException
from application.security.authorization import ActorContext
 
 
def get_actor_context(request: Request) -> ActorContext:
    """Extracts actor from request context. Use in routes that need the current user."""
    actor_data = getattr(request.state, "ctx", {}).get("actor", {})
    return ActorContext(
        actor_id=actor_data.get("actor_id"),
        actor_type=actor_data.get("actor_type", "anonymous"),
        roles=tuple(actor_data.get("roles", [])),
        account_ids=tuple(actor_data.get("account_ids", [])),
    )
 
 
def require_authenticated(actor: ActorContext = Depends(get_actor_context)) -> ActorContext:
    if not actor.is_authenticated:
        raise HTTPException(status_code=401, detail="Authentication required")
    return actor
 
 
def require_admin(actor: ActorContext = Depends(get_actor_context)) -> ActorContext:
    if not actor.is_admin:
        raise HTTPException(status_code=403, detail="Admin access required")
    return actor
 
 
def require_account_access(account_id: str):
    """Factory for route-specific account access check."""
    def dependency(actor: ActorContext = Depends(require_authenticated)) -> ActorContext:
        if not actor.can_access_account(account_id):
            raise HTTPException(status_code=403, detail="Access denied")
        return actor
    return dependency

Usage in routes:

@router.get("/{account_id}")
async def get_account(
    account_id: UUID,
    actor: ActorContext = Depends(require_authenticated),
    handler=Depends(get_account_handler),
):
    if not actor.can_access_account(str(account_id)):
        raise HTTPException(status_code=403)
    return await handler.execute(str(account_id))
 
@router.get("/admin/all")
async def list_all_accounts(
    actor: ActorContext = Depends(require_admin),
    handler=Depends(get_list_accounts_handler),
):
    return await handler.execute()

Password Hashing (infrastructure/authentication/password_bcrypt.py)

import bcrypt
from application.authentication.password_service import PasswordService
 
 
class BcryptPasswordService(PasswordService):
    def hash(self, plain: str) -> str:
        return bcrypt.hashpw(plain.encode(), bcrypt.gensalt()).decode()
 
    def verify(self, plain: str, hashed: str) -> bool:
        return bcrypt.checkpw(plain.encode(), hashed.encode())

Auth Factories (infrastructure/authentication/factories.py)

from config.settings import Settings
from infrastructure.authentication.jwt_token import PyJWTService
from infrastructure.authentication.password_bcrypt import BcryptPasswordService
 
def build_jwt_service(settings: Settings) -> PyJWTService:
    return PyJWTService(
        secret=settings.auth_jwt_secret,
        algorithm=settings.auth_jwt_algorithm,
        access_expire_minutes=settings.auth_jwt_access_token_expire_minutes,
        issuer=settings.auth_jwt_issuer,    # e.g. "https://api.your-service.com"
        audience=settings.auth_jwt_audience, # e.g. "your-service-api"
    )
 
def build_password_service(settings: Settings) -> BcryptPasswordService:
    return BcryptPasswordService()

Rules

  • JWT validation happens only in middleware, never in route handlers or application handlers.
  • Route handlers receive an ActorContext — they do not decode tokens or read auth headers.
  • Application handlers apply business-rule-based authorisation (e.g. "can this actor modify this account?") using ActorContext. They do not raise HTTP exceptions — they raise domain exceptions (NotAuthorizedError), which exception handlers map to 403 responses.
  • Public routes (health, docs, login, signup) are listed in JWTAuthMiddleware.PUBLIC_PATHS — middleware skips validation for these.
Authentication PatternsConfiguration & Settings