diff --git a/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py b/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py index 8cfdfa0..95acacf 100644 --- a/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py +++ b/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py @@ -29,6 +29,47 @@ class CampaignMailProfileBoundaryError(ValueError): """ +def _validated_opt_ins(value: Any) -> dict[str, bool]: + if not isinstance(value, dict) or any( + key not in CAMPAIGN_OPT_IN_KEYS for key in value + ): + raise CampaignMailProfileBoundaryError( + "Campaign editor opt_ins contains unsupported fields" + ) + if any(not isinstance(item, bool) for item in value.values()): + raise CampaignMailProfileBoundaryError( + "Campaign editor opt_ins values must be booleans" + ) + return copy.deepcopy(value) + + +def _is_valid_field_override(key: Any, value: Any) -> bool: + return ( + isinstance(key, str) + and bool(key.strip()) + and len(key) <= 256 + and isinstance(value, bool) + ) + + +def _validated_field_overrides(value: Any) -> dict[str, bool]: + if not isinstance(value, dict) or len(value) > 10_000: + raise CampaignMailProfileBoundaryError( + "Campaign editor field_overrides must be a bounded object" + ) + if any(not _is_valid_field_override(key, item) for key, item in value.items()): + raise CampaignMailProfileBoundaryError( + "Campaign editor field_overrides must map short field names to booleans" + ) + return copy.deepcopy(value) + + +def _validated_required_string(value: Any, *, max_length: int, error: str) -> str: + if not isinstance(value, str) or not value.strip() or len(value) > max_length: + raise CampaignMailProfileBoundaryError(error) + return value.strip() + + def validate_campaign_editor_state( value: dict[str, Any] | None, *, @@ -55,32 +96,17 @@ def validate_campaign_editor_state( 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() + result["created_from"] = _validated_required_string( + value["created_from"], + max_length=128, + error="Campaign editor created_from must be a short string", + ) 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) + result["opt_ins"] = _validated_opt_ins(value["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) + result["field_overrides"] = _validated_field_overrides( + value["field_overrides"] + ) if "review_send" in value: result["review_send"] = _validated_server_review_state(value["review_send"]) return result @@ -122,34 +148,62 @@ def campaign_editor_state_for_edit(value: Any) -> dict[str, Any]: return state +def _is_valid_reviewed_message_key(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) and len(value) <= 512 + + +def _validated_reviewed_message_keys(value: Any) -> list[str]: + if not isinstance(value, list) or len(value) > 100_000: + raise CampaignMailProfileBoundaryError( + "Campaign reviewed message keys are invalid" + ) + if any(not _is_valid_reviewed_message_key(key) for key in value): + raise CampaignMailProfileBoundaryError( + "Campaign reviewed message keys are invalid" + ) + return list(dict.fromkeys(key.strip() for key in value)) + + +def _validated_review_actor(value: Any) -> str | None: + if value is not None and (not isinstance(value, str) or len(value) > 256): + raise CampaignMailProfileBoundaryError("Campaign review actor is invalid") + return value + + 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") + 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") + validated_build_token = _validated_required_string( + build_token, + max_length=256, + error="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") + raise CampaignMailProfileBoundaryError( + "Campaign review completion state is invalid" + ) + validated_keys = _validated_reviewed_message_keys(keys) + validated_updated_at = _validated_required_string( + updated_at, + max_length=128, + error="Campaign review timestamp is invalid", + ) + validated_updated_by = _validated_review_actor(updated_by) return { - "build_token": build_token.strip(), + "build_token": validated_build_token, "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, + "reviewed_message_keys": validated_keys, + "updated_at": validated_updated_at, + "updated_by_user_id": validated_updated_by, }