From af833ca38c1083fbbfe604aa1d5c83daf588f8b4 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 17:55:30 +0200 Subject: [PATCH] security: bound Campaign editor metadata --- .../backend/campaign/mail_profile_boundary.py | 138 ++++++++++++++++++ .../backend/persistence/versions.py | 10 +- src/govoplan_campaign/backend/schemas.py | 19 ++- tests/test_editor_state_security.py | 110 ++++++++++++++ 4 files changed, 274 insertions(+), 3 deletions(-) create mode 100644 tests/test_editor_state_security.py diff --git a/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py b/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py index 2da7dff..8cfdfa0 100644 --- a/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py +++ b/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py @@ -1,9 +1,23 @@ from __future__ import annotations +import copy from typing import Any CAMPAIGN_MAIL_SERVER_KEYS = frozenset({"mail_profile_id"}) +CAMPAIGN_CLIENT_EDITOR_STATE_KEYS = frozenset({"created_from", "field_overrides", "opt_ins"}) +CAMPAIGN_OPT_IN_KEYS = frozenset( + {"campaign_address_suggestions", "remember_used_addresses", "inline_guidance"} +) +CAMPAIGN_REVIEW_STATE_KEYS = frozenset( + { + "build_token", + "inspection_complete", + "reviewed_message_keys", + "updated_at", + "updated_by_user_id", + } +) class CampaignMailProfileBoundaryError(ValueError): @@ -15,6 +29,130 @@ class CampaignMailProfileBoundaryError(ValueError): """ +def validate_campaign_editor_state( + value: dict[str, Any] | None, + *, + allow_server_review_state: bool = False, +) -> dict[str, Any]: + """Validate the bounded Campaign-owned UI metadata contract. + + Arbitrary editor metadata would be a second, weakly typed persistence and + response channel for Mail credentials. Review evidence is server-owned and + cannot be supplied through ordinary version create/update requests. + """ + + if value is None: + return {} + if not isinstance(value, dict): + raise CampaignMailProfileBoundaryError("Campaign editor state must be an object") + allowed = set(CAMPAIGN_CLIENT_EDITOR_STATE_KEYS) + if allow_server_review_state: + allowed.add("review_send") + if any(key not in allowed for key in value): + raise CampaignMailProfileBoundaryError( + "Campaign editor state contains unsupported or transport-owned fields" + ) + + result: dict[str, Any] = {} + if "created_from" in value: + created_from = value["created_from"] + if not isinstance(created_from, str) or not created_from.strip() or len(created_from) > 128: + raise CampaignMailProfileBoundaryError("Campaign editor created_from must be a short string") + result["created_from"] = created_from.strip() + if "opt_ins" in value: + opt_ins = value["opt_ins"] + if not isinstance(opt_ins, dict) or any(key not in CAMPAIGN_OPT_IN_KEYS for key in opt_ins): + raise CampaignMailProfileBoundaryError("Campaign editor opt_ins contains unsupported fields") + if any(not isinstance(item, bool) for item in opt_ins.values()): + raise CampaignMailProfileBoundaryError("Campaign editor opt_ins values must be booleans") + result["opt_ins"] = copy.deepcopy(opt_ins) + if "field_overrides" in value: + overrides = value["field_overrides"] + if not isinstance(overrides, dict) or len(overrides) > 10_000: + raise CampaignMailProfileBoundaryError("Campaign editor field_overrides must be a bounded object") + if any( + not isinstance(key, str) + or not key.strip() + or len(key) > 256 + or not isinstance(item, bool) + for key, item in overrides.items() + ): + raise CampaignMailProfileBoundaryError( + "Campaign editor field_overrides must map short field names to booleans" + ) + result["field_overrides"] = copy.deepcopy(overrides) + if "review_send" in value: + result["review_send"] = _validated_server_review_state(value["review_send"]) + return result + + +def public_campaign_editor_state( + value: Any, + *, + include_diagnostics: bool = False, +) -> dict[str, Any]: + """Return only known non-secret UI metadata from current or legacy rows.""" + + if not isinstance(value, dict): + return {} + result: dict[str, Any] = {} + for key in CAMPAIGN_CLIENT_EDITOR_STATE_KEYS: + if key not in value: + continue + try: + result.update(validate_campaign_editor_state({key: value[key]})) + except CampaignMailProfileBoundaryError: + continue + if "review_send" in value: + try: + review_state = _validated_server_review_state(value["review_send"]) + if not include_diagnostics: + review_state.pop("build_token", None) + result["review_send"] = review_state + except CampaignMailProfileBoundaryError: + pass + return result + + +def campaign_editor_state_for_edit(value: Any) -> dict[str, Any]: + """Copy only client-owned safe metadata into a new editable version.""" + + state = public_campaign_editor_state(value) + state.pop("review_send", None) + return state + + +def _validated_server_review_state(value: Any) -> dict[str, Any]: + if not isinstance(value, dict) or any(key not in CAMPAIGN_REVIEW_STATE_KEYS for key in value): + raise CampaignMailProfileBoundaryError("Campaign review editor state is invalid") + build_token = value.get("build_token") + inspected = value.get("inspection_complete") + keys = value.get("reviewed_message_keys", []) + updated_at = value.get("updated_at") + updated_by = value.get("updated_by_user_id") + if not isinstance(build_token, str) or not build_token.strip() or len(build_token) > 256: + raise CampaignMailProfileBoundaryError("Campaign review build token is invalid") + if not isinstance(inspected, bool): + raise CampaignMailProfileBoundaryError("Campaign review completion state is invalid") + if ( + not isinstance(keys, list) + or len(keys) > 100_000 + or any(not isinstance(key, str) or not key.strip() or len(key) > 512 for key in keys) + ): + raise CampaignMailProfileBoundaryError("Campaign reviewed message keys are invalid") + if not isinstance(updated_at, str) or not updated_at.strip() or len(updated_at) > 128: + raise CampaignMailProfileBoundaryError("Campaign review timestamp is invalid") + if updated_by is not None and (not isinstance(updated_by, str) or len(updated_by) > 256): + raise CampaignMailProfileBoundaryError("Campaign review actor is invalid") + return { + "build_token": build_token.strip(), + "inspection_complete": inspected, + "reviewed_message_keys": list(dict.fromkeys(key.strip() for key in keys)), + "updated_at": updated_at.strip(), + "updated_by_user_id": updated_by, + } + + def campaign_mail_profile_id(raw_json: dict[str, Any] | None) -> str | None: server = raw_json.get("server") if isinstance(raw_json, dict) else None if not isinstance(server, dict): diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index a0bb5e9..3edaa55 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -20,10 +20,12 @@ from govoplan_campaign.backend.db.models import ( ) from govoplan_campaign.backend.sending.execution import clear_execution_snapshot from govoplan_campaign.backend.campaign.mail_profile_boundary import ( + campaign_editor_state_for_edit, campaign_mail_profile_boundary_violations, campaign_mail_profile_id, assert_campaign_uses_mail_profile_reference, public_campaign_mail_server, + validate_campaign_editor_state, ) from govoplan_campaign.backend.integrations import files_integration, mail_integration from govoplan_campaign.backend.persistence.campaigns import ( @@ -391,7 +393,11 @@ def fork_campaign_version_for_edit( current_flow=current_flow if current_flow is not None else (source.current_flow or CampaignVersionFlow.MANUAL.value), current_step=current_step if current_step is not None else source.current_step, is_complete=False, - editor_state=editor_state if editor_state is not None else copy.deepcopy(source.editor_state or {}), + editor_state=( + validate_campaign_editor_state(editor_state) + if editor_state is not None + else campaign_editor_state_for_edit(source.editor_state) + ), autosaved_at=datetime.now(UTC) if autosave else None, ) session.add(new_version) @@ -573,7 +579,7 @@ def update_campaign_version( if is_complete is not None: version.is_complete = is_complete if editor_state is not None: - version.editor_state = editor_state + version.editor_state = validate_campaign_editor_state(editor_state) if source_filename is not None: version.source_filename = source_filename if source_base_path is not None: diff --git a/src/govoplan_campaign/backend/schemas.py b/src/govoplan_campaign/backend/schemas.py index e14abcc..d388112 100644 --- a/src/govoplan_campaign/backend/schemas.py +++ b/src/govoplan_campaign/backend/schemas.py @@ -3,9 +3,13 @@ from __future__ import annotations from datetime import datetime from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator from govoplan_core.api.v1.schemas import DeltaDeletedItem +from govoplan_campaign.backend.campaign.mail_profile_boundary import ( + public_campaign_editor_state, + validate_campaign_editor_state, +) from govoplan_campaign.backend.response_security import ( public_campaign_configuration, public_campaign_payload, @@ -55,6 +59,11 @@ class CampaignVersionUpdateRequest(BaseModel): source_base_path: str | None = None migrate_legacy_mail_settings: bool = False + @field_validator("editor_state") + @classmethod + def validate_editor_state(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: + return validate_campaign_editor_state(value) if value is not None else None + class CampaignVersionSetStepRequest(BaseModel): model_config = ConfigDict(extra="forbid") @@ -104,6 +113,14 @@ class CampaignVersionResponse(BaseModel): execution_snapshot_hash: str | None = None execution_snapshot_at: datetime | None = None + @field_validator("editor_state", mode="before") + @classmethod + def remove_unsupported_editor_state(cls, value: Any, info: ValidationInfo) -> dict[str, Any]: + return public_campaign_editor_state( + value, + include_diagnostics=bool((info.context or {}).get("include_diagnostics")), + ) + @field_validator("source_filename", mode="before") @classmethod def remove_source_directory(cls, value: Any) -> str | None: diff --git a/tests/test_editor_state_security.py b/tests/test_editor_state_security.py new file mode 100644 index 0000000..1c5495c --- /dev/null +++ b/tests/test_editor_state_security.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from govoplan_campaign.backend.campaign.mail_profile_boundary import ( + campaign_editor_state_for_edit, +) +from govoplan_campaign.backend.schemas import ( + CampaignVersionResponse, + CampaignVersionUpdateRequest, +) + + +@pytest.mark.parametrize( + "editor_state", + [ + {"smtp": {"host": "smtp.example.test", "password": "secret"}}, + {"transport": {"imap_password": "secret"}}, + { + "review_send": { + "build_token": "forged", + "inspection_complete": True, + "reviewed_message_keys": [], + "updated_at": "2026-07-21T00:00:00+00:00", + "updated_by_user_id": "user-1", + } + }, + ], +) +def test_version_updates_reject_arbitrary_or_server_owned_editor_state( + editor_state: dict[str, object], +) -> None: + with pytest.raises(ValidationError, match="unsupported or transport-owned"): + CampaignVersionUpdateRequest(editor_state=editor_state) + + +def test_version_response_omits_legacy_secret_bearing_editor_state() -> None: + response = CampaignVersionResponse.model_validate( + { + "id": "version-1", + "campaign_id": "campaign-1", + "version_number": 1, + "schema_version": "1.0", + "editor_state": { + "opt_ins": {"inline_guidance": True}, + "smtp": {"host": "smtp.internal.example", "password": "provider-secret"}, + }, + "created_at": datetime.now(UTC), + "updated_at": datetime.now(UTC), + } + ) + + assert response.editor_state == {"opt_ins": {"inline_guidance": True}} + assert "provider-secret" not in repr(response) + assert "internal.example" not in repr(response) + + +def test_review_build_token_is_visible_only_in_operator_diagnostics() -> None: + value = { + "id": "version-1", + "campaign_id": "campaign-1", + "version_number": 1, + "schema_version": "1.0", + "editor_state": { + "review_send": { + "build_token": "internal-build-token", + "inspection_complete": True, + "reviewed_message_keys": ["entry-1"], + "updated_at": "2026-07-21T00:00:00+00:00", + "updated_by_user_id": "reviewer-1", + } + }, + "created_at": datetime.now(UTC), + "updated_at": datetime.now(UTC), + } + + public = CampaignVersionResponse.model_validate(value) + operator = CampaignVersionResponse.model_validate( + value, + context={"include_diagnostics": True}, + ) + + assert "build_token" not in public.editor_state["review_send"] + assert operator.editor_state["review_send"]["build_token"] == "internal-build-token" + + +def test_fork_copy_keeps_only_client_owned_bounded_metadata() -> None: + copied = campaign_editor_state_for_edit( + { + "created_from": "minimal_campaign", + "field_overrides": {"department": False}, + "review_send": { + "build_token": "build-1", + "inspection_complete": True, + "reviewed_message_keys": ["entry-1"], + "updated_at": "2026-07-21T00:00:00+00:00", + "updated_by_user_id": "reviewer-1", + }, + "credentials": {"password": "legacy-secret"}, + } + ) + + assert copied == { + "created_from": "minimal_campaign", + "field_overrides": {"department": False}, + } + assert "legacy-secret" not in repr(copied)