104 lines
4.0 KiB
Python
104 lines
4.0 KiB
Python
"""Root-owned generic enrollment template, separate from Forge and user SSO."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Literal
|
|
from uuid import UUID
|
|
from urllib.parse import urlsplit
|
|
|
|
from pydantic import BaseModel, ConfigDict, field_validator, model_validator
|
|
from sqlmodel import Session
|
|
|
|
from ..connectors.authentik import AuthentikConnector
|
|
from ..db.models import AppSetting
|
|
from ..schemas.enrollment import EnrollmentTrustConfigV1
|
|
from ..core.config import settings
|
|
|
|
OIDC_TEMPLATE_SETTING_KEY = 'deployment_enrollment:oidc_template'
|
|
TEMPLATE_FIELDS = {
|
|
'authorization_flow': 'AUTHENTIK_OIDC_AUTHORIZATION_FLOW',
|
|
'invalidation_flow': 'AUTHENTIK_OIDC_INVALIDATION_FLOW',
|
|
'openid_mapping': 'AUTHENTIK_OIDC_OPENID_CLAIM_MAPPING',
|
|
'email_mapping': 'AUTHENTIK_OIDC_EMAIL_CLAIM_MAPPING',
|
|
'profile_mapping': 'AUTHENTIK_OIDC_PROFILE_CLAIM_MAPPING',
|
|
'roles_mapping': 'AUTHENTIK_OIDC_GROUP_CLAIM_MAPPING',
|
|
'signing_key': 'AUTHENTIK_OIDC_SIGNING_KEY_ID',
|
|
}
|
|
|
|
|
|
def trusted_issuer_origin() -> str | None:
|
|
"""Use AppOps' established native Authentik SSO trust; never caller URLs."""
|
|
try:
|
|
value = urlsplit(settings.authentik_issuer or '')
|
|
except ValueError:
|
|
return None
|
|
if value.scheme != 'https' or not value.netloc or value.username or value.password:
|
|
return None
|
|
return f'https://{value.netloc.lower()}'
|
|
|
|
|
|
class EnrollmentOidcProfileV1(BaseModel):
|
|
model_config = ConfigDict(extra='forbid', frozen=True)
|
|
|
|
issuer: str
|
|
authorization_flow: UUID
|
|
invalidation_flow: UUID
|
|
openid_mapping: UUID
|
|
email_mapping: UUID
|
|
profile_mapping: UUID
|
|
roles_mapping: UUID
|
|
signing_key: UUID
|
|
administrator_group: Literal['Administrator'] = 'Administrator'
|
|
support_agent_group: Literal['SupportAgent'] = 'SupportAgent'
|
|
read_only_group: Literal['ReadOnly'] = 'ReadOnly'
|
|
|
|
@field_validator('issuer')
|
|
@classmethod
|
|
def issuer_origin(cls, value: str) -> str:
|
|
if not re.fullmatch(r'https://[a-z0-9]+(?:[.-][a-z0-9]+)*(?:/)?', value):
|
|
raise ValueError('An HTTPS Authentik origin without credentials or application path is required')
|
|
return value.rstrip('/')
|
|
|
|
@model_validator(mode='after')
|
|
def separate_mappings(self):
|
|
if len({self.openid_mapping, self.email_mapping, self.profile_mapping, self.roles_mapping}) != 4:
|
|
raise ValueError('Four distinct scope mappings are required')
|
|
return self
|
|
|
|
def template_config(self) -> dict[str, str]:
|
|
return {**{env: str(getattr(self, field)) for field, env in TEMPLATE_FIELDS.items()}, 'AUTHENTIK_OIDC_ROLE_CLAIM': 'roles'}
|
|
|
|
|
|
def load_profile(session: Session) -> EnrollmentOidcProfileV1 | None:
|
|
row = session.get(AppSetting, OIDC_TEMPLATE_SETTING_KEY)
|
|
if row is None or row.is_secret:
|
|
return None
|
|
try:
|
|
return EnrollmentOidcProfileV1.model_validate(row.value_json)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def bind_template(session: Session, connector: AuthentikConnector) -> None:
|
|
"""A corrupt stored binding blocks fallback; never mutate global/Forge IDs."""
|
|
row = session.get(AppSetting, OIDC_TEMPLATE_SETTING_KEY)
|
|
if row is None:
|
|
connector.configure_deployment_oidc_template(None)
|
|
return
|
|
profile = load_profile(session)
|
|
trust_row = session.get(AppSetting, 'deployment_enrollment:trust')
|
|
try:
|
|
trust = EnrollmentTrustConfigV1.model_validate(trust_row.value_json) if trust_row and not trust_row.is_secret else None
|
|
except ValueError:
|
|
trust = None
|
|
if not trust or not profile or profile.issuer != trusted_issuer_origin() or (
|
|
trust.oidc_issuer != profile.issuer
|
|
or trust.administrator_group != profile.administrator_group
|
|
or trust.support_agent_group != profile.support_agent_group
|
|
or trust.read_only_group != profile.read_only_group
|
|
):
|
|
profile = None
|
|
connector.configure_deployment_oidc_template(
|
|
profile.template_config() if profile else {}, issuer_origin=profile.issuer if profile else None,
|
|
)
|