diff --git a/src/govoplan_policy/backend/distribution_channels.py b/src/govoplan_policy/backend/distribution_channels.py new file mode 100644 index 0000000..669fc53 --- /dev/null +++ b/src/govoplan_policy/backend/distribution_channels.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +import hashlib +from sqlalchemy.orm import Session + +from govoplan_core.core.distribution_lists import ( + DistributionChannelPolicyDecision, + DistributionChannelPolicyRequest, +) +from govoplan_policy.backend.policy_overrides import resolution_policy_overrides + + +CHANNELS = frozenset({"email", "postal", "internal_mail", "portal"}) + + +@dataclass(frozen=True, slots=True) +class DistributionChannelPolicyResolution: + allowed_channels: frozenset[str] + source_path: tuple[Mapping[str, object], ...] = () + diagnostics: tuple[Mapping[str, object], ...] = () + + +class DistributionChannelPolicyProvider: + def resolve_distribution_channel( + self, + session: object, + principal: object, + *, + request: DistributionChannelPolicyRequest, + ) -> DistributionChannelPolicyDecision: + resolution = _resolve(session, principal, request) + channel = request.candidate.channel + allowed = channel in resolution.allowed_channels + malformed = any( + item.get("code") == "distribution_channel_policy.invalid" + for item in resolution.diagnostics + ) + if malformed and not allowed: + reason_code = "policy.invalid_fail_closed" + explanation = ( + "Distribution is blocked because an applicable channel policy is invalid." + ) + elif allowed: + reason_code = "policy.channel_allowed" + explanation = f"The {channel} channel is permitted by the effective Policy." + else: + reason_code = "policy.channel_blocked" + explanation = f"The {channel} channel is blocked by the effective Policy." + return DistributionChannelPolicyDecision( + allowed=allowed, + reason_code=reason_code, + explanation=explanation, + source_path=resolution.source_path, + requirements=(() if allowed else (f"policy.channel.{channel}",)), + details={ + "allowed_channels": sorted(resolution.allowed_channels), + "diagnostics": [dict(item) for item in resolution.diagnostics], + }, + ) + + +def resolve_distribution_channel_policy_rows( + rows: object, +) -> DistributionChannelPolicyResolution: + allowed_channels = set(CHANNELS) + source_path: list[Mapping[str, object]] = [] + diagnostics: list[Mapping[str, object]] = [] + for row in rows if isinstance(rows, (list, tuple)) else (): + policy, malformed = validate_distribution_channel_policy(row.policy) + if malformed: + allowed_channels.clear() + applied_fields: tuple[str, ...] = ("configuration_status",) + source_policy: Mapping[str, object] = { + "configuration_status": "invalid_fail_closed", + "target_key": row.target_key, + } + diagnostics.append( + { + "code": "distribution_channel_policy.invalid", + "severity": "error", + "scope": row.scope_key, + "target_key": row.target_key, + } + ) + else: + configured = policy.get("allowed_channels") + blocked = policy.get("blocked_channels", ()) + if isinstance(configured, tuple): + allowed_channels.intersection_update(configured) + if isinstance(blocked, tuple): + allowed_channels.difference_update(blocked) + applied_fields = tuple(sorted(policy)) + source_policy = { + key: list(value) if isinstance(value, tuple) else value + for key, value in policy.items() + } + source_policy = {**source_policy, "target_key": row.target_key} + source_path.append( + { + "scope_type": row.scope_type, + "scope_id": row.scope_id, + "label": f"{row.scope_type.capitalize()} distribution-channel policy", + "applied_fields": list(applied_fields), + "policy": dict(source_policy), + } + ) + return DistributionChannelPolicyResolution( + allowed_channels=frozenset(allowed_channels), + source_path=tuple(source_path), + diagnostics=tuple(diagnostics), + ) + + +def validate_distribution_channel_policy( + value: object, +) -> tuple[dict[str, tuple[str, ...]], bool]: + if not isinstance(value, Mapping): + return {}, True + supported = {"allowed_channels", "blocked_channels"} + if any(str(key) not in supported for key in value): + return {}, True + policy: dict[str, tuple[str, ...]] = {} + for key, raw in value.items(): + if not isinstance(raw, Sequence) or isinstance(raw, (str, bytes)): + return {}, True + channels = tuple(dict.fromkeys(str(item).strip() for item in raw)) + if any(channel not in CHANNELS for channel in channels): + return {}, True + policy[str(key)] = channels + return policy, False + + +def _resolve( + session: object, + principal: object, + request: DistributionChannelPolicyRequest, +) -> DistributionChannelPolicyResolution: + if not isinstance(session, Session): + return DistributionChannelPolicyResolution(allowed_channels=CHANNELS) + group_ids = tuple(getattr(principal, "group_ids", ()) or ()) + account_id = str(getattr(principal, "account_id", "") or "") + membership_id = str(getattr(principal, "membership_id", "") or "") + target_keys = ["*", _target_key("list", request.list_id)] + if request.purpose: + target_keys.extend( + ( + _target_key("purpose", request.purpose), + _target_key( + f"list:{request.list_id}:purpose", + request.purpose, + ), + ) + ) + rows = resolution_policy_overrides( + session, + policy_family="distribution_channels", + target_keys=target_keys, + tenant_id=request.tenant_id, + group_ids=group_ids, + user_ids=(account_id, membership_id), + ) + return resolve_distribution_channel_policy_rows(rows) + + +def _target_key(prefix: str, value: str) -> str: + candidate = f"{prefix}:{value}" + if len(candidate) <= 120: + return candidate + digest = hashlib.sha256(value.encode("utf-8")).hexdigest() + return f"{prefix[:78]}:sha256:{digest[:32]}" + + +__all__ = [ + "CHANNELS", + "DistributionChannelPolicyProvider", + "DistributionChannelPolicyResolution", + "resolve_distribution_channel_policy_rows", + "validate_distribution_channel_policy", +] diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index 1d6fe69..16247e3 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -6,6 +6,9 @@ from govoplan_core.core.access import ( CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) +from govoplan_core.core.distribution_lists import ( + CAPABILITY_POLICY_DISTRIBUTION_CHANNELS, +) from govoplan_core.core.module_guards import ( drop_table_retirement_provider, persistent_table_uninstall_guard, @@ -79,6 +82,15 @@ def _function_assignment_governance_policy(context: ModuleContext) -> object: return FunctionAssignmentGovernancePolicyProvider() +def _distribution_channel_policy(context: ModuleContext) -> object: + del context + from govoplan_policy.backend.distribution_channels import ( + DistributionChannelPolicyProvider, + ) + + return DistributionChannelPolicyProvider() + + manifest = ModuleManifest( id="policy", name="Policy", @@ -100,6 +112,10 @@ manifest = ModuleManifest( name="policy.function_assignment_governance", version="1.0.0", ), + ModuleInterfaceProvider( + name=CAPABILITY_POLICY_DISTRIBUTION_CHANNELS, + version="1.0.0", + ), ), route_factory=_route_factory, migration_spec=MigrationSpec( @@ -162,6 +178,7 @@ manifest = ModuleManifest( CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE: ( _function_assignment_governance_policy ), + CAPABILITY_POLICY_DISTRIBUTION_CHANNELS: _distribution_channel_policy, CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy, }, diff --git a/src/govoplan_policy/backend/policy_overrides.py b/src/govoplan_policy/backend/policy_overrides.py index 7f4821d..4498e6e 100644 --- a/src/govoplan_policy/backend/policy_overrides.py +++ b/src/govoplan_policy/backend/policy_overrides.py @@ -10,7 +10,7 @@ from govoplan_policy.backend.db.models import PolicyOverride POLICY_SCOPE_TYPES = frozenset({"system", "tenant", "group", "user"}) -POLICY_FAMILIES = frozenset({"definition", "view"}) +POLICY_FAMILIES = frozenset({"definition", "distribution_channels", "view"}) class PolicyOverrideError(ValueError): @@ -21,7 +21,9 @@ def normalize_policy_target(policy_family: str, target_key: str) -> tuple[str, s family = policy_family.strip().casefold() target = target_key.strip().casefold() if family not in POLICY_FAMILIES: - raise PolicyOverrideError("Policy family must be definition or view") + raise PolicyOverrideError( + "Policy family must be definition, distribution_channels, or view" + ) if not target or len(target) > 120: raise PolicyOverrideError("Policy target must contain 1 to 120 characters") if family == "view" and target != "*": diff --git a/tests/test_distribution_channels.py b/tests/test_distribution_channels.py new file mode 100644 index 0000000..046a466 --- /dev/null +++ b/tests/test_distribution_channels.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +from datetime import UTC, datetime +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.core.distribution_lists import ( + DistributionChannelCandidate, + DistributionChannelPolicyRequest, + DistributionRecipientRef, +) +from govoplan_policy.backend.db.models import PolicyOverride +from govoplan_policy.backend.distribution_channels import ( + DistributionChannelPolicyProvider, + resolve_distribution_channel_policy_rows, + validate_distribution_channel_policy, +) + + +class _Principal: + account_id = "account-1" + membership_id = "membership-1" + group_ids = frozenset({"group-1"}) + + +def _request(channel: str = "email") -> DistributionChannelPolicyRequest: + candidate = DistributionChannelCandidate( + channel=channel, # type: ignore[arg-type] + target="recipient@example.test", + target_key=f"{channel}:recipient@example.test", + ) + return DistributionChannelPolicyRequest( + tenant_id="tenant-1", + list_id="list-1", + purpose="monthly-notice", + effective_at=datetime(2026, 1, 1, tzinfo=UTC), + recipient=DistributionRecipientRef( + recipient_key="recipient-1", + display_name="Recipient", + status="usable", + channels=(candidate,), + ), + candidate=candidate, + ) + + +class DistributionChannelPolicyTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + PolicyOverride.__table__.create(self.engine) + + def tearDown(self) -> None: + self.engine.dispose() + + def test_hierarchy_can_only_reduce_permitted_channels(self) -> None: + with Session(self.engine) as session: + session.add_all( + ( + PolicyOverride( + policy_family="distribution_channels", + target_key="*", + tenant_id=None, + scope_type="system", + scope_id=None, + scope_key="system", + policy={"allowed_channels": ["email", "postal"]}, + ), + PolicyOverride( + policy_family="distribution_channels", + target_key="purpose:monthly-notice", + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + scope_key="tenant:tenant-1", + policy={"blocked_channels": ["postal"]}, + ), + ) + ) + session.commit() + provider = DistributionChannelPolicyProvider() + + email = provider.resolve_distribution_channel( + session, + _Principal(), + request=_request("email"), + ) + postal = provider.resolve_distribution_channel( + session, + _Principal(), + request=_request("postal"), + ) + + self.assertTrue(email.allowed) + self.assertFalse(postal.allowed) + self.assertEqual("policy.channel_blocked", postal.reason_code) + self.assertEqual(2, len(postal.source_path)) + + def test_malformed_policy_fails_closed_with_diagnostic(self) -> None: + row = type( + "Row", + (), + { + "policy": {"allowed_channels": "email"}, + "target_key": "*", + "scope_type": "tenant", + "scope_id": "tenant-1", + "scope_key": "tenant:tenant-1", + }, + )() + result = resolve_distribution_channel_policy_rows((row,)) + + self.assertEqual(frozenset(), result.allowed_channels) + self.assertEqual("distribution_channel_policy.invalid", result.diagnostics[0]["code"]) + + def test_schema_rejects_unknown_channels_and_fields(self) -> None: + self.assertEqual( + ({"allowed_channels": ("email",)}, False), + validate_distribution_channel_policy({"allowed_channels": ["email"]}), + ) + self.assertTrue(validate_distribution_channel_policy({"allowed_channels": ["fax"]})[1]) + self.assertTrue(validate_distribution_channel_policy({"unknown": []})[1]) + + def test_long_purpose_is_resolved_without_exceeding_storage_key_limit(self) -> None: + request = _request("email") + request = DistributionChannelPolicyRequest( + tenant_id=request.tenant_id, + list_id=request.list_id, + purpose="purpose-" + "x" * 120, + effective_at=request.effective_at, + recipient=request.recipient, + candidate=request.candidate, + ) + with Session(self.engine) as session: + decision = DistributionChannelPolicyProvider().resolve_distribution_channel( + session, + _Principal(), + request=request, + ) + + self.assertTrue(decision.allowed) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_policy_module_contract.py b/tests/test_policy_module_contract.py index 72f0adf..7e015df 100644 --- a/tests/test_policy_module_contract.py +++ b/tests/test_policy_module_contract.py @@ -11,6 +11,9 @@ from govoplan_core.core.policy import ( CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY, CAPABILITY_POLICY_VIEW_GOVERNANCE, ) +from govoplan_core.core.distribution_lists import ( + CAPABILITY_POLICY_DISTRIBUTION_CHANNELS, +) from govoplan_policy.backend.manifest import manifest @@ -44,6 +47,7 @@ class PolicyModuleContractTests(unittest.TestCase): self.assertEqual( { CAPABILITY_POLICY_DEFINITION_GOVERNANCE, + CAPABILITY_POLICY_DISTRIBUTION_CHANNELS, CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE, CAPABILITY_POLICY_PRIVACY_RETENTION, CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,