diff --git a/posthog/settings/ses.py b/posthog/settings/ses.py
index 91f636f4289a..a0438d852078 100644
--- a/posthog/settings/ses.py
+++ b/posthog/settings/ses.py
@@ -13,3 +13,10 @@
SES_SECRET_ACCESS_KEY = os.getenv("SES_SECRET_ACCESS_KEY", "") or None
SES_REGION = os.getenv("SES_REGION", "us-east-1")
+
+# 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.
+WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS: list[str] = [
+ arn.strip() for arn in os.getenv("WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS", "").split(",") if arn.strip()
+]
diff --git a/posthog/tasks/email.py b/posthog/tasks/email.py
index 23672c995a70..239f50213b8c 100644
--- a/posthog/tasks/email.py
+++ b/posthog/tasks/email.py
@@ -631,6 +631,41 @@ def send_email_sending_suspended(team_id: int, reason: str, suspended_at: str) -
message.send()
+@shared_task(**EMAIL_TASK_KWARGS)
+@with_team_scope()
+def send_email_sending_reputation_finding(team_id: int, impact: str, found_at: str) -> None:
+ """
+ Warn a team's admins that our email provider raised reputation findings against their
+ project's sending. LOW impact is a fix-this warning; HIGH impact means sending can be
+ paused automatically. Not gated by notification settings — inaction escalates to a pause.
+ """
+ if not is_email_available(with_absolute_urls=True):
+ return
+ team = Team.objects.get(id=team_id)
+ memberships_to_email = _get_project_admins_to_notify_of_email_sending_suspension(team)
+ if not memberships_to_email:
+ return
+ high_impact = impact == "HIGH"
+ subject = (
+ f"[Action required] Email sending for project '{team}' is at risk of being paused"
+ if high_impact
+ else f"Reputation warning for email sending in project '{team}'"
+ )
+ message = EmailMessage(
+ campaign_key=f"email_sending_reputation_finding_{team_id}_{impact}_{found_at}",
+ subject=subject,
+ template_name="email_sending_reputation_finding",
+ template_context={
+ "team": team,
+ "high_impact": high_impact,
+ "reputation_path": f"/project/{team.id}/workflows/reputation",
+ },
+ )
+ for membership in memberships_to_email:
+ message.add_user_recipient(membership.user)
+ message.send()
+
+
@shared_task(**EMAIL_TASK_KWARGS)
@with_team_scope()
def send_email_sending_unsuspended(team_id: int, unsuspended_at: str) -> None:
diff --git a/posthog/tasks/scheduled.py b/posthog/tasks/scheduled.py
index a77cb77fc9dd..2fc20cd96494 100644
--- a/posthog/tasks/scheduled.py
+++ b/posthog/tasks/scheduled.py
@@ -116,6 +116,7 @@
reap_stale_prewarm_heatmaps,
report_stuck_heatmap_screenshots,
)
+from products.workflows.backend.tasks.ses_tenant_state import reconcile_ses_tenant_states
TWENTY_FOUR_HOURS = 24 * 60 * 60
@@ -236,6 +237,15 @@ def setup_periodic_tasks(sender: Celery, **kwargs: Any) -> None:
name="team metadata expiry tracking cleanup",
)
+ # SES tenant reputation reconciliation - daily at 6:30 AM UTC. EventBridge events are the
+ # real-time path; this sweep catches missed deliveries. Sequential SES API calls per team
+ # with an SES email integration, so kept daily to stay well inside SES API rate limits.
+ sender.add_periodic_task(
+ crontab(hour="6", minute="30"),
+ reconcile_ses_tenant_states.s(),
+ name="ses tenant reputation reconciliation",
+ )
+
# LLM gateway policy cache sync - hourly at :05 to stagger from team_metadata at :00
sender.add_periodic_task(
crontab(hour="*", minute="5"),
diff --git a/posthog/templates/email/email_sending_reputation_finding.html b/posthog/templates/email/email_sending_reputation_finding.html
new file mode 100644
index 000000000000..8b877ff93fd3
--- /dev/null
+++ b/posthog/templates/email/email_sending_reputation_finding.html
@@ -0,0 +1,38 @@
+{% extends "email/base.html" %} {% load posthog_assets %}
+{% block heading %}{% if high_impact %}Email sending is at risk of being paused{% else %}Reputation warning for your email sending{% endif %}{% endblock %}
+{% block section %}
+
+
+ Our email provider has raised {% if high_impact %}high-impact{% else %}low-impact{% endif %} reputation findings against workflow email sending in project {{ team }}.
+
+{% if high_impact %}
+
+ At this level, email sending for your project can be paused automatically to protect deliverability. Act now to avoid an interruption.
+
+{% else %}
+
+ Sending still works, but if these findings are not fixed they can escalate and email sending for your project can be paused.
+
+{% endif %}
+
+ Open the Reputation tab to see each finding with the provider's recommended fix, and which workflows have the highest bounce or spam complaint rates. Typical fixes: remove old or purchased addresses from your lists, and stop sending to recipients who never engage.
+
+
+
+{% endblock %}
+
+{% block footer %}
+Need help?
+Contact support
+or
+read our documentation.
+{% endblock %}
diff --git a/posthog/urls.py b/posthog/urls.py
index d8d7eafae097..0ebbebc7efcf 100644
--- a/posthog/urls.py
+++ b/posthog/urls.py
@@ -82,6 +82,7 @@
)
from products.warehouse_sources.backend.presentation.views.public_source_configs import PublicSourceConfigViewSet
from products.workflows.backend.api import hog_flow, hog_flow_template
+from products.workflows.backend.api.ses_events_webhook import ses_tenant_events_webhook
from .utils import opt_slash_path, render_template
from .views import (
@@ -675,6 +676,8 @@ def authorize_and_redirect(request: HttpRequest) -> HttpResponse:
opt_slash_path("webhooks/github", github_webhook),
# Stamphog runs as its own GitHub App with a dedicated inbound endpoint (not the fan-out above)
opt_slash_path("webhooks/stamphog/github", stamphog_github_webhook),
+ # AWS SES tenant reputation events (EventBridge -> SNS HTTPS subscription)
+ opt_slash_path("webhooks/workflows/ses-events", ses_tenant_events_webhook),
# Message preferences
path("messaging-preferences//", preferences_page, name="message_preferences"),
opt_slash_path("messaging-preferences/update", update_preferences, name="message_preferences_update"),
diff --git a/products/workflows/backend/api/ses_events_webhook.py b/products/workflows/backend/api/ses_events_webhook.py
new file mode 100644
index 000000000000..34801f52785b
--- /dev/null
+++ b/products/workflows/backend/api/ses_events_webhook.py
@@ -0,0 +1,106 @@
+"""
+Inbound webhook for AWS SES tenant reputation events, delivered via EventBridge → SNS HTTPS
+subscription. Events are treated as change signals only — the handler never trusts the payload's
+state; it enqueues a sync that reads the authoritative tenant state from the SES API. That makes
+duplicate, reordered, or partially-shaped events harmless.
+
+Authentication is the SNS message signature (proves "from AWS") plus a topic-ARN allowlist
+(proves "from our topic"). AWS-side wiring: an EventBridge rule on `source = aws.ses` for the
+tenant detail-types, targeting an SNS topic with an HTTPS subscription to this endpoint.
+"""
+
+import re
+import json
+from typing import Any
+
+from django.conf import settings
+from django.http import HttpRequest, HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+import requests
+import structlog
+
+from products.workflows.backend.services.sns_verification import is_valid_sns_url, verify_sns_message
+from products.workflows.backend.tasks.ses_tenant_state import sync_ses_tenant_state_task
+
+logger = structlog.get_logger(__name__)
+
+_TENANT_TEAM_RE = re.compile(r"\bteam-(\d+)\b")
+
+
+def _extract_team_id(event: dict[str, Any]) -> int | None:
+ """Pull the `team-` tenant name out of an EventBridge event, wherever AWS put it."""
+ detail = event.get("detail") or {}
+ candidates: list[Any] = [
+ detail.get("tenantName"),
+ detail.get("tenant-name"),
+ detail.get("reputationEntityReference"),
+ ]
+ resources = event.get("resources")
+ if isinstance(resources, list):
+ candidates.extend(resources)
+ for candidate in candidates:
+ if isinstance(candidate, str) and (match := _TENANT_TEAM_RE.search(candidate)):
+ return int(match.group(1))
+ return None
+
+
+@csrf_exempt
+def ses_tenant_events_webhook(request: HttpRequest) -> HttpResponse:
+ """Verify an SNS delivery and enqueue a tenant-state sync for the affected team."""
+ if request.method != "POST":
+ return HttpResponse(status=405)
+
+ allowed_topics = {arn for arn in settings.WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS if arn}
+ if not allowed_topics:
+ logger.error("ses_tenant_events_webhook_not_configured")
+ return HttpResponse("Webhook not configured", status=500)
+
+ try:
+ # RecursionError: deeply nested JSON from an unauthenticated caller must 400, not 500.
+ message = json.loads(request.body)
+ except (json.JSONDecodeError, UnicodeDecodeError, RecursionError):
+ return HttpResponse("Invalid payload", status=400)
+ if not isinstance(message, dict):
+ return HttpResponse("Invalid payload", status=400)
+
+ if message.get("TopicArn") not in allowed_topics:
+ logger.warning("ses_tenant_events_webhook_unknown_topic", topic=message.get("TopicArn"))
+ return HttpResponse("Unknown topic", status=403)
+ if not verify_sns_message(message):
+ logger.warning("ses_tenant_events_webhook_invalid_signature", message_id=message.get("MessageId"))
+ return HttpResponse("Invalid signature", status=403)
+
+ message_type = message.get("Type")
+ if message_type == "SubscriptionConfirmation":
+ subscribe_url = message.get("SubscribeURL")
+ if not isinstance(subscribe_url, str) or not is_valid_sns_url(subscribe_url):
+ return HttpResponse("Invalid subscribe URL", status=400)
+ try:
+ requests.get(subscribe_url, timeout=5).raise_for_status()
+ except requests.RequestException:
+ # Non-2xx makes SNS retry the confirmation later.
+ logger.exception("ses_tenant_events_webhook_subscription_confirm_failed", topic=message.get("TopicArn"))
+ return HttpResponse("Subscription confirmation failed", status=502)
+ logger.info("ses_tenant_events_webhook_subscription_confirmed", topic=message.get("TopicArn"))
+ return HttpResponse(status=200)
+
+ if message_type != "Notification":
+ # UnsubscribeConfirmation and anything unknown: ack so SNS stops retrying.
+ return HttpResponse(status=200)
+
+ try:
+ event = json.loads(message.get("Message", ""))
+ except (json.JSONDecodeError, TypeError):
+ return HttpResponse(status=200)
+ if not isinstance(event, dict) or event.get("source") != "aws.ses":
+ return HttpResponse(status=200)
+
+ team_id = _extract_team_id(event)
+ if team_id is None:
+ logger.warning("ses_tenant_events_webhook_no_tenant", detail_type=event.get("detail-type"))
+ return HttpResponse(status=200)
+
+ # Ack fast; the sync fetches authoritative state and sends any transition emails.
+ sync_ses_tenant_state_task.delay(team_id)
+ return HttpResponse(status=202)
diff --git a/products/workflows/backend/migrations/0017_teamworkflowsconfig_ses_tenant_state.py b/products/workflows/backend/migrations/0017_teamworkflowsconfig_ses_tenant_state.py
new file mode 100644
index 000000000000..4b71acbc94d4
--- /dev/null
+++ b/products/workflows/backend/migrations/0017_teamworkflowsconfig_ses_tenant_state.py
@@ -0,0 +1,27 @@
+# Generated by Django 5.2.14 on 2026-07-30 08:27
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ("workflows", "0016_drop_emailreputationsnapshot_fk"),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name="teamworkflowsconfig",
+ name="ses_tenant_reputation_impact",
+ field=models.CharField(blank=True, db_default="", default="", max_length=8),
+ ),
+ migrations.AddField(
+ model_name="teamworkflowsconfig",
+ name="ses_tenant_sending_status",
+ field=models.CharField(blank=True, db_default="", default="", max_length=16),
+ ),
+ migrations.AddField(
+ model_name="teamworkflowsconfig",
+ name="ses_tenant_state_synced_at",
+ field=models.DateTimeField(blank=True, null=True),
+ ),
+ ]
diff --git a/products/workflows/backend/migrations/max_migration.txt b/products/workflows/backend/migrations/max_migration.txt
index 92b4c0e3895c..65a8d79947f0 100644
--- a/products/workflows/backend/migrations/max_migration.txt
+++ b/products/workflows/backend/migrations/max_migration.txt
@@ -1 +1 @@
-0016_drop_emailreputationsnapshot_fk
+0017_teamworkflowsconfig_ses_tenant_state
diff --git a/products/workflows/backend/models/team_workflows_config.py b/products/workflows/backend/models/team_workflows_config.py
index e13c994ae0ea..2c3b2162be31 100644
--- a/products/workflows/backend/models/team_workflows_config.py
+++ b/products/workflows/backend/models/team_workflows_config.py
@@ -35,5 +35,13 @@ class TeamWorkflowsConfig(models.Model):
email_sending_suspended_at = models.DateTimeField(null=True, blank=True)
email_sending_suspension_reason = models.TextField(blank=True, default="")
+ # Last-known state of the team's AWS SES tenant, mirrored so state *changes* can trigger
+ # customer emails exactly once (EventBridge events are best-effort; a periodic sweep
+ # reconciles). Empty string = never synced; the first sync sets a baseline without notifying.
+ # db_default so raw INSERTs from non-Django writers keep working.
+ ses_tenant_sending_status = models.CharField(max_length=16, blank=True, default="", db_default="")
+ ses_tenant_reputation_impact = models.CharField(max_length=8, blank=True, default="", db_default="")
+ ses_tenant_state_synced_at = models.DateTimeField(null=True, blank=True)
+
register_team_extension_signal(TeamWorkflowsConfig, logger=logger)
diff --git a/products/workflows/backend/services/ses_tenant_state.py b/products/workflows/backend/services/ses_tenant_state.py
new file mode 100644
index 000000000000..2d7657cd1371
--- /dev/null
+++ b/products/workflows/backend/services/ses_tenant_state.py
@@ -0,0 +1,108 @@
+from collections.abc import Callable
+
+from django.db import transaction
+from django.utils import timezone
+
+import structlog
+
+from posthog.tasks.email import (
+ send_email_sending_reputation_finding,
+ send_email_sending_suspended,
+ send_email_sending_unsuspended,
+)
+
+from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig
+from products.workflows.backend.providers.ses import SESProvider
+
+logger = structlog.get_logger(__name__)
+
+PROVIDER_PAUSE_REASON = "Our email provider paused sending for this project because of high-impact reputation findings"
+
+_IMPACT_SEVERITY = {"": 0, "LOW": 1, "HIGH": 2}
+
+_STATE_FIELDS = ["ses_tenant_sending_status", "ses_tenant_reputation_impact", "ses_tenant_state_synced_at"]
+
+
+def sync_ses_tenant_state(team_id: int) -> None:
+ """
+ Fetch the authoritative AWS SES tenant state for a team and apply it. Called from the
+ EventBridge webhook (events only say "something changed" — the API is the source of truth)
+ and from the periodic reconciliation sweep (event delivery is best-effort).
+ """
+ tenant = SESProvider().get_tenant_reputation(team_id)
+ if tenant is None:
+ return
+ apply_ses_tenant_state(
+ team_id, sending_status=tenant["sending_status"], reputation_impact=tenant["reputation_impact"]
+ )
+
+
+def apply_ses_tenant_state(team_id: int, *, sending_status: str, reputation_impact: str | None) -> None:
+ """
+ Persist the tenant state and email the project's admins on meaningful transitions:
+ sending paused, sending re-enabled, or reputation findings escalating. The stored state is
+ the dedupe mechanism — the row is locked for the read-compare-write so overlapping syncs
+ (webhook + sweep, or duplicate events) serialize, and the loser sees an unchanged state.
+ """
+ impact = reputation_impact or ""
+ TeamWorkflowsConfig.objects.get_or_create(team_id=team_id)
+
+ with transaction.atomic():
+ config = TeamWorkflowsConfig.objects.select_for_update().get(team_id=team_id)
+ previous_status = config.ses_tenant_sending_status
+ previous_impact = config.ses_tenant_reputation_impact
+ if previous_status == sending_status and previous_impact == impact:
+ config.ses_tenant_state_synced_at = timezone.now()
+ config.save(update_fields=["ses_tenant_state_synced_at"])
+ return
+
+ now = timezone.now()
+ config.ses_tenant_sending_status = sending_status
+ config.ses_tenant_reputation_impact = impact
+ config.ses_tenant_state_synced_at = now
+ config.save(update_fields=_STATE_FIELDS)
+ logger.info(
+ "SES tenant state changed",
+ team_id=team_id,
+ from_status=previous_status,
+ to_status=sending_status,
+ from_impact=previous_impact,
+ to_impact=impact,
+ )
+
+ notify = _pick_notification(
+ previous_status=previous_status,
+ previous_impact=previous_impact,
+ sending_status=sending_status,
+ impact=impact,
+ team_id=team_id,
+ now_iso=now.isoformat(),
+ )
+ if notify is not None:
+ # Dispatch after commit so a rollback can't leave an email claiming a state
+ # that was never persisted.
+ transaction.on_commit(notify)
+
+
+def _pick_notification(
+ *, previous_status: str, previous_impact: str, sending_status: str, impact: str, team_id: int, now_iso: str
+) -> Callable[[], None] | None:
+ if previous_status == "":
+ # First sync is baseline adoption, not a transition — healthy teams stay silent so
+ # rollout doesn't mass-email. But a tenant that is ALREADY paused or already carries
+ # high-impact findings is exactly who this feature exists to tell, so those do notify.
+ if sending_status == "DISABLED":
+ return lambda: send_email_sending_suspended.delay(team_id, PROVIDER_PAUSE_REASON, now_iso)
+ if impact == "HIGH":
+ return lambda: send_email_sending_reputation_finding.delay(team_id, impact, now_iso)
+ return None
+
+ if sending_status == "DISABLED" and previous_status != "DISABLED":
+ return lambda: send_email_sending_suspended.delay(team_id, PROVIDER_PAUSE_REASON, now_iso)
+ if previous_status == "DISABLED" and sending_status in ("ENABLED", "REINSTATED"):
+ return lambda: send_email_sending_unsuspended.delay(team_id, now_iso)
+ if sending_status != "DISABLED" and _IMPACT_SEVERITY.get(impact, 0) > _IMPACT_SEVERITY.get(previous_impact, 0):
+ # Escalation only, and only while still sending — a paused tenant already got the
+ # suspension email, and de-escalations don't need action.
+ return lambda: send_email_sending_reputation_finding.delay(team_id, impact, now_iso)
+ return None
diff --git a/products/workflows/backend/services/sns_verification.py b/products/workflows/backend/services/sns_verification.py
new file mode 100644
index 000000000000..9fc8dda146e7
--- /dev/null
+++ b/products/workflows/backend/services/sns_verification.py
@@ -0,0 +1,107 @@
+import re
+import base64
+import logging
+from typing import Any
+from urllib.parse import urlparse
+
+from django.core.cache import cache
+
+import requests
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
+from cryptography.x509 import load_pem_x509_certificate
+
+logger = logging.getLogger(__name__)
+
+# The signing cert must come from SNS itself — a signature check against an attacker-supplied cert
+# proves nothing. Mirrors the cert-URL validation in the Node SES webhook
+# (nodejs/src/cdp/services/messaging/helpers/ses.ts).
+_SNS_HOST_RE = re.compile(r"^sns\.[a-z0-9-]+\.amazonaws\.com$")
+
+_CERT_CACHE_SECONDS = 60 * 60
+_FETCH_TIMEOUT_SECONDS = 5
+
+# Per https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html the string to
+# sign is "Key\nValue\n" pairs in this exact order, skipping absent keys.
+_SIGNED_KEYS_BY_TYPE = {
+ "Notification": ["Message", "MessageId", "Subject", "Timestamp", "TopicArn", "Type"],
+ "SubscriptionConfirmation": ["Message", "MessageId", "SubscribeURL", "Timestamp", "Token", "TopicArn", "Type"],
+ "UnsubscribeConfirmation": ["Message", "MessageId", "SubscribeURL", "Timestamp", "Token", "TopicArn", "Type"],
+}
+
+
+def is_valid_sns_url(url: str | None) -> bool:
+ """True when the URL is HTTPS on a real *.amazonaws.com SNS host (cert or subscribe URL)."""
+ if not url:
+ return False
+ parsed = urlparse(url)
+ return parsed.scheme == "https" and bool(parsed.hostname) and bool(_SNS_HOST_RE.match(parsed.hostname or ""))
+
+
+def _fetch_signing_cert(cert_url: str) -> bytes | None:
+ cache_key = f"sns_signing_cert_{cert_url}"
+ cached = cache.get(cache_key)
+ if cached is not None:
+ return cached
+ try:
+ response = requests.get(cert_url, timeout=_FETCH_TIMEOUT_SECONDS)
+ response.raise_for_status()
+ except requests.RequestException:
+ logger.exception("Failed to fetch SNS signing certificate", extra={"cert_url": cert_url})
+ return None
+ cache.set(cache_key, response.content, _CERT_CACHE_SECONDS)
+ return response.content
+
+
+def _string_to_sign(message: dict[str, Any]) -> str | None:
+ keys = _SIGNED_KEYS_BY_TYPE.get(message.get("Type", ""))
+ if keys is None:
+ return None
+ parts = []
+ for key in keys:
+ value = message.get(key)
+ if value is None:
+ continue
+ parts.append(f"{key}\n{value}\n")
+ return "".join(parts)
+
+
+def verify_sns_message(message: dict[str, Any]) -> bool:
+ """
+ Verify an SNS message's authenticity: signing cert served by SNS over HTTPS, RSA signature over
+ the canonical string-to-sign. Returns False (never raises) on any mismatch so callers fail
+ closed. Signature proves "from AWS SNS" — callers must still check TopicArn against an
+ allowlist to prove "from *our* topic".
+ """
+ # Only SignatureVersion 2 (SHA256) is accepted. Version 1 signs with SHA1, which is not
+ # collision resistant — and since we own the SNS topic, we simply configure it with
+ # SignatureVersion=2 (a one-time SetTopicAttributes call, in the rollout runbook) instead of
+ # ever verifying SHA1 here.
+ if message.get("SignatureVersion") != "2":
+ return False
+ cert_url = message.get("SigningCertURL")
+ if not isinstance(cert_url, str) or not is_valid_sns_url(cert_url):
+ return False
+ string_to_sign = _string_to_sign(message)
+ if string_to_sign is None:
+ return False
+ try:
+ signature = base64.b64decode(message.get("Signature", ""))
+ except (ValueError, TypeError):
+ return False
+ cert_pem = _fetch_signing_cert(cert_url)
+ if cert_pem is None:
+ return False
+ try:
+ public_key = load_pem_x509_certificate(cert_pem).public_key()
+ if not isinstance(public_key, RSAPublicKey):
+ return False
+ public_key.verify(signature, string_to_sign.encode(), padding.PKCS1v15(), hashes.SHA256())
+ return True
+ except InvalidSignature:
+ return False
+ except Exception:
+ logger.exception("SNS signature verification errored")
+ return False
diff --git a/products/workflows/backend/tasks/ses_tenant_state.py b/products/workflows/backend/tasks/ses_tenant_state.py
new file mode 100644
index 000000000000..75d35ee97a89
--- /dev/null
+++ b/products/workflows/backend/tasks/ses_tenant_state.py
@@ -0,0 +1,52 @@
+from celery import shared_task
+from structlog import get_logger
+
+from posthog.models.integration import Integration
+from posthog.scoping_audit import skip_team_scope_audit
+from posthog.tasks.utils import CeleryQueue
+
+from products.workflows.backend.services.ses_tenant_state import sync_ses_tenant_state
+
+logger = get_logger(__name__)
+
+
+@shared_task(
+ ignore_result=True,
+ queue=CeleryQueue.DEFAULT.value,
+ # The sync hits the SES API; transient throttling/outages should retry with backoff instead
+ # of dropping the event (the daily sweep would catch it, but a day late).
+ autoretry_for=(Exception,),
+ retry_backoff=60,
+ retry_backoff_max=600,
+ max_retries=5,
+ retry_jitter=True,
+)
+def sync_ses_tenant_state_task(team_id: int) -> None:
+ """Webhook-triggered: an EventBridge event said this team's tenant changed — fetch and apply."""
+ sync_ses_tenant_state(team_id)
+
+
+@shared_task(ignore_result=True, queue=CeleryQueue.LONG_RUNNING.value)
+@skip_team_scope_audit
+def reconcile_ses_tenant_states() -> None:
+ """
+ Periodic backstop: EventBridge delivery is best-effort, so sweep every team that has an SES
+ tenant (i.e. a verified email integration) and apply the authoritative state. The transition
+ logic dedupes against stored state, so overlap with webhook-triggered syncs is harmless.
+ """
+ team_ids = (
+ Integration.objects.filter(kind="email", config__provider="ses")
+ .values_list("team_id", flat=True)
+ .distinct()
+ .order_by("team_id")
+ )
+ synced = 0
+ failed = 0
+ for team_id in team_ids.iterator(chunk_size=500):
+ try:
+ sync_ses_tenant_state(team_id)
+ synced += 1
+ except Exception:
+ failed += 1
+ logger.exception("SES tenant reconciliation failed for team", team_id=team_id)
+ logger.info("SES tenant reconciliation finished", synced=synced, failed=failed)
diff --git a/products/workflows/backend/test/test_ses_events_webhook.py b/products/workflows/backend/test/test_ses_events_webhook.py
new file mode 100644
index 000000000000..2212a0e2d667
--- /dev/null
+++ b/products/workflows/backend/test/test_ses_events_webhook.py
@@ -0,0 +1,131 @@
+import json
+from typing import Any
+
+from unittest.mock import patch
+
+from django.test import Client, TestCase, override_settings
+
+TOPIC = "arn:aws:sns:us-east-1:123456789012:ses-tenant-events"
+WEBHOOK_PATH = "/webhooks/workflows/ses-events"
+
+
+def _sns_notification(event: dict[str, Any], topic: str = TOPIC) -> dict[str, Any]:
+ return {
+ "Type": "Notification",
+ "MessageId": "mid-1",
+ "TopicArn": topic,
+ "Message": json.dumps(event),
+ "Timestamp": "2026-07-30T00:00:00.000Z",
+ "SignatureVersion": "1",
+ "Signature": "sig",
+ "SigningCertURL": "https://sns.us-east-1.amazonaws.com/cert.pem",
+ }
+
+
+def _eventbridge_event(**overrides: Any) -> dict[str, Any]:
+ event: dict[str, Any] = {
+ "version": "0",
+ "source": "aws.ses",
+ "detail-type": "Sending Status Disabled",
+ "resources": [],
+ "detail": {"tenantName": "team-42"},
+ }
+ event.update(overrides)
+ return event
+
+
+@override_settings(WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS=[TOPIC])
+class TestSesTenantEventsWebhook(TestCase):
+ def setUp(self):
+ self.client = Client()
+ verify = patch("products.workflows.backend.api.ses_events_webhook.verify_sns_message", return_value=True)
+ self.verify_mock = verify.start()
+ self.addCleanup(verify.stop)
+ sync = patch("products.workflows.backend.api.ses_events_webhook.sync_ses_tenant_state_task")
+ self.sync_mock = sync.start()
+ self.addCleanup(sync.stop)
+
+ def _post(self, payload: dict[str, Any]):
+ return self.client.post(WEBHOOK_PATH, data=json.dumps(payload), content_type="text/plain")
+
+ def test_enqueues_a_sync_for_the_tenant_named_in_the_event(self):
+ response = self._post(_sns_notification(_eventbridge_event()))
+
+ assert response.status_code == 202
+ self.sync_mock.delay.assert_called_once_with(42)
+
+ def test_finds_the_tenant_in_resource_arns_when_detail_has_no_name(self):
+ event = _eventbridge_event(detail={}, resources=["arn:aws:ses:us-east-1:123456789012:tenant/team-7/deadbeef"])
+
+ response = self._post(_sns_notification(event))
+
+ assert response.status_code == 202
+ self.sync_mock.delay.assert_called_once_with(7)
+
+ def test_acks_but_ignores_events_from_other_sources(self):
+ response = self._post(_sns_notification(_eventbridge_event(source="aws.health")))
+
+ assert response.status_code == 200
+ assert not self.sync_mock.delay.called
+
+ def test_rejects_messages_from_unknown_topics(self):
+ response = self._post(_sns_notification(_eventbridge_event(), topic="arn:aws:sns:us-east-1:999:other"))
+
+ assert response.status_code == 403
+ assert not self.sync_mock.delay.called
+
+ def test_rejects_messages_with_invalid_signatures(self):
+ self.verify_mock.return_value = False
+
+ response = self._post(_sns_notification(_eventbridge_event()))
+
+ assert response.status_code == 403
+ assert not self.sync_mock.delay.called
+
+ @override_settings(WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS=[])
+ def test_is_inert_when_no_topic_is_allowlisted(self):
+ response = self._post(_sns_notification(_eventbridge_event()))
+
+ assert response.status_code == 500
+ assert not self.sync_mock.delay.called
+
+ def test_confirms_subscriptions_by_fetching_the_subscribe_url(self):
+ subscribe_url = "https://sns.us-east-1.amazonaws.com/?Action=ConfirmSubscription&Token=tok"
+ payload = {
+ "Type": "SubscriptionConfirmation",
+ "MessageId": "mid-2",
+ "TopicArn": TOPIC,
+ "Token": "tok",
+ "Message": "You have chosen to subscribe...",
+ "SubscribeURL": subscribe_url,
+ "Timestamp": "2026-07-30T00:00:00.000Z",
+ "SignatureVersion": "1",
+ "Signature": "sig",
+ "SigningCertURL": "https://sns.us-east-1.amazonaws.com/cert.pem",
+ }
+
+ with patch("products.workflows.backend.api.ses_events_webhook.requests.get") as get_mock:
+ response = self._post(payload)
+
+ assert response.status_code == 200
+ get_mock.assert_called_once_with(subscribe_url, timeout=5)
+
+ def test_rejects_subscription_confirmations_pointing_off_aws(self):
+ payload = {
+ "Type": "SubscriptionConfirmation",
+ "MessageId": "mid-3",
+ "TopicArn": TOPIC,
+ "Token": "tok",
+ "Message": "...",
+ "SubscribeURL": "https://attacker.example.com/confirm",
+ "Timestamp": "2026-07-30T00:00:00.000Z",
+ "SignatureVersion": "1",
+ "Signature": "sig",
+ "SigningCertURL": "https://sns.us-east-1.amazonaws.com/cert.pem",
+ }
+
+ with patch("products.workflows.backend.api.ses_events_webhook.requests.get") as get_mock:
+ response = self._post(payload)
+
+ assert response.status_code == 400
+ assert not get_mock.called
diff --git a/products/workflows/backend/test/test_ses_tenant_state.py b/products/workflows/backend/test/test_ses_tenant_state.py
new file mode 100644
index 000000000000..82db85e7b149
--- /dev/null
+++ b/products/workflows/backend/test/test_ses_tenant_state.py
@@ -0,0 +1,107 @@
+from posthog.test.base import BaseTest
+from unittest.mock import patch
+
+from parameterized import parameterized
+
+from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig
+from products.workflows.backend.services.ses_tenant_state import PROVIDER_PAUSE_REASON, apply_ses_tenant_state
+
+
+class TestApplySesTenantState(BaseTest):
+ def _apply(self, sending_status: str, reputation_impact: str | None) -> dict:
+ with (
+ patch("products.workflows.backend.services.ses_tenant_state.send_email_sending_suspended") as suspended,
+ patch("products.workflows.backend.services.ses_tenant_state.send_email_sending_unsuspended") as unsuspended,
+ patch(
+ "products.workflows.backend.services.ses_tenant_state.send_email_sending_reputation_finding"
+ ) as finding,
+ self.captureOnCommitCallbacks(execute=True),
+ ):
+ apply_ses_tenant_state(self.team.id, sending_status=sending_status, reputation_impact=reputation_impact)
+ return {"suspended": suspended.delay, "unsuspended": unsuspended.delay, "finding": finding.delay}
+
+ def _seed(self, sending_status: str, reputation_impact: str = "") -> None:
+ TeamWorkflowsConfig.objects.update_or_create(
+ team=self.team,
+ defaults={
+ "ses_tenant_sending_status": sending_status,
+ "ses_tenant_reputation_impact": reputation_impact,
+ },
+ )
+
+ @parameterized.expand(
+ [
+ # A healthy or low-impact baseline is silent (rollout must not mass-email)...
+ ("healthy", "ENABLED", None, None),
+ ("low_findings", "ENABLED", "LOW", None),
+ # ...but a tenant that is already paused or already critical is exactly who this
+ # feature exists to tell — the first sync must not swallow that.
+ ("already_paused", "DISABLED", "HIGH", "suspended"),
+ ("already_critical", "ENABLED", "HIGH", "finding"),
+ ]
+ )
+ def test_first_sync_baseline_notifies_only_already_bad_tenants(
+ self, _name: str, sending_status: str, impact: str | None, expected_email: str | None
+ ):
+ emails = self._apply(sending_status, impact)
+
+ config = TeamWorkflowsConfig.objects.get(team=self.team)
+ assert config.ses_tenant_sending_status == sending_status
+ assert config.ses_tenant_state_synced_at is not None
+ for kind, mock in emails.items():
+ assert mock.called == (kind == expected_email), f"{kind} called={mock.called}"
+
+ def test_unchanged_state_only_refreshes_synced_at(self):
+ self._seed("ENABLED", "LOW")
+
+ emails = self._apply("ENABLED", "LOW")
+
+ assert all(not mock.called for mock in emails.values())
+ assert TeamWorkflowsConfig.objects.get(team=self.team).ses_tenant_state_synced_at is not None
+
+ @parameterized.expand(
+ [
+ # (previous_status, previous_impact, new_status, new_impact, expected_email)
+ ("pause", "ENABLED", "", "DISABLED", "HIGH", "suspended"),
+ ("pause_from_reinstated", "REINSTATED", "LOW", "DISABLED", "HIGH", "suspended"),
+ ("unpause", "DISABLED", "HIGH", "ENABLED", "", "unsuspended"),
+ ("reinstate", "DISABLED", "HIGH", "REINSTATED", "HIGH", "unsuspended"),
+ ("first_finding", "ENABLED", "", "ENABLED", "LOW", "finding"),
+ ("escalation", "ENABLED", "LOW", "ENABLED", "HIGH", "finding"),
+ ("deescalation", "ENABLED", "HIGH", "ENABLED", "LOW", None),
+ # Impact escalating while already paused: the suspension email already covers it
+ ("escalation_while_paused", "DISABLED", "LOW", "DISABLED", "HIGH", None),
+ ("finding_resolved", "ENABLED", "LOW", "ENABLED", "", None),
+ ]
+ )
+ def test_transitions_send_the_right_email(
+ self,
+ _name: str,
+ previous_status: str,
+ previous_impact: str,
+ new_status: str,
+ new_impact: str,
+ expected_email: str | None,
+ ):
+ self._seed(previous_status, previous_impact)
+
+ emails = self._apply(new_status, new_impact or None)
+
+ for kind, mock in emails.items():
+ assert mock.called == (kind == expected_email), f"{kind} called={mock.called}"
+
+ def test_provider_pause_email_carries_the_provider_reason(self):
+ self._seed("ENABLED")
+
+ emails = self._apply("DISABLED", "HIGH")
+
+ args = emails["suspended"].call_args.args
+ assert args[0] == self.team.id
+ assert args[1] == PROVIDER_PAUSE_REASON
+
+ def test_finding_email_carries_the_new_impact(self):
+ self._seed("ENABLED", "LOW")
+
+ emails = self._apply("ENABLED", "HIGH")
+
+ assert emails["finding"].call_args.args[:2] == (self.team.id, "HIGH")
diff --git a/products/workflows/backend/test/test_sns_verification.py b/products/workflows/backend/test/test_sns_verification.py
new file mode 100644
index 000000000000..31ff00cf4578
--- /dev/null
+++ b/products/workflows/backend/test/test_sns_verification.py
@@ -0,0 +1,84 @@
+import base64
+import datetime
+from typing import Any
+
+from unittest import TestCase
+from unittest.mock import patch
+
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.x509.oid import NameOID
+from parameterized import parameterized
+
+from products.workflows.backend.services.sns_verification import is_valid_sns_url, verify_sns_message
+
+_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+_SUBJECT = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "sns.amazonaws.com")])
+_CERT_PEM = (
+ x509.CertificateBuilder()
+ .subject_name(_SUBJECT)
+ .issuer_name(_SUBJECT)
+ .public_key(_KEY.public_key())
+ .serial_number(x509.random_serial_number())
+ .not_valid_before(datetime.datetime(2020, 1, 1))
+ .not_valid_after(datetime.datetime(2040, 1, 1))
+ .sign(_KEY, hashes.SHA256())
+ .public_bytes(serialization.Encoding.PEM)
+)
+
+
+def _signed_notification(tamper: dict[str, Any] | None = None) -> dict[str, Any]:
+ message = {
+ "Type": "Notification",
+ "MessageId": "mid-1",
+ "TopicArn": "arn:aws:sns:us-east-1:123456789012:topic",
+ "Message": '{"source":"aws.ses"}',
+ "Timestamp": "2026-07-30T00:00:00.000Z",
+ "SignatureVersion": "2",
+ "SigningCertURL": "https://sns.us-east-1.amazonaws.com/cert.pem",
+ }
+ string_to_sign = "".join(
+ f"{key}\n{message[key]}\n" for key in ["Message", "MessageId", "Timestamp", "TopicArn", "Type"]
+ )
+ signature = _KEY.sign(string_to_sign.encode(), padding.PKCS1v15(), hashes.SHA256())
+ message["Signature"] = base64.b64encode(signature).decode()
+ message.update(tamper or {})
+ return message
+
+
+class TestSnsVerification(TestCase):
+ def _verify(self, message: dict[str, Any]) -> bool:
+ with patch("products.workflows.backend.services.sns_verification._fetch_signing_cert", return_value=_CERT_PEM):
+ return verify_sns_message(message)
+
+ def test_accepts_a_correctly_signed_notification(self):
+ assert self._verify(_signed_notification()) is True
+
+ @parameterized.expand(
+ [
+ ("tampered_message", {"Message": '{"source":"aws.ses","evil":true}'}),
+ ("tampered_topic", {"TopicArn": "arn:aws:sns:us-east-1:666:other"}),
+ ("garbage_signature", {"Signature": base64.b64encode(b"nope").decode()}),
+ ("cert_not_from_sns", {"SigningCertURL": "https://attacker.example.com/cert.pem"}),
+ ("cert_over_http", {"SigningCertURL": "http://sns.us-east-1.amazonaws.com/cert.pem"}),
+ ("unknown_type", {"Type": "SomethingElse"}),
+ # SignatureVersion 1 is SHA1-signed and rejected outright; the topic must be
+ # configured with SignatureVersion=2
+ ("sha1_signature_version", {"SignatureVersion": "1"}),
+ ]
+ )
+ def test_rejects(self, _name: str, tamper: dict[str, Any]):
+ assert self._verify(_signed_notification(tamper)) is False
+
+ @parameterized.expand(
+ [
+ ("sns_regional", "https://sns.eu-west-1.amazonaws.com/SimpleNotificationService.pem", True),
+ ("subdomain_spoof", "https://sns.us-east-1.amazonaws.com.evil.com/cert.pem", False),
+ ("wrong_service", "https://s3.us-east-1.amazonaws.com/cert.pem", False),
+ ("empty", "", False),
+ ("none", None, False),
+ ]
+ )
+ def test_url_validation(self, _name: str, url: str | None, expected: bool):
+ assert is_valid_sns_url(url) is expected