from __future__ import annotations from fastapi import HTTPException, status from sqlalchemy.orm import Session from govoplan_access.backend.admin.service import ( AdminConflictError, AdminValidationError, assert_tenant_owner_exists, role_assignment_counts, set_user_groups, set_user_roles, tenant_owner_user_ids, ) from govoplan_access.backend.api.v1.admin_schemas import ( ApiKeyAdminItem, GroupSummary, RoleSummary, SystemAccountItem, UserAdminItem, ) from govoplan_access.backend.security.sessions import ( collect_direct_user_roles, collect_system_roles, collect_user_groups, collect_user_scopes, ) from govoplan_access.backend.semantic import ( collect_external_function_roles, collect_function_assignment_ids, collect_function_delegation_ids, ) from govoplan_access.backend.auth.dependencies import ApiPrincipal, has_scope from govoplan_access.backend.db.models import ( Account, ApiKey, Group, GroupRoleAssignment, Role, SystemRoleAssignment, Tenant, User, UserGroupMembership, ) from govoplan_core.core.idm import OrganizationFunctionAssignmentRef from govoplan_core.core.organizations import OrganizationDirectory from govoplan_access.backend.permissions.catalog import effective_permission_count, expand_scopes def _http_admin_error(exc: Exception) -> HTTPException: if isinstance(exc, AdminConflictError): return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) if isinstance(exc, AdminValidationError): return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) def _require_system_role_assignment(principal: ApiPrincipal) -> None: if not (has_scope(principal, "system:roles:assign") or has_scope(principal, "system:access:assign")): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: system:roles:assign") def _require_permission(principal: ApiPrincipal, scope: str) -> None: if not has_scope(principal, scope): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}") def _resolve_tenant( session: Session, principal: ApiPrincipal, tenant_id: str | None, ) -> Tenant: target_id = tenant_id or principal.tenant_id if target_id != principal.tenant_id: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Switch to the target tenant before using tenant-administration endpoints.", ) tenant = session.get(Tenant, target_id) if tenant is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found") return tenant def _role_summary(session: Session, role: Role) -> RoleSummary: group_count = 0 if role.tenant_id is None: user_count = session.query(SystemRoleAssignment).filter(SystemRoleAssignment.role_id == role.id).count() permission_level = "system" else: user_count, group_count = role_assignment_counts(session, role.id) permission_level = "tenant" permissions = list(role.permissions or []) return RoleSummary( id=role.id, slug=role.slug, name=role.name, description=role.description, permissions=permissions, effective_permission_count=effective_permission_count(permissions, level=permission_level), is_builtin=role.is_builtin, is_assignable=role.is_assignable, user_assignments=user_count, group_assignments=group_count, level="system" if role.tenant_id is None else "tenant", system_template_id=role.system_template_id, system_required=role.system_required, ) def _group_summary(session: Session, group: Group, *, include_members: bool = True) -> GroupSummary: member_ids = [ row[0] for row in session.query(UserGroupMembership.user_id) .filter(UserGroupMembership.tenant_id == group.tenant_id, UserGroupMembership.group_id == group.id) .all() ] roles = ( session.query(Role) .join(GroupRoleAssignment, GroupRoleAssignment.role_id == Role.id) .filter(GroupRoleAssignment.tenant_id == group.tenant_id, GroupRoleAssignment.group_id == group.id) .order_by(Role.name.asc()) .all() ) return GroupSummary( id=group.id, slug=group.slug, name=group.name, description=group.description, is_active=group.is_active, member_count=len(member_ids), member_ids=member_ids if include_members else [], roles=[_role_summary(session, role) for role in roles], created_at=group.created_at, updated_at=group.updated_at, system_template_id=group.system_template_id, system_required=group.system_required, ) def _user_item( session: Session, user: User, *, owner_ids: set[str] | None = None, idm_assignments: tuple[OrganizationFunctionAssignmentRef, ...] = (), organization_directory: OrganizationDirectory | None = None, ) -> UserAdminItem: account = session.get(Account, user.account_id) if account is None: raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="User account is missing") groups = collect_user_groups(session, user) roles = collect_direct_user_roles(session, user) external_roles = ( collect_external_function_roles( session, user, idm_assignments, organization_directory=organization_directory, ) if idm_assignments else [] ) effective_scopes = set(collect_user_scopes(session, user, include_system=False)) for role in external_roles: effective_scopes.update(role.permissions or []) function_assignment_ids = collect_function_assignment_ids(session, user) function_assignment_ids.extend(item.id for item in idm_assignments) effective_owner_ids = owner_ids if owner_ids is not None else tenant_owner_user_ids(session, user.tenant_id) return UserAdminItem( id=user.id, account_id=account.id, tenant_id=user.tenant_id, email=account.email, display_name=user.display_name or account.display_name, is_active=user.is_active, account_is_active=account.is_active, password_reset_required=account.password_reset_required, last_login_at=account.last_login_at, groups=[_group_summary(session, group, include_members=False) for group in groups], roles=[_role_summary(session, role) for role in roles], function_assignment_ids=sorted(dict.fromkeys(function_assignment_ids)), function_delegation_ids=collect_function_delegation_ids(session, user), effective_scopes=expand_scopes(effective_scopes), is_owner=user.id in effective_owner_ids, is_last_active_owner=user.id in effective_owner_ids and len(effective_owner_ids) == 1, created_at=user.created_at, updated_at=user.updated_at, ) def _system_account_item(session: Session, account: Account) -> SystemAccountItem: memberships = ( session.query(User, Tenant) .join(Tenant, Tenant.id == User.tenant_id) .filter(User.account_id == account.id) .order_by(Tenant.name.asc()) .all() ) owner_ids_by_tenant = { tenant.id: tenant_owner_user_ids(session, tenant.id) for _, tenant in memberships } return SystemAccountItem( account_id=account.id, email=account.email, display_name=account.display_name, is_active=account.is_active, memberships=[ { "tenant_id": tenant.id, "tenant_name": tenant.name, "user_id": user.id, "is_active": user.is_active and tenant.is_active, "role_ids": [role.id for role in collect_direct_user_roles(session, user)], "group_ids": [group.id for group in collect_user_groups(session, user)], "is_owner": user.id in owner_ids_by_tenant[tenant.id], "is_last_active_owner": ( user.id in owner_ids_by_tenant[tenant.id] and len(owner_ids_by_tenant[tenant.id]) == 1 ), } for user, tenant in memberships ], roles=[_role_summary(session, role) for role in collect_system_roles(session, account)], last_login_at=account.last_login_at, ) def _api_key_item(session: Session, item: ApiKey) -> ApiKeyAdminItem: user = session.get(User, item.user_id) account = session.get(Account, user.account_id) if user else None return ApiKeyAdminItem( id=item.id, user_id=item.user_id, user_email=account.email if account else "Unknown account", name=item.name, prefix=item.prefix, scopes=item.scopes or [], expires_at=item.expires_at, last_used_at=item.last_used_at, revoked_at=item.revoked_at, created_at=item.created_at, ) def _set_system_memberships(session: Session, account: Account, requested: list[dict]) -> None: desired = {item["tenant_id"]: item for item in requested} existing = {item.tenant_id: item for item in session.query(User).filter(User.account_id == account.id).all()} affected = set(existing) | set(desired) for tenant_id, user in existing.items(): if tenant_id not in desired: user.is_active = False session.add(user) for tenant_id, item in desired.items(): tenant = session.get(Tenant, tenant_id) if tenant is None: raise AdminValidationError(f"Unknown tenant: {tenant_id}") user = existing.get(tenant_id) if user is None: user = User( tenant_id=tenant.id, account_id=account.id, email=account.email, display_name=account.display_name, is_active=bool(item.get("is_active", True)), auth_provider=account.auth_provider, password_hash=account.password_hash, ) session.add(user) session.flush() else: user.is_active = bool(item.get("is_active", True)) user.email = account.email session.add(user) set_user_roles(session, user=user, role_ids=item.get("role_ids", [])) set_user_groups(session, user=user, group_ids=item.get("group_ids", [])) session.flush() for tenant_id in affected: tenant = session.get(Tenant, tenant_id) if tenant and tenant.is_active: assert_tenant_owner_exists(session, tenant_id)