"""Headless application service for the isolated user-engine MVP.""" from __future__ import annotations import hashlib import json from dataclasses import dataclass, field, replace from datetime import datetime, timedelta from typing import Any, Callable, Iterable, Mapping from user_engine.custody import CustodyObservationAdapter from user_engine.domain import ( Account, AccessControlFact, AccessMembershipRequirement, AccessProfile, AccessScopeType, AccountStatus, ActiveAccessContext, Actor, Application, ApplicationBinding, AttributeDefinition, AuditRecord, AuthorizationDecision, AuthorizationRequest, CanonEntityReference, CanonRelationshipReference, Catalog, CatalogLifecycle, ExternalIdentity, FactorVerification, FamilyDataspaceRequest, FamilyInvitation, FamilyMemberSpec, FamilyRole, IdentityFactor, IdentityFactorType, InvitationStatus, Membership, Mutability, OnboardingJourney, OnboardingJourneyStatus, OnboardingStep, OnboardingStepStatus, OnboardingTask, OnboardingTriggerType, OutboxEvent, PreparedAccount, PreparedAccountStatus, PreparedEntitlement, PreparedEntitlementKind, PreparedFactorRequirement, ProfileScope, ProfileValue, ProjectionType, RegistrationSession, RegistrationStatus, Sensitivity, SubsystemHandoff, TenantAccount, User, Visibility, WelcomeProtocol, WelcomeProtocolStep, new_id, utc_now, ) from user_engine.errors import ( AuthorizationDenied, ConflictError, NotFoundError, ValidationError, ) from user_engine.evidence import ( HEARTBEAT_EVENT_TYPE, load_bearing_counts, ) from user_engine.ports import ( AuthorizationCheckPort, FactorVerificationAdapter, IdentityClaimsAdapter, UserEngineStore, ) REDACTED = "" PLATFORM_TENANT = "platform:root" PLATFORM_OPERATOR_ROLE = "platform-operator" TENANT_ADMIN_ROLE = "tenant-admin" _TERMINAL_REGISTRATION_STATUSES = { RegistrationStatus.COMPLETED, RegistrationStatus.ABANDONED, RegistrationStatus.EXPIRED, RegistrationStatus.REJECTED, } _UNSET = object() @dataclass(frozen=True) class HealthReport: status: str schema_version: str | None @dataclass(frozen=True) class ReadinessReport: ready: bool checks: Mapping[str, bool] @dataclass(frozen=True) class AccountAuthority: """Account state and optional observed custody fact; never authorization.""" issuer: str subject: str tenant: str user_id: str account_id: str identity_id: str account_status: AccountStatus tenant_account_status: AccountStatus | None observed_at: datetime decision_id: str | None root_grant: Mapping[str, Any] | None = None @dataclass(frozen=True) class UserSession: actor: Actor user: User account: Account identities: tuple[ExternalIdentity, ...] @dataclass(frozen=True) class TenantContext: tenant: str actor: Actor platform_operator: bool @dataclass(frozen=True) class EffectiveProfile: user_id: str application_id: str | None values: Mapping[str, Any] source_layers: Mapping[str, str] trace: Mapping[str, tuple[str, ...]] @dataclass(frozen=True) class Projection: user_id: str projection_type: ProjectionType application_id: str | None values: Mapping[str, Any] redactions: Mapping[str, str] access_context: Mapping[str, Any] = field(default_factory=dict) @dataclass(frozen=True) class IdentityContext: actor: Actor user: User account: Account identities: tuple[ExternalIdentity, ...] tenant: str application_id: str | None memberships: tuple[Membership, ...] entity_refs: Mapping[str, CanonEntityReference] relationship_refs: tuple[CanonRelationshipReference, ...] grant_like_refs: tuple[CanonEntityReference, ...] evidence_refs: tuple[CanonEntityReference, ...] profile: EffectiveProfile | None = None gaps: tuple[str, ...] = () active_access_context: ActiveAccessContext | None = None access_control_facts: tuple[AccessControlFact, ...] = () onboarding_journeys: tuple[OnboardingJourney, ...] = () @dataclass(frozen=True) class AccessProfileSelection: profile: AccessProfile active_context: ActiveAccessContext access_control_facts: tuple[AccessControlFact, ...] @dataclass(frozen=True) class AccessControlFactExport: tenant: str facts: tuple[AccessControlFact, ...] manifest: Mapping[str, Any] @dataclass(frozen=True) class AccessProfileDiagnostics: tenant: str profile_count: int active_context_count: int required_factor_types: Mapping[str, tuple[str, ...]] issues: tuple[str, ...] @dataclass(frozen=True) class OnboardingJourneyStart: journey: OnboardingJourney protocol: WelcomeProtocol @dataclass(frozen=True) class OnboardingDiagnostics: tenant: str protocol_count: int journey_count: int statuses: Mapping[str, int] blocked_steps: tuple[str, ...] lifecycle_gaps: tuple[str, ...] @dataclass(frozen=True) class FamilyMemberInvitation: user: User tenant_account: TenantAccount membership: Membership invitation: FamilyInvitation @dataclass(frozen=True) class FamilyDataspaceOnboarding: tenant: str family_scope_id: str family_display_name: str owner_session: UserSession owner_membership: Membership application: Application binding: ApplicationBinding catalog: Catalog invitations: tuple[FamilyMemberInvitation, ...] identity_context: IdentityContext claims_projection: Projection @dataclass(frozen=True) class FamilyInvitationAcceptance: session: UserSession invitation: FamilyInvitation identity_context: IdentityContext claims_projection: Projection @dataclass(frozen=True) class RegistrationCompletion: session: RegistrationSession user: User account: Account netkingdom_id: str identity_context: IdentityContext @dataclass(frozen=True) class PreparedAccountClaim: prepared_account: PreparedAccount registration: RegistrationSession user: User memberships: tuple[Membership, ...] tenant_accounts: tuple[TenantAccount, ...] profile_values: tuple[ProfileValue, ...] application_bindings: tuple[ApplicationBinding, ...] onboarding_journeys: tuple[str, ...] @dataclass(frozen=True) class TenantDiagnostics: tenant: str checks: Mapping[str, bool] issues: tuple[str, ...] memberships: tuple[Membership, ...] @dataclass(frozen=True) class OutboxDiagnostics: pending_count: int event_types: Mapping[str, int] oldest_correlation_id: str | None @dataclass(frozen=True) class RegistrationDiagnostics: tenant: str total_sessions: int statuses: Mapping[str, int] verified_factor_count: int pending_session_count: int @dataclass(frozen=True) class OperabilitySnapshot: ready: bool checks: Mapping[str, bool] metrics: Mapping[str, int] issues: tuple[str, ...] class UserEngineService: """Headless service API for isolated user and profile management.""" def __init__( self, *, store: UserEngineStore, identity_adapter: IdentityClaimsAdapter, authorization: AuthorizationCheckPort, factor_verifier: FactorVerificationAdapter | None = None, custody: CustodyObservationAdapter | None = None, ) -> None: self.store = store self.identity_adapter = identity_adapter self.authorization = authorization self.factor_verifier = factor_verifier self.custody = custody if not self.store.ready: self.store.migrate() def health(self) -> HealthReport: return HealthReport(status="ok", schema_version=self.store.schema_version) def readiness(self) -> ReadinessReport: return ReadinessReport( ready=self.store.ready, checks={ "store_schema": self.store.ready, "identity_adapter": self.identity_adapter is not None, "authorization_port": self.authorization is not None, }, ) def resolve_tenant_context( self, actor: Actor, tenant: str | None = None ) -> TenantContext: resolved_tenant = tenant or actor.tenant if not resolved_tenant: raise ValidationError("tenant context is required") platform_operator = PLATFORM_OPERATOR_ROLE in actor.roles if resolved_tenant == PLATFORM_TENANT and not platform_operator: raise AuthorizationDenied("platform tenant requires platform operator role") if resolved_tenant != actor.tenant and not platform_operator: raise AuthorizationDenied("cross-tenant access denied") return TenantContext( tenant=resolved_tenant, actor=actor, platform_operator=platform_operator, ) def update_self_service_profile( self, actor: Actor, *, display_name: str, consent_accepted: bool, consent_version: str, correlation_id: str | None = None, ) -> User: """Update safe self-service fields; verified email remains immutable.""" display_name = display_name.strip() consent_version = consent_version.strip() if not 1 <= len(display_name) <= 200: raise ValidationError("display_name must contain 1 to 200 characters") if consent_accepted and not consent_version: raise ValidationError("consent_version is required when consent is accepted") correlation_id = correlation_id or new_id("corr") identity = self.store.find_identity(actor.issuer, actor.subject) if identity is None: raise NotFoundError("current identity is not linked") user = self._require_user(identity.user_id) decision = self._authorize( actor, action="profile.self.update", resource_type="user-engine:user", resource_id=user.user_id, tenant=actor.tenant, correlation_id=correlation_id, target_user_id=user.user_id, context={"self": True}, ) now = utc_now() updated = replace( user, display_name=display_name, updated_at=now, profile_completed_at=now, consented_at=now if consent_accepted else None, consent_version=consent_version if consent_accepted else None, ) with self.store.transaction(): self.store.save_user(updated) self._record_mutation( actor, action="profile.self.update", subject=user.user_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="user.self_service_profile_updated", aggregate_id=user.user_id, payload={ "user_id": user.user_id, "profile_complete": True, "consent_accepted": consent_accepted, "consent_version": consent_version if consent_accepted else None, }, ) return updated def start_registration( self, actor: Actor, *, tenant: str | None = None, required_factor_types: Iterable[IdentityFactorType | str] = ( IdentityFactorType.EMAIL, ), correlation_id: str | None = None, applicant_username: str | None = None, client_id: str | None = None, ) -> RegistrationSession: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") required = tuple(_factor_type(item) for item in required_factor_types) decision = self._authorize( actor, action="registration.start", resource_type="user-engine:registration", resource_id="new", tenant=tenant_context.tenant, correlation_id=correlation_id, context={"required_factor_types": tuple(item.value for item in required)}, ) session = RegistrationSession( tenant=tenant_context.tenant, status=( RegistrationStatus.FACTOR_PENDING if required else RegistrationStatus.STARTED ), required_factor_types=required, started_by_subject=actor.subject, applicant_username=applicant_username, client_id=client_id, correlation_id=correlation_id, ) with self.store.transaction(): self.store.save_registration_session(session) self._record_mutation( actor, action="registration.start", subject=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="registration.started", aggregate_id=session.registration_id, payload={ "registration_id": session.registration_id, "status": session.status, "required_factor_types": tuple( item.value for item in session.required_factor_types ), }, ) return session def attach_registration_factor( self, actor: Actor, registration_id: str, verification: FactorVerification | Mapping[str, Any], *, correlation_id: str | None = None, ) -> RegistrationSession: session = self._require_registration_session(registration_id) self._ensure_registration_mutable(session) tenant_context = self.resolve_tenant_context(actor, session.tenant) correlation_id = correlation_id or new_id("corr") verified = self._normalize_factor_verification(verification) if not verified.normalized_value: raise ValidationError("verified factor normalized_value is required") decision = self._authorize( actor, action="registration.factor.write", resource_type="user-engine:registration-factor", resource_id=f"{registration_id}:{verified.factor_type.value}", tenant=tenant_context.tenant, correlation_id=correlation_id, context={ "registration_id": registration_id, "factor_type": verified.factor_type.value, "source_system": verified.source_system, }, ) factor = IdentityFactor( factor_type=verified.factor_type, normalized_value=verified.normalized_value, registration_id=session.registration_id, display_value=verified.display_value, source_system=verified.source_system, assurance=dict(verified.assurance), evidence_refs=verified.evidence_refs, verified_at=verified.verified_at, expires_at=verified.expires_at, ) current_factor_ids = tuple( factor.factor_id for factor in self.store.factors_for_registration( session.registration_id ) ) verified_factor_ids = (*current_factor_ids, factor.factor_id) all_factor_types = { item.factor_type for item in self.store.factors_for_registration(session.registration_id) } all_factor_types.add(factor.factor_type) next_status = ( RegistrationStatus.FACTOR_VERIFIED if set(session.required_factor_types).issubset(all_factor_types) else RegistrationStatus.FACTOR_PENDING ) updated = replace( session, status=next_status, verified_factor_ids=verified_factor_ids, updated_at=utc_now(), ) with self.store.transaction(): self.store.save_identity_factor(factor) self.store.save_registration_session(updated) self._record_mutation( actor, action="registration.factor.write", subject=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="registration.factor_verified", aggregate_id=session.registration_id, payload={ "registration_id": session.registration_id, "factor_id": factor.factor_id, "factor_type": factor.factor_type, "source_system": factor.source_system, "evidence_ref_count": len(factor.evidence_refs), "status": updated.status, }, ) return updated def complete_registration( self, actor: Actor, registration_id: str, *, display_name: str | None = None, primary_email: str | None = None, correlation_id: str | None = None, ) -> RegistrationCompletion: session = self._require_registration_session(registration_id) self._ensure_registration_mutable(session) tenant_context = self.resolve_tenant_context(actor, session.tenant) factors = self.store.factors_for_registration(session.registration_id) missing = _missing_required_factor_types(session, factors) if missing: raise ValidationError( "registration is missing required factors: " + ", ".join(item.value for item in missing) ) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="registration.complete", resource_type="user-engine:registration", resource_id=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, context={ "factor_types": tuple( sorted({factor.factor_type.value for factor in factors}) ) }, ) existing_identity = self.store.find_identity(*actor.identity_key) if existing_identity is not None: user = self._require_user(existing_identity.user_id) account = self._require_account(user.user_id) tenant_account = self.store.tenant_account( tenant_context.tenant, user.user_id ) identity = existing_identity else: user = User( display_name=display_name or actor.preferred_username, primary_email=primary_email or _registration_primary_email( factors, actor ), ) account = Account( account_id=new_id("acct"), user_id=user.user_id, status=AccountStatus.ACTIVE, ) tenant_account = TenantAccount( user_id=user.user_id, tenant=tenant_context.tenant, status=AccountStatus.ACTIVE, ) identity = ExternalIdentity( identity_id=new_id("idn"), user_id=user.user_id, issuer=actor.issuer, subject=actor.subject, provider=actor.authorized_party, ) completed_at = utc_now() completed = replace( session, status=RegistrationStatus.COMPLETED, user_id=user.user_id, netkingdom_id=user.user_id, updated_at=completed_at, completed_at=completed_at, ) attached_factors = tuple( replace(factor, user_id=user.user_id) for factor in factors ) with self.store.transaction(): if existing_identity is None: self.store.save_user(user) self.store.save_account(account) self.store.save_tenant_account(tenant_account) self.store.save_identity(identity) elif tenant_account is None: self.store.save_tenant_account( TenantAccount( user_id=user.user_id, tenant=tenant_context.tenant, status=AccountStatus.ACTIVE, ) ) for factor in attached_factors: self.store.save_identity_factor(factor) self.store.save_registration_session(completed) self._record_mutation( actor, action="registration.complete", subject=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="registration.completed", aggregate_id=user.user_id, payload={ "registration_id": session.registration_id, "netkingdom_id": user.user_id, "user_id": user.user_id, "factor_types": tuple( sorted({factor.factor_type.value for factor in factors}) ), }, ) self._auto_start_onboarding_journeys( actor, user.user_id, self._matching_welcome_protocols_for_registration(completed), trigger_type=OnboardingTriggerType.REGISTRATION, source_id=session.registration_id, source_event_type="registration.completed", correlation_id=correlation_id, decision_id=decision.decision_id, ) identity_context = self.identity_context( actor, user_id=user.user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, ) return RegistrationCompletion( session=completed, user=user, account=account, netkingdom_id=user.user_id, identity_context=identity_context, ) def abandon_registration( self, actor: Actor, registration_id: str, *, correlation_id: str | None = None, ) -> RegistrationSession: return self._close_registration( actor, registration_id, status=RegistrationStatus.ABANDONED, action="registration.abandon", event_type="registration.abandoned", correlation_id=correlation_id, ) def expire_registration( self, actor: Actor, registration_id: str, *, correlation_id: str | None = None, ) -> RegistrationSession: return self._close_registration( actor, registration_id, status=RegistrationStatus.EXPIRED, action="registration.expire", event_type="registration.expired", correlation_id=correlation_id, ) def resume_registration( self, actor: Actor, registration_id: str, *, correlation_id: str | None = None, ) -> RegistrationSession: session = self._require_registration_session(registration_id) if session.status in _TERMINAL_REGISTRATION_STATUSES: raise ValidationError("terminal registration sessions cannot be resumed") tenant_context = self.resolve_tenant_context(actor, session.tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="registration.resume", resource_type="user-engine:registration", resource_id=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, ) return session def registration_diagnostics( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> RegistrationDiagnostics: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="registration.diagnostics.read", resource_type="user-engine:registration", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) sessions = tuple( session for session in self.store.all_registration_sessions() if session.tenant == tenant_context.tenant ) statuses: dict[str, int] = {} verified_factor_count = 0 for session in sessions: statuses[session.status.value] = statuses.get(session.status.value, 0) + 1 verified_factor_count += len( self.store.factors_for_registration(session.registration_id) ) pending_count = sum( statuses.get(status.value, 0) for status in ( RegistrationStatus.STARTED, RegistrationStatus.FACTOR_PENDING, RegistrationStatus.FACTOR_VERIFIED, ) ) return RegistrationDiagnostics( tenant=tenant_context.tenant, total_sessions=len(sessions), statuses=statuses, verified_factor_count=verified_factor_count, pending_session_count=pending_count, ) def prepare_account( self, actor: Actor, *, tenant: str, required_factor_matches: Iterable[PreparedFactorRequirement], entitlements: Iterable[PreparedEntitlement], display_name: str | None = None, primary_email: str | None = None, expires_at: datetime | None = None, correlation_id: str | None = None, ) -> PreparedAccount: tenant_context = self.resolve_tenant_context(actor, tenant) required = tuple(required_factor_matches) prepared_entitlements = tuple(entitlements) self._validate_prepared_factor_requirements(required) if not prepared_entitlements: raise ValidationError("prepared account requires at least one entitlement") self._validate_prepared_entitlements( prepared_entitlements, tenant_context.tenant ) self._ensure_no_duplicate_prepared_account( tenant_context.tenant, required, excluding_prepared_account_id=None, ) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="prepared_account.create", resource_type="user-engine:prepared-account", resource_id="new", tenant=tenant_context.tenant, correlation_id=correlation_id, context={ "factor_types": tuple( sorted({item.factor_type.value for item in required}) ), "entitlement_count": len(prepared_entitlements), }, ) prepared = PreparedAccount( tenant=tenant_context.tenant, required_factor_matches=required, entitlements=prepared_entitlements, display_name=display_name, primary_email=primary_email, prepared_by_subject=actor.subject, correlation_id=correlation_id, expires_at=expires_at, ) with self.store.transaction(): self.store.save_prepared_account(prepared) self._record_mutation( actor, action="prepared_account.create", subject=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="prepared_account.created", aggregate_id=prepared.prepared_account_id, payload={ "prepared_account_id": prepared.prepared_account_id, "factor_types": tuple( sorted( { item.factor_type.value for item in prepared.required_factor_matches } ) ), "entitlement_count": len(prepared.entitlements), "expires_at": prepared.expires_at.isoformat() if prepared.expires_at else None, }, ) return prepared def update_prepared_account( self, actor: Actor, prepared_account_id: str, *, required_factor_matches: Iterable[PreparedFactorRequirement] | None = None, entitlements: Iterable[PreparedEntitlement] | None = None, expires_at: datetime | None | object = _UNSET, correlation_id: str | None = None, ) -> PreparedAccount: prepared = self._require_prepared_account(prepared_account_id) self._ensure_prepared_account_pending(prepared) tenant_context = self.resolve_tenant_context(actor, prepared.tenant) required = ( tuple(required_factor_matches) if required_factor_matches is not None else prepared.required_factor_matches ) prepared_entitlements = ( tuple(entitlements) if entitlements is not None else prepared.entitlements ) self._validate_prepared_factor_requirements(required) if not prepared_entitlements: raise ValidationError("prepared account requires at least one entitlement") self._validate_prepared_entitlements( prepared_entitlements, tenant_context.tenant ) self._ensure_no_duplicate_prepared_account( tenant_context.tenant, required, excluding_prepared_account_id=prepared.prepared_account_id, ) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="prepared_account.update", resource_type="user-engine:prepared-account", resource_id=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, ) updated = replace( prepared, required_factor_matches=required, entitlements=prepared_entitlements, expires_at=prepared.expires_at if expires_at is _UNSET else expires_at, updated_at=utc_now(), ) with self.store.transaction(): self.store.save_prepared_account(updated) self._record_mutation( actor, action="prepared_account.update", subject=updated.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="prepared_account.updated", aggregate_id=updated.prepared_account_id, payload={ "prepared_account_id": updated.prepared_account_id, "factor_types": tuple( sorted( { item.factor_type.value for item in updated.required_factor_matches } ) ), "entitlement_count": len(updated.entitlements), "expires_at": updated.expires_at.isoformat() if updated.expires_at else None, }, ) return updated def list_prepared_accounts( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> tuple[PreparedAccount, ...]: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="prepared_account.read", resource_type="user-engine:prepared-account", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) return self.store.prepared_accounts_for_tenant(tenant_context.tenant) def revoke_prepared_account( self, actor: Actor, prepared_account_id: str, *, correlation_id: str | None = None, ) -> PreparedAccount: return self._close_prepared_account( actor, prepared_account_id, status=PreparedAccountStatus.REVOKED, action="prepared_account.revoke", event_type="prepared_account.revoked", correlation_id=correlation_id, ) def expire_prepared_account( self, actor: Actor, prepared_account_id: str, *, correlation_id: str | None = None, ) -> PreparedAccount: return self._close_prepared_account( actor, prepared_account_id, status=PreparedAccountStatus.EXPIRED, action="prepared_account.expire", event_type="prepared_account.expired", correlation_id=correlation_id, ) def claim_prepared_account( self, actor: Actor, registration_id: str, *, prepared_account_id: str | None = None, correlation_id: str | None = None, ) -> PreparedAccountClaim: session = self._require_registration_session(registration_id) if session.status != RegistrationStatus.COMPLETED or session.user_id is None: raise ValidationError("registration must be completed before claim") tenant_context = self.resolve_tenant_context(actor, session.tenant) correlation_id = correlation_id or new_id("corr") factors = self.store.factors_for_registration(session.registration_id) prepared = self._resolve_prepared_account_claim( actor, session, factors, prepared_account_id=prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, ) if any(entitlement.requires_approval for entitlement in prepared.entitlements): self._record_audit_only( actor, action="prepared_account.claim", subject=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, summary="prepared account claim denied: approval required", ) raise AuthorizationDenied("prepared account claim requires approval") decision = self._authorize( actor, action="prepared_account.claim", resource_type="user-engine:prepared-account", resource_id=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=session.user_id, context={ "registration_id": session.registration_id, "entitlement_count": len(prepared.entitlements), }, ) user = self._require_user(session.user_id) ( tenant_accounts, memberships, profile_values, application_bindings, onboarding_journeys, ) = self._prepared_entitlement_records( prepared, user.user_id, correlation_id=correlation_id, ) claimed_at = utc_now() claimed = replace( prepared, status=PreparedAccountStatus.CLAIMED, claimed_by_user_id=user.user_id, claimed_registration_id=session.registration_id, claimed_at=claimed_at, updated_at=claimed_at, ) with self.store.transaction(): for account in tenant_accounts: self.store.save_tenant_account(account) for membership in memberships: self.store.save_membership(membership) for value in profile_values: self.store.save_profile_value(value) for binding in application_bindings: self.store.save_binding(binding) self.store.save_prepared_account(claimed) self._record_mutation( actor, action="prepared_account.claim", subject=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="prepared_account.claimed", aggregate_id=user.user_id, payload={ "prepared_account_id": prepared.prepared_account_id, "registration_id": session.registration_id, "user_id": user.user_id, "entitlement_count": len(prepared.entitlements), }, ) for journey in onboarding_journeys: self._record_mutation( actor, action="prepared_account.onboarding.start", subject=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="prepared_account.onboarding_requested", aggregate_id=user.user_id, payload={ "prepared_account_id": prepared.prepared_account_id, "registration_id": session.registration_id, "user_id": user.user_id, "journey": journey, }, ) self._auto_start_onboarding_journeys( actor, user.user_id, self._matching_welcome_protocols_for_prepared_account(claimed), trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT, source_id=prepared.prepared_account_id, source_event_type="prepared_account.claimed", correlation_id=correlation_id, decision_id=decision.decision_id, ) return PreparedAccountClaim( prepared_account=claimed, registration=session, user=user, memberships=memberships, tenant_accounts=tenant_accounts, profile_values=profile_values, application_bindings=application_bindings, onboarding_journeys=onboarding_journeys, ) def register_access_profile( self, actor: Actor, profile: AccessProfile, *, correlation_id: str | None = None, ) -> AccessProfile: tenant_context = self.resolve_tenant_context(actor, profile.tenant) self._validate_access_profile(profile, tenant_context.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="access_profile.write", resource_type="user-engine:access-profile", resource_id=profile.access_profile_id, tenant=tenant_context.tenant, correlation_id=correlation_id, context={ "hat": profile.hat, "scope_type": profile.scope_type.value, "scope_id": _access_profile_scope_id(profile), "required_factor_types": tuple( factor_type.value for factor_type in profile.required_factor_types ), }, ) updated = replace(profile, tenant=tenant_context.tenant, updated_at=utc_now()) with self.store.transaction(): self.store.save_access_profile(updated) self._record_mutation( actor, action="access_profile.write", subject=updated.access_profile_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="access_profile.registered", aggregate_id=updated.access_profile_id, payload={ "access_profile_id": updated.access_profile_id, "hat": updated.hat, "scope_type": updated.scope_type, "scope_id": _access_profile_scope_id(updated), "required_factor_types": tuple( factor_type.value for factor_type in updated.required_factor_types ), "membership_requirement_count": len( updated.membership_requirements ), }, ) return updated def list_access_profiles( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> tuple[AccessProfile, ...]: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="access_profile.read", resource_type="user-engine:access-profile", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) return self.store.access_profiles_for_tenant(tenant_context.tenant) def select_active_hat( self, actor: Actor, user_id: str, access_profile_id: str, *, tenant: str | None = None, correlation_id: str | None = None, ) -> AccessProfileSelection: profile = self._require_access_profile(access_profile_id) tenant_context = self.resolve_tenant_context(actor, tenant or profile.tenant) if profile.tenant != tenant_context.tenant: raise AuthorizationDenied("access profile tenant mismatch") if profile.requires_approval: raise AuthorizationDenied("access profile requires approval") user = self._require_user(user_id) account = self._require_account(user.user_id) tenant_account = self.store.tenant_account(tenant_context.tenant, user.user_id) if tenant_account is None or tenant_account.status != AccountStatus.ACTIVE: raise ValidationError("active tenant account is required") memberships = self.store.memberships_for_user( user.user_id, tenant=tenant_context.tenant ) matched_memberships = self._satisfying_access_memberships( profile, memberships ) factor_ids = self._verified_factor_ids_for_user( user.user_id, profile.required_factor_types ) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="active_access_context.select", resource_type="user-engine:active-access-context", resource_id=user.user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=user.user_id, context={ "access_profile_id": profile.access_profile_id, "hat": profile.hat, "scope_type": profile.scope_type.value, "scope_id": _access_profile_scope_id(profile), "membership_ids": tuple( membership.membership_id for membership in matched_memberships ), "required_factor_types": tuple( factor_type.value for factor_type in profile.required_factor_types ), }, ) group_ids = _active_group_ids(profile, memberships) active_context = ActiveAccessContext( user_id=user.user_id, tenant=tenant_context.tenant, access_profile_id=profile.access_profile_id, hat=profile.hat, scope_type=profile.scope_type, scope_id=_access_profile_scope_id(profile), realm_id=profile.realm_id, service_id=profile.service_id, asset_id=profile.asset_id, membership_ids=tuple( membership.membership_id for membership in matched_memberships ), factor_ids=factor_ids, group_ids=group_ids, claims=dict(profile.claims), profile_defaults=dict(profile.profile_defaults), selected_by_subject=actor.subject, ) facts = self._access_control_facts( tenant_context.tenant, user_id=user.user_id, memberships=memberships, active_context=active_context, ) with self.store.transaction(): self.store.save_active_access_context(active_context) self._record_mutation( actor, action="active_access_context.select", subject=active_context.active_context_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="active_access_context.selected", aggregate_id=user.user_id, payload={ "active_context_id": active_context.active_context_id, "user_id": user.user_id, "access_profile_id": profile.access_profile_id, "hat": profile.hat, "scope_type": profile.scope_type, "scope_id": active_context.scope_id, "membership_ids": active_context.membership_ids, "factor_types": tuple( factor_type.value for factor_type in profile.required_factor_types ), }, ) return AccessProfileSelection( profile=profile, active_context=active_context, access_control_facts=facts, ) def export_access_control_facts( self, actor: Actor, *, tenant: str, user_id: str | None = None, correlation_id: str | None = None, ) -> AccessControlFactExport: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="access_control_facts.export", resource_type="user-engine:access-control-facts", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=user_id, ) memberships = ( self.store.memberships_for_user(user_id, tenant=tenant_context.tenant) if user_id is not None else self.store.memberships_for_tenant(tenant_context.tenant) ) active_context = ( self.store.active_access_context(user_id, tenant_context.tenant) if user_id is not None else None ) facts = self._access_control_facts( tenant_context.tenant, user_id=user_id, memberships=memberships, active_context=active_context, ) manifest = { "tenant": tenant_context.tenant, "fact_count": len(facts), "scope_types": tuple(sorted({fact.scope_type for fact in facts})), "subject_types": tuple(sorted({fact.subject_type for fact in facts})), } return AccessControlFactExport( tenant=tenant_context.tenant, facts=facts, manifest=manifest, ) def access_profile_diagnostics( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> AccessProfileDiagnostics: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="access_profile.diagnostics.read", resource_type="user-engine:access-profile", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) profiles = self.store.access_profiles_for_tenant(tenant_context.tenant) active_contexts = self.store.active_access_contexts_for_tenant( tenant_context.tenant ) issues = [] for profile in profiles: if profile.requires_approval: issues.append(f"approval-required:{profile.access_profile_id}") return AccessProfileDiagnostics( tenant=tenant_context.tenant, profile_count=len(profiles), active_context_count=len(active_contexts), required_factor_types={ profile.access_profile_id: tuple( factor_type.value for factor_type in profile.required_factor_types ) for profile in profiles }, issues=tuple(issues), ) def register_welcome_protocol( self, actor: Actor, protocol: WelcomeProtocol, *, correlation_id: str | None = None, ) -> WelcomeProtocol: tenant_context = self.resolve_tenant_context(actor, protocol.tenant) self._validate_welcome_protocol(protocol, tenant_context.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="welcome_protocol.write", resource_type="user-engine:welcome-protocol", resource_id=protocol.protocol_id, tenant=tenant_context.tenant, correlation_id=correlation_id, context={ "trigger_type": protocol.trigger_type.value, "journey_key": protocol.journey_key, "step_count": len(protocol.steps), }, ) updated = replace(protocol, tenant=tenant_context.tenant, updated_at=utc_now()) with self.store.transaction(): self.store.save_welcome_protocol(updated) self._record_mutation( actor, action="welcome_protocol.write", subject=updated.protocol_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="welcome_protocol.registered", aggregate_id=updated.protocol_id, payload={ "protocol_id": updated.protocol_id, "trigger_type": updated.trigger_type, "journey_key": updated.journey_key, "step_count": len(updated.steps), }, ) return updated def list_welcome_protocols( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> tuple[WelcomeProtocol, ...]: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="welcome_protocol.read", resource_type="user-engine:welcome-protocol", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) return self.store.welcome_protocols_for_tenant(tenant_context.tenant) def start_onboarding_journey( self, actor: Actor, user_id: str, protocol_id: str, *, trigger_type: OnboardingTriggerType = OnboardingTriggerType.MANUAL, source_id: str | None = None, source_event_type: str | None = None, correlation_id: str | None = None, ) -> OnboardingJourneyStart: protocol = self._require_welcome_protocol(protocol_id) tenant_context = self.resolve_tenant_context(actor, protocol.tenant) user = self._require_user(user_id) if self._onboarding_journey_exists(user.user_id, protocol.protocol_id, source_id): raise ConflictError("onboarding journey already exists for source") correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="onboarding_journey.start", resource_type="user-engine:onboarding-journey", resource_id=protocol.protocol_id, tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=user.user_id, context={ "protocol_id": protocol.protocol_id, "trigger_type": trigger_type.value, "source_event_type": source_event_type, }, ) journey = self._build_onboarding_journey( protocol, user.user_id, trigger_type=trigger_type, source_id=source_id, source_event_type=source_event_type, correlation_id=correlation_id, ) with self.store.transaction(): self.store.save_onboarding_journey(journey) self._record_onboarding_journey_started( actor, journey, decision_id=decision.decision_id, correlation_id=correlation_id, ) return OnboardingJourneyStart(journey=journey, protocol=protocol) def start_onboarding_for_registration( self, actor: Actor, registration_id: str, *, protocol_id: str | None = None, correlation_id: str | None = None, ) -> tuple[OnboardingJourneyStart, ...]: session = self._require_registration_session(registration_id) if session.status != RegistrationStatus.COMPLETED or session.user_id is None: raise ValidationError("registration must be completed before onboarding") protocols = ( (self._require_welcome_protocol(protocol_id),) if protocol_id is not None else self._matching_welcome_protocols_for_registration(session) ) starts = [] for protocol in protocols: starts.append( self.start_onboarding_journey( actor, session.user_id, protocol.protocol_id, trigger_type=OnboardingTriggerType.REGISTRATION, source_id=session.registration_id, source_event_type="registration.completed", correlation_id=correlation_id, ) ) return tuple(starts) def start_onboarding_for_prepared_account( self, actor: Actor, prepared_account_id: str, *, protocol_id: str | None = None, correlation_id: str | None = None, ) -> tuple[OnboardingJourneyStart, ...]: prepared = self._require_prepared_account(prepared_account_id) if ( prepared.status != PreparedAccountStatus.CLAIMED or prepared.claimed_by_user_id is None ): raise ValidationError("prepared account must be claimed before onboarding") protocols = ( (self._require_welcome_protocol(protocol_id),) if protocol_id is not None else self._matching_welcome_protocols_for_prepared_account(prepared) ) starts = [] for protocol in protocols: starts.append( self.start_onboarding_journey( actor, prepared.claimed_by_user_id, protocol.protocol_id, trigger_type=OnboardingTriggerType.PREPARED_ACCOUNT, source_id=prepared.prepared_account_id, source_event_type="prepared_account.claimed", correlation_id=correlation_id, ) ) return tuple(starts) def progress_onboarding_step( self, actor: Actor, journey_id: str, step_key: str, *, correlation_id: str | None = None, ) -> OnboardingJourney: return self._transition_onboarding_step( actor, journey_id, step_key, status=OnboardingStepStatus.IN_PROGRESS, action="onboarding_step.progress", event_type="onboarding_step.progressed", correlation_id=correlation_id, ) def complete_onboarding_step( self, actor: Actor, journey_id: str, step_key: str, *, correlation_id: str | None = None, ) -> OnboardingJourney: return self._transition_onboarding_step( actor, journey_id, step_key, status=OnboardingStepStatus.COMPLETED, action="onboarding_step.complete", event_type="onboarding_step.completed", correlation_id=correlation_id, ) def skip_onboarding_step( self, actor: Actor, journey_id: str, step_key: str, *, correlation_id: str | None = None, ) -> OnboardingJourney: return self._transition_onboarding_step( actor, journey_id, step_key, status=OnboardingStepStatus.SKIPPED, action="onboarding_step.skip", event_type="onboarding_step.skipped", correlation_id=correlation_id, ) def fail_onboarding_step( self, actor: Actor, journey_id: str, step_key: str, *, lifecycle_gap: str | None = None, correlation_id: str | None = None, ) -> OnboardingJourney: return self._transition_onboarding_step( actor, journey_id, step_key, status=OnboardingStepStatus.FAILED, action="onboarding_step.fail", event_type="onboarding_step.failed", lifecycle_gap=lifecycle_gap, correlation_id=correlation_id, ) def resume_onboarding_journey( self, actor: Actor, journey_id: str, *, callback_refs: Mapping[str, str] | None = None, correlation_id: str | None = None, ) -> OnboardingJourney: journey = self._require_onboarding_journey(journey_id) tenant_context = self.resolve_tenant_context(actor, journey.tenant) if journey.status not in ( OnboardingJourneyStatus.BLOCKED, OnboardingJourneyStatus.FAILED, OnboardingJourneyStatus.PENDING, ): raise ValidationError("only pending, blocked, or failed journeys can resume") correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="onboarding_journey.resume", resource_type="user-engine:onboarding-journey", resource_id=journey.journey_id, tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=journey.user_id, ) updated_steps = _resume_onboarding_steps(journey.steps, callback_refs or {}) updated = replace( journey, steps=updated_steps, status=_journey_status_from_steps(updated_steps), active_step_key=_active_onboarding_step_key(updated_steps), updated_at=utc_now(), failed_at=None, ) with self.store.transaction(): self.store.save_onboarding_journey(updated) self._record_mutation( actor, action="onboarding_journey.resume", subject=updated.journey_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="onboarding_journey.resumed", aggregate_id=updated.user_id, payload={ "journey_id": updated.journey_id, "user_id": updated.user_id, "status": updated.status, "active_step_key": updated.active_step_key, }, ) return updated def onboarding_diagnostics( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> OnboardingDiagnostics: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="onboarding.diagnostics.read", resource_type="user-engine:onboarding", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) protocols = self.store.welcome_protocols_for_tenant(tenant_context.tenant) journeys = self.store.onboarding_journeys_for_tenant(tenant_context.tenant) statuses: dict[str, int] = {} blocked_steps = [] lifecycle_gaps = [] for journey in journeys: statuses[journey.status.value] = statuses.get(journey.status.value, 0) + 1 for step in journey.steps: if step.status == OnboardingStepStatus.BLOCKED: blocked_steps.append(f"{journey.journey_id}:{step.step_key}") if step.lifecycle_gap: lifecycle_gaps.append(step.lifecycle_gap) return OnboardingDiagnostics( tenant=tenant_context.tenant, protocol_count=len(protocols), journey_count=len(journeys), statuses=statuses, blocked_steps=tuple(blocked_steps), lifecycle_gaps=tuple(sorted(set(lifecycle_gaps))), ) def me( self, claims: Mapping[str, Any], *, correlation_id: str | None = None, ) -> UserSession: actor = self.identity_adapter.normalize(claims) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="me.read", resource_type="user-engine:me", resource_id=actor.subject, tenant=actor.tenant, correlation_id=correlation_id, context={"self": True}, ) identity = self.store.find_identity(*actor.identity_key) if identity is not None: user = self._require_user(identity.user_id) account = self._require_account(user.user_id) return self._session(actor, user, account) user = User( display_name=actor.preferred_username, primary_email=_optional_claim(actor, "email"), ) account = Account( account_id=new_id("acct"), user_id=user.user_id, status=AccountStatus.ACTIVE, ) tenant_account = TenantAccount(user_id=user.user_id, tenant=actor.tenant) identity = ExternalIdentity( identity_id=new_id("idn"), user_id=user.user_id, issuer=actor.issuer, subject=actor.subject, provider=actor.authorized_party, ) with self.store.transaction(): self.store.save_user(user) self.store.save_account(account) self.store.save_tenant_account(tenant_account) self.store.save_identity(identity) self._record_mutation( actor, action="user.create_from_identity", subject=user.user_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="user.created", aggregate_id=user.user_id, payload={ "user_id": user.user_id, "account_id": account.account_id, "identity": {"issuer": actor.issuer, "subject": actor.subject}, }, ) return self._session(actor, user, account) def lookup_account_authority( self, caller: Actor, *, issuer: str, subject: str, tenant: str, correlation_id: str | None = None, ) -> AccountAuthority: """Authorize a read of an existing identity; never call provisioning /me. The host must authenticate caller independently of the lookup target. This action needs its own policy grant; caller roles do not grant it here. Missing scoped accounts are explicit, never defaulted to active. """ if any(not isinstance(value, str) or not value.strip() for value in (issuer, subject, tenant)): raise ValidationError("exact issuer, subject and tenant are required") correlation_id = correlation_id or new_id("corr") lookup_id = "sha256:" + hashlib.sha256(json.dumps( [issuer, subject, tenant], separators=(",", ":"), ).encode()).hexdigest() decision = self._authorize( caller, action="account.authority.read", resource_type="user-engine:account-authority", resource_id=lookup_id, tenant=tenant, correlation_id=correlation_id, context={"lookup_issuer": issuer, "lookup_subject": subject}, ) observed_at = utc_now() identity = self.store.find_identity(issuer, subject) if identity is None: raise NotFoundError("identity link not found") user = self._require_user(identity.user_id) account = self._require_account(user.user_id) scoped = self.store.tenant_account(tenant, user.user_id) return AccountAuthority( issuer=issuer, subject=subject, tenant=tenant, user_id=user.user_id, account_id=account.account_id, identity_id=identity.identity_id, account_status=account.status, tenant_account_status=scoped.status if scoped is not None else None, observed_at=observed_at, decision_id=decision.decision_id, root_grant=(self.custody.observe(issuer=issuer, subject=subject, tenant=tenant) if self.custody is not None else None), ) def create_user( self, actor: Actor, *, display_name: str | None, primary_email: str | None, correlation_id: str | None = None, ) -> User: correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="user.create", resource_type="user-engine:user", resource_id="new", tenant=actor.tenant, correlation_id=correlation_id, ) user = User(display_name=display_name, primary_email=primary_email) account = Account(account_id=new_id("acct"), user_id=user.user_id) tenant_account = TenantAccount(user_id=user.user_id, tenant=actor.tenant) with self.store.transaction(): self.store.save_user(user) self.store.save_account(account) self.store.save_tenant_account(tenant_account) self._record_mutation( actor, action="user.create", subject=user.user_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="user.created", aggregate_id=user.user_id, payload={"user_id": user.user_id, "account_id": account.account_id}, ) return user def set_account_status( self, actor: Actor, user_id: str, status: AccountStatus, *, correlation_id: str | None = None, ) -> Account: account = self._require_account(user_id) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="account.update", resource_type="user-engine:account", resource_id=account.account_id, tenant=actor.tenant, correlation_id=correlation_id, target_user_id=user_id, ) updated = replace(account, status=status) with self.store.transaction(): self.store.save_account(updated) self._record_mutation( actor, action="account.update", subject=user_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="account.status_changed", aggregate_id=account.account_id, payload={"user_id": user_id, "status": status}, ) return updated def authorize_tenant_member_action( self, actor: Actor, user_id: str, *, tenant: str, correlation_id: str, ) -> User: """Authorize before any credential-provider side effect or target readout.""" self.resolve_tenant_context(actor, tenant) if not {"tenant-admin", PLATFORM_OPERATOR_ROLE}.intersection(actor.roles): raise AuthorizationDenied("tenant administrator role required") memberships = self.store.memberships_for_user(user_id, tenant=tenant) if not any(m.scope_type == "tenant" and m.scope_id == tenant for m in memberships): raise NotFoundError("account is not a member of this tenant") self._authorize(actor, action="tenant.account.update", resource_type="user-engine:tenant-account", resource_id=f"{tenant}:{user_id}", tenant=tenant, correlation_id=correlation_id, target_user_id=user_id) return self._require_user(user_id) def validate_tenant_role_change(self, actor: Actor, user_id: str, role: str, *, tenant: str, correlation_id: str) -> None: self.authorize_tenant_member_action(actor, user_id, tenant=tenant, correlation_id=correlation_id) if role not in {"user", "tenant-admin"}: raise ValidationError("Choose User or Tenant administrator.") self._authorize(actor, action="membership.write", resource_type="user-engine:membership", resource_id=f"{tenant}:{user_id}:tenant:{tenant}", tenant=tenant, correlation_id=correlation_id, target_user_id=user_id, context={"scope_type":"tenant", "scope_id":tenant, "kind":role}) if role != "tenant-admin": self.require_admin_successor(user_id, tenant=tenant) def set_tenant_role(self, actor: Actor, user_id: str, role: str, *, tenant: str, correlation_id: str) -> None: with self.store.tenant_lifecycle_guard(tenant): self._set_tenant_role(actor, user_id, role, tenant=tenant, correlation_id=correlation_id) def _set_tenant_role(self, actor: Actor, user_id: str, role: str, *, tenant: str, correlation_id: str) -> None: self.validate_tenant_role_change(actor, user_id, role, tenant=tenant, correlation_id=correlation_id) with self.store.transaction(): for membership in self.store.memberships_for_user(user_id, tenant=tenant): if membership.scope_type == "tenant" and membership.scope_id == tenant: self.store.save_membership(replace(membership, kind=role, freshness_version=correlation_id)) self._record_mutation(actor, action="membership.write", subject=user_id, tenant=tenant, correlation_id=correlation_id, decision_id=None, event_type="membership.role_changed", aggregate_id=user_id, payload={"user_id":user_id, "tenant":tenant, "role":role}) def require_admin_successor(self, user_id: str, *, tenant: str) -> None: """Do not deactivate the last active tenant administrator.""" members = self.store.memberships_for_tenant(tenant) admins = {m.user_id for m in members if m.scope_type == "tenant" and m.scope_id == tenant and m.kind == "tenant-admin"} if user_id not in admins: return account = self.store.tenant_account(tenant, user_id) if account is not None and account.status != AccountStatus.ACTIVE: return for other in admins - {user_id}: state = self.store.tenant_account(tenant, other) if state is not None and state.status == AccountStatus.ACTIVE: return raise ConflictError("Assign another active tenant administrator before disabling this account.") def set_tenant_account_status(self, actor: Actor, user_id: str, status: AccountStatus, *, tenant: str, correlation_id: str | None = None) -> TenantAccount: with self.store.tenant_lifecycle_guard(tenant): return self._set_tenant_account_status(actor, user_id, status, tenant=tenant, correlation_id=correlation_id) def _set_tenant_account_status( self, actor: Actor, user_id: str, status: AccountStatus, *, tenant: str, correlation_id: str | None = None, ) -> TenantAccount: self._require_user(user_id) tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") account = self.store.tenant_account(tenant_context.tenant, user_id) if account is None: account = TenantAccount(user_id=user_id, tenant=tenant_context.tenant) decision = self._authorize( actor, action="tenant.account.update", resource_type="user-engine:tenant-account", resource_id=f"{tenant_context.tenant}:{user_id}", tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=user_id, ) updated = replace(account, status=status) with self.store.transaction(): if status in {AccountStatus.SUSPENDED, AccountStatus.DISABLED}: self.require_admin_successor(user_id, tenant=tenant_context.tenant) self.store.save_tenant_account(updated) self._record_mutation( actor, action="tenant.account.update", subject=user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="tenant_account.status_changed", aggregate_id=user_id, payload={ "user_id": user_id, "tenant": tenant_context.tenant, "status": status, }, ) return updated def add_membership( self, actor: Actor, user_id: str, *, tenant: str, scope_type: str, scope_id: str, kind: str, correlation_id: str | None = None, ) -> Membership: self._require_user(user_id) tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="membership.write", resource_type="user-engine:membership", resource_id=f"{tenant_context.tenant}:{user_id}:{scope_type}:{scope_id}", tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=user_id, context={"scope_type": scope_type, "scope_id": scope_id, "kind": kind}, ) membership = Membership( membership_id=new_id("mem"), user_id=user_id, tenant=tenant_context.tenant, scope_type=scope_type, scope_id=scope_id, kind=kind, freshness_version=correlation_id, ) with self.store.transaction(): self.store.save_membership(membership) self._record_mutation( actor, action="membership.write", subject=user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="membership.added", aggregate_id=membership.membership_id, payload={ "membership_id": membership.membership_id, "user_id": user_id, "tenant": tenant_context.tenant, "scope_type": scope_type, "scope_id": scope_id, "kind": kind, }, ) return membership def link_identity( self, actor: Actor, user_id: str, *, issuer: str, subject: str, provider: str | None = None, correlation_id: str | None = None, ) -> ExternalIdentity: self._require_user(user_id) existing = self.store.find_identity(issuer, subject) if existing is not None and existing.user_id != user_id: raise ConflictError("external identity is already linked") if existing is not None: return existing correlation_id = correlation_id or new_id("corr") actor_identity = self.store.find_identity(*actor.identity_key) decision = self._authorize( actor, action="identity.link", resource_type="user-engine:identity", resource_id=f"{issuer}:{subject}", tenant=actor.tenant, correlation_id=correlation_id, target_user_id=user_id, context={ "self": actor_identity is not None and actor_identity.user_id == user_id, }, ) identity = ExternalIdentity( identity_id=new_id("idn"), user_id=user_id, issuer=issuer, subject=subject, provider=provider, ) with self.store.transaction(): self.store.save_identity(identity) self._record_mutation( actor, action="identity.link", subject=user_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="identity.linked", aggregate_id=user_id, payload={"issuer": issuer, "subject": subject, "user_id": user_id}, ) return identity def register_application( self, actor: Actor, application: Application, *, binding: ApplicationBinding | None = None, correlation_id: str | None = None, ) -> Application: if self.store.application(application.application_id) is not None: raise ConflictError("application already exists") correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="application.register", resource_type="user-engine:application", resource_id=application.application_id, tenant=actor.tenant, correlation_id=correlation_id, application_id=application.application_id, ) with self.store.transaction(): self.store.save_application(application) if binding is not None: if binding.application_id != application.application_id: raise ValidationError("binding application id must match application") self.store.save_binding(binding) self._record_mutation( actor, action="application.register", subject=application.application_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="application.registered", aggregate_id=application.application_id, payload={"application_id": application.application_id}, application_id=application.application_id, ) return application def publish_catalog( self, actor: Actor, catalog: Catalog, *, correlation_id: str | None = None, ) -> Catalog: self._validate_catalog(catalog) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="catalog.publish", resource_type="user-engine:catalog", resource_id=catalog.catalog_id, tenant=actor.tenant, correlation_id=correlation_id, application_id=catalog.owning_application_id, ) with self.store.transaction(): self.store.save_catalog(catalog) self._record_mutation( actor, action="catalog.publish", subject=catalog.catalog_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="catalog.published", aggregate_id=catalog.catalog_id, payload={ "catalog_id": catalog.catalog_id, "namespace": catalog.namespace, "version": catalog.version, }, application_id=catalog.owning_application_id, ) return catalog def set_profile_value( self, actor: Actor, user_id: str, attribute_key: str, value: Any, *, scope: ProfileScope = ProfileScope.GLOBAL, scope_id: str | None = None, tenant: str | None = None, application_id: str | None = None, correlation_id: str | None = None, ) -> ProfileValue: self._require_user(user_id) definition = self._require_attribute(attribute_key) self._validate_profile_scope(definition, scope, scope_id) self._validate_value(definition, value) operation_tenant = self._operation_tenant(actor, tenant, scope, scope_id) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="profile.write", resource_type="user-engine:profile", resource_id=f"{user_id}:{attribute_key}", tenant=operation_tenant, correlation_id=correlation_id, application_id=application_id, target_user_id=user_id, context={"scope": scope, "scope_id": scope_id}, ) profile_value = ProfileValue( user_id=user_id, attribute_key=attribute_key, value=value, scope=scope, scope_id=scope_id, source=f"{actor.issuer}:{actor.subject}", ) with self.store.transaction(): self.store.save_profile_value(profile_value) self._record_mutation( actor, action="profile.write", subject=user_id, tenant=operation_tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="profile.value_set", aggregate_id=user_id, payload={ "user_id": user_id, "attribute_key": attribute_key, "scope": scope, "scope_id": scope_id, }, application_id=application_id, ) return profile_value def effective_profile( self, actor: Actor, user_id: str, *, application_id: str | None = None, tenant: str | None = None, correlation_id: str | None = None, ) -> EffectiveProfile: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="profile.read", resource_type="user-engine:profile", resource_id=user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=application_id, target_user_id=user_id, ) return self._resolve_effective_profile(user_id, application_id, tenant_context.tenant) def projection( self, actor: Actor, user_id: str, projection_type: ProjectionType, *, application_id: str | None = None, tenant: str | None = None, correlation_id: str | None = None, ) -> Projection: if _is_application_projection(projection_type) and application_id is None: raise ValidationError("application projections require application_id") tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="projection.read", resource_type="user-engine:projection", resource_id=user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=application_id, target_user_id=user_id, context={"projection_type": projection_type}, ) effective = self._resolve_effective_profile( user_id, application_id, tenant_context.tenant ) owning_application_id = ( application_id if _is_application_projection(projection_type) else None ) values: dict[str, Any] = {} redactions: dict[str, str] = {} for key, definition in self._active_attribute_definitions( owning_application_id=owning_application_id ).items(): if key not in effective.values: continue if not _visible_in_projection(definition, projection_type): continue if _must_redact(definition, projection_type): values[key] = REDACTED redactions[key] = "sensitivity" else: values[key] = effective.values[key] return Projection( user_id=user_id, projection_type=projection_type, application_id=application_id, values=values, redactions=redactions, access_context=self._projection_access_context( user_id, tenant_context.tenant, application_id=application_id, ), ) def identity_context( self, actor: Actor, *, user_id: str | None = None, tenant: str | None = None, application_id: str | None = None, include_profile: bool = False, correlation_id: str | None = None, ) -> IdentityContext: tenant_context = self.resolve_tenant_context(actor, tenant) identity = self.store.find_identity(*actor.identity_key) if user_id is None: if identity is None: raise NotFoundError("identity link not found") user_id = identity.user_id user = self._require_user(user_id) account = self._require_account(user_id) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="identity_context.read", resource_type="user-engine:identity-context", resource_id=user_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=application_id, target_user_id=user_id, context={ "include_profile": include_profile, "self": identity is not None and identity.user_id == user_id, }, ) identities = self.store.identities_for_user(user.user_id) memberships = self.store.memberships_for_user( user.user_id, tenant=tenant_context.tenant ) active_access_context = self.store.active_access_context( user.user_id, tenant_context.tenant ) evidence_refs = self._identity_evidence_refs( user, account, memberships, tenant_context.tenant, active_access_context=active_access_context, ) access_control_facts = self._access_control_facts( tenant_context.tenant, user_id=user.user_id, memberships=memberships, active_context=active_access_context, ) onboarding_journeys = self.store.onboarding_journeys_for_user( user.user_id, tenant=tenant_context.tenant ) entity_refs = self._identity_entity_refs( actor, user, account, identities, memberships, tenant_context.tenant, application_id, active_access_context, ) relationship_refs, grant_like_refs = self._identity_relationship_refs( entity_refs, memberships, tenant_context.tenant, evidence_refs[0].identifier if evidence_refs else None, active_access_context=active_access_context, ) profile = ( self._resolve_effective_profile( user.user_id, application_id, tenant_context.tenant ) if include_profile else None ) gaps = () if evidence_refs else ("evidence:no-audit-records",) return IdentityContext( actor=actor, user=user, account=account, identities=identities, tenant=tenant_context.tenant, application_id=application_id, memberships=memberships, entity_refs=entity_refs, relationship_refs=relationship_refs, grant_like_refs=grant_like_refs, evidence_refs=evidence_refs, profile=profile, gaps=gaps, active_access_context=active_access_context, access_control_facts=access_control_facts, onboarding_journeys=onboarding_journeys, ) def onboard_family_dataspace( self, actor: Actor, request: FamilyDataspaceRequest, *, correlation_id: str | None = None, ) -> FamilyDataspaceOnboarding: tenant_context = self.resolve_tenant_context(actor, request.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="family_dataspace.onboard", resource_type="user-engine:family-dataspace", resource_id=request.family_scope_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=request.application_id, context={"family_scope_id": request.family_scope_id}, ) with self.store.transaction(): owner_session = self._ensure_actor_session(actor, correlation_id) application, binding = self._ensure_family_dataspace_application( actor, request, correlation_id ) catalog = self._ensure_family_dataspace_catalog( actor, request, correlation_id ) owner_membership = self._ensure_membership( actor, owner_session.user.user_id, tenant=tenant_context.tenant, scope_type="family", scope_id=request.family_scope_id, kind=FamilyRole.OWNER.value, correlation_id=correlation_id, ) owner_defaults = dict(request.owner_profile_defaults) owner_defaults.setdefault( "member_display_name", owner_session.user.display_name or actor.preferred_username or owner_session.user.primary_email or actor.subject, ) self._apply_family_profile_defaults( actor, owner_session.user.user_id, tenant=tenant_context.tenant, application_id=request.application_id, catalog_namespace=request.catalog_namespace, values=owner_defaults, correlation_id=correlation_id, ) invitations = tuple( self.invite_family_member( actor, tenant=tenant_context.tenant, family_scope_id=request.family_scope_id, application_id=request.application_id, catalog_namespace=request.catalog_namespace, member=member, correlation_id=correlation_id, ) for member in request.member_specs ) self._record_mutation( actor, action="family_dataspace.onboard", subject=request.family_scope_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="family_dataspace.onboarded", aggregate_id=request.family_scope_id, payload={ "tenant": tenant_context.tenant, "family_scope_id": request.family_scope_id, "application_id": request.application_id, "member_invitation_count": len(invitations), }, application_id=request.application_id, ) identity_context = self.identity_context( actor, user_id=owner_session.user.user_id, tenant=tenant_context.tenant, application_id=request.application_id, include_profile=True, correlation_id=correlation_id, ) claims_projection = self.projection( actor, owner_session.user.user_id, ProjectionType.CLAIMS_ENRICHMENT, tenant=tenant_context.tenant, application_id=request.application_id, correlation_id=correlation_id, ) return FamilyDataspaceOnboarding( tenant=tenant_context.tenant, family_scope_id=request.family_scope_id, family_display_name=request.family_display_name, owner_session=owner_session, owner_membership=owner_membership, application=application, binding=binding, catalog=catalog, invitations=invitations, identity_context=identity_context, claims_projection=claims_projection, ) def invite_family_member( self, actor: Actor, *, tenant: str, family_scope_id: str, application_id: str, member: FamilyMemberSpec, catalog_namespace: str = "dataspace", correlation_id: str | None = None, expires_in: timedelta = timedelta(days=7), ) -> FamilyMemberInvitation: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") role = _family_role_value(member.role) if not member.primary_email: raise ValidationError("family member primary_email is required") if any( item.primary_email.casefold() == member.primary_email.casefold() and item.status == InvitationStatus.PENDING for item in self.store.family_invitations_for_tenant(tenant_context.tenant) ): raise ConflictError("a pending invitation already exists for this address") if member.issuer and member.subject: existing = self.store.find_identity(member.issuer, member.subject) if existing is not None: raise ConflictError("external identity is already linked") decision = self._authorize( actor, action="family_member.invite", resource_type="user-engine:family-invitation", resource_id=member.primary_email, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=application_id, context={ "family_scope_id": family_scope_id, "role": role, }, ) with self.store.transaction(): user = self.create_user( actor, display_name=member.display_name, primary_email=member.primary_email, correlation_id=correlation_id, ) tenant_account = self.set_tenant_account_status( actor, user.user_id, AccountStatus.INVITED, tenant=tenant_context.tenant, correlation_id=correlation_id, ) membership = self._ensure_membership( actor, user.user_id, tenant=tenant_context.tenant, scope_type="family", scope_id=family_scope_id, kind=role, correlation_id=correlation_id, ) profile_defaults = dict(member.profile_defaults) if member.display_name and role not in {"user", TENANT_ADMIN_ROLE}: profile_defaults.setdefault("member_display_name", member.display_name) self._apply_family_profile_defaults( actor, user.user_id, tenant=tenant_context.tenant, application_id=application_id, catalog_namespace=catalog_namespace, values=profile_defaults, correlation_id=correlation_id, ) if member.issuer and member.subject: self.link_identity( actor, user.user_id, issuer=member.issuer, subject=member.subject, provider=member.provider, correlation_id=correlation_id, ) invitation = FamilyInvitation( invitation_id=new_id("inv"), tenant=tenant_context.tenant, family_scope_id=family_scope_id, application_id=application_id, user_id=user.user_id, primary_email=member.primary_email, role=role, invited_by=actor.subject, correlation_id=correlation_id, last_sent_correlation_id=correlation_id, expires_at=utc_now() + expires_in, ) self.store.save_family_invitation(invitation) self._record_mutation( actor, action="family_member.invite", subject=invitation.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="family_member.invited", aggregate_id=invitation.invitation_id, payload={ "invitation_id": invitation.invitation_id, "user_id": user.user_id, "tenant": tenant_context.tenant, "family_scope_id": family_scope_id, "application_id": application_id, "role": role, "primary_email": invitation.primary_email, }, application_id=application_id, ) return FamilyMemberInvitation( user=user, tenant_account=tenant_account, membership=membership, invitation=invitation, ) def resend_family_invitation( self, actor: Actor, invitation_id: str, *, correlation_id: str | None = None, expected_version: int | None = None, ) -> FamilyInvitation: invitation = self._require_family_invitation(invitation_id) if invitation.status != InvitationStatus.PENDING: raise ValidationError("only pending invitations can be resent") if expected_version is not None and invitation.version != expected_version: raise ConflictError("invitation version does not match") tenant_context = self.resolve_tenant_context(actor, invitation.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="family_invitation.resend", resource_type="user-engine:family-invitation", resource_id=invitation.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=invitation.application_id, target_user_id=invitation.user_id, ) updated = replace( invitation, resend_count=invitation.resend_count + 1, last_sent_correlation_id=correlation_id, updated_at=utc_now(), expires_at=utc_now() + timedelta(days=7), version=invitation.version + 1, ) with self.store.transaction(): self.store.save_family_invitation(updated) self._record_mutation( actor, action="family_invitation.resend", subject=updated.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="family_invitation.resent", aggregate_id=updated.invitation_id, payload={ "invitation_id": updated.invitation_id, "user_id": updated.user_id, "resend_count": updated.resend_count, "primary_email": updated.primary_email, }, application_id=updated.application_id, ) return updated def revoke_family_invitation( self, actor: Actor, invitation_id: str, *, correlation_id: str | None = None, expected_version: int | None = None, ) -> FamilyInvitation: invitation = self._require_family_invitation(invitation_id) if invitation.status != InvitationStatus.PENDING: raise ValidationError("only pending invitations can be revoked") if expected_version is not None and invitation.version != expected_version: raise ConflictError("invitation version does not match") tenant_context = self.resolve_tenant_context(actor, invitation.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="family_invitation.revoke", resource_type="user-engine:family-invitation", resource_id=invitation.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=invitation.application_id, target_user_id=invitation.user_id, ) with self.store.transaction(): self.set_tenant_account_status( actor, invitation.user_id, AccountStatus.DISABLED, tenant=tenant_context.tenant, correlation_id=correlation_id, ) updated = replace( invitation, status=InvitationStatus.REVOKED, updated_at=utc_now(), revoked_at=utc_now(), version=invitation.version + 1, ) self.store.save_family_invitation(updated) self._record_mutation( actor, action="family_invitation.revoke", subject=updated.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="family_invitation.revoked", aggregate_id=updated.invitation_id, payload={ "invitation_id": updated.invitation_id, "user_id": updated.user_id, "status": updated.status, }, application_id=updated.application_id, ) return updated def accept_family_invitation( self, claims: Mapping[str, Any], invitation_id: str, *, correlation_id: str | None = None, ) -> FamilyInvitationAcceptance: invitation = self._require_family_invitation(invitation_id) if invitation.status == InvitationStatus.REVOKED: raise ValidationError("revoked invitations cannot be accepted") if invitation.status == InvitationStatus.ACCEPTED: raise ValidationError("invitation is already accepted") if invitation.expires_at is not None and invitation.expires_at <= utc_now(): raise ValidationError("invitation has expired") actor = self.identity_adapter.normalize(claims) tenant_context = self.resolve_tenant_context(actor, invitation.tenant) claimed_email = str(claims.get("email") or "").strip().casefold() if not claimed_email or claimed_email != invitation.primary_email.casefold(): raise AuthorizationDenied("invitation identity does not match") correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action="family_invitation.accept", resource_type="user-engine:family-invitation", resource_id=invitation.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, application_id=invitation.application_id, target_user_id=invitation.user_id, context={"family_scope_id": invitation.family_scope_id}, ) with self.store.transaction(): self.link_identity( actor, invitation.user_id, issuer=actor.issuer, subject=actor.subject, provider=actor.authorized_party, correlation_id=correlation_id, ) account = self.set_account_status( actor, invitation.user_id, AccountStatus.ACTIVE, correlation_id=correlation_id, ) self.set_tenant_account_status( actor, invitation.user_id, AccountStatus.ACTIVE, tenant=tenant_context.tenant, correlation_id=correlation_id, ) accepted_at = utc_now() accepted = replace( invitation, status=InvitationStatus.ACCEPTED, updated_at=accepted_at, accepted_at=accepted_at, version=invitation.version + 1, ) self.store.save_family_invitation(accepted) self._record_mutation( actor, action="family_invitation.accept", subject=accepted.invitation_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="family_invitation.accepted", aggregate_id=accepted.invitation_id, payload={ "invitation_id": accepted.invitation_id, "user_id": accepted.user_id, "tenant": accepted.tenant, "family_scope_id": accepted.family_scope_id, "application_id": accepted.application_id, "status": accepted.status, }, application_id=accepted.application_id, ) session = self._session(actor, self._require_user(accepted.user_id), account) identity_context = self.identity_context( actor, user_id=accepted.user_id, tenant=tenant_context.tenant, application_id=accepted.application_id, include_profile=True, correlation_id=correlation_id, ) claims_projection = self.projection( actor, accepted.user_id, ProjectionType.CLAIMS_ENRICHMENT, tenant=tenant_context.tenant, application_id=accepted.application_id, correlation_id=correlation_id, ) return FamilyInvitationAcceptance( session=session, invitation=accepted, identity_context=identity_context, claims_projection=claims_projection, ) def tenant_diagnostics( self, actor: Actor, *, tenant: str, correlation_id: str | None = None, ) -> TenantDiagnostics: tenant_context = self.resolve_tenant_context(actor, tenant) correlation_id = correlation_id or new_id("corr") self._authorize( actor, action="tenant.diagnostics.read", resource_type="user-engine:tenant", resource_id=tenant_context.tenant, tenant=tenant_context.tenant, correlation_id=correlation_id, ) memberships = self.store.memberships_for_tenant(tenant_context.tenant) checks = { "store_schema": self.store.ready, "authorization_port": self.authorization is not None, "membership_facts": bool(memberships), "catalog_scopes": bool(self._active_attribute_definitions()), "audit_ready": self.store.ready, } issues = tuple(key for key, passed in checks.items() if not passed) return TenantDiagnostics( tenant=tenant_context.tenant, checks=checks, issues=issues, memberships=memberships, ) def audit_records(self) -> tuple[AuditRecord, ...]: return self.store.audit_log() def outbox_events(self) -> tuple[OutboxEvent, ...]: return self.store.pending_outbox() def deliver_outbox( self, actor: Actor, deliver: Callable[[OutboxEvent], None], *, worker_id: str, max_attempts: int = 3, correlation_id: str | None = None, event_id: str | None = None, expected_attempts: int | None = None, ) -> tuple[OutboxEvent, ...]: """Serialize each external attempt and retain only classified failures.""" if not worker_id.strip() or not 1 <= max_attempts <= 20: raise ValidationError("invalid outbox worker or attempt limit") correlation_id = correlation_id or new_id("corr") self._authorize(actor, action="outbox.deliver", resource_type="user-engine:outbox", resource_id="pending", tenant=actor.tenant, correlation_id=correlation_id) selected = self.store.outbox_event(event_id) if event_id else None if event_id and selected is None: raise NotFoundError("delivery record not found") events = (selected,) if selected else self.store.pending_outbox() results = [] for candidate in events: with self.store.outbox_delivery_guard(candidate.event_id): event = self.store.outbox_event(candidate.event_id) if event is None or event.delivered_at is not None: continue if expected_attempts is not None and event.delivery_attempts != expected_attempts: raise ConflictError("Delivery changed. Refresh and review the attempt again.") # Owning this cross-process guard proves any previous claim is orphaned. # Receiver idempotency reconciles a lost receipt without blind resending. if event.dead_lettered_at and not event_id: continue claimed = replace(event, claimed_by=worker_id, claimed_at=utc_now(), delivery_attempts=event.delivery_attempts+1, failed_at=None, failure_reason=None) with self.store.transaction(): self.store.save_outbox(claimed) try: deliver(claimed) except Exception as exc: code = getattr(exc, "code", "delivery_unavailable") if code not in {"mail_unconfigured", "delivery_outcome_unknown", "temporary_deferral", "permanent_rejection", "recipient_suppressed", "provider_timeout", "provider_unavailable"}: code = "delivery_unavailable" outcome = replace(claimed, claimed_by=None, claimed_at=None, failed_at=utc_now(), failure_reason=code, dead_lettered_at=utc_now() if claimed.delivery_attempts >= max_attempts else None) else: outcome = replace(claimed, claimed_by=None, claimed_at=None, delivered_at=utc_now(), dead_lettered_at=None) with self.store.transaction(): self.store.save_outbox(outcome) results.append(outcome) return tuple(results) def replay_outbox(self, actor: Actor, event_id: str, *, correlation_id: str | None = None) -> OutboxEvent: correlation_id = correlation_id or new_id("corr") with self.store.outbox_delivery_guard(event_id): event = self.store.outbox_event(event_id) if event is None: raise NotFoundError("outbox event not found") self.resolve_tenant_context(actor, event.tenant) self._authorize(actor, action="outbox.replay", resource_type="user-engine:outbox-event", resource_id=event_id, tenant=event.tenant, correlation_id=correlation_id) if event.delivered_at or (event.claimed_by and not event.failed_at): raise ConflictError("Delivery is completed or its previous attempt is unresolved.") replayed = replace(event, claimed_by=None, claimed_at=None, failed_at=None, failure_reason=None, dead_lettered_at=None) with self.store.transaction(): self.store.save_outbox(replayed) return replayed def outbox_diagnostics(self) -> OutboxDiagnostics: event_types: dict[str, int] = {} pending = self.store.pending_outbox() for event in pending: event_types[event.event_type] = event_types.get(event.event_type, 0) + 1 oldest = pending[0].correlation_id if pending else None return OutboxDiagnostics( pending_count=len(pending), event_types=event_types, oldest_correlation_id=oldest, ) def operability_snapshot(self) -> OperabilitySnapshot: audit_correlation_ok = all( bool(record.correlation_id) for record in self.store.audit_log() ) checks = { "ready": self.readiness().ready, "audit_correlation": audit_correlation_ok, "outbox_diagnostics": self.outbox_diagnostics().pending_count >= 0, } metrics = self.store.record_counts() issues = tuple(key for key, passed in checks.items() if not passed) return OperabilitySnapshot( ready=all(checks.values()), checks=checks, metrics=metrics, issues=issues, ) def record_evidence_heartbeat( self, *, correlation_id: str | None = None ) -> OutboxEvent: """Emit a positive evidence claim that can itself go missing. Counts load-bearing denials and revocations already on the local trail. Does not claim the trail is complete. """ correlation_id = correlation_id or new_id("corr") counts = load_bearing_counts( self.store.audit_log(), self.store.pending_outbox() ) event = OutboxEvent( event_id=new_id("evt"), event_type=HEARTBEAT_EVENT_TYPE, aggregate_id="user-engine:evidence", payload={ "completeness_claimed": False, "form": "heartbeat", "bound": ( "the archive proves the records it holds were not " "altered or truncated after arrival" ), "load_bearing": counts, }, tenant=PLATFORM_TENANT, correlation_id=correlation_id, ) self.store.append_outbox(event) return event def structured_log_context( self, *, correlation_id: str, tenant: str, actor: Actor | None = None, ) -> Mapping[str, str]: context = { "correlation_id": correlation_id, "tenant": tenant, } if actor is not None: context["actor_issuer"] = actor.issuer context["actor_subject"] = actor.subject return context def _close_registration( self, actor: Actor, registration_id: str, *, status: RegistrationStatus, action: str, event_type: str, correlation_id: str | None, ) -> RegistrationSession: session = self._require_registration_session(registration_id) self._ensure_registration_mutable(session) tenant_context = self.resolve_tenant_context(actor, session.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action=action, resource_type="user-engine:registration", resource_id=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, ) now = utc_now() updated = replace( session, status=status, updated_at=now, abandoned_at=now if status == RegistrationStatus.ABANDONED else None, expired_at=now if status == RegistrationStatus.EXPIRED else None, rejected_at=now if status == RegistrationStatus.REJECTED else None, ) with self.store.transaction(): self.store.save_registration_session(updated) self._record_mutation( actor, action=action, subject=session.registration_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type=event_type, aggregate_id=session.registration_id, payload={ "registration_id": session.registration_id, "status": updated.status, }, ) return updated def _close_prepared_account( self, actor: Actor, prepared_account_id: str, *, status: PreparedAccountStatus, action: str, event_type: str, correlation_id: str | None, ) -> PreparedAccount: prepared = self._require_prepared_account(prepared_account_id) self._ensure_prepared_account_pending(prepared) tenant_context = self.resolve_tenant_context(actor, prepared.tenant) correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action=action, resource_type="user-engine:prepared-account", resource_id=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, ) now = utc_now() updated = replace( prepared, status=status, updated_at=now, revoked_at=now if status == PreparedAccountStatus.REVOKED else None, expired_at=now if status == PreparedAccountStatus.EXPIRED else None, ) with self.store.transaction(): self.store.save_prepared_account(updated) self._record_mutation( actor, action=action, subject=prepared.prepared_account_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type=event_type, aggregate_id=prepared.prepared_account_id, payload={ "prepared_account_id": prepared.prepared_account_id, "status": updated.status, }, ) return updated def _require_registration_session( self, registration_id: str ) -> RegistrationSession: session = self.store.registration_session(registration_id) if session is None: raise NotFoundError("registration session not found") return session def _ensure_registration_mutable(self, session: RegistrationSession) -> None: if session.status in _TERMINAL_REGISTRATION_STATUSES: raise ValidationError("registration session is already terminal") def _normalize_factor_verification( self, verification: FactorVerification | Mapping[str, Any] ) -> FactorVerification: if isinstance(verification, FactorVerification): return verification if self.factor_verifier is None: raise ValidationError("factor verifier adapter is required") return self.factor_verifier.normalize(verification) def _require_prepared_account(self, prepared_account_id: str) -> PreparedAccount: prepared = self.store.prepared_account(prepared_account_id) if prepared is None: raise NotFoundError("prepared account not found") return prepared def _ensure_prepared_account_pending(self, prepared: PreparedAccount) -> None: if prepared.status != PreparedAccountStatus.PENDING: raise ValidationError("prepared account is not pending") def _ensure_no_duplicate_prepared_account( self, tenant: str, requirements: tuple[PreparedFactorRequirement, ...], *, excluding_prepared_account_id: str | None, ) -> None: signature = _prepared_factor_signature(requirements) for prepared in self.store.prepared_accounts_for_tenant(tenant): if prepared.prepared_account_id == excluding_prepared_account_id: continue if prepared.status != PreparedAccountStatus.PENDING: continue if _prepared_account_expired(prepared): continue if _prepared_factor_signature(prepared.required_factor_matches) == signature: raise ConflictError("matching prepared account already exists") def _validate_prepared_factor_requirements( self, requirements: tuple[PreparedFactorRequirement, ...] ) -> None: if not requirements: raise ValidationError("prepared account requires at least one factor match") for requirement in requirements: if not requirement.normalized_value.strip(): raise ValidationError( "prepared account factor match requires normalized value" ) def _validate_prepared_entitlements( self, entitlements: tuple[PreparedEntitlement, ...], tenant: str ) -> None: for entitlement in entitlements: if entitlement.tenant != tenant: raise ValidationError("prepared entitlement tenant must match account") if entitlement.kind == PreparedEntitlementKind.TENANT_ACCOUNT: continue if entitlement.kind == PreparedEntitlementKind.MEMBERSHIP: if not entitlement.scope_type or not entitlement.scope_id: raise ValidationError("membership entitlement requires scope") if not entitlement.role: raise ValidationError("membership entitlement requires role") elif entitlement.kind == PreparedEntitlementKind.PROFILE_VALUE: if not entitlement.attribute_key: raise ValidationError("profile entitlement requires attribute_key") elif entitlement.kind == PreparedEntitlementKind.APPLICATION_BINDING: if entitlement.application_binding is None: raise ValidationError( "application binding entitlement requires binding" ) elif entitlement.kind == PreparedEntitlementKind.ONBOARDING_JOURNEY: if not entitlement.onboarding_journey: raise ValidationError( "onboarding journey entitlement requires journey" ) else: raise ValidationError(f"unsupported entitlement kind: {entitlement.kind}") def _resolve_prepared_account_claim( self, actor: Actor, session: RegistrationSession, factors: tuple[IdentityFactor, ...], *, prepared_account_id: str | None, tenant: str, correlation_id: str, ) -> PreparedAccount: candidates = ( (self._require_prepared_account(prepared_account_id),) if prepared_account_id is not None else self.store.prepared_accounts_for_tenant(tenant) ) matches = tuple( prepared for prepared in candidates if prepared.tenant == tenant and prepared.status == PreparedAccountStatus.PENDING and not _prepared_account_expired(prepared) and _prepared_account_matches_factors(prepared, factors) ) if prepared_account_id is not None and not matches: self._record_audit_only( actor, action="prepared_account.claim", subject=prepared_account_id, tenant=tenant, correlation_id=correlation_id, summary="prepared account claim denied: factor mismatch or closed", ) raise ValidationError("prepared account claim requirements are not met") if not matches: self._record_audit_only( actor, action="prepared_account.claim", subject=session.registration_id, tenant=tenant, correlation_id=correlation_id, summary="prepared account claim denied: no match", ) raise NotFoundError("no matching prepared account found") if len(matches) > 1: self._record_audit_only( actor, action="prepared_account.claim", subject=session.registration_id, tenant=tenant, correlation_id=correlation_id, summary="prepared account claim denied: ambiguous match", ) raise ConflictError("multiple prepared accounts match registration") return matches[0] def _prepared_entitlement_records( self, prepared: PreparedAccount, user_id: str, *, correlation_id: str, ) -> tuple[ tuple[TenantAccount, ...], tuple[Membership, ...], tuple[ProfileValue, ...], tuple[ApplicationBinding, ...], tuple[str, ...], ]: tenant_accounts: list[TenantAccount] = [] memberships: list[Membership] = [] profile_values: list[ProfileValue] = [] application_bindings: list[ApplicationBinding] = [] onboarding_journeys: list[str] = [] for entitlement in prepared.entitlements: if entitlement.kind == PreparedEntitlementKind.TENANT_ACCOUNT: tenant_accounts.append( TenantAccount( user_id=user_id, tenant=entitlement.tenant, status=entitlement.tenant_account_status, ) ) elif entitlement.kind == PreparedEntitlementKind.MEMBERSHIP: memberships.append( Membership( membership_id=new_id("mem"), user_id=user_id, tenant=entitlement.tenant, scope_type=entitlement.scope_type or "tenant", scope_id=entitlement.scope_id or entitlement.tenant, kind=entitlement.role or "member", source_system="prepared-account", freshness_version=correlation_id, ) ) elif entitlement.kind == PreparedEntitlementKind.PROFILE_VALUE: definition = self._require_attribute(entitlement.attribute_key or "") self._validate_profile_scope( definition, entitlement.profile_scope, entitlement.profile_scope_id, ) self._validate_value(definition, entitlement.value) profile_values.append( ProfileValue( user_id=user_id, attribute_key=entitlement.attribute_key or "", value=entitlement.value, scope=entitlement.profile_scope, scope_id=entitlement.profile_scope_id, source=f"prepared-account:{prepared.prepared_account_id}", ) ) elif entitlement.kind == PreparedEntitlementKind.APPLICATION_BINDING: if entitlement.application_binding is not None: if self.store.application( entitlement.application_binding.application_id ) is None: raise ValidationError( "application binding entitlement requires registered application" ) application_bindings.append(entitlement.application_binding) elif entitlement.kind == PreparedEntitlementKind.ONBOARDING_JOURNEY: if entitlement.onboarding_journey: onboarding_journeys.append(entitlement.onboarding_journey) return ( tuple(tenant_accounts), tuple(memberships), tuple(profile_values), tuple(application_bindings), tuple(onboarding_journeys), ) def _record_audit_only( self, actor: Actor, *, action: str, subject: str, tenant: str, correlation_id: str, summary: str, ) -> None: self.store.append_audit( AuditRecord( audit_id=new_id("aud"), actor=actor, action=action, subject=subject, tenant=tenant, correlation_id=correlation_id, summary=summary, ) ) def _require_welcome_protocol(self, protocol_id: str) -> WelcomeProtocol: protocol = self.store.welcome_protocol(protocol_id) if protocol is None: raise NotFoundError("welcome protocol not found") return protocol def _require_onboarding_journey(self, journey_id: str) -> OnboardingJourney: journey = self.store.onboarding_journey(journey_id) if journey is None: raise NotFoundError("onboarding journey not found") return journey def _validate_welcome_protocol( self, protocol: WelcomeProtocol, tenant: str ) -> None: if protocol.tenant != tenant: raise ValidationError("welcome protocol tenant must match context") if not protocol.name.strip(): raise ValidationError("welcome protocol name is required") if not protocol.steps: raise ValidationError("welcome protocol requires at least one step") step_keys = [step.step_key for step in protocol.steps] if len(step_keys) != len(set(step_keys)): raise ValidationError("welcome protocol step keys must be unique") for step in protocol.steps: if not step.step_key.strip() or not step.title.strip(): raise ValidationError("welcome protocol steps require key and title") if not step.subsystem.strip(): raise ValidationError("welcome protocol steps require subsystem") def _build_onboarding_journey( self, protocol: WelcomeProtocol, user_id: str, *, trigger_type: OnboardingTriggerType, source_id: str | None, source_event_type: str | None, correlation_id: str, ) -> OnboardingJourney: if trigger_type != OnboardingTriggerType.MANUAL and ( protocol.trigger_type != trigger_type ): raise ValidationError("welcome protocol trigger does not match journey") steps = tuple(_onboarding_step_from_template(step) for step in protocol.steps) steps = _activate_initial_onboarding_step(steps) status = _journey_status_from_steps(steps) return OnboardingJourney( tenant=protocol.tenant, user_id=user_id, protocol_id=protocol.protocol_id, trigger_type=trigger_type, steps=steps, status=status, source_id=source_id, source_event_type=source_event_type, journey_key=protocol.journey_key or protocol.prepared_journey, active_step_key=_active_onboarding_step_key(steps), correlation_id=correlation_id, ) def _auto_start_onboarding_journeys( self, actor: Actor, user_id: str, protocols: tuple[WelcomeProtocol, ...], *, trigger_type: OnboardingTriggerType, source_id: str, source_event_type: str, correlation_id: str, decision_id: str | None, ) -> tuple[OnboardingJourney, ...]: journeys = [] for protocol in protocols: if self._onboarding_journey_exists( user_id, protocol.protocol_id, source_id, ): continue journey = self._build_onboarding_journey( protocol, user_id, trigger_type=trigger_type, source_id=source_id, source_event_type=source_event_type, correlation_id=correlation_id, ) self.store.save_onboarding_journey(journey) self._record_onboarding_journey_started( actor, journey, decision_id=decision_id, correlation_id=correlation_id, ) journeys.append(journey) return tuple(journeys) def _record_onboarding_journey_started( self, actor: Actor, journey: OnboardingJourney, *, decision_id: str | None, correlation_id: str, ) -> None: self._record_mutation( actor, action="onboarding_journey.start", subject=journey.journey_id, tenant=journey.tenant, correlation_id=correlation_id, decision_id=decision_id, event_type="onboarding_journey.started", aggregate_id=journey.user_id, payload={ "journey_id": journey.journey_id, "user_id": journey.user_id, "protocol_id": journey.protocol_id, "trigger_type": journey.trigger_type, "status": journey.status, "source_id": journey.source_id, "source_event_type": journey.source_event_type, "active_step_key": journey.active_step_key, "lifecycle_gaps": tuple( step.lifecycle_gap for step in journey.steps if step.lifecycle_gap ), }, ) def _matching_welcome_protocols_for_registration( self, session: RegistrationSession ) -> tuple[WelcomeProtocol, ...]: factors = self.store.factors_for_registration(session.registration_id) factor_types = {factor.factor_type for factor in factors} return tuple( protocol for protocol in self.store.welcome_protocols_for_tenant(session.tenant) if protocol.trigger_type == OnboardingTriggerType.REGISTRATION and ( protocol.registration_status is None or protocol.registration_status == session.status ) and set(protocol.required_factor_types).issubset(factor_types) ) def _matching_welcome_protocols_for_prepared_account( self, prepared: PreparedAccount ) -> tuple[WelcomeProtocol, ...]: prepared_journeys = { entitlement.onboarding_journey for entitlement in prepared.entitlements if entitlement.onboarding_journey } return tuple( protocol for protocol in self.store.welcome_protocols_for_tenant(prepared.tenant) if protocol.trigger_type == OnboardingTriggerType.PREPARED_ACCOUNT and ( not protocol.prepared_journey or protocol.prepared_journey in prepared_journeys ) ) def _onboarding_journey_exists( self, user_id: str, protocol_id: str, source_id: str | None, ) -> bool: return any( journey.protocol_id == protocol_id and journey.source_id == source_id for journey in self.store.onboarding_journeys_for_user(user_id) ) def _transition_onboarding_step( self, actor: Actor, journey_id: str, step_key: str, *, status: OnboardingStepStatus, action: str, event_type: str, lifecycle_gap: str | None = None, correlation_id: str | None, ) -> OnboardingJourney: journey = self._require_onboarding_journey(journey_id) tenant_context = self.resolve_tenant_context(actor, journey.tenant) if journey.status in ( OnboardingJourneyStatus.COMPLETED, OnboardingJourneyStatus.SKIPPED, ): raise ValidationError("completed onboarding journeys cannot change") correlation_id = correlation_id or new_id("corr") decision = self._authorize( actor, action=action, resource_type="user-engine:onboarding-step", resource_id=f"{journey.journey_id}:{step_key}", tenant=tenant_context.tenant, correlation_id=correlation_id, target_user_id=journey.user_id, ) updated_steps = _replace_onboarding_step_status( journey.steps, step_key, status=status, lifecycle_gap=lifecycle_gap, ) updated = _replace_onboarding_journey_steps(journey, updated_steps) with self.store.transaction(): self.store.save_onboarding_journey(updated) self._record_mutation( actor, action=action, subject=updated.journey_id, tenant=tenant_context.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type=event_type, aggregate_id=updated.user_id, payload={ "journey_id": updated.journey_id, "user_id": updated.user_id, "step_key": step_key, "step_status": status, "journey_status": updated.status, "active_step_key": updated.active_step_key, "lifecycle_gap": lifecycle_gap, }, ) return updated def _require_access_profile(self, access_profile_id: str) -> AccessProfile: profile = self.store.access_profile(access_profile_id) if profile is None: raise NotFoundError("access profile not found") return profile def _validate_access_profile(self, profile: AccessProfile, tenant: str) -> None: if profile.tenant != tenant: raise ValidationError("access profile tenant must match context") if not profile.display_name.strip(): raise ValidationError("access profile display_name is required") if not profile.hat.strip(): raise ValidationError("access profile hat is required") if profile.scope_type != AccessScopeType.TENANT and not profile.scope_id: raise ValidationError("access profile scope_id is required") if not profile.membership_requirements: raise ValidationError( "access profile requires at least one membership requirement" ) for requirement in profile.membership_requirements: if ( not requirement.scope_type.strip() or not requirement.scope_id.strip() or not requirement.kind.strip() ): raise ValidationError("membership requirements must be complete") def _satisfying_access_memberships( self, profile: AccessProfile, memberships: tuple[Membership, ...], ) -> tuple[Membership, ...]: matched = [] for requirement in profile.membership_requirements: match = next( ( membership for membership in memberships if membership.scope_type == requirement.scope_type and membership.scope_id == requirement.scope_id and membership.kind == requirement.kind ), None, ) if match is None: raise ValidationError( "access profile membership requirements are not met" ) matched.append(match) return tuple(matched) def _verified_factor_ids_for_user( self, user_id: str, required_factor_types: tuple[IdentityFactorType, ...], ) -> tuple[str, ...]: if not required_factor_types: return () now = utc_now() factors = self.store.factors_for_user(user_id) factor_ids = [] for required_type in required_factor_types: match = next( ( factor for factor in factors if factor.factor_type == required_type and (factor.expires_at is None or factor.expires_at > now) ), None, ) if match is None: raise ValidationError("access profile factor requirements are not met") factor_ids.append(match.factor_id) return tuple(factor_ids) def _access_control_facts( self, tenant: str, *, user_id: str | None, memberships: tuple[Membership, ...], active_context: ActiveAccessContext | None, ) -> tuple[AccessControlFact, ...]: facts: list[AccessControlFact] = [] active_membership_ids = ( set(active_context.membership_ids) if active_context is not None else set() ) for membership in memberships: is_active = membership.membership_id in active_membership_ids facts.append( AccessControlFact( tenant=tenant, subject_type="group" if membership.scope_type == AccessScopeType.GROUP.value else "user", subject_id=membership.scope_id if membership.scope_type == AccessScopeType.GROUP.value else membership.user_id, user_id=membership.user_id, scope_type=membership.scope_type, scope_id=membership.scope_id, role=membership.kind, hat=active_context.hat if is_active and active_context else None, access_profile_id=active_context.access_profile_id if is_active and active_context else None, membership_id=membership.membership_id, evidence_refs=self._access_fact_evidence_refs( tenant, {membership.membership_id} ), ) ) if active_context is not None: evidence_refs = self._access_fact_evidence_refs( tenant, {active_context.active_context_id} ) facts.append( AccessControlFact( tenant=tenant, subject_type="user", subject_id=active_context.user_id, user_id=active_context.user_id, scope_type=active_context.scope_type.value, scope_id=active_context.scope_id, role=active_context.hat, hat=active_context.hat, access_profile_id=active_context.access_profile_id, evidence_refs=evidence_refs, ) ) for group_id in active_context.group_ids: facts.append( AccessControlFact( tenant=tenant, subject_type="group", subject_id=group_id, user_id=active_context.user_id, scope_type=active_context.scope_type.value, scope_id=active_context.scope_id, role=active_context.hat, hat=active_context.hat, access_profile_id=active_context.access_profile_id, evidence_refs=evidence_refs, ) ) return tuple(facts) def _access_fact_evidence_refs( self, tenant: str, subjects: set[str] ) -> tuple[CanonEntityReference, ...]: return tuple( CanonEntityReference( concept="Evidence Source", identifier=record.audit_id, scope=tenant, source_system="user-engine:audit", local_type=record.summary or record.action, ) for record in self.store.audit_log() if record.tenant == tenant and record.subject in subjects ) def _projection_access_context( self, user_id: str, tenant: str, *, application_id: str | None, ) -> Mapping[str, Any]: active_context = self.store.active_access_context(user_id, tenant) if active_context is None: return {} if ( application_id is not None and active_context.service_id is not None and active_context.service_id != application_id ): return {} return { "active_hat": active_context.hat, "access_profile_id": active_context.access_profile_id, "scope_type": active_context.scope_type.value, "scope_id": active_context.scope_id, "realm_id": active_context.realm_id, "service_id": active_context.service_id, "asset_id": active_context.asset_id, "group_ids": active_context.group_ids, "claims": dict(active_context.claims), "profile_defaults": dict(active_context.profile_defaults), "factor_ids": active_context.factor_ids, } def _ensure_actor_session( self, actor: Actor, correlation_id: str ) -> UserSession: identity = self.store.find_identity(*actor.identity_key) if identity is not None: user = self._require_user(identity.user_id) account = self._require_account(user.user_id) return self._session(actor, user, account) decision = self._authorize( actor, action="me.read", resource_type="user-engine:me", resource_id=actor.subject, tenant=actor.tenant, correlation_id=correlation_id, ) user = User( display_name=actor.preferred_username, primary_email=_optional_claim(actor, "email"), ) account = Account( account_id=new_id("acct"), user_id=user.user_id, status=AccountStatus.ACTIVE, ) tenant_account = TenantAccount(user_id=user.user_id, tenant=actor.tenant) external_identity = ExternalIdentity( identity_id=new_id("idn"), user_id=user.user_id, issuer=actor.issuer, subject=actor.subject, provider=actor.authorized_party, ) with self.store.transaction(): self.store.save_user(user) self.store.save_account(account) self.store.save_tenant_account(tenant_account) self.store.save_identity(external_identity) self._record_mutation( actor, action="user.create_from_identity", subject=user.user_id, tenant=actor.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="user.created", aggregate_id=user.user_id, payload={ "user_id": user.user_id, "account_id": account.account_id, "identity": {"issuer": actor.issuer, "subject": actor.subject}, }, ) return self._session(actor, user, account) def _ensure_family_dataspace_application( self, actor: Actor, request: FamilyDataspaceRequest, correlation_id: str, ) -> tuple[Application, ApplicationBinding]: desired_binding = _family_dataspace_binding(request) application = self.store.application(request.application_id) if application is not None: binding = self.store.binding(request.application_id) if binding is None: decision = self._authorize( actor, action="application.bind", resource_type="user-engine:application", resource_id=request.application_id, tenant=request.tenant, correlation_id=correlation_id, application_id=request.application_id, ) with self.store.transaction(): self.store.save_binding(desired_binding) self._record_mutation( actor, action="application.bind", subject=request.application_id, tenant=request.tenant, correlation_id=correlation_id, decision_id=decision.decision_id, event_type="application.bound", aggregate_id=request.application_id, payload={ "application_id": request.application_id, "catalog_namespaces": desired_binding.catalog_namespaces, }, application_id=request.application_id, ) binding = desired_binding return application, binding application = Application( application_id=request.application_id, display_name=request.application_display_name, owner=request.family_scope_id, allowed_profile_scopes=( ProfileScope.GLOBAL, ProfileScope.TENANT, ProfileScope.APPLICATION, ProfileScope.MEMBERSHIP, ), allowed_projection_types=( ProjectionType.APPLICATION_RUNTIME, ProjectionType.CLAIMS_ENRICHMENT, ), ) return ( self.register_application( actor, application, binding=desired_binding, correlation_id=correlation_id, ), desired_binding, ) def _ensure_family_dataspace_catalog( self, actor: Actor, request: FamilyDataspaceRequest, correlation_id: str, ) -> Catalog: catalog_id = f"{request.application_id}.profile" existing = self.store.catalog(catalog_id) if existing is not None: return existing catalog = Catalog( catalog_id=catalog_id, namespace=request.catalog_namespace, version="0.1.0", owning_application_id=request.application_id, lifecycle=CatalogLifecycle.ACTIVE, attributes=( AttributeDefinition( key=f"{request.catalog_namespace}.family_display_name", value_type="string", scope=ProfileScope.TENANT, sensitivity=Sensitivity.PERSONAL, visibility=( Visibility.USER, Visibility.ADMIN, Visibility.APPLICATION, ), mutability=(Mutability.ADMIN,), default=request.family_display_name, ), AttributeDefinition( key=f"{request.catalog_namespace}.member_display_name", value_type="string", scope=ProfileScope.APPLICATION, sensitivity=Sensitivity.PERSONAL, visibility=( Visibility.USER, Visibility.ADMIN, Visibility.APPLICATION, ), mutability=(Mutability.USER, Mutability.ADMIN), ), AttributeDefinition( key=f"{request.catalog_namespace}.home_view", value_type="string", scope=ProfileScope.APPLICATION, sensitivity=Sensitivity.INTERNAL, visibility=(Visibility.USER, Visibility.APPLICATION), mutability=(Mutability.USER,), default="family", validation={"enum": ["family", "personal"]}, ), ), ) return self.publish_catalog( actor, catalog, correlation_id=correlation_id, ) def _ensure_membership( self, actor: Actor, user_id: str, *, tenant: str, scope_type: str, scope_id: str, kind: str, correlation_id: str, ) -> Membership: for membership in self.store.memberships_for_user(user_id, tenant=tenant): if ( membership.scope_type == scope_type and membership.scope_id == scope_id and membership.kind == kind ): return membership return self.add_membership( actor, user_id, tenant=tenant, scope_type=scope_type, scope_id=scope_id, kind=kind, correlation_id=correlation_id, ) def _apply_family_profile_defaults( self, actor: Actor, user_id: str, *, tenant: str, application_id: str, catalog_namespace: str, values: Mapping[str, Any], correlation_id: str, ) -> None: for key, value in values.items(): self.set_profile_value( actor, user_id, _family_profile_key(catalog_namespace, key), value, scope=ProfileScope.APPLICATION, scope_id=application_id, tenant=tenant, application_id=application_id, correlation_id=correlation_id, ) def _require_family_invitation(self, invitation_id: str) -> FamilyInvitation: invitation = self.store.family_invitation(invitation_id) if invitation is None: raise NotFoundError("family invitation not found") return invitation def _identity_entity_refs( self, actor: Actor, user: User, account: Account, identities: tuple[ExternalIdentity, ...], memberships: tuple[Membership, ...], tenant: str, application_id: str | None, active_access_context: ActiveAccessContext | None, ) -> Mapping[str, CanonEntityReference]: actor_identifier = f"{actor.issuer}:{actor.subject}" refs: dict[str, CanonEntityReference] = { "actor": CanonEntityReference( concept="Actor", identifier=actor_identifier, scope=actor.tenant, source_system=actor.issuer, local_type=actor.principal_type.value, ), "user": CanonEntityReference( concept="User", identifier=user.user_id, scope=tenant, local_type="user-engine-user-record", ), "account": CanonEntityReference( concept="Account", identifier=account.account_id, scope=tenant, local_type=account.status.value, ), "authenticated_subject": CanonEntityReference( concept="Authenticated Subject", identifier=actor_identifier, scope=actor.tenant, source_system=actor.issuer, local_type="issuer-subject", ), "authorization_principal": CanonEntityReference( concept="Authorization Principal", identifier=f"{tenant}:{actor_identifier}", scope=tenant, local_type=actor.principal_type.value, ), "tenant": CanonEntityReference( concept="Tenant", identifier=tenant, scope=tenant, local_type="isolation-scope", ), } if application_id is not None: refs["application_scope"] = CanonEntityReference( concept="Scope", identifier=application_id, scope=tenant, local_type="application", ) if active_access_context is not None: refs["active_access_context"] = CanonEntityReference( concept="Active Access Context", identifier=active_access_context.active_context_id, scope=tenant, local_type=active_access_context.scope_type.value, ) refs["active_hat"] = CanonEntityReference( concept="Hat", identifier=active_access_context.hat, scope=tenant, local_type="active", ) refs["access_profile"] = CanonEntityReference( concept="Access Profile", identifier=active_access_context.access_profile_id, scope=tenant, local_type=active_access_context.hat, ) if active_access_context.realm_id is not None: refs["realm"] = CanonEntityReference( concept="Realm", identifier=active_access_context.realm_id, scope=tenant, local_type="realm", ) if active_access_context.service_id is not None: refs["service_area"] = CanonEntityReference( concept="Service Area", identifier=active_access_context.service_id, scope=tenant, local_type="service", ) if active_access_context.asset_id is not None: refs["asset_scope"] = CanonEntityReference( concept="Asset Scope", identifier=active_access_context.asset_id, scope=tenant, local_type="asset", ) for group_id in active_access_context.group_ids: refs[f"group:{group_id}"] = CanonEntityReference( concept="Group", identifier=group_id, scope=tenant, local_type="group", ) for identity in identities: refs[f"identity_record:{identity.identity_id}"] = CanonEntityReference( concept="Identity Record", identifier=identity.identity_id, scope=identity.issuer, source_system=identity.issuer, local_type=identity.provider, ) refs[f"identifier:{identity.identity_id}"] = CanonEntityReference( concept="Scoped Identifier", identifier=f"{identity.issuer}:{identity.subject}", scope=identity.issuer, source_system=identity.issuer, local_type="issuer-subject", ) for membership in memberships: scope_key = f"{membership.scope_type}:{membership.scope_id}" refs.setdefault( scope_key, CanonEntityReference( concept=_scope_concept(membership.scope_type), identifier=membership.scope_id, scope=tenant, local_type=membership.scope_type, ), ) refs[f"membership:{membership.membership_id}"] = CanonEntityReference( concept="Membership Relationship", identifier=membership.membership_id, scope=tenant, local_type=membership.kind, ) refs[f"role:{membership.membership_id}"] = CanonEntityReference( concept="Role", identifier=f"{membership.scope_id}:{membership.kind}", scope=membership.scope_id, local_type=membership.kind, ) return refs def _identity_relationship_refs( self, entity_refs: Mapping[str, CanonEntityReference], memberships: tuple[Membership, ...], tenant: str, evidence_id: str | None, *, active_access_context: ActiveAccessContext | None = None, ) -> tuple[tuple[CanonRelationshipReference, ...], tuple[CanonEntityReference, ...]]: account_ref = entity_refs["account"] subject_ref = entity_refs["authenticated_subject"] principal_ref = entity_refs["authorization_principal"] user_ref = entity_refs["user"] tenant_ref = entity_refs["tenant"] relationships = [ CanonRelationshipReference( relationship_type="belongs_to_tenant", source=user_ref, target=tenant_ref, scope=tenant, evidence_id=evidence_id, ), CanonRelationshipReference( relationship_type="authenticates_as", source=account_ref, target=subject_ref, scope=tenant, evidence_id=evidence_id, ), CanonRelationshipReference( relationship_type="evaluated_as", source=subject_ref, target=principal_ref, scope=tenant, evidence_id=evidence_id, ), ] grant_like_refs: list[CanonEntityReference] = [] for key, ref in entity_refs.items(): if key.startswith("identity_record:"): relationships.append( CanonRelationshipReference( relationship_type="identity_link", source=ref, target=user_ref, scope=ref.scope, evidence_id=evidence_id, ) ) for membership in memberships: membership_ref = entity_refs[f"membership:{membership.membership_id}"] role_ref = entity_refs[f"role:{membership.membership_id}"] scope_ref = entity_refs[f"{membership.scope_type}:{membership.scope_id}"] grant_ref = CanonEntityReference( concept="Access Grant", identifier=membership.membership_id, scope=tenant, local_type=membership.kind, evidence_id=evidence_id, ) grant_like_refs.append(grant_ref) relationships.extend( ( CanonRelationshipReference( relationship_type="member_of", source=user_ref, target=scope_ref, scope=tenant, evidence_id=evidence_id, ), CanonRelationshipReference( relationship_type="role_label", source=membership_ref, target=role_ref, scope=tenant, evidence_id=evidence_id, ), CanonRelationshipReference( relationship_type="scoped_to", source=grant_ref, target=tenant_ref, scope=tenant, evidence_id=evidence_id, ), ) ) if active_access_context is not None: active_ref = entity_refs["active_access_context"] hat_ref = entity_refs["active_hat"] profile_ref = entity_refs["access_profile"] relationships.extend( ( CanonRelationshipReference( relationship_type="wears_hat", source=user_ref, target=hat_ref, scope=tenant, evidence_id=evidence_id, ), CanonRelationshipReference( relationship_type="selected_access_profile", source=active_ref, target=profile_ref, scope=tenant, evidence_id=evidence_id, ), ) ) return tuple(relationships), tuple(grant_like_refs) def _identity_evidence_refs( self, user: User, account: Account, memberships: tuple[Membership, ...], tenant: str, *, active_access_context: ActiveAccessContext | None = None, ) -> tuple[CanonEntityReference, ...]: membership_ids = {membership.membership_id for membership in memberships} access_subjects = ( {active_access_context.active_context_id} if active_access_context is not None else set() ) subjects = {user.user_id, account.account_id, *membership_ids, *access_subjects} evidence = [] for record in self.store.audit_log(): if record.tenant != tenant: continue if record.subject not in subjects and record.summary != "membership.added": continue evidence.append( CanonEntityReference( concept="Evidence Source", identifier=record.audit_id, scope=tenant, source_system="user-engine:audit", local_type=record.summary or record.action, ) ) return tuple(evidence) def _session(self, actor: Actor, user: User, account: Account) -> UserSession: identities = self.store.identities_for_user(user.user_id) return UserSession(actor=actor, user=user, account=account, identities=identities) def _authorize( self, actor: Actor, *, action: str, resource_type: str, resource_id: str, tenant: str, correlation_id: str, application_id: str | None = None, target_user_id: str | None = None, context: Mapping[str, Any] | None = None, ): request = AuthorizationRequest( actor=actor, resource_type=resource_type, resource_id=resource_id, action=action, tenant=tenant, correlation_id=correlation_id, application_id=application_id, target_user_id=target_user_id, context=self._authorization_context( actor, target_user_id=target_user_id, context=context, tenant=tenant, ), ) decision = self.authorization.check(request) if decision.allowed: if decision.lifetime is None: self._record_authorization_denial( actor, action=action, resource_id=resource_id, tenant=tenant, correlation_id=correlation_id, application_id=application_id, decision=AuthorizationDecision( effect=decision.effect, reason="authorization allow has no lifetime", ), summary="authorization denied (allow has no lifetime)", ) raise AuthorizationDenied("authorization allow has no lifetime") if decision.expired(): self._record_authorization_denial( actor, action=action, resource_id=resource_id, tenant=tenant, correlation_id=correlation_id, application_id=application_id, decision=decision, summary="authorization denied (allow expired)", ) raise AuthorizationDenied("authorization allow expired") return decision summary = "authorization denied" if decision.stance_applied: summary = f"authorization denied (stance {decision.stance_applied})" self._record_authorization_denial( actor, action=action, resource_id=resource_id, tenant=tenant, correlation_id=correlation_id, application_id=application_id, decision=decision, summary=summary, ) raise AuthorizationDenied(decision.reason or "authorization denied") def _record_authorization_denial( self, actor: Actor, *, action: str, resource_id: str, tenant: str, correlation_id: str, application_id: str | None, decision: AuthorizationDecision, summary: str, ) -> None: self.store.append_audit( AuditRecord( audit_id=new_id("aud"), actor=actor, action=action, subject=resource_id, tenant=tenant, correlation_id=correlation_id, decision_id=decision.decision_id, application_id=application_id, summary=summary, ) ) def _record_mutation( self, actor: Actor, *, action: str, subject: str, tenant: str, correlation_id: str, decision_id: str | None, event_type: str, aggregate_id: str, payload: Mapping[str, Any], application_id: str | None = None, ) -> None: self.store.append_audit( AuditRecord( audit_id=new_id("aud"), actor=actor, action=action, subject=subject, tenant=tenant, correlation_id=correlation_id, decision_id=decision_id, application_id=application_id, summary=event_type, ) ) self.store.append_outbox( OutboxEvent( event_id=new_id("evt"), event_type=event_type, aggregate_id=aggregate_id, payload=dict(payload), tenant=tenant, correlation_id=correlation_id, ) ) def _resolve_effective_profile( self, user_id: str, application_id: str | None, tenant: str | None ) -> EffectiveProfile: self._require_user(user_id) values: dict[str, Any] = {} source_layers: dict[str, str] = {} trace: dict[str, list[str]] = {} definitions = self._active_attribute_definitions() for key, definition in definitions.items(): if definition.default is not None: values[key] = definition.default source_layers[key] = "catalog_default" trace[key] = ["catalog_default"] profile_values = self.store.values_for_user(user_id) for profile_value in profile_values: if profile_value.attribute_key not in definitions: continue if profile_value.scope != ProfileScope.GLOBAL: continue values[profile_value.attribute_key] = profile_value.value source_layers[profile_value.attribute_key] = "global" trace.setdefault(profile_value.attribute_key, []).append("global") for profile_value in profile_values: if profile_value.attribute_key not in definitions: continue if profile_value.scope != ProfileScope.TENANT: continue if profile_value.scope_id != tenant: continue values[profile_value.attribute_key] = profile_value.value layer = f"tenant:{tenant}" source_layers[profile_value.attribute_key] = layer trace.setdefault(profile_value.attribute_key, []).append(layer) for profile_value in profile_values: if profile_value.attribute_key not in definitions: continue if profile_value.scope != ProfileScope.APPLICATION: continue if profile_value.scope_id != application_id: continue values[profile_value.attribute_key] = profile_value.value layer = f"application:{application_id}" source_layers[profile_value.attribute_key] = layer trace.setdefault(profile_value.attribute_key, []).append(layer) return EffectiveProfile( user_id=user_id, application_id=application_id, values=values, source_layers=source_layers, trace={key: tuple(layers) for key, layers in trace.items()}, ) def _authorization_context( self, actor: Actor, *, target_user_id: str | None, context: Mapping[str, Any] | None, tenant: str, ) -> Mapping[str, Any]: facts: dict[str, Any] = { "actor_roles": actor.roles, "actor_scopes": actor.scopes, "assurance": dict(actor.assurance), } if target_user_id is not None: memberships = self.store.memberships_for_user(target_user_id, tenant=tenant) facts["target_memberships"] = tuple( { "membership_id": membership.membership_id, "scope_type": membership.scope_type, "scope_id": membership.scope_id, "kind": membership.kind, "freshness_version": membership.freshness_version, } for membership in memberships ) active_context = self.store.active_access_context(target_user_id, tenant) if active_context is not None: facts["active_access_context"] = { "active_context_id": active_context.active_context_id, "access_profile_id": active_context.access_profile_id, "hat": active_context.hat, "scope_type": active_context.scope_type.value, "scope_id": active_context.scope_id, "realm_id": active_context.realm_id, "service_id": active_context.service_id, "asset_id": active_context.asset_id, "group_ids": active_context.group_ids, } if context: facts.update(context) return facts def _operation_tenant( self, actor: Actor, tenant: str | None, scope: ProfileScope, scope_id: str | None, ) -> str: if scope == ProfileScope.TENANT: return self.resolve_tenant_context(actor, scope_id).tenant return self.resolve_tenant_context(actor, tenant).tenant def _validate_catalog(self, catalog: Catalog) -> None: if self.store.application(catalog.owning_application_id) is None: raise ValidationError("catalog owning application is not registered") if catalog.lifecycle != CatalogLifecycle.ACTIVE: raise ValidationError("only active catalogs can be published") if not catalog.attributes: raise ValidationError("catalog must define at least one attribute") keys = [attribute.key for attribute in catalog.attributes] if len(keys) != len(set(keys)): raise ValidationError("catalog attribute keys must be unique") binding = self.store.binding(catalog.owning_application_id) if binding and catalog.namespace not in binding.catalog_namespaces: raise ValidationError("catalog namespace is not bound to application") existing_catalog = self.store.catalog(catalog.catalog_id) if existing_catalog is not None: self._validate_catalog_update(existing_catalog, catalog) active_definitions = self._active_attribute_definitions( excluding_catalog_id=catalog.catalog_id ) active_catalogs = ( existing for existing in self.store.all_catalogs() if existing.catalog_id != catalog.catalog_id and existing.lifecycle == CatalogLifecycle.ACTIVE ) for existing in active_catalogs: if ( existing.namespace == catalog.namespace and existing.owning_application_id != catalog.owning_application_id ): raise ConflictError("active catalog namespace is already owned") for attribute in catalog.attributes: if not attribute.key.startswith(f"{catalog.namespace}."): raise ValidationError("attribute keys must use the catalog namespace") if attribute.key in active_definitions: raise ConflictError("active catalog already owns attribute key") if not attribute.visibility: raise ValidationError("attribute must declare visibility") if not attribute.mutability: raise ValidationError("attribute must declare mutability") if Mutability.READ_ONLY in attribute.mutability and len(attribute.mutability) > 1: raise ValidationError("read-only mutability cannot be combined") if attribute.default is not None: self._validate_value(attribute, attribute.default) def _validate_catalog_update(self, previous: Catalog, catalog: Catalog) -> None: if previous.namespace != catalog.namespace: raise ValidationError("catalog namespace cannot change") if previous.owning_application_id != catalog.owning_application_id: raise ValidationError("catalog owner cannot change") if _version_tuple(catalog.version) < _version_tuple(previous.version): raise ValidationError("catalog version cannot move backwards") previous_attributes = { attribute.key: attribute for attribute in previous.attributes } for attribute in catalog.attributes: previous_attribute = previous_attributes.get(attribute.key) if previous_attribute is None: continue if _sensitivity_rank(attribute.sensitivity) < _sensitivity_rank( previous_attribute.sensitivity ): raise ValidationError("catalog update cannot downgrade sensitivity") def _validate_profile_scope( self, definition: AttributeDefinition, scope: ProfileScope, scope_id: str | None, ) -> None: allowed_scopes = { ProfileScope.GLOBAL, ProfileScope.TENANT, ProfileScope.APPLICATION, definition.scope, } if scope not in allowed_scopes: raise ValidationError("profile value scope is not allowed for attribute") if scope == ProfileScope.APPLICATION: if scope_id is None: raise ValidationError("application profile values require scope_id") if self.store.application(scope_id) is None: raise ValidationError("application profile scope_id is not registered") elif scope == ProfileScope.TENANT: if scope_id is None: raise ValidationError("tenant profile values require scope_id") elif scope_id is not None: raise ValidationError( "only tenant or application scoped values may set scope_id" ) def _validate_value(self, definition: AttributeDefinition, value: Any) -> None: if value is None: return validators = { "string": lambda item: isinstance(item, str), "integer": lambda item: isinstance(item, int) and not isinstance(item, bool), "number": lambda item: isinstance(item, (int, float)) and not isinstance(item, bool), "boolean": lambda item: isinstance(item, bool), "object": lambda item: isinstance(item, Mapping), "array": lambda item: isinstance(item, (list, tuple)), } validator = validators.get(definition.value_type) if validator is None: raise ValidationError(f"unsupported attribute value_type: {definition.value_type}") if not validator(value): raise ValidationError(f"{definition.key} must be {definition.value_type}") enum_values = definition.validation.get("enum") if enum_values is not None and value not in enum_values: raise ValidationError(f"{definition.key} is not an allowed value") def _require_user(self, user_id: str) -> User: user = self.store.user(user_id) if user is None: raise NotFoundError("user not found") return user def _require_account(self, user_id: str) -> Account: account = self.store.user_account(user_id) if account is None: raise NotFoundError("account not found") return account def _require_attribute(self, attribute_key: str) -> AttributeDefinition: definition = self._active_attribute_definitions().get(attribute_key) if definition is None: raise NotFoundError("active attribute definition not found") return definition def _active_attribute_definitions( self, *, excluding_catalog_id: str | None = None, owning_application_id: str | None = None, ) -> dict[str, AttributeDefinition]: definitions: dict[str, AttributeDefinition] = {} for catalog in self.store.all_catalogs(): if catalog.catalog_id == excluding_catalog_id: continue if ( owning_application_id is not None and catalog.owning_application_id != owning_application_id ): continue if catalog.lifecycle != CatalogLifecycle.ACTIVE: continue for attribute in catalog.attributes: definitions[attribute.key] = attribute return dict(sorted(definitions.items())) def _optional_claim(actor: Actor, key: str) -> str | None: value = actor.claims.get(key) if value is None: return None return str(value) def _factor_type(value: IdentityFactorType | str) -> IdentityFactorType: try: return IdentityFactorType(str(value)) except ValueError as exc: raise ValidationError(f"unsupported identity factor type: {value}") from exc def _missing_required_factor_types( session: RegistrationSession, factors: Iterable[IdentityFactor] ) -> tuple[IdentityFactorType, ...]: verified = {factor.factor_type for factor in factors} return tuple( factor_type for factor_type in session.required_factor_types if factor_type not in verified ) def _registration_primary_email( factors: Iterable[IdentityFactor], actor: Actor ) -> str | None: for factor in factors: if factor.factor_type == IdentityFactorType.EMAIL: return factor.display_value or factor.normalized_value return _optional_claim(actor, "email") def _prepared_factor_signature( requirements: Iterable[PreparedFactorRequirement], ) -> tuple[tuple[str, str], ...]: return tuple( sorted( ( requirement.factor_type.value, requirement.normalized_value.casefold(), ) for requirement in requirements ) ) def _prepared_account_expired(prepared: PreparedAccount) -> bool: return prepared.expires_at is not None and prepared.expires_at <= utc_now() def _prepared_account_matches_factors( prepared: PreparedAccount, factors: Iterable[IdentityFactor] ) -> bool: now = utc_now() factor_values = { (factor.factor_type, factor.normalized_value.casefold()) for factor in factors if factor.expires_at is None or factor.expires_at > now } return all( ( requirement.factor_type, requirement.normalized_value.casefold(), ) in factor_values for requirement in prepared.required_factor_matches ) def _access_profile_scope_id(profile: AccessProfile) -> str: if profile.scope_id: return profile.scope_id if profile.scope_type == AccessScopeType.TENANT: return profile.tenant raise ValidationError("access profile scope_id is required") def _active_group_ids( profile: AccessProfile, memberships: Iterable[Membership] ) -> tuple[str, ...]: group_ids = { membership.scope_id for membership in memberships if membership.scope_type == AccessScopeType.GROUP.value } group_ids.update(profile.group_scope_ids) return tuple(sorted(group_ids)) def _onboarding_step_from_template(template: WelcomeProtocolStep) -> OnboardingStep: lifecycle_gap = ( f"subsystem-callback-missing:{template.subsystem}:{template.step_key}" if template.requires_subsystem_callback and not template.callback_ref else None ) initial_status = ( OnboardingStepStatus.BLOCKED if lifecycle_gap is not None else OnboardingStepStatus.PENDING ) task = OnboardingTask( subsystem=template.subsystem, task_kind=template.task_kind, external_ref=template.task_ref, status=initial_status, ) handoff = SubsystemHandoff( subsystem=template.subsystem, callback_ref=template.callback_ref, lifecycle_gap=lifecycle_gap, status=initial_status, ) return OnboardingStep( step_key=template.step_key, title=template.title, subsystem=template.subsystem, status=initial_status, task=task, handoff=handoff, support_ref=template.support_ref, lifecycle_gap=lifecycle_gap, ) def _activate_initial_onboarding_step( steps: tuple[OnboardingStep, ...] ) -> tuple[OnboardingStep, ...]: if any(step.status == OnboardingStepStatus.BLOCKED for step in steps): return steps for index, step in enumerate(steps): if step.status == OnboardingStepStatus.PENDING: return ( *steps[:index], _onboarding_step_with_status(step, OnboardingStepStatus.IN_PROGRESS), *steps[index + 1 :], ) return steps def _replace_onboarding_step_status( steps: tuple[OnboardingStep, ...], step_key: str, *, status: OnboardingStepStatus, lifecycle_gap: str | None, ) -> tuple[OnboardingStep, ...]: updated = [] found = False for step in steps: if step.step_key != step_key: updated.append(step) continue found = True updated.append( _onboarding_step_with_status( step, status, lifecycle_gap=lifecycle_gap, ) ) if not found: raise NotFoundError("onboarding step not found") return _activate_next_onboarding_step(tuple(updated)) def _activate_next_onboarding_step( steps: tuple[OnboardingStep, ...] ) -> tuple[OnboardingStep, ...]: if any( step.status in (OnboardingStepStatus.IN_PROGRESS, OnboardingStepStatus.BLOCKED) for step in steps ): return steps for index, step in enumerate(steps): if step.status == OnboardingStepStatus.PENDING: return ( *steps[:index], _onboarding_step_with_status(step, OnboardingStepStatus.IN_PROGRESS), *steps[index + 1 :], ) return steps def _resume_onboarding_steps( steps: tuple[OnboardingStep, ...], callback_refs: Mapping[str, str] ) -> tuple[OnboardingStep, ...]: updated = [] for step in steps: if step.status not in ( OnboardingStepStatus.BLOCKED, OnboardingStepStatus.FAILED, ): updated.append(step) continue callback_ref = callback_refs.get(step.step_key) handoff = step.handoff if handoff is not None and callback_ref is not None: handoff = replace( handoff, callback_ref=callback_ref, lifecycle_gap=None, status=OnboardingStepStatus.IN_PROGRESS, ) updated.append( _onboarding_step_with_status( step, OnboardingStepStatus.IN_PROGRESS, lifecycle_gap=None, handoff=handoff, ) ) return tuple(updated) def _onboarding_step_with_status( step: OnboardingStep, status: OnboardingStepStatus, *, lifecycle_gap: str | None = None, handoff: SubsystemHandoff | None = None, ) -> OnboardingStep: task = replace(step.task, status=status) if step.task is not None else None resolved_handoff = handoff if handoff is not None else step.handoff if resolved_handoff is not None: resolved_handoff = replace( resolved_handoff, status=status, lifecycle_gap=lifecycle_gap, ) return replace( step, status=status, task=task, handoff=resolved_handoff, lifecycle_gap=lifecycle_gap, updated_at=utc_now(), ) def _replace_onboarding_journey_steps( journey: OnboardingJourney, steps: tuple[OnboardingStep, ...] ) -> OnboardingJourney: now = utc_now() status = _journey_status_from_steps(steps) return replace( journey, steps=steps, status=status, active_step_key=_active_onboarding_step_key(steps), updated_at=now, completed_at=now if status == OnboardingJourneyStatus.COMPLETED else None, failed_at=now if status == OnboardingJourneyStatus.FAILED else journey.failed_at, skipped_at=now if status == OnboardingJourneyStatus.SKIPPED else None, ) def _journey_status_from_steps( steps: tuple[OnboardingStep, ...] ) -> OnboardingJourneyStatus: if any(step.status == OnboardingStepStatus.FAILED for step in steps): return OnboardingJourneyStatus.FAILED if any(step.status == OnboardingStepStatus.BLOCKED for step in steps): return OnboardingJourneyStatus.BLOCKED if all(step.status == OnboardingStepStatus.SKIPPED for step in steps): return OnboardingJourneyStatus.SKIPPED if all( step.status in (OnboardingStepStatus.COMPLETED, OnboardingStepStatus.SKIPPED) for step in steps ): return OnboardingJourneyStatus.COMPLETED if any(step.status == OnboardingStepStatus.IN_PROGRESS for step in steps): return OnboardingJourneyStatus.IN_PROGRESS return OnboardingJourneyStatus.PENDING def _active_onboarding_step_key(steps: tuple[OnboardingStep, ...]) -> str | None: for status in (OnboardingStepStatus.IN_PROGRESS, OnboardingStepStatus.BLOCKED): for step in steps: if step.status == status: return step.step_key return None def _scope_concept(scope_type: str) -> str: concepts = { "team": "Team", "tenant": "Tenant", "application": "Scope", "family": "Group", "group": "Group", "realm": "Realm", "service": "Service Area", "asset": "Asset Scope", } return concepts.get(scope_type, "Scope") def _family_dataspace_binding( request: FamilyDataspaceRequest, ) -> ApplicationBinding: event_source = request.event_source or request.application_id return ApplicationBinding( application_id=request.application_id, oidc_client_id=request.oidc_client_id, protected_system_id=request.protected_system_id, catalog_namespaces=(request.catalog_namespace,), event_source=event_source, deployment_ref=request.deployment_ref, ) def _family_profile_key(catalog_namespace: str, key: str) -> str: if key.startswith(f"{catalog_namespace}."): return key return f"{catalog_namespace}.{key}" def _family_role_value(role: FamilyRole | str) -> str: if str(role) in {"user", TENANT_ADMIN_ROLE}: return str(role) try: return FamilyRole(str(role)).value except ValueError as exc: raise ValidationError(f"unsupported family role: {role}") from exc def _visible_in_projection( definition: AttributeDefinition, projection_type: ProjectionType ) -> bool: if projection_type == ProjectionType.AUDIT: return True visibility = { ProjectionType.SELF_SERVICE: Visibility.USER, ProjectionType.ADMIN: Visibility.ADMIN, ProjectionType.APPLICATION_RUNTIME: Visibility.APPLICATION, ProjectionType.AGENT_CONTEXT: Visibility.APPLICATION, ProjectionType.CLAIMS_ENRICHMENT: Visibility.APPLICATION, }.get(projection_type) if visibility is None: return False return visibility in definition.visibility def _is_application_projection(projection_type: ProjectionType) -> bool: return projection_type in { ProjectionType.APPLICATION_RUNTIME, ProjectionType.AGENT_CONTEXT, ProjectionType.CLAIMS_ENRICHMENT, } def _must_redact( definition: AttributeDefinition, projection_type: ProjectionType ) -> bool: if projection_type in { ProjectionType.ADMIN, ProjectionType.AUDIT, ProjectionType.SELF_SERVICE, }: return False return definition.sensitivity in {Sensitivity.SENSITIVE, Sensitivity.SECRET} def _version_tuple(version: str) -> tuple[int, int, int]: parts = version.split(".") if len(parts) != 3 or any(not part.isdigit() for part in parts): raise ValidationError("catalog version must be semantic MAJOR.MINOR.PATCH") return tuple(int(part) for part in parts) def _sensitivity_rank(sensitivity: Sensitivity) -> int: return { Sensitivity.PUBLIC: 0, Sensitivity.INTERNAL: 1, Sensitivity.PERSONAL: 2, Sensitivity.SENSITIVE: 3, Sensitivity.SECRET: 4, }[sensitivity]