GovOPlaN/govoplan-campaign
1
0
Fork 0
Code Issues 46 Pull Requests Actions Packages Projects Releases Wiki Activity

Block local file paths at Campaign server boundaries

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 03:17:11 +02:00
parent 78f52a36d4
commit 1d291377c0
9 changed files with 576 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@@ -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.

63
src/govoplan_campaign/backend/attachments/resolver.py
View File

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

5
src/govoplan_campaign/backend/dev/mock_campaign.py
View File

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

170
src/govoplan_campaign/backend/path_security.py Normal file
View File

@@ -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,
)

18
src/govoplan_campaign/backend/persistence/campaigns.py
View File

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

20
src/govoplan_campaign/backend/persistence/versions.py
View File

@@ -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")

27
src/govoplan_campaign/backend/router.py
View File

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

11
src/govoplan_campaign/backend/sending/execution.py
View File

@@ -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"))

279
tests/test_server_path_security.py Normal file
View File

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