From 5e505de825ced7d59c03fd89c822d1ef8f1aae4f Mon Sep 17 00:00:00 2001 From: meikel Date: Thu, 30 Jul 2026 10:37:18 +0200 Subject: [PATCH 1/3] feat(workflows): email admins on AWS SES tenant reputation changes Generated-By: PostHog Code Task-Id: 0f3e9835-ed9b-469d-aa52-1152818572a2 --- posthog/settings/ses.py | 7 + posthog/tasks/email.py | 35 +++++ posthog/tasks/scheduled.py | 10 ++ .../email_sending_reputation_finding.html | 38 +++++ posthog/urls.py | 3 + .../backend/api/ses_events_webhook.py | 98 +++++++++++++ ...17_teamworkflowsconfig_ses_tenant_state.py | 27 ++++ .../backend/migrations/max_migration.txt | 2 +- .../backend/models/team_workflows_config.py | 8 ++ .../backend/services/ses_tenant_state.py | 80 +++++++++++ .../backend/services/sns_verification.py | 102 ++++++++++++++ .../backend/tasks/ses_tenant_state.py | 42 ++++++ .../backend/test/test_ses_events_webhook.py | 131 ++++++++++++++++++ .../backend/test/test_ses_tenant_state.py | 91 ++++++++++++ .../backend/test/test_sns_verification.py | 81 +++++++++++ 15 files changed, 754 insertions(+), 1 deletion(-) create mode 100644 posthog/templates/email/email_sending_reputation_finding.html create mode 100644 products/workflows/backend/api/ses_events_webhook.py create mode 100644 products/workflows/backend/migrations/0017_teamworkflowsconfig_ses_tenant_state.py create mode 100644 products/workflows/backend/services/ses_tenant_state.py create mode 100644 products/workflows/backend/services/sns_verification.py create mode 100644 products/workflows/backend/tasks/ses_tenant_state.py create mode 100644 products/workflows/backend/test/test_ses_events_webhook.py create mode 100644 products/workflows/backend/test/test_ses_tenant_state.py create mode 100644 products/workflows/backend/test/test_sns_verification.py 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. +

+
+ View findings +
+
+

+ If the button above doesn't work, paste this link into your browser:
+ {% absolute_uri reputation_path %} +

+
+{% 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..07b0d36c0967 --- /dev/null +++ b/products/workflows/backend/api/ses_events_webhook.py @@ -0,0 +1,98 @@ +""" +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: + message = json.loads(request.body) + except (json.JSONDecodeError, UnicodeDecodeError): + 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: + return HttpResponse("Unknown topic", status=403) + if not verify_sns_message(message): + 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) + requests.get(subscribe_url, timeout=5) + 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..e89795948827 --- /dev/null +++ b/products/workflows/backend/services/ses_tenant_state.py @@ -0,0 +1,80 @@ +import logging + +from django.utils import timezone + +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 = logging.getLogger(__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 — reapplying an unchanged state sends nothing, so the webhook and the + sweep can safely overlap. + """ + impact = reputation_impact or "" + config, _ = TeamWorkflowsConfig.objects.get_or_create(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", + extra={ + "team_id": team_id, + "from_status": previous_status, + "to_status": sending_status, + "from_impact": previous_impact, + "to_impact": impact, + }, + ) + + if previous_status == "": + # First sync is baseline adoption, not a transition — don't notify on rollout. + return + + if sending_status == "DISABLED" and previous_status != "DISABLED": + send_email_sending_suspended.delay(team_id, PROVIDER_PAUSE_REASON, now.isoformat()) + elif previous_status == "DISABLED" and sending_status in ("ENABLED", "REINSTATED"): + send_email_sending_unsuspended.delay(team_id, now.isoformat()) + elif _IMPACT_SEVERITY.get(impact, 0) > _IMPACT_SEVERITY.get(previous_impact, 0): + # Escalation only, and only while still sending — the pause email above already covers + # the paused case, and de-escalations don't need action. + send_email_sending_reputation_finding.delay(team_id, impact, now.isoformat()) diff --git a/products/workflows/backend/services/sns_verification.py b/products/workflows/backend/services/sns_verification.py new file mode 100644 index 000000000000..b468531354b5 --- /dev/null +++ b/products/workflows/backend/services/sns_verification.py @@ -0,0 +1,102 @@ +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". + """ + 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 + hash_algorithm = hashes.SHA256() if message.get("SignatureVersion") == "2" else hashes.SHA1() # noqa: S303 + public_key.verify(signature, string_to_sign.encode(), padding.PKCS1v15(), hash_algorithm) + 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..dac5dc81140d --- /dev/null +++ b/products/workflows/backend/tasks/ses_tenant_state.py @@ -0,0 +1,42 @@ +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) +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..af7e566cf324 --- /dev/null +++ b/products/workflows/backend/test/test_ses_tenant_state.py @@ -0,0 +1,91 @@ +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, + ): + 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, + }, + ) + + def test_first_sync_sets_a_baseline_without_notifying(self): + emails = self._apply("DISABLED", "HIGH") + + config = TeamWorkflowsConfig.objects.get(team=self.team) + assert config.ses_tenant_sending_status == "DISABLED" + assert config.ses_tenant_reputation_impact == "HIGH" + assert config.ses_tenant_state_synced_at is not None + assert all(not mock.called for mock in emails.values()) + + 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), + ("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..377f53aba369 --- /dev/null +++ b/products/workflows/backend/test/test_sns_verification.py @@ -0,0 +1,81 @@ +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": "1", + "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.SHA1()) # noqa: S303 + 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"}), + ] + ) + 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 From b62336ff50a625e58752f0105a1601a4efb863ab Mon Sep 17 00:00:00 2001 From: meikel Date: Thu, 30 Jul 2026 12:28:31 +0200 Subject: [PATCH 2/3] fix(workflows): lock tenant-state transitions, notify already-bad baselines Generated-By: PostHog Code Task-Id: 0f3e9835-ed9b-469d-aa52-1152818572a2 --- .../backend/api/ses_events_webhook.py | 12 ++- .../backend/services/ses_tenant_state.py | 98 ++++++++++++------- .../backend/tasks/ses_tenant_state.py | 12 ++- .../backend/test/test_ses_tenant_state.py | 26 ++++- 4 files changed, 105 insertions(+), 43 deletions(-) diff --git a/products/workflows/backend/api/ses_events_webhook.py b/products/workflows/backend/api/ses_events_webhook.py index 07b0d36c0967..34801f52785b 100644 --- a/products/workflows/backend/api/ses_events_webhook.py +++ b/products/workflows/backend/api/ses_events_webhook.py @@ -57,15 +57,18 @@ def ses_tenant_events_webhook(request: HttpRequest) -> HttpResponse: 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): + 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") @@ -73,7 +76,12 @@ def ses_tenant_events_webhook(request: HttpRequest) -> HttpResponse: 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) - requests.get(subscribe_url, timeout=5) + 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) diff --git a/products/workflows/backend/services/ses_tenant_state.py b/products/workflows/backend/services/ses_tenant_state.py index e89795948827..2d7657cd1371 100644 --- a/products/workflows/backend/services/ses_tenant_state.py +++ b/products/workflows/backend/services/ses_tenant_state.py @@ -1,7 +1,10 @@ -import logging +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, @@ -11,7 +14,7 @@ from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig from products.workflows.backend.providers.ses import SESProvider -logger = logging.getLogger(__name__) +logger = structlog.get_logger(__name__) PROVIDER_PAUSE_REASON = "Our email provider paused sending for this project because of high-impact reputation findings" @@ -38,43 +41,68 @@ def apply_ses_tenant_state(team_id: int, *, sending_status: str, reputation_impa """ 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 — reapplying an unchanged state sends nothing, so the webhook and the - sweep can safely overlap. + 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 "" - config, _ = TeamWorkflowsConfig.objects.get_or_create(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 + TeamWorkflowsConfig.objects.get_or_create(team_id=team_id) - 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", - extra={ - "team_id": team_id, - "from_status": previous_status, - "to_status": sending_status, - "from_impact": previous_impact, - "to_impact": impact, - }, - ) + 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 — don't notify on rollout. - return + # 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": - send_email_sending_suspended.delay(team_id, PROVIDER_PAUSE_REASON, now.isoformat()) - elif previous_status == "DISABLED" and sending_status in ("ENABLED", "REINSTATED"): - send_email_sending_unsuspended.delay(team_id, now.isoformat()) - elif _IMPACT_SEVERITY.get(impact, 0) > _IMPACT_SEVERITY.get(previous_impact, 0): - # Escalation only, and only while still sending — the pause email above already covers - # the paused case, and de-escalations don't need action. - send_email_sending_reputation_finding.delay(team_id, impact, now.isoformat()) + 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/tasks/ses_tenant_state.py b/products/workflows/backend/tasks/ses_tenant_state.py index dac5dc81140d..75d35ee97a89 100644 --- a/products/workflows/backend/tasks/ses_tenant_state.py +++ b/products/workflows/backend/tasks/ses_tenant_state.py @@ -10,7 +10,17 @@ logger = get_logger(__name__) -@shared_task(ignore_result=True, queue=CeleryQueue.DEFAULT.value) +@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) diff --git a/products/workflows/backend/test/test_ses_tenant_state.py b/products/workflows/backend/test/test_ses_tenant_state.py index af7e566cf324..82db85e7b149 100644 --- a/products/workflows/backend/test/test_ses_tenant_state.py +++ b/products/workflows/backend/test/test_ses_tenant_state.py @@ -15,6 +15,7 @@ def _apply(self, sending_status: str, reputation_impact: str | None) -> dict: 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} @@ -28,14 +29,27 @@ def _seed(self, sending_status: str, reputation_impact: str = "") -> None: }, ) - def test_first_sync_sets_a_baseline_without_notifying(self): - emails = self._apply("DISABLED", "HIGH") + @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 == "DISABLED" - assert config.ses_tenant_reputation_impact == "HIGH" + assert config.ses_tenant_sending_status == sending_status assert config.ses_tenant_state_synced_at is not None - assert all(not mock.called for mock in emails.values()) + 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") @@ -55,6 +69,8 @@ def test_unchanged_state_only_refreshes_synced_at(self): ("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), ] ) From c128f3d08db12fc410781ff92090137a043cc539 Mon Sep 17 00:00:00 2001 From: meikel Date: Thu, 30 Jul 2026 16:52:09 +0200 Subject: [PATCH 3/3] fix(workflows): require SNS SignatureVersion 2, drop SHA1 verification Generated-By: PostHog Code Task-Id: 0f3e9835-ed9b-469d-aa52-1152818572a2 --- products/workflows/backend/services/sns_verification.py | 9 +++++++-- products/workflows/backend/test/test_sns_verification.py | 7 +++++-- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/products/workflows/backend/services/sns_verification.py b/products/workflows/backend/services/sns_verification.py index b468531354b5..9fc8dda146e7 100644 --- a/products/workflows/backend/services/sns_verification.py +++ b/products/workflows/backend/services/sns_verification.py @@ -75,6 +75,12 @@ def verify_sns_message(message: dict[str, Any]) -> bool: 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 @@ -92,8 +98,7 @@ def verify_sns_message(message: dict[str, Any]) -> bool: public_key = load_pem_x509_certificate(cert_pem).public_key() if not isinstance(public_key, RSAPublicKey): return False - hash_algorithm = hashes.SHA256() if message.get("SignatureVersion") == "2" else hashes.SHA1() # noqa: S303 - public_key.verify(signature, string_to_sign.encode(), padding.PKCS1v15(), hash_algorithm) + public_key.verify(signature, string_to_sign.encode(), padding.PKCS1v15(), hashes.SHA256()) return True except InvalidSignature: return False diff --git a/products/workflows/backend/test/test_sns_verification.py b/products/workflows/backend/test/test_sns_verification.py index 377f53aba369..31ff00cf4578 100644 --- a/products/workflows/backend/test/test_sns_verification.py +++ b/products/workflows/backend/test/test_sns_verification.py @@ -35,13 +35,13 @@ def _signed_notification(tamper: dict[str, Any] | None = None) -> dict[str, Any] "TopicArn": "arn:aws:sns:us-east-1:123456789012:topic", "Message": '{"source":"aws.ses"}', "Timestamp": "2026-07-30T00:00:00.000Z", - "SignatureVersion": "1", + "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.SHA1()) # noqa: S303 + signature = _KEY.sign(string_to_sign.encode(), padding.PKCS1v15(), hashes.SHA256()) message["Signature"] = base64.b64encode(signature).decode() message.update(tamper or {}) return message @@ -63,6 +63,9 @@ def test_accepts_a_correctly_signed_notification(self): ("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]):