diff --git a/src/govoplan_campaign/backend/attachments/resolver.py b/src/govoplan_campaign/backend/attachments/resolver.py index f675c04..a909a48 100644 --- a/src/govoplan_campaign/backend/attachments/resolver.py +++ b/src/govoplan_campaign/backend/attachments/resolver.py @@ -390,6 +390,27 @@ def _issue_for_ambiguous(config: AttachmentConfig, behavior: Behavior, match_cou ) +def _send_without_attachments_behavior(config: CampaignConfig) -> Behavior: + return config.attachments.send_without_attachments_behavior or ( + Behavior.CONTINUE if config.attachments.send_without_attachments else Behavior.BLOCK + ) + + +def _issue_for_missing_attachment_coverage(behavior: Behavior) -> AttachmentIssue: + messages = { + Behavior.BLOCK: "No attachment file was resolved for this message, and campaign policy blocks sending without attachments.", + Behavior.ASK: "No attachment file was resolved for this message. Confirm the message should still be sent.", + Behavior.DROP: "No attachment file was resolved for this message. Campaign policy excludes messages without attachments.", + Behavior.WARN: "No attachment file was resolved for this message. Campaign policy allows sending with a warning.", + } + return AttachmentIssue( + severity=ResolutionSeverity.ERROR if behavior == Behavior.BLOCK else ResolutionSeverity.WARNING, + code="missing_attachment_coverage", + message=messages.get(behavior, "No attachment file was resolved for this message."), + behavior=behavior, + ) + + def _resolve_one_config( *, campaign_file: str | Path, @@ -495,6 +516,15 @@ def resolve_entry_attachments( ) issues = [issue for item in resolved for issue in item.issues] + missing_coverage_behavior = _send_without_attachments_behavior(config) + if ( + entry.active + and resolved + and missing_coverage_behavior != Behavior.CONTINUE + and sum(len(item.matches) for item in resolved) == 0 + and not any(issue.behavior == Behavior.BLOCK for issue in issues) + ): + issues.append(_issue_for_missing_attachment_coverage(missing_coverage_behavior)) return EntryAttachmentResolution( entry_index=entry_index, entry_id=entry.id, diff --git a/src/govoplan_campaign/backend/campaign/models.py b/src/govoplan_campaign/backend/campaign/models.py index a02b59f..f2ace24 100644 --- a/src/govoplan_campaign/backend/campaign/models.py +++ b/src/govoplan_campaign/backend/campaign/models.py @@ -404,11 +404,20 @@ class AttachmentsConfig(StrictModel): base_paths: list[AttachmentBasePathConfig] = Field(default_factory=list) allow_individual: bool = False send_without_attachments: bool = True + send_without_attachments_behavior: Behavior | None = None zip: ZipCollectionConfig = Field(default_factory=ZipCollectionConfig) global_: list[AttachmentConfig] = Field(default_factory=list, alias="global") missing_behavior: Behavior = Behavior.ASK ambiguous_behavior: Behavior = Behavior.ASK + @model_validator(mode="after") + def normalize_send_without_attachments_behavior(self) -> "AttachmentsConfig": + if self.send_without_attachments_behavior is None: + self.send_without_attachments_behavior = Behavior.CONTINUE if self.send_without_attachments else Behavior.BLOCK + else: + self.send_without_attachments = self.send_without_attachments_behavior != Behavior.BLOCK + return self + @property def individual_base_path_values(self) -> set[str]: return {base_path.path for base_path in self.base_paths if base_path.allow_individual} diff --git a/src/govoplan_campaign/backend/campaign/validation.py b/src/govoplan_campaign/backend/campaign/validation.py index f19408f..e511bb1 100644 --- a/src/govoplan_campaign/backend/campaign/validation.py +++ b/src/govoplan_campaign/backend/campaign/validation.py @@ -10,6 +10,7 @@ from pydantic import BaseModel, ConfigDict, Field from .field_values import ignored_entry_field_overrides from .models import AttachmentConfig, CampaignConfig, EntryConfig, FieldType, SourceType, ZipArchiveConfig, ZipPasswordMode, ZipPasswordScope, ZipRuleMode +from ..attachments.resolver import resolve_campaign_attachments class Severity(StrEnum): @@ -507,6 +508,7 @@ def _entries_source_file_issues(config: CampaignConfig, campaign_path: Path, map def _file_check_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] issues.extend(_attachment_file_check_issues(config, campaign_path)) + issues.extend(_attachment_resolution_check_issues(config, campaign_path)) issues.extend(_template_source_file_issues(config, campaign_path)) return issues @@ -536,6 +538,37 @@ def _attachment_file_check_issues(config: CampaignConfig, campaign_path: Path) - ] +def _attachment_resolution_check_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: + try: + report = resolve_campaign_attachments(config, campaign_file=campaign_path) + except Exception as exc: + return [ + _issue( + Severity.ERROR, + "attachment_resolution_failed", + f"attachment rules could not be resolved: {exc}", + "/attachments", + ) + ] + + issues: list[SemanticIssue] = [] + for entry in report.entries: + entry_path = f"/entries/{entry.entry_id or entry.entry_index}" + for issue in entry.issues: + if any(issue is attachment_issue for attachment in entry.attachments for attachment_issue in attachment.issues): + continue + issues.append(_issue(Severity(issue.severity.value), issue.code, issue.message, entry_path)) + for attachment in entry.attachments: + attachment_path = ( + f"/attachments/global/{attachment.index}" + if attachment.scope.value == "global" + else f"{entry_path}/attachments/{attachment.index}" + ) + for issue in attachment.issues: + issues.append(_issue(Severity(issue.severity.value), issue.code, issue.message, attachment_path)) + return issues + + def _template_source_file_issues(config: CampaignConfig, campaign_path: Path) -> list[SemanticIssue]: issues: list[SemanticIssue] = [] for schema_path, raw_path in _iter_template_source_paths(config): diff --git a/src/govoplan_campaign/backend/messages/builder.py b/src/govoplan_campaign/backend/messages/builder.py index ce629bc..1caa164 100644 --- a/src/govoplan_campaign/backend/messages/builder.py +++ b/src/govoplan_campaign/backend/messages/builder.py @@ -245,6 +245,26 @@ def _message_issues_from_attachment_resolution(resolution: EntryAttachmentResolu ] +def _append_no_attachment_coverage_issue(issues: list[MessageIssue]) -> None: + if any(issue.code == "missing_attachment_coverage" for issue in issues): + return + issues.append( + MessageIssue( + severity="error", + code="missing_attachment_coverage", + message="No attachment file was resolved for this message, and sending without attachments is not allowed.", + behavior=Behavior.BLOCK.value, + source="attachments", + ) + ) + + +def _send_without_attachments_behavior(config: CampaignConfig) -> Behavior: + return config.attachments.send_without_attachments_behavior or ( + Behavior.CONTINUE if config.attachments.send_without_attachments else Behavior.BLOCK + ) + + def _safe_filename(value: str | None, fallback: str) -> str: raw = value or fallback safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", raw).strip("._") @@ -391,6 +411,7 @@ def _attach_files( for attachment in resolution.attachments: attachment.message_filenames = [] attachment.zip_entry_names = [] + attachment.zip_filename = None for attachment in resolution.attachments: if attachment.status != AttachmentMatchStatus.OK or not attachment.matches: @@ -435,6 +456,7 @@ def _attach_files( work_dir / "_zip" / f"entry-{entry_index:04d}" / _safe_filename(archive.id, "archive") / filename, members, password, + archive.method.value, ) data, maintype, subtype = _attachment_bytes(archive_path) message.add_attachment(data, maintype=maintype, subtype=subtype, filename=filename) @@ -740,6 +762,14 @@ def _build_mime_message( values=rendered.values, work_dir=work_dir, ) + if attachment_count == 0 and context.resolution.attachments and _send_without_attachments_behavior(config) == Behavior.BLOCK: + _append_no_attachment_coverage_issue(context.issues) + return _MimeBuildResult( + message=None, + build_status=BuildStatus.BUILD_FAILED, + validation_status=MessageValidationStatus.BLOCKED, + attachment_count=0, + ) return _MimeBuildResult( message=message, build_status=BuildStatus.BUILT, diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index 272256f..41624cb 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -158,6 +158,7 @@ def minimal_campaign_json(*, external_id: str, name: str, description: str | Non ], "allow_individual": True, "send_without_attachments": False, + "send_without_attachments_behavior": "block", "global": [], "zip": {"enabled": False, "archives": []}, "missing_behavior": "ask", diff --git a/src/govoplan_campaign/backend/schema/campaign.schema.json b/src/govoplan_campaign/backend/schema/campaign.schema.json index aa93351..28ca9ee 100644 --- a/src/govoplan_campaign/backend/schema/campaign.schema.json +++ b/src/govoplan_campaign/backend/schema/campaign.schema.json @@ -419,7 +419,19 @@ "send_without_attachments": { "type": "boolean", "default": true, - "description": "Legacy compatibility flag. Prefer validation_policy and per-config missing_behavior for new campaigns." + "description": "Legacy compatibility flag. Use send_without_attachments_behavior for new campaigns." + }, + "send_without_attachments_behavior": { + "type": "string", + "enum": [ + "block", + "ask", + "drop", + "continue", + "warn" + ], + "default": "continue", + "description": "Campaign-wide behavior when no attachment file is resolved for an active message." }, "zip": { "$ref": "#/$defs/zip_collection_config", diff --git a/src/govoplan_campaign/backend/services/zip_service.py b/src/govoplan_campaign/backend/services/zip_service.py index d732a8a..df81b02 100644 --- a/src/govoplan_campaign/backend/services/zip_service.py +++ b/src/govoplan_campaign/backend/services/zip_service.py @@ -1,8 +1,14 @@ from __future__ import annotations +import binascii +from datetime import datetime +import secrets +import stat +import struct import zipfile from pathlib import Path from typing import Iterable +import zlib try: import pyzipper @@ -10,6 +16,8 @@ except ImportError: # pragma: no cover pyzipper = None ArchiveMember = tuple[Path, str] +ZIP_METHOD_AES = "aes" +ZIP_METHOD_STANDARD = "zip_standard" def _normalized_members(files: Iterable[Path | ArchiveMember]) -> list[ArchiveMember]: @@ -34,23 +42,17 @@ def create_zip_archive( output_path: Path, files: Iterable[Path | ArchiveMember], password: str = "", + method: str = ZIP_METHOD_AES, ) -> Path: - """Create a ZIP archive, using AES encryption when a password is supplied.""" + """Create a ZIP archive, optionally using AES or legacy ZipCrypto encryption.""" output_path.parent.mkdir(parents=True, exist_ok=True) members = _normalized_members(files) if password: - if pyzipper is None: - raise RuntimeError("pyzipper is required for writing password-protected ZIP files") - with pyzipper.AESZipFile( - output_path, - "w", - compression=pyzipper.ZIP_DEFLATED, - encryption=pyzipper.WZ_AES, - ) as zip_file: - zip_file.setpassword(password.encode("utf-8")) - for file_path, archive_name in members: - zip_file.write(file_path, arcname=archive_name) + if method == ZIP_METHOD_STANDARD: + _create_zipcrypto_archive(output_path, members, password) + return output_path + _create_aes_archive(output_path, members, password) return output_path with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zip_file: @@ -59,7 +61,214 @@ def create_zip_archive( return output_path -def create_encrypted_zip(output_path: Path, files: list[Path], password: str) -> Path: +def create_encrypted_zip(output_path: Path, files: list[Path], password: str, method: str = ZIP_METHOD_AES) -> Path: """Backward-compatible wrapper for the original per-rule ZIP helper.""" - return create_zip_archive(output_path, files, password) + return create_zip_archive(output_path, files, password, method) + + +def _create_aes_archive(output_path: Path, members: list[ArchiveMember], password: str) -> None: + if pyzipper is None: + raise RuntimeError("pyzipper is required for writing AES password-protected ZIP files") + with pyzipper.AESZipFile( + output_path, + "w", + compression=pyzipper.ZIP_DEFLATED, + encryption=pyzipper.WZ_AES, + ) as zip_file: + zip_file.setpassword(password.encode("utf-8")) + for file_path, archive_name in members: + zip_file.write(file_path, arcname=archive_name) + + +def _create_zipcrypto_archive(output_path: Path, members: list[ArchiveMember], password: str) -> None: + entries: list[_CentralDirectoryEntry] = [] + password_bytes = password.encode("utf-8") + with output_path.open("wb") as zip_file: + for file_path, archive_name in members: + local_header_offset = zip_file.tell() + file_data = file_path.read_bytes() + crc = binascii.crc32(file_data) & 0xFFFFFFFF + compressed_data = _deflate(file_data) + encrypted_data = _zipcrypto_encrypt(compressed_data, password_bytes, crc) + compressed_size = len(encrypted_data) + uncompressed_size = len(file_data) + modified_at = datetime.fromtimestamp(file_path.stat().st_mtime) + dos_time, dos_date = _dos_timestamp(modified_at) + filename_bytes, flags = _filename_bytes(archive_name, encrypted=True) + + zip_file.write( + struct.pack( + " None: + self.filename_bytes = filename_bytes + self.flags = flags + self.dos_time = dos_time + self.dos_date = dos_date + self.crc = crc + self.compressed_size = compressed_size + self.uncompressed_size = uncompressed_size + self.external_attributes = external_attributes + self.local_header_offset = local_header_offset + + +def _deflate(data: bytes) -> bytes: + compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -zlib.MAX_WBITS) + return compressor.compress(data) + compressor.flush() + + +def _zipcrypto_encrypt(data: bytes, password: bytes, crc: int) -> bytes: + cipher = _ZipCrypto(password) + header = secrets.token_bytes(11) + bytes([(crc >> 24) & 0xFF]) + return cipher.encrypt(header + data) + + +class _ZipCrypto: + def __init__(self, password: bytes) -> None: + self.key0 = 0x12345678 + self.key1 = 0x23456789 + self.key2 = 0x34567890 + for value in password: + self._update_keys(value) + + def encrypt(self, data: bytes) -> bytes: + encrypted = bytearray() + for value in data: + encrypted.append(value ^ self._decrypt_byte()) + self._update_keys(value) + return bytes(encrypted) + + def _decrypt_byte(self) -> int: + temp = self.key2 | 2 + return ((temp * (temp ^ 1)) >> 8) & 0xFF + + def _update_keys(self, value: int) -> None: + self.key0 = _zipcrypto_crc32_byte(value, self.key0) + self.key1 = (self.key1 + (self.key0 & 0xFF)) & 0xFFFFFFFF + self.key1 = (self.key1 * 134775813 + 1) & 0xFFFFFFFF + self.key2 = _zipcrypto_crc32_byte((self.key1 >> 24) & 0xFF, self.key2) + + +def _build_zipcrypto_crc(value: int) -> int: + for _ in range(8): + if value & 1: + value = 0xEDB88320 ^ (value >> 1) + else: + value >>= 1 + return value + + +_ZIPCRYPTO_CRC_TABLE = [_build_zipcrypto_crc(value) for value in range(256)] + + +def _zipcrypto_crc32_byte(value: int, crc: int) -> int: + return ((crc >> 8) ^ _ZIPCRYPTO_CRC_TABLE[(crc ^ value) & 0xFF]) & 0xFFFFFFFF + + +def _filename_bytes(filename: str, *, encrypted: bool) -> tuple[bytes, int]: + flags = 0x1 if encrypted else 0 + try: + return filename.encode("ascii"), flags + except UnicodeEncodeError: + return filename.encode("utf-8"), flags | 0x800 + + +def _dos_timestamp(value: datetime) -> tuple[int, int]: + year = min(max(value.year, 1980), 2107) + month = value.month if value.year == year else 1 + day = value.day if value.year == year else 1 + dos_time = value.hour << 11 | value.minute << 5 | value.second // 2 + dos_date = (year - 1980) << 9 | month << 5 | day + return dos_time, dos_date + + +def _external_attributes(path: Path) -> int: + try: + permissions = stat.S_IMODE(path.stat().st_mode) + except OSError: + return 0 + return permissions << 16 diff --git a/tests/test_attachment_building.py b/tests/test_attachment_building.py new file mode 100644 index 0000000..1ce0f91 --- /dev/null +++ b/tests/test_attachment_building.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import io +import tempfile +import unittest +import zipfile +from pathlib import Path + +from govoplan_campaign.backend.campaign.models import CampaignConfig +from govoplan_campaign.backend.campaign.validation import validate_campaign_config +from govoplan_campaign.backend.messages.builder import build_campaign_messages + + +class CampaignAttachmentBuildTests(unittest.TestCase): + def _no_attachment_config( + self, + behavior: str | None = None, + legacy_allow: bool | None = None, + *, + configure_missing_rule: bool = False, + ) -> CampaignConfig: + attachments: dict[str, object] = { + "base_path": ".", + "global": [], + "zip": {"enabled": False, "archives": []}, + } + if configure_missing_rule: + attachments["global"] = [ + { + "id": "missing", + "label": "Missing attachment", + "base_dir": ".", + "file_filter": "missing.pdf", + "required": False, + "missing_behavior": "continue", + } + ] + if behavior is not None: + attachments["send_without_attachments_behavior"] = behavior + if legacy_allow is not None: + attachments["send_without_attachments"] = legacy_allow + return CampaignConfig.model_validate({ + "version": "1.0", + "campaign": {"id": f"no-attachment-{behavior or legacy_allow}", "name": "No attachment policy", "mode": "test"}, + "fields": [], + "global_values": {}, + "server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, + "recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True}, + "template": {"subject": "Subject", "text": "Body"}, + "attachments": attachments, + "entries": {"inline": [{"id": "recipient-1", "to": [{"email": "recipient@example.org", "type": "to"}]}]}, + "validation_policy": {"missing_email": "block", "template_error": "block"}, + "delivery": {"imap_append_sent": {"enabled": False}}, + }) + + def test_send_without_attachments_policy_does_not_block_when_no_rules_are_configured(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + campaign_file = root / "campaign.json" + campaign_file.write_text("{}", encoding="utf-8") + config = self._no_attachment_config(legacy_allow=False) + + validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True) + self.assertTrue(validation.ok) + self.assertNotIn("missing_attachment_coverage", {issue.code for issue in validation.issues}) + + result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=True) + self.assertEqual(result.report.build_failed_count, 0) + self.assertEqual(result.report.queueable_count, 1) + message = result.report.messages[0] + self.assertEqual(message.attachment_count, 0) + self.assertIsNotNone(message.eml_path) + self.assertNotIn("missing_attachment_coverage", {issue.code for issue in message.issues}) + self.assertIsNotNone(result.built_messages[0].mime) + + def test_configured_attachment_rule_blocks_generation_when_no_files_resolve(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + campaign_file = root / "campaign.json" + campaign_file.write_text("{}", encoding="utf-8") + config = self._no_attachment_config(behavior="block", configure_missing_rule=True) + + validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True) + self.assertFalse(validation.ok) + self.assertIn("missing_attachment_coverage", {issue.code for issue in validation.issues}) + + result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=True) + self.assertEqual(result.report.build_failed_count, 1) + self.assertEqual(result.report.queueable_count, 0) + message = result.report.messages[0] + self.assertEqual(message.attachment_count, 0) + self.assertIsNone(message.eml_path) + self.assertIn("missing_attachment_coverage", {issue.code for issue in message.issues}) + self.assertIsNone(result.built_messages[0].mime) + + def test_send_without_attachments_behavior_modes(self) -> None: + cases = { + "block": ("build_failed", "blocked", 0, "block", False), + "ask": ("built", "needs_review", 0, "ask", True), + "drop": ("built", "excluded", 0, "drop", True), + "warn": ("built", "warning", 1, "warn", True), + "continue": ("built", "ready", 1, None, True), + } + for behavior, (build_status, validation_status, queueable_count, issue_behavior, has_mime) in cases.items(): + with self.subTest(behavior=behavior): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + campaign_file = root / "campaign.json" + campaign_file.write_text("{}", encoding="utf-8") + config = self._no_attachment_config(behavior=behavior, configure_missing_rule=True) + + validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True) + if behavior == "block": + self.assertFalse(validation.ok) + else: + self.assertTrue(validation.ok) + + result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=True) + self.assertEqual(result.report.queueable_count, queueable_count) + message = result.report.messages[0] + self.assertEqual(message.build_status.value, build_status) + self.assertEqual(message.validation_status.value, validation_status) + coverage_issues = [issue for issue in message.issues if issue.code == "missing_attachment_coverage"] + if issue_behavior is None: + self.assertEqual(coverage_issues, []) + else: + self.assertEqual(coverage_issues[0].behavior, issue_behavior) + self.assertEqual(result.built_messages[0].mime is not None, has_mime) + + def test_missing_pattern_does_not_create_zip_member_or_count_as_attachment(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + documents = root / "documents" + documents.mkdir() + (documents / "matched.xlsx").write_bytes(b"matched workbook") + campaign_file = root / "campaign.json" + campaign_file.write_text("{}", encoding="utf-8") + config = CampaignConfig.model_validate({ + "version": "1.0", + "campaign": {"id": "zip-missing-pattern", "name": "ZIP missing pattern", "mode": "test"}, + "fields": [], + "global_values": {}, + "server": {"smtp": {"host": "smtp.example.invalid", "port": 587, "security": "starttls"}}, + "recipients": {"from": {"email": "sender@example.org", "type": "to"}, "allow_individual_to": True}, + "template": {"subject": "Subject", "text": "Body"}, + "attachments": { + "base_path": ".", + "allow_individual": True, + "send_without_attachments": False, + "zip": { + "enabled": True, + "archives": [{"id": "bundle", "name": "recipient-bundle.zip", "standard": True}], + }, + "global": [], + }, + "entries": { + "inline": [{ + "id": "recipient-1", + "to": [{"email": "recipient@example.org", "type": "to"}], + "attachments": [ + {"id": "matched", "base_dir": "documents", "file_filter": "matched.xlsx", "required": True}, + { + "id": "missing", + "base_dir": "documents", + "file_filter": "missing.xlsx", + "required": False, + "missing_behavior": "warn", + }, + ], + }] + }, + "validation_policy": { + "missing_email": "block", + "template_error": "block", + "missing_optional_attachment": "warn", + }, + "delivery": {"imap_append_sent": {"enabled": False}}, + }) + + validation = validate_campaign_config(config, campaign_file=campaign_file, check_files=True) + self.assertTrue(validation.ok) + self.assertIn("missing_optional_attachment", {issue.code for issue in validation.issues}) + + result = build_campaign_messages(config, campaign_file=campaign_file, output_dir=root / "out", write_eml=False) + self.assertEqual(result.report.built_count, 1) + self.assertEqual(result.report.queueable_count, 1) + message = result.report.messages[0] + self.assertEqual(message.attachment_count, 1) + summaries = {attachment.attachment_id: attachment for attachment in message.attachments} + self.assertEqual(summaries["matched"].zip_filename, "recipient-bundle.zip") + self.assertEqual(summaries["matched"].zip_entry_names, ["matched.xlsx"]) + self.assertIsNone(summaries["missing"].zip_filename) + self.assertEqual(summaries["missing"].zip_entry_names, []) + self.assertEqual(summaries["missing"].matches, []) + + mime = result.built_messages[0].mime + self.assertIsNotNone(mime) + attachments = {part.get_filename(): part.get_payload(decode=True) for part in mime.iter_attachments()} + self.assertEqual(set(attachments), {"recipient-bundle.zip"}) + with zipfile.ZipFile(io.BytesIO(attachments["recipient-bundle.zip"])) as archive: + self.assertEqual(archive.namelist(), ["matched.xlsx"]) + self.assertEqual(archive.read("matched.xlsx"), b"matched workbook") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_zip_service.py b/tests/test_zip_service.py new file mode 100644 index 0000000..a82bfee --- /dev/null +++ b/tests/test_zip_service.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +from pathlib import Path +import tempfile +import unittest +import zipfile + +try: + import pyzipper +except ImportError: # pragma: no cover + pyzipper = None + +from govoplan_campaign.backend.services.zip_service import create_zip_archive + + +class ZipServiceTests(unittest.TestCase): + def test_standard_password_zip_uses_zipcrypto_readable_by_stdlib(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "source.txt" + source.write_text("Hello Windows ZIP", encoding="utf-8") + output = root / "standard.zip" + + create_zip_archive(output, [(source, "message.txt")], "secret", "zip_standard") + + with zipfile.ZipFile(output) as archive: + info = archive.getinfo("message.txt") + 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") + + @unittest.skipIf(pyzipper is None, "pyzipper is not installed") + def test_aes_password_zip_keeps_aes_encryption(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "source.txt" + source.write_text("Hello AES ZIP", encoding="utf-8") + output = root / "aes.zip" + + create_zip_archive(output, [(source, "message.txt")], "secret", "aes") + + with zipfile.ZipFile(output) as archive: + info = archive.getinfo("message.txt") + self.assertEqual(info.compress_type, 99) + self.assertTrue(info.flag_bits & 0x1) + + with pyzipper.AESZipFile(output) as archive: + archive.setpassword(b"secret") + self.assertEqual(archive.read("message.txt"), b"Hello AES ZIP") + + def test_unprotected_zip_ignores_encryption_mode(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + source = root / "source.txt" + source.write_text("Plain ZIP", encoding="utf-8") + output = root / "plain.zip" + + create_zip_archive(output, [(source, "message.txt")], "", "zip_standard") + + with zipfile.ZipFile(output) as archive: + info = archive.getinfo("message.txt") + self.assertEqual(info.compress_type, zipfile.ZIP_DEFLATED) + self.assertFalse(info.flag_bits & 0x1) + self.assertEqual(archive.read("message.txt"), b"Plain ZIP") + + +if __name__ == "__main__": + unittest.main()