6 Commits

13 changed files with 966 additions and 22 deletions

View File

@@ -20,3 +20,19 @@ Hierarchical policy evaluation, delegation ceilings, and write simulations are
implemented in `govoplan_policy.backend.hierarchy`. Privacy retention uses that implemented in `govoplan_policy.backend.hierarchy`. Privacy retention uses that
shared helper and exposes `/api/v1/admin/privacy-retention/policies/{scope}/simulate` shared helper and exposes `/api/v1/admin/privacy-retention/policies/{scope}/simulate`
for preflight checks before saving lower-level policy changes. for preflight checks before saving lower-level policy changes.
The module also provides the optional
`policy.schedulingParticipantPrivacy` capability. Scheduling owns each
request's participant-visibility setting; Policy can only preserve or narrow
it. The resolver reads an optional `maximum_visibility` ceiling from the
`scheduling_participant_privacy_policy` object in system and tenant settings.
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.

View File

@@ -20,6 +20,34 @@ When retention needs audit-log storage behavior, it requests the
`audit.retention` capability; it does not import audit module tables or `audit.retention` capability; it does not import audit module tables or
providers directly. providers directly.
## Scheduling Participant Privacy
Policy exposes `policy.schedulingParticipantPrivacy` as an optional restriction
hook. Scheduling supplies the request-level visibility choice and remains
responsible for its secure fallback when Policy is absent. The provider never
broadens that choice.
System and tenant settings may contain:
```json
{
"scheduling_participant_privacy_policy": {
"maximum_visibility": "aggregates_only"
}
}
```
`maximum_visibility` is either `aggregates_only` or `names_and_statuses`.
Omitted settings impose no additional restriction. Effective visibility is the
most restrictive of the Scheduling request, the system ceiling, and the tenant
ceiling. Explicit malformed policy fails closed to `aggregates_only` and is
reported in decision details without echoing the invalid stored value. Policy
sources include only explicitly configured or invalid system and tenant steps.
The current slice is resolver-only. A managed write API and administration UI
must add validation, audit, parent-ceiling enforcement, and configuration
safety registration before operators can edit this setting through GovOPlaN.
## Backend DTOs ## Backend DTOs
Use `govoplan_core.core.policy.PolicyDecision` for explainable policy results: Use `govoplan_core.core.policy.PolicyDecision` for explainable policy results:

View File

@@ -1,6 +1,6 @@
{ {
"name": "@govoplan/policy-webui", "name": "@govoplan/policy-webui",
"version": "0.1.8", "version": "0.1.9",
"private": true, "private": true,
"type": "module", "type": "module",
"main": "webui/src/index.ts", "main": "webui/src/index.ts",
@@ -18,7 +18,7 @@
"LICENSE" "LICENSE"
], ],
"peerDependencies": { "peerDependencies": {
"@govoplan/core-webui": "^0.1.8", "@govoplan/core-webui": "^0.1.9",
"lucide-react": "^1.23.0", "lucide-react": "^1.23.0",
"react": "^19.0.0", "react": "^19.0.0",
"react-dom": "^19.0.0", "react-dom": "^19.0.0",

View File

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "govoplan-policy" name = "govoplan-policy"
version = "0.1.8" version = "0.1.9"
description = "GovOPlaN policy platform module." description = "GovOPlaN policy platform module."
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
authors = [{ name = "GovOPlaN" }] authors = [{ name = "GovOPlaN" }]
dependencies = [ dependencies = [
"govoplan-core>=0.1.8", "govoplan-core>=0.1.9",
] ]
[tool.setuptools.packages.find] [tool.setuptools.packages.find]

View File

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

View File

@@ -66,21 +66,27 @@ def validate_hierarchical_policy_patch(
relock_message: Callable[[str], str] | None = None, relock_message: Callable[[str], str] | None = None,
) -> tuple[PolicyValidationIssue, ...]: ) -> tuple[PolicyValidationIssue, ...]:
issues: list[PolicyValidationIssue] = [] issues: list[PolicyValidationIssue] = []
lock_message = locked_field_message or (lambda field: f"{field} is locked by the parent policy.") lock_message = locked_field_message or (lambda field_name: f"{field_name} is locked by the parent policy.")
reenable_message = relock_message or (lambda field: f"{field} limiting cannot be re-enabled below a parent policy lock.") reenable_message = relock_message or (
lambda field_name: f"{field_name} limiting cannot be re-enabled below a parent policy lock."
)
for field in field_keys: for field_name in field_keys:
if field in patch and patch.get(field) is not None and not parent_allow_lower_level_limits.get(field, True): if (
field_name in patch
and patch.get(field_name) is not None
and not parent_allow_lower_level_limits.get(field_name, True)
):
issues.append(PolicyValidationIssue( issues.append(PolicyValidationIssue(
code="field_locked_by_parent", code="field_locked_by_parent",
field=field, field=field_name,
message=lock_message(field), message=lock_message(field_name),
)) ))
patch_allow = patch.get("allow_lower_level_limits") patch_allow = patch.get("allow_lower_level_limits")
if isinstance(patch_allow, Mapping): if isinstance(patch_allow, Mapping):
for field, allowed in patch_allow.items(): for field_name, allowed in patch_allow.items():
clean_field = str(field) clean_field = str(field_name)
if bool(allowed) and not parent_allow_lower_level_limits.get(clean_field, True): if bool(allowed) and not parent_allow_lower_level_limits.get(clean_field, True):
issues.append(PolicyValidationIssue( issues.append(PolicyValidationIssue(
code="lower_level_limit_locked_by_parent", code="lower_level_limit_locked_by_parent",
@@ -125,9 +131,9 @@ def simulate_hierarchical_policy_change(
relock_message=relock_message, relock_message=relock_message,
) )
changed_fields = tuple( changed_fields = tuple(
field field_name
for field in (*field_keys, "allow_lower_level_limits") for field_name in (*field_keys, "allow_lower_level_limits")
if field in patch and current_policy.get(field) != patch.get(field) if field_name in patch and current_policy.get(field_name) != patch.get(field_name)
) )
allowed = not any(issue.severity == "blocker" for issue in issues) allowed = not any(issue.severity == "blocker" for issue in issues)
decision = PolicyDecision( decision = PolicyDecision(

View File

@@ -1,8 +1,18 @@
from __future__ import annotations from __future__ import annotations
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
from govoplan_core.core.policy import CAPABILITY_POLICY_PRIVACY_RETENTION from govoplan_core.core.policy import (
from govoplan_core.core.modules import FrontendModule, ModuleContext, ModuleManifest CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
CAPABILITY_POLICY_PRIVACY_RETENTION,
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
)
from govoplan_core.core.modules import (
FrontendModule,
ModuleContext,
ModuleInterfaceProvider,
ModuleManifest,
)
from govoplan_core.core.views import ViewSurface
def _route_factory(context: ModuleContext): def _route_factory(context: ModuleContext):
@@ -19,18 +29,48 @@ def _privacy_retention_service(context: ModuleContext) -> object:
return SqlPrivacyRetentionService() return SqlPrivacyRetentionService()
def _scheduling_participant_privacy_policy(context: ModuleContext) -> object:
del context
from govoplan_policy.backend.scheduling_privacy import SqlSchedulingParticipantPrivacyPolicy
return SqlSchedulingParticipantPrivacyPolicy()
def _definition_governance_policy(context: ModuleContext) -> object:
del context
from govoplan_policy.backend.definition_governance import (
DefinitionGovernancePolicyProvider,
)
return DefinitionGovernancePolicyProvider()
manifest = ModuleManifest( manifest = ModuleManifest(
id="policy", id="policy",
name="Policy", name="Policy",
version="0.1.8", version="0.1.9",
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), 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, route_factory=_route_factory,
frontend=FrontendModule( frontend=FrontendModule(
module_id="policy", module_id="policy",
package_name="@govoplan/policy-webui", package_name="@govoplan/policy-webui",
view_surfaces=(
ViewSurface(id="policy.admin.system-retention", module_id="policy", kind="section", label="System retention", order=80),
ViewSurface(id="policy.admin.tenant-retention", module_id="policy", kind="section", label="Tenant retention", order=80),
ViewSurface(id="policy.admin.group-retention", module_id="policy", kind="section", label="Group retention", order=80),
ViewSurface(id="policy.admin.user-retention", module_id="policy", kind="section", label="User retention", order=80),
),
), ),
capability_factories={ capability_factories={
CAPABILITY_POLICY_DEFINITION_GOVERNANCE: _definition_governance_policy,
CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service,
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy,
}, },
) )

View File

@@ -0,0 +1,195 @@
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Literal, cast
from sqlalchemy.orm import Session
from govoplan_core.admin.models import SystemSettings
from govoplan_core.admin.settings import SYSTEM_SETTINGS_ID
from govoplan_core.core.policy import (
PolicySourceStep,
SchedulingParticipantPrivacyDecision,
SchedulingParticipantPrivacyRequest,
SchedulingParticipantVisibility,
)
from govoplan_core.tenancy.scope import Tenant
SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY = "scheduling_participant_privacy_policy"
MAXIMUM_VISIBILITY_KEY = "maximum_visibility"
_AGGREGATES_ONLY: SchedulingParticipantVisibility = "aggregates_only"
_NAMES_AND_STATUSES: SchedulingParticipantVisibility = "names_and_statuses"
@dataclass(frozen=True, slots=True)
class _VisibilityCeiling:
value: SchedulingParticipantVisibility
configured: bool = False
issue: str | None = None
def _visibility_ceiling(settings_payload: object) -> _VisibilityCeiling:
"""Read one ceiling, treating malformed explicit policy as restrictive."""
if settings_payload is None:
return _VisibilityCeiling(value=_NAMES_AND_STATUSES)
if not isinstance(settings_payload, Mapping):
return _VisibilityCeiling(
value=_AGGREGATES_ONLY,
configured=True,
issue="invalid_settings_shape",
)
if SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY not in settings_payload:
return _VisibilityCeiling(value=_NAMES_AND_STATUSES)
raw_policy = settings_payload.get(SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY)
if not isinstance(raw_policy, Mapping) or MAXIMUM_VISIBILITY_KEY not in raw_policy:
return _VisibilityCeiling(
value=_AGGREGATES_ONLY,
configured=True,
issue="invalid_policy_shape",
)
raw_visibility = raw_policy.get(MAXIMUM_VISIBILITY_KEY)
if raw_visibility == _AGGREGATES_ONLY:
return _VisibilityCeiling(value=_AGGREGATES_ONLY, configured=True)
if raw_visibility == _NAMES_AND_STATUSES:
return _VisibilityCeiling(value=_NAMES_AND_STATUSES, configured=True)
return _VisibilityCeiling(
value=_AGGREGATES_ONLY,
configured=True,
issue="invalid_maximum_visibility",
)
def _restrict_visibility(
current: SchedulingParticipantVisibility,
ceiling: SchedulingParticipantVisibility,
) -> SchedulingParticipantVisibility:
if current == _AGGREGATES_ONLY or ceiling == _AGGREGATES_ONLY:
return _AGGREGATES_ONLY
return _NAMES_AND_STATUSES
def _source_step(
*,
scope_type: Literal["system", "tenant"],
scope_id: str | None,
label: str,
ceiling: _VisibilityCeiling,
) -> PolicySourceStep | None:
if not ceiling.configured:
return None
policy: dict[str, Any] = {MAXIMUM_VISIBILITY_KEY: ceiling.value}
if ceiling.issue is not None:
policy["configuration_status"] = "invalid_fail_closed"
return PolicySourceStep(
scope_type=scope_type,
scope_id=scope_id,
label=label,
applied_fields=(MAXIMUM_VISIBILITY_KEY,),
policy=policy,
)
def _invalid_requested_visibility(value: object) -> bool:
return value not in {_AGGREGATES_ONLY, _NAMES_AND_STATUSES}
class SqlSchedulingParticipantPrivacyPolicy:
"""Resolve Scheduling roster disclosure against system and tenant ceilings.
Scheduling owns the request-level visibility setting. This provider is an
optional upper bound: absent policy preserves that setting, while an
explicit or malformed policy may only reduce it.
"""
def resolve_scheduling_participant_visibility(
self,
session: object,
*,
request: SchedulingParticipantPrivacyRequest,
) -> SchedulingParticipantPrivacyDecision:
db_session = cast(Session, session)
configuration_errors: list[dict[str, str]] = []
requested_invalid = _invalid_requested_visibility(request.requested_visibility)
requested_visibility: SchedulingParticipantVisibility = (
_AGGREGATES_ONLY if requested_invalid else request.requested_visibility
)
if requested_invalid:
configuration_errors.append({"scope": "request", "code": "invalid_requested_visibility"})
system_settings = db_session.get(SystemSettings, SYSTEM_SETTINGS_ID)
system_ceiling = _visibility_ceiling(system_settings.settings if system_settings is not None else None)
if system_ceiling.issue is not None:
configuration_errors.append({"scope": "system", "code": system_ceiling.issue})
tenant = db_session.get(Tenant, request.tenant_id)
if tenant is None:
tenant_ceiling = _VisibilityCeiling(
value=_AGGREGATES_ONLY,
configured=True,
issue="tenant_not_found",
)
else:
tenant_ceiling = _visibility_ceiling(tenant.settings)
if tenant_ceiling.issue is not None:
configuration_errors.append({"scope": "tenant", "code": tenant_ceiling.issue})
policy_ceiling = _restrict_visibility(system_ceiling.value, tenant_ceiling.value)
effective_visibility = _restrict_visibility(requested_visibility, policy_ceiling)
source_path = tuple(
step
for step in (
_source_step(
scope_type="system",
scope_id=None,
label="System",
ceiling=system_ceiling,
),
_source_step(
scope_type="tenant",
scope_id=request.tenant_id,
label="Tenant",
ceiling=tenant_ceiling,
),
)
if step is not None
)
reason: str | None = None
if configuration_errors:
reason = (
"Participant roster visibility was restricted because policy configuration "
"could not be validated."
)
elif effective_visibility != requested_visibility:
reason = "Participant roster visibility is restricted by policy."
return SchedulingParticipantPrivacyDecision(
effective_visibility=effective_visibility,
reason=reason,
source_path=source_path,
details={
"requested_visibility": request.requested_visibility,
"policy_ceiling": policy_ceiling,
"configured_scopes": [
scope
for scope, ceiling in (("system", system_ceiling), ("tenant", tenant_ceiling))
if ceiling.configured
],
"configuration_errors": configuration_errors,
},
)
__all__ = [
"MAXIMUM_VISIBILITY_KEY",
"SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY",
"SqlSchedulingParticipantPrivacyPolicy",
]

View File

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

View File

@@ -4,6 +4,13 @@ import pathlib
import tomllib import tomllib
import unittest import unittest
from govoplan_core.core.policy import (
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
CAPABILITY_POLICY_PRIVACY_RETENTION,
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
)
from govoplan_policy.backend.manifest import manifest
ROOT = pathlib.Path(__file__).resolve().parents[1] ROOT = pathlib.Path(__file__).resolve().parents[1]
@@ -25,6 +32,16 @@ class PolicyModuleContractTests(unittest.TestCase):
self.assertEqual([], offenders) self.assertEqual([], offenders)
def test_policy_manifest_exposes_policy_capabilities(self) -> None:
self.assertEqual(
{
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
CAPABILITY_POLICY_PRIVACY_RETENTION,
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
},
set(manifest.capability_factories),
)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

View File

@@ -0,0 +1,178 @@
from __future__ import annotations
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.admin.models import SystemSettings
from govoplan_core.core.modules import ModuleContext
from govoplan_core.core.policy import (
CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
SchedulingParticipantPrivacyPolicy,
SchedulingParticipantPrivacyRequest,
)
from govoplan_core.db.base import Base
from govoplan_core.tenancy.scope import Tenant, scope_registry
from govoplan_policy.backend.manifest import manifest
from govoplan_policy.backend.scheduling_privacy import (
SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY,
SqlSchedulingParticipantPrivacyPolicy,
)
def _policy_settings(maximum_visibility: object) -> dict[str, object]:
return {
SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: {
"maximum_visibility": maximum_visibility,
}
}
class SchedulingParticipantPrivacyTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(bind=self.engine, tables=[SystemSettings.__table__])
scope_registry.metadata.create_all(bind=self.engine, tables=[Tenant.__table__])
self.session = Session(self.engine)
self.provider = SqlSchedulingParticipantPrivacyPolicy()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def _add_settings(
self,
*,
system: dict[str, object] | None = None,
tenant: dict[str, object] | None = None,
) -> None:
self.session.add(SystemSettings(id="global", settings=system or {}))
self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings=tenant or {}))
self.session.flush()
def _resolve(self, requested_visibility: str = "names_and_statuses"):
return self.provider.resolve_scheduling_participant_visibility(
self.session,
request=SchedulingParticipantPrivacyRequest(
tenant_id="tenant-1",
scheduling_request_id="request-1",
participant_id="participant-1",
requested_visibility=requested_visibility, # type: ignore[arg-type]
actor_user_id="user-1",
),
)
def test_manifest_registers_the_core_capability(self) -> None:
factory = manifest.capability_factories[CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY]
provider = factory(ModuleContext(registry=object(), settings=object()))
self.assertIsInstance(provider, SchedulingParticipantPrivacyPolicy)
def test_missing_policy_preserves_scheduling_visibility(self) -> None:
self._add_settings()
decision = self._resolve()
self.assertEqual("names_and_statuses", decision.effective_visibility)
self.assertIsNone(decision.reason)
self.assertEqual((), decision.source_path)
self.assertEqual([], decision.details["configured_scopes"])
def test_missing_system_settings_is_an_unrestricted_read_only_default(self) -> None:
self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings={}))
self.session.flush()
decision = self._resolve()
self.assertEqual("names_and_statuses", decision.effective_visibility)
self.assertIsNone(self.session.get(SystemSettings, "global"))
def test_system_ceiling_restricts_participant_roster(self) -> None:
self._add_settings(system=_policy_settings("aggregates_only"))
decision = self._resolve()
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertEqual("Participant roster visibility is restricted by policy.", decision.reason)
self.assertEqual(["system"], [step.path for step in decision.source_path])
self.assertEqual("aggregates_only", decision.source_path[0].policy["maximum_visibility"])
def test_tenant_ceiling_can_narrow_system_but_cannot_widen_it(self) -> None:
self._add_settings(
system=_policy_settings("aggregates_only"),
tenant=_policy_settings("names_and_statuses"),
)
decision = self._resolve()
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertEqual(["system", "tenant:tenant-1"], [step.path for step in decision.source_path])
self.assertEqual("aggregates_only", decision.details["policy_ceiling"])
def test_policy_never_broadens_an_aggregate_only_request(self) -> None:
self._add_settings(
system=_policy_settings("names_and_statuses"),
tenant=_policy_settings("names_and_statuses"),
)
decision = self._resolve("aggregates_only")
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertIsNone(decision.reason)
def test_invalid_explicit_policy_fails_closed_without_echoing_value(self) -> None:
self._add_settings(tenant=_policy_settings("surprise"))
decision = self._resolve()
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertIn("could not be validated", decision.reason or "")
self.assertEqual(
[{"scope": "tenant", "code": "invalid_maximum_visibility"}],
decision.details["configuration_errors"],
)
self.assertNotIn("surprise", repr(decision.to_dict()))
self.assertEqual("invalid_fail_closed", decision.source_path[0].policy["configuration_status"])
def test_invalid_policy_shape_fails_closed(self) -> None:
self._add_settings(
system={SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: "not-an-object"},
)
decision = self._resolve()
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertEqual(
[{"scope": "system", "code": "invalid_policy_shape"}],
decision.details["configuration_errors"],
)
def test_missing_tenant_fails_closed(self) -> None:
self.session.add(SystemSettings(id="global", settings={}))
self.session.flush()
decision = self._resolve()
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertEqual(
[{"scope": "tenant", "code": "tenant_not_found"}],
decision.details["configuration_errors"],
)
self.assertEqual(["tenant:tenant-1"], [step.path for step in decision.source_path])
def test_invalid_requested_visibility_fails_closed(self) -> None:
self._add_settings()
decision = self._resolve("invalid")
self.assertEqual("aggregates_only", decision.effective_visibility)
self.assertEqual(
[{"scope": "request", "code": "invalid_requested_visibility"}],
decision.details["configuration_errors"],
)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,6 +1,6 @@
{ {
"name": "@govoplan/policy-webui", "name": "@govoplan/policy-webui",
"version": "0.1.8", "version": "0.1.9",
"private": true, "private": true,
"type": "module", "type": "module",
"main": "src/index.ts", "main": "src/index.ts",
@@ -13,11 +13,11 @@
} }
}, },
"peerDependencies": { "peerDependencies": {
"@govoplan/core-webui": "^0.1.8", "@govoplan/core-webui": "^0.1.9",
"lucide-react": "^1.23.0", "lucide-react": "^1.23.0",
"react": "^19.0.0", "react": "^19.0.0",
"react-dom": "^19.0.0", "react-dom": "^19.0.0",
"react-router-dom": "^7.1.1" "react-router-dom": ">=7.18.2 <8"
}, },
"peerDependenciesMeta": { "peerDependenciesMeta": {
"@govoplan/core-webui": { "@govoplan/core-webui": {

View File

@@ -7,6 +7,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
sections: [ sections: [
{ {
id: "system-retention", id: "system-retention",
surfaceId: "policy.admin.system-retention",
label: "Retention", label: "Retention",
group: "SYSTEM", group: "SYSTEM",
order: 80, order: 80,
@@ -19,6 +20,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
}, },
{ {
id: "tenant-retention", id: "tenant-retention",
surfaceId: "policy.admin.tenant-retention",
label: "Retention", label: "Retention",
group: "TENANT", group: "TENANT",
order: 80, order: 80,
@@ -31,6 +33,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
}, },
{ {
id: "tenant-group-retention", id: "tenant-group-retention",
surfaceId: "policy.admin.group-retention",
label: "Retention", label: "Retention",
group: "GROUP", group: "GROUP",
order: 30, order: 30,
@@ -43,6 +46,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
}, },
{ {
id: "tenant-user-retention", id: "tenant-user-retention",
surfaceId: "policy.admin.user-retention",
label: "Retention", label: "Retention",
group: "USER", group: "USER",
order: 30, order: 30,
@@ -59,8 +63,14 @@ const policyAdminSections: AdminSectionsUiCapability = {
export const policyModule: PlatformWebModule = { export const policyModule: PlatformWebModule = {
id: "policy", id: "policy",
label: "Policy", label: "Policy",
version: "0.1.6", version: "0.1.9",
dependencies: ["access", "admin"], dependencies: ["access", "admin"],
viewSurfaces: [
{ id: "policy.admin.system-retention", moduleId: "policy", kind: "section", label: "System retention", order: 80 },
{ id: "policy.admin.tenant-retention", moduleId: "policy", kind: "section", label: "Tenant retention", order: 80 },
{ id: "policy.admin.group-retention", moduleId: "policy", kind: "section", label: "Group retention", order: 80 },
{ id: "policy.admin.user-retention", moduleId: "policy", kind: "section", label: "User retention", order: 80 }
],
uiCapabilities: { uiCapabilities: {
"admin.sections": policyAdminSections "admin.sections": policyAdminSections
} }