731 lines
26 KiB
Python
731 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from collections import Counter
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from govoplan_campaign.backend.campaign.loader import validate_against_schema
|
|
from govoplan_campaign.backend.db.models import (
|
|
Campaign,
|
|
CampaignIssue,
|
|
CampaignJob,
|
|
CampaignVersion,
|
|
)
|
|
from govoplan_campaign.backend.persistence.versions import minimal_campaign_json
|
|
from govoplan_campaign.backend.response_security import (
|
|
public_campaign_configuration,
|
|
public_campaign_payload,
|
|
)
|
|
|
|
|
|
PORTABLE_CAMPAIGN_FORMAT = "govoplan.campaign-portable"
|
|
PORTABLE_CAMPAIGN_FORMAT_VERSION = "1.0"
|
|
PORTABLE_CAMPAIGN_SCOPE_ORDER = (
|
|
"metadata",
|
|
"template_config",
|
|
"recipients",
|
|
"attachments",
|
|
"review_state",
|
|
"delivery_history",
|
|
)
|
|
DEFAULT_PORTABLE_CAMPAIGN_SCOPES = ("metadata", "template_config")
|
|
OPERATIONAL_EVIDENCE_SCOPES = frozenset(("review_state", "delivery_history"))
|
|
_CONFIG_STRUCTURAL_KEYS = frozenset(
|
|
("version", "campaign", "recipients", "entries", "attachments")
|
|
)
|
|
_SENSITIVE_SETTING_FRAGMENTS = (
|
|
"api_key",
|
|
"credential",
|
|
"password",
|
|
"private_key",
|
|
"secret",
|
|
"token",
|
|
)
|
|
|
|
|
|
class CampaignTransferError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CampaignImportInspection:
|
|
preview: dict[str, Any]
|
|
configuration: dict[str, Any] | None
|
|
portable_settings: dict[str, Any]
|
|
|
|
|
|
def canonical_sha256(value: object) -> str:
|
|
encoded = json.dumps(
|
|
value,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
allow_nan=False,
|
|
).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def normalize_transfer_scopes(scopes: Iterable[str]) -> tuple[str, ...]:
|
|
selected = set(scopes)
|
|
invalid = sorted(selected.difference(PORTABLE_CAMPAIGN_SCOPE_ORDER))
|
|
if invalid:
|
|
raise CampaignTransferError(
|
|
f"Unsupported campaign transfer scope(s): {', '.join(invalid)}"
|
|
)
|
|
if not selected:
|
|
raise CampaignTransferError("Select at least one campaign transfer scope.")
|
|
return tuple(scope for scope in PORTABLE_CAMPAIGN_SCOPE_ORDER if scope in selected)
|
|
|
|
|
|
def build_campaign_portable_package(
|
|
*,
|
|
campaign: Campaign,
|
|
version: CampaignVersion,
|
|
scopes: Iterable[str],
|
|
jobs: Iterable[CampaignJob] = (),
|
|
issues: Iterable[CampaignIssue] = (),
|
|
module_version: str,
|
|
) -> dict[str, Any]:
|
|
selected = normalize_transfer_scopes(scopes)
|
|
configuration = public_campaign_configuration(version.raw_json)
|
|
if not isinstance(configuration, dict):
|
|
raise CampaignTransferError("The campaign configuration is not portable JSON.")
|
|
configuration, password_redactions = _redact_password_field_values(configuration)
|
|
payload: dict[str, Any] = {}
|
|
item_counts: dict[str, int] = {}
|
|
redactions: Counter[str] = Counter(password_redactions)
|
|
|
|
if "metadata" in selected:
|
|
payload["metadata"] = {
|
|
"external_id": campaign.external_id,
|
|
"name": campaign.name,
|
|
"description": campaign.description,
|
|
"source_status": campaign.status,
|
|
}
|
|
item_counts["metadata"] = 1
|
|
|
|
if "template_config" in selected:
|
|
settings, setting_redactions = _redact_sensitive_settings(
|
|
campaign.settings or {}
|
|
)
|
|
mail_policy, mail_policy_redactions = _redact_sensitive_settings(
|
|
campaign.mail_profile_policy or {}
|
|
)
|
|
template_configuration = {
|
|
key: copy.deepcopy(value)
|
|
for key, value in configuration.items()
|
|
if key not in _CONFIG_STRUCTURAL_KEYS
|
|
}
|
|
server = template_configuration.get("server")
|
|
if isinstance(server, dict):
|
|
for key in ("smtp_credential_id", "imap_credential_id"):
|
|
if server.pop(key, None) is not None:
|
|
redactions["deployment_credential_reference"] += 1
|
|
payload["template_config"] = {
|
|
"schema_version": version.schema_version,
|
|
"configuration": template_configuration,
|
|
"campaign_settings": settings,
|
|
"mail_profile_policy": mail_policy,
|
|
}
|
|
redactions.update(setting_redactions)
|
|
redactions.update(mail_policy_redactions)
|
|
item_counts["template_config"] = len(template_configuration)
|
|
|
|
if "recipients" in selected:
|
|
entries = copy.deepcopy(configuration.get("entries") or {})
|
|
_remove_entry_attachments(entries)
|
|
payload["recipients"] = {
|
|
"recipients": copy.deepcopy(configuration.get("recipients") or {}),
|
|
"entries": entries,
|
|
}
|
|
item_counts["recipients"] = _recipient_entry_count(entries)
|
|
|
|
if "attachments" in selected:
|
|
entry_attachments = _entry_attachment_projection(
|
|
configuration.get("entries")
|
|
)
|
|
payload["attachments"] = {
|
|
"configuration": copy.deepcopy(configuration.get("attachments") or {}),
|
|
"entry_attachments": entry_attachments,
|
|
"content_included": False,
|
|
}
|
|
item_counts["attachments"] = _attachment_rule_count(
|
|
payload["attachments"]
|
|
)
|
|
|
|
issue_rows = tuple(issues)
|
|
if "review_state" in selected:
|
|
review_state = _review_state_projection(version, issue_rows)
|
|
payload["review_state"] = review_state
|
|
item_counts["review_state"] = int(review_state["decision_count"])
|
|
|
|
job_rows = tuple(jobs)
|
|
if "delivery_history" in selected:
|
|
payload["delivery_history"] = {
|
|
"jobs": [_delivery_job_projection(job) for job in job_rows],
|
|
"counts": _delivery_counts(job_rows),
|
|
}
|
|
item_counts["delivery_history"] = len(job_rows)
|
|
|
|
exported_at = datetime.now(UTC)
|
|
package: dict[str, Any] = {
|
|
"format": PORTABLE_CAMPAIGN_FORMAT,
|
|
"format_version": PORTABLE_CAMPAIGN_FORMAT_VERSION,
|
|
"package_id": str(uuid4()),
|
|
"exported_at": exported_at.isoformat(),
|
|
"source": {
|
|
"module": "campaigns",
|
|
"module_version": module_version,
|
|
"tenant_ref_sha256": hashlib.sha256(
|
|
campaign.tenant_id.encode("utf-8")
|
|
).hexdigest(),
|
|
"campaign_id": campaign.id,
|
|
"campaign_external_id": campaign.external_id,
|
|
"campaign_name": campaign.name,
|
|
"version_id": version.id,
|
|
"version_number": version.version_number,
|
|
"campaign_schema_version": version.schema_version,
|
|
},
|
|
"scopes": list(selected),
|
|
"manifest": {
|
|
"item_counts": item_counts,
|
|
"redactions": dict(sorted(redactions.items())),
|
|
"privacy_default_scopes": list(DEFAULT_PORTABLE_CAMPAIGN_SCOPES),
|
|
"attachments_are_references_only": True,
|
|
"operational_evidence_is_not_replayed": True,
|
|
"secrets_included": False,
|
|
},
|
|
"payload": payload,
|
|
}
|
|
package["integrity"] = {
|
|
"algorithm": "sha256",
|
|
"package_sha256": canonical_sha256(package),
|
|
}
|
|
return package
|
|
|
|
|
|
def inspect_campaign_portable_package(
|
|
package: Mapping[str, Any],
|
|
*,
|
|
selected_scopes: Iterable[str] | None,
|
|
external_id: str,
|
|
name: str,
|
|
) -> CampaignImportInspection:
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
package_dict = copy.deepcopy(dict(package))
|
|
package_id = _optional_text(package_dict.get("package_id"))
|
|
format_version = _optional_text(package_dict.get("format_version"))
|
|
source = package_dict.get("source")
|
|
source_dict = copy.deepcopy(source) if isinstance(source, dict) else {}
|
|
integrity = package_dict.get("integrity")
|
|
expected_hash = (
|
|
_optional_text(integrity.get("package_sha256"))
|
|
if isinstance(integrity, dict)
|
|
else None
|
|
)
|
|
hash_input = copy.deepcopy(package_dict)
|
|
hash_input.pop("integrity", None)
|
|
actual_hash = canonical_sha256(hash_input)
|
|
|
|
if package_dict.get("format") != PORTABLE_CAMPAIGN_FORMAT:
|
|
errors.append("The file is not a GovOPlaN portable Campaign package.")
|
|
if format_version != PORTABLE_CAMPAIGN_FORMAT_VERSION:
|
|
errors.append(
|
|
"The Campaign package format version is not supported by this installation."
|
|
)
|
|
if not package_id:
|
|
errors.append("The Campaign package has no package identifier.")
|
|
if not expected_hash or expected_hash != actual_hash:
|
|
errors.append("The Campaign package integrity checksum does not match its content.")
|
|
if not isinstance(integrity, dict) or integrity.get("algorithm") != "sha256":
|
|
errors.append("The Campaign package does not use the supported SHA-256 integrity algorithm.")
|
|
if not source_dict:
|
|
errors.append("The Campaign package has no source provenance.")
|
|
elif source_dict.get("campaign_schema_version") != "1.0":
|
|
errors.append("The Campaign configuration schema version is not supported by this installation.")
|
|
|
|
available: tuple[str, ...] = ()
|
|
try:
|
|
raw_scopes = package_dict.get("scopes")
|
|
if not isinstance(raw_scopes, list):
|
|
raise CampaignTransferError("The Campaign package has no valid scope list.")
|
|
available = normalize_transfer_scopes(str(item) for item in raw_scopes)
|
|
except CampaignTransferError as exc:
|
|
errors.append(str(exc))
|
|
|
|
try:
|
|
selected = normalize_transfer_scopes(
|
|
available if selected_scopes is None else selected_scopes
|
|
)
|
|
except CampaignTransferError as exc:
|
|
errors.append(str(exc))
|
|
selected = ()
|
|
unavailable = sorted(set(selected).difference(available))
|
|
if unavailable:
|
|
errors.append(
|
|
f"Selected scope(s) are absent from the package: {', '.join(unavailable)}"
|
|
)
|
|
|
|
payload = package_dict.get("payload")
|
|
payload_dict = payload if isinstance(payload, dict) else {}
|
|
if not isinstance(payload, dict):
|
|
errors.append("The Campaign package has no valid payload object.")
|
|
if not isinstance(package_dict.get("manifest"), dict):
|
|
errors.append("The Campaign package has no valid manifest.")
|
|
for scope in available:
|
|
if scope not in payload_dict:
|
|
errors.append(f"The Campaign package payload is missing scope '{scope}'.")
|
|
elif not isinstance(payload_dict[scope], dict):
|
|
errors.append(f"The Campaign package scope '{scope}' is not a valid object.")
|
|
|
|
template_scope = payload_dict.get("template_config")
|
|
if (
|
|
"template_config" in available
|
|
and isinstance(template_scope, dict)
|
|
and template_scope.get("schema_version") != "1.0"
|
|
):
|
|
errors.append("The portable template/configuration schema version is not supported.")
|
|
|
|
configuration: dict[str, Any] | None = None
|
|
portable_settings: dict[str, Any] = {}
|
|
will_create: list[dict[str, Any]] = []
|
|
will_skip: list[dict[str, Any]] = []
|
|
if not errors:
|
|
configuration, portable_settings, created, skipped, materialize_warnings = (
|
|
_materialize_import(
|
|
payload_dict,
|
|
available=available,
|
|
selected=selected,
|
|
external_id=external_id,
|
|
name=name,
|
|
)
|
|
)
|
|
will_create.extend(created)
|
|
will_skip.extend(skipped)
|
|
warnings.extend(materialize_warnings)
|
|
try:
|
|
validate_against_schema(configuration)
|
|
except Exception as exc:
|
|
errors.append(f"The imported Campaign configuration is incompatible: {exc}")
|
|
configuration = None
|
|
|
|
manifest = package_dict.get("manifest")
|
|
if isinstance(manifest, dict) and manifest.get("redactions"):
|
|
warnings.append(
|
|
"The source export redacted sensitive or deployment-bound values; review the package manifest and reconfigure them locally."
|
|
)
|
|
|
|
preview = {
|
|
"compatible": not errors,
|
|
"package_id": package_id,
|
|
"package_sha256": actual_hash,
|
|
"format_version": format_version,
|
|
"source": source_dict,
|
|
"available_scopes": list(available),
|
|
"selected_scopes": list(selected),
|
|
"destination": {
|
|
"external_id": external_id,
|
|
"name": name,
|
|
"status": "draft",
|
|
},
|
|
"will_create": will_create,
|
|
"will_skip": will_skip,
|
|
"warnings": list(dict.fromkeys(warnings)),
|
|
"errors": list(dict.fromkeys(errors)),
|
|
}
|
|
return CampaignImportInspection(
|
|
preview=preview,
|
|
configuration=configuration,
|
|
portable_settings=portable_settings,
|
|
)
|
|
|
|
|
|
def _materialize_import(
|
|
payload: Mapping[str, Any],
|
|
*,
|
|
available: tuple[str, ...],
|
|
selected: tuple[str, ...],
|
|
external_id: str,
|
|
name: str,
|
|
) -> tuple[
|
|
dict[str, Any],
|
|
dict[str, Any],
|
|
list[dict[str, Any]],
|
|
list[dict[str, Any]],
|
|
list[str],
|
|
]:
|
|
selected_set = set(selected)
|
|
configuration = minimal_campaign_json(external_id=external_id, name=name)
|
|
portable_settings: dict[str, Any] = {}
|
|
created: list[dict[str, Any]] = [
|
|
_plan_item("metadata", "campaign_draft", "A new Campaign draft and editable version will be created.", 1)
|
|
]
|
|
skipped: list[dict[str, Any]] = []
|
|
warnings: list[str] = []
|
|
|
|
metadata = payload.get("metadata")
|
|
if "metadata" in selected_set and isinstance(metadata, dict):
|
|
description = metadata.get("description")
|
|
if isinstance(description, str):
|
|
configuration["campaign"]["description"] = description
|
|
|
|
template_payload = payload.get("template_config")
|
|
if "template_config" in selected_set and isinstance(template_payload, dict):
|
|
source_configuration = template_payload.get("configuration")
|
|
if isinstance(source_configuration, dict):
|
|
for key, value in source_configuration.items():
|
|
if key in _CONFIG_STRUCTURAL_KEYS:
|
|
continue
|
|
configuration[key] = copy.deepcopy(value)
|
|
source_server = configuration.get("server")
|
|
if isinstance(source_server, dict) and source_server:
|
|
configuration["server"] = {}
|
|
skipped.append(
|
|
_plan_item(
|
|
"template_config",
|
|
"deployment_bound_mail_profile",
|
|
"Mail profile and server references are not applied across installations; select local Mail resources after import.",
|
|
len(source_server),
|
|
)
|
|
)
|
|
settings = template_payload.get("campaign_settings")
|
|
if isinstance(settings, dict):
|
|
portable_settings = copy.deepcopy(settings)
|
|
created.append(
|
|
_plan_item(
|
|
"template_config",
|
|
"editable_configuration",
|
|
"Portable fields, template, delivery settings, and validation policy will be applied to the draft.",
|
|
len(source_configuration),
|
|
)
|
|
)
|
|
|
|
recipients_payload = payload.get("recipients")
|
|
if "recipients" in selected_set and isinstance(recipients_payload, dict):
|
|
recipients = recipients_payload.get("recipients")
|
|
entries = recipients_payload.get("entries")
|
|
if isinstance(recipients, dict):
|
|
configuration["recipients"] = copy.deepcopy(recipients)
|
|
if isinstance(entries, dict):
|
|
configuration["entries"] = copy.deepcopy(entries)
|
|
created.append(
|
|
_plan_item(
|
|
"recipients",
|
|
"recipient_rows",
|
|
"Campaign-local recipient rows and source provenance will be copied into the draft.",
|
|
_recipient_entry_count(entries),
|
|
)
|
|
)
|
|
|
|
attachments_payload = payload.get("attachments")
|
|
if "attachments" in selected_set and isinstance(attachments_payload, dict):
|
|
attachment_configuration = attachments_payload.get("configuration")
|
|
if isinstance(attachment_configuration, dict):
|
|
configuration["attachments"] = copy.deepcopy(attachment_configuration)
|
|
per_entry = attachments_payload.get("entry_attachments")
|
|
applied_entry_rules = 0
|
|
if "recipients" in selected_set and isinstance(per_entry, list):
|
|
inline = configuration.get("entries", {}).get("inline", [])
|
|
if isinstance(inline, list):
|
|
for item in per_entry:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
index = item.get("entry_index")
|
|
rules = item.get("attachments")
|
|
if (
|
|
isinstance(index, int)
|
|
and 0 <= index < len(inline)
|
|
and isinstance(inline[index], dict)
|
|
and isinstance(rules, list)
|
|
):
|
|
inline[index]["attachments"] = copy.deepcopy(rules)
|
|
applied_entry_rules += len(rules)
|
|
elif isinstance(per_entry, list) and per_entry:
|
|
skipped.append(
|
|
_plan_item(
|
|
"attachments",
|
|
"recipient_scope_required",
|
|
"Per-recipient attachment rules are skipped unless recipient rows are also imported.",
|
|
sum(
|
|
len(item.get("attachments") or [])
|
|
for item in per_entry
|
|
if isinstance(item, dict)
|
|
),
|
|
)
|
|
)
|
|
created.append(
|
|
_plan_item(
|
|
"attachments",
|
|
"attachment_references",
|
|
"Portable attachment rules will be applied; file content is never embedded in the package.",
|
|
_attachment_rule_count(attachments_payload) - max(0, _entry_rule_count(per_entry) - applied_entry_rules),
|
|
)
|
|
)
|
|
warnings.append(
|
|
"Attachment rules contain references only. Reconnect or upload the required files and validate the draft before use."
|
|
)
|
|
|
|
for scope in PORTABLE_CAMPAIGN_SCOPE_ORDER:
|
|
if scope not in OPERATIONAL_EVIDENCE_SCOPES:
|
|
continue
|
|
if scope in selected_set:
|
|
item_count = _manifest_scope_count(payload.get(scope))
|
|
skipped.append(
|
|
_plan_item(
|
|
scope,
|
|
"operational_evidence_not_replayed",
|
|
"Historical review or delivery evidence remains in the source package and import receipt but is never replayed as live Campaign state.",
|
|
item_count,
|
|
)
|
|
)
|
|
|
|
for scope in available:
|
|
if scope not in selected_set:
|
|
skipped.append(
|
|
_plan_item(
|
|
scope,
|
|
"scope_not_selected",
|
|
"This available package scope was not selected for import.",
|
|
_manifest_scope_count(payload.get(scope)),
|
|
)
|
|
)
|
|
|
|
campaign_metadata = configuration.get("campaign")
|
|
if not isinstance(campaign_metadata, dict):
|
|
raise CampaignTransferError("The imported Campaign metadata is invalid.")
|
|
campaign_metadata.update({"id": external_id, "name": name, "mode": "draft"})
|
|
return configuration, portable_settings, created, skipped, warnings
|
|
|
|
|
|
def _review_state_projection(
|
|
version: CampaignVersion, issues: tuple[CampaignIssue, ...]
|
|
) -> dict[str, Any]:
|
|
editor_state = version.editor_state if isinstance(version.editor_state, dict) else {}
|
|
review = editor_state.get("review_send")
|
|
review = review if isinstance(review, dict) else {}
|
|
decisions = [
|
|
item
|
|
for item in (review.get("issue_decisions") or [])
|
|
if isinstance(item, dict)
|
|
]
|
|
decision_evidence = [
|
|
{
|
|
"decision": item.get("decision"),
|
|
"issue_codes": sorted(str(code) for code in item.get("issue_codes") or []),
|
|
"issue_fingerprint": item.get("issue_fingerprint"),
|
|
"message_sha256": item.get("message_sha256"),
|
|
"reason_recorded": bool(str(item.get("reason") or "").strip()),
|
|
}
|
|
for item in decisions
|
|
]
|
|
issue_counts = Counter(str(issue.severity) for issue in issues)
|
|
return {
|
|
"workflow_state": version.workflow_state,
|
|
"inspection_complete": bool(review.get("inspection_complete")),
|
|
"reviewed_message_count": len(review.get("reviewed_message_keys") or []),
|
|
"decision_count": len(decisions),
|
|
"decision_evidence_sha256": canonical_sha256(decision_evidence),
|
|
"issue_counts": dict(sorted(issue_counts.items())),
|
|
"validation_summary": public_campaign_payload(version.validation_summary or {}),
|
|
"build_summary": public_campaign_payload(version.build_summary or {}),
|
|
}
|
|
|
|
|
|
def _delivery_job_projection(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,
|
|
"message_id_header": job.message_id_header,
|
|
"message_sha256": job.eml_sha256,
|
|
"build_status": job.build_status,
|
|
"validation_status": job.validation_status,
|
|
"queue_status": job.queue_status,
|
|
"send_status": job.send_status,
|
|
"postbox_status": job.postbox_status,
|
|
"print_status": job.print_status,
|
|
"imap_status": job.imap_status,
|
|
"attempt_count": job.attempt_count,
|
|
"sent_at": _isoformat(job.sent_at),
|
|
"outcome_unknown_at": _isoformat(job.outcome_unknown_at),
|
|
"delivery_provenance": public_campaign_payload(job.delivery_provenance or {}),
|
|
}
|
|
|
|
|
|
def _delivery_counts(jobs: tuple[CampaignJob, ...]) -> dict[str, dict[str, int]]:
|
|
return {
|
|
field: dict(
|
|
sorted(Counter(str(getattr(job, field) or "unknown") for job in jobs).items())
|
|
)
|
|
for field in ("validation_status", "queue_status", "send_status")
|
|
}
|
|
|
|
|
|
def _redact_sensitive_settings(
|
|
value: Mapping[str, Any],
|
|
) -> tuple[dict[str, Any], Counter[str]]:
|
|
redactions: Counter[str] = Counter()
|
|
|
|
def visit(item: Any) -> Any:
|
|
if isinstance(item, dict):
|
|
result: dict[str, Any] = {}
|
|
for raw_key, child in item.items():
|
|
key = str(raw_key)
|
|
normalized = key.lower().replace("-", "_")
|
|
if any(fragment in normalized for fragment in _SENSITIVE_SETTING_FRAGMENTS):
|
|
redactions["sensitive_setting"] += 1
|
|
continue
|
|
result[key] = visit(child)
|
|
return result
|
|
if isinstance(item, list):
|
|
return [visit(child) for child in item]
|
|
return copy.deepcopy(item)
|
|
|
|
return visit(dict(value)), redactions
|
|
|
|
|
|
def _redact_password_field_values(
|
|
configuration: dict[str, Any],
|
|
) -> tuple[dict[str, Any], Counter[str]]:
|
|
result = copy.deepcopy(configuration)
|
|
password_fields = {
|
|
str(field.get("name"))
|
|
for field in result.get("fields") or []
|
|
if isinstance(field, dict)
|
|
and field.get("type") == "password"
|
|
and field.get("name")
|
|
}
|
|
redactions: Counter[str] = Counter()
|
|
if not password_fields:
|
|
return result, redactions
|
|
global_values = result.get("global_values")
|
|
if isinstance(global_values, dict):
|
|
for key in password_fields:
|
|
if global_values.pop(key, None) is not None:
|
|
redactions["password_field_value"] += 1
|
|
entries = result.get("entries")
|
|
if isinstance(entries, dict):
|
|
for entry in entries.get("inline") or []:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
fields = entry.get("fields")
|
|
if not isinstance(fields, dict):
|
|
continue
|
|
for key in password_fields:
|
|
if fields.pop(key, None) is not None:
|
|
redactions["password_field_value"] += 1
|
|
return result, redactions
|
|
|
|
|
|
def _remove_entry_attachments(entries: Any) -> None:
|
|
if not isinstance(entries, dict):
|
|
return
|
|
for entry in entries.get("inline") or []:
|
|
if isinstance(entry, dict):
|
|
entry["attachments"] = []
|
|
defaults = entries.get("defaults")
|
|
if isinstance(defaults, dict):
|
|
defaults["attachments"] = []
|
|
|
|
|
|
def _entry_attachment_projection(entries: Any) -> list[dict[str, Any]]:
|
|
if not isinstance(entries, dict):
|
|
return []
|
|
result: list[dict[str, Any]] = []
|
|
for index, entry in enumerate(entries.get("inline") or []):
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
rules = entry.get("attachments")
|
|
if isinstance(rules, list) and rules:
|
|
result.append(
|
|
{
|
|
"entry_index": index,
|
|
"attachments": copy.deepcopy(rules),
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def _recipient_entry_count(entries: Any) -> int:
|
|
if not isinstance(entries, dict):
|
|
return 0
|
|
inline = entries.get("inline")
|
|
return len(inline) if isinstance(inline, list) else 0
|
|
|
|
|
|
def _attachment_rule_count(value: Any) -> int:
|
|
if not isinstance(value, dict):
|
|
return 0
|
|
configuration = value.get("configuration")
|
|
global_rules = (
|
|
configuration.get("global") if isinstance(configuration, dict) else []
|
|
)
|
|
return (len(global_rules) if isinstance(global_rules, list) else 0) + _entry_rule_count(
|
|
value.get("entry_attachments")
|
|
)
|
|
|
|
|
|
def _entry_rule_count(value: Any) -> int:
|
|
if not isinstance(value, list):
|
|
return 0
|
|
return sum(
|
|
len(item.get("attachments") or [])
|
|
for item in value
|
|
if isinstance(item, dict)
|
|
)
|
|
|
|
|
|
def _manifest_scope_count(value: Any) -> int:
|
|
if not isinstance(value, dict):
|
|
return 0
|
|
if isinstance(value.get("jobs"), list):
|
|
return len(value["jobs"])
|
|
if "decision_count" in value:
|
|
return int(value.get("decision_count") or 0)
|
|
if "entries" in value:
|
|
return _recipient_entry_count(value.get("entries"))
|
|
return 1
|
|
|
|
|
|
def _plan_item(
|
|
scope: str, code: str, summary: str, item_count: int | None
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"scope": scope,
|
|
"code": code,
|
|
"summary": summary,
|
|
"item_count": item_count,
|
|
}
|
|
|
|
|
|
def _optional_text(value: object) -> str | None:
|
|
text = str(value or "").strip()
|
|
return text or None
|
|
|
|
|
|
def _isoformat(value: datetime | None) -> str | None:
|
|
return value.isoformat() if value is not None else None
|
|
|
|
|
|
__all__ = [
|
|
"CampaignImportInspection",
|
|
"CampaignTransferError",
|
|
"DEFAULT_PORTABLE_CAMPAIGN_SCOPES",
|
|
"OPERATIONAL_EVIDENCE_SCOPES",
|
|
"PORTABLE_CAMPAIGN_FORMAT",
|
|
"PORTABLE_CAMPAIGN_FORMAT_VERSION",
|
|
"PORTABLE_CAMPAIGN_SCOPE_ORDER",
|
|
"build_campaign_portable_package",
|
|
"canonical_sha256",
|
|
"inspect_campaign_portable_package",
|
|
"normalize_transfer_scopes",
|
|
]
|