From d6a0d6241d52761a5cc2343062a73d9db958bc1b Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sat, 11 Jul 2026 00:39:39 +0200 Subject: [PATCH] Expose resource access explanation API --- .../backend/api/v1/admin_schemas.py | 17 ++++ src/govoplan_access/backend/api/v1/routes.py | 84 ++++++++++++++++++- src/govoplan_access/backend/explanation.py | 28 ++++++- src/govoplan_access/backend/manifest.py | 15 ++++ webui/src/api/admin.ts | 31 +++++++ 5 files changed, 171 insertions(+), 4 deletions(-) diff --git a/src/govoplan_access/backend/api/v1/admin_schemas.py b/src/govoplan_access/backend/api/v1/admin_schemas.py index 70be955..491d77f 100644 --- a/src/govoplan_access/backend/api/v1/admin_schemas.py +++ b/src/govoplan_access/backend/api/v1/admin_schemas.py @@ -486,6 +486,15 @@ class AccessScopeExplanationItem(BaseModel): sources: list[AccessRoleSourceItem] = Field(default_factory=list) +class AccessDecisionProvenanceItem(BaseModel): + kind: str + id: str | None = None + label: str | None = None + tenant_id: str | None = None + source: str | None = None + details: dict[str, object] = Field(default_factory=dict) + + class FunctionFactExplanationItem(BaseModel): source_module: str assignment_id: str @@ -512,6 +521,14 @@ class UserAccessExplanationResponse(BaseModel): function_facts: list[FunctionFactExplanationItem] = Field(default_factory=list) +class ResourceAccessExplanationResponse(BaseModel): + user: UserAdminItem + resource_type: str + resource_id: str + action: str + provenance: list[AccessDecisionProvenanceItem] = Field(default_factory=list) + + class UserListResponse(BaseModel): users: list[UserAdminItem] diff --git a/src/govoplan_access/backend/api/v1/routes.py b/src/govoplan_access/backend/api/v1/routes.py index 22a5f0e..d0ac600 100644 --- a/src/govoplan_access/backend/api/v1/routes.py +++ b/src/govoplan_access/backend/api/v1/routes.py @@ -110,6 +110,7 @@ from govoplan_access.backend.api.v1.admin_schemas import ( RoleListResponse, RoleSummary, RoleUpdateRequest, + ResourceAccessExplanationResponse, SystemAccountCreateRequest, SystemAccountCreateResponse, SystemAccountItem, @@ -154,6 +155,7 @@ from govoplan_core.core.configuration_control import ( record_configuration_change_applied, ) from govoplan_core.core.configuration_safety import configuration_safety_catalog, plan_configuration_change +from govoplan_core.core.access import CAPABILITY_ACCESS_EXPLANATION, AccessExplanationService, AccessDecisionProvenance, PrincipalRef from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, IdmDirectory, OrganizationFunctionAssignmentRef from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, ORGANIZATIONS_MODULE_ID, OrganizationDirectory from govoplan_core.api.v1.schemas import DeltaDeletedItem @@ -185,7 +187,8 @@ from govoplan_access.backend.db.models import ( UserGroupMembership, UserRoleAssignment, ) -from govoplan_access.backend.semantic import identity_id_for_account +from govoplan_access.backend.semantic import collect_external_function_roles, collect_function_assignment_ids, collect_function_delegation_ids, identity_id_for_account +from govoplan_access.backend.security.sessions import collect_user_groups, collect_user_roles, collect_user_scopes from govoplan_core.db.session import get_session from govoplan_access.backend.permissions.catalog import ( normalize_email, @@ -514,6 +517,21 @@ def _optional_idm_directory() -> IdmDirectory | None: return capability +def _access_explanation_service_or_error() -> AccessExplanationService: + registry = get_registry() + if registry is not None and registry.has_capability(CAPABILITY_ACCESS_EXPLANATION): + capability = registry.require_capability(CAPABILITY_ACCESS_EXPLANATION) + if not isinstance(capability, AccessExplanationService): + raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Invalid capability: {CAPABILITY_ACCESS_EXPLANATION}") + return capability + from govoplan_access.backend.explanation import SqlAccessExplanationService + + return SqlAccessExplanationService( + idm_directory=_optional_idm_directory(), + organization_directory=_optional_organization_directory(), + ) + + def _idm_assignments_for_user( idm_directory: IdmDirectory | None, user: User, @@ -523,6 +541,39 @@ def _idm_assignments_for_user( return tuple(idm_directory.organization_function_assignments_for_account(user.account_id, tenant_id=user.tenant_id)) +def _principal_ref_for_user(session: Session, user: User, *, idm_directory: IdmDirectory | None = None) -> PrincipalRef: + account = session.get(Account, user.account_id) + if account is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found") + idm_assignments = _idm_assignments_for_user(idm_directory, user) + idm_roles = tuple(collect_external_function_roles(session, user, idm_assignments)) + scopes = set(collect_user_scopes(session, user, include_system=True)) + for role in idm_roles: + scopes.update(role.permissions or []) + role_ids = [role.id for role in collect_user_roles(session, user)] + role_ids.extend(role.id for role in idm_roles) + function_assignment_ids = list(collect_function_assignment_ids(session, user)) + function_assignment_ids.extend(item.id for item in idm_assignments) + return PrincipalRef( + account_id=account.id, + membership_id=user.id, + tenant_id=user.tenant_id, + identity_id=identity_id_for_account(session, account.id), + scopes=frozenset(sorted(scopes)), + group_ids=frozenset(group.id for group in collect_user_groups(session, user)), + role_ids=frozenset(sorted(dict.fromkeys(role_ids))), + function_assignment_ids=frozenset(sorted(dict.fromkeys(function_assignment_ids))), + delegation_ids=frozenset(collect_function_delegation_ids(session, user)), + auth_method="session", + email=account.email, + display_name=account.display_name or user.display_name, + ) + + +def _provenance_payload(items: Iterable[AccessDecisionProvenance]) -> list[dict[str, object]]: + return [item.to_dict() for item in items] + + def _validate_external_function_source(*, tenant_id: str, source_module: str, function_id: str) -> None: if source_module != ORGANIZATIONS_MODULE_ID: raise HTTPException( @@ -1988,6 +2039,37 @@ def get_user_access_explanation( ) +@router.get("/access/resource-explanation", response_model=ResourceAccessExplanationResponse) +def get_resource_access_explanation( + user_id: str = Query(...), + resource_type: str = Query(..., min_length=1, max_length=100), + resource_id: str = Query(..., min_length=1, max_length=2048), + action: str = Query(..., min_length=1, max_length=255), + tenant_id: str | None = Query(default=None), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(require_any_scope("admin:users:read", "admin:roles:read", "access:membership:read", "access:role:read")), +): + tenant = _resolve_tenant(session, principal, tenant_id) + user = session.query(User).filter(User.id == user_id, User.tenant_id == tenant.id).one_or_none() + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") + idm_directory = _optional_idm_directory() + target_principal = _principal_ref_for_user(session, user, idm_directory=idm_directory) + provenance = _access_explanation_service_or_error().explain_resource_provenance( + target_principal, + resource_type=resource_type, + resource_id=resource_id, + action=action, + ) + return ResourceAccessExplanationResponse( + user=_user_item_for_response(session, user, idm_directory=idm_directory), + resource_type=resource_type, + resource_id=resource_id, + action=action, + provenance=_provenance_payload(provenance), + ) + + @router.post("/users", response_model=UserCreateResponse, status_code=status.HTTP_201_CREATED) def create_user( payload: UserCreateRequest, diff --git a/src/govoplan_access/backend/explanation.py b/src/govoplan_access/backend/explanation.py index fad6e97..f78f5f5 100644 --- a/src/govoplan_access/backend/explanation.py +++ b/src/govoplan_access/backend/explanation.py @@ -26,7 +26,7 @@ from govoplan_access.backend.semantic import ( active_function_delegations_for_account, identity_id_for_account, ) -from govoplan_core.core.access import AccessDecisionProvenance, AccessExplanationService, PrincipalRef +from govoplan_core.core.access import AccessDecisionProvenance, AccessExplanationService, PrincipalRef, ResourceAccessExplanationProvider from govoplan_core.core.idm import IdmDirectory, OrganizationFunctionAssignmentRef from govoplan_core.core.organizations import OrganizationDirectory from govoplan_core.db.session import get_database @@ -161,9 +161,11 @@ class SqlAccessExplanationService(AccessExplanationService): *, idm_directory: IdmDirectory | None = None, organization_directory: OrganizationDirectory | None = None, + resource_explanation_providers: Iterable[ResourceAccessExplanationProvider] = (), ) -> None: self._idm_directory = idm_directory self._organization_directory = organization_directory + self._resource_explanation_providers = tuple(resource_explanation_providers) def explain_scope_provenance( self, @@ -189,8 +191,28 @@ class SqlAccessExplanationService(AccessExplanationService): resource_id: str, action: str, ) -> tuple[AccessDecisionProvenance, ...]: - del resource_type, resource_id - return self.explain_scope_provenance(principal, action) + with get_database().session() as session: + items = _scope_provenance( + session, + principal, + action, + idm_directory=self._idm_directory, + organization_directory=self._organization_directory, + ) + for provider in self._resource_explanation_providers: + provider_items = tuple( + provider.explain_resource_provenance( + session, + principal, + resource_type=resource_type, + resource_id=resource_id, + action=action, + ) + ) + if provider_items: + items.extend(provider_items) + break + return tuple(_dedupe_provenance(items)) def build_user_access_explanation( diff --git a/src/govoplan_access/backend/manifest.py b/src/govoplan_access/backend/manifest.py index 536d308..32c55ad 100644 --- a/src/govoplan_access/backend/manifest.py +++ b/src/govoplan_access/backend/manifest.py @@ -17,7 +17,10 @@ from govoplan_core.core.access import ( CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_TENANT_CONTEXT_SWITCHER, + ResourceAccessExplanationProvider, ) +from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_ACCESS +from govoplan_core.core.files import CAPABILITY_FILES_ACCESS from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, IdmDirectory from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory from govoplan_core.core.module_guards import persistent_table_uninstall_guard @@ -507,12 +510,24 @@ def _optional_organization_directory(context: ModuleContext) -> OrganizationDire return capability +def _resource_explanation_providers(context: ModuleContext) -> tuple[ResourceAccessExplanationProvider, ...]: + providers: list[ResourceAccessExplanationProvider] = [] + for capability_name in (CAPABILITY_FILES_ACCESS, CAPABILITY_CAMPAIGNS_ACCESS): + if not context.registry.has_capability(capability_name): + continue + capability = context.registry.require_capability(capability_name) + if isinstance(capability, ResourceAccessExplanationProvider): + providers.append(capability) + return tuple(providers) + + def _access_explanation_service(context: ModuleContext) -> object: from govoplan_access.backend.explanation import SqlAccessExplanationService return SqlAccessExplanationService( idm_directory=_optional_idm_directory(context), organization_directory=_optional_organization_directory(context), + resource_explanation_providers=_resource_explanation_providers(context), ) diff --git a/webui/src/api/admin.ts b/webui/src/api/admin.ts index c6907fd..ade7ed3 100644 --- a/webui/src/api/admin.ts +++ b/webui/src/api/admin.ts @@ -132,6 +132,15 @@ export type AccessScopeExplanationItem = { sources: AccessRoleSourceItem[]; }; +export type AccessDecisionProvenanceItem = { + kind: string; + id?: string | null; + label?: string | null; + tenant_id?: string | null; + source?: string | null; + details: Record; +}; + export type FunctionFactExplanationItem = { source_module: string; assignment_id: string; @@ -158,6 +167,14 @@ export type UserAccessExplanationResponse = { function_facts: FunctionFactExplanationItem[]; }; +export type ResourceAccessExplanationResponse = { + user: UserAdminItem; + resource_type: string; + resource_id: string; + action: string; + provenance: AccessDecisionProvenanceItem[]; +}; + export type SystemAccountItem = { account_id: string; email: string; @@ -463,6 +480,20 @@ export function fetchUserAccessExplanation(settings: ApiSettings, userId: string return apiFetch(settings, `/api/v1/admin/users/${userId}/access-explanation`); } +export function fetchResourceAccessExplanation( + settings: ApiSettings, + options: { userId: string; resourceType: string; resourceId: string; action: string; tenantId?: string | null } +): Promise { + const params = new URLSearchParams({ + user_id: options.userId, + resource_type: options.resourceType, + resource_id: options.resourceId, + action: options.action + }); + if (options.tenantId) params.set("tenant_id", options.tenantId); + return apiFetch(settings, `/api/v1/admin/access/resource-explanation?${params.toString()}`); +} + export async function fetchGroups(settings: ApiSettings): Promise { const response = await apiFetch<{ groups: GroupSummary[] }>(settings, "/api/v1/admin/groups"); return response.groups;