from __future__ import annotations import re from typing import Any, Literal from uuid import RFC_4122, UUID from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr, model_validator FORGE_OIDC_ACTION = "forge.oidc.provision" FORGE_OIDC_TARGET = "itworx-forge" FORGE_OIDC_CLIENT_ID = "itworx-forge" FORGE_OIDC_APPLICATION_NAME = "ITWorx Forge" FORGE_OIDC_PROVIDER_NAME = "ITWorx Forge OIDC" FORGE_OIDC_LAUNCH_URL = "https://forge.example.com" FORGE_OIDC_CALLBACK_URL = "https://forge.example.com/auth/callback" FORGE_OIDC_ISSUER = "https://auth.example-host.com/application/o/itworx-forge/" FORGE_OIDC_DISCOVERY_URL = ( "https://auth.example-host.com/application/o/itworx-forge/.well-known/openid-configuration" ) FORGE_OIDC_SECRET_REF = "ITWORX_FORGE_OIDC_CLIENT_SECRET" FORGE_OIDC_BINDING_SECRET_REF = "AUTHENTIK_FORGE_OIDC_BINDING" FORGE_OIDC_GROUPS = ("forge-viewers", "forge-operators", "forge-owners") FORGE_OIDC_MANAGED_DESCRIPTION = "Managed by AppOps Forge OIDC v1." FORGE_OIDC_PROFILE_MAPPING_NAME = "ITWorx Forge profile and groups claim v1" FORGE_OIDC_PROFILE_MAPPING_EXPRESSION = '''allowed_groups = {"forge-viewers", "forge-operators", "forge-owners"} return delete_none_values({ "name": request.user.name, "preferred_username": request.user.username, "groups": sorted(group.name for group in request.user.ak_groups.all() if group.name in allowed_groups), })''' FORGE_OIDC_PROFILE_MAPPING_EXPRESSION_SHA256 = ( "648b3f6dbf2212940309a581b239c9996562ca017a89be22bcfa93f9aa2c26ae" ) FORGE_OIDC_OPENID_MAPPING_MANAGED = "goauthentik.io/providers/oauth2/scope-openid" FORGE_OIDC_EMAIL_MAPPING_MANAGED = "goauthentik.io/providers/oauth2/scope-email" FORGE_OIDC_OPENID_MAPPING_EXPRESSION_SHA256 = ( "38e58da331cc5378d4f269389242aecb80adf30093c405524dc1ab5aab9d452f" ) FORGE_OIDC_EMAIL_MAPPING_EXPRESSION_SHA256 = ( "d1200eabeffc3b7f379e9c1636367b69befb0b1ac2b5d82b7be20e78667f2b94" ) class ForgeOidcProvisionRequestV1(BaseModel): """Empty, strict request contract; the target and resource fields are fixed.""" model_config = ConfigDict(extra="forbid", frozen=True) class ForgeOidcScopeMappingIdsV1(BaseModel): """The only three scopes attached to the fixed Forge provider.""" model_config = ConfigDict(extra="forbid", frozen=True) openid: UUID email: UUID profile: UUID @model_validator(mode="before") @classmethod def mapping_ids_are_canonical_rfc_uuids(cls, value: Any) -> Any: if not isinstance(value, dict) or set(value) != {"openid", "email", "profile"}: raise ValueError("forge_oidc_scope_mapping_ids_invalid") for field in ("openid", "email", "profile"): raw = value[field] if not isinstance(raw, str): # Pydantic converts ValueError into structured ValidationError output. raise ValueError("forge_oidc_scope_mapping_id_invalid") # noqa: TRY004 try: parsed = UUID(raw) except (TypeError, ValueError, AttributeError): raise ValueError("forge_oidc_scope_mapping_id_invalid") from None if ( str(parsed) != raw or parsed.version not in {1, 2, 3, 4, 5, 6, 7, 8} or parsed.variant != RFC_4122 ): raise ValueError("forge_oidc_scope_mapping_id_invalid") return value @model_validator(mode="after") def mapping_ids_are_distinct(self) -> ForgeOidcScopeMappingIdsV1: values = (self.openid, self.email, self.profile) if len(set(values)) != len(values): raise ValueError("forge_oidc_scope_mapping_ids_must_be_distinct") return self class ForgeOidcBindingV1(BaseModel): """Protected, encrypted reference to the remote resources owned by Forge OIDC.""" model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True) version: StrictInt provider_id: StrictStr | None = Field(default=None, alias="providerId") scope_mapping_ids: ForgeOidcScopeMappingIdsV1 = Field(alias="scopeMappingIds") @model_validator(mode="after") def provider_id_is_positive_decimal(self) -> ForgeOidcBindingV1: if self.version != 1: raise ValueError("forge_oidc_binding_version_unsupported") if self.provider_id is not None and not self.provider_id.isascii(): raise ValueError("forge_oidc_provider_id_must_be_positive_decimal") if self.provider_id is not None and ( not re.fullmatch(r"[1-9][0-9]{0,9}", self.provider_id) ): raise ValueError("forge_oidc_provider_id_must_be_positive_decimal") return self class ForgeOidcProvisionResultV1(BaseModel): """Safe result projection. It intentionally has no client-secret field.""" model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True) success: bool status: Literal["created", "already_configured", "blocked", "conflict", "failed", "partial_failure"] message: str client_id: Literal["itworx-forge"] = Field(default=FORGE_OIDC_CLIENT_ID, alias="clientId") callback_url: Literal["https://forge.example.com/auth/callback"] = Field( default=FORGE_OIDC_CALLBACK_URL, alias="callbackUrl", ) issuer: Literal["https://auth.example-host.com/application/o/itworx-forge/"] = FORGE_OIDC_ISSUER discovery_url: Literal[ "https://auth.example-host.com/application/o/itworx-forge/.well-known/openid-configuration" ] = Field(default=FORGE_OIDC_DISCOVERY_URL, alias="discoveryUrl") groups: tuple[Literal["forge-viewers", "forge-operators", "forge-owners"], ...] = FORGE_OIDC_GROUPS secret_ref: Literal["ITWORX_FORGE_OIDC_CLIENT_SECRET"] = Field( default=FORGE_OIDC_SECRET_REF, alias="secretRef", ) secret_created: bool = Field(default=False, alias="secretCreated") application_id: str | None = Field(default=None, alias="applicationId") provider_id: str | None = Field(default=None, alias="providerId")