from __future__ import annotations import stat import unittest from datetime import UTC, datetime from types import SimpleNamespace from govoplan_campaign.backend.response_security import public_campaign_payload from govoplan_campaign.backend.router import ( _job_attempts_payload, _job_detail_payload, _job_diagnostics_payload, ) from govoplan_campaign.backend.schemas import CampaignVersionDetailResponse, CampaignVersionResponse def _now() -> datetime: return datetime(2026, 7, 21, 12, 0, tzinfo=UTC) def _job() -> SimpleNamespace: return SimpleNamespace( id="job-1", tenant_id="tenant-1", campaign_id="campaign-1", campaign_version_id="version-1", entry_index=1, entry_id="recipient-1", recipient_email="person@example.test", subject="Subject", message_id_header="", build_status="built", validation_status="ready", queue_status="claimed", send_status="sending", imap_status="pending", eml_size_bytes=123, eml_sha256="sha256", eml_local_path="/runtime/campaign/job-1.eml", eml_storage_key="campaign/job-1.eml", claim_token="job-claim-secret", attempt_count=1, last_error=None, queued_at=_now(), claimed_at=_now(), smtp_started_at=_now(), outcome_unknown_at=None, sent_at=None, created_at=_now(), updated_at=_now(), issues_snapshot=[], resolved_attachments=[ { "filename": "public.pdf", "local_path": "/tmp/materialized/public.pdf", "storage_key": "private/blob-key", } ], resolved_recipients={"to": [{"email": "person@example.test"}]}, ) def _smtp_attempt() -> SimpleNamespace: return SimpleNamespace( id="smtp-attempt-1", attempt_number=1, status="started", claim_token="attempt-claim-secret", smtp_status_code=None, smtp_response=None, error_type=None, error_message=None, started_at=_now(), finished_at=None, ) def _imap_attempt() -> SimpleNamespace: return SimpleNamespace( id="imap-attempt-1", attempt_number=1, status="claimed", claim_token="imap-claim-secret", folder="Sent", error_message=None, created_at=_now(), updated_at=_now(), ) def test_public_payload_recursively_removes_infrastructure_locators() -> None: source = { "campaign_file": "/tmp/campaign.json", "messages": [ { "subject": "Public", "eml_path": "/tmp/message.eml", "managed": {"storage_bucket": "private", "storage_key": "object"}, } ], } result = public_campaign_payload(source) assert result == {"messages": [{"subject": "Public", "managed": {}}]} assert source["messages"][0]["eml_path"] == "/tmp/message.eml" def test_version_response_omits_source_base_path_and_sanitizes_summaries() -> None: version = SimpleNamespace( id="version-1", campaign_id="campaign-1", version_number=1, schema_version="1", source_filename="/srv/govoplan/imports/campaign.json", source_base_path="/private/import/path", workflow_state="editing", current_flow="manual", current_step=None, is_complete=False, editor_state={}, autosaved_at=None, published_at=None, locked_at=None, locked_by_user_id=None, user_lock_state=None, user_locked_at=None, user_locked_by_user_id=None, created_at=_now(), updated_at=_now(), validation_summary={"campaign_file": "/tmp/campaign.json", "ok": True}, build_summary={"messages": [{"eml_path": "/tmp/message.eml", "subject": "Public"}]}, execution_snapshot_hash=None, execution_snapshot_at=None, ) payload = CampaignVersionResponse.model_validate(version).model_dump() assert "source_base_path" not in payload assert payload["source_filename"] == "campaign.json" assert payload["validation_summary"] == {"ok": True} assert payload["build_summary"] == {"messages": [{"subject": "Public"}]} detail = CampaignVersionDetailResponse.model_validate( SimpleNamespace( **version.__dict__, raw_json={ "campaign": {"title": "Public"}, "files": [{"storage_key": "private/object", "filename": "public.pdf"}], "server": { "smtp": {"host": "smtp.example.invalid", "password": "smtp-secret"}, "imap": {"host": "imap.example.invalid", "password": "imap-secret"}, "credentials": { "smtp": {"username": "sender", "password": "legacy-smtp-secret"}, "imap": {"username": "archive", "password": "legacy-imap-secret"}, }, }, "archives": [{"password": "business-zip-password"}], "template": { "source": { "subject_path": "/srv/govoplan/templates/subject.txt", "html_path": "templates/body.html", } }, "entries": {"source": {"type": "csv", "path": "C:\\imports\\recipients.csv"}}, "attachments": { "base_path": "/srv/govoplan/attachments", "base_paths": [ {"name": "Shared", "path": "/mnt/shared/campaign", "source": "/mnt/shared"} ], "global": [{"base_dir": "/srv/govoplan/attachments/global", "file_filter": "*.pdf"}], }, }, ) ).model_dump() assert detail["raw_json"] == { "campaign": {"title": "Public"}, "files": [{"filename": "public.pdf"}], "server": { "smtp": {"host": "smtp.example.invalid"}, "imap": {"host": "imap.example.invalid"}, "credentials": { "smtp": {"username": "sender"}, "imap": {"username": "archive"}, }, }, "archives": [{"password": "business-zip-password"}], "template": { "source": { "subject_path": "subject.txt", "html_path": "templates/body.html", } }, "entries": {"source": {"type": "csv", "path": "recipients.csv"}}, "attachments": { "base_path": "attachments", "base_paths": [{"name": "Shared", "path": "campaign", "source": "shared"}], "global": [{"base_dir": "global", "file_filter": "*.pdf"}], }, } def test_ordinary_job_detail_and_attempts_do_not_expose_diagnostics() -> None: job_payload = _job_detail_payload(_job()) # type: ignore[arg-type] attempts = _job_attempts_payload([_smtp_attempt()], [_imap_attempt()]) # type: ignore[list-item] assert "eml_local_path" not in job_payload assert "eml_storage_key" not in job_payload assert job_payload["attachments"] == [{"filename": "public.pdf"}] assert "claim_token" not in attempts["smtp"][0] assert "claim_token" not in attempts["imap"][0] def test_operator_diagnostics_include_claim_and_storage_details() -> None: diagnostics = _job_diagnostics_payload( # type: ignore[arg-type, list-item] _job(), [_smtp_attempt()], [_imap_attempt()], ).model_dump() assert diagnostics["storage"]["eml_local_path"] == "/runtime/campaign/job-1.eml" assert diagnostics["storage"]["eml_storage_key"] == "campaign/job-1.eml" assert diagnostics["worker_claim"]["claim_token"] == "job-claim-secret" assert diagnostics["attempts"]["smtp"][0]["claim_token"] == "attempt-claim-secret" assert diagnostics["attempts"]["imap"][0]["claim_token"] == "imap-claim-secret" def test_diagnostics_permission_is_operator_only_by_default() -> None: from govoplan_campaign.backend.manifest import PERMISSIONS, ROLE_TEMPLATES permission_scopes = {permission.scope for permission in PERMISSIONS} role_permissions = {role.slug: set(role.permissions) for role in ROLE_TEMPLATES} assert "campaigns:diagnostic:read" in permission_scopes assert "campaigns:diagnostic:read" in role_permissions["campaign_sender"] assert "campaigns:diagnostic:read" not in role_permissions["campaign_manager"] assert "campaigns:diagnostic:read" not in role_permissions["campaign_reviewer"] def test_campaign_snapshot_with_inline_credentials_is_owner_only(tmp_path, monkeypatch) -> None: from govoplan_campaign.backend.persistence import campaigns as persistence snapshots = tmp_path / "snapshots" output = tmp_path / "generated" monkeypatch.setattr(persistence, "CAMPAIGN_SNAPSHOT_DIR", snapshots) monkeypatch.setattr(persistence, "BUILD_OUTPUT_DIR", output) path = persistence._write_campaign_snapshot( # type: ignore[arg-type] SimpleNamespace(id="version-secret", raw_json={"server": {"smtp": {"password": "secret"}}}) ) assert stat.S_IMODE(path.stat().st_mode) == 0o600 assert stat.S_IMODE(snapshots.stat().st_mode) == 0o700 assert stat.S_IMODE(output.stat().st_mode) == 0o700 class ResponseSecurityTests(unittest.TestCase): def test_public_payload(self) -> None: test_public_payload_recursively_removes_infrastructure_locators() def test_version_response(self) -> None: test_version_response_omits_source_base_path_and_sanitizes_summaries() def test_ordinary_job_response(self) -> None: test_ordinary_job_detail_and_attempts_do_not_expose_diagnostics() def test_operator_diagnostics(self) -> None: test_operator_diagnostics_include_claim_and_storage_details() def test_operator_permission(self) -> None: test_diagnostics_permission_is_operator_only_by_default() if __name__ == "__main__": unittest.main()