From 1d291377c0791bfe6a9f9073d944576ca65266a8 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 03:17:11 +0200 Subject: [PATCH] Block local file paths at Campaign server boundaries --- README.md | 2 +- .../backend/attachments/resolver.py | 63 +++- .../backend/dev/mock_campaign.py | 5 + .../backend/path_security.py | 170 +++++++++++ .../backend/persistence/campaigns.py | 18 +- .../backend/persistence/versions.py | 20 +- src/govoplan_campaign/backend/router.py | 27 +- .../backend/sending/execution.py | 11 +- tests/test_server_path_security.py | 279 ++++++++++++++++++ 9 files changed, 576 insertions(+), 19 deletions(-) create mode 100644 src/govoplan_campaign/backend/path_security.py create mode 100644 tests/test_server_path_security.py diff --git a/README.md b/README.md index 6a44816..8d75abf 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ The module has one required runtime dependency: Files and mail are optional module integrations declared in the campaign manifest: -- `govoplan-files` enables managed attachment selection, frozen file-version evidence, and managed-file usage tracking. Without it, campaigns can still use legacy/local attachment paths where configured. +- `govoplan-files` enables managed attachment selection, frozen file-version evidence, and managed-file usage tracking. Server/API campaigns require this integration for attachments and never resolve caller-supplied local filesystem paths. Legacy file-oriented loading remains available only to explicitly trusted operator/library workflows. - `govoplan-mail` enables reusable mail profiles, delivery policy checks, SMTP sending, and IMAP append behavior. Without it, campaigns can still be authored, validated, built, and reported, but real delivery/profile features are unavailable. Backend optional behavior is accessed through core-provided capabilities, not direct required imports. WebUI optional behavior uses core module metadata/capabilities so campaign pages can build and run without files or mail WebUI packages installed. diff --git a/src/govoplan_campaign/backend/attachments/resolver.py b/src/govoplan_campaign/backend/attachments/resolver.py index a909a48..fa0655e 100644 --- a/src/govoplan_campaign/backend/attachments/resolver.py +++ b/src/govoplan_campaign/backend/attachments/resolver.py @@ -13,6 +13,11 @@ from pydantic import BaseModel, ConfigDict, Field from govoplan_campaign.backend.campaign.entries import load_campaign_entries from govoplan_campaign.backend.campaign.template_values import build_template_values from govoplan_campaign.backend.campaign.models import AttachmentBasePathConfig, AttachmentConfig, Behavior, CampaignConfig, EntryConfig, ZipArchiveConfig, ZipRuleMode +from govoplan_campaign.backend.path_security import ( + CampaignPathSecurityError, + assert_logical_relative_path, + is_managed_source, +) class AttachmentScope(StrEnum): @@ -369,6 +374,23 @@ def _match_files(directory: Path, file_filter: str, include_subdirs: bool, match return sorted(path for path in directory.glob(file_filter) if path.is_file()) +def _confine_managed_matches(directory: Path, matches: list[Path]) -> tuple[list[Path], bool]: + root = directory.resolve() + confined: list[Path] = [] + rejected = False + for match in matches: + try: + resolved = match.resolve() + except (OSError, RuntimeError): + rejected = True + continue + if not resolved.is_relative_to(root): + rejected = True + continue + confined.append(match) + return confined, rejected + + def _issue_for_missing(config: AttachmentConfig, behavior: Behavior) -> AttachmentIssue: code = "missing_required_attachment" if config.required else "missing_optional_attachment" severity = ResolutionSeverity.ERROR if config.required and behavior == Behavior.BLOCK else ResolutionSeverity.WARNING @@ -429,13 +451,50 @@ def _resolve_one_config( attachment_config=config, rendered_base_dir=rendered_base_dir, ) - matches = _match_files(directory, rendered_file_filter, config.include_subdirs, match_index) allow_multiple = _rule_allows_multiple(config, rendered_file_filter) issues: list[AttachmentIssue] = [] behavior: Behavior | None = None + managed_source = selected_base_path is not None and is_managed_source(selected_base_path.source) + unsafe_managed_path = False + if managed_source: + try: + assert_logical_relative_path( + rendered_file_filter, + field="rendered managed attachment file_filter", + ) + except CampaignPathSecurityError as exc: + matches = [] + unsafe_managed_path = True + issues.append( + AttachmentIssue( + severity=ResolutionSeverity.ERROR, + code="unsafe_managed_attachment_path", + message=str(exc), + behavior=Behavior.BLOCK, + ) + ) + else: + matches = _match_files(directory, rendered_file_filter, config.include_subdirs, match_index) + matches, rejected = _confine_managed_matches(directory, matches) + if rejected: + matches = [] + unsafe_managed_path = True + issues.append( + AttachmentIssue( + severity=ResolutionSeverity.ERROR, + code="managed_attachment_path_escape", + message="A managed attachment match resolved outside its materialized source directory.", + behavior=Behavior.BLOCK, + ) + ) + else: + matches = _match_files(directory, rendered_file_filter, config.include_subdirs, match_index) - if not matches: + if unsafe_managed_path: + status = AttachmentMatchStatus.MISSING + behavior = Behavior.BLOCK + elif not matches: status = AttachmentMatchStatus.MISSING behavior = _missing_behavior(campaign_config, config) issues.append(_issue_for_missing(config, behavior)) diff --git a/src/govoplan_campaign/backend/dev/mock_campaign.py b/src/govoplan_campaign/backend/dev/mock_campaign.py index 271b241..c12903a 100644 --- a/src/govoplan_campaign/backend/dev/mock_campaign.py +++ b/src/govoplan_campaign/backend/dev/mock_campaign.py @@ -14,6 +14,7 @@ from govoplan_campaign.backend.persistence.campaigns import load_campaign_config from govoplan_campaign.backend.messages.builder import build_campaign_messages from govoplan_campaign.backend.messages.models import MessageAddress, MessageDraft, MessageValidationStatus from govoplan_campaign.backend.integrations import files_integration, mail_integration +from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths class MockCampaignSendError(RuntimeError): @@ -334,6 +335,10 @@ def run_mock_campaign_send( mailbox.clear_records() files = files_integration() + assert_server_safe_campaign_paths( + version.raw_json if isinstance(version.raw_json, dict) else {}, + managed_files_available=files.available, + ) with files.prepared_campaign_snapshot( session, tenant_id=tenant_id, diff --git a/src/govoplan_campaign/backend/path_security.py b/src/govoplan_campaign/backend/path_security.py new file mode 100644 index 0000000..e149bff --- /dev/null +++ b/src/govoplan_campaign/backend/path_security.py @@ -0,0 +1,170 @@ +from __future__ import annotations + +from pathlib import PureWindowsPath +from typing import Any, Iterable + + +class CampaignPathSecurityError(ValueError): + """Raised when server-provided campaign JSON could read local files.""" + + +def _text(value: object) -> str: + return value.strip() if isinstance(value, str) else "" + + +def assert_logical_relative_path(value: object, *, field: str) -> None: + raw = _text(value) + if not raw: + return + if "\x00" in raw: + raise CampaignPathSecurityError(f"{field} contains a NUL byte") + + normalized = raw.replace("\\", "/") + windows_path = PureWindowsPath(raw) + if ( + normalized.startswith("/") + or normalized.startswith("~") + or windows_path.is_absolute() + or bool(windows_path.drive) + ): + raise CampaignPathSecurityError(f"{field} must be a relative managed-file path") + if any(part == ".." for part in normalized.split("/")): + raise CampaignPathSecurityError(f"{field} must not contain parent-directory traversal") + + +def is_managed_source(value: object) -> bool: + raw = _text(value) + if not raw.startswith("managed:"): + return False + parts = raw.split(":", 2) + return len(parts) == 3 and parts[1] in {"user", "group"} and bool(parts[2].strip()) + + +def _attachment_rules(raw_json: dict[str, Any]) -> Iterable[tuple[str, dict[str, Any]]]: + attachments = raw_json.get("attachments") + if isinstance(attachments, dict): + global_rules = attachments.get("global") + if isinstance(global_rules, list): + for index, rule in enumerate(global_rules): + if isinstance(rule, dict): + yield f"attachments.global[{index}]", rule + + entries = raw_json.get("entries") + inline = entries.get("inline") if isinstance(entries, dict) else None + if isinstance(inline, list): + for entry_index, entry in enumerate(inline): + rules = entry.get("attachments") if isinstance(entry, dict) else None + if not isinstance(rules, list): + continue + for rule_index, rule in enumerate(rules): + if isinstance(rule, dict): + yield f"entries.inline[{entry_index}].attachments[{rule_index}]", rule + defaults = entries.get("defaults") if isinstance(entries, dict) else None + default_rules = defaults.get("attachments") if isinstance(defaults, dict) else None + if isinstance(default_rules, list): + for rule_index, rule in enumerate(default_rules): + if isinstance(rule, dict): + yield f"entries.defaults.attachments[{rule_index}]", rule + + +def _assert_inline_server_sources( + raw_json: dict[str, Any], + *, + source_filename: str | None, + source_base_path: str | None, +) -> None: + if _text(source_filename): + raise CampaignPathSecurityError("source_filename is not accepted for server-managed campaigns") + if _text(source_base_path): + raise CampaignPathSecurityError("source_base_path is not accepted for server-managed campaigns") + + template = raw_json.get("template") + template_source = template.get("source") if isinstance(template, dict) else None + path_fields = ("subject_path", "text_path", "html_path") + if isinstance(template_source, dict) and any(_text(template_source.get(key)) for key in path_fields): + raise CampaignPathSecurityError( + "template source paths are not accepted for server-managed campaigns; store template content inline" + ) + + entries = raw_json.get("entries") + entries_source = entries.get("source") if isinstance(entries, dict) else None + if isinstance(entries_source, dict) and _text(entries_source.get("path")): + raise CampaignPathSecurityError( + "entries.source.path is not accepted for server-managed campaigns; import recipients into the campaign snapshot" + ) + + +def _managed_attachment_base_paths(attachments: dict[str, Any]) -> dict[str, dict[str, Any]]: + assert_logical_relative_path(attachments.get("base_path"), field="attachments.base_path") + raw_base_paths = attachments.get("base_paths") + base_paths = raw_base_paths if isinstance(raw_base_paths, list) else [] + managed: dict[str, dict[str, Any]] = {} + for index, item in enumerate(base_paths): + if not isinstance(item, dict): + continue + assert_logical_relative_path( + item.get("path"), + field=f"attachments.base_paths[{index}].path", + ) + source_is_managed = is_managed_source(item.get("source")) + base_path_id = _text(item.get("id")) + if base_path_id and source_is_managed: + managed[base_path_id] = item + if item.get("unsent_warning") is True and not source_is_managed: + raise CampaignPathSecurityError( + f"attachments.base_paths[{index}] must use a managed Files source before unsent-file scanning is enabled" + ) + return managed + + +def _assert_managed_attachment_rules( + raw_json: dict[str, Any], + *, + managed_base_paths: dict[str, dict[str, Any]], + managed_files_available: bool, +) -> None: + rules = list(_attachment_rules(raw_json)) + if not rules: + return + if not managed_files_available: + raise CampaignPathSecurityError("campaign attachments require the Files module in server mode") + + for field, rule in rules: + base_path_id = _text(rule.get("base_path_id")) + if not base_path_id or base_path_id not in managed_base_paths: + raise CampaignPathSecurityError( + f"{field}.base_path_id must select an attachments.base_paths entry backed by managed Files" + ) + assert_logical_relative_path(rule.get("base_dir"), field=f"{field}.base_dir") + assert_logical_relative_path(rule.get("file_filter"), field=f"{field}.file_filter") + + +def assert_server_safe_campaign_paths( + raw_json: dict[str, Any], + *, + source_filename: str | None = None, + source_base_path: str | None = None, + managed_files_available: bool, +) -> None: + """Reject local-file references at the HTTP/server trust boundary. + + File-oriented campaign loading remains available to trusted operator code + through the campaign loader and message builder. Persisted/API campaigns + must use inline templates and recipients plus managed Files attachment + sources, which are materialized into an isolated snapshot before use. + """ + + _assert_inline_server_sources( + raw_json, + source_filename=source_filename, + source_base_path=source_base_path, + ) + + attachments = raw_json.get("attachments") + if not isinstance(attachments, dict): + return + _assert_managed_attachment_rules( + raw_json, + managed_base_paths=_managed_attachment_base_paths(attachments), + managed_files_available=managed_files_available, + ) diff --git a/src/govoplan_campaign/backend/persistence/campaigns.py b/src/govoplan_campaign/backend/persistence/campaigns.py index 40e23d6..9b6edb1 100644 --- a/src/govoplan_campaign/backend/persistence/campaigns.py +++ b/src/govoplan_campaign/backend/persistence/campaigns.py @@ -32,6 +32,7 @@ from govoplan_campaign.backend.messages.models import MessageDraft from govoplan_campaign.backend.sending.execution import create_execution_snapshot from govoplan_campaign.backend.campaign.models import CampaignConfig from govoplan_campaign.backend.integrations import files_integration, mail_integration +from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths RUNTIME_DIR = Path(__file__).resolve().parents[3] / "runtime" CAMPAIGN_SNAPSHOT_DIR = RUNTIME_DIR / "campaign_snapshots" @@ -92,13 +93,14 @@ def _resolve_runtime_path(base_path: Path | None, value: str | None) -> str | No def normalize_campaign_paths(raw_json: dict[str, Any], source_base_path: str | Path | None) -> dict[str, Any]: - """Return a DB/runtime-safe campaign JSON snapshot. + """Resolve paths for an explicitly trusted, file-oriented import. The CLI naturally resolves relative paths against the campaign.json file. Once the campaign is stored in the database, the JSON snapshot lives in app/mailer/runtime/campaign_snapshots. To keep existing file-based campaigns working, relative file paths are normalized to absolute paths at - import time when a source_base_path is known. + import time when a source_base_path is known. HTTP/API callers are rejected + by ``assert_server_safe_campaign_paths`` before they can reach this helper. """ base = Path(source_base_path).expanduser().resolve() if source_base_path else None data = copy.deepcopy(raw_json) @@ -128,6 +130,12 @@ def create_campaign_version_from_json( source_filename: str | None = None, source_base_path: str | None = None, ) -> tuple[Campaign, CampaignVersion]: + assert_server_safe_campaign_paths( + raw_json, + source_filename=source_filename, + source_base_path=source_base_path, + managed_files_available=files_integration().available, + ) if source_base_path is None and source_filename: source_path = Path(source_filename).expanduser() source_base_path = str(source_path.parent if source_path.suffix else source_path) @@ -235,8 +243,12 @@ def load_version_config(session: Session, version_id: str): campaign = session.get(Campaign, version.campaign_id) if not campaign: raise CampaignPersistenceError(f"Campaign not found for version: {version_id}") - path = _write_campaign_snapshot(version) raw_json = version.raw_json if isinstance(version.raw_json, dict) else {} + assert_server_safe_campaign_paths( + raw_json, + managed_files_available=files_integration().available, + ) + path = _write_campaign_snapshot(version) return version, path, load_campaign_config_from_json(session, tenant_id=campaign.tenant_id, raw_json=raw_json, campaign_id=campaign.id) diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index 41624cb..e5e001e 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -19,13 +19,14 @@ from govoplan_campaign.backend.db.models import ( JobSendStatus, ) from govoplan_campaign.backend.sending.execution import clear_execution_snapshot -from govoplan_campaign.backend.integrations import mail_integration +from govoplan_campaign.backend.integrations import files_integration, mail_integration from govoplan_campaign.backend.persistence.campaigns import ( CampaignPersistenceError, _next_version_number, _write_campaign_snapshot, normalize_campaign_paths, ) +from govoplan_campaign.backend.path_security import assert_server_safe_campaign_paths class LockedCampaignVersionError(CampaignPersistenceError): @@ -120,8 +121,8 @@ def minimal_campaign_json(*, external_id: str, name: str, description: str | Non "sent_folder": "auto", }, "credentials": { - "smtp": {"username": "", "password": ""}, - "imap": {"username": "", "password": ""}, + "smtp": {"username": "", "password": ""}, # nosec B105 - empty draft placeholder. + "imap": {"username": "", "password": ""}, # nosec B105 - empty draft placeholder. }, }, "recipients": { @@ -371,6 +372,12 @@ def fork_campaign_version_for_edit( ) base_json = raw_json if raw_json is not None else copy.deepcopy(source.raw_json) + assert_server_safe_campaign_paths( + base_json, + source_filename=source_filename, + source_base_path=source_base_path, + managed_files_available=files_integration().available, + ) runtime_json = normalize_campaign_paths(base_json, source_base_path) if source_base_path else copy.deepcopy(base_json) mail_integration().assert_campaign_mail_policy_allows_json(session, tenant_id=tenant_id, raw_json=runtime_json, campaign_id=campaign.id) @@ -522,6 +529,13 @@ def update_campaign_version( source_base_path: str | None = None, autosave: bool = False, ) -> CampaignVersion: + if raw_json is not None or source_filename is not None or source_base_path is not None: + assert_server_safe_campaign_paths( + raw_json if raw_json is not None else {}, + source_filename=source_filename, + source_base_path=source_base_path, + managed_files_available=files_integration().available, + ) version = get_campaign_version_for_tenant(session, tenant_id=tenant_id, campaign_id=campaign_id, version_id=version_id) campaign = _require_campaign(session, campaign_id) ensure_current_working_version(campaign, version, action="edit") diff --git a/src/govoplan_campaign/backend/router.py b/src/govoplan_campaign/backend/router.py index dc51091..c541169 100644 --- a/src/govoplan_campaign/backend/router.py +++ b/src/govoplan_campaign/backend/router.py @@ -107,6 +107,7 @@ from govoplan_campaign.backend.persistence.campaigns import ( validate_campaign_version, ) from govoplan_campaign.backend.integrations import files_integration +from govoplan_campaign.backend.path_security import CampaignPathSecurityError, assert_server_safe_campaign_paths from govoplan_campaign.backend.campaign.loader import load_campaign_json from govoplan_campaign.backend.attachments.resolver import resolve_campaign_attachments from govoplan_core.security.time import utc_now @@ -373,6 +374,8 @@ def _campaign_version_detail_response( return CampaignVersionDetailResponse.model_validate(version) except LockedCampaignVersionError as exc: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + except CampaignPathSecurityError as exc: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except CampaignPersistenceError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: @@ -1487,6 +1490,8 @@ def fork_version_for_edit( ) except LockedCampaignVersionError as exc: raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc + except CampaignPathSecurityError as exc: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except CampaignPersistenceError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @@ -3203,6 +3208,7 @@ def _attachment_preview_for_version( include_unlinked_candidates: bool, ) -> CampaignAttachmentPreviewResponse: files = files_integration() + assert_server_safe_campaign_paths(raw, managed_files_available=files.available) with files.prepared_campaign_snapshot( session, tenant_id=principal.tenant_id, @@ -3352,15 +3358,18 @@ def preview_campaign_attachments( raw = payload.campaign_json if isinstance(payload.campaign_json, dict) else version.raw_json raw = raw if isinstance(raw, dict) else {} _require_mail_profile_use_if_needed(principal, raw) - return _attachment_preview_for_version( - session, - principal, - campaign=campaign, - version=version, - raw=raw, - include_unmatched=payload.include_unmatched, - include_unlinked_candidates=payload.include_unlinked_candidates, - ) + try: + return _attachment_preview_for_version( + session, + principal, + campaign=campaign, + version=version, + raw=raw, + include_unmatched=payload.include_unmatched, + include_unlinked_candidates=payload.include_unlinked_candidates, + ) + except CampaignPathSecurityError as exc: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc @router.post("/{campaign_id}/versions/{version_id}/attachments/link-matches", response_model=CampaignAttachmentLinkMatchesResponse) diff --git a/src/govoplan_campaign/backend/sending/execution.py b/src/govoplan_campaign/backend/sending/execution.py index 0550280..89a7dc4 100644 --- a/src/govoplan_campaign/backend/sending/execution.py +++ b/src/govoplan_campaign/backend/sending/execution.py @@ -10,7 +10,8 @@ from sqlalchemy.orm import Session from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion, JobValidationStatus from govoplan_campaign.backend.campaign.models import DeliveryConfig, ImapConfig, SmtpConfig -from govoplan_campaign.backend.integrations import MailProfileError, mail_integration +from govoplan_campaign.backend.integrations import MailProfileError, files_integration, mail_integration +from govoplan_campaign.backend.path_security import CampaignPathSecurityError, assert_server_safe_campaign_paths SNAPSHOT_VERSION = "3" @@ -250,6 +251,14 @@ def ensure_execution_snapshot(session: Session, version: CampaignVersion) -> Exe without a manual data migration. """ + try: + assert_server_safe_campaign_paths( + version.raw_json if isinstance(version.raw_json, dict) else {}, + managed_files_available=files_integration().available, + ) + except CampaignPathSecurityError as exc: + raise ExecutionSnapshotError(str(exc)) from exc + if isinstance(version.execution_snapshot, dict): snapshot = ExecutionSnapshot.model_validate(version.execution_snapshot) expected = snapshot_hash(snapshot.model_dump(mode="json")) diff --git a/tests/test_server_path_security.py b/tests/test_server_path_security.py new file mode 100644 index 0000000..135dc9f --- /dev/null +++ b/tests/test_server_path_security.py @@ -0,0 +1,279 @@ +from __future__ import annotations + +import copy +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from govoplan_campaign.backend.attachments.resolver import ( + MessageAttachmentStatus, + resolve_entry_attachments, +) +from govoplan_campaign.backend.campaign.models import CampaignConfig +from govoplan_campaign.backend.path_security import ( + CampaignPathSecurityError, + assert_server_safe_campaign_paths, +) +from govoplan_campaign.backend.persistence.campaigns import ( + create_campaign_version_from_json, + load_version_config, +) +from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError, ensure_execution_snapshot + + +def _campaign_json() -> dict[str, object]: + return { + "version": "1.0", + "campaign": {"id": "path-security", "name": "Path security", "mode": "test"}, + "template": {"subject": "Subject", "text": "Body"}, + "attachments": {"base_path": ".", "base_paths": [], "global": []}, + "entries": {"inline": []}, + } + + +class ServerCampaignPathSecurityTests(unittest.TestCase): + def test_api_import_rejects_absolute_source_filename_before_persistence(self) -> None: + with patch( + "govoplan_campaign.backend.persistence.campaigns.files_integration", + return_value=SimpleNamespace(available=False), + ): + with self.assertRaisesRegex(CampaignPathSecurityError, "source_filename"): + create_campaign_version_from_json( + None, # type: ignore[arg-type] + tenant_id="tenant-1", + user_id="user-1", + raw_json=_campaign_json(), + source_filename="/etc/passwd", + ) + + def test_template_parent_traversal_is_rejected(self) -> None: + raw = _campaign_json() + raw["template"] = {"source": {"text_path": "../../etc/passwd"}} + + with self.assertRaisesRegex(CampaignPathSecurityError, "template source paths"): + assert_server_safe_campaign_paths(raw, managed_files_available=True) + + def test_existing_unsafe_version_cannot_reuse_a_delivery_snapshot(self) -> None: + raw = _campaign_json() + raw["template"] = {"source": {"text_path": "/etc/passwd"}} + version = SimpleNamespace(raw_json=raw, execution_snapshot={}) + + with patch( + "govoplan_campaign.backend.sending.execution.files_integration", + return_value=SimpleNamespace(available=True), + ): + with self.assertRaisesRegex(ExecutionSnapshotError, "template source paths"): + ensure_execution_snapshot(None, version) # type: ignore[arg-type] + + def test_existing_unsafe_version_is_rejected_before_config_loading(self) -> None: + raw = _campaign_json() + raw["template"] = {"source": {"text_path": "/etc/passwd"}} + version = SimpleNamespace(campaign_id="campaign-1", raw_json=raw) + campaign = SimpleNamespace(id="campaign-1", tenant_id="tenant-1") + session = SimpleNamespace(get=lambda _model, key: version if key == "version-1" else campaign) + + with ( + patch( + "govoplan_campaign.backend.persistence.campaigns.files_integration", + return_value=SimpleNamespace(available=True), + ), + patch("govoplan_campaign.backend.persistence.campaigns._write_campaign_snapshot") as write_snapshot, + patch("govoplan_campaign.backend.persistence.campaigns.load_campaign_config_from_json") as load_config, + ): + with self.assertRaisesRegex(CampaignPathSecurityError, "template source paths"): + load_version_config(session, "version-1") # type: ignore[arg-type] + + write_snapshot.assert_not_called() + load_config.assert_not_called() + + def test_source_base_symlink_escape_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + outside = root / "outside" + outside.mkdir() + (outside / "secret.txt").write_text("secret", encoding="utf-8") + trusted = root / "trusted" + trusted.mkdir() + (trusted / "escape").symlink_to(outside, target_is_directory=True) + + raw = _campaign_json() + raw["template"] = {"source": {"text_path": "escape/secret.txt"}} + with self.assertRaisesRegex(CampaignPathSecurityError, "source_base_path"): + assert_server_safe_campaign_paths( + raw, + source_base_path=str(trusted), + managed_files_available=True, + ) + + def test_attachment_absolute_path_and_parent_filter_are_rejected(self) -> None: + raw = _managed_attachment_campaign() + attachments = raw["attachments"] + assert isinstance(attachments, dict) + base_paths = attachments["base_paths"] + assert isinstance(base_paths, list) + assert isinstance(base_paths[0], dict) + base_paths[0]["path"] = "/etc" + with self.assertRaisesRegex(CampaignPathSecurityError, "relative managed-file path"): + assert_server_safe_campaign_paths(raw, managed_files_available=True) + + raw = _managed_attachment_campaign() + attachments = raw["attachments"] + assert isinstance(attachments, dict) + rules = attachments["global"] + assert isinstance(rules, list) + assert isinstance(rules[0], dict) + rules[0]["file_filter"] = "../../etc/passwd" + with self.assertRaisesRegex(CampaignPathSecurityError, "parent-directory traversal"): + assert_server_safe_campaign_paths(raw, managed_files_available=True) + + def test_local_attachment_rule_is_rejected_in_server_mode(self) -> None: + raw = _managed_attachment_campaign() + attachments = raw["attachments"] + assert isinstance(attachments, dict) + base_paths = attachments["base_paths"] + assert isinstance(base_paths, list) + assert isinstance(base_paths[0], dict) + base_paths[0].pop("source") + + with self.assertRaisesRegex(CampaignPathSecurityError, "backed by managed Files"): + assert_server_safe_campaign_paths(raw, managed_files_available=True) + + def test_default_attachment_rule_must_use_managed_files(self) -> None: + raw = _managed_attachment_campaign() + attachments = raw["attachments"] + assert isinstance(attachments, dict) + attachments["global"] = [] + entries = raw["entries"] + assert isinstance(entries, dict) + entries["defaults"] = { + "attachments": [{"base_dir": "invoices", "file_filter": "invoice.pdf"}] + } + + with self.assertRaisesRegex(CampaignPathSecurityError, "entries.defaults.attachments"): + assert_server_safe_campaign_paths(raw, managed_files_available=True) + + def test_normal_managed_file_campaign_is_allowed(self) -> None: + raw = _managed_attachment_campaign() + assert_server_safe_campaign_paths(raw, managed_files_available=True) + + copied = copy.deepcopy(raw) + attachments = copied["attachments"] + assert isinstance(attachments, dict) + rules = attachments["global"] + assert isinstance(rules, list) + assert isinstance(rules[0], dict) + rules[0]["file_filter"] = "**/{{local:invoice_number}}-report.XLSX" + assert_server_safe_campaign_paths(copied, managed_files_available=True) + + def test_rendered_managed_filter_cannot_traverse_materialized_source(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + managed_root = root / "managed" + managed_root.mkdir() + (root / "secret.pdf").write_bytes(b"server secret") + + config = _runtime_managed_attachment_config( + managed_root, + file_filter="{{local:file_name}}", + file_name="../secret.pdf", + ) + resolution = resolve_entry_attachments( + config=config, + campaign_file=root / "campaign.json", + entry=config.entries.inline[0], # type: ignore[index] + entry_index=1, + ) + + self.assertEqual(resolution.status, MessageAttachmentStatus.BLOCKED) + self.assertEqual(resolution.attachments[0].matches, []) + self.assertIn( + "unsafe_managed_attachment_path", + {issue.code for issue in resolution.issues}, + ) + + def test_managed_match_symlink_cannot_escape_materialized_source(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + managed_root = root / "managed" + managed_root.mkdir() + outside = root / "secret.pdf" + outside.write_bytes(b"server secret") + (managed_root / "escape.pdf").symlink_to(outside) + + config = _runtime_managed_attachment_config( + managed_root, + file_filter="escape.pdf", + file_name="unused.pdf", + ) + resolution = resolve_entry_attachments( + config=config, + campaign_file=root / "campaign.json", + entry=config.entries.inline[0], # type: ignore[index] + entry_index=1, + ) + + self.assertEqual(resolution.status, MessageAttachmentStatus.BLOCKED) + self.assertEqual(resolution.attachments[0].matches, []) + self.assertIn( + "managed_attachment_path_escape", + {issue.code for issue in resolution.issues}, + ) + + +def _managed_attachment_campaign() -> dict[str, object]: + raw = _campaign_json() + raw["attachments"] = { + "base_path": "invoices", + "base_paths": [ + { + "id": "personal-invoices", + "name": "My invoices", + "source": "managed:user:user-1", + "path": "invoices", + } + ], + "global": [ + { + "base_path_id": "personal-invoices", + "base_dir": "invoices", + "file_filter": "invoice.pdf", + } + ], + } + return raw + + +def _runtime_managed_attachment_config( + managed_root: Path, + *, + file_filter: str, + file_name: str, +) -> CampaignConfig: + raw = _managed_attachment_campaign() + raw["fields"] = [{"name": "file_name", "type": "string", "required": True}] + attachments = raw["attachments"] + assert isinstance(attachments, dict) + base_paths = attachments["base_paths"] + assert isinstance(base_paths, list) + assert isinstance(base_paths[0], dict) + base_paths[0]["path"] = str(managed_root) + rules = attachments["global"] + assert isinstance(rules, list) + assert isinstance(rules[0], dict) + rules[0]["file_filter"] = file_filter + entries = raw["entries"] + assert isinstance(entries, dict) + entries["inline"] = [ + { + "id": "recipient-1", + "to": [{"email": "recipient@example.test", "type": "to"}], + "fields": {"file_name": file_name}, + } + ] + return CampaignConfig.model_validate(raw) + + +if __name__ == "__main__": + unittest.main()