127 lines
4.1 KiB
Python
127 lines
4.1 KiB
Python
"""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
|