2 Commits
Author SHA1 Message Date
zemion 5b8baa6cde Make tenancy an optional access integration 2026-07-11 00:40:09 +02:00
zemion d6a0d6241d Expose resource access explanation API 2026-07-11 00:39:39 +02:00
8 changed files with 252 additions and 6 deletions
+21
View File
@@ -64,6 +64,27 @@ Access declares tenancy as an optional module integration. It uses the
core-owned `core_scopes` table as the scope table, but it must not import
`govoplan_tenancy` or require the tenancy package to start.
## Core-Only Startup Contract
A core-only installation must be able to start far enough to expose process
health, module metadata, and the unauthenticated shell needed for installation
or recovery work. It is not a usable authenticated product installation.
Authenticated product use requires the `access` module or another module that
provides the same kernel auth capabilities:
- `auth.apiPrincipalProvider`
- `auth.principalResolver`
- `auth.permissionEvaluator`
- `auth.tenantContextSwitcher`
Access contributes the default implementations for those capabilities plus the
interactive `/api/v1/auth/*` routes. Product modules should express auth needs
as required capabilities or route permission requirements instead of importing
access internals. Runtime configurations that intentionally omit access should
hide authenticated navigation and return capability errors for authenticated
product routes rather than failing process startup.
## Principal Context Contract
The stable runtime principal is `govoplan_core.core.access.PrincipalRef`.
+1 -2
View File
@@ -5,14 +5,13 @@ build-backend = "setuptools.build_meta"
[project]
name = "govoplan-access"
version = "0.1.6"
description = "GovOPlaN access platform module with identity, auth, RBAC, and tenancy primitives."
description = "GovOPlaN access platform module with identity, auth, RBAC, and scope primitives."
readme = "README.md"
requires-python = ">=3.12"
license = { file = "LICENSE" }
authors = [{ name = "GovOPlaN" }]
dependencies = [
"govoplan-core>=0.1.6",
"govoplan-tenancy>=0.1.6",
"SQLAlchemy>=2,<3",
]
@@ -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]
+83 -1
View File
@@ -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,
+25 -3
View File
@@ -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(
+15
View File
@@ -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),
)
+59
View File
@@ -0,0 +1,59 @@
from __future__ import annotations
import ast
import pathlib
import tomllib
import unittest
ROOT = pathlib.Path(__file__).resolve().parents[1]
class OptionalTenancyContractTests(unittest.TestCase):
def test_access_package_does_not_require_tenancy_to_install(self) -> None:
project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]
dependencies = tuple(project["dependencies"])
self.assertIn("govoplan-core>=0.1.6", dependencies)
self.assertNotIn("govoplan-tenancy>=0.1.6", dependencies)
self.assertFalse(any(item.startswith("govoplan-tenancy") for item in dependencies))
def test_tenancy_is_declared_as_optional_module_integration(self) -> None:
manifest_path = ROOT / "src" / "govoplan_access" / "backend" / "manifest.py"
tree = ast.parse(manifest_path.read_text(encoding="utf-8"))
manifest_call = next(
node.value
for node in ast.walk(tree)
if isinstance(node, ast.Assign)
and any(isinstance(target, ast.Name) and target.id == "manifest" for target in node.targets)
and isinstance(node.value, ast.Call)
)
optional_dependencies = next(
keyword.value
for keyword in manifest_call.keywords
if keyword.arg == "optional_dependencies"
)
self.assertIsInstance(optional_dependencies, ast.Tuple)
self.assertIn(
"tenancy",
{
item.value
for item in optional_dependencies.elts
if isinstance(item, ast.Constant) and isinstance(item.value, str)
},
)
def test_access_source_does_not_import_tenancy_module_internals(self) -> None:
offenders: list[str] = []
for path in (ROOT / "src" / "govoplan_access").rglob("*.py"):
source = path.read_text(encoding="utf-8")
if "govoplan_tenancy" in source:
offenders.append(str(path.relative_to(ROOT)))
self.assertEqual([], offenders)
if __name__ == "__main__":
unittest.main()
+31
View File
@@ -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<string, unknown>;
};
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<ResourceAccessExplanationResponse> {
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<GroupSummary[]> {
const response = await apiFetch<{ groups: GroupSummary[] }>(settings, "/api/v1/admin/groups");
return response.groups;