from __future__ import annotations from dataclasses import dataclass from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from sqlalchemy.orm import Session from govoplan_core.api.v1.schemas import ( AuthGroupsResponse, AuthProfileResponse, AuthRolesResponse, AuthShellResponse, AuthSessionResponse, AuthSessionUserInfo, GroupInfo, LoginRequest, LoginResponse, MeResponse, PrincipalContextInfo, ProfileUpdateRequest, RoleInfo, SwitchTenantRequest, TenantInfo, TenantMembershipInfo, UserInfo, UserUiPreferences, ) from govoplan_core.core.access import AuthMethod, PrincipalRef from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, IdentityDirectory from govoplan_core.core.registry import PlatformRegistry from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal from govoplan_core.admin.settings import get_system_settings from govoplan_core.audit.logging import audit_event from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, maintenance_response_detail, saved_maintenance_mode from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, Group, Role, Tenant, User from govoplan_core.db.session import get_session from govoplan_core.i18n import ( i18n_settings, normalize_enabled_language_codes, normalize_language_code, preferred_language_code, system_enabled_language_codes, system_i18n_payload, tenant_enabled_language_codes, user_enabled_language_codes, ) from govoplan_access.backend.permissions.catalog import intersect_api_key_scopes, normalize_email, scopes_grant from govoplan_core.security.time import utc_now from govoplan_core.settings import settings from govoplan_access.backend.semantic import collect_function_assignment_ids, collect_function_delegation_ids, identity_id_for_account from govoplan_access.backend.auth.tenant_context import AccessTenantContextSwitcher from govoplan_access.backend.security.api_keys import authenticate_api_key from govoplan_access.backend.security.passwords import DUMMY_PASSWORD_HASH, verify_password from govoplan_access.backend.security.sessions import ( authenticate_session_token, collect_user_authorization_context, collect_system_roles, collect_tenant_memberships, collect_user_groups, collect_user_roles, collect_user_scopes, create_auth_session, verify_auth_session_csrf, ) router = APIRouter(prefix="/auth", tags=["auth"]) @dataclass(slots=True) class AuthContext: account: Account user: User tenant: Tenant auth_method: str source: str auth_session: AuthSession | None = None api_key: ApiKey | None = None def _cookie_samesite() -> str: value = settings.auth_cookie_samesite.lower().strip() if value not in {"lax", "strict", "none"}: return "lax" if value == "none" and not settings.auth_cookie_secure: return "lax" return value def _set_auth_cookies(response: Response, created) -> None: max_age = max(0, int((created.model.expires_at - utc_now()).total_seconds())) common = { "secure": settings.auth_cookie_secure, "samesite": _cookie_samesite(), "max_age": max_age, "path": "/", } if settings.auth_cookie_domain: common["domain"] = settings.auth_cookie_domain response.set_cookie(settings.auth_session_cookie_name, created.token, httponly=True, **common) response.set_cookie(settings.auth_csrf_cookie_name, created.csrf_token, httponly=False, **common) def _clear_auth_cookies(response: Response) -> None: kwargs = {"path": "/"} if settings.auth_cookie_domain: kwargs["domain"] = settings.auth_cookie_domain response.delete_cookie(settings.auth_session_cookie_name, **kwargs) response.delete_cookie(settings.auth_csrf_cookie_name, **kwargs) def _tenant_info(tenant: Tenant, *, enabled_language_codes: list[str] | None = None) -> TenantInfo: return TenantInfo( id=tenant.id, slug=tenant.slug, name=tenant.name, is_active=tenant.is_active, default_locale=tenant.default_locale, enabled_language_codes=enabled_language_codes or [], ) def _user_ui_preferences(settings_payload: object) -> UserUiPreferences: raw = settings_payload.get("ui") if isinstance(settings_payload, dict) else None try: return UserUiPreferences.model_validate(raw if isinstance(raw, dict) else {}) except ValueError: return UserUiPreferences() def _user_info( user: User, account: Account, *, preferred_language: str | None = None, enabled_language_codes: list[str] | None = None, ) -> UserInfo: return UserInfo( id=user.id, account_id=account.id, email=account.email, display_name=account.display_name or user.display_name, tenant_display_name=user.display_name, is_tenant_admin=user.is_tenant_admin, password_reset_required=account.password_reset_required, preferred_language=preferred_language, enabled_language_codes=enabled_language_codes or [], ui_preferences=_user_ui_preferences(user.settings), ) def _session_user_info(user: User, account: Account) -> AuthSessionUserInfo: return AuthSessionUserInfo( id=user.id, account_id=account.id, email=account.email, display_name=account.display_name or user.display_name, tenant_display_name=user.display_name, is_tenant_admin=user.is_tenant_admin, password_reset_required=account.password_reset_required, ) def _roles_info(roles, *, level: str = "tenant") -> list[RoleInfo]: return [ RoleInfo( id=role.id, slug=role.slug, name=role.name, permissions=role.permissions or [], level=level, ) for role in roles ] def _groups_info(groups) -> list[GroupInfo]: return [GroupInfo(id=group.id, slug=group.slug, name=group.name) for group in groups] def _tenant_memberships( session: Session, account: Account, *, active_user_id: str | None = None, active_tenant_roles: list[Role] | None = None, ) -> list[TenantMembershipInfo]: memberships: list[TenantMembershipInfo] = [] for user, tenant in collect_tenant_memberships(session, account): roles = ( active_tenant_roles if active_tenant_roles is not None and user.id == active_user_id else collect_user_roles(session, user) ) memberships.append( TenantMembershipInfo( id=tenant.id, slug=tenant.slug, name=tenant.name, is_active=tenant.is_active and user.is_active, default_locale=tenant.default_locale, roles=[role.slug for role in roles], ) ) return memberships def _tenant_membership_summaries(session: Session, account: Account) -> list[TenantMembershipInfo]: return [ TenantMembershipInfo( id=tenant.id, slug=tenant.slug, name=tenant.name, is_active=tenant.is_active and user.is_active, default_locale=tenant.default_locale, roles=[], ) for user, tenant in collect_tenant_memberships(session, account) ] def _language_context(session: Session, *, tenant: Tenant, user: User) -> dict[str, object]: system_settings = get_system_settings(session) system_payload = system_i18n_payload(system_settings) system_enabled = system_enabled_language_codes(system_settings.settings, default_locale=system_settings.default_locale) tenant_enabled = tenant_enabled_language_codes(tenant.settings, system_enabled, default_locale=tenant.default_locale) user_enabled = user_enabled_language_codes(user.settings, tenant_enabled) preferred = preferred_language_code( user.settings, user_enabled, default_locale=tenant.default_locale or system_payload.get("default_language"), ) return { "available_languages": system_payload["available_languages"], "tenant_enabled_language_codes": tenant_enabled, "user_enabled_language_codes": user_enabled, "preferred_language": preferred, } def _identity_directory_from_request(request: Request) -> IdentityDirectory | None: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry) or not registry.has_capability(CAPABILITY_IDENTITY_DIRECTORY): return None capability = registry.require_capability(CAPABILITY_IDENTITY_DIRECTORY) if not isinstance(capability, IdentityDirectory): raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Invalid capability: {CAPABILITY_IDENTITY_DIRECTORY}") return capability def _resolve_login_user(session: Session, payload: LoginRequest) -> tuple[Account, User, Tenant]: account = ( session.query(Account) .filter(Account.normalized_email == normalize_email(payload.email), Account.is_active.is_(True)) .one_or_none() ) password_hash = account.password_hash if account is not None and account.password_hash else DUMMY_PASSWORD_HASH password_matches = verify_password(payload.password, password_hash) if account is None or not account.password_hash or not password_matches: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid login") query = ( session.query(User, Tenant) .join(Tenant, Tenant.id == User.tenant_id) .filter( User.account_id == account.id, User.is_active.is_(True), Tenant.is_active.is_(True), ) ) if payload.tenant_slug: query = query.filter(Tenant.slug == payload.tenant_slug) row = query.order_by(Tenant.name.asc()).first() if row is None: # Keep every authentication failure generic so callers cannot infer # whether an account exists but lacks an active tenant membership. raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid login") return account, row[0], row[1] def _extract_auth_token(request: Request) -> tuple[str | None, str]: x_api_key = request.headers.get("x-api-key") if x_api_key: return x_api_key.strip(), "api_key" authorization = request.headers.get("authorization") if authorization and authorization.lower().startswith("bearer "): return authorization[7:].strip(), "bearer" cookie_token = request.cookies.get(settings.auth_session_cookie_name) if cookie_token: return cookie_token.strip(), "cookie" return None, "none" def _active_context_or_401( *, user: User | None, account: Account | None, tenant: Tenant | None, expected_tenant_id: str, ) -> tuple[Account, User, Tenant]: if ( not user or not account or not tenant or not user.is_active or not account.is_active or not tenant.is_active or user.account_id != account.id or user.tenant_id != expected_tenant_id or tenant.id != expected_tenant_id ): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Inactive or inconsistent session") return account, user, tenant def _session_response( *, account: Account, user: User, tenant: Tenant, auth_method: str, auth_session: AuthSession | None = None, api_key: ApiKey | None = None, ) -> AuthSessionResponse: active_tenant = _tenant_info(tenant) return AuthSessionResponse( authenticated=True, auth_method=auth_method, # type: ignore[arg-type] user=_session_user_info(user, account), tenant=active_tenant, active_tenant=active_tenant, session_id=auth_session.id if auth_session else None, api_key_id=api_key.id if api_key else None, expires_at=auth_session.expires_at if auth_session else api_key.expires_at if api_key else None, ) def _resolve_auth_context(request: Request, session: Session) -> AuthContext: token, source = _extract_auth_token(request) if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing API key or session token") api_key = authenticate_api_key(session, token) if source != "cookie" else None if api_key is not None: user = session.get(User, api_key.user_id) account = session.get(Account, user.account_id) if user else None tenant = session.get(Tenant, api_key.tenant_id) account, user, tenant = _active_context_or_401( user=user, account=account, tenant=tenant, expected_tenant_id=api_key.tenant_id, ) return AuthContext( account=account, user=user, tenant=tenant, auth_method="api_key", source=source, api_key=api_key, ) auth_session = authenticate_session_token(session, token) if auth_session is None: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key or session token") user = session.get(User, auth_session.user_id) account = session.get(Account, auth_session.account_id) tenant = session.get(Tenant, auth_session.tenant_id) account, user, tenant = _active_context_or_401( user=user, account=account, tenant=tenant, expected_tenant_id=auth_session.tenant_id, ) return AuthContext( account=account, user=user, tenant=tenant, auth_method="session", source=source, auth_session=auth_session, ) def _shell_response( session: Session, *, account: Account, user: User, tenant: Tenant, scopes: list[str], auth_method: str, auth_session: AuthSession | None = None, api_key: ApiKey | None = None, ) -> AuthShellResponse: active_tenant = _tenant_info(tenant) memberships = ( _tenant_membership_summaries(session, account) if auth_session is not None else [ TenantMembershipInfo( id=tenant.id, slug=tenant.slug, name=tenant.name, is_active=tenant.is_active and user.is_active, default_locale=tenant.default_locale, roles=[], ) ] ) return AuthShellResponse( user=_user_info(user, account), tenant=active_tenant, active_tenant=active_tenant, tenants=memberships, scopes=scopes, principal=PrincipalContextInfo( account_id=account.id, membership_id=user.id, tenant_id=tenant.id, scopes=scopes, auth_method=auth_method, # type: ignore[arg-type] api_key_id=api_key.id if api_key else None, session_id=auth_session.id if auth_session else None, email=account.email, display_name=account.display_name or user.display_name, ), ) def _resolve_lightweight_session(request: Request, session: Session) -> AuthSessionResponse: context = _resolve_auth_context(request, session) return _session_response( account=context.account, user=context.user, tenant=context.tenant, auth_method=context.auth_method, auth_session=context.auth_session, api_key=context.api_key, ) def _resolve_shell_auth(request: Request, session: Session) -> AuthShellResponse: context = _resolve_auth_context(request, session) if context.api_key is not None: user_scopes = collect_user_scopes(session, context.user, include_system=False) scopes = intersect_api_key_scopes(user_scopes, context.api_key.scopes or []) return _shell_response( session, account=context.account, user=context.user, tenant=context.tenant, scopes=scopes, auth_method="api_key", api_key=context.api_key, ) scopes = collect_user_scopes(session, context.user, include_system=True) return _shell_response( session, account=context.account, user=context.user, tenant=context.tenant, scopes=scopes, auth_method="session", auth_session=context.auth_session, ) def _profile_response(session: Session, context: AuthContext) -> AuthProfileResponse: languages = _language_context(session, tenant=context.tenant, user=context.user) tenant_enabled = list(languages["tenant_enabled_language_codes"]) user_enabled = list(languages["user_enabled_language_codes"]) preferred_language = str(languages["preferred_language"]) active_tenant = _tenant_info(context.tenant, enabled_language_codes=tenant_enabled) return AuthProfileResponse( user=_user_info( context.user, context.account, preferred_language=preferred_language, enabled_language_codes=user_enabled, ), tenant=active_tenant, active_tenant=active_tenant, available_languages=languages["available_languages"], enabled_language_codes=tenant_enabled, default_language=preferred_language, ) def _roles_response(session: Session, context: AuthContext) -> AuthRolesResponse: tenant_roles = collect_user_roles(session, context.user) system_roles = collect_system_roles(session, context.account) if context.auth_session is not None else [] return AuthRolesResponse(roles=_roles_info(tenant_roles) + _roles_info(system_roles, level="system")) def _groups_response(session: Session, context: AuthContext) -> AuthGroupsResponse: return AuthGroupsResponse(groups=_groups_info(collect_user_groups(session, context.user))) def _verify_profile_mutation_allowed(request: Request, context: AuthContext) -> None: if context.auth_session is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="API keys cannot edit an interactive user profile") if context.source != "cookie": return header_token = request.headers.get("x-csrf-token") cookie_token = request.cookies.get(settings.auth_csrf_cookie_name) if not header_token or not cookie_token or header_token != cookie_token or not verify_auth_session_csrf(context.auth_session, header_token): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid or missing CSRF token") def _roles_from_principal(session: Session, principal: ApiPrincipal, *, tenant_id: str) -> tuple[list[Role], list[Role]]: roles = ( session.query(Role) .filter(Role.id.in_(sorted(principal.role_ids))) .order_by(Role.name.asc()) .all() ) tenant_roles = [role for role in roles if role.tenant_id == tenant_id] system_roles = [role for role in roles if role.tenant_id is None and principal.auth_session is not None] return tenant_roles, system_roles def _groups_from_principal(session: Session, principal: ApiPrincipal, *, tenant_id: str) -> list[Group]: return ( session.query(Group) .filter(Group.tenant_id == tenant_id, Group.id.in_(sorted(principal.group_ids))) .order_by(Group.name.asc()) .all() ) def _me_response( session: Session, *, account: Account, user: User, tenant: Tenant, effective_scopes: list[str] | None = None, auth_method: AuthMethod = "session", api_key_id: str | None = None, session_id: str | None = None, service_account_id: str | None = None, include_system: bool = True, include_all_memberships: bool = True, identity_directory: IdentityDirectory | None = None, identity_id: str | None = None, tenant_roles: list[Role] | None = None, system_roles: list[Role] | None = None, groups: list[Group] | None = None, function_assignment_ids: tuple[str, ...] | None = None, delegation_ids: tuple[str, ...] | None = None, ) -> MeResponse: if tenant_roles is None: tenant_roles = collect_user_roles(session, user) if system_roles is None: system_roles = collect_system_roles(session, account) if include_system else [] elif not include_system: system_roles = [] if groups is None: groups = collect_user_groups(session, user) scopes = effective_scopes if effective_scopes is not None else collect_user_scopes(session, user, include_system=include_system) resolved_function_assignment_ids = ( function_assignment_ids if function_assignment_ids is not None else tuple(collect_function_assignment_ids(session, user)) ) resolved_delegation_ids = ( delegation_ids if delegation_ids is not None else tuple(collect_function_delegation_ids(session, user)) ) languages = _language_context(session, tenant=tenant, user=user) tenant_enabled = list(languages["tenant_enabled_language_codes"]) user_enabled = list(languages["user_enabled_language_codes"]) preferred_language = str(languages["preferred_language"]) active_tenant = _tenant_info(tenant, enabled_language_codes=tenant_enabled) memberships = _tenant_memberships( session, account, active_user_id=user.id, active_tenant_roles=tenant_roles, ) if include_all_memberships else [ TenantMembershipInfo( id=tenant.id, slug=tenant.slug, name=tenant.name, is_active=tenant.is_active and user.is_active, default_locale=tenant.default_locale, roles=[role.slug for role in tenant_roles], ) ] return MeResponse( user=_user_info(user, account, preferred_language=preferred_language, enabled_language_codes=user_enabled), tenant=active_tenant, active_tenant=active_tenant, tenants=memberships, scopes=scopes, roles=_roles_info(tenant_roles) + _roles_info(system_roles, level="system"), groups=_groups_info(groups), principal=PrincipalContextInfo.model_validate( PrincipalRef( account_id=account.id, membership_id=user.id, tenant_id=tenant.id, identity_id=identity_id or identity_id_for_account(session, account.id, identity_directory=identity_directory), scopes=frozenset(scopes), group_ids=frozenset(group.id for group in groups), role_ids=frozenset(role.id for role in tenant_roles + system_roles), function_assignment_ids=frozenset(resolved_function_assignment_ids), delegation_ids=frozenset(resolved_delegation_ids), auth_method=auth_method, api_key_id=api_key_id, session_id=session_id, service_account_id=service_account_id, email=account.email, display_name=account.display_name or user.display_name, ).to_dict() ), available_languages=languages["available_languages"], enabled_language_codes=tenant_enabled, default_language=preferred_language, ) @router.post("/login", response_model=LoginResponse) def login(payload: LoginRequest, request: Request, response: Response, session: Session = Depends(get_session)): account, user, tenant = _resolve_login_user(session, payload) identity_directory = _identity_directory_from_request(request) authorization_context = collect_user_authorization_context( session, user, account=account, include_system=True, ) tenant_roles = authorization_context.tenant_roles system_roles = authorization_context.system_roles groups = authorization_context.groups effective_scopes = authorization_context.scopes maintenance_mode = saved_maintenance_mode(session) if maintenance_mode.enabled and not scopes_grant(effective_scopes, MAINTENANCE_ACCESS_SCOPE): raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=maintenance_response_detail(maintenance_mode), ) user_agent = request.headers.get("user-agent") ip_address = request.client.host if request.client else None created = create_auth_session( session, user=user, hours=settings.auth_session_hours, user_agent=user_agent, ip_address=ip_address, ) session.commit() _set_auth_cookies(response, created) return LoginResponse( access_token=created.token, expires_at=created.model.expires_at, **_me_response( session, account=account, user=user, tenant=tenant, effective_scopes=effective_scopes, auth_method="session", session_id=created.model.id, identity_directory=identity_directory, tenant_roles=tenant_roles, system_roles=system_roles, groups=groups, function_assignment_ids=authorization_context.function_assignment_ids, delegation_ids=authorization_context.function_delegation_ids, ).model_dump(), ) @router.get("/session", response_model=AuthSessionResponse) def auth_session(request: Request, session: Session = Depends(get_session)): return _resolve_lightweight_session(request, session) @router.get("/shell", response_model=AuthShellResponse) def auth_shell(request: Request, session: Session = Depends(get_session)): return _resolve_shell_auth(request, session) @router.get("/profile", response_model=AuthProfileResponse) def auth_profile(request: Request, session: Session = Depends(get_session)): return _profile_response(session, _resolve_auth_context(request, session)) @router.get("/roles", response_model=AuthRolesResponse) def auth_roles(request: Request, session: Session = Depends(get_session)): context = _resolve_auth_context(request, session) authorization_context = collect_user_authorization_context( session, context.user, account=context.account, include_system=context.auth_session is not None, ) return AuthRolesResponse( roles=_roles_info(authorization_context.tenant_roles) + _roles_info(authorization_context.system_roles, level="system") ) @router.get("/groups", response_model=AuthGroupsResponse) def auth_groups(request: Request, session: Session = Depends(get_session)): return _groups_response(session, _resolve_auth_context(request, session)) @router.get("/me", response_model=MeResponse) def me(principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session)): tenant = session.get(Tenant, principal.tenant_id) if tenant is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Active tenant not found") tenant_roles: list[Role] | None = None system_roles: list[Role] | None = None groups: list[Group] | None = None if principal.role_ids: tenant_roles, system_roles = _roles_from_principal(session, principal, tenant_id=tenant.id) if principal.group_ids: groups = _groups_from_principal(session, principal, tenant_id=tenant.id) return _me_response( session, account=principal.account, user=principal.user, tenant=tenant, effective_scopes=principal.scopes, auth_method=principal.auth_method, api_key_id=principal.api_key_id, session_id=principal.session_id, service_account_id=principal.principal.service_account_id, include_system=principal.auth_session is not None, include_all_memberships=principal.auth_session is not None, identity_id=principal.principal.identity_id, tenant_roles=tenant_roles, system_roles=system_roles, groups=groups, function_assignment_ids=tuple(principal.function_assignment_ids), delegation_ids=tuple(principal.delegation_ids), ) @router.patch("/profile", response_model=AuthProfileResponse) def update_profile( payload: ProfileUpdateRequest, request: Request, session: Session = Depends(get_session), ): context = _resolve_auth_context(request, session) _verify_profile_mutation_allowed(request, context) if "display_name" in payload.model_fields_set: context.account.display_name = payload.display_name.strip() if payload.display_name else None session.add(context.account) if "tenant_display_name" in payload.model_fields_set: context.user.display_name = payload.tenant_display_name.strip() if payload.tenant_display_name else None session.add(context.user) next_settings = dict(context.user.settings or {}) settings_changed = False if {"preferred_language", "enabled_language_codes"}.intersection(payload.model_fields_set): system_settings = get_system_settings(session) system_enabled = system_enabled_language_codes( system_settings.settings, default_locale=system_settings.default_locale, ) tenant_enabled = tenant_enabled_language_codes( context.tenant.settings, system_enabled, default_locale=context.tenant.default_locale, ) next_i18n = i18n_settings(next_settings) if "preferred_language" in payload.model_fields_set: preferred = normalize_language_code(payload.preferred_language) if preferred: normalized_preferred = normalize_enabled_language_codes( [preferred], [{"code": code} for code in tenant_enabled], fallback_codes=tenant_enabled, )[0] next_i18n["preferred_language"] = normalized_preferred else: next_i18n.pop("preferred_language", None) if "enabled_language_codes" in payload.model_fields_set: next_i18n["enabled_language_codes"] = normalize_enabled_language_codes( payload.enabled_language_codes, [{"code": code} for code in tenant_enabled], fallback_codes=tenant_enabled, ) next_settings["i18n"] = next_i18n settings_changed = True if "ui_preferences" in payload.model_fields_set: if payload.ui_preferences is None: next_settings["ui"] = UserUiPreferences().model_dump() else: next_ui = _user_ui_preferences(next_settings).model_dump() next_ui.update(payload.ui_preferences.model_dump(exclude_unset=True)) next_settings["ui"] = UserUiPreferences.model_validate(next_ui).model_dump() settings_changed = True if settings_changed: context.user.settings = next_settings session.add(context.user) audit_event( session, tenant_id=context.tenant.id, user_id=context.user.id, action="profile.updated", object_type="account", object_id=context.account.id, details={"fields": sorted(payload.model_fields_set)}, ) session.commit() return _profile_response(session, context) @router.post("/switch-tenant", response_model=AuthShellResponse) def switch_tenant( payload: SwitchTenantRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): if principal.auth_session is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="API keys cannot switch tenant context") try: switched = AccessTenantContextSwitcher().switch_tenant_context(session, principal=principal, tenant_id=payload.tenant_id) except LookupError as exc: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc tenant = session.get(Tenant, switched.tenant_id) membership = session.get(User, switched.membership_id) if tenant is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found") if membership is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant membership not found") session.commit() authorization_context = collect_user_authorization_context( session, membership, account=principal.account, include_system=True, ) return _shell_response( session, account=principal.account, user=membership, tenant=tenant, scopes=authorization_context.scopes, auth_method="session", auth_session=principal.auth_session, ) @router.post("/logout") def logout( response: Response, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): if principal.auth_session is not None: principal.auth_session.revoked_at = utc_now() session.add(principal.auth_session) session.commit() _clear_auth_cookies(response) return {"ok": True}