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

fix(api): hide campaign operational internals

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:03:56 +02:00
parent 04bc6c430e
commit ce92499333
9 changed files with 593 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
tests/test_response_security.py Normal file
View File

@@ -0,0 +1,274 @@
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="<message@example.test>",
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()