Authentication Patterns

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: ...

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,
        )

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

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

@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()

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())

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()