from __future__ import annotations import shutil import tempfile import unittest from pathlib import Path from types import SimpleNamespace from fastapi import APIRouter, Depends, FastAPI from fastapi.testclient import TestClient from govoplan_access.auth import ApiPrincipal, get_api_principal from govoplan_core.core.access import ( ACCESS_CAPABILITY_NAMES, ACCESS_MODULE_ID, CAPABILITY_ACCESS_ADMINISTRATION, CAPABILITY_ACCESS_DIRECTORY, CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER, CAPABILITY_ACCESS_PERMISSION_EVALUATOR, CAPABILITY_ACCESS_PRINCIPAL_RESOLVER, CAPABILITY_ACCESS_RESOURCE_ACCESS, CAPABILITY_ACCESS_EXPLANATION, CAPABILITY_ACCESS_SEMANTIC_DIRECTORY, CAPABILITY_ACCESS_TENANT_PROVISIONER, CAPABILITY_AUDIT_RECORDER, CAPABILITY_AUDIT_RETENTION, CAPABILITY_AUDIT_SINK, CAPABILITY_TENANCY_TENANT_RESOLVER, AccessDecision, AccessAdministration, AccessDirectory, AccessDecisionProvenance, AccessExplanationService, AccessGovernanceMaterializer, ResourceAccessExplanationProvider, AccessSemanticDirectory, AccessSubjectRef, AccountRef, AuditEvent, AuditRecordRef, AuditRecorder, AuditRetentionProvider, AuditSink, CreatedApiKeyRef, DevelopmentBootstrapRef, FunctionAssignmentRef, FunctionDelegationRef, FunctionRef, GovernanceTemplateMaterialization, GroupRef, IdentityRef, OrganizationUnitRef, PermissionEvaluator, PrincipalRef, PrincipalResolver, ResourceAccessService, RoleRef, TenantRef, TenantAccessProvisioner, TenantOwnerCandidateRef, TenantResolver, UserRef, ) from govoplan_core.security.secrets import CAPABILITY_SECURITY_SECRET_PROVIDER, SecretProvider from govoplan_core.core.campaigns import ( CAPABILITY_CAMPAIGNS_ACCESS, CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT, CAPABILITY_CAMPAIGNS_POLICY_CONTEXT, CAPABILITY_CAMPAIGNS_RETENTION, CampaignAccessProvider, CampaignDeliveryTaskProvider, CampaignMailPolicyContext, CampaignMailPolicyContextProvider, CampaignPolicyContext, CampaignPolicyContextProvider, CampaignRetentionProvider, ) from govoplan_core.core.files import CAPABILITY_FILES_ACCESS, FileAccessProvider from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, OrganizationFunctionAssignmentRef from govoplan_core.core.organizations import CAPABILITY_ORGANIZATION_DIRECTORY, OrganizationDirectory as CoreOrganizationDirectory, OrganizationFunctionRef as CoreOrganizationFunctionRef, OrganizationUnitRef as CoreOrganizationUnitRef from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.runtime import clear_runtime, configure_runtime from govoplan_core.db.base import Base from govoplan_core.server.app import create_app from govoplan_core.server.config import GovoplanServerConfig from govoplan_core.tenancy.scope import create_scope_tables from govoplan_access.backend.db.models import ( Account, ExternalFunctionRoleAssignment, Function, FunctionAssignment, FunctionRoleAssignment, Group, Identity, IdentityAccountLink, OrganizationUnit, Role, User, UserGroupMembership, ) from govoplan_tenancy.backend.db.models import Tenant from tests.db_isolation import temporary_database class _FakeAccessDirectory: account = AccountRef(id="account-1", email="demo@example.test", display_name="Demo") user = UserRef(id="user-1", account_id=account.id, tenant_id="tenant-1", email=account.email) group = GroupRef(id="group-1", tenant_id="tenant-1", name="Team") def get_account(self, account_id: str) -> AccountRef | None: return self.account if account_id == self.account.id else None def get_user(self, user_id: str) -> UserRef | None: return self.user if user_id == self.user.id else None def get_users(self, user_ids): return {self.user.id: self.user} if self.user.id in set(user_ids) else {} def users_for_tenant(self, tenant_id: str): return (self.user,) if tenant_id == self.user.tenant_id else () def get_group(self, group_id: str) -> GroupRef | None: return self.group if group_id == self.group.id else None def get_groups(self, group_ids): return {self.group.id: self.group} if self.group.id in set(group_ids) else {} def groups_for_tenant(self, tenant_id: str): return (self.group,) if tenant_id == self.group.tenant_id else () def groups_for_user(self, user_id: str, *, tenant_id: str): if user_id == self.user.id and tenant_id == self.user.tenant_id: return (self.group,) return () def display_label(self, subject: AccessSubjectRef) -> str | None: if subject.kind == "user" and subject.id == self.user.id: return self.user.display_name or self.user.email if subject.kind == "group" and subject.id == self.group.id: return self.group.name return subject.label class _FakeAccessSemanticDirectory(_FakeAccessDirectory): identity = IdentityRef( id="identity-1", display_name="Demo Person", primary_account_id=_FakeAccessDirectory.account.id, account_ids=(_FakeAccessDirectory.account.id, "account-2"), ) organization_unit = OrganizationUnitRef(id="ou-1", tenant_id="tenant-1", name="Registry Office") function = FunctionRef( id="function-1", tenant_id="tenant-1", organization_unit_id=organization_unit.id, slug="registry-clerk", name="Registry Clerk", role_ids=("role-1",), delegable=True, act_in_place_allowed=True, ) assignment = FunctionAssignmentRef( id="assignment-1", tenant_id="tenant-1", account_id=_FakeAccessDirectory.account.id, identity_id=identity.id, function_id=function.id, organization_unit_id=organization_unit.id, applies_to_subunits=True, ) def get_identity(self, identity_id: str) -> IdentityRef | None: return self.identity if identity_id == self.identity.id else None def accounts_for_identity(self, identity_id: str): return (self.account,) if identity_id == self.identity.id else () def get_organization_unit(self, organization_unit_id: str) -> OrganizationUnitRef | None: return self.organization_unit if organization_unit_id == self.organization_unit.id else None def organization_units_for_tenant(self, tenant_id: str): return (self.organization_unit,) if tenant_id == self.organization_unit.tenant_id else () def get_function(self, function_id: str) -> FunctionRef | None: return self.function if function_id == self.function.id else None def functions_for_organization_unit(self, organization_unit_id: str, *, include_subunits: bool = False): del include_subunits return (self.function,) if organization_unit_id == self.organization_unit.id else () def get_function_assignment(self, assignment_id: str) -> FunctionAssignmentRef | None: return self.assignment if assignment_id == self.assignment.id else None def function_assignments_for_account(self, account_id: str, *, tenant_id: str | None = None): if account_id != self.account.id: return () if tenant_id is not None and tenant_id != self.assignment.tenant_id: return () return (self.assignment,) class _FakePrincipalResolver: def resolve_request(self, request: object, *, session: object | None = None) -> PrincipalRef: del request, session return PrincipalRef( account_id="account-1", membership_id="user-1", tenant_id="tenant-1", scopes=frozenset({"files:file:read"}), group_ids=frozenset({"group-1"}), ) class _FakeTenantResolver: tenant = TenantRef(id="tenant-1", name="Tenant") def get_tenant(self, tenant_id: str) -> TenantRef | None: return self.tenant if tenant_id == self.tenant.id else None def current_tenant(self, principal: PrincipalRef) -> TenantRef | None: return self.get_tenant(principal.tenant_id or "") class _FakePermissionEvaluator: def scopes_grant(self, scopes, required_scope: str) -> bool: return required_scope in set(scopes) def has_scope(self, principal: PrincipalRef, required_scope: str) -> bool: return self.scopes_grant(principal.scopes, required_scope) def explain_scope(self, principal: PrincipalRef, required_scope: str) -> AccessDecision: allowed = self.has_scope(principal, required_scope) return AccessDecision(allowed=allowed, reason=None if allowed else "missing scope", requirements=(required_scope,)) class _FakeResourceAccessService: def explain(self, principal: PrincipalRef, *, resource_type: str, resource_id: str, action: str) -> AccessDecision: del principal, resource_type, resource_id, action return AccessDecision(allowed=True) class _FakeResourceAccessExplanationProvider: def explain_resource_provenance( self, session: object, principal: PrincipalRef, *, resource_type: str, resource_id: str, action: str, ): del session, principal, resource_type, action return (AccessDecisionProvenance(kind="resource", id=resource_id, label="Demo resource"),) class _FakeFileAccessProvider(_FakeResourceAccessExplanationProvider): pass class _FakeAccessExplanationService: provenance = ( AccessDecisionProvenance(kind="identity", id="identity-1", label="Demo Person"), AccessDecisionProvenance(kind="function", id="assignment-1", label="Registry Clerk", tenant_id="tenant-1"), AccessDecisionProvenance(kind="role", id="role-1", label="Clerk", tenant_id="tenant-1"), AccessDecisionProvenance(kind="right", id="files:file:read", label="Read files"), ) def explain_scope_provenance(self, principal: PrincipalRef, required_scope: str): del principal, required_scope return self.provenance def explain_resource_provenance( self, principal: PrincipalRef, *, resource_type: str, resource_id: str, action: str, ): del principal, resource_type, resource_id, action return self.provenance class _FakeTenantAccessProvisioner: def ensure_default_roles(self, session: object, tenant: object | None = None): del session, tenant return {} def tenant_owner_candidates(self, session: object): del session return (TenantOwnerCandidateRef(account_id="account-1", email="demo@example.test", display_name="Demo"),) def ensure_tenant_owner_membership(self, session: object, *, tenant: object, owner_account_id: str) -> str: del session, tenant, owner_account_id return "user-1" def ensure_development_admin(self, session: object, *, tenant: object, user_email: str, user_password: str, api_key_secret: str | None, scopes): del session, tenant, user_password, scopes return DevelopmentBootstrapRef( user=UserRef(id="user-1", account_id="account-1", tenant_id="tenant-1", email=user_email), created_api_key=CreatedApiKeyRef(id="api-key-1", secret=api_key_secret) if api_key_secret else None, ) class _FakeAccessAdministration: def tenant_counts(self, session: object, tenant_id: str): del session, tenant_id return {"users": 1, "active_users": 1, "groups": 2, "api_keys": 3} def system_account_count(self, session: object) -> int: del session return 1 def role_count_for_tenant(self, session: object, tenant_id: str) -> int: del session, tenant_id return 2 def active_api_key_count_for_tenant(self, session: object, tenant_id: str) -> int: del session, tenant_id return 3 def actor_email_by_user_id(self, session: object, user_ids): del session return {user_id: "demo@example.test" for user_id in user_ids} def user_ids_for_actor_filter(self, session: object, *, operator: str, value: str): del session, operator, value return ("user-1",) def user_settings(self, session: object, user_id: str, *, tenant_id: str): del session, user_id, tenant_id return {"privacy_retention_policy": {"audit_detail_level": "minimal"}} def set_user_settings(self, session: object, user_id: str, *, tenant_id: str, settings): del session, user_id, tenant_id return dict(settings) def group_settings(self, session: object, group_id: str, *, tenant_id: str): del session, group_id, tenant_id return {"privacy_retention_policy": {"audit_detail_level": "redacted"}} def set_group_settings(self, session: object, group_id: str, *, tenant_id: str, settings): del session, group_id, tenant_id return dict(settings) class _FakeAccessGovernanceMaterializer: def __init__(self) -> None: self.synced: list[GovernanceTemplateMaterialization] = [] self.removed: list[GovernanceTemplateMaterialization] = [] def sync_template(self, session: object, template: GovernanceTemplateMaterialization) -> None: del session self.synced.append(template) def remove_template(self, session: object, template: GovernanceTemplateMaterialization) -> None: del session self.removed.append(template) class _FakeCampaignMailPolicyContextProvider: context = CampaignMailPolicyContext( id="campaign-1", tenant_id="tenant-1", owner_user_id="user-1", owner_group_id="group-1", mail_profile_policy={"allow_campaign_profiles": False}, ) def get_campaign_mail_policy_context(self, session: object, *, tenant_id: str, campaign_id: str): del session if tenant_id == self.context.tenant_id and campaign_id == self.context.id: return self.context return None def set_campaign_mail_profile_policy(self, session: object, *, tenant_id: str, campaign_id: str, policy): del session if tenant_id != self.context.tenant_id or campaign_id != self.context.id: raise ValueError("Campaign policy not found") return dict(policy) class _FakeCampaignAccessProvider: def campaign_exists(self, session: object, *, tenant_id: str, campaign_id: str) -> bool: del session return tenant_id == "tenant-1" and campaign_id == "campaign-1" def can_read_campaign( self, session: object, *, tenant_id: str, campaign_id: str, user_id: str, group_ids=(), tenant_admin: bool = False, ) -> bool: del session return self.campaign_exists(object(), tenant_id=tenant_id, campaign_id=campaign_id) and ( tenant_admin or user_id == "user-1" or "group-1" in set(group_ids) ) class _FakeCampaignPolicyContextProvider: context = CampaignPolicyContext( id="campaign-1", tenant_id="tenant-1", owner_user_id="user-1", owner_group_id="group-1", settings={"privacy_retention_policy": {"audit_detail_level": "redacted"}}, ) def get_campaign_policy_context(self, session: object, *, campaign_id: str, tenant_id: str | None = None): del session if campaign_id != self.context.id or (tenant_id is not None and tenant_id != self.context.tenant_id): return None return self.context def set_campaign_settings(self, session: object, *, tenant_id: str, campaign_id: str, settings): del session if tenant_id != self.context.tenant_id or campaign_id != self.context.id: raise ValueError("Campaign privacy policy not found") return dict(settings) class _FakeCampaignDeliveryTaskProvider: def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True): del session return {"job_id": job_id, "enqueue_imap_task": enqueue_imap_task} def append_sent_for_job(self, session: object, *, job_id: str): del session return {"job_id": job_id, "status": "appended"} class _FakeCampaignRetentionProvider: def apply_retention(self, session: object, *, dry_run: bool, now, policy_for_campaign_id): del session, now policy_for_campaign_id("campaign-1") return {"raw_campaign_json": {"eligible": int(dry_run)}} class _FakeSecretProvider: def __init__(self) -> None: self._values: dict[str, str] = {} def store_secret(self, *, scope: str, name: str, value: str) -> str: ref = f"{scope}/{name}" self._values[ref] = value return ref def read_secret(self, secret_ref: str) -> str | None: return self._values.get(secret_ref) def delete_secret(self, secret_ref: str) -> None: self._values.pop(secret_ref, None) class _FakeAuditSink: def __init__(self) -> None: self.events: list[AuditEvent] = [] def record(self, event: AuditEvent) -> None: self.events.append(event) class _FakeAuditRecorder: def record_event(self, session: object, event: AuditEvent) -> AuditRecordRef: del session return AuditRecordRef( id="audit-1", scope=event.scope, tenant_id=event.tenant_id, user_id=event.user_id, api_key_id=event.api_key_id, action=event.action, object_type=event.resource_type, object_id=event.resource_id, details=event.details, ) class _FakeAuditRetentionProvider: def apply_detail_retention(self, session: object, *, dry_run: bool, now, policy_for_record): del session, dry_run, now, policy_for_record return {"eligible": 1, "redacted": 0, "already_redacted": 0} class AccessContractTests(unittest.TestCase): def test_access_capability_names_are_stable(self) -> None: self.assertEqual("access", ACCESS_MODULE_ID) self.assertEqual("access.principalResolver", CAPABILITY_ACCESS_PRINCIPAL_RESOLVER) self.assertEqual("access.directory", CAPABILITY_ACCESS_DIRECTORY) self.assertEqual("access.permissionEvaluator", CAPABILITY_ACCESS_PERMISSION_EVALUATOR) self.assertEqual("access.resourceAccess", CAPABILITY_ACCESS_RESOURCE_ACCESS) self.assertEqual("access.semanticDirectory", CAPABILITY_ACCESS_SEMANTIC_DIRECTORY) self.assertEqual("access.explanation", CAPABILITY_ACCESS_EXPLANATION) self.assertEqual("access.tenantProvisioner", CAPABILITY_ACCESS_TENANT_PROVISIONER) self.assertEqual("access.administration", CAPABILITY_ACCESS_ADMINISTRATION) self.assertEqual("access.governanceMaterializer", CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER) self.assertEqual("campaigns.access", CAPABILITY_CAMPAIGNS_ACCESS) self.assertEqual("files.access", CAPABILITY_FILES_ACCESS) self.assertEqual("campaigns.deliveryTasks", CAPABILITY_CAMPAIGNS_DELIVERY_TASKS) self.assertEqual("campaigns.mailPolicyContext", CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT) self.assertEqual("campaigns.policyContext", CAPABILITY_CAMPAIGNS_POLICY_CONTEXT) self.assertEqual("campaigns.retention", CAPABILITY_CAMPAIGNS_RETENTION) self.assertEqual("tenancy.tenantResolver", CAPABILITY_TENANCY_TENANT_RESOLVER) self.assertEqual("security.secretProvider", CAPABILITY_SECURITY_SECRET_PROVIDER) self.assertEqual("audit.sink", CAPABILITY_AUDIT_SINK) self.assertEqual("audit.recorder", CAPABILITY_AUDIT_RECORDER) self.assertEqual("audit.retention", CAPABILITY_AUDIT_RETENTION) self.assertEqual(len(ACCESS_CAPABILITY_NAMES), len(set(ACCESS_CAPABILITY_NAMES))) def test_access_refs_are_small_immutable_dtos(self) -> None: principal = PrincipalRef( account_id="account-1", membership_id="user-1", tenant_id="tenant-1", identity_id="identity-1", scopes=frozenset({"campaigns:campaign:read"}), group_ids=frozenset({"group-1"}), role_ids=frozenset({"role-1"}), function_assignment_ids=frozenset({"assignment-1"}), delegation_ids=frozenset({"delegation-1"}), acting_for_account_id="account-2", email="demo@example.test", ) role = RoleRef(id="role-1", name="Reviewer", level="tenant", tenant_id="tenant-1") self.assertEqual("account-1", principal.account_id) self.assertEqual(frozenset({"campaigns:campaign:read"}), principal.scopes) self.assertEqual("Reviewer", role.name) payload = principal.to_dict() self.assertEqual(["campaigns:campaign:read"], payload["scopes"]) self.assertEqual(["group-1"], payload["group_ids"]) self.assertEqual(["role-1"], payload["role_ids"]) self.assertEqual(["assignment-1"], payload["function_assignment_ids"]) self.assertEqual(["delegation-1"], payload["delegation_ids"]) self.assertEqual("identity-1", payload["identity_id"]) self.assertEqual("account-2", payload["acting_for_account_id"]) self.assertEqual("session", payload["auth_method"]) self.assertIsNone(payload["api_key_id"]) self.assertIsNone(payload["service_account_id"]) self.assertEqual(principal, PrincipalRef.from_mapping(payload)) with self.assertRaises(AttributeError): principal.account_id = "changed" # type: ignore[misc] def test_identity_function_and_delegation_refs_model_organization_scope(self) -> None: identity = IdentityRef( id="identity-1", display_name="Demo Person", primary_account_id="account-1", account_ids=("account-1", "account-2"), ) organization_unit = OrganizationUnitRef(id="ou-1", tenant_id="tenant-1", name="Registry Office") function = FunctionRef( id="function-1", tenant_id="tenant-1", organization_unit_id=organization_unit.id, slug="registry-clerk", name="Registry Clerk", role_ids=("role-1",), delegable=True, act_in_place_allowed=True, ) assignment = FunctionAssignmentRef( id="assignment-1", tenant_id="tenant-1", account_id="account-1", identity_id=identity.id, function_id=function.id, organization_unit_id=organization_unit.id, applies_to_subunits=True, ) delegation = FunctionDelegationRef( id="delegation-1", tenant_id="tenant-1", function_assignment_id=assignment.id, delegator_account_id="account-1", delegate_account_id="account-2", mode="act_in_place", ) provenance = AccessDecisionProvenance( kind="function", id=assignment.id, label=function.name, tenant_id="tenant-1", source="delegation", details={"organization_unit_id": organization_unit.id, "applies_to_subunits": True}, ) self.assertEqual(("account-1", "account-2"), identity.account_ids) self.assertEqual("ou-1", function.organization_unit_id) self.assertTrue(assignment.applies_to_subunits) self.assertEqual("act_in_place", delegation.mode) self.assertEqual("function", provenance.to_dict()["kind"]) self.assertEqual({"organization_unit_id": "ou-1", "applies_to_subunits": True}, provenance.to_dict()["details"]) def test_protocol_shapes_match_reference_implementations(self) -> None: self.assertIsInstance(_FakePrincipalResolver(), PrincipalResolver) self.assertIsInstance(_FakeTenantResolver(), TenantResolver) self.assertIsInstance(_FakeAccessDirectory(), AccessDirectory) self.assertIsInstance(_FakeAccessSemanticDirectory(), AccessSemanticDirectory) self.assertIsInstance(_FakePermissionEvaluator(), PermissionEvaluator) self.assertIsInstance(_FakeResourceAccessService(), ResourceAccessService) self.assertIsInstance(_FakeResourceAccessExplanationProvider(), ResourceAccessExplanationProvider) self.assertIsInstance(_FakeFileAccessProvider(), FileAccessProvider) self.assertIsInstance(_FakeAccessExplanationService(), AccessExplanationService) self.assertIsInstance(_FakeTenantAccessProvisioner(), TenantAccessProvisioner) self.assertIsInstance(_FakeAccessAdministration(), AccessAdministration) self.assertIsInstance(_FakeAccessGovernanceMaterializer(), AccessGovernanceMaterializer) self.assertIsInstance(_FakeCampaignAccessProvider(), CampaignAccessProvider) self.assertIsInstance(_FakeCampaignDeliveryTaskProvider(), CampaignDeliveryTaskProvider) self.assertIsInstance(_FakeCampaignMailPolicyContextProvider(), CampaignMailPolicyContextProvider) self.assertIsInstance(_FakeCampaignPolicyContextProvider(), CampaignPolicyContextProvider) self.assertIsInstance(_FakeCampaignRetentionProvider(), CampaignRetentionProvider) self.assertIsInstance(_FakeSecretProvider(), SecretProvider) self.assertIsInstance(_FakeAuditSink(), AuditSink) self.assertIsInstance(_FakeAuditRecorder(), AuditRecorder) self.assertIsInstance(_FakeAuditRetentionProvider(), AuditRetentionProvider) def test_campaign_mail_policy_context_capability_shape(self) -> None: provider = _FakeCampaignMailPolicyContextProvider() context = provider.get_campaign_mail_policy_context(object(), tenant_id="tenant-1", campaign_id="campaign-1") self.assertEqual("campaign-1", context.id) # type: ignore[union-attr] self.assertEqual("user-1", context.owner_user_id) # type: ignore[union-attr] self.assertEqual({"allow_campaign_profiles": False}, context.mail_profile_policy) # type: ignore[union-attr] self.assertEqual({"allow_campaign_profiles": True}, provider.set_campaign_mail_profile_policy(object(), tenant_id="tenant-1", campaign_id="campaign-1", policy={"allow_campaign_profiles": True})) def test_campaign_access_capability_shape(self) -> None: provider = _FakeCampaignAccessProvider() self.assertTrue(provider.campaign_exists(object(), tenant_id="tenant-1", campaign_id="campaign-1")) self.assertTrue(provider.can_read_campaign(object(), tenant_id="tenant-1", campaign_id="campaign-1", user_id="other", group_ids=("group-1",))) self.assertFalse(provider.can_read_campaign(object(), tenant_id="tenant-1", campaign_id="missing", user_id="user-1")) def test_campaign_policy_delivery_and_retention_capability_shapes(self) -> None: policy_provider = _FakeCampaignPolicyContextProvider() delivery_provider = _FakeCampaignDeliveryTaskProvider() retention_provider = _FakeCampaignRetentionProvider() context = policy_provider.get_campaign_policy_context(object(), tenant_id="tenant-1", campaign_id="campaign-1") self.assertEqual("tenant-1", context.tenant_id) # type: ignore[union-attr] self.assertEqual({"privacy_retention_policy": {"audit_detail_level": "redacted"}}, context.settings) # type: ignore[union-attr] self.assertEqual({"demo": True}, policy_provider.set_campaign_settings(object(), tenant_id="tenant-1", campaign_id="campaign-1", settings={"demo": True})) self.assertEqual({"job_id": "job-1", "enqueue_imap_task": False}, delivery_provider.send_campaign_job(object(), job_id="job-1", enqueue_imap_task=False)) self.assertEqual({"job_id": "job-1", "status": "appended"}, delivery_provider.append_sent_for_job(object(), job_id="job-1")) self.assertEqual({"raw_campaign_json": {"eligible": 1}}, retention_provider.apply_retention(object(), dry_run=True, now=object(), policy_for_campaign_id=lambda campaign_id: object())) def test_access_capabilities_register_and_resolve_through_platform_registry(self) -> None: directory = _FakeAccessDirectory() semantic_directory = _FakeAccessSemanticDirectory() explanation = _FakeAccessExplanationService() resolver = _FakePrincipalResolver() evaluator = _FakePermissionEvaluator() tenant_resolver = _FakeTenantResolver() administration = _FakeAccessAdministration() materializer = _FakeAccessGovernanceMaterializer() manifest = ModuleManifest( id=ACCESS_MODULE_ID, name="Access", version="test", capability_factories={ CAPABILITY_ACCESS_DIRECTORY: lambda context: directory, CAPABILITY_ACCESS_SEMANTIC_DIRECTORY: lambda context: semantic_directory, CAPABILITY_ACCESS_EXPLANATION: lambda context: explanation, CAPABILITY_ACCESS_PRINCIPAL_RESOLVER: lambda context: resolver, CAPABILITY_ACCESS_PERMISSION_EVALUATOR: lambda context: evaluator, CAPABILITY_ACCESS_ADMINISTRATION: lambda context: administration, CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER: lambda context: materializer, CAPABILITY_TENANCY_TENANT_RESOLVER: lambda context: tenant_resolver, }, ) registry = PlatformRegistry() registry.register(manifest) registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) self.assertIs(directory, registry.require_capability(CAPABILITY_ACCESS_DIRECTORY)) self.assertIs(semantic_directory, registry.require_capability(CAPABILITY_ACCESS_SEMANTIC_DIRECTORY)) self.assertIs(explanation, registry.require_capability(CAPABILITY_ACCESS_EXPLANATION)) self.assertIs(resolver, registry.require_capability(CAPABILITY_ACCESS_PRINCIPAL_RESOLVER)) self.assertIs(evaluator, registry.require_capability(CAPABILITY_ACCESS_PERMISSION_EVALUATOR)) self.assertIs(administration, registry.require_capability(CAPABILITY_ACCESS_ADMINISTRATION)) self.assertIs(materializer, registry.require_capability(CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER)) self.assertIs(tenant_resolver, registry.require_capability(CAPABILITY_TENANCY_TENANT_RESOLVER)) def test_sql_directory_and_tenant_resolver_return_access_dtos(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-access-directory-")) try: with temporary_database(f"sqlite:///{root / 'directory.db'}") as database: create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.session() as session: tenant = Tenant(id="tenant-directory", slug="directory", name="Directory Tenant", settings={}) account = Account( id="account-directory", email="directory@example.test", normalized_email="directory@example.test", display_name="Directory User", ) user = User( id="user-directory", tenant_id=tenant.id, account_id=account.id, email=account.email, display_name=None, settings={}, mail_profile_policy={}, ) group = Group(id="group-directory", tenant_id=tenant.id, slug="review", name="Review", description=None) membership = UserGroupMembership(tenant_id=tenant.id, user_id=user.id, group_id=group.id) identity = Identity(id="identity-directory", display_name="Directory Person", source="local") identity_link = IdentityAccountLink( identity_id=identity.id, account_id=account.id, is_primary=True, source="local", ) organization_unit = OrganizationUnit( id="ou-directory", tenant_id=tenant.id, slug="registry", name="Registry", ) role = Role( id="role-directory", tenant_id=tenant.id, slug="registry-reader", name="Registry Reader", permissions=["files:read"], ) function = Function( id="function-directory", tenant_id=tenant.id, organization_unit_id=organization_unit.id, slug="registry-clerk", name="Registry Clerk", delegable=True, ) function_role = FunctionRoleAssignment(tenant_id=tenant.id, function_id=function.id, role_id=role.id) function_assignment = FunctionAssignment( id="assignment-directory", tenant_id=tenant.id, account_id=account.id, identity_id=identity.id, function_id=function.id, organization_unit_id=organization_unit.id, applies_to_subunits=True, ) session.add_all([ tenant, account, user, group, membership, identity, identity_link, organization_unit, role, function, function_role, function_assignment, ]) session.commit() from govoplan_access.backend.directory import SqlAccessDirectory from govoplan_access.backend.security.sessions import collect_user_scopes from govoplan_tenancy.backend.capabilities import SqlTenantResolver directory = SqlAccessDirectory() tenant_resolver = SqlTenantResolver() self.assertEqual("directory@example.test", directory.get_account("account-directory").email) # type: ignore[union-attr] self.assertEqual("Directory User", directory.get_user("user-directory").display_name) # type: ignore[union-attr] self.assertEqual(["user-directory"], [user.id for user in directory.users_for_tenant("tenant-directory")]) self.assertEqual(["group-directory"], [group.id for group in directory.groups_for_tenant("tenant-directory")]) self.assertEqual(["group-directory"], [group.id for group in directory.groups_for_user("user-directory", tenant_id="tenant-directory")]) self.assertEqual("Review", directory.display_label(AccessSubjectRef(kind="group", id="group-directory", tenant_id="tenant-directory"))) self.assertEqual("Directory Person", directory.get_identity("identity-directory").display_name) # type: ignore[union-attr] self.assertEqual(["account-directory"], [account.id for account in directory.accounts_for_identity("identity-directory")]) self.assertEqual("Registry", directory.get_organization_unit("ou-directory").name) # type: ignore[union-attr] self.assertEqual(["role-directory"], list(directory.get_function("function-directory").role_ids)) # type: ignore[union-attr] self.assertEqual( ["assignment-directory"], [item.id for item in directory.function_assignments_for_account("account-directory", tenant_id="tenant-directory")], ) with database.session() as session: user = session.get(User, "user-directory") self.assertIn("files:read", collect_user_scopes(session, user, include_system=False)) # type: ignore[arg-type] self.assertEqual("directory", tenant_resolver.get_tenant("tenant-directory").slug) # type: ignore[union-attr] self.assertEqual( "tenant-directory", tenant_resolver.current_tenant( PrincipalRef(account_id="account-directory", membership_id="user-directory", tenant_id="tenant-directory") ).id, # type: ignore[union-attr] ) finally: shutil.rmtree(root, ignore_errors=True) def test_semantic_access_admin_crud_endpoints(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-access-semantic-api-")) try: with temporary_database(f"sqlite:///{root / 'semantic.db'}") as database: create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) tenant_id = "tenant-semantic" account_id = "account-semantic" delegate_account_id = "account-semantic-delegate" user_id = "user-semantic" role_id = "role-semantic" with database.session() as session: tenant = Tenant(id=tenant_id, slug="semantic", name="Semantic Tenant", settings={}) account = Account( id=account_id, email="semantic@example.test", normalized_email="semantic@example.test", display_name="Semantic User", ) delegate_account = Account( id=delegate_account_id, email="delegate@example.test", normalized_email="delegate@example.test", display_name="Delegate User", ) user = User( id=user_id, tenant_id=tenant_id, account_id=account_id, email=account.email, display_name=account.display_name, settings={}, mail_profile_policy={}, ) role = Role( id=role_id, tenant_id=tenant_id, slug="semantic-reader", name="Semantic Reader", permissions=["files:read"], ) session.add_all([tenant, account, delegate_account, user, role]) session.commit() from govoplan_access.backend.api.v1.routes import router as admin_router app = FastAPI() app.include_router(admin_router) def principal_override() -> ApiPrincipal: with database.session() as session: account = session.get(Account, account_id) user = session.get(User, user_id) assert account is not None assert user is not None return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id=user_id, tenant_id=tenant_id, scopes=frozenset({ "admin:roles:read", "admin:roles:write", "admin:roles:assign", "system:accounts:read", "system:accounts:update", }), ), account=account, user=user, ) app.dependency_overrides[get_api_principal] = principal_override with TestClient(app) as client: identity_response = client.post( "/admin/identities", json={"display_name": "Semantic Person", "account_ids": [account_id], "primary_account_id": account_id}, ) self.assertEqual(201, identity_response.status_code, identity_response.text) identity_id = identity_response.json()["id"] unit_response = client.post( "/admin/organization-units", json={"slug": "registry", "name": "Registry"}, ) self.assertEqual(201, unit_response.status_code, unit_response.text) organization_unit_id = unit_response.json()["id"] function_response = client.post( "/admin/functions", json={ "organization_unit_id": organization_unit_id, "slug": "registry-clerk", "name": "Registry Clerk", "role_ids": [role_id], "delegable": True, }, ) self.assertEqual(201, function_response.status_code, function_response.text) function_id = function_response.json()["id"] self.assertEqual([role_id], function_response.json()["role_ids"]) assignment_response = client.post( "/admin/function-assignments", json={ "account_id": account_id, "identity_id": identity_id, "function_id": function_id, "applies_to_subunits": True, }, ) self.assertEqual(201, assignment_response.status_code, assignment_response.text) assignment_id = assignment_response.json()["id"] self.assertTrue(assignment_response.json()["applies_to_subunits"]) delegation_response = client.post( "/admin/function-delegations", json={"function_assignment_id": assignment_id, "delegate_account_id": delegate_account_id}, ) self.assertEqual(201, delegation_response.status_code, delegation_response.text) self.assertEqual(delegate_account_id, delegation_response.json()["delegate_account_id"]) list_response = client.get(f"/admin/function-assignments?account_id={account_id}") self.assertEqual(200, list_response.status_code, list_response.text) self.assertEqual([assignment_id], [item["id"] for item in list_response.json()["assignments"]]) finally: shutil.rmtree(root, ignore_errors=True) def test_external_function_role_mappings_require_organizations_directory(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-access-external-function-map-")) try: tenant_id = "tenant-external-function-map" account_id = "account-external-function-map" user_id = "user-external-function-map" role_id = "role-external-function-map" function_id = "function-external-function-map" with temporary_database(f"sqlite:///{root / 'external-function-map.db'}") as database: create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.session() as session: tenant = Tenant(id=tenant_id, slug="external-function-map", name="External Function Map", settings={}) account = Account( id=account_id, email="external-function-map@example.test", normalized_email="external-function-map@example.test", display_name="External Function Map Admin", ) user = User( id=user_id, tenant_id=tenant_id, account_id=account_id, email=account.email, display_name=account.display_name, settings={}, mail_profile_policy={}, ) role = Role( id=role_id, tenant_id=tenant_id, slug="external-function-reader", name="External Function Reader", permissions=["files:file:read"], is_assignable=True, ) session.add_all([tenant, account, user, role]) session.commit() from govoplan_access.backend.api.v1.routes import router as admin_router app = FastAPI() app.include_router(admin_router) def principal_override() -> ApiPrincipal: with database.session() as session: account = session.get(Account, account_id) user = session.get(User, user_id) assert account is not None assert user is not None return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id=user_id, tenant_id=tenant_id, scopes=frozenset({"admin:roles:read", "admin:roles:write", "access:role:assign"}), ), account=account, user=user, ) app.dependency_overrides[get_api_principal] = principal_override clear_runtime() with TestClient(app) as client: blocked_response = client.post( "/admin/external-function-role-mappings", json={"source_module": "organizations", "function_id": function_id, "role_id": role_id}, ) self.assertEqual(409, blocked_response.status_code, blocked_response.text) self.assertEqual("organizations_required", blocked_response.json()["detail"]["code"]) class FakeOrganizationDirectory(CoreOrganizationDirectory): def get_organization_unit(self, organization_unit_id: str): del organization_unit_id return None def organization_units_for_tenant(self, tenant_id: str): del tenant_id return () def get_function(self, requested_function_id: str): if requested_function_id != function_id: return None return CoreOrganizationFunctionRef( id=function_id, tenant_id=tenant_id, organization_unit_id="unit-external-function-map", slug="external-function", name="External Function", ) def functions_for_organization_unit(self, organization_unit_id: str, *, include_subunits: bool = False): del organization_unit_id, include_subunits return () registry = PlatformRegistry() registry.register( ModuleManifest( id="organizations", name="Organizations", version="test", capability_factories={CAPABILITY_ORGANIZATION_DIRECTORY: lambda context: FakeOrganizationDirectory()}, ) ) context = ModuleContext(registry=registry, settings=object()) registry.configure_capability_context(context) configure_runtime(context) created_response = client.post( "/admin/external-function-role-mappings", json={"source_module": "organizations", "function_id": function_id, "role_id": role_id}, ) self.assertEqual(201, created_response.status_code, created_response.text) self.assertEqual(function_id, created_response.json()["function_id"]) unsupported_response = client.post( "/admin/external-function-role-mappings", json={"source_module": "legacy", "function_id": function_id, "role_id": role_id}, ) self.assertEqual(422, unsupported_response.status_code, unsupported_response.text) finally: clear_runtime() shutil.rmtree(root, ignore_errors=True) def test_access_explanation_includes_idm_function_role_sources(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-access-explanation-")) try: tenant_id = "tenant-access-explanation" account_id = "account-access-explanation" user_id = "user-access-explanation" role_id = "role-access-explanation" function_id = "function-access-explanation" organization_unit_id = "unit-access-explanation" assignment_id = "assignment-access-explanation" identity_id = "identity-access-explanation" assignment = OrganizationFunctionAssignmentRef( id=assignment_id, tenant_id=tenant_id, identity_id=identity_id, account_id=account_id, function_id=function_id, organization_unit_id=organization_unit_id, applies_to_subunits=True, source="direct", ) class FakeIdmDirectory: def get_organization_function_assignment(self, requested_assignment_id: str): return assignment if requested_assignment_id == assignment_id else None def organization_function_assignments_for_identity(self, requested_identity_id: str, *, tenant_id: str | None = None): if requested_identity_id == identity_id and tenant_id in (None, assignment.tenant_id): return (assignment,) return () def organization_function_assignments_for_account(self, requested_account_id: str, *, tenant_id: str | None = None): if requested_account_id == account_id and tenant_id in (None, assignment.tenant_id): return (assignment,) return () class FakeOrganizationDirectory(CoreOrganizationDirectory): def get_organization_unit(self, requested_organization_unit_id: str): if requested_organization_unit_id != organization_unit_id: return None return CoreOrganizationUnitRef( id=organization_unit_id, tenant_id=tenant_id, slug="registry", name="Registry Office", ) def organization_units_for_tenant(self, requested_tenant_id: str): if requested_tenant_id != tenant_id: return () unit = self.get_organization_unit(organization_unit_id) return (unit,) if unit is not None else () def get_function(self, requested_function_id: str): if requested_function_id != function_id: return None return CoreOrganizationFunctionRef( id=function_id, tenant_id=tenant_id, organization_unit_id=organization_unit_id, slug="registry-clerk", name="Registry Clerk", ) def functions_for_organization_unit(self, requested_organization_unit_id: str, *, include_subunits: bool = False): del include_subunits if requested_organization_unit_id != organization_unit_id: return () function = self.get_function(function_id) return (function,) if function is not None else () idm_directory = FakeIdmDirectory() organization_directory = FakeOrganizationDirectory() class FakeFileAccessProvider: def explain_resource_provenance( self, session: object, principal: PrincipalRef, *, resource_type: str, resource_id: str, action: str, ): del session, principal, action if resource_type != "file": return () return ( AccessDecisionProvenance( kind="resource", id=resource_id, label="Registry.pdf", tenant_id=tenant_id, source="files.file", details={"resource_type": "file"}, ), ) resource_provider = FakeFileAccessProvider() with temporary_database(f"sqlite:///{root / 'test.db'}") as database: create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.session() as session: tenant = Tenant(id=tenant_id, slug="access-explanation", name="Access Explanation", settings={}) account = Account( id=account_id, email="explanation@example.test", normalized_email="explanation@example.test", display_name="Access Explanation User", ) user = User( id=user_id, tenant_id=tenant_id, account_id=account_id, email=account.email, display_name=account.display_name, settings={}, mail_profile_policy={}, ) role = Role( id=role_id, tenant_id=tenant_id, slug="file-reader", name="File Reader", permissions=["files:file:read"], ) mapping = ExternalFunctionRoleAssignment( tenant_id=tenant_id, source_module="organizations", function_id=function_id, role_id=role_id, ) session.add_all([tenant, account, user, role, mapping]) session.commit() from govoplan_access.backend.explanation import SqlAccessExplanationService, build_user_access_explanation explanation = build_user_access_explanation( session, user, idm_directory=idm_directory, organization_directory=organization_directory, ) self.assertEqual(1, len(explanation.function_facts)) self.assertEqual("Registry Clerk", explanation.function_facts[0].function_name) self.assertEqual((role_id,), explanation.function_facts[0].role_ids) source = next(item for item in explanation.role_sources if item.source_type == "idm_function_role") self.assertEqual("Registry Clerk", source.function_name) self.assertEqual("File Reader", source.role_name) scope = next(item for item in explanation.scopes if item.scope == "files:file:read") self.assertEqual(["idm_function_role"], [item.source_type for item in scope.sources]) service = SqlAccessExplanationService( idm_directory=idm_directory, organization_directory=organization_directory, resource_explanation_providers=(resource_provider,), ) provenance = service.explain_scope_provenance( PrincipalRef( account_id=account_id, membership_id=user_id, tenant_id=tenant_id, identity_id=identity_id, scopes=frozenset({"files:file:read"}), ), "files:file:read", ) self.assertTrue(any(item.kind == "function" and item.label == "Registry Clerk" for item in provenance)) self.assertTrue(any(item.kind == "role" and item.label == "File Reader" for item in provenance)) resource_provenance = service.explain_resource_provenance( PrincipalRef( account_id=account_id, membership_id=user_id, tenant_id=tenant_id, identity_id=identity_id, scopes=frozenset({"files:file:read"}), ), resource_type="file", resource_id="file-access-explanation", action="files:file:read", ) self.assertTrue(any(item.kind == "resource" and item.id == "file-access-explanation" for item in resource_provenance)) self.assertTrue(any(item.kind == "right" and item.id == "files:file:read" for item in resource_provenance)) registry = PlatformRegistry() registry.register( ModuleManifest( id="access", name="Access", version="test", capability_factories={CAPABILITY_ACCESS_EXPLANATION: lambda context: service}, ) ) registry.register( ModuleManifest( id="idm", name="IDM", version="test", capability_factories={CAPABILITY_IDM_DIRECTORY: lambda context: idm_directory}, ) ) registry.register( ModuleManifest( id="organizations", name="Organizations", version="test", capability_factories={CAPABILITY_ORGANIZATION_DIRECTORY: lambda context: organization_directory}, ) ) context = ModuleContext(registry=registry, settings=object()) registry.configure_capability_context(context) configure_runtime(context) from govoplan_access.backend.api.v1.routes import router as admin_router app = FastAPI() app.include_router(admin_router) def principal_override() -> ApiPrincipal: with database.session() as session: account = session.get(Account, account_id) user = session.get(User, user_id) assert account is not None assert user is not None return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id=user_id, tenant_id=tenant_id, scopes=frozenset({"admin:users:read"}), ), account=account, user=user, ) app.dependency_overrides[get_api_principal] = principal_override with TestClient(app) as client: response = client.get(f"/admin/users/{user_id}/access-explanation") self.assertEqual(200, response.status_code, response.text) payload = response.json() self.assertIn("files:file:read", payload["user"]["effective_scopes"]) self.assertEqual("idm_function_role", payload["role_sources"][0]["source_type"]) self.assertEqual("Registry Clerk", payload["function_facts"][0]["function_name"]) resource_response = client.get( "/admin/access/resource-explanation", params={ "user_id": user_id, "resource_type": "file", "resource_id": "file-access-explanation", "action": "files:file:read", }, ) self.assertEqual(200, resource_response.status_code, resource_response.text) resource_payload = resource_response.json() self.assertTrue(any(item["kind"] == "resource" and item["id"] == "file-access-explanation" for item in resource_payload["provenance"])) finally: clear_runtime() shutil.rmtree(root, ignore_errors=True) def test_api_principal_dependency_uses_access_resolver_capability(self) -> None: root = Path(tempfile.mkdtemp(prefix="govoplan-access-contract-")) try: account_id = "account-capability" tenant_id = "tenant-capability" user_id = "user-capability" class CapabilityPrincipalResolver: def resolve_request(self, request: object, *, session: object | None = None) -> PrincipalRef: del request, session return PrincipalRef( account_id=account_id, membership_id=user_id, tenant_id=tenant_id, scopes=frozenset(), ) class CapabilityPermissionEvaluator: def scopes_grant(self, scopes, required_scope: str) -> bool: del scopes return required_scope == "demo:scope:read" def has_scope(self, principal: PrincipalRef, required_scope: str) -> bool: del principal return required_scope == "demo:scope:read" def explain_scope(self, principal: PrincipalRef, required_scope: str) -> AccessDecision: return AccessDecision(allowed=self.has_scope(principal, required_scope), requirements=(required_scope,)) router = APIRouter() @router.get("/capability-principal") def capability_principal(principal: ApiPrincipal = Depends(get_api_principal)) -> dict[str, object]: return { "account_id": principal.account_id, "tenant_id": principal.tenant_id, "user_id": principal.user.id, "has_demo_scope": principal.has("demo:scope:read"), } def access_manifest() -> ModuleManifest: return ModuleManifest( id=ACCESS_MODULE_ID, name="Access", version="test", capability_factories={ CAPABILITY_ACCESS_PRINCIPAL_RESOLVER: lambda context: CapabilityPrincipalResolver(), CAPABILITY_ACCESS_PERMISSION_EVALUATOR: lambda context: CapabilityPermissionEvaluator(), }, ) config = GovoplanServerConfig( title="access contract test", version="test", settings=SimpleNamespace(database_url=f"sqlite:///{root / 'test.db'}"), enabled_modules=(), manifest_factories=(access_manifest,), base_routers=(router,), post_module_routers=(), extra_routers=(), lifespan=None, app_configurators=(), ) with temporary_database(f"sqlite:///{root / 'test.db'}") as database: app = create_app(config) create_scope_tables(database.engine) Base.metadata.create_all(bind=database.engine) with database.session() as session: tenant = Tenant(id=tenant_id, slug="capability", name="Capability Tenant", settings={}) account = Account( id=account_id, email="capability@example.test", normalized_email="capability@example.test", display_name="Capability User", ) user = User( id=user_id, tenant_id=tenant_id, account_id=account_id, email=account.email, display_name=account.display_name, settings={}, mail_profile_policy={}, ) session.add_all([tenant, account, user]) session.commit() with TestClient(app) as client: response = client.get("/api/v1/capability-principal") self.assertEqual(response.status_code, 200, response.text) self.assertEqual( { "account_id": account_id, "tenant_id": tenant_id, "user_id": user_id, "has_demo_scope": True, }, response.json(), ) finally: shutil.rmtree(root, ignore_errors=True) if __name__ == "__main__": unittest.main()