diff --git a/README.md b/README.md index 97562a0..e39542c 100644 --- a/README.md +++ b/README.md @@ -29,3 +29,10 @@ it. The resolver reads an optional `maximum_visibility` ceiling from the Missing policy is unrestricted, while malformed explicit policy fails closed to aggregate-only visibility. This resolver slice intentionally has no policy management endpoint or UI yet. + +Policy also provides `policy.definitionGovernance` for Dataflow and Workflow +libraries. It evaluates view, edit, run/start, reuse, derive, and automation +actions across system, tenant, group, and user scopes. Templates cannot run or +be automated. Derived definitions retain ancestor ceilings, and every +decision includes the ordered Policy source path and effective limits so a UI +can explain why an action is available or blocked. diff --git a/src/govoplan_policy/backend/definition_governance.py b/src/govoplan_policy/backend/definition_governance.py new file mode 100644 index 0000000..6913ded --- /dev/null +++ b/src/govoplan_policy/backend/definition_governance.py @@ -0,0 +1,307 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from govoplan_core.core.policy import ( + DefinitionGovernanceRequest, + PolicyDecision, + PolicySourceStep, +) +from govoplan_core.security.module_permissions import scopes_grant_compatible + + +class DefinitionGovernancePolicyProvider: + """Resolve flow-library actions without importing a domain module.""" + + def resolve_definition_action( + self, + *, + request: DefinitionGovernanceRequest, + ) -> PolicyDecision: + source_path = _source_path(request) + effective = _effective_limits(request) + visible, visibility_reason = _visible( + request, + effective=effective, + ) + action = request.action + + if action == "view": + return _decision( + visible, + visibility_reason, + source_path=source_path, + request=request, + effective=effective, + ) + + if action == "edit": + editable, reason = _editable(request) + return _decision( + editable, + reason, + source_path=source_path, + request=request, + effective=effective, + ) + + if not visible: + return _decision( + False, + visibility_reason, + source_path=source_path, + request=request, + effective=effective, + ) + + if action == "run": + if request.definition_kind == "template": + return _decision( + False, + "Templates must be derived into a complete flow before execution.", + source_path=source_path, + request=request, + effective=effective, + ) + if request.status != "active": + return _decision( + False, + "Only active flow revisions can be executed.", + source_path=source_path, + request=request, + effective=effective, + ) + return _decision( + effective["allow_run"], + ( + None + if effective["allow_run"] + else "Execution is disabled by definition or ancestor policy." + ), + source_path=source_path, + request=request, + effective=effective, + ) + + if action in {"reuse", "derive"}: + return _decision( + effective["allow_reuse"], + ( + None + if effective["allow_reuse"] + else "Reuse is disabled by definition or ancestor policy." + ), + source_path=source_path, + request=request, + effective=effective, + ) + + if action == "automate": + if request.definition_kind == "template": + return _decision( + False, + "Templates cannot be automated.", + source_path=source_path, + request=request, + effective=effective, + ) + if request.status != "active": + return _decision( + False, + "Automation requires an active flow revision.", + source_path=source_path, + request=request, + effective=effective, + ) + allowed = effective["allow_run"] and effective["allow_automation"] + return _decision( + allowed, + ( + None + if allowed + else "Automation is disabled by definition or ancestor policy." + ), + source_path=source_path, + request=request, + effective=effective, + ) + + return _decision( + False, + f"Unsupported definition action: {action}", + source_path=source_path, + request=request, + effective=effective, + ) + + +def _visible( + request: DefinitionGovernanceRequest, + *, + effective: Mapping[str, bool], +) -> tuple[bool, str | None]: + scope = request.definition_scope + actor = request.actor + if scope.scope_type == "system": + if _has_scope(actor.scopes, "system:governance:read") or _has_scope( + actor.scopes, + "system:governance:write", + ): + return True, None + if effective["inherit_to_lower_scopes"]: + return True, None + return False, "The system definition is not inherited by lower scopes." + if actor.tenant_id != request.tenant_id: + return False, "The definition belongs to another tenant." + if scope.scope_type == "tenant": + if scope.scope_id != request.tenant_id: + return False, "The definition belongs to another tenant." + return True, None + if scope.scope_type == "group": + if scope.scope_id in actor.group_ids: + return True, None + return False, "The definition is limited to another group." + if scope.scope_type == "user": + if scope.scope_id in {actor.membership_id, actor.account_id}: + return True, None + return False, "The definition is limited to another user." + return False, "The definition scope is invalid." + + +def _editable(request: DefinitionGovernanceRequest) -> tuple[bool, str | None]: + scope = request.definition_scope + actor = request.actor + if scope.scope_type == "system": + allowed = _has_scope(actor.scopes, "system:governance:write") + return ( + allowed, + None if allowed else "System definitions require system governance permission.", + ) + if actor.tenant_id != request.tenant_id: + return False, "Definitions from another tenant are read-only." + if scope.scope_type == "tenant": + allowed = scope.scope_id == request.tenant_id + return allowed, None if allowed else "Inherited tenant definitions are read-only." + if scope.scope_type == "group": + allowed = scope.scope_id in actor.group_ids + return allowed, None if allowed else "Definitions from another group are read-only." + if scope.scope_type == "user": + allowed = scope.scope_id in {actor.membership_id, actor.account_id} + return allowed, None if allowed else "Definitions from another user are read-only." + return False, "The definition scope is invalid." + + +def _effective_limits( + request: DefinitionGovernanceRequest, +) -> dict[str, bool]: + ancestor = request.context.get("ancestor_limits") + ancestor_limits = ancestor if isinstance(ancestor, Mapping) else {} + return { + "inherit_to_lower_scopes": request.inherit_to_lower_scopes + and _ancestor_flag(ancestor_limits, "inherit_to_lower_scopes"), + "allow_run": request.allow_run + and _ancestor_flag(ancestor_limits, "allow_run"), + "allow_reuse": request.allow_reuse + and _ancestor_flag(ancestor_limits, "allow_reuse"), + "allow_automation": request.allow_automation + and _ancestor_flag(ancestor_limits, "allow_automation"), + } + + +def _ancestor_flag(value: Mapping[str, Any], key: str) -> bool: + raw = value.get(key) + return True if raw is None else raw is True + + +def _source_path( + request: DefinitionGovernanceRequest, +) -> tuple[PolicySourceStep, ...]: + steps: list[PolicySourceStep] = [] + ancestor = request.context.get("ancestor_limits") + if isinstance(ancestor, Mapping): + source = request.context.get("ancestor_source") + source_mapping = source if isinstance(source, Mapping) else {} + scope_type = str(source_mapping.get("scope_type") or "system") + scope_id = source_mapping.get("scope_id") + if scope_type in {"system", "tenant", "group", "user"}: + steps.append( + PolicySourceStep( + scope_type=scope_type, # type: ignore[arg-type] + scope_id=str(scope_id) if scope_id is not None else None, + label=str(source_mapping.get("label") or "Ancestor definition"), + applied_fields=tuple(sorted(str(key) for key in ancestor)), + policy=dict(ancestor), + ) + ) + scope = request.definition_scope + steps.append( + PolicySourceStep( + scope_type=scope.scope_type, + scope_id=scope.scope_id, + label=f"{scope.scope_type.capitalize()} definition", + applied_fields=( + "definition_kind", + "inherit_to_lower_scopes", + "allow_run", + "allow_reuse", + "allow_automation", + ), + policy={ + "definition_kind": request.definition_kind, + "status": request.status, + "inherit_to_lower_scopes": request.inherit_to_lower_scopes, + "allow_run": request.allow_run, + "allow_reuse": request.allow_reuse, + "allow_automation": request.allow_automation, + }, + ) + ) + target = request.target_scope + if target.path != scope.path: + steps.append( + PolicySourceStep( + scope_type=target.scope_type, + scope_id=target.scope_id, + label=f"{target.scope_type.capitalize()} use context", + applied_fields=("action",), + policy={"action": request.action}, + ) + ) + return tuple(steps) + + +def _decision( + allowed: bool, + reason: str | None, + *, + source_path: tuple[PolicySourceStep, ...], + request: DefinitionGovernanceRequest, + effective: Mapping[str, bool], +) -> PolicyDecision: + return PolicyDecision( + allowed=allowed, + reason=reason, + source_path=source_path, + requirements=( + () + if allowed + else (f"{request.module_id}.definition.{request.action}",) + ), + details={ + "module_id": request.module_id, + "definition_ref": request.definition_ref, + "action": request.action, + "definition_scope": request.definition_scope.path, + "target_scope": request.target_scope.path, + "definition_kind": request.definition_kind, + "effective_limits": dict(effective), + }, + ) + + +def _has_scope(scopes: object, required: str) -> bool: + return scopes_grant_compatible(scopes, required) # type: ignore[arg-type] + + +__all__ = ["DefinitionGovernancePolicyProvider"] diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index dff84bb..17dc08b 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -2,10 +2,16 @@ from __future__ import annotations from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER from govoplan_core.core.policy import ( + CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, ) -from govoplan_core.core.modules import FrontendModule, ModuleContext, ModuleManifest +from govoplan_core.core.modules import ( + FrontendModule, + ModuleContext, + ModuleInterfaceProvider, + ModuleManifest, +) def _route_factory(context: ModuleContext): @@ -29,17 +35,33 @@ def _scheduling_participant_privacy_policy(context: ModuleContext) -> object: return SqlSchedulingParticipantPrivacyPolicy() +def _definition_governance_policy(context: ModuleContext) -> object: + del context + from govoplan_policy.backend.definition_governance import ( + DefinitionGovernancePolicyProvider, + ) + + return DefinitionGovernancePolicyProvider() + + manifest = ModuleManifest( id="policy", name="Policy", version="0.1.9", required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), + provides_interfaces=( + ModuleInterfaceProvider( + name="policy.definition_governance", + version="0.1.0", + ), + ), route_factory=_route_factory, frontend=FrontendModule( module_id="policy", package_name="@govoplan/policy-webui", ), capability_factories={ + CAPABILITY_POLICY_DEFINITION_GOVERNANCE: _definition_governance_policy, CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy, }, diff --git a/tests/test_definition_governance.py b/tests/test_definition_governance.py new file mode 100644 index 0000000..8d7a581 --- /dev/null +++ b/tests/test_definition_governance.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.policy import ( + DefinitionGovernanceRequest, + DefinitionScopeRef, +) +from govoplan_policy.backend.definition_governance import ( + DefinitionGovernancePolicyProvider, +) + + +class DefinitionGovernancePolicyTests(unittest.TestCase): + def setUp(self) -> None: + self.policy = DefinitionGovernancePolicyProvider() + self.actor = PrincipalRef( + account_id="account-1", + membership_id="user-1", + tenant_id="tenant-1", + group_ids=frozenset({"group-1"}), + ) + + def _resolve(self, **overrides): + values = { + "module_id": "dataflow", + "definition_ref": "pipeline:1", + "tenant_id": "tenant-1", + "definition_scope": DefinitionScopeRef( + "tenant", + "tenant-1", + ), + "target_scope": DefinitionScopeRef("tenant", "tenant-1"), + "definition_kind": "flow", + "action": "view", + "actor": self.actor, + "status": "active", + "inherit_to_lower_scopes": False, + "allow_run": True, + "allow_reuse": False, + "allow_automation": False, + } + values.update(overrides) + return self.policy.resolve_definition_action( + request=DefinitionGovernanceRequest(**values) + ) + + def test_system_definition_is_read_only_when_inherited(self) -> None: + view = self._resolve( + definition_scope=DefinitionScopeRef("system"), + inherit_to_lower_scopes=True, + ) + edit = self._resolve( + definition_scope=DefinitionScopeRef("system"), + inherit_to_lower_scopes=True, + action="edit", + ) + + self.assertTrue(view.allowed) + self.assertFalse(edit.allowed) + self.assertEqual( + ["system", "tenant:tenant-1"], + [step.path for step in view.source_path], + ) + + def test_templates_cannot_run_or_be_automated(self) -> None: + run = self._resolve( + definition_kind="template", + action="run", + allow_run=True, + ) + automated = self._resolve( + definition_kind="template", + action="automate", + allow_run=True, + allow_automation=True, + ) + + self.assertFalse(run.allowed) + self.assertFalse(automated.allowed) + + def test_reuse_and_automation_are_independent_narrowing_grants(self) -> None: + reuse = self._resolve(action="reuse", allow_reuse=True) + automated = self._resolve( + action="automate", + allow_run=True, + allow_automation=False, + ) + + self.assertTrue(reuse.allowed) + self.assertFalse(automated.allowed) + + def test_ancestor_limits_cannot_be_broadened(self) -> None: + decision = self._resolve( + action="reuse", + allow_reuse=True, + context={ + "ancestor_limits": {"allow_reuse": False}, + "ancestor_source": { + "scope_type": "system", + "label": "System template", + }, + }, + ) + + self.assertFalse(decision.allowed) + self.assertFalse( + decision.details["effective_limits"]["allow_reuse"] + ) + self.assertEqual("system", decision.source_path[0].path) + + def test_ancestor_inheritance_limit_controls_visibility(self) -> None: + decision = self._resolve( + definition_scope=DefinitionScopeRef("system"), + inherit_to_lower_scopes=True, + context={ + "ancestor_limits": { + "inherit_to_lower_scopes": False, + }, + "ancestor_source": { + "scope_type": "system", + "label": "Restricted ancestor", + }, + }, + ) + + self.assertFalse(decision.allowed) + self.assertEqual( + "The system definition is not inherited by lower scopes.", + decision.reason, + ) + + def test_group_and_user_scopes_are_visible_only_to_the_actor(self) -> None: + group = self._resolve( + definition_scope=DefinitionScopeRef("group", "group-1") + ) + user = self._resolve( + definition_scope=DefinitionScopeRef("user", "user-2") + ) + + self.assertTrue(group.allowed) + self.assertFalse(user.allowed) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py index f1b4d7e..30a6dac 100644 --- a/tests/test_policy_module_contract.py +++ b/tests/test_policy_module_contract.py @@ -5,6 +5,7 @@ import tomllib import unittest from govoplan_core.core.policy import ( + CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, ) @@ -34,6 +35,7 @@ class PolicyModuleContractTests(unittest.TestCase): def test_policy_manifest_exposes_policy_capabilities(self) -> None: self.assertEqual( { + CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, },