Add function assignment governance profiles

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",
+]

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,
     },

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()

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.assertTrue(
-        self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies))
+            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,