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