from __future__ import annotations from collections.abc import Mapping from typing import Literal, cast from govoplan_core.core.access import PrincipalRef from govoplan_core.core.policy import ( DefinitionGovernanceAction, DefinitionGovernanceRequest, DefinitionScopeRef, PolicyDecision, PolicySourceStep, definition_governance_policy, ) from govoplan_reporting.backend.domain import ReportingDefinitionRecord from govoplan_reporting.backend.schemas import DefinitionGovernance _LIMITS = ( "inherit_to_lower_scopes", "allow_run", "allow_reuse", "allow_automation", ) _SCOPE_RANK = {"system": 0, "tenant": 1, "group": 2, "user": 3} class ReportingGovernanceError(ValueError): pass def normalize_definition_governance( payload: Mapping[str, object], principal: object, *, administrative: bool, ) -> dict[str, object]: result = dict(payload) raw = result.get("governance") governance = DefinitionGovernance.model_validate( raw if isinstance(raw, Mapping) else {} ) scope_type = governance.scope_type scope_id = str(governance.scope_id or "").strip() or None tenant_id = _tenant(principal) if scope_type == "system": if not _has_scope(principal, "system:governance:write"): raise PermissionError( "System Reporting definitions require system governance permission." ) elif scope_type == "tenant": if scope_id not in {None, tenant_id}: raise PermissionError( "Reporting definitions can only target the active tenant." ) scope_id = tenant_id elif scope_type == "group": if scope_id not in _string_set(getattr(principal, "group_ids", ())): if not administrative: raise PermissionError( "Group Reporting definitions require membership in that group." ) elif scope_type == "user": own_ids = { str(getattr(principal, "account_id", "") or ""), str(getattr(principal, "membership_id", "") or ""), } if scope_id not in own_ids and not administrative: raise PermissionError( "User Reporting definitions can only target the current account." ) if scope_id == str(getattr(principal, "membership_id", "") or ""): scope_id = str(getattr(principal, "account_id", "") or "") effective = _effective_limits(governance) result["governance"] = governance.model_copy( update={ "scope_id": scope_id, "inherit_to_lower_scopes": effective["inherit_to_lower_scopes"], "allow_run": effective["allow_run"], "allow_reuse": effective["allow_reuse"], "allow_automation": effective["allow_automation"], "source_effective_limits": dict(effective), } ).model_dump(mode="json") return result def validate_parent_governance( child_payload: Mapping[str, object], parent_payload: Mapping[str, object], ) -> None: child = _governance(child_payload) parent = _governance(parent_payload) child_scope = _scope(child) parent_scope = _scope(parent) if _SCOPE_RANK[child_scope.scope_type] < _SCOPE_RANK[parent_scope.scope_type]: raise ReportingGovernanceError( "A Reporting definition cannot broaden the scope of its parent." ) if child_scope != parent_scope and not parent.inherit_to_lower_scopes: raise ReportingGovernanceError( "The parent Reporting definition is not inherited by lower scopes." ) parent_limits = _effective_limits(parent) child_limits = _effective_limits(child) broadened = [key for key in _LIMITS if child_limits[key] and not parent_limits[key]] if broadened: raise ReportingGovernanceError( "A child Reporting definition cannot broaden inherited limits: " + ", ".join(sorted(broadened)) ) def apply_parent_governance( child_payload: Mapping[str, object], parent_payload: Mapping[str, object], ) -> dict[str, object]: """Persist the effective parent restriction and its immediate provenance.""" validate_parent_governance(child_payload, parent_payload) child = _governance(child_payload) parent = _governance(parent_payload) parent_limits = _effective_limits(parent) effective = { key: bool(getattr(child, key)) and parent_limits[key] for key in _LIMITS } parent_scope = { "scope_type": parent.scope_type, "scope_id": parent.scope_id, } if parent.source_scope: parent_scope["inherited_from"] = dict(parent.source_scope) result = dict(child_payload) result["governance"] = child.model_copy( update={ "inherit_to_lower_scopes": effective["inherit_to_lower_scopes"], "allow_run": effective["allow_run"], "allow_reuse": effective["allow_reuse"], "allow_automation": effective["allow_automation"], "source_scope": parent_scope, "source_effective_limits": effective, "derivation_provenance": { **dict(child.derivation_provenance), "parent_scope": parent_scope, "restriction_mode": "intersection", }, } ).model_dump(mode="json") return result def definition_decision( session: object, principal: object, *, registry: object | None, record: ReportingDefinitionRecord, action: DefinitionGovernanceAction, ) -> PolicyDecision: governance = _governance(record.payload) source = _scope(governance) target = _target_scope(source, principal) request = DefinitionGovernanceRequest( module_id="reporting", definition_ref=f"{record.definition_kind}:{record.definition_id}:{record.revision}", tenant_id=_tenant(principal), definition_scope=source, target_scope=target, definition_kind=cast(Literal["flow", "template"], "flow"), action=action, actor=_principal_ref(principal), status=record.status, inherit_to_lower_scopes=governance.inherit_to_lower_scopes, allow_run=governance.allow_run, allow_reuse=governance.allow_reuse, allow_automation=governance.allow_automation, context={ "ancestor_limits": dict(governance.source_effective_limits), "ancestor_source": dict(governance.source_scope or {}), "reporting_definition_kind": record.definition_kind, }, ) provider = definition_governance_policy(registry) if provider is not None: return provider.resolve_definition_action(session, request=request) return _fallback_decision(request) def require_definition_action( session: object, principal: object, *, registry: object | None, record: ReportingDefinitionRecord, action: DefinitionGovernanceAction, ) -> PolicyDecision: decision = definition_decision( session, principal, registry=registry, record=record, action=action, ) if not decision.allowed: raise PermissionError( decision.reason or f"Reporting definition action is denied: {action}." ) return decision def governance_payload(payload: Mapping[str, object]) -> dict[str, object]: governance = _governance(payload) return { **governance.model_dump(mode="json"), "effective_limits": _effective_limits(governance), } def scope_visible(payload: Mapping[str, object], principal: object) -> bool: governance = _governance(payload) scope = _scope(governance) if scope.scope_type == "system": return governance.inherit_to_lower_scopes or _has_scope( principal, "reporting:definition:admin" ) if scope.scope_type == "tenant": return scope.scope_id in {None, _tenant(principal)} if scope.scope_type == "group": return scope.scope_id in _string_set(getattr(principal, "group_ids", ())) return scope.scope_id in { str(getattr(principal, "account_id", "") or ""), str(getattr(principal, "membership_id", "") or ""), } def _fallback_decision(request: DefinitionGovernanceRequest) -> PolicyDecision: source = request.definition_scope target = request.target_scope same_scope = source == target inherited = ( _SCOPE_RANK[target.scope_type] >= _SCOPE_RANK[source.scope_type] and request.inherit_to_lower_scopes ) visible = same_scope or inherited if request.action == "view": allowed = visible elif request.action == "edit": allowed = same_scope elif request.action == "run": allowed = visible and request.status == "active" and request.allow_run elif request.action == "reuse": allowed = visible and request.allow_reuse elif request.action == "automate": allowed = visible and request.allow_automation else: allowed = visible and request.allow_reuse reason = ( None if allowed else ( "The Reporting definition's scope or inherited limits do not allow this action." ) ) return PolicyDecision( allowed=allowed, reason=reason, source_path=( PolicySourceStep( scope_type=source.scope_type, scope_id=source.scope_id, label="Reporting definition governance", applied_fields=_LIMITS, policy={ "inherit_to_lower_scopes": request.inherit_to_lower_scopes, "allow_run": request.allow_run, "allow_reuse": request.allow_reuse, "allow_automation": request.allow_automation, }, ), ), requirements=() if allowed else (f"reporting.definition.{request.action}",), details={ "provider": "reporting.conservative_fallback", "definition_scope": source.path, "target_scope": target.path, "action": request.action, }, ) def _governance(payload: Mapping[str, object]) -> DefinitionGovernance: raw = payload.get("governance") return DefinitionGovernance.model_validate(raw if isinstance(raw, Mapping) else {}) def _scope(governance: DefinitionGovernance) -> DefinitionScopeRef: return DefinitionScopeRef( scope_type=governance.scope_type, scope_id=governance.scope_id, ) def _target_scope(source: DefinitionScopeRef, principal: object) -> DefinitionScopeRef: if source.scope_type == "group" and source.scope_id in _string_set( getattr(principal, "group_ids", ()) ): return source own_ids = { str(getattr(principal, "account_id", "") or ""), str(getattr(principal, "membership_id", "") or ""), } if source.scope_type == "user" and source.scope_id in own_ids: return source return DefinitionScopeRef("tenant", _tenant(principal)) def _effective_limits(governance: DefinitionGovernance) -> dict[str, bool]: source = governance.source_effective_limits return { key: bool(getattr(governance, key)) and source.get(key, True) is True for key in _LIMITS } def _principal_ref(principal: object) -> PrincipalRef: converter = getattr(principal, "to_platform_principal", None) if callable(converter): converted = converter() if isinstance(converted, PrincipalRef): return converted return PrincipalRef( account_id=str(getattr(principal, "account_id", "") or "system"), membership_id=_optional(getattr(principal, "membership_id", None)), tenant_id=_tenant(principal), identity_id=_optional(getattr(principal, "identity_id", None)), scopes=frozenset(_string_set(getattr(principal, "scopes", ()))), group_ids=frozenset(_string_set(getattr(principal, "group_ids", ()))), role_ids=frozenset(_string_set(getattr(principal, "role_ids", ()))), function_assignment_ids=frozenset( _string_set(getattr(principal, "function_assignment_ids", ())) ), service_account_id=_optional(getattr(principal, "service_account_id", None)), acting_assignment_id=_optional( getattr(principal, "acting_assignment_id", None) ), ) def _has_scope(principal: object, scope: str) -> bool: method = getattr(principal, "has", None) if callable(method): return bool(method(scope)) return scope in _string_set(getattr(principal, "scopes", ())) def _tenant(principal: object) -> str: tenant_id = str(getattr(principal, "tenant_id", "") or "").strip() if not tenant_id: raise ReportingGovernanceError( "Reporting governance requires a tenant-bound principal." ) return tenant_id def _string_set(value: object) -> set[str]: if isinstance(value, (str, bytes)): return {str(value)} if value else set() try: return {str(item) for item in value or () if str(item).strip()} # type: ignore[union-attr] except TypeError: return set() def _optional(value: object) -> str | None: clean = str(value or "").strip() return clean or None __all__ = [ "ReportingGovernanceError", "apply_parent_governance", "definition_decision", "governance_payload", "normalize_definition_governance", "require_definition_action", "scope_visible", "validate_parent_governance", ] __all__ = [ "ReportingGovernanceError", "definition_decision", "governance_payload", "normalize_definition_governance", "require_definition_action", "scope_visible", "validate_parent_governance", ]