293 lines
11 KiB
Python
293 lines
11 KiB
Python
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="smtp.internal.example /srv/private provider-secret",
|
|
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="smtp.internal.example provider-secret",
|
|
error_type="/srv/private/ProviderError",
|
|
error_message="credential=provider-secret",
|
|
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="imap.internal.example /srv/private provider-secret",
|
|
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={
|
|
"build_token": "internal-build-token",
|
|
"smtp_transport_revision": "internal-smtp-revision",
|
|
"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"}]}
|
|
|
|
operator_payload = CampaignVersionResponse.model_validate(
|
|
version,
|
|
context={"include_diagnostics": True},
|
|
).model_dump()
|
|
assert operator_payload["build_summary"]["build_token"] == "internal-build-token"
|
|
assert operator_payload["build_summary"]["smtp_transport_revision"] == "internal-smtp-revision"
|
|
|
|
detail = CampaignVersionDetailResponse.model_validate(
|
|
SimpleNamespace(
|
|
**version.__dict__,
|
|
raw_json={
|
|
"campaign": {"title": "Public"},
|
|
"files": [{"storage_key": "private/object", "filename": "public.pdf"}],
|
|
"server": {
|
|
"mail_profile_id": "profile-1",
|
|
"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": {"mail_profile_id": "profile-1"},
|
|
"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 "claimed_at" not in job_payload
|
|
assert "smtp_started_at" 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]
|
|
assert "smtp_response" not in attempts["smtp"][0]
|
|
assert "error_type" not in attempts["smtp"][0]
|
|
assert "error_message" not in attempts["smtp"][0]
|
|
assert "error_message" not in attempts["imap"][0]
|
|
ordinary = repr({"job": job_payload, "attempts": attempts})
|
|
assert "internal.example" not in ordinary
|
|
assert "/srv/private" not in ordinary
|
|
assert "provider-secret" not in ordinary
|
|
|
|
|
|
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"
|
|
assert "smtp.internal.example" in diagnostics["attempts"]["smtp"][0]["smtp_response"]
|
|
assert "imap.internal.example" in diagnostics["attempts"]["imap"][0]["error_message"]
|
|
assert "provider-secret" in diagnostics["worker_claim"]["last_error"]
|
|
|
|
|
|
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()
|