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