govoplan-campaign/tests/test_archive_encryption_governance.py

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