fix(workflows): enforce sandbox sender save validation

posthog/api/integration.py

@@ -20,11 +20,11 @@
 from django_filters.rest_framework import DjangoFilterBackend
 from django_redis.cache import RedisCache
 from django_redis.exceptions import ConnectionInterrupted
-from drf_spectacular.utils import extend_schema, extend_schema_field, extend_schema_serializer
+from drf_spectacular.utils import OpenApiResponse, extend_schema, extend_schema_field, extend_schema_serializer
 from prometheus_client import Counter
 from redis.exceptions import RedisError
 from rest_framework import mixins, serializers, status, viewsets
-from rest_framework.exceptions import APIException, PermissionDenied, Throttled, ValidationError
+from rest_framework.exceptions import APIException, NotFound, PermissionDenied, Throttled, ValidationError
 from rest_framework.permissions import BasePermission, IsAuthenticated
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -125,7 +125,11 @@
 from products.slack_app.backend.services.slack_auth import SLACK_AUTH_FAILURE_CODES
 from products.tasks.backend.facade.api import get_in_progress_runs_for_github_integration
 from products.tasks.backend.facade.contracts import InProgressGithubRunsDTO
-from products.workflows.backend.facade.api import get_active_workflows_using_integration
+from products.workflows.backend.facade.api import (
+    SandboxSenderUnavailable,
+    ensure_sandbox_email_sender,
+    get_active_workflows_using_integration,
+)
 
 logger = structlog.get_logger(__name__)
 
@@ -1381,6 +1385,7 @@ class IntegrationViewSet(
         "github_prepare_callback",
         "github_link_existing",
         "github_oauth_authorize",
+        "email_sandbox_sender",
         # Side-effecting POST (emails admins) — a read-only token must not be able to trigger it.
         "request_access",
     ]
@@ -1419,7 +1424,13 @@ def dangerously_get_permissions(self):
         ]
         # Adding an integration only requires project membership. Every edit and removal uses the
         # viewset permission class, including the creator exception for Google account removal.
-        if self.action in ("create", "github_link_existing", "github_oauth_authorize", "request_access"):
+        if self.action in (
+            "create",
+            "github_link_existing",
+            "github_oauth_authorize",
+            "request_access",
+            "email_sandbox_sender",
+        ):
             return base_permissions
         if self.action == "refresh_github_repos":
             return [*base_permissions, TeamMemberLightManagementPermission()]
@@ -1431,6 +1442,9 @@ def get_throttles(self):
         return super().get_throttles()
 
     def perform_destroy(self, instance: Integration) -> None:
+        if instance.kind == "email":
+            EmailIntegration(instance).ensure_user_managed()
+
         flows_using_integration = get_active_workflows_using_integration(
             team_id=instance.team_id, integration_id=instance.id
         )
@@ -2430,6 +2444,23 @@ def jira_projects(self, request: Request, *args: Any, **kwargs: Any) -> Response
         jira = JiraIntegration(instance)
         return Response({"projects": jira.list_projects()})
 
+    @extend_schema(
+        request=None,
+        responses={200: IntegrationSerializer, 404: OpenApiResponse(description="No sandbox sender for this project.")},
+    )
+    @action(methods=["POST"], detail=False, url_path="email_sandbox_sender")
+    def email_sandbox_sender(self, request: Request, *args: Any, **kwargs: Any) -> Response:
+        """Create or refresh the project's PostHog-managed sandbox email sender and return it."""
+        try:
+            sandbox_sender = ensure_sandbox_email_sender(self.team_id)
+        except SandboxSenderUnavailable:
+            raise NotFound("The sandbox sender is not available for this project.")
+        if sandbox_sender.created:
+            report_user_action(
+                cast(User, request.user), "workflows sandbox sender provisioned", team=self.team, request=request
+            )
+        return Response(IntegrationSerializer(sandbox_sender.integration, context=self.get_serializer_context()).data)
+
     @action(methods=["POST"], detail=True, url_path="email/verify")
     def email_verify(self, request, **kwargs):
         email = EmailIntegration(self.get_object())
@@ -2439,14 +2470,13 @@ def email_verify(self, request, **kwargs):
     @extend_schema(responses={200: IntegrationSerializer})
     @action(methods=["PATCH"], detail=True, url_path="email")
     def email_update(self, request, **kwargs) -> Response:
-        instance = self.get_object()
-        config = request.data.get("config", {})
+        email = EmailIntegration(self.get_object())
+        email.ensure_user_managed()
 
-        serializer = NativeEmailIntegrationSerializer(data=config)
+        serializer = NativeEmailIntegrationSerializer(data=request.data.get("config", {}))
         serializer.is_valid(raise_exception=True)
 
-        email = EmailIntegration(instance)
-        email.update_native_integration(serializer.validated_data, instance.team_id)
+        email.update_native_integration(serializer.validated_data, email.integration.team_id)
 
         return Response(IntegrationSerializer(email.integration).data)
 

posthog/cdp/test/test_validation.py

@@ -614,7 +614,7 @@ def _validate_email_from(self, from_value, existing_from=None, get_team=True, ca
         if get_team:
             context_extra["get_team"] = lambda: self.team
         if cache is not None:
-            context_extra["email_integration_domain_cache"] = cache
+            context_extra["email_integration_cache"] = cache
         return validate_inputs(inputs_schema, {"email": {"value": value}}, context_extra=context_extra)
 
     @parameterized.expand(

posthog/cdp/validation.py

@@ -25,7 +25,8 @@
     compile_filters_bytecode,
     compile_filters_expr,
 )
-from posthog.models.integration import POSTHOG_CONNECT_KIND, Integration
+from posthog.models.integration import POSTHOG_CONNECT_KIND, SANDBOX_EMAIL_PROVIDER, Integration
+from posthog.permissions import posthog_feature_flag_enabled
 
 from products.cdp.backend.models.hog_functions.hog_function import (
     TYPES_WITH_JAVASCRIPT_SOURCE,
@@ -72,13 +73,95 @@ def masked_secret_input_keys(stored_inputs: object) -> list[str]:
 
 
 def _sender_integration_ids(from_value: dict) -> set[int]:
+    rotation = from_value.get("integrationIds")
     return {
-        integration_id
-        for integration_id in [from_value.get("integrationId"), *(from_value.get("integrationIds") or [])]
-        if isinstance(integration_id, int) and not isinstance(integration_id, bool)
+        int(integration_id)
+        for integration_id in [from_value.get("integrationId"), *(rotation if isinstance(rotation, list) else [])]
+        if isinstance(integration_id, (int, float))
+        and not isinstance(integration_id, bool)
+        and (not isinstance(integration_id, float) or integration_id.is_integer())
     }
 
 
+def _email_integrations(integration_ids: set[int], context: dict[str, Any]) -> dict[int, Integration | None]:
+    cache: dict[int, Integration | None] = context.setdefault("email_integration_cache", {})
+    missing_ids = integration_ids - cache.keys()
+    if missing_ids:
+        integrations = {
+            integration.id: integration
+            for integration in Integration.objects.filter(
+                team_id=context["get_team"]().id, id__in=missing_ids, kind="email"
+            )
+        }
+        cache.update({integration_id: integrations.get(integration_id) for integration_id in missing_ids})
+    return cache
+
+
+def validate_sandbox_email_sender(email_value: object, context: dict[str, Any]) -> None:
+    if not isinstance(email_value, dict) or not isinstance(email_value.get("from"), dict):
+        return
+    integration_ids = _sender_integration_ids(email_value["from"])
+    get_team = context.get("get_team")
+    if not integration_ids or get_team is None:
+        return
+    integrations = _email_integrations(integration_ids, context)
+    if not any(
+        (integration := integrations[integration_id]) is not None
+        and (integration.config or {}).get("provider") == SANDBOX_EMAIL_PROVIDER
+        for integration_id in integration_ids
+    ):
+        return
+    if context.get("workflow_origin_product") == "broadcasts":
+        raise serializers.ValidationError(
+            {"input": "The sandbox sender cannot be used in broadcasts. Select a sender on your own verified domain."}
+        )
+    is_test_send = getattr(context.get("view"), "action", None) == "invocations"
+    if context.get("workflow_action_type") != "function_email" and not is_test_send:
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender can only be used in workflow email steps and test sends. "
+                "Select a sender on your own verified domain for destinations."
+            }
+        )
+    for stored in context.get("existing_email_values") or []:
+        if (
+            isinstance(stored, dict)
+            and email_value["from"] == stored.get("from")
+            and (email_value.get("replyTo") or "") == (stored.get("replyTo") or "")
+        ):
+            return
+    if len(integration_ids) != 1:
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender must be the only sender. Remove the other senders "
+                "or select a sender on your own verified domain."
+            }
+        )
+    if email_value["from"].get("email") or email_value["from"].get("name") or email_value.get("replyTo"):
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender uses a fixed From address and name and does not support Reply-To. "
+                "Remove these overrides or select a sender on your own verified domain."
+            }
+        )
+    team = get_team()
+    flag_cache: dict[int, bool] = context.setdefault("sandbox_sender_enabled_cache", {})
+    if team.id not in flag_cache:
+        try:
+            flag_cache[team.id] = posthog_feature_flag_enabled(
+                "workflows-sandbox-sender", str(team.uuid), organization_id=team.organization_id, team_id=team.id
+            )
+        except Exception:
+            flag_cache[team.id] = False
+    if not flag_cache[team.id]:
+        raise serializers.ValidationError(
+            {
+                "input": "The sandbox sender is not available for this project. "
+                "Select a sender on your own verified domain."
+            }
+        )
+
+
 def _validate_not_posthog_connection(integration_ids: list[int], context: dict) -> None:
     # A PostHog connection acts as the user who created it, so only that user may use it. A function
     # runs for the whole team and the runtime never resolves one, so reject it where the author sees why.
@@ -137,20 +220,11 @@ def _validate_email_sender_override(from_value: dict, context: dict) -> None:
         return
 
     override_domain = override.split("@")[1].lower()
-    # An empty cached domain means the id resolved to no email integration for this team; the
-    # save is not blocked on it (there is no domain to compare), matching the uncached behavior.
-    shared_cache = context.get("email_integration_domain_cache")
-    domain_cache: dict[int, str] = shared_cache if isinstance(shared_cache, dict) else {}
-    missing_ids = [integration_id for integration_id in integration_ids if integration_id not in domain_cache]
-    if missing_ids:
-        for integration in Integration.objects.filter(team_id=get_team().id, id__in=missing_ids, kind="email"):
-            config = integration.config or {}
-            domain_cache[integration.id] = (config.get("domain") or (config.get("email") or "").split("@")[-1]).lower()
-        for integration_id in missing_ids:
-            domain_cache.setdefault(integration_id, "")
-
+    integrations = _email_integrations(integration_ids, context)
     for integration_id in sorted(integration_ids):
-        integration_domain = domain_cache.get(integration_id) or ""
+        integration = integrations[integration_id]
+        config = (integration.config or {}) if integration is not None else {}
+        integration_domain = (config.get("domain") or (config.get("email") or "").split("@")[-1]).lower()
         if integration_domain and override_domain != integration_domain:
             raise serializers.ValidationError(
                 {
@@ -751,6 +825,7 @@ def validate(self, attrs):
         elif item_type == "email" or item_type == "native_email":
             if not isinstance(value, dict):
                 raise serializers.ValidationError({"input": f"Value must be an email object."})
+            validate_sandbox_email_sender(value, self.context)
             # Report every missing key in one error: these objects are typically authored
             # programmatically, and a one-at-a-time raise forces a round trip per missing key.
             missing = [f"'{key_}'" for key_ in ("from", "to", "subject") if not value.get(key_)]

posthog/models/integration/__init__.py

@@ -31,7 +31,12 @@
 from .clickup import ClickUpIntegration
 from .common import ERROR_TOKEN_REFRESH_FAILED, META_GRAPH_API_VERSION, IntegrationError, dot_get
 from .databricks import DatabricksIntegration, DatabricksIntegrationError
-from .email import EmailIntegration, cleanup_ses_identity_on_integration_delete
+from .email import (
+    SANDBOX_EMAIL_INTEGRATION_ID,
+    SANDBOX_EMAIL_PROVIDER,
+    EmailIntegration,
+    cleanup_ses_identity_on_integration_delete,
+)
 from .external_issues import (
     SUPPORTED_EXTERNAL_ISSUE_PROVIDERS,
     external_issue_url,
@@ -204,6 +209,8 @@
     "ClickUpIntegration",
     "EmailIntegration",
     "cleanup_ses_identity_on_integration_delete",
+    "SANDBOX_EMAIL_INTEGRATION_ID",
+    "SANDBOX_EMAIL_PROVIDER",
     "LinearIntegration",
     "SUPPORTED_EXTERNAL_ISSUE_PROVIDERS",
     "external_issue_url",

posthog/models/integration/email.py

@@ -20,6 +20,16 @@
     from products.workflows.backend.facade.contracts import EmailDomainVerification
 
 
+SANDBOX_EMAIL_PROVIDER = "sandbox"
+SANDBOX_EMAIL_INTEGRATION_ID = "posthog-sandbox"
+SANDBOX_SENDER_MANAGED_MESSAGE = "The sandbox sender is managed by PostHog."
+
+
+def is_sandbox_sender_domain(domain: str) -> bool:
+    sandbox_domain = settings.WORKFLOWS_SANDBOX_SENDER_DOMAIN
+    return bool(sandbox_domain) and (domain == sandbox_domain or domain.endswith(f".{sandbox_domain}"))
+
+
 class EmailIntegration:
     integration: model.Integration
 
@@ -28,6 +38,14 @@ def __init__(self, integration: model.Integration) -> None:
             raise Exception("EmailIntegration init called with Integration with wrong 'kind'")
         self.integration = integration
 
+    @property
+    def is_sandbox_sender(self) -> bool:
+        return self.integration.config.get("provider") == SANDBOX_EMAIL_PROVIDER
+
+    def ensure_user_managed(self) -> None:
+        if self.is_sandbox_sender:
+            raise ValidationError(SANDBOX_SENDER_MANAGED_MESSAGE)
+
     @classmethod
     def create_native_integration(
         cls, config: dict, team_id: int, organization_id: str, created_by: User | None = None
@@ -41,6 +59,11 @@ def create_native_integration(
         if domain in free_email_domains_list or domain in disposable_email_domains_list:
             raise ValidationError(f"Email domain {domain} is not supported. Please use a custom domain.")
 
+        if is_sandbox_sender_domain(domain):
+            raise ValidationError(
+                f"Email domain {domain} belongs to the PostHog sandbox sender. Please use a domain you own."
+            )
+
         # Check if any other integration already exists in a different team with the same domain,
         # if so, ensure this team is part of the same organization. If not, we block creation.
         same_domain_integrations = model.Integration.objects.filter(kind="email", config__domain=domain)
@@ -123,6 +146,7 @@ def update_native_integration(self, config: dict, team_id: int) -> model.Integra
         return self.integration
 
     def verify(self) -> "EmailDomainVerification":
+        self.ensure_user_managed()
         domain = self.integration.config.get("domain")
         provider = self.integration.config.get("provider", "ses")
         mail_from_subdomain = self.integration.config.get("mail_from_subdomain", "feedback")

posthog/settings/ses.py

@@ -29,6 +29,13 @@
     if cs.strip()
 ]
 
+# The PostHog-owned sandbox sender, which lets a project send email before it verifies a domain.
+# Empty means not provisioned: projects get no sandbox sender row.
+# The sandbox tenant keeps its own configuration set, which must not join SES_TENANT_CONFIGURATION_SETS.
+WORKFLOWS_SANDBOX_SENDER_DOMAIN: str = os.getenv("WORKFLOWS_SANDBOX_SENDER_DOMAIN", "").strip().lower()
+WORKFLOWS_SANDBOX_SENDER_FROM_ADDRESS: str = os.getenv("WORKFLOWS_SANDBOX_SENDER_FROM_ADDRESS", "").strip().lower()
+SES_SANDBOX_TENANT_NAME: str = os.getenv("SES_SANDBOX_TENANT_NAME", "").strip()
+
 # SNS topics allowed to deliver SES tenant reputation events (EventBridge -> SNS -> webhook).
 # Empty (the default) leaves the webhook inert: the SNS signature proves a message came from AWS,
 # but only the allowlist proves it came from *our* topic.

products/integrations/mcp/tools.yaml

@@ -94,6 +94,9 @@ tools:
     integrations-email-partial-update:
         operation: integrations_email_partial_update
         enabled: false
+    integrations-email-sandbox-sender-create:
+        operation: integrations_email_sandbox_sender_create
+        enabled: false
     integrations-email-verify-create:
         operation: integrations_email_verify_create
         enabled: false

products/workflows/backend/facade/api.py

@@ -29,6 +29,7 @@
     unsuspend_email_sending,
 )
 from products.workflows.backend.services.integration_usage import get_active_hog_flows_using_integration
+from products.workflows.backend.services.sandbox_sender import SandboxSenderUnavailable, ensure_sandbox_email_sender
 from products.workflows.backend.services.template_input_usage import (
     filter_hog_flow_references_by_access_level,
     get_hog_flows_referencing_template_input_keys,
@@ -42,8 +43,10 @@
 
 __all__ = [
     "MIN_EMAIL_SENDING_TIER",
+    "SandboxSenderUnavailable",
     "compute_next_occurrences",
     "create_batch_job",
+    "ensure_sandbox_email_sender",
     "ensure_workflows_config",
     "filter_hog_flow_references_by_access_level",
     "get_email_sending_state",