feat: govern legacy archive encryption

2026-08-20 12:24:36 +02:00
parent d5b874c469
commit 75e8f864a7
19 changed files with 1102 additions and 24 deletions
@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_campaign.backend.archive_encryption import (
+    CampaignArchiveEncryptionError,
+    LEGACY_ZIPCRYPTO_SCOPE,
+    assert_archive_encryption_allowed,
+    effective_archive_encryption_policy,
+    stamp_legacy_zipcrypto_acknowledgements,
+)
+from govoplan_campaign.backend.db.models import Campaign
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.policy import (
+    CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION,
+    CampaignArchiveEncryptionDecision,
+    PolicySourceStep,
+)
+
+
+class _PolicyProvider:
+    def __init__(self, methods: set[str]) -> None:
+        self.methods = methods
+
+    def resolve_campaign_archive_encryption(self, session=None, *, request):
+        del session, request
+        return CampaignArchiveEncryptionDecision(
+            allowed_password_encryption_methods=frozenset(self.methods),
+            allowed_password_delivery_channels=frozenset(
+                {"separate_mail", "sms", "letter", "phone", "in_person"}
+            ),
+            policy_hash="f" * 64,
+            source_path=(
+                PolicySourceStep(
+                    scope_type="system",
+                    label="System archive-encryption policy",
+                    applied_fields=("allowed_password_encryption_methods",),
+                    policy={"allowed_password_encryption_methods": sorted(self.methods)},
+                ),
+            ),
+        )
+
+
+class _Registry:
+    def __init__(self, provider) -> None:
+        self.provider = provider
+
+    def has_capability(self, name: str) -> bool:
+        return name == CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION
+
+    def capability(self, name: str):
+        return self.provider if self.has_capability(name) else None
+
+
+class CampaignArchiveEncryptionGovernanceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        self.session = Session(self.engine)
+        self.campaign = Campaign(
+            id="campaign-1",
+            tenant_id="tenant-1",
+            external_id="example",
+            name="Example",
+            owner_user_id="user-1",
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_unavailable_policy_keeps_aes_and_fails_closed_for_legacy(self) -> None:
+        with patch("govoplan_campaign.backend.archive_encryption.get_registry", return_value=None):
+            policy = effective_archive_encryption_policy(self.session, self.campaign)
+            self.assertFalse(policy.available)
+            self.assertEqual(frozenset({"aes"}), policy.allowed_password_encryption_methods)
+            assert_archive_encryption_allowed(
+                self.session,
+                self.campaign,
+                _raw_archive("aes"),
+            )
+            with self.assertRaisesRegex(CampaignArchiveEncryptionError, "blocked"):
+                assert_archive_encryption_allowed(
+                    self.session,
+                    self.campaign,
+                    _raw_archive("zip_standard", stamped=True),
+                )
+
+    def test_existing_password_archive_inherits_separate_mail_channel(self) -> None:
+        raw = _raw_archive("aes")
+        raw["attachments"]["zip"]["archives"][0].pop("password_delivery_channel")
+        with patch(
+            "govoplan_campaign.backend.archive_encryption.get_registry",
+            return_value=None,
+        ):
+            decision = assert_archive_encryption_allowed(
+                self.session,
+                self.campaign,
+                raw,
+            )
+        self.assertIn(
+            "separate_mail",
+            decision.allowed_password_delivery_channels,
+        )
+
+    def test_legacy_selection_requires_permission_and_gets_server_stamp(self) -> None:
+        registry = _Registry(_PolicyProvider({"aes", "zip_standard"}))
+        candidate = _raw_archive("zip_standard")
+        candidate["attachments"]["zip"]["archives"][0].update(
+            {
+                "legacy_zipcrypto_acknowledged": True,
+                "legacy_zipcrypto_reason": "Recipient requires built-in Windows extraction",
+            }
+        )
+        with patch("govoplan_campaign.backend.archive_encryption.get_registry", return_value=registry):
+            with self.assertRaisesRegex(CampaignArchiveEncryptionError, "Missing scope"):
+                stamp_legacy_zipcrypto_acknowledgements(
+                    self.session,
+                    self.campaign,
+                    {},
+                    candidate,
+                    principal=_principal(set()),
+                )
+            stamped, evidence = stamp_legacy_zipcrypto_acknowledgements(
+                self.session,
+                self.campaign,
+                {},
+                candidate,
+                principal=_principal({LEGACY_ZIPCRYPTO_SCOPE}),
+            )
+            archive = stamped["attachments"]["zip"]["archives"][0]
+            self.assertEqual("user-1", archive["legacy_zipcrypto_acknowledged_by"])
+            self.assertTrue(archive["legacy_zipcrypto_acknowledged_at"])
+            self.assertEqual("f" * 64, evidence[0]["policy_hash"])
+            assert_archive_encryption_allowed(
+                self.session,
+                self.campaign,
+                stamped,
+                principal=_principal({LEGACY_ZIPCRYPTO_SCOPE}),
+            )
+
+
+def _raw_archive(method: str, *, stamped: bool = False) -> dict:
+    archive = {
+        "id": "archive-1",
+        "method": method,
+        "password_enabled": True,
+        "password_delivery_channel": "separate_mail",
+        "legacy_zipcrypto_acknowledged": method == "zip_standard",
+        "legacy_zipcrypto_reason": "Windows recipient compatibility required"
+        if method == "zip_standard"
+        else None,
+    }
+    if stamped:
+        archive.update(
+            {
+                "legacy_zipcrypto_acknowledged_by": "user-1",
+                "legacy_zipcrypto_acknowledged_at": "2026-08-20T10:00:00+00:00",
+            }
+        )
+    return {"attachments": {"zip": {"enabled": True, "archives": [archive]}}}
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -10,7 +10,10 @@ try:
 except ImportError:  # pragma: no cover
    pyzipper = None

-from govoplan_campaign.backend.services.zip_service import create_zip_archive
+from govoplan_campaign.backend.services.zip_service import (
+    create_zip_archive,
+    zip_archive_evidence,
+)


 class ZipServiceTests(unittest.TestCase):
@@ -28,6 +31,20 @@ class ZipServiceTests(unittest.TestCase):
                self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED)
                self.assertTrue(info.flag_bits & 0x1)
                self.assertEqual(archive.read("message.txt", pwd=b"secret"), b"Hello Windows ZIP")
+                with self.assertRaises(RuntimeError):
+                    archive.read("message.txt", pwd=b"wrong-password")
+
+            evidence = zip_archive_evidence(
+                output,
+                [(source, "message.txt")],
+                password_protected=True,
+                method="zip_standard",
+            )
+            self.assertEqual("Legacy ZipCrypto", evidence["format"])
+            self.assertEqual("govoplan-campaign.zipcrypto", evidence["implementation"])
+            self.assertNotIn("password", evidence)
+            self.assertEqual(64, len(str(evidence["archive_sha256"])))
+            self.assertEqual(64, len(str(evidence["members"][0]["sha256"])))

    @unittest.skipIf(pyzipper is None, "pyzipper is not installed")
    def test_aes_password_zip_keeps_aes_encryption(self) -> None:
@@ -63,6 +80,18 @@ class ZipServiceTests(unittest.TestCase):
                self.assertFalse(info.flag_bits & 0x1)
                self.assertEqual(archive.read("message.txt"), b"Plain ZIP")

+    def test_unknown_password_method_fails_without_downgrade_or_output(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            source = root / "source.txt"
+            source.write_text("No downgrade", encoding="utf-8")
+            output = root / "unknown.zip"
+
+            with self.assertRaisesRegex(ValueError, "Unsupported"):
+                create_zip_archive(output, [source], "secret", "unknown")
+
+            self.assertFalse(output.exists())
+

 if __name__ == "__main__":
    unittest.main()