From e061f230f2accda5ab12154a63bdb8946bbd78df Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 31 Jul 2026 19:40:22 +0200 Subject: [PATCH] Add function assignment governance profiles --- .../backend/function_assignment_governance.py | 304 ++++++++++++++++++ src/govoplan_policy/backend/manifest.py | 17 + tests/test_function_assignment_governance.py | 137 ++++++++ tests/test_policy_module_contract.py | 14 +- 4 files changed, 469 insertions(+), 3 deletions(-) create mode 100644 src/govoplan_policy/backend/function_assignment_governance.py create mode 100644 tests/test_function_assignment_governance.py diff --git a/src/govoplan_policy/backend/function_assignment_governance.py b/src/govoplan_policy/backend/function_assignment_governance.py new file mode 100644 index 0000000..fed2096 --- /dev/null +++ b/src/govoplan_policy/backend/function_assignment_governance.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +from collections.abc import Mapping + +from govoplan_core.core.policy import ( + FunctionAssignmentGovernanceDecision, + FunctionAssignmentGovernanceRequest, + PolicySourceStep, +) +from govoplan_core.security.module_permissions import scopes_grant_compatible + + +SUPPORTED_PROFILES = { + "holder_grant", + "holder_with_authority_clearance", + "authority_only", + "unavailable", +} + + +class FunctionAssignmentGovernancePolicyProvider: + def resolve_function_assignment_action( + self, + session: object | None = None, + *, + request: FunctionAssignmentGovernanceRequest, + ) -> FunctionAssignmentGovernanceDecision: + del session + policy = _policy(request.function_settings) + profile = ( + str(policy.get(f"{request.kind}_profile") or "unavailable") + .strip() + .casefold() + ) + if profile not in SUPPORTED_PROFILES: + return _decision( + request, + policy, + profile="unavailable", + allowed=False, + reason=f"Unsupported function assignment profile: {profile}.", + requirements=("valid_profile",), + ) + required_steps = _required_steps( + request.kind, + profile, + recipient_acceptance=_bool( + policy.get("recipient_acceptance_required"), + default=request.kind == "grant", + ), + ) + authority_function_id = _text(policy.get("authority_function_id")) + requirements: list[str] = [] + if "authority" in required_steps and authority_function_id is None: + requirements.append("authority_function") + evidence_required = _bool( + policy.get("evidence_required"), + default=False, + ) + if evidence_required and not request.context.get("has_evidence"): + requirements.append("evidence") + allowed, reason = _action_decision( + request, + profile=profile, + required_steps=required_steps, + ) + if profile == "unavailable": + allowed = False + reason = "Function assignment requests and grants are disabled." + if requirements and request.action == "submit": + allowed = False + reason = _requirements_reason(requirements) + return _decision( + request, + policy, + profile=profile, + allowed=allowed, + reason=reason, + required_steps=required_steps, + authority_function_id=authority_function_id, + evidence_required=evidence_required, + requirements=tuple(requirements), + ) + + +def _action_decision( + request: FunctionAssignmentGovernanceRequest, + *, + profile: str, + required_steps: tuple[str, ...], +) -> tuple[bool, str | None]: + context = request.context + action = request.action + if action == "submit": + if request.kind == "request": + if not bool(context.get("candidate_is_actor")): + return False, "A function request must target the requesting identity." + if profile == "authority_only": + allowed = bool(context.get("actor_is_authority")) + return ( + allowed, + None + if allowed + else "Only the designated authority may initiate this assignment.", + ) + return True, None + if profile == "authority_only": + allowed = bool(context.get("actor_is_authority")) + reason = "Only the designated authority may initiate this grant." + else: + allowed = bool(context.get("actor_is_holder")) + reason = "An effective function holder must initiate this grant." + return allowed, None if allowed else reason + if action == "approve_holder": + allowed = "holder" in required_steps and bool(context.get("actor_is_holder")) + return allowed, None if allowed else "A current holder must approve." + if action == "approve_authority": + allowed = "authority" in required_steps and bool( + context.get("actor_is_authority") + ) + return allowed, None if allowed else "The designated authority must approve." + if action == "accept_recipient": + allowed = "recipient" in required_steps and bool( + context.get("candidate_is_actor") + ) + return allowed, None if allowed else "The candidate must accept this grant." + if action in {"reject", "request_changes"}: + if request.current_state == "awaiting_holder": + allowed = bool(context.get("actor_is_holder")) + reason = "A current holder must act at the holder review step." + elif request.current_state == "awaiting_authority": + allowed = bool(context.get("actor_is_authority")) + reason = "The designated authority must act at the authority step." + elif request.current_state == "awaiting_recipient": + allowed = bool(context.get("candidate_is_actor")) + reason = "Only the candidate may act at recipient acceptance." + else: + allowed = False + reason = "The current state does not accept this review action." + return allowed, None if allowed else reason + if action == "respond": + allowed = request.current_state == "changes_requested" and ( + bool(context.get("actor_is_initiator")) + or bool(context.get("candidate_is_actor")) + ) + return ( + allowed, + None + if allowed + else "Only the initiator or candidate may respond to requested changes.", + ) + if action == "withdraw": + allowed = bool(context.get("actor_is_initiator")) + return ( + allowed, + None if allowed else "Only the initiator may withdraw this change.", + ) + if action == "recover": + allowed = _has_scope(request, "idm:function_change:admin") + return ( + allowed, + None if allowed else "Administrative recovery permission is required.", + ) + if action == "apply": + allowed = bool(context.get("approvals_complete")) + return allowed, None if allowed else "Required decisions are incomplete." + return False, f"Unsupported function assignment action: {action}." + + +def _required_steps( + kind: str, + profile: str, + *, + recipient_acceptance: bool, +) -> tuple[str, ...]: + if profile == "holder_grant": + steps = ["holder"] + elif profile == "holder_with_authority_clearance": + steps = ["holder", "authority"] + elif profile == "authority_only": + steps = ["authority"] + else: + steps = [] + if kind == "grant" and recipient_acceptance: + steps.append("recipient") + return tuple(steps) + + +def _decision( + request: FunctionAssignmentGovernanceRequest, + policy: Mapping[str, object], + *, + profile: str, + allowed: bool, + reason: str | None, + required_steps: tuple[str, ...] = (), + authority_function_id: str | None = None, + evidence_required: bool = False, + requirements: tuple[str, ...] = (), +) -> FunctionAssignmentGovernanceDecision: + recipient_required = "recipient" in required_steps + return FunctionAssignmentGovernanceDecision( + allowed=allowed, + reason=reason, + profile=profile, + required_steps=required_steps, + authority_function_id=authority_function_id, + evidence_required=evidence_required, + recipient_acceptance_required=recipient_required, + separation_of_duties=_bool( + policy.get("separation_of_duties"), + default=True, + ), + quorum=_bounded_int(policy.get("quorum"), default=1, minimum=1, maximum=20), + maximum_validity_days=_optional_positive_int( + policy.get("maximum_validity_days"), + maximum=3650, + ), + request_expiry_hours=_bounded_int( + policy.get("request_expiry_hours"), + default=336, + minimum=1, + maximum=8760, + ), + source_path=( + PolicySourceStep( + scope_type="tenant", + scope_id=request.tenant_id, + label="Organization function assignment policy", + applied_fields=tuple(sorted(policy)), + policy=policy, + ), + ), + requirements=requirements, + details={ + "function_id": request.function_id, + "kind": request.kind, + "action": request.action, + }, + ) + + +def _policy(settings: Mapping[str, object]) -> dict[str, object]: + raw = settings.get("assignment_governance") + return dict(raw) if isinstance(raw, Mapping) else {} + + +def _requirements_reason(requirements: list[str]) -> str: + labels = { + "authority_function": "a designated authority function", + "evidence": "the required evidence", + "valid_profile": "a supported governance profile", + } + return ( + "Submission requires " + + ", ".join(labels.get(item, item.replace("_", " ")) for item in requirements) + + "." + ) + + +def _has_scope( + request: FunctionAssignmentGovernanceRequest, + scope: str, +) -> bool: + return scopes_grant_compatible(request.actor.scopes, scope) + + +def _text(value: object) -> str | None: + text = str(value).strip() if value is not None else "" + return text or None + + +def _bool(value: object, *, default: bool) -> bool: + return value if isinstance(value, bool) else default + + +def _bounded_int( + value: object, + *, + default: int, + minimum: int, + maximum: int, +) -> int: + try: + number = int(value) if value is not None else default + except (TypeError, ValueError): + return default + return max(minimum, min(number, maximum)) + + +def _optional_positive_int(value: object, *, maximum: int) -> int | None: + if value is None: + return None + try: + number = int(value) + except (TypeError, ValueError): + return None + return max(1, min(number, maximum)) + + +__all__ = [ + "FunctionAssignmentGovernancePolicyProvider", + "SUPPORTED_PROFILES", +] diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index cbe86cb..1d6fe69 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -12,6 +12,7 @@ from govoplan_core.core.module_guards import ( ) from govoplan_core.core.policy import ( CAPABILITY_POLICY_DEFINITION_GOVERNANCE, + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, CAPABILITY_POLICY_VIEW_GOVERNANCE, @@ -69,6 +70,15 @@ def _view_governance_policy(context: ModuleContext) -> object: return ViewGovernancePolicyProvider() +def _function_assignment_governance_policy(context: ModuleContext) -> object: + del context + from govoplan_policy.backend.function_assignment_governance import ( + FunctionAssignmentGovernancePolicyProvider, + ) + + return FunctionAssignmentGovernancePolicyProvider() + + manifest = ModuleManifest( id="policy", name="Policy", @@ -86,6 +96,10 @@ manifest = ModuleManifest( name="policy.view_governance", version="0.1.0", ), + ModuleInterfaceProvider( + name="policy.function_assignment_governance", + version="1.0.0", + ), ), route_factory=_route_factory, migration_spec=MigrationSpec( @@ -145,6 +159,9 @@ manifest = ModuleManifest( capability_factories={ CAPABILITY_POLICY_DEFINITION_GOVERNANCE: _definition_governance_policy, CAPABILITY_POLICY_VIEW_GOVERNANCE: _view_governance_policy, + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE: ( + _function_assignment_governance_policy + ), CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy, }, diff --git a/tests/test_function_assignment_governance.py b/tests/test_function_assignment_governance.py new file mode 100644 index 0000000..5cbe58b --- /dev/null +++ b/tests/test_function_assignment_governance.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.policy import FunctionAssignmentGovernanceRequest +from govoplan_policy.backend.function_assignment_governance import ( + FunctionAssignmentGovernancePolicyProvider, +) + + +class FunctionAssignmentGovernancePolicyTests(unittest.TestCase): + def setUp(self) -> None: + self.provider = FunctionAssignmentGovernancePolicyProvider() + self.actor = PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + identity_id="identity-1", + ) + + def resolve(self, **overrides): + values = { + "tenant_id": "tenant-1", + "kind": "request", + "action": "submit", + "function_id": "function-1", + "actor": self.actor, + "candidate_identity_id": "identity-1", + "function_settings": { + "assignment_governance": { + "request_profile": "holder_with_authority_clearance", + "grant_profile": "holder_with_authority_clearance", + "authority_function_id": "authority-1", + }, + }, + "context": { + "candidate_is_actor": True, + "actor_is_holder": False, + "actor_is_authority": False, + "has_evidence": True, + }, + } + values.update(overrides) + return self.provider.resolve_function_assignment_action( + request=FunctionAssignmentGovernanceRequest(**values) + ) + + def test_self_request_resolves_holder_and_authority_steps(self) -> None: + decision = self.resolve() + + self.assertTrue(decision.allowed) + self.assertEqual(("holder", "authority"), decision.required_steps) + self.assertEqual("authority-1", decision.authority_function_id) + + def test_grant_profiles_recheck_holder_and_authority(self) -> None: + holder_grant = self.resolve( + kind="grant", + context={"actor_is_holder": True, "has_evidence": True}, + ) + unauthorized = self.resolve( + kind="grant", + context={"actor_is_holder": False, "has_evidence": True}, + ) + authority_approval = self.resolve( + kind="grant", + action="approve_authority", + context={"actor_is_authority": True, "has_evidence": True}, + ) + + self.assertTrue(holder_grant.allowed) + self.assertIn("recipient", holder_grant.required_steps) + self.assertFalse(unauthorized.allowed) + self.assertTrue(authority_approval.allowed) + + def test_missing_authority_and_evidence_fail_closed(self) -> None: + decision = self.resolve( + function_settings={ + "assignment_governance": { + "request_profile": "holder_with_authority_clearance", + "evidence_required": True, + }, + }, + context={"candidate_is_actor": True, "has_evidence": False}, + ) + + self.assertFalse(decision.allowed) + self.assertEqual( + ("authority_function", "evidence"), + decision.requirements, + ) + + def test_rejection_is_limited_to_the_current_reviewer(self) -> None: + holder_reject = self.resolve( + action="reject", + current_state="awaiting_holder", + context={"actor_is_holder": True}, + ) + authority_cannot_reject_holder_step = self.resolve( + action="reject", + current_state="awaiting_holder", + context={"actor_is_authority": True}, + ) + recipient_reject = self.resolve( + action="reject", + current_state="awaiting_recipient", + context={"candidate_is_actor": True}, + ) + + self.assertTrue(holder_reject.allowed) + self.assertFalse(authority_cannot_reject_holder_step.allowed) + self.assertTrue(recipient_reject.allowed) + + def test_change_request_and_response_follow_current_participants(self) -> None: + reviewer = self.resolve( + action="request_changes", + current_state="awaiting_holder", + context={"actor_is_holder": True}, + ) + responder = self.resolve( + action="respond", + current_state="changes_requested", + context={"actor_is_initiator": True}, + ) + unrelated = self.resolve( + action="respond", + current_state="changes_requested", + context={}, + ) + + self.assertTrue(reviewer.allowed) + self.assertTrue(responder.allowed) + self.assertFalse(unrelated.allowed) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py index 8da8913..72f0adf 100644 --- a/tests/test_policy_module_contract.py +++ b/tests/test_policy_module_contract.py @@ -6,6 +6,7 @@ import unittest from govoplan_core.core.policy import ( CAPABILITY_POLICY_DEFINITION_GOVERNANCE, + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, CAPABILITY_POLICY_VIEW_GOVERNANCE, @@ -18,11 +19,17 @@ ROOT = pathlib.Path(__file__).resolve().parents[1] class PolicyModuleContractTests(unittest.TestCase): def test_policy_package_does_not_hard_require_access(self) -> None: - project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"] + project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))[ + "project" + ] dependencies = tuple(project["dependencies"]) - self.assertTrue(any(item.startswith("govoplan-core>=") for item in dependencies)) - self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies)) + self.assertTrue( + any(item.startswith("govoplan-core>=") for item in dependencies) + ) + self.assertFalse( + any(item.startswith("govoplan-access") for item in dependencies) + ) def test_policy_source_does_not_import_access_implementation(self) -> None: offenders: list[str] = [] @@ -37,6 +44,7 @@ class PolicyModuleContractTests(unittest.TestCase): self.assertEqual( { CAPABILITY_POLICY_DEFINITION_GOVERNANCE, + CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, CAPABILITY_POLICY_VIEW_GOVERNANCE,