"""Typed identity models with no raw token or project-binding claims.""" from __future__ import annotations from datetime import UTC, datetime from enum import StrEnum from typing import Any, Literal from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, field_validator class ManagementRole(StrEnum): ADMIN = "admin" OPERATOR = "operator" AUDITOR = "auditor" VIEWER = "viewer" class PrincipalKind(StrEnum): MANAGEMENT_USER = "management_user" MACHINE_CLIENT = "machine_client" class ValidatedClaims(BaseModel): """The small validated projection used by authorization mapping.""" model_config = ConfigDict(extra="ignore", frozen=True) subject: str = Field(min_length=1, max_length=512, alias="sub") issuer: str = Field(min_length=1, max_length=2048, alias="iss") audience: tuple[str, ...] = Field(alias="aud") expires_at: datetime = Field(alias="exp") issued_at: datetime = Field(alias="iat") not_before: datetime | None = Field(default=None, alias="nbf") client_subject: str | None = Field(default=None, alias="azp") client_id: str | None = None nonce: str | None = None groups: tuple[str, ...] = () roles: tuple[str, ...] = () @field_validator("expires_at", "issued_at", "not_before", mode="before") @classmethod def timezone_aware(cls, value: Any) -> Any: if isinstance(value, int | float): return datetime.fromtimestamp(value, tz=UTC) if isinstance(value, datetime) and value.tzinfo is None: raise ValueError("JWT timestamps must be timezone-aware") return value @field_validator("audience", mode="before") @classmethod def normalize_audience(cls, value: Any) -> tuple[str, ...]: if isinstance(value, str): return (value,) if isinstance(value, list | tuple) and all(isinstance(item, str) for item in value): return tuple(value) raise ValueError("invalid JWT audience") @field_validator("groups", "roles", mode="before") @classmethod def normalize_strings(cls, value: Any) -> tuple[str, ...]: if value is None: return () if isinstance(value, str): return (value,) if isinstance(value, list | tuple) and all(isinstance(item, str) for item in value): return tuple(value) raise ValueError("invalid identity group or role claim") class AuthenticatedPrincipal(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) subject: str = Field(min_length=1, max_length=512) issuer: str = Field(min_length=1, max_length=2048) audience: tuple[str, ...] kind: PrincipalKind client_id: UUID | None = None machine_subject: str | None = None groups: tuple[str, ...] = () roles: frozenset[ManagementRole] = frozenset() expires_at: datetime issued_at: datetime @field_validator("expires_at", "issued_at") @classmethod def require_aware_time(cls, value: datetime) -> datetime: if value.tzinfo is None or value.utcoffset() is None: raise ValueError("principal timestamps must be timezone-aware") return value class SessionMetadata(BaseModel): """Safe browser-visible session data; never contains bearer tokens.""" model_config = ConfigDict(extra="forbid", frozen=True) subject: str client_id: UUID | None kind: PrincipalKind roles: frozenset[ManagementRole] expires_at: datetime class TokenSet(BaseModel): """Internal token-exchange result. Never serialize this model to a browser.""" model_config = ConfigDict(extra="ignore") access_token: str = Field(min_length=1, max_length=65536) token_type: Literal["Bearer", "bearer"] = "Bearer" # noqa: S105 expires_in: int | None = Field(default=None, ge=1, le=86400) refresh_token: str | None = Field(default=None, max_length=65536) id_token: str | None = Field(default=None, max_length=65536) class SessionCookie(BaseModel): """Internal result used to set an opaque session cookie and CSRF token.""" model_config = ConfigDict(extra="forbid", frozen=True) value: str csrf_token: str metadata: SessionMetadata set_cookie: str