Expose resource access explanation API

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,