GovOPlaN/govoplan-access
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: secure resource explanation subjects

Browse Source
This commit is contained in:
zemion
2026-08-20 06:16:43 +02:00
parent e4bae0121d
commit 566b3b83ad
4 changed files with 310 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_access/backend/api/v1/admin_schemas.py
+15
View File
@@ -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]
src/govoplan_access/backend/api/v1/routes.py
+135 -1
View File
@@ -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,
src/govoplan_access/backend/manifest.py
+51
View File
@@ -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",
tests/test_resource_access_explanation_subjects.py
+109
View File
@@ -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()