API Reference¶

@configured(target="self", prefix="auth_client", mapping="tree")
@dataclass
class AuthClientSettings:
    """Type-safe settings for the auth client, loaded from configuration sources.

    Populated automatically from configuration sources using the ``auth_client``
    prefix via pico-ioc's ``@configured`` decorator.

    Attributes:
        enabled: Whether authentication middleware is active.
        issuer: Expected JWT issuer (``iss`` claim).
        audience: Expected JWT audience (``aud`` claim).
        jwks_ttl_seconds: How long to cache the JWKS key set (seconds).
        jwks_endpoint: URL to fetch JWKS from. Defaults to ``{issuer}/api/v1/auth/jwks``.
    """

    enabled: bool = True
    issuer: str = ""
    audience: str = ""
    jwks_ttl_seconds: int = 300
    jwks_endpoint: str = ""
    accepted_algorithms: tuple[str, ...] = ("RS256",)

class AuthClientError(Exception):
    """Base error for all pico-client-auth exceptions."""

class AuthConfigurationError(AuthClientError):
    """Authentication configuration is missing or invalid."""

class InsufficientPermissionsError(AuthClientError):
    """The authenticated user lacks the required role(s)."""

class MissingTokenError(AuthClientError):
    """No Bearer token found in the Authorization header."""

class TokenExpiredError(AuthClientError):
    """The JWT token has expired."""

class TokenInvalidError(AuthClientError):
    """The JWT token is malformed or has an invalid signature."""

@dataclass(frozen=True)
class TokenClaims:
    """Immutable representation of the essential JWT claims.

    Attributes:
        sub: Subject identifier (user ID).
        email: User email address.
        role: Primary role claim from the token.
        org_id: Organisation identifier.
        jti: Unique token identifier (JWT ID).
    """

    sub: str
    email: str
    role: str
    org_id: str
    jti: str
    groups: tuple[str, ...] = ()

@runtime_checkable
class RoleResolver(Protocol):
    """Protocol for resolving user roles from JWT claims.

    Implement this protocol to customise how roles are extracted
    (e.g. from a ``roles`` array claim, from an external service, etc.).
    Register your implementation as a ``@component`` and it will
    automatically replace the default resolver.
    """

    async def resolve(self, claims: TokenClaims, raw_claims: dict) -> list[str]: ...

class SecurityContext:
    """Singleton-style accessor for the current request's authentication state.

    All methods are static; the underlying storage uses ``ContextVar`` so
    each async task / thread has its own isolated copy.
    """

    @staticmethod
    def get() -> Optional[TokenClaims]:
        """Return the current claims, or ``None`` if not authenticated."""
        return _claims_var.get()

    @staticmethod
    def require() -> TokenClaims:
        """Return the current claims, raising if not authenticated.

        Raises:
            MissingTokenError: If no authentication context is set.
        """
        claims = _claims_var.get()
        if claims is None:
            raise MissingTokenError("No authenticated user in SecurityContext")
        return claims

    @staticmethod
    def get_roles() -> list[str]:
        """Return the resolved roles for the current request."""
        return list(_roles_var.get())

    @staticmethod
    def has_role(role: str) -> bool:
        """Check whether the current user has the given role."""
        return role in _roles_var.get()

    @staticmethod
    def require_role(*roles: str) -> None:
        """Assert that the current user has at least one of the given roles.

        Raises:
            InsufficientPermissionsError: If none of the roles match.
        """
        current = set(_roles_var.get())
        if not current.intersection(roles):
            raise InsufficientPermissionsError(f"Required one of {roles}, but user has {sorted(current)}")

    @staticmethod
    def get_groups() -> tuple[str, ...]:
        """Return the group IDs for the current request."""
        return _groups_var.get()

    @staticmethod
    def has_group(group_id: str) -> bool:
        """Check whether the current user belongs to the given group."""
        return group_id in _groups_var.get()

    @staticmethod
    def require_group(*group_ids: str) -> None:
        """Assert that the current user belongs to at least one of the given groups.

        Raises:
            InsufficientPermissionsError: If none of the groups match.
        """
        current = set(_groups_var.get())
        if not current.intersection(group_ids):
            raise InsufficientPermissionsError(f"Required one of groups {group_ids}, but user has {sorted(current)}")

    @staticmethod
    def set(claims: TokenClaims, roles: list[str]) -> None:
        """Populate the security context (called by the middleware)."""
        _claims_var.set(claims)
        _roles_var.set(roles)
        _groups_var.set(claims.groups)

    @staticmethod
    def clear() -> None:
        """Clear the security context (called by the middleware in ``finally``)."""
        _claims_var.set(None)
        _roles_var.set([])
        _groups_var.set(())