diff --git a/README.md b/README.md index 4263173..97562a0 100644 --- a/README.md +++ b/README.md @@ -20,3 +20,12 @@ Hierarchical policy evaluation, delegation ceilings, and write simulations are implemented in `govoplan_policy.backend.hierarchy`. Privacy retention uses that shared helper and exposes `/api/v1/admin/privacy-retention/policies/{scope}/simulate` 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. diff --git a/docs/POLICY_DECISION_PROVENANCE.md b/docs/POLICY_DECISION_PROVENANCE.md index c64c7d9..9daa603 100644 --- a/docs/POLICY_DECISION_PROVENANCE.md +++ b/docs/POLICY_DECISION_PROVENANCE.md @@ -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 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 Use `govoplan_core.core.policy.PolicyDecision` for explainable policy results: diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index 765f77b..dff84bb 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -1,7 +1,10 @@ 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_PRIVACY_RETENTION +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_PRIVACY_RETENTION, + CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, +) from govoplan_core.core.modules import FrontendModule, ModuleContext, ModuleManifest @@ -19,6 +22,13 @@ def _privacy_retention_service(context: ModuleContext) -> object: return SqlPrivacyRetentionService() +def _scheduling_participant_privacy_policy(context: ModuleContext) -> object: + del context + from govoplan_policy.backend.scheduling_privacy import SqlSchedulingParticipantPrivacyPolicy + + return SqlSchedulingParticipantPrivacyPolicy() + + manifest = ModuleManifest( id="policy", name="Policy", @@ -31,6 +41,7 @@ manifest = ModuleManifest( ), capability_factories={ CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, + CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy, }, ) diff --git a/src/govoplan_policy/backend/scheduling_privacy.py b/src/govoplan_policy/backend/scheduling_privacy.py new file mode 100644 index 0000000..26ad31b --- /dev/null +++ b/src/govoplan_policy/backend/scheduling_privacy.py @@ -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", +] diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py index 1780da4..f1b4d7e 100644 --- a/tests/test_policy_module_contract.py +++ b/tests/test_policy_module_contract.py @@ -4,6 +4,12 @@ import pathlib import tomllib import unittest +from govoplan_core.core.policy import ( + CAPABILITY_POLICY_PRIVACY_RETENTION, + CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, +) +from govoplan_policy.backend.manifest import manifest + ROOT = pathlib.Path(__file__).resolve().parents[1] @@ -25,6 +31,15 @@ class PolicyModuleContractTests(unittest.TestCase): self.assertEqual([], offenders) + def test_policy_manifest_exposes_policy_capabilities(self) -> None: + self.assertEqual( + { + CAPABILITY_POLICY_PRIVACY_RETENTION, + CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, + }, + set(manifest.capability_factories), + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_scheduling_privacy.py b/tests/test_scheduling_privacy.py new file mode 100644 index 0000000..58ba889 --- /dev/null +++ b/tests/test_scheduling_privacy.py @@ -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()