378 lines
16 KiB
Python
378 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Iterable
|
|
|
|
from pydantic import BaseModel, ConfigDict
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion, JobValidationStatus
|
|
from govoplan_campaign.backend.campaign.models import DeliveryConfig
|
|
from govoplan_campaign.backend.campaign.mail_profile_boundary import (
|
|
CampaignMailProfileBoundaryError,
|
|
assert_campaign_uses_mail_profile_reference,
|
|
campaign_mail_profile_id,
|
|
campaign_mail_resource_ids,
|
|
)
|
|
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 = "6"
|
|
|
|
|
|
class ExecutionSnapshotError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class ExecutionSnapshot(BaseModel):
|
|
"""Immutable delivery inputs for one built campaign version.
|
|
|
|
Rendered messages and attachment evidence remain normalized in
|
|
``CampaignJob`` and ``CampaignAttachmentUse``. This record freezes the
|
|
Mail-profile reference, delivery policy, and opaque transport revisions and
|
|
cryptographically binds them to the build and exact persisted jobs. Mail
|
|
owns the resolved transport configuration and credentials.
|
|
"""
|
|
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
snapshot_version: str = SNAPSHOT_VERSION
|
|
campaign_version_id: str
|
|
campaign_json_sha256: str
|
|
mail_profile_id: str
|
|
smtp_server_id: str | None = None
|
|
smtp_credential_id: str | None = None
|
|
imap_server_id: str | None = None
|
|
imap_credential_id: str | None = None
|
|
created_at: str
|
|
build_token: str | None = None
|
|
built_at: str | None = None
|
|
job_count: int = 0
|
|
queueable_job_count: int = 0
|
|
job_manifest_sha256: str | None = None
|
|
effective_policy_sha256: str | None = None
|
|
smtp_transport_revision: str | None = None
|
|
imap_transport_revision: str | None = None
|
|
delivery: DeliveryConfig
|
|
|
|
|
|
def _canonical_json(value: Any) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
|
|
|
|
|
|
def _sha256(value: Any) -> str:
|
|
return hashlib.sha256(_canonical_json(value)).hexdigest()
|
|
|
|
|
|
def snapshot_hash(payload: dict[str, Any]) -> str:
|
|
return _sha256(payload)
|
|
|
|
|
|
def profile_delivery_summary(session: Session, version: CampaignVersion) -> dict[str, Any]:
|
|
raw_json = version.raw_json if isinstance(version.raw_json, dict) else {}
|
|
_assert_version_mail_profile_boundary(raw_json)
|
|
mail = mail_integration()
|
|
profile_id = campaign_mail_profile_id(raw_json)
|
|
if profile_id is None: # Kept explicit for static typing; the assertion above requires it.
|
|
raise ExecutionSnapshotError("Campaign has no Mail profile reference")
|
|
campaign = session.get(Campaign, version.campaign_id)
|
|
if campaign is None:
|
|
raise ExecutionSnapshotError("Campaign not found for mail-server profile resolution")
|
|
references = campaign_mail_resource_ids(raw_json)
|
|
try:
|
|
return mail.campaign_profile_delivery_summary(
|
|
session,
|
|
tenant_id=campaign.tenant_id,
|
|
campaign_id=campaign.id,
|
|
profile_id=profile_id,
|
|
smtp_server_id=references["smtp_server_id"],
|
|
smtp_credential_id=references["smtp_credential_id"],
|
|
imap_server_id=references["imap_server_id"],
|
|
imap_credential_id=references["imap_credential_id"],
|
|
)
|
|
except MailProfileError as exc:
|
|
raise ExecutionSnapshotError(str(exc)) from exc
|
|
|
|
|
|
def profile_transport_revisions(session: Session, version: CampaignVersion) -> dict[str, str | None]:
|
|
summary = profile_delivery_summary(session, version)
|
|
return {
|
|
"smtp": summary.get("smtp_transport_revision"),
|
|
"imap": summary.get("imap_transport_revision"),
|
|
}
|
|
|
|
|
|
def _assert_snapshot_profile_matches_version(version: CampaignVersion, snapshot: ExecutionSnapshot) -> None:
|
|
raw_json = version.raw_json if isinstance(version.raw_json, dict) else {}
|
|
_assert_version_mail_profile_boundary(raw_json)
|
|
if campaign_mail_profile_id(raw_json) != snapshot.mail_profile_id:
|
|
raise ExecutionSnapshotError(
|
|
"The campaign's Mail profile reference differs from the built execution snapshot. "
|
|
"Revalidate and rebuild the campaign before delivery."
|
|
)
|
|
references = campaign_mail_resource_ids(raw_json)
|
|
for key in (
|
|
"smtp_server_id",
|
|
"smtp_credential_id",
|
|
"imap_server_id",
|
|
"imap_credential_id",
|
|
):
|
|
configured = references[key]
|
|
if configured and configured != getattr(snapshot, key):
|
|
raise ExecutionSnapshotError(
|
|
"The campaign's Mail server or credential selection differs from the built "
|
|
"execution snapshot. Revalidate and rebuild before delivery."
|
|
)
|
|
|
|
|
|
def _assert_version_mail_profile_boundary(raw_json: dict[str, Any]) -> None:
|
|
try:
|
|
assert_campaign_uses_mail_profile_reference(raw_json, require_profile=True)
|
|
except CampaignMailProfileBoundaryError as exc:
|
|
raise ExecutionSnapshotError(str(exc)) from exc
|
|
|
|
|
|
def _policy_fingerprint(raw_json: dict[str, Any], delivery: DeliveryConfig) -> str:
|
|
return _sha256(
|
|
{
|
|
"validation_policy": raw_json.get("validation_policy"),
|
|
"policy": raw_json.get("policy"),
|
|
"delivery": delivery.model_dump(mode="json"),
|
|
"attachment_defaults": (raw_json.get("attachments") or {}).get("defaults")
|
|
if isinstance(raw_json.get("attachments"), dict)
|
|
else None,
|
|
}
|
|
)
|
|
|
|
|
|
def _job_execution_input_payload(job: CampaignJob) -> dict[str, Any]:
|
|
return {
|
|
"job_id": job.id,
|
|
"entry_index": job.entry_index,
|
|
"entry_id": job.entry_id,
|
|
"recipient_email": job.recipient_email,
|
|
"subject": job.subject,
|
|
"message_id_header": job.message_id_header,
|
|
"eml_size_bytes": job.eml_size_bytes,
|
|
"eml_sha256": job.eml_sha256,
|
|
"build_status": job.build_status,
|
|
"validation_status": job.validation_status,
|
|
"resolved_recipients_sha256": _sha256(job.resolved_recipients or {}),
|
|
"resolved_attachments_sha256": _sha256(job.resolved_attachments or []),
|
|
"issues_sha256": _sha256(job.issues_snapshot or []),
|
|
}
|
|
|
|
|
|
def job_execution_input_hash(job: CampaignJob) -> str:
|
|
return _sha256(_job_execution_input_payload(job))
|
|
|
|
|
|
def job_manifest_hash(jobs: Iterable[CampaignJob]) -> str:
|
|
"""Hash the immutable per-message execution records in stable order."""
|
|
|
|
payload = [
|
|
_job_execution_input_payload(job)
|
|
for job in sorted(jobs, key=lambda item: (item.entry_index, item.id))
|
|
]
|
|
return _sha256(payload)
|
|
|
|
|
|
def create_execution_snapshot(
|
|
version: CampaignVersion,
|
|
*,
|
|
mail_profile_id: str,
|
|
smtp_transport_revision: str,
|
|
imap_transport_revision: str | None,
|
|
delivery: DeliveryConfig,
|
|
smtp_server_id: str | None = None,
|
|
smtp_credential_id: str | None = None,
|
|
imap_server_id: str | None = None,
|
|
imap_credential_id: str | None = None,
|
|
jobs: Iterable[CampaignJob] = (),
|
|
build_summary: dict[str, Any] | None = None,
|
|
) -> tuple[dict[str, Any], str]:
|
|
raw_json = version.raw_json if isinstance(version.raw_json, dict) else {}
|
|
job_list = list(jobs)
|
|
for job in job_list:
|
|
job.execution_input_sha256 = job_execution_input_hash(job)
|
|
summary = build_summary if isinstance(build_summary, dict) else {}
|
|
queueable_statuses = {JobValidationStatus.READY.value, JobValidationStatus.WARNING.value}
|
|
payload = ExecutionSnapshot(
|
|
campaign_version_id=version.id,
|
|
campaign_json_sha256=_sha256(raw_json),
|
|
mail_profile_id=mail_profile_id,
|
|
smtp_server_id=smtp_server_id,
|
|
smtp_credential_id=smtp_credential_id,
|
|
imap_server_id=imap_server_id,
|
|
imap_credential_id=imap_credential_id,
|
|
build_token=str(summary.get("build_token") or "") or None,
|
|
built_at=str(summary.get("built_at") or "") or None,
|
|
job_count=len(job_list),
|
|
queueable_job_count=sum(1 for job in job_list if job.validation_status in queueable_statuses),
|
|
job_manifest_sha256=job_manifest_hash(job_list) if job_list else None,
|
|
effective_policy_sha256=_policy_fingerprint(raw_json, delivery),
|
|
smtp_transport_revision=smtp_transport_revision,
|
|
imap_transport_revision=imap_transport_revision,
|
|
created_at=datetime.now(timezone.utc).isoformat(),
|
|
delivery=delivery,
|
|
).model_dump(mode="json")
|
|
return payload, snapshot_hash(payload)
|
|
|
|
|
|
def _assert_snapshot_matches_persisted_inputs(
|
|
session: Session,
|
|
version: CampaignVersion,
|
|
snapshot: ExecutionSnapshot,
|
|
*,
|
|
effect_job: CampaignJob | None = None,
|
|
) -> None:
|
|
"""Fail closed when any build-bound input drifted after snapshot creation."""
|
|
|
|
raw_json = version.raw_json if isinstance(version.raw_json, dict) else {}
|
|
if snapshot.campaign_version_id != version.id:
|
|
raise ExecutionSnapshotError("Execution snapshot campaign version mismatch")
|
|
if snapshot.campaign_json_sha256 != _sha256(raw_json):
|
|
raise ExecutionSnapshotError(
|
|
"Campaign inputs changed after this execution snapshot was built. "
|
|
"Revalidate and rebuild the campaign before delivery."
|
|
)
|
|
if not snapshot.smtp_transport_revision:
|
|
raise ExecutionSnapshotError("Execution snapshot has no SMTP transport revision")
|
|
if not snapshot.job_manifest_sha256:
|
|
raise ExecutionSnapshotError("Execution snapshot has no built-job manifest checksum")
|
|
if not snapshot.effective_policy_sha256:
|
|
raise ExecutionSnapshotError("Execution snapshot has no effective-policy checksum")
|
|
if snapshot.effective_policy_sha256 != _policy_fingerprint(raw_json, snapshot.delivery):
|
|
raise ExecutionSnapshotError(
|
|
"Campaign delivery policy changed after the execution snapshot was created. "
|
|
"Revalidate and rebuild the campaign before delivery."
|
|
)
|
|
|
|
if effect_job is not None:
|
|
if effect_job.campaign_version_id != version.id:
|
|
raise ExecutionSnapshotError("Campaign job does not belong to the snapshotted version")
|
|
if not getattr(effect_job, "execution_input_sha256", None):
|
|
raise ExecutionSnapshotError("Campaign job has no execution-input checksum; rebuild before delivery")
|
|
if effect_job.execution_input_sha256 != job_execution_input_hash(effect_job):
|
|
raise ExecutionSnapshotError(
|
|
"Built campaign job inputs changed after the execution snapshot was created. "
|
|
"Revalidate and rebuild the campaign before delivery."
|
|
)
|
|
return
|
|
|
|
jobs = (
|
|
session.query(CampaignJob)
|
|
.filter(CampaignJob.campaign_version_id == version.id)
|
|
.order_by(CampaignJob.entry_index.asc(), CampaignJob.id.asc())
|
|
.all()
|
|
)
|
|
queueable_statuses = {JobValidationStatus.READY.value, JobValidationStatus.WARNING.value}
|
|
if snapshot.job_count != len(jobs):
|
|
raise ExecutionSnapshotError(
|
|
"Built campaign jobs changed after the execution snapshot was created. "
|
|
"Revalidate and rebuild the campaign before delivery."
|
|
)
|
|
queueable_count = sum(1 for job in jobs if job.validation_status in queueable_statuses)
|
|
if (
|
|
snapshot.queueable_job_count != queueable_count
|
|
or snapshot.job_manifest_sha256 != job_manifest_hash(jobs)
|
|
or any(getattr(job, "execution_input_sha256", None) != job_execution_input_hash(job) for job in jobs)
|
|
):
|
|
raise ExecutionSnapshotError(
|
|
"Built campaign job inputs changed after the execution snapshot was created. "
|
|
"Revalidate and rebuild the campaign before delivery."
|
|
)
|
|
|
|
|
|
def ensure_execution_snapshot(
|
|
session: Session,
|
|
version: CampaignVersion,
|
|
*,
|
|
effect_job: CampaignJob | None = None,
|
|
) -> ExecutionSnapshot:
|
|
"""Return a validated snapshot, creating one for pre-migration builds.
|
|
|
|
New builds create the snapshot after persisting their jobs. The fallback is
|
|
intentionally limited to legacy built versions so they can be operated
|
|
without a manual data migration.
|
|
"""
|
|
|
|
raw_json = version.raw_json if isinstance(version.raw_json, dict) else {}
|
|
try:
|
|
assert_server_safe_campaign_paths(
|
|
raw_json,
|
|
managed_files_available=files_integration().available,
|
|
)
|
|
except CampaignPathSecurityError as exc:
|
|
raise ExecutionSnapshotError(str(exc)) from exc
|
|
_assert_version_mail_profile_boundary(raw_json)
|
|
|
|
if isinstance(version.execution_snapshot, dict):
|
|
if str(version.execution_snapshot.get("snapshot_version") or "") != SNAPSHOT_VERSION:
|
|
raise ExecutionSnapshotError(
|
|
"This campaign has a legacy execution snapshot that may contain campaign-owned transport data. "
|
|
"It is preserved for audit only and cannot be delivered; select a Mail profile, then revalidate "
|
|
"and rebuild a new campaign version."
|
|
)
|
|
snapshot = ExecutionSnapshot.model_validate(version.execution_snapshot)
|
|
expected = snapshot_hash(snapshot.model_dump(mode="json"))
|
|
if not version.execution_snapshot_hash:
|
|
raise ExecutionSnapshotError("Execution snapshot checksum is missing")
|
|
if version.execution_snapshot_hash != expected:
|
|
raise ExecutionSnapshotError("Execution snapshot checksum mismatch")
|
|
_assert_snapshot_profile_matches_version(version, snapshot)
|
|
_assert_snapshot_matches_persisted_inputs(
|
|
session,
|
|
version,
|
|
snapshot,
|
|
effect_job=effect_job,
|
|
)
|
|
return snapshot
|
|
|
|
from govoplan_campaign.backend.persistence.campaigns import load_version_config
|
|
|
|
_, _, config = load_version_config(session, version.id)
|
|
profile_id = campaign_mail_profile_id(raw_json)
|
|
if not config.server.profile_capabilities.smtp_available:
|
|
raise ExecutionSnapshotError("The selected Mail profile has no SMTP configuration")
|
|
if profile_id is None:
|
|
raise ExecutionSnapshotError("Campaign has no Mail profile reference")
|
|
jobs = (
|
|
session.query(CampaignJob)
|
|
.filter(CampaignJob.campaign_version_id == version.id)
|
|
.order_by(CampaignJob.entry_index.asc(), CampaignJob.id.asc())
|
|
.all()
|
|
)
|
|
if not jobs:
|
|
raise ExecutionSnapshotError("Campaign version has no built jobs; rebuild it before delivery")
|
|
summary = profile_delivery_summary(session, version)
|
|
if not summary.get("smtp_transport_revision"):
|
|
raise ExecutionSnapshotError("The selected Mail profile has no SMTP transport revision")
|
|
payload, digest = create_execution_snapshot(
|
|
version,
|
|
mail_profile_id=profile_id,
|
|
smtp_server_id=summary.get("smtp_server_id"),
|
|
smtp_credential_id=summary.get("smtp_credential_id"),
|
|
imap_server_id=summary.get("imap_server_id"),
|
|
imap_credential_id=summary.get("imap_credential_id"),
|
|
smtp_transport_revision=summary["smtp_transport_revision"],
|
|
imap_transport_revision=summary.get("imap_transport_revision"),
|
|
delivery=config.delivery,
|
|
jobs=jobs,
|
|
build_summary=version.build_summary if isinstance(version.build_summary, dict) else {},
|
|
)
|
|
version.execution_snapshot = payload
|
|
version.execution_snapshot_hash = digest
|
|
version.execution_snapshot_at = datetime.now(timezone.utc)
|
|
session.add(version)
|
|
session.flush()
|
|
return ExecutionSnapshot.model_validate(payload)
|
|
|
|
|
|
def clear_execution_snapshot(version: CampaignVersion) -> None:
|
|
version.execution_snapshot = None
|
|
version.execution_snapshot_hash = None
|
|
version.execution_snapshot_at = None
|