Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5b8baa6cde | ||
|
|
d6a0d6241d |
@@ -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
@@ -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]
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user