From 566b3b83addb66d4a16caa9b767226651e12274f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 20 Aug 2026 06:16:43 +0200 Subject: [PATCH] feat: secure resource explanation subjects --- .../backend/api/v1/admin_schemas.py | 15 ++ src/govoplan_access/backend/api/v1/routes.py | 136 +++++++++++++++++- src/govoplan_access/backend/manifest.py | 51 +++++++ ...st_resource_access_explanation_subjects.py | 109 ++++++++++++++ 4 files changed, 310 insertions(+), 1 deletion(-) create mode 100644 tests/test_resource_access_explanation_subjects.py diff --git a/src/govoplan_access/backend/api/v1/admin_schemas.py b/src/govoplan_access/backend/api/v1/admin_schemas.py index 7c43f45..77670c4 100644 --- a/src/govoplan_access/backend/api/v1/admin_schemas.py +++ b/src/govoplan_access/backend/api/v1/admin_schemas.py @@ -541,6 +541,21 @@ class ResourceAccessExplanationResponse(BaseModel): provenance: list[AccessDecisionProvenanceItem] = Field(default_factory=list) +class ResourceAccessExplanationSubjectItem(BaseModel): + id: str + email: str | None = None + display_name: str | None = None + + +class ResourceAccessExplanationSubjectsResponse(BaseModel): + mode: Literal["current_user", "cross_user"] + can_select_other_users: bool + reason: str + source: str + required_scope: str | None = None + users: list[ResourceAccessExplanationSubjectItem] = Field(default_factory=list) + + class UserListResponse(PagedListResponse): users: list[UserAdminItem] diff --git a/src/govoplan_access/backend/api/v1/routes.py b/src/govoplan_access/backend/api/v1/routes.py index 76fd2b8..fb770b3 100644 --- a/src/govoplan_access/backend/api/v1/routes.py +++ b/src/govoplan_access/backend/api/v1/routes.py @@ -6,6 +6,7 @@ from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.exc import IntegrityError +from sqlalchemy import or_ from sqlalchemy.orm import Session from govoplan_access.backend.admin.governance import ( @@ -125,6 +126,8 @@ from govoplan_access.backend.api.v1.admin_schemas import ( RoleSummary, RoleUpdateRequest, ResourceAccessExplanationResponse, + ResourceAccessExplanationSubjectItem, + ResourceAccessExplanationSubjectsResponse, SystemAccountCreateRequest, SystemAccountCreateResponse, SystemAccountItem, @@ -186,7 +189,15 @@ from govoplan_core.core.provider_governance import ( ExternalProviderStateContext, collect_external_provider_states, ) -from govoplan_core.core.access import CAPABILITY_ACCESS_EXPLANATION, AccessExplanationService, AccessDecisionProvenance, PrincipalRef +from govoplan_core.core.access import ( + CAPABILITY_ACCESS_EXPLANATION, + CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS, + AccessDecisionProvenance, + AccessExplanationService, + AccessExplanationSubjectDecision, + AccessExplanationSubjectPolicy, + PrincipalRef, +) from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, IdentityDirectory from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, IdmDirectory, OrganizationFunctionAssignmentRef from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, ORGANIZATIONS_MODULE_ID, OrganizationDirectory @@ -673,6 +684,40 @@ def _access_explanation_service_or_error() -> AccessExplanationService: ) +def _access_explanation_subject_decision( + session: Session, + principal: ApiPrincipal, + *, + tenant_id: str, +) -> AccessExplanationSubjectDecision: + registry = get_registry() + if registry is None or not registry.has_capability( + CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS + ): + return AccessExplanationSubjectDecision( + allow_other_users=False, + reason="Access explanations are limited to the signed-in user because no subject policy is active.", + source="access.safe_default", + provenance={"tenant_id": tenant_id, "mode": "current_user"}, + ) + capability = registry.require_capability( + CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS + ) + if not isinstance(capability, AccessExplanationSubjectPolicy): + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=( + "Invalid capability: " + f"{CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS}" + ), + ) + return capability.decide_subject_selection( + session, + principal.principal, + tenant_id=tenant_id, + ) + + def _idm_assignments_for_user( idm_directory: IdmDirectory | None, user: User, @@ -2511,6 +2556,70 @@ def revoke_user_session( ) +@router.get( + "/access/resource-explanation/subjects", + response_model=ResourceAccessExplanationSubjectsResponse, +) +def get_resource_access_explanation_subjects( + tenant_id: str | None = Query(default=None), + query: str | None = Query(default=None, max_length=200), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends( + require_any_scope( + "admin:users:read", + "admin:roles:read", + "access:membership:read", + "access:role:read", + ) + ), +) -> ResourceAccessExplanationSubjectsResponse: + tenant = _resolve_tenant(session, principal, tenant_id) + decision = _access_explanation_subject_decision( + session, + principal, + tenant_id=tenant.id, + ) + users_query = session.query(User).filter( + User.tenant_id == tenant.id, + User.is_active.is_(True), + ) + if not decision.allow_other_users: + users_query = users_query.filter(User.id == principal.membership_id) + elif query and query.strip(): + pattern = f"%{query.strip()}%" + users_query = users_query.filter( + or_(User.display_name.ilike(pattern), User.email.ilike(pattern)) + ) + users = users_query.order_by(User.display_name.asc(), User.email.asc()).limit(100).all() + if decision.allow_other_users and not query: + current_user = ( + session.query(User) + .filter( + User.id == principal.membership_id, + User.tenant_id == tenant.id, + User.is_active.is_(True), + ) + .one_or_none() + ) + if current_user is not None and all(user.id != current_user.id for user in users): + users = [current_user, *users[:99]] + return ResourceAccessExplanationSubjectsResponse( + mode="cross_user" if decision.allow_other_users else "current_user", + can_select_other_users=decision.allow_other_users, + reason=decision.reason, + source=decision.source, + required_scope=decision.required_scope, + users=[ + ResourceAccessExplanationSubjectItem( + id=user.id, + email=user.email, + display_name=user.display_name, + ) + for user in users + ], + ) + + @router.get("/access/resource-explanation", response_model=ResourceAccessExplanationResponse) def get_resource_access_explanation( user_id: str = Query(...), @@ -2522,6 +2631,16 @@ def get_resource_access_explanation( 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) + decision = _access_explanation_subject_decision( + session, + principal, + tenant_id=tenant.id, + ) + if user_id != principal.membership_id and not decision.allow_other_users: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=decision.reason, + ) 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") @@ -2541,6 +2660,21 @@ def get_resource_access_explanation( resource_id=resource_id, action=action, ) + if user_id != principal.membership_id: + audit_from_principal( + session, + principal, + action="access.resource_explanation.selected_user_viewed", + scope="tenant", + object_type=resource_type, + object_id=resource_id, + details={ + "target_membership_id": user.id, + "requested_action": action, + "policy_source": decision.source, + }, + ) + session.commit() return ResourceAccessExplanationResponse( user=_user_item_for_response(session, user, idm_directory=idm_directory, organization_directory=organization_directory), resource_type=resource_type, diff --git a/src/govoplan_access/backend/manifest.py b/src/govoplan_access/backend/manifest.py index 306e5c2..17ce8b2 100644 --- a/src/govoplan_access/backend/manifest.py +++ b/src/govoplan_access/backend/manifest.py @@ -301,6 +301,57 @@ ADMIN_READ_SCOPES = ( ) ACCESS_DOCUMENTATION: tuple[DocumentationTopic, ...] = ( + DocumentationTopic( + id="access.reference.resource-explanation-subjects", + title="Select a user for resource-access diagnostics", + summary=( + "Resource explanations default to the signed-in user and expose " + "other tenant users only when Policy permits the diagnostic." + ), + body=( + "The shared Files and Campaign explanation dialog asks Access for " + "the permitted subject list. If Policy is unavailable or the actor " + "lacks policy:access_explanation:select_user, Access returns only " + "the signed-in active membership and no metadata for other users. " + "When Policy permits selection, the picker is limited to active " + "users in the current tenant. Every explanation run for another " + "user creates audit evidence with the target membership, resource, " + "requested action, and policy source. The explanation is diagnostic " + "and never grants resource access." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("tenant_admin", "access_admin", "security_reviewer"), + order=29, + conditions=( + DocumentationCondition( + required_modules=("access",), + any_scopes=( + "admin:users:read", + "admin:roles:read", + "access:membership:read", + "access:role:read", + ), + ), + ), + links=( + DocumentationLink( + label="Permitted explanation subjects API", + href="/api/v1/admin/access/resource-explanation/subjects", + kind="api", + ), + DocumentationLink( + label="Resource explanation API", + href="/api/v1/admin/access/resource-explanation", + kind="api", + ), + ), + related_modules=("audit", "campaigns", "files", "policy"), + metadata={ + "kind": "reference", + "help_contexts": ["access.resource-explanation.subject"], + }, + ), DocumentationTopic( id="access.operator.enroll-first-administrator", title="Enroll the first production administrator", diff --git a/tests/test_resource_access_explanation_subjects.py b/tests/test_resource_access_explanation_subjects.py new file mode 100644 index 0000000..3f10e5b --- /dev/null +++ b/tests/test_resource_access_explanation_subjects.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import pathlib +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import ( + CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS, + AccessExplanationSubjectDecision, + PrincipalRef, +) +from govoplan_access.backend.api.v1.routes import ( + _access_explanation_subject_decision, +) + + +ROOT = pathlib.Path(__file__).resolve().parents[1] + + +def _principal() -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="user-1", + tenant_id="tenant-1", + ), + account=SimpleNamespace(id="account-1"), + user=SimpleNamespace(id="user-1"), + ) + + +class _SubjectPolicy: + def decide_subject_selection( + self, + session: object, + principal: PrincipalRef, + *, + tenant_id: str, + ) -> AccessExplanationSubjectDecision: + del session, principal, tenant_id + return AccessExplanationSubjectDecision( + allow_other_users=True, + reason="Permitted by test policy.", + source="test.policy", + ) + + +class _Registry: + def __init__(self, provider: object | None = None) -> None: + self.provider = provider + + def has_capability(self, name: str) -> bool: + return ( + name == CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS + and self.provider is not None + ) + + def require_capability(self, name: str) -> object: + if not self.has_capability(name): + raise KeyError(name) + return self.provider + + +class ResourceAccessExplanationSubjectTests(unittest.TestCase): + def test_missing_policy_defaults_to_current_user(self) -> None: + with patch( + "govoplan_access.backend.api.v1.routes.get_registry", + return_value=None, + ): + decision = _access_explanation_subject_decision( + object(), # type: ignore[arg-type] + _principal(), + tenant_id="tenant-1", + ) + + self.assertFalse(decision.allow_other_users) + self.assertEqual("access.safe_default", decision.source) + + def test_policy_capability_controls_cross_user_selection(self) -> None: + with patch( + "govoplan_access.backend.api.v1.routes.get_registry", + return_value=_Registry(_SubjectPolicy()), + ): + decision = _access_explanation_subject_decision( + object(), # type: ignore[arg-type] + _principal(), + tenant_id="tenant-1", + ) + + self.assertTrue(decision.allow_other_users) + self.assertEqual("test.policy", decision.source) + + def test_route_contract_is_tenant_bounded_and_audited(self) -> None: + source = ( + ROOT / "src/govoplan_access/backend/api/v1/routes.py" + ).read_text(encoding="utf-8") + + self.assertIn('User.tenant_id == tenant.id', source) + self.assertIn('User.id == principal.membership_id', source) + self.assertIn( + 'action="access.resource_explanation.selected_user_viewed"', + source, + ) + + +if __name__ == "__main__": + unittest.main()