from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass, replace from datetime import timedelta from fastapi import Depends, Header, HTTPException, Request, status from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import ( CAPABILITY_ACCESS_PERMISSION_EVALUATOR, CAPABILITY_ACCESS_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, PermissionEvaluator, PrincipalRef, PrincipalResolver, ) from govoplan_core.core.automation import ( AutomationPrincipalRequest, AutomationPrincipalResolution, ) from govoplan_core.core.identity import IdentityDirectory from govoplan_core.core.idm import IdmDirectory, OrganizationFunctionAssignmentRef from govoplan_core.core.organizations import OrganizationDirectory from govoplan_core.core.principal_cache import ( auth_principal_revision, ) from govoplan_core.core.modules import AccessDecision from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, maintenance_response_detail, saved_maintenance_mode from govoplan_core.db.session import get_database, get_session from govoplan_access.backend.db.models import ( Account, ApiKey, AuthSession, Role, ServiceAccount, Tenant, User, ) from govoplan_access.backend.semantic import collect_external_function_roles, identity_id_for_account from govoplan_access.backend.auth.principal_cache import principal_summary_cache from govoplan_access.backend.auth.tokens import hash_secret from govoplan_access.backend.security.api_keys import authenticate_api_key from govoplan_access.backend.security.sessions import ( authenticate_session_token, collect_user_authorization_context, UserAuthorizationContext, verify_auth_session_csrf, ) from govoplan_core.security.module_permissions import scopes_grant_compatible from govoplan_core.security.time import ensure_aware_utc, utc_now from govoplan_access.backend.permissions.catalog import intersect_api_key_scopes from govoplan_core.settings import settings @dataclass(slots=True) class ResolvedPrincipalContext: principal: PrincipalRef account: Account user: User tenant: Tenant api_key: ApiKey | None = None auth_session: AuthSession | None = None def _extract_token(request: Request, authorization: str | None, x_api_key: str | None) -> tuple[str | None, str]: if x_api_key: return x_api_key.strip(), "api_key" if authorization and authorization.lower().startswith("bearer "): return authorization[7:].strip(), "bearer" cookie_token = request.cookies.get(settings.auth_session_cookie_name) if cookie_token: return cookie_token.strip(), "cookie" return None, "none" def _requires_csrf(request: Request) -> bool: return request.method.upper() not in {"GET", "HEAD", "OPTIONS", "TRACE"} def _build_principal_ref( session: Session, *, account: Account, user: User, tenant_id: str, scopes: list[str], auth_method: str, api_key: ApiKey | None = None, auth_session: AuthSession | None = None, idm_assignments: tuple[OrganizationFunctionAssignmentRef, ...] = (), identity_directory: IdentityDirectory | None = None, extra_roles: tuple[Role, ...] = (), authorization_context: UserAuthorizationContext | None = None, include_system_roles: bool = False, ) -> PrincipalRef: if authorization_context is None: authorization_context = collect_user_authorization_context( session, user, account=account, include_system=include_system_roles, extra_roles=extra_roles, ) function_assignment_ids = list(authorization_context.function_assignment_ids) function_assignment_ids.extend(item.id for item in idm_assignments) role_ids = [role.id for role in authorization_context.tenant_roles] if include_system_roles: role_ids.extend(role.id for role in authorization_context.system_roles) return PrincipalRef( account_id=account.id, membership_id=user.id, tenant_id=tenant_id, identity_id=identity_id_for_account(session, account.id, identity_directory=identity_directory), scopes=frozenset(scopes), group_ids=frozenset(group.id for group in authorization_context.groups), role_ids=frozenset(sorted(dict.fromkeys(role_ids))), function_assignment_ids=frozenset(sorted(dict.fromkeys(function_assignment_ids))), delegation_ids=frozenset(authorization_context.function_delegation_ids), auth_method=auth_method, # type: ignore[arg-type] api_key_id=api_key.id if api_key else None, session_id=auth_session.id if auth_session else None, email=account.email, display_name=account.display_name or user.display_name, ) def _api_principal_from_ref( session: Session, principal: PrincipalRef, *, permission_evaluator: PermissionEvaluator | None = None, ) -> ApiPrincipal: account = session.get(Account, principal.account_id) user = session.get(User, principal.membership_id) if principal.membership_id else None tenant = session.get(Tenant, principal.tenant_id) if principal.tenant_id else None if ( not account or not user or not tenant or not account.is_active or not user.is_active or not tenant.is_active or user.account_id != account.id or user.tenant_id != tenant.id ): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Inactive or inconsistent API principal") api_key = session.get(ApiKey, principal.api_key_id) if principal.api_key_id else None auth_session = session.get(AuthSession, principal.session_id) if principal.session_id else None return ApiPrincipal( principal=principal, account=account, user=user, api_key=api_key, auth_session=auth_session, permission_evaluator=permission_evaluator, ) def _api_principal_from_context( context: ResolvedPrincipalContext, *, permission_evaluator: PermissionEvaluator | None = None, ) -> ApiPrincipal: return ApiPrincipal( principal=context.principal, account=context.account, user=context.user, api_key=context.api_key, auth_session=context.auth_session, permission_evaluator=permission_evaluator, ) def _resolve_legacy_principal_ref( request: Request, session: Session, *, authorization: str | None, x_api_key: str | None, idm_directory: IdmDirectory | None = None, identity_directory: IdentityDirectory | None = None, organization_directory: OrganizationDirectory | None = None, ) -> PrincipalRef: return _resolve_legacy_principal_context( request, session, authorization=authorization, x_api_key=x_api_key, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ).principal def _resolve_legacy_principal_context( request: Request, session: Session, *, authorization: str | None, x_api_key: str | None, idm_directory: IdmDirectory | None = None, identity_directory: IdentityDirectory | None = None, organization_directory: OrganizationDirectory | None = None, ) -> ResolvedPrincipalContext: token, source = _extract_token(request, authorization, x_api_key) if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing API key or session token") cached = _cached_principal_context( request, session, token=token, source=source, ) if cached is not None: return cached if source != "cookie": context = _resolve_api_key_principal_context( session, token=token, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) if context is not None: return _cache_resolved_principal_context( session, token=token, context=context, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) context = _resolve_session_principal_context( request, session, token=token, source=source, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) if context is not None: return _cache_resolved_principal_context( session, token=token, context=context, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key or session token") def _cached_principal_context( request: Request, session: Session, *, token: str, source: str, ) -> ResolvedPrincipalContext | None: if not settings.auth_principal_cache_enabled: return None token_digest = hash_secret(token) entry = principal_summary_cache.get( token_digest, session_ttl_seconds=settings.auth_principal_cache_session_ttl_seconds, api_key_ttl_seconds=settings.auth_principal_cache_api_key_ttl_seconds, ) if entry is None: return None before = auth_principal_revision(session, tenant_id=entry.principal.tenant_id) if before != entry.revision: principal_summary_cache.discard(token_digest) return None context = _rehydrate_cached_principal( request, session, token_digest=token_digest, source=source, principal=entry.principal, ) after = auth_principal_revision(session, tenant_id=entry.principal.tenant_id) if context is None or after != before: principal_summary_cache.discard(token_digest) return None return context def _rehydrate_cached_principal( request: Request, session: Session, *, token_digest: str, source: str, principal: PrincipalRef, ) -> ResolvedPrincipalContext | None: account = session.get(Account, principal.account_id) user = session.get(User, principal.membership_id) if principal.membership_id else None tenant = session.get(Tenant, principal.tenant_id) if principal.tenant_id else None if ( not account or not user or not tenant or not account.is_active or not user.is_active or not tenant.is_active or user.account_id != account.id or user.tenant_id != tenant.id ): return None if principal.auth_method == "api_key": api_key = session.get(ApiKey, principal.api_key_id) if principal.api_key_id else None if ( api_key is None or api_key.key_hash != token_digest or api_key.revoked_at is not None or api_key.user_id != user.id or api_key.tenant_id != tenant.id or _is_expired(api_key.expires_at, allow_none=True) ): return None _touch_auth_activity( session, api_key, field="last_used_at", ) return ResolvedPrincipalContext( principal=principal, account=account, user=user, tenant=tenant, api_key=api_key, ) auth_session = session.get(AuthSession, principal.session_id) if principal.session_id else None if ( auth_session is None or auth_session.token_hash != token_digest or auth_session.revoked_at is not None or auth_session.user_id != user.id or auth_session.account_id != account.id or auth_session.tenant_id != tenant.id or _is_expired(auth_session.expires_at) ): return None if source == "cookie": _verify_session_csrf(request, auth_session) _touch_auth_activity( session, auth_session, field="last_seen_at", ) return ResolvedPrincipalContext( principal=principal, account=account, user=user, tenant=tenant, auth_session=auth_session, ) def _is_expired(value, *, allow_none: bool = False) -> bool: expires_at = ensure_aware_utc(value) return (not allow_none and expires_at is None) or ( expires_at is not None and expires_at < utc_now() ) def _touch_auth_activity( session: Session, model: ApiKey | AuthSession, *, field: str, ) -> None: now = utc_now() previous = ensure_aware_utc(getattr(model, field)) interval = settings.auth_activity_touch_interval_seconds if interval <= 0 or previous is None or now - previous >= timedelta(seconds=interval): setattr(model, field, now) session.add(model) session.commit() def _cache_resolved_principal_context( session: Session, *, token: str, context: ResolvedPrincipalContext, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> ResolvedPrincipalContext: if not settings.auth_principal_cache_enabled: return context before = auth_principal_revision(session, tenant_id=context.principal.tenant_id) refreshed = _refresh_principal_context( session, context=context, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) after = auth_principal_revision(session, tenant_id=context.principal.tenant_id) if before == after: principal_summary_cache.put( hash_secret(token), principal=refreshed.principal, revision=after, max_entries=settings.auth_principal_cache_max_entries, ) return refreshed def _refresh_principal_context( session: Session, *, context: ResolvedPrincipalContext, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> ResolvedPrincipalContext: idm_assignments, idm_roles = _principal_idm_context( session, user=context.user, account=context.account, tenant_id=context.tenant.id, idm_directory=idm_directory, organization_directory=organization_directory, ) include_system = context.auth_session is not None authorization_context = collect_user_authorization_context( session, context.user, account=context.account, include_system=include_system, extra_roles=idm_roles, ) scopes = authorization_context.scopes auth_method = "session" if context.api_key is not None: scopes = intersect_api_key_scopes(scopes, context.api_key.scopes or []) auth_method = "api_key" principal = _build_principal_ref( session, account=context.account, user=context.user, tenant_id=context.tenant.id, scopes=scopes, auth_method=auth_method, api_key=context.api_key, auth_session=context.auth_session, idm_assignments=idm_assignments, identity_directory=identity_directory, extra_roles=idm_roles, authorization_context=authorization_context, include_system_roles=include_system, ) return replace(context, principal=principal) def _resolve_api_key_principal_ref( session: Session, *, token: str, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> PrincipalRef | None: context = _resolve_api_key_principal_context( session, token=token, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) return context.principal if context is not None else None def _resolve_api_key_principal_context( session: Session, *, token: str, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> ResolvedPrincipalContext | None: # API keys remain supported for CLI/automation. Their permissions are the # intersection of the key grant and the owner's current tenant roles. api_key = authenticate_api_key( session, token, touch_interval_seconds=settings.auth_activity_touch_interval_seconds, ) if api_key is None: return None activity_touch_pending = session.is_modified( api_key, include_collections=False, ) user = api_key.user account = user.account if user else None tenant = session.get(Tenant, api_key.tenant_id) if ( not user or not account or not tenant or not user.is_active or not account.is_active or not tenant.is_active or user.tenant_id != api_key.tenant_id ): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Inactive or inconsistent API-key principal") idm_assignments, idm_roles = _principal_idm_context( session, user=user, account=account, tenant_id=api_key.tenant_id, idm_directory=idm_directory, organization_directory=organization_directory, ) authorization_context = collect_user_authorization_context( session, user, account=account, include_system=False, extra_roles=idm_roles, ) effective_scopes = intersect_api_key_scopes(authorization_context.scopes, api_key.scopes or []) principal = _build_principal_ref( session, api_key=api_key, account=account, user=user, tenant_id=api_key.tenant_id, scopes=effective_scopes, auth_method="api_key", idm_assignments=idm_assignments, identity_directory=identity_directory, extra_roles=idm_roles, authorization_context=authorization_context, ) if activity_touch_pending: session.commit() return ResolvedPrincipalContext(principal=principal, account=account, user=user, tenant=tenant, api_key=api_key) def _resolve_session_principal_ref( request: Request, session: Session, *, token: str, source: str, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> PrincipalRef | None: context = _resolve_session_principal_context( request, session, token=token, source=source, idm_directory=idm_directory, identity_directory=identity_directory, organization_directory=organization_directory, ) return context.principal if context is not None else None def _resolve_session_principal_context( request: Request, session: Session, *, token: str, source: str, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> ResolvedPrincipalContext | None: auth_session = authenticate_session_token( session, token, touch_interval_seconds=settings.auth_activity_touch_interval_seconds, ) if auth_session is None: return None activity_touch_pending = session.is_modified( auth_session, include_collections=False, ) user = auth_session.user account = auth_session.account tenant = session.get(Tenant, auth_session.tenant_id) if ( not user or not account or not tenant or not user.is_active or not account.is_active or not tenant.is_active or user.account_id != account.id or user.tenant_id != tenant.id ): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Inactive or inconsistent session principal") if source == "cookie": _verify_session_csrf(request, auth_session) idm_assignments, idm_roles = _principal_idm_context( session, user=user, account=account, tenant_id=user.tenant_id, idm_directory=idm_directory, organization_directory=organization_directory, ) authorization_context = collect_user_authorization_context( session, user, account=account, include_system=True, extra_roles=idm_roles, ) scopes = authorization_context.scopes principal = _build_principal_ref( session, auth_session=auth_session, account=account, user=user, tenant_id=user.tenant_id, scopes=scopes, auth_method="session", idm_assignments=idm_assignments, identity_directory=identity_directory, extra_roles=idm_roles, authorization_context=authorization_context, include_system_roles=True, ) if activity_touch_pending: session.commit() return ResolvedPrincipalContext(principal=principal, account=account, user=user, tenant=tenant, auth_session=auth_session) def _verify_session_csrf(request: Request, auth_session: AuthSession) -> None: if not _requires_csrf(request): return header_token = request.headers.get("x-csrf-token") cookie_token = request.cookies.get(settings.auth_csrf_cookie_name) if not header_token or not cookie_token or header_token != cookie_token or not verify_auth_session_csrf(auth_session, header_token): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid or missing CSRF token") def _principal_idm_context( session: Session, *, user: User, account: Account, tenant_id: str, idm_directory: IdmDirectory | None, organization_directory: OrganizationDirectory | None, ) -> tuple[tuple[OrganizationFunctionAssignmentRef, ...], tuple[Role, ...]]: idm_assignments = _idm_assignments_for_account(idm_directory, account.id, tenant_id=tenant_id) idm_roles = tuple(collect_external_function_roles(session, user, idm_assignments, organization_directory=organization_directory)) return idm_assignments, idm_roles def _idm_assignments_for_account( idm_directory: IdmDirectory | None, account_id: str, *, tenant_id: str, ) -> tuple[OrganizationFunctionAssignmentRef, ...]: if idm_directory is None: return () return tuple(idm_directory.organization_function_assignments_for_account(account_id, tenant_id=tenant_id)) def _registry_from_request(request: Request) -> PlatformRegistry | None: registry = getattr(request.app.state, "govoplan_registry", None) return registry if isinstance(registry, PlatformRegistry) else None def _principal_resolver_from_request(request: Request) -> PrincipalResolver | None: registry = _registry_from_request(request) if registry is None: return None capability_name = ( CAPABILITY_AUTH_PRINCIPAL_RESOLVER if registry.has_capability(CAPABILITY_AUTH_PRINCIPAL_RESOLVER) else CAPABILITY_ACCESS_PRINCIPAL_RESOLVER ) if not registry.has_capability(capability_name): return None capability = registry.require_capability(capability_name) if not isinstance(capability, PrincipalResolver): raise HTTPException(status_code=500, detail=f"Invalid capability: {capability_name}") return capability def _permission_evaluator_from_request(request: Request) -> PermissionEvaluator | None: registry = _registry_from_request(request) if registry is None: return None capability_name = ( CAPABILITY_AUTH_PERMISSION_EVALUATOR if registry.has_capability(CAPABILITY_AUTH_PERMISSION_EVALUATOR) else CAPABILITY_ACCESS_PERMISSION_EVALUATOR ) if not registry.has_capability(capability_name): return None capability = registry.require_capability(capability_name) if not isinstance(capability, PermissionEvaluator): raise HTTPException(status_code=500, detail=f"Invalid capability: {capability_name}") return capability class LegacyPrincipalResolver: def __init__( self, idm_directory: IdmDirectory | None = None, identity_directory: IdentityDirectory | None = None, organization_directory: OrganizationDirectory | None = None, ) -> None: self._idm_directory = idm_directory self._identity_directory = identity_directory self._organization_directory = organization_directory def resolve_request(self, request: object, *, session: object | None = None) -> PrincipalRef: if not isinstance(request, Request): raise TypeError("LegacyPrincipalResolver requires a FastAPI request") authorization = request.headers.get("authorization") x_api_key = request.headers.get("x-api-key") if isinstance(session, Session): return _resolve_legacy_principal_ref( request, session, authorization=authorization, x_api_key=x_api_key, idm_directory=self._idm_directory, identity_directory=self._identity_directory, organization_directory=self._organization_directory, ) with get_database().session() as managed_session: return _resolve_legacy_principal_ref( request, managed_session, authorization=authorization, x_api_key=x_api_key, idm_directory=self._idm_directory, identity_directory=self._identity_directory, organization_directory=self._organization_directory, ) class LegacyPermissionEvaluator: def scopes_grant(self, scopes, required_scope: str) -> bool: return scopes_grant_compatible(scopes, required_scope) def has_scope(self, principal: PrincipalRef, required_scope: str) -> bool: return self.scopes_grant(principal.scopes, required_scope) def explain_scope(self, principal: PrincipalRef, required_scope: str) -> AccessDecision: allowed = self.has_scope(principal, required_scope) return AccessDecision( allowed=allowed, reason=None if allowed else f"Missing scope: {required_scope}", requirements=(required_scope,), ) def resolve_api_principal( request: Request, session: Session, *, authorization: str | None = None, x_api_key: str | None = None, ) -> ApiPrincipal: resolver = _principal_resolver_from_request(request) permission_evaluator = _permission_evaluator_from_request(request) if isinstance(resolver, LegacyPrincipalResolver): context = _resolve_legacy_principal_context( request, session, authorization=authorization, x_api_key=x_api_key, idm_directory=resolver._idm_directory, identity_directory=resolver._identity_directory, organization_directory=resolver._organization_directory, ) api_principal = _api_principal_from_context(context, permission_evaluator=permission_evaluator) _enforce_maintenance_mode(session, api_principal) return api_principal if resolver is not None: principal = resolver.resolve_request(request, session=session) api_principal = _api_principal_from_ref(session, principal, permission_evaluator=permission_evaluator) _enforce_maintenance_mode(session, api_principal) return api_principal context = _resolve_legacy_principal_context( request, session, authorization=authorization, x_api_key=x_api_key, ) api_principal = _api_principal_from_context(context, permission_evaluator=permission_evaluator) _enforce_maintenance_mode(session, api_principal) return api_principal class AccessApiPrincipalProvider: def resolve_api_principal( self, request: object, session: object, *, authorization: str | None = None, x_api_key: str | None = None, ) -> ApiPrincipal: if not isinstance(request, Request): raise TypeError("AccessApiPrincipalProvider requires a FastAPI request") if not isinstance(session, Session): raise TypeError("AccessApiPrincipalProvider requires a SQLAlchemy session") return resolve_api_principal( request, session, authorization=authorization, x_api_key=x_api_key, ) class AccessAutomationPrincipalProvider: """Rebuild a trigger owner against current tenant authorization.""" def __init__( self, *, idm_directory: IdmDirectory | None = None, identity_directory: IdentityDirectory | None = None, organization_directory: OrganizationDirectory | None = None, ) -> None: self._idm_directory = idm_directory self._identity_directory = identity_directory self._organization_directory = organization_directory def resolve_automation_principal( self, session: object, *, request: AutomationPrincipalRequest, ) -> AutomationPrincipalResolution: if not isinstance(session, Session): raise TypeError( "Access automation principal resolution requires a " "SQLAlchemy session" ) if request.subject_kind == "service_account": return _resolve_service_account_automation( session, request=request, ) return _resolve_delegated_user_automation( session, request=request, idm_directory=self._idm_directory, identity_directory=self._identity_directory, organization_directory=self._organization_directory, ) def _resolve_delegated_user_automation( session: Session, *, request: AutomationPrincipalRequest, idm_directory: IdmDirectory | None, identity_directory: IdentityDirectory | None, organization_directory: OrganizationDirectory | None, ) -> AutomationPrincipalResolution: account = session.get(Account, request.account_id) user = session.get(User, request.membership_id) tenant = session.get(Tenant, request.tenant_id) if ( account is None or user is None or tenant is None or not account.is_active or not user.is_active or not tenant.is_active or user.account_id != account.id or user.tenant_id != tenant.id ): return _automation_denied( request, status="inactive_or_inconsistent", reason=( "The automation owner is inactive, missing, or no longer " "belongs to the tenant." ), ) idm_assignments, idm_roles = _principal_idm_context( session, user=user, account=account, tenant_id=request.tenant_id, idm_directory=idm_directory, organization_directory=organization_directory, ) authorization_context = collect_user_authorization_context( session, user, account=account, include_system=False, extra_roles=idm_roles, ) granted_scopes, missing_scopes = _current_trigger_grants( request.grant_scopes, current_scopes=authorization_context.scopes, ) if missing_scopes: return _automation_denied( request, status="authorization_reduced", reason=( "The automation owner no longer has every scope granted " "to this trigger." ), granted_scopes=granted_scopes, missing_scopes=missing_scopes, ) principal_ref = _build_principal_ref( session, account=account, user=user, tenant_id=request.tenant_id, scopes=list(granted_scopes), auth_method="service_account", idm_assignments=idm_assignments, identity_directory=identity_directory, extra_roles=idm_roles, authorization_context=authorization_context, include_system_roles=False, ) principal_ref = replace( principal_ref, service_account_id=None, acting_for_account_id=account.id, ) return _automation_allowed( session, request=request, principal_ref=principal_ref, granted_scopes=granted_scopes, ) def _resolve_service_account_automation( session: Session, *, request: AutomationPrincipalRequest, ) -> AutomationPrincipalResolution: item = session.get(ServiceAccount, request.service_account_id) tenant = session.get(Tenant, request.tenant_id) account = ( session.get(Account, item.account_id) if item is not None else None ) user = ( session.get(User, item.membership_id) if item is not None else None ) if ( item is None or tenant is None or account is None or user is None or item.tenant_id != request.tenant_id or not item.is_active or not tenant.is_active or not account.is_active or not user.is_active or item.account_id != account.id or item.membership_id != user.id or user.account_id != account.id or user.tenant_id != tenant.id or account.auth_provider != "service_account" or user.auth_provider != "service_account" ): return _automation_denied( request, status="inactive_or_inconsistent", reason=( "The service account is inactive, missing, or no longer " "belongs to the tenant." ), ) granted_scopes, missing_scopes = _current_trigger_grants( request.grant_scopes, current_scopes=item.scope_ceiling, ) if missing_scopes: return _automation_denied( request, status="authorization_reduced", reason=( "The service account no longer has every scope granted " "to this trigger." ), granted_scopes=granted_scopes, missing_scopes=missing_scopes, ) principal_ref = PrincipalRef( account_id=account.id, membership_id=user.id, tenant_id=tenant.id, scopes=frozenset(granted_scopes), auth_method="service_account", service_account_id=item.id, email=None, display_name=item.name, ) return _automation_allowed( session, request=request, principal_ref=principal_ref, granted_scopes=granted_scopes, ) def _current_trigger_grants( trigger_scopes: tuple[str, ...], *, current_scopes: list[str] | tuple[str, ...], ) -> tuple[tuple[str, ...], tuple[str, ...]]: granted = tuple( sorted( { required for required in trigger_scopes if scopes_grant_compatible( current_scopes, required, ) } ) ) missing = tuple( sorted(set(trigger_scopes) - set(granted)) ) return granted, missing def _automation_allowed( session: Session, *, request: AutomationPrincipalRequest, principal_ref: PrincipalRef, granted_scopes: tuple[str, ...], ) -> AutomationPrincipalResolution: api_principal = _api_principal_from_ref( session, principal_ref, permission_evaluator=LegacyPermissionEvaluator(), ) provenance = _automation_provenance( request, status="current_authorization_resolved", current_principal={ "kind": request.subject_kind, "account_id": principal_ref.account_id, "membership_id": principal_ref.membership_id, "service_account_id": principal_ref.service_account_id, "role_ids": sorted(principal_ref.role_ids), "group_ids": sorted(principal_ref.group_ids), "function_assignment_ids": sorted( principal_ref.function_assignment_ids ), "delegation_ids": sorted( principal_ref.delegation_ids ), "granted_scopes": list(granted_scopes), }, ) return AutomationPrincipalResolution( allowed=True, principal=api_principal, granted_scopes=granted_scopes, provenance=provenance, ) def _automation_denied( request: AutomationPrincipalRequest, *, status: str, reason: str, granted_scopes: tuple[str, ...] = (), missing_scopes: tuple[str, ...] | None = None, ) -> AutomationPrincipalResolution: missing = ( tuple(sorted(set(request.grant_scopes))) if missing_scopes is None else missing_scopes ) return AutomationPrincipalResolution( allowed=False, reason=reason, granted_scopes=granted_scopes, missing_scopes=missing, provenance=_automation_provenance( request, status=status, current_principal=None, ), ) def _automation_provenance( request: AutomationPrincipalRequest, *, status: str, current_principal: Mapping[str, object] | None, ) -> dict[str, object]: return { "contract_version": request.contract_version, "authorization_artifact": { "ref": request.authorization_ref, }, "trigger_owner": _automation_trigger_owner(request), "current_automation_principal": ( dict(current_principal) if current_principal is not None else None ), "event_actor": _automation_context_actor( request.context.get("event_actor") ), "operator_override": _automation_context_actor( request.context.get("operator_override") ), "trigger_ref": _optional_context_text( request.context.get("trigger_ref") ), "delivery_ref": _optional_context_text( request.context.get("delivery_ref") ), "status": status, } def _automation_trigger_owner( request: AutomationPrincipalRequest, ) -> dict[str, object]: return { "kind": request.subject_kind, "tenant_id": request.tenant_id, "account_id": request.account_id, "membership_id": request.membership_id, "service_account_id": request.service_account_id, } def _automation_context_actor( value: object, ) -> dict[str, str] | None: if not isinstance(value, Mapping): return None result = { key: str(value[key]).strip() for key in ( "kind", "type", "id", "label", "account_id", "membership_id", "service_account_id", "reason", ) if value.get(key) is not None and str(value[key]).strip() } return result or None def _optional_context_text(value: object) -> str | None: if value is None: return None clean = str(value).strip() return clean[:300] or None def get_api_principal( request: Request, session: Session = Depends(get_session), authorization: str | None = Header(default=None), x_api_key: str | None = Header(default=None, alias="X-API-Key"), ) -> ApiPrincipal: return resolve_api_principal( request, session, authorization=authorization, x_api_key=x_api_key, ) def _enforce_maintenance_mode(session: Session, principal: ApiPrincipal) -> None: mode = saved_maintenance_mode(session) if not mode.enabled or principal.has(MAINTENANCE_ACCESS_SCOPE): return raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=maintenance_response_detail(mode), ) def has_scope(principal: ApiPrincipal, required_scope: str) -> bool: return principal.has(required_scope) def require_scope(required_scope: str): def dependency(principal: ApiPrincipal = Depends(get_api_principal)) -> ApiPrincipal: if not has_scope(principal, required_scope): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {required_scope}") return principal return dependency def require_any_scope(*required_scopes: str): def dependency(principal: ApiPrincipal = Depends(get_api_principal)) -> ApiPrincipal: if not any(has_scope(principal, required) for required in required_scopes): joined = ", ".join(required_scopes) raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Requires one of: {joined}") return principal return dependency