"""project_auth domain — business logic. Core owns: - Auth config (providers, JWT settings, Google OAuth mode + BYOC credentials) - RS256 key pairs (private key encrypted, public key pushed to Cloud KV) - Custom role definitions (synced to Cloud via auth-config push) Cloud owns app users, sessions, and audit events — Core never touches those. """ from __future__ import annotations import secrets from typing import Optional from uuid import UUID import structlog from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from src.core.exceptions import ConflictError, NotFoundError from src.domain.project_auth.permissions import ( ADMIN_PRESET, MEMBER_PRESET, OWNER_PRESET, STAFF_PRESET, validate_role_permissions, ) from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.project_auth.models import AuthAssignmentTier, AuthRoleCreator, ProjectAuthConfig from src.domain.project_auth.auth_contract import authz_policy_fingerprint, normalize_auth_contract from src.domain.project_auth.repository import ( ProjectAuthConfigRepository, ProjectAuthKeyRepository, ProjectAuthRoleRepository, ) from src.domain.project_auth.schemas import ( AuthConfigResponse, AuthKeyResponse, AuthRoleResponse, CreateAuthRoleRequest, RotateKeyResponse, RevokeAppUserRequest, UpdateAuthRoleRequest, UpsertAuthConfigRequest, ) from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() SUPER_ADMIN_ROLE = "super_admin" STAFF_ADMIN_ROLE = "admin" CUSTOMER_MEMBER_ROLE = "member" STAFF_ROLE = "staff" class ProjectAuthService: def __init__( self, *, config_repo: ProjectAuthConfigRepository, key_repo: ProjectAuthKeyRepository, role_repo: ProjectAuthRoleRepository, cloud_client: CloudControlClient, ) -> None: self._config = config_repo self._keys = key_repo self._roles = role_repo self._cloud = cloud_client # ------------------------------------------------------------------ # Auth Config # ------------------------------------------------------------------ async def upsert_config( self, project_id: UUID, body: UpsertAuthConfigRequest ) -> AuthConfigResponse: row = await self._config.get_by_project(project_id) fields: dict = {} if body.enabled is not None: fields["enabled"] = body.enabled if body.providers is not None: fields["providers"] = list(body.providers) if body.auth_methods is not None: fields["auth_methods"] = normalize_auth_contract(auth_methods=body.auth_methods)["auth_methods"] elif row is None and body.providers is not None: fields.update(normalize_auth_contract(providers=list(body.providers))) if body.identity_channels is not None: fields["identity_channels"] = list(body.identity_channels) if body.auth_connections is not None: fields["auth_connections"] = list(body.auth_connections) if body.audiences is not None: fields["audiences"] = list(body.audiences) if body.mfa_methods is not None: fields["mfa_methods"] = list(dict.fromkeys(body.mfa_methods)) if body.require_mfa is not None: fields["require_mfa"] = body.require_mfa if body.allow_signup is not None: fields["allow_signup"] = body.allow_signup if body.require_email_verify is not None: fields["require_email_verify"] = body.require_email_verify if body.google_oauth_mode is not None: fields["google_oauth_mode"] = body.google_oauth_mode if body.google_client_id is not None: fields["google_client_id_enc"] = encrypt_value(body.google_client_id) if body.google_client_secret is not None: fields["google_client_secret_enc"] = encrypt_value(body.google_client_secret) if body.access_token_ttl_seconds is not None: fields["access_token_ttl_seconds"] = body.access_token_ttl_seconds if body.refresh_token_ttl_days is not None: fields["refresh_token_ttl_days"] = body.refresh_token_ttl_days if body.password_min_length is not None: fields["password_min_length"] = body.password_min_length if row: row = await self._config.update(row, **fields) else: row = await self._config.create(project_id, **fields) await self._push_auth_config(project_id, row) return _config_response(row) async def get_config(self, project_id: UUID) -> AuthConfigResponse: row = await self._config.get_by_project(project_id) if not row: raise NotFoundError("Auth not configured for this project") return _config_response(row) # ------------------------------------------------------------------ # RS256 Keys # ------------------------------------------------------------------ async def rotate_key(self, project_id: UUID) -> RotateKeyResponse: """Generate a new RS256 key pair and push public key to Cloud. The previous active key is marked 'rotated' (not deleted — old tokens referencing it will expire naturally). """ previous_key = await self._keys.get_active_for_project(project_id) previous_kid: Optional[str] = None if previous_key: await self._keys.mark_rotated(previous_key) previous_kid = previous_key.key_id kid, private_pem, public_pem = _generate_rsa_key_pair() private_enc = encrypt_value(private_pem) new_key = await self._keys.create( project_id=project_id, key_id=kid, public_key_pem=public_pem, private_key_enc=private_enc, ) await self._push_auth_keys(project_id, kid, public_pem, private_pem) logger.info("auth_key_rotated", project_id=str(project_id), new_kid=kid, prev_kid=previous_kid) return RotateKeyResponse( new_key=_key_response(new_key), previous_key_id=previous_kid, ) async def list_keys(self, project_id: UUID) -> list[AuthKeyResponse]: rows = await self._keys.list_for_project(project_id) return [_key_response(r) for r in rows] # ------------------------------------------------------------------ # Roles # ------------------------------------------------------------------ async def list_roles(self, project_id: UUID) -> list[AuthRoleResponse]: rows = await self._roles.list_for_project(project_id) return [_role_response(r) for r in rows] async def create_role( self, project_id: UUID, body: CreateAuthRoleRequest ) -> AuthRoleResponse: existing = await self._roles.get_by_name(project_id, body.name) if existing: raise ConflictError(f"Role {body.name!r} already exists for this project") permissions = validate_role_permissions( body.permissions, protected=False, system_created=False ) if body.is_default: await self._roles.clear_default_flag(project_id) row = await self._roles.create( project_id=project_id, name=body.name, description=body.description, is_default=body.is_default, permissions_json=permissions, created_by=AuthRoleCreator.user, assignment_tier=body.assignment_tier, ) await self.push_config_with_roles(project_id) return _role_response(row) async def update_role( self, project_id: UUID, role_name: str, body: UpdateAuthRoleRequest ) -> AuthRoleResponse: row = await self._roles.get_by_name(project_id, role_name) if not row: raise NotFoundError(f"Role {role_name!r} not found") if row.protected: raise ConflictError(f"Protected role {role_name!r} cannot be modified") fields: dict = {} if body.description is not None: fields["description"] = body.description if body.assignment_tier is not None: if row.assignment_tier is not None and row.assignment_tier != body.assignment_tier: raise ConflictError(f"Role {role_name!r} assignment tier is fixed once set") fields["assignment_tier"] = body.assignment_tier if body.permissions is not None: fields["permissions_json"] = validate_role_permissions( body.permissions, protected=False, system_created=row.created_by == AuthRoleCreator.codegen, ) if body.is_default is not None: if body.is_default: await self._roles.clear_default_flag(project_id) fields["is_default"] = body.is_default row = await self._roles.update(row, **fields) await self.push_config_with_roles(project_id) return _role_response(row) async def delete_role(self, project_id: UUID, role_name: str) -> None: row = await self._roles.get_by_name(project_id, role_name) if not row: raise NotFoundError(f"Role {role_name!r} not found") if row.protected: raise ConflictError(f"Protected role {role_name!r} cannot be deleted") await self._roles.delete(row) await self.push_config_with_roles(project_id) # ------------------------------------------------------------------ # Auto-provisioning # ------------------------------------------------------------------ async def ensure_default_auth(self, project_id: UUID) -> bool: """Auto-provision default auth config and RS256 key pair if neither exists. Idempotent — returns True on first provision, False when already configured. Auth is a required runtime capability for generated applications. A Core row without its Cloud projection is therefore not a successful provision: push failures are deliberately propagated so the setup step and capability status remain blocked instead of allowing a deploy that cannot serve auth traffic. """ existing = await self._config.get_by_project(project_id) created = False if existing: # Config exists; generate a key only if the project somehow has none. active_key = await self._keys.get_active_for_project(project_id) if not active_key: kid, private_pem, public_pem = _generate_rsa_key_pair() await self._keys.create( project_id=project_id, key_id=kid, public_key_pem=public_pem, private_key_enc=encrypt_value(private_pem), ) await self._push_auth_keys(project_id, kid, public_pem, private_pem) else: await self._config.create( project_id, enabled=True, providers=["email", "magic_link"], auth_methods=["passwordless"], identity_channels=["email"], auth_connections=[{"id": "local_magic_link", "method": "passwordless", "channel": "email"}], audiences=["staff", "customer"], allow_signup=True, ) kid, private_pem, public_pem = _generate_rsa_key_pair() await self._keys.create( project_id=project_id, key_id=kid, public_key_pem=public_pem, private_key_enc=encrypt_value(private_pem), ) created = True await self._push_auth_keys(project_id, kid, public_pem, private_pem) await self.ensure_staff_roles_and_owner_admin(project_id) if created: logger.info("project_default_auth_provisioned", project_id=str(project_id)) return created async def ensure_staff_roles_and_owner_admin(self, project_id: UUID) -> None: """Ensure default roles exist and workspace collaborators are Cloud app users. The workspace owner is generated-app super_admin. Other members, including workspace admins, are generated-app admin. Idempotent. Cloud upsert is non-fatal so a transient outage cannot block project setup or codegen. """ await self._ensure_default_roles(project_id) config = await self._config.get_by_project(project_id) if not config or not config.enabled: return # Auth config and roles are required for the generated runtime. Keep the # exception visible to the setup worker so the project is marked blocked. await self._push_auth_config(project_id, config) try: await self._seed_workspace_collaborators(project_id) except Exception: logger.warning( "workspace_owner_admin_seed_skipped", project_id=str(project_id), ) await self._reconcile_canonical_memberships(project_id) async def _reconcile_canonical_memberships(self, project_id: UUID) -> None: """Create the owner and workspace-admin memberships in Core at provisioning time. Same canonical planner as the repair sweep, so a new project starts converged. It is Core-only (no Cloud write) and gated by the system flag and the project's eligibility. """ from src.config import get_settings from src.domain.project_auth.membership_projection import ( reconcile_project_from_workspace, ) from src.domain.project_auth.projection_eligibility import may_reconcile, project_state db = self._config._db try: if not may_reconcile( get_settings().project_membership_projection_enabled, await project_state(db, project_id), ): return async with db.begin_nested(): await reconcile_project_from_workspace(db, project_id) except Exception: logger.warning("provisioning_membership_reconcile_skipped", project_id=str(project_id)) async def _ensure_default_roles(self, project_id: UUID) -> None: """Seed the v2 default roles for a new project. Four system roles, each with a fixed assignment tier and a v2 permission preset: member (tier member), staff (tier staff), admin (tier admin), and super_admin, the protected owner role (no tier, not assignable). The owner's authority comes from is_owner, and the protected role carries OWNER_PRESET. Existing projects are not touched here; they are handled by the backfill. """ roles = await self._roles.list_for_project(project_id) names = {r.name for r in roles} has_default = any(r.is_default for r in roles) if CUSTOMER_MEMBER_ROLE not in names: await self._roles.create( project_id=project_id, name=CUSTOMER_MEMBER_ROLE, description="Default customer / end-user role", is_default=not has_default, permissions_json=list(MEMBER_PRESET), created_by=AuthRoleCreator.codegen, assignment_tier=AuthAssignmentTier.member, ) has_default = True if STAFF_ROLE not in names: await self._roles.create( project_id=project_id, name=STAFF_ROLE, description="Staff role — operational access, granted by an admin or the owner", is_default=False, permissions_json=list(STAFF_PRESET), created_by=AuthRoleCreator.codegen, assignment_tier=AuthAssignmentTier.staff, ) if STAFF_ADMIN_ROLE not in names: await self._roles.create( project_id=project_id, name=STAFF_ADMIN_ROLE, description="Admin role — manages members and staff, granted by the owner", is_default=False, permissions_json=list(ADMIN_PRESET), created_by=AuthRoleCreator.codegen, assignment_tier=AuthAssignmentTier.admin, ) if SUPER_ADMIN_ROLE not in names: await self._roles.create( project_id=project_id, name=SUPER_ADMIN_ROLE, description="Project owner — protected, not assignable; authority comes from ownership", is_default=False, permissions_json=list(OWNER_PRESET), created_by=AuthRoleCreator.codegen, assignment_tier=None, protected=True, ) async def _seed_workspace_collaborators(self, project_id: UUID) -> None: from sqlalchemy import select from src.domain.project_auth.staff_directory import identities_for_workspace from src.domain.projects.models import Project from src.domain.workspaces.models import Workspace db = self._config._db project = ( await db.execute(select(Project).where(Project.id == project_id)) ).scalar_one_or_none() if not project: return workspace = ( await db.execute(select(Workspace).where(Workspace.id == project.workspace_id)) ).scalar_one_or_none() if not workspace: return identities = await identities_for_workspace( db, project.workspace_id, workspace.owner_id ) current_emails = {identity.email for identity in identities} for identity in identities: await self._cloud.upsert_app_user( str(project_id), email=identity.email, role=identity.app_role, email_verified=True, user_metadata={ "source": "workspace_owner" if identity.is_owner else "workspace_member", "workspace_role": identity.workspace_role, "name": identity.name, "p2hs_authority": { "source": "workspace", "workspace_role": identity.workspace_role, "app_role": identity.app_role, }, "p2hs_auth": { "identity": {"channel": "email", "value": identity.email}, "method": "passwordless", "connection": "local_magic_link", "audience": "staff", "membership": "workspace_inherited", "role": identity.app_role, }, }, ) # Converge only identities previously marked as workspace-inherited. # Application-local users are intentionally untouched. try: cloud_users = await self._cloud.list_app_users(str(project_id)) except Exception: logger.warning("workspace_staff_reconciliation_list_failed", project_id=str(project_id)) return for user in cloud_users: email = str(user.get("email") or "").strip().lower() metadata = user.get("user_metadata") or {} authority = metadata.get("p2hs_authority") if isinstance(metadata, dict) else None if ( email and email not in current_emails and isinstance(authority, dict) and authority.get("source") == "workspace" and user.get("id") ): await self._cloud.revoke_app_user( project_id=str(project_id), user_id=str(user["id"]), reason="workspace_membership_reconciled", ) # ------------------------------------------------------------------ # Revocation # ------------------------------------------------------------------ async def revoke_app_user( self, project_id: UUID, body: RevokeAppUserRequest ) -> None: """Revoke an app user's access by pushing to Cloud KV.""" await self._cloud.revoke_app_user( project_id=str(project_id), user_id=body.user_id, reason=body.reason, ) logger.info( "app_user_revoked", project_id=str(project_id), user_id=body.user_id, reason=body.reason, ) # ------------------------------------------------------------------ # Internal — control plane pushes # ------------------------------------------------------------------ async def _push_auth_config(self, project_id: UUID, config: ProjectAuthConfig) -> None: from sqlalchemy import select from src.domain.projects.models import Project roles = await self._roles.list_for_project(project_id) role_dicts = [] for r in roles: perms = r.permissions_json if isinstance(perms, list): perm_map = {str(p): True for p in perms} elif isinstance(perms, dict): perm_map = perms else: perm_map = {} role_dicts.append({ "name": r.name, "is_default": r.is_default, "permissions": perms if isinstance(perms, list) else list(perm_map), "permissions_json": perm_map, }) google_client_id: Optional[str] = None if config.google_client_id_enc: try: google_client_id = decrypt_value(config.google_client_id_enc) except Exception: pass from src.domain.projects.runtime_invariants import ( ProjectRuntimeInvariants, validate_runtime_routes, ) workspace_id = None project = ( await self._config._db.execute(select(Project).where(Project.id == project_id)) ).scalar_one_or_none() if project: workspace_id = str(project.workspace_id) auth_contract = _effective_auth_contract(config) policy = { "schema_version": 1, "workspace_id": workspace_id, "project_id": str(project_id), "roles": role_dicts, "auth_methods": auth_contract["auth_methods"], "identity_channels": auth_contract["identity_channels"], "connections": auth_contract["connections"], "audiences": auth_contract["audiences"], "require_mfa": bool(getattr(config, "require_mfa", False)), } policy_hash = authz_policy_fingerprint(policy) policy_version = config.authz_policy_version if config.authz_policy_hash and config.authz_policy_hash != policy_hash: policy_version += 1 invariants = ProjectRuntimeInvariants.model_validate( getattr(project, "runtime_invariants", None) or {} ) route_errors = validate_runtime_routes(invariants) if route_errors: raise ConflictError("Invalid project runtime routes: " + ", ".join(route_errors)) from datetime import datetime, timezone try: pushed = await self._cloud.push_auth_config( project_id=str(project_id), config={ "workspace_id": workspace_id, "admin_panel_path": invariants.admin_panel_path, "customer_login_path": invariants.customer_login_path, "enabled": config.enabled, "providers": config.providers or [], "auth_methods": auth_contract["auth_methods"], "identity_channels": auth_contract["identity_channels"], "connections": auth_contract["connections"], "audiences": auth_contract["audiences"], "mfa_methods": getattr(config, "mfa_methods", None) or [], "require_mfa": bool(getattr(config, "require_mfa", False)), "allow_signup": config.allow_signup, "require_email_verify": config.require_email_verify, "google_oauth_mode": config.google_oauth_mode or "platform", "google_client_id": google_client_id, "access_token_ttl_seconds": config.access_token_ttl_seconds, "jwt_expiry_seconds": config.access_token_ttl_seconds, "refresh_token_ttl_days": config.refresh_token_ttl_days, "refresh_token_expiry_seconds": config.refresh_token_ttl_days * 86400, "password_min_length": config.password_min_length, "roles": role_dicts, "authz_policy_version": policy_version, "authz_policy_hash": policy_hash, "authz_rollout_mode": config.projection_state, }, ) if not pushed: raise RuntimeError("cloud_not_configured") except Exception as exc: await self._config.update(config, cloud_sync_error=str(exc)[:500]) raise if pushed: await self._config.update( config, cloud_synced_at=datetime.now(timezone.utc), cloud_sync_error=None, authz_policy_version=policy_version, authz_policy_hash=policy_hash, ) async def push_config_with_roles(self, project_id: UUID) -> None: """Re-sync the project's already-configured auth (including current roles and `Project.runtime_invariants`, e.g. admin_panel_path) to Cloud. A no-op if the project has no auth configured yet. Public so other domains can trigger a resync after changing something Cloud's copy depends on without themselves editing project_auth_config -- e.g. codegen/service.py after a runtime invariants lock changes admin_panel_path.""" config = await self._config.get_by_project(project_id) if config: await self._push_auth_config(project_id, config) async def push_active_key(self, project_id: UUID) -> None: """Repair only the active-key projection; never generates or rotates a key.""" key = await self._keys.get_active_for_project(project_id) if not key: raise ConflictError("auth_public_key_missing") await self._push_auth_keys( project_id, key.key_id, key.public_key_pem, decrypt_value(key.private_key_enc), ) async def _push_auth_keys( self, project_id: UUID, key_id: str, public_key_pem: str, private_key_pem: str ) -> None: pushed = await self._cloud.push_auth_keys( project_id=str(project_id), key_id=key_id, public_key_pem=public_key_pem, private_key_pem=private_key_pem, algorithm="RS256", ) if pushed is False: raise RuntimeError("cloud_not_configured") # --------------------------------------------------------------------------- # Key generation helpers # --------------------------------------------------------------------------- def _generate_rsa_key_pair() -> tuple[str, str, str]: """Generate a 2048-bit RSA key pair. Returns: (kid, private_pem, public_pem) kid is a 16-char hex token unique enough for a per-project key identifier. """ private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, ) private_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ).decode() public_pem = private_key.public_key().public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo, ).decode() kid = secrets.token_hex(8) return kid, private_pem, public_pem # --------------------------------------------------------------------------- # Response builders # --------------------------------------------------------------------------- def _config_response(row: ProjectAuthConfig) -> AuthConfigResponse: return AuthConfigResponse( project_id=row.project_id, enabled=row.enabled, providers=row.providers or [], auth_methods=_effective_auth_contract(row)["auth_methods"], identity_channels=_effective_auth_contract(row)["identity_channels"], auth_connections=_effective_auth_contract(row)["connections"], audiences=_effective_auth_contract(row)["audiences"], mfa_methods=row.mfa_methods or [], require_mfa=row.require_mfa, allow_signup=row.allow_signup, require_email_verify=row.require_email_verify, google_oauth_mode=row.google_oauth_mode, google_client_id_set=bool(row.google_client_id_enc), access_token_ttl_seconds=row.access_token_ttl_seconds, refresh_token_ttl_days=row.refresh_token_ttl_days, password_min_length=row.password_min_length, created_at=row.created_at, updated_at=row.updated_at, ) def _effective_auth_methods(row: ProjectAuthConfig) -> list[str]: return _effective_auth_contract(row)["auth_methods"] def _effective_auth_contract(row: ProjectAuthConfig) -> dict: return normalize_auth_contract( providers=getattr(row, "providers", None), auth_methods=getattr(row, "auth_methods", None), identity_channels=getattr(row, "identity_channels", None), connections=getattr(row, "auth_connections", None), audiences=getattr(row, "audiences", None), ) def _key_response(row) -> AuthKeyResponse: return AuthKeyResponse( id=row.id, project_id=row.project_id, key_id=row.key_id, status=row.status, public_key_pem=row.public_key_pem, created_at=row.created_at, rotated_at=row.rotated_at, ) def _role_response(row) -> AuthRoleResponse: return AuthRoleResponse( id=row.id, project_id=row.project_id, name=row.name, description=row.description, is_default=row.is_default, permissions=row.permissions_json or [], assignment_tier=row.assignment_tier, protected=row.protected, created_by=row.created_by, created_at=row.created_at, updated_at=row.updated_at, )