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

security: separate Campaign reader and operator data

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 17:55:54 +02:00
parent af833ca38c
commit 9a709b1264
6 changed files with 128 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
src/govoplan_campaign/backend/reports/campaigns.py
View File

@@ -64,7 +64,11 @@ def _selected_version(
return version
def _version_info(version: CampaignVersion | None) -> dict[str, Any] | None:
def _version_info(
version: CampaignVersion | None,
*,
include_diagnostics: bool = False,
) -> dict[str, Any] | None:
if not version:
return None
return {
@@ -73,14 +77,25 @@ def _version_info(version: CampaignVersion | None) -> dict[str, Any] | None:
"schema_version": version.schema_version,
"source_filename": public_source_filename(version.source_filename),
"created_at": version.created_at.isoformat() if version.created_at else None,
"validation_summary": public_campaign_payload(version.validation_summary),
"build_summary": public_campaign_payload(version.build_summary),
"validation_summary": public_campaign_payload(
version.validation_summary,
include_diagnostics=include_diagnostics,
),
"build_summary": public_campaign_payload(
version.build_summary,
include_diagnostics=include_diagnostics,
),
"execution_snapshot_hash": version.execution_snapshot_hash,
"execution_snapshot_at": version.execution_snapshot_at.isoformat() if version.execution_snapshot_at else None,
}
def _load_delivery_info(version: CampaignVersion | None, jobs: list[CampaignJob]) -> dict[str, Any]:
def _load_delivery_info(
version: CampaignVersion | None,
jobs: list[CampaignJob],
*,
include_diagnostics: bool = False,
) -> dict[str, Any]:
"""Read deterministic delivery settings from the immutable execution snapshot."""
default = {
@@ -90,18 +105,23 @@ def _load_delivery_info(version: CampaignVersion | None, jobs: list[CampaignJob]
"execution_snapshot_hash": None,
"execution_snapshot_at": None,
"snapshot_version": None,
"build_token": None, # nosec B105 - absent report value, not a credential.
"built_at": None,
"job_manifest_sha256": None,
"job_count": 0,
"queueable_job_count": 0,
"effective_policy_sha256": None,
"smtp_transport_revision": None,
"imap_transport_revision": None,
"estimated_remaining_send_seconds": None,
"estimated_remaining_send_human": None,
"background_workers_enabled": bool(core_settings.celery_enabled),
}
if include_diagnostics:
default.update(
{
"build_token": None, # nosec B105 - absent report value, not a credential.
"job_manifest_sha256": None,
"effective_policy_sha256": None,
"smtp_transport_revision": None,
"imap_transport_revision": None,
"background_workers_enabled": bool(core_settings.celery_enabled),
}
)
if not version or not isinstance(version.execution_snapshot, dict):
default["load_error"] = "No execution snapshot exists; rebuild the current locked version before delivery."
return default
@@ -118,7 +138,7 @@ def _load_delivery_info(version: CampaignVersion | None, jobs: list[CampaignJob]
if messages_per_minute and pending:
estimated_seconds = int(math.ceil((len(pending) / messages_per_minute) * 60))
return {
result = {
"rate_limit": {
"messages_per_minute": messages_per_minute,
"concurrency": snapshot.delivery.rate_limit.concurrency,
@@ -134,18 +154,24 @@ def _load_delivery_info(version: CampaignVersion | None, jobs: list[CampaignJob]
"execution_snapshot_hash": version.execution_snapshot_hash,
"execution_snapshot_at": version.execution_snapshot_at.isoformat() if version.execution_snapshot_at else None,
"snapshot_version": snapshot.snapshot_version,
"build_token": snapshot.build_token,
"built_at": snapshot.built_at,
"job_manifest_sha256": snapshot.job_manifest_sha256,
"job_count": snapshot.job_count,
"queueable_job_count": snapshot.queueable_job_count,
"effective_policy_sha256": snapshot.effective_policy_sha256,
"smtp_transport_revision": snapshot.smtp_transport_revision,
"imap_transport_revision": snapshot.imap_transport_revision,
"estimated_remaining_send_seconds": estimated_seconds,
"estimated_remaining_send_human": _human_duration(estimated_seconds),
"background_workers_enabled": bool(core_settings.celery_enabled),
}
if include_diagnostics:
result.update(
{
"build_token": snapshot.build_token,
"job_manifest_sha256": snapshot.job_manifest_sha256,
"effective_policy_sha256": snapshot.effective_policy_sha256,
"smtp_transport_revision": snapshot.smtp_transport_revision,
"imap_transport_revision": snapshot.imap_transport_revision,
"background_workers_enabled": bool(core_settings.celery_enabled),
}
)
return result
def _human_duration(seconds: int | None) -> str | None:
@@ -243,8 +269,8 @@ def _recent_failures(jobs: list[CampaignJob], *, limit: int = 20) -> list[dict[s
]
def _job_row(job: CampaignJob) -> dict[str, Any]:
return {
def _job_row(job: CampaignJob, *, include_diagnostics: bool = False) -> dict[str, Any]:
row = {
"job_id": job.id,
"entry_index": job.entry_index,
"entry_id": job.entry_id,
@@ -257,8 +283,6 @@ def _job_row(job: CampaignJob) -> dict[str, Any]:
"imap_status": job.imap_status,
"attempt_count": job.attempt_count,
"queued_at": job.queued_at.isoformat() if job.queued_at else None,
"claimed_at": job.claimed_at.isoformat() if job.claimed_at else None,
"smtp_started_at": job.smtp_started_at.isoformat() if job.smtp_started_at else None,
"outcome_unknown_at": job.outcome_unknown_at.isoformat() if job.outcome_unknown_at else None,
"sent_at": job.sent_at.isoformat() if job.sent_at else None,
"last_error": public_delivery_result_message(
@@ -272,6 +296,14 @@ def _job_row(job: CampaignJob) -> dict[str, Any]:
"attachment_config_count": len(job.resolved_attachments or []),
"matched_file_count": sum(len(item.get("matches") or []) for item in (job.resolved_attachments or []) if isinstance(item, dict)),
}
if include_diagnostics:
row.update(
{
"claimed_at": job.claimed_at.isoformat() if job.claimed_at else None,
"smtp_started_at": job.smtp_started_at.isoformat() if job.smtp_started_at else None,
}
)
return row
def _address_summary(value: Any) -> str:
@@ -334,8 +366,9 @@ def _job_evidence_row(
*,
latest_smtp: SendAttempt | None = None,
latest_imap: ImapAppendAttempt | None = None,
include_diagnostics: bool = False,
) -> dict[str, Any]:
row = _job_row(job)
row = _job_row(job, include_diagnostics=include_diagnostics)
recipients = job.resolved_recipients or {}
row.update({
"campaign_id": job.campaign_id,
@@ -369,6 +402,7 @@ def generate_campaign_report(
version_id: str | None = None,
include_jobs: bool = False,
include_recent_failures: bool = False,
include_diagnostics: bool = False,
) -> dict[str, Any]:
"""Generate a dashboard/report payload for one campaign.
@@ -387,9 +421,13 @@ def generate_campaign_report(
jobs=jobs,
tenant_id=tenant_id,
include_recent_failures=include_recent_failures,
include_diagnostics=include_diagnostics,
)
if include_jobs:
report["jobs"] = [_job_row(job) for job in jobs]
report["jobs"] = [
_job_row(job, include_diagnostics=include_diagnostics)
for job in jobs
]
return report
@@ -419,12 +457,13 @@ def _campaign_report_payload(
jobs: list[CampaignJob],
tenant_id: str,
include_recent_failures: bool,
include_diagnostics: bool,
) -> dict[str, Any]:
job_ids = [job.id for job in jobs]
report = {
"generated_at": _utcnow_iso(),
"campaign": _campaign_report_campaign_payload(campaign),
"current_version": _version_info(version),
"current_version": _version_info(version, include_diagnostics=include_diagnostics),
"selected_version_id": version.id if version else None,
"cards": _campaign_report_cards(version, jobs),
"status_counts": _campaign_report_status_counts(version, jobs),
@@ -439,7 +478,11 @@ def _campaign_report_payload(
},
"attachments": _attachment_summary(jobs),
"attempts": _campaign_report_attempt_counts(session, job_ids),
"delivery": _load_delivery_info(version, jobs),
"delivery": _load_delivery_info(
version,
jobs,
include_diagnostics=include_diagnostics,
),
}
if include_recent_failures:
report["recent_failures"] = _recent_failures(jobs)
@@ -544,6 +587,7 @@ def generate_jobs_csv(
tenant_id: str,
campaign_id: str,
version_id: str | None = None,
include_diagnostics: bool = False,
) -> str:
campaign = _get_campaign(session, tenant_id=tenant_id, campaign_id=campaign_id)
version = _selected_version(session, campaign, version_id)
@@ -584,6 +628,7 @@ def generate_jobs_csv(
job,
latest_smtp=latest_smtp.get(job.id),
latest_imap=latest_imap.get(job.id),
include_diagnostics=include_diagnostics,
)
for job in jobs
]
@@ -608,8 +653,6 @@ def generate_jobs_csv(
"imap_status",
"attempt_count",
"queued_at",
"claimed_at",
"smtp_started_at",
"outcome_unknown_at",
"sent_at",
"last_error",
@@ -630,6 +673,9 @@ def generate_jobs_csv(
"latest_imap_created_at",
"latest_imap_updated_at",
]
if include_diagnostics:
sent_at_index = fieldnames.index("outcome_unknown_at")
fieldnames[sent_at_index:sent_at_index] = ["claimed_at", "smtp_started_at"]
buffer = io.StringIO()
writer = csv.DictWriter(buffer, fieldnames=fieldnames)
writer.writeheader()

29
src/govoplan_campaign/backend/response_security.py
View File

@@ -24,20 +24,39 @@ CAMPAIGN_INTERNAL_RESPONSE_KEYS = frozenset(
}
)
CAMPAIGN_DIAGNOSTIC_RESPONSE_KEYS = frozenset(
{
"background_workers_enabled",
"build_token",
"celery_enabled",
"claimed_at",
"effective_policy_sha256",
"execution_input_sha256",
"imap_claimed_at",
"imap_transport_revision",
"job_manifest_sha256",
"smtp_started_at",
"smtp_transport_revision",
}
)
def public_campaign_payload(value: Any) -> Any:
def public_campaign_payload(value: Any, *, include_diagnostics: bool = False) -> Any:
"""Return a detached payload without infrastructure-only locators."""
if isinstance(value, dict):
blocked_keys = CAMPAIGN_INTERNAL_RESPONSE_KEYS
if not include_diagnostics:
blocked_keys = blocked_keys | CAMPAIGN_DIAGNOSTIC_RESPONSE_KEYS
return {
key: public_campaign_payload(item)
key: public_campaign_payload(item, include_diagnostics=include_diagnostics)
for key, item in value.items()
if key not in CAMPAIGN_INTERNAL_RESPONSE_KEYS
if key not in blocked_keys
}
if isinstance(value, list):
return [public_campaign_payload(item) for item in value]
return [public_campaign_payload(item, include_diagnostics=include_diagnostics) for item in value]
if isinstance(value, tuple):
return tuple(public_campaign_payload(item) for item in value)
return tuple(public_campaign_payload(item, include_diagnostics=include_diagnostics) for item in value)
return copy.deepcopy(value)

7
src/govoplan_campaign/backend/schemas.py
View File

@@ -128,8 +128,11 @@ class CampaignVersionResponse(BaseModel):
@field_validator("validation_summary", "build_summary", mode="before")
@classmethod
def remove_internal_summary_fields(cls, value: Any) -> Any:
return public_campaign_payload(value)
def remove_internal_summary_fields(cls, value: Any, info: ValidationInfo) -> Any:
return public_campaign_payload(
value,
include_diagnostics=bool((info.context or {}).get("include_diagnostics")),
)
class CampaignVersionDetailResponse(CampaignVersionResponse):