280 lines
11 KiB
Python
280 lines
11 KiB
Python
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()
|