Fix Campaign navigation action markup

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.

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

src/govoplan_campaign/backend/campaign/models.py

@@ -31,7 +31,7 @@ class FieldType(StrEnum):
     INTEGER = "integer"
     DOUBLE = "double"
     DATE = "date"
-    PASSWORD = "password"
+    PASSWORD = "password"  # noqa: S105  # nosec B105 - field type vocabulary.
 
 
 class RecipientType(StrEnum):

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,

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

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)
 
 

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

src/govoplan_campaign/backend/reports/campaigns.py

@@ -81,7 +81,7 @@ def _load_delivery_info(version: CampaignVersion | None, jobs: list[CampaignJob]
         "execution_snapshot_hash": None,
         "execution_snapshot_at": None,
         "snapshot_version": None,
-        "build_token": None,
+        "build_token": None,  # nosec B105 - absent report value, not a credential.
         "built_at": None,
         "job_manifest_sha256": None,
         "job_count": 0,

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,6 +3358,7 @@ 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)
+    try:
         return _attachment_preview_for_version(
             session,
             principal,
@@ -3361,6 +3368,8 @@ def preview_campaign_attachments(
             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)

src/govoplan_campaign/backend/schemas.py

@@ -597,6 +597,7 @@ class AppendSentRequest(BaseModel):
 class CampaignActionResponse(BaseModel):
     result: dict[str, Any]
 
+
 class ReportEmailRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -610,111 +611,3 @@ class ReportEmailRequest(BaseModel):
 
 class ReportEmailResponse(BaseModel):
     result: dict[str, Any]
-
-
-class AuditLogItemResponse(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-
-    id: str
-    tenant_id: str | None = None
-    user_id: str | None = None
-    api_key_id: str | None = None
-    action: str
-    object_type: str | None = None
-    object_id: str | None = None
-    details: dict[str, Any] | None = None
-    created_at: datetime
-
-
-class AuditLogListResponse(BaseModel):
-    items: list[AuditLogItemResponse]
-
-
-
-class LoginRequest(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    email: str
-    password: str
-    # Kept optional for backwards compatibility and future tenant-switch login flows.
-    # The WebUI no longer sends it. If omitted, the backend resolves the user by email.
-    tenant_slug: str | None = None
-
-
-
-
-class SwitchTenantRequest(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    tenant_id: str
-
-
-class TenantInfo(BaseModel):
-    id: str
-    slug: str
-    name: str
-    is_active: bool = True
-    default_locale: str = "en"
-
-
-class TenantMembershipInfo(TenantInfo):
-    roles: list[str] = Field(default_factory=list)
-    is_active: bool = True
-
-
-class UserInfo(BaseModel):
-    id: str
-    account_id: str
-    email: str
-    # Global account identity used by the title bar and account settings.
-    display_name: str | None = None
-    # Optional tenant-local alias used in tenant administration and ownership.
-    tenant_display_name: str | None = None
-    is_tenant_admin: bool = False
-    password_reset_required: bool = False
-
-
-class ProfileUpdateRequest(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    display_name: str | None = Field(default=None, max_length=255)
-    tenant_display_name: str | None = Field(default=None, max_length=255)
-
-
-class RoleInfo(BaseModel):
-    id: str
-    slug: str
-    name: str
-    permissions: list[str] = Field(default_factory=list)
-    level: Literal["tenant", "system"] = "tenant"
-
-
-class GroupInfo(BaseModel):
-    id: str
-    slug: str
-    name: str
-
-
-class LoginResponse(BaseModel):
-    access_token: str
-    token_type: str = "bearer"
-    expires_at: datetime
-    user: UserInfo
-    # Backwards-compatible alias for the active tenant.
-    tenant: TenantInfo
-    active_tenant: TenantInfo
-    tenants: list[TenantMembershipInfo] = Field(default_factory=list)
-    scopes: list[str]
-    roles: list[RoleInfo] = Field(default_factory=list)
-    groups: list[GroupInfo] = Field(default_factory=list)
-
-
-class MeResponse(BaseModel):
-    user: UserInfo
-    # Backwards-compatible alias for the active tenant.
-    tenant: TenantInfo
-    active_tenant: TenantInfo
-    tenants: list[TenantMembershipInfo] = Field(default_factory=list)
-    scopes: list[str]
-    roles: list[RoleInfo] = Field(default_factory=list)
-    groups: list[GroupInfo] = Field(default_factory=list)

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

src/govoplan_campaign/backend/services/zip_service.py

@@ -41,7 +41,7 @@ def _normalized_members(files: Iterable[Path | ArchiveMember]) -> list[ArchiveMe
 def create_zip_archive(
     output_path: Path,
     files: Iterable[Path | ArchiveMember],
-    password: str = "",
+    password: str = "",  # nosec B107 - empty means an unencrypted archive.
     method: str = ZIP_METHOD_AES,
 ) -> Path:
     """Create a ZIP archive, optionally using AES or legacy ZipCrypto encryption."""

tests/test_access_provider.py

@@ -21,6 +21,7 @@ GROUP_ID = "group-1"
 class CampaignAccessProviderTests(unittest.TestCase):
     def test_campaign_access_provider_explains_owner_share_admin_and_missing_resources(self) -> None:
         session = _session()
+        self.addCleanup(_close_session, session)
         _seed_access_subjects(session)
         owned = Campaign(id="campaign-owned", tenant_id=TENANT_ID, owner_user_id=USER_ID, external_id="owned", name="Owned campaign")
         shared = Campaign(id="campaign-shared", tenant_id=TENANT_ID, owner_user_id=OTHER_USER_ID, external_id="shared", name="Shared campaign")
@@ -46,6 +47,7 @@ class CampaignAccessProviderTests(unittest.TestCase):
 
     def test_campaign_access_provider_requires_write_share_for_write_actions(self) -> None:
         session = _session()
+        self.addCleanup(_close_session, session)
         _seed_access_subjects(session)
         campaign = Campaign(id="campaign-write", tenant_id=TENANT_ID, owner_user_id=OTHER_USER_ID, external_id="write", name="Write campaign")
         session.add_all([
@@ -74,6 +76,12 @@ def _session():
     return sessionmaker(bind=engine, future=True)()
 
 
+def _close_session(session) -> None:
+    engine = session.get_bind()
+    session.close()
+    engine.dispose()
+
+
 def _seed_access_subjects(session) -> None:
     session.add_all([
         Account(id="account-1", email="one@example.test", normalized_email="one@example.test"),

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

webui/src/features/campaigns/CampaignOverviewPage.tsx

@@ -171,7 +171,7 @@ export default function CampaignOverviewPage({ settings, campaignId }: {settings
         <Card
           title="i18n:govoplan-campaign.campaign_identity.a00ca574"
           collapsible
-          actions={<Link to="wizard/create"><Button>i18n:govoplan-campaign.edit_with_wizard.672a7d1a</Button></Link>}>
+          actions={<Link className="btn btn-secondary" to="wizard/create">i18n:govoplan-campaign.edit_with_wizard.672a7d1a</Link>}>
           <div className="form-grid campaign-identity-grid">
             <FormField label="i18n:govoplan-campaign.campaign_id.4c4ed79e">
               <input value={identity.external_id} onChange={(event) => patchIdentity("external_id", event.target.value)} />

webui/src/features/campaigns/components/MessagePreviewOverlay.tsx

@@ -1,6 +1,6 @@
-import { useEffect, type ReactNode } from "react";
+import { useEffect, useRef, type ReactNode } from "react";
 import { ArrowBigLeft, ArrowBigLeftDash, ArrowBigRight, ArrowBigRightDash } from "lucide-react";
-import { Button, MessageDisplayPanel, type MessageDisplayAttachment } from "@govoplan/core-webui";
+import { Button, Dialog, MessageDisplayPanel, type MessageDisplayAttachment } from "@govoplan/core-webui";
 
 // Campaign review/template/mock previews need recipient navigation, review notes,
 // raw MIME inspection and attachment grouping. Generic mailbox reading uses
@@ -60,15 +60,13 @@ export default function CampaignMessagePreviewOverlay({
 }: CampaignMessagePreviewOverlayProps) {
   const shownSubject = subject?.trim() || "i18n:govoplan-campaign.no_subject.7b4e8035";
   const fields = metaItems.map((item) => ({ label: item.label, value: item.value }));
+  const contentRef = useRef<HTMLDivElement | null>(null);
 
   useEffect(() => {
     const handleKeyDown = (event: KeyboardEvent) => {
+      const dialogPanel = contentRef.current?.closest<HTMLElement>("[data-dialog-stack-state]");
+      if (dialogPanel?.dataset.dialogStackState !== "topmost") return;
       if (isEditableTarget(event.target)) return;
-      if (event.key === "Escape") {
-        event.preventDefault();
-        onClose();
-        return;
-      }
       if (!navigation) return;
       if (event.key === "ArrowLeft") {
         event.preventDefault();
@@ -87,16 +85,26 @@ export default function CampaignMessagePreviewOverlay({
 
     window.addEventListener("keydown", handleKeyDown);
     return () => window.removeEventListener("keydown", handleKeyDown);
-  }, [navigation, onClose]);
+  }, [navigation]);
 
   return (
-    <div className="overlay-backdrop message-preview-backdrop" role="dialog" aria-modal="true" aria-labelledby="message-preview-title">
-      <div className="modal-panel template-preview-modal message-preview-modal">
-        <header className="modal-header">
-          <h2 id="message-preview-title">{title}</h2>
-          <button type="button" className="modal-close" aria-label={closeLabel} title={closeLabel} onClick={onClose}>×</button>
-        </header>
-        <div className="modal-body">
+    <Dialog
+      open
+      title={title}
+      onClose={onClose}
+      closeLabel={closeLabel}
+      closeOnBackdrop={false}
+      backdropClassName="overlay-backdrop message-preview-backdrop"
+      className="modal-panel template-preview-modal message-preview-modal"
+      headerClassName="modal-header"
+      bodyClassName="modal-body"
+      footerClassName="modal-footer"
+      footer={<>
+        {actions && <div className="button-row compact-actions">{actions}</div>}
+        <Button variant="primary" onClick={onClose}>{closeLabel}</Button>
+      </>}
+    >
+      <div ref={contentRef} className="message-preview-content">
           {(recipientLabel || recipientNote || navigation) &&
           <div className="template-preview-toolbar">
               <div>
@@ -133,12 +141,7 @@ export default function CampaignMessagePreviewOverlay({
             </details>
           }
       </div>
-        <footer className="modal-footer">
-          {actions && <div className="button-row compact-actions">{actions}</div>}
-          <Button variant="primary" onClick={onClose}>{closeLabel}</Button>
-        </footer>
-      </div>
-    </div>);
+    </Dialog>);
 
 }
 

webui/src/features/campaigns/wizard/CreateWizard.tsx

@@ -98,7 +98,7 @@ export default function CreateWizard({ settings, campaignId }: {settings: ApiSet
                 message="i18n:govoplan-campaign.this_wizard_is_read_only_for_the_selected_versio.b0865947" />
               
               <div className="button-row">
-                <Link to="../.."><Button>i18n:govoplan-campaign.back_to_overview.ec986cba</Button></Link>
+                <Link className="btn btn-secondary" to="../..">i18n:govoplan-campaign.back_to_overview.ec986cba</Link>
               </div>
             </Card>
           </div>

webui/src/styles/campaign-workspace.css

@@ -1335,6 +1335,10 @@
   scrollbar-gutter: stable;
 }
 
+.message-preview-content {
+  display: contents;
+}
+
 .message-preview-modal .modal-footer {
   align-items: center;
   min-height: 68px;

webui/tests/message-preview-overlay-structure.test.mjs

@@ -15,8 +15,10 @@ assert(
   "message previews expose their responsive backdrop hook"
 );
 assert(
-  overlaySource.includes('type="button" className="modal-close" aria-label={closeLabel} title={closeLabel}'),
-  "the close control has stable button semantics, an accessible name, and a tooltip"
+  overlaySource.includes("<Dialog") &&
+    overlaySource.includes("closeLabel={closeLabel}") &&
+    overlaySource.includes('dataset.dialogStackState !== "topmost"'),
+  "message previews use the central stacked Dialog with an accessible close label"
 );
 assert(
   styles.includes("height: min(780px, calc(100dvh - 48px));"),

webui/tests/review-preview-ui.test.ts

@@ -103,7 +103,7 @@ for (const [key, english, german] of newTranslations) {
 const overlaySource = readFileSync("src/features/campaigns/components/MessagePreviewOverlay.tsx", "utf8");
 assert(overlaySource.includes("message-preview-backdrop"), "message previews expose a layout-specific responsive backdrop hook");
 assert(overlaySource.includes("{actions && <div"), "built-message footer actions remain in the stable shared preview footer");
-assert(overlaySource.includes('aria-label={closeLabel}'), "the preview close control has an accessible label");
+assert(overlaySource.includes("<Dialog") && overlaySource.includes("closeLabel={closeLabel}"), "the preview uses the central Dialog and its accessible close label");
 
 const styles = readFileSync("src/styles/campaign-workspace.css", "utf8");
 assert(styles.includes("height: min(780px, calc(100dvh - 48px));"), "desktop previews use a stable responsive height");