Govern distribution delivery channels

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

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,
     },

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 != "*":

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

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,