From 25af2dbab43bac1aaad5b33cee0e4839cd5c4ef6 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Mon, 19 Oct 2026 09:09:23 +0000 Subject: [PATCH 1/2] fix(workflows): refresh pending email sender verification Background sweeps check pending SES email senders and save the verified flag, so Channels and MCP reads see a verified sender without the user opening the verification UI. Senders whose verification was requested in the last hour are checked every minute, older ones up to 7 days every 5 minutes. A resolved SES VDM advisor finding marks the pending senders on its domain as requested again. SES status reads are batched per 100 senders and paced in background providers. Matching senders are row-locked while marked verified, and the first verified sender becomes the project's default when none exists. Co-Authored-By: Claude Opus 5.5 --- posthog/api/test/test_integration.py | 17 +- posthog/models/integration/__init__.py | 3 +- posthog/models/integration/email.py | 78 ++-- posthog/tasks/scheduled.py | 18 + products/workflows/backend/facade/tasks.py | 6 + products/workflows/backend/providers/ses.py | 185 +++++++--- .../services/email_sender_verification.py | 179 +++++++++ .../backend/services/ses_tenant_events.py | 32 +- products/workflows/backend/tasks/__init__.py | 18 +- .../tasks/email_sender_verification.py | 22 ++ .../test/test_email_integration_ses.py | 4 +- .../test/test_email_sender_verification.py | 346 ++++++++++++++++++ .../backend/test/test_ses_provider.py | 13 +- .../backend/test/test_ses_tenant_events.py | 15 + 14 files changed, 847 insertions(+), 89 deletions(-) create mode 100644 products/workflows/backend/services/email_sender_verification.py create mode 100644 products/workflows/backend/tasks/email_sender_verification.py create mode 100644 products/workflows/backend/test/test_email_sender_verification.py diff --git a/posthog/api/test/test_integration.py b/posthog/api/test/test_integration.py index 0b639d08feb8..e42ac51274e6 100644 --- a/posthog/api/test/test_integration.py +++ b/posthog/api/test/test_integration.py @@ -659,6 +659,7 @@ def test_integration_from_domain( "reply_to": "support@posthog.com", "verified": False, "provider": "ses", + "verification_requested_at": ANY, } assert integration.sensitive_config == {} assert integration.created_by == self.user @@ -717,7 +718,9 @@ def test_email_verify_returns_ses_result(self, mock_create_email_domain, mock_ve created_by=self.user, ) email_integration = EmailIntegration(integration) - verification_result = email_integration.verify() + clicked_at = timezone.now() + timedelta(hours=2) + with time_machine.travel(clicked_at, tick=False): + verification_result = email_integration.verify() assert verification_result == expected_result @@ -734,6 +737,7 @@ def test_email_verify_returns_ses_result(self, mock_create_email_domain, mock_ve "reply_to": "", "verified": False, "provider": "ses", + "verification_requested_at": int(clicked_at.timestamp()), } @patch("products.workflows.backend.facade.api.verify_ses_email_domain") @@ -771,6 +775,7 @@ def test_email_verify_updates_integration(self, mock_create_email_domain, mock_v "reply_to": "", "verified": True, "provider": "ses", + "verification_requested_at": ANY, } @patch("products.workflows.backend.facade.api.verify_ses_email_domain") @@ -795,7 +800,13 @@ def test_email_verify_updates_all_other_integrations_with_same_domain( created_by=self.user, ) integration2 = EmailIntegration.create_native_integration( - {**self.valid_config, "provider": "ses"}, + {**self.valid_config, "email": "other@posthog.com", "provider": "ses"}, + team_id=self.team.id, + organization_id=self.organization.id, + created_by=self.user, + ) + integrationOtherMailFrom = EmailIntegration.create_native_integration( + {**self.valid_config, "email": "bounces@posthog.com", "mail_from_subdomain": "mail", "provider": "ses"}, team_id=self.team.id, organization_id=self.organization.id, created_by=self.user, @@ -822,10 +833,12 @@ def test_email_verify_updates_all_other_integrations_with_same_domain( integration1.refresh_from_db() integration2.refresh_from_db() + integrationOtherMailFrom.refresh_from_db() integrationOtherDomain.refresh_from_db() assert integration1.config["verified"] assert integration2.config["verified"] + assert not integrationOtherMailFrom.config["verified"] assert not integrationOtherDomain.config["verified"] diff --git a/posthog/models/integration/__init__.py b/posthog/models/integration/__init__.py index 89b6b633d166..545203e35b73 100644 --- a/posthog/models/integration/__init__.py +++ b/posthog/models/integration/__init__.py @@ -32,7 +32,7 @@ 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 EmailIntegration, cleanup_ses_identity_on_integration_delete, mark_verification_requested from .external_issues import ( SUPPORTED_EXTERNAL_ISSUE_PROVIDERS, external_issue_url, @@ -260,4 +260,5 @@ "RedshiftIntegration", "SnowflakeIntegrationError", "SnowflakeIntegration", + "mark_verification_requested", ] diff --git a/posthog/models/integration/email.py b/posthog/models/integration/email.py index 2dc30b43d6cc..0fe7a16b3dd2 100644 --- a/posthog/models/integration/email.py +++ b/posthog/models/integration/email.py @@ -4,7 +4,9 @@ from django.conf import settings from django.db import models, transaction +from django.db.models import F, Func, JSONField, Q, Value from django.dispatch import receiver +from django.utils import timezone from disposable_email_domains import blocklist as disposable_email_domains_list from free_email_domains import whitelist as free_email_domains_list @@ -20,6 +22,23 @@ from products.workflows.backend.facade.contracts import EmailDomainVerification +VERIFICATION_REQUESTED_AT = "verification_requested_at" + + +def _with_config_key(key: str, value: object) -> Func: + return Func( + F("config"), + Value([key]), + Value(value, output_field=JSONField()), + function="jsonb_set", + output_field=JSONField(), + ) + + +def mark_verification_requested(senders: "models.QuerySet[model.Integration]") -> int: + return senders.update(config=_with_config_key(VERIFICATION_REQUESTED_AT, int(timezone.now().timestamp()))) + + class EmailIntegration: integration: model.Integration @@ -81,6 +100,7 @@ def create_native_integration( "name": name, "provider": provider, "verified": True if provider == "maildev" else False, + VERIFICATION_REQUESTED_AT: int(timezone.now().timestamp()), }, "created_by": created_by, }, @@ -125,6 +145,38 @@ def update_native_integration(self, config: dict, team_id: int) -> model.Integra return self.integration + def apply_verification(self, verification_result: "EmailDomainVerification") -> "EmailDomainVerification": + if verification_result.get("status") == "success": + verified_ids = self._mark_matching_senders_verified() + if verified_ids: + reload_integrations_on_workers(self.integration.team_id, verified_ids) + return verification_result + + def _mark_matching_senders_verified(self) -> list[int]: + with transaction.atomic(): + verified_ids = list(self._matching_senders().select_for_update().values_list("id", flat=True)) + model.Integration.objects.filter(id__in=verified_ids).update(config=_with_config_key("verified", True)) + if self.integration.id in verified_ids: + from products.workflows.backend.facade.api import ( + set_default_email_sender_if_unset, # noqa: PLC0415 — keeps the workflows facade off the model import path + ) + + set_default_email_sender_if_unset(self.integration.team_id, self.integration.id) + return verified_ids + + def _matching_senders(self) -> "models.QuerySet[model.Integration]": + mail_from_subdomain = self.integration.config.get("mail_from_subdomain", "feedback") + matching_mail_from = Q(config__mail_from_subdomain=mail_from_subdomain) + if mail_from_subdomain == "feedback": + matching_mail_from |= Q(config__mail_from_subdomain__isnull=True) + return model.Integration.objects.filter( + matching_mail_from, + team_id=self.integration.team_id, + kind="email", + config__domain=self.integration.config.get("domain"), + config__provider=self.integration.config.get("provider", "ses"), + ) + def verify(self) -> "EmailDomainVerification": domain = self.integration.config.get("domain") provider = self.integration.config.get("provider", "ses") @@ -152,29 +204,9 @@ def verify(self) -> "EmailDomainVerification": else: raise ValueError(f"Invalid provider: {provider}") - if verification_result.get("status") == "success": - # We can validate all other integrations with the same domain and provider - all_integrations_for_domain = model.Integration.objects.filter( - team_id=self.integration.team_id, - kind="email", - config__domain=domain, - config__provider=provider, - ) - for integration in all_integrations_for_domain: - integration.config["verified"] = True - integration.save() - - reload_integrations_on_workers( - self.integration.team_id, [integration.id for integration in all_integrations_for_domain] - ) - - from products.workflows.backend.facade.api import ( - set_default_email_sender_if_unset, # noqa: PLC0415 — keeps the workflows facade off the model import path - ) - - set_default_email_sender_if_unset(self.integration.team_id, self.integration.id) - - return verification_result + if verification_result.get("status") != "success": + mark_verification_requested(model.Integration.objects.filter(id=self.integration.id)) + return self.apply_verification(verification_result) @receiver(models.signals.post_delete, sender=model.Integration) diff --git a/posthog/tasks/scheduled.py b/posthog/tasks/scheduled.py index 96dea56cb627..deaebc724581 100644 --- a/posthog/tasks/scheduled.py +++ b/posthog/tasks/scheduled.py @@ -158,6 +158,8 @@ poll_ses_reputation_findings, recompute_workflows_email_sending_tiers, reconcile_ses_tenant_states, + refresh_new_email_senders, + refresh_settled_email_senders, sweep_workflow_email_deliverability, ) @@ -338,6 +340,22 @@ def setup_periodic_tasks(sender: Celery, **kwargs: Any) -> None: expires_seconds=12 * 3600, ) + add_periodic_task_with_expiry( + sender, + crontab(minute="*"), + refresh_new_email_senders.s(), + name="refresh new workflow email senders", + expires_seconds=60, + ) + + add_periodic_task_with_expiry( + sender, + crontab(minute="*/5"), + refresh_settled_email_senders.s(), + name="refresh settled workflow email senders", + expires_seconds=300, + ) + # 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. diff --git a/products/workflows/backend/facade/tasks.py b/products/workflows/backend/facade/tasks.py index f4258abea162..e4e576e64e56 100644 --- a/products/workflows/backend/facade/tasks.py +++ b/products/workflows/backend/facade/tasks.py @@ -1,3 +1,7 @@ +from products.workflows.backend.tasks.email_sender_verification import ( + refresh_new_email_senders, + refresh_settled_email_senders, +) from products.workflows.backend.tasks.email_sending_tiers import recompute_workflows_email_sending_tiers from products.workflows.backend.tasks.hog_flows import refresh_affected_hog_flows from products.workflows.backend.tasks.ses_account_reputation import ( @@ -13,5 +17,7 @@ "recompute_workflows_email_sending_tiers", "reconcile_ses_tenant_states", "refresh_affected_hog_flows", + "refresh_new_email_senders", + "refresh_settled_email_senders", "sweep_workflow_email_deliverability", ] diff --git a/products/workflows/backend/providers/ses.py b/products/workflows/backend/providers/ses.py index 5173b5324559..869ae5a9d10e 100644 --- a/products/workflows/backend/providers/ses.py +++ b/products/workflows/backend/providers/ses.py @@ -5,7 +5,7 @@ from datetime import UTC, datetime, timedelta from functools import cached_property from itertools import batched -from time import monotonic +from time import monotonic, sleep from typing import TYPE_CHECKING, Any from django.conf import settings @@ -27,6 +27,7 @@ if TYPE_CHECKING: from types_boto3_ses.client import SESClient + from types_boto3_ses.type_defs import IdentityMailFromDomainAttributesTypeDef from types_boto3_sesv2.client import SESV2Client from types_boto3_sesv2.type_defs import RecommendationTypeDef @@ -76,6 +77,42 @@ # mode adds a client-side rate limiter that slows the following pages after a throttle response. RECOMMENDATIONS_RETRY_CONFIG = Config(retries={"mode": "adaptive", "max_attempts": 8}) +IDENTITY_ATTRIBUTES_BATCH_SIZE = 100 +READ_MIN_INTERVAL_SECONDS = 2 + +BACKGROUND_CONNECT_TIMEOUT_SECONDS = 2 +BACKGROUND_READ_TIMEOUT_SECONDS = 5 +BACKGROUND_TOTAL_ATTEMPTS = 2 +BACKGROUND_FIRST_RETRY_MAX_BACKOFF_SECONDS = 1 +BACKGROUND_CALL_WORST_CASE_SECONDS = ( + BACKGROUND_CONNECT_TIMEOUT_SECONDS + BACKGROUND_READ_TIMEOUT_SECONDS +) * BACKGROUND_TOTAL_ATTEMPTS + BACKGROUND_FIRST_RETRY_MAX_BACKOFF_SECONDS +BACKGROUND_CLIENT_CONFIG = Config( + connect_timeout=BACKGROUND_CONNECT_TIMEOUT_SECONDS, + read_timeout=BACKGROUND_READ_TIMEOUT_SECONDS, + retries={"mode": "standard", "total_max_attempts": BACKGROUND_TOTAL_ATTEMPTS}, +) + + +def _same_domain(left: str | None, right: str) -> bool: + return left is not None and left.lower().rstrip(".") == right.lower().rstrip(".") + + +@frozen +class SESIdentityStatus: + verification: str = "Unknown" + dkim: str = "Unknown" + mail_from: str = "Unknown" + mail_from_domain: str | None = None + + def mail_from_status_for(self, mail_from_domain: str) -> str: + if self.mail_from != "Unknown" and not _same_domain(self.mail_from_domain, mail_from_domain): + return "Pending" + return self.mail_from + + def is_ready_for(self, mail_from_domain: str) -> bool: + return self.verification == self.dkim == self.mail_from_status_for(mail_from_domain) == "Success" + @frozen class IspSendingMetrics: @@ -282,25 +319,29 @@ class SESProvider: ses_v2_metrics_client: "SESV2Client" ses_v2_recommendations_client: "SESV2Client" - def __init__(self): - # Initialize the boto3 clients + def __init__(self, client_config: Config | None = None, read_interval_seconds: float = 0.0): + self._read_interval_seconds = read_interval_seconds + self._next_read_at: dict[str, float] = {} self.sts_client = boto3.client( "sts", aws_access_key_id=settings.SES_ACCESS_KEY_ID, aws_secret_access_key=settings.SES_SECRET_ACCESS_KEY, region_name=settings.SES_REGION, + config=client_config, ) self.ses_client = boto3.client( "ses", aws_access_key_id=settings.SES_ACCESS_KEY_ID, aws_secret_access_key=settings.SES_SECRET_ACCESS_KEY, region_name=settings.SES_REGION, + config=client_config, ) self.ses_v2_client = boto3.client( "sesv2", aws_access_key_id=settings.SES_ACCESS_KEY_ID, aws_secret_access_key=settings.SES_SECRET_ACCESS_KEY, region_name=settings.SES_REGION, + config=client_config, ) # Separate client for the metric fan-out only, so bounding it cannot slow identity and # tenant calls, which are allowed to take longer. @@ -323,6 +364,10 @@ def __init__(self): config=RECOMMENDATIONS_RETRY_CONFIG, ) + @classmethod + def for_background(cls) -> "SESProvider": + return cls(client_config=BACKGROUND_CLIENT_CONFIG, read_interval_seconds=READ_MIN_INTERVAL_SECONDS) + def _tenant_name_for_team(self, team_id: int) -> str: return f"team-{team_id}" @@ -456,6 +501,7 @@ def _associate_tenant_resource(self, tenant_name: str, resource_arn: str) -> Non raise def _list_identity_tenants(self, domain: str) -> set[str]: + self._pace_read("resource_tenants") try: resp = self.ses_v2_client.list_resource_tenants(ResourceArn=self._identity_arn(domain)) except ClientError as e: @@ -532,8 +578,6 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in if not re.match(DOMAIN_REGEX, domain): raise exceptions.ValidationError("Please enter a valid domain or subdomain name.") - dns_records: list[EmailDomainDnsRecord] = [] - # Start/ensure domain verification (TXT at _amazonses.domain) --- verification_token: str | None = None try: @@ -544,6 +588,41 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in if e.response["Error"]["Code"] not in ("InvalidParameterValue",): raise + # Start/ensure DKIM (three CNAMEs) --- + dkim_tokens: list[str] = [] + try: + dkim_resp = self.ses_client.verify_domain_dkim(Domain=domain) + dkim_tokens = dkim_resp["DkimTokens"] + except ClientError as e: + if e.response["Error"]["Code"] not in ("InvalidParameterValue",): + raise + + # Start/ensure MAIL FROM setup (MX + TXT) --- + try: + self.ses_client.set_identity_mail_from_domain( + Identity=domain, + MailFromDomain=f"{mail_from_subdomain}.{domain}", + BehaviorOnMXFailure="UseDefaultValue", + ) + except ClientError as e: + if e.response["Error"]["Code"] not in ("InvalidParameterValue",): + raise + + return self.get_email_domain_verification( + domain, mail_from_subdomain, team_id, verification_token=verification_token, dkim_tokens=dkim_tokens + ) + + def get_email_domain_verification( + self, + domain: str, + mail_from_subdomain: str, + team_id: int, + *, + verification_token: str | None = None, + dkim_tokens: Sequence[str] = (), + identity_status: SESIdentityStatus | None = None, + ) -> EmailDomainVerification: + dns_records: list[EmailDomainDnsRecord] = [] if verification_token: dns_records.append( { @@ -555,15 +634,6 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in } ) - # Start/ensure DKIM (three CNAMEs) --- - dkim_tokens: list[str] = [] - try: - dkim_resp = self.ses_client.verify_domain_dkim(Domain=domain) - dkim_tokens = dkim_resp["DkimTokens"] - except ClientError as e: - if e.response["Error"]["Code"] not in ("InvalidParameterValue",): - raise - for t in dkim_tokens: dns_records.append( { @@ -585,17 +655,6 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in } ) - # Start/ensure MAIL FROM setup (MX + TXT) --- - try: - self.ses_client.set_identity_mail_from_domain( - Identity=domain, - MailFromDomain=f"{mail_from_subdomain}.{domain}", - BehaviorOnMXFailure="UseDefaultValue", - ) - except ClientError as e: - if e.response["Error"]["Code"] not in ("InvalidParameterValue",): - raise - ses_region = getattr(settings, "SES_REGION", "us-east-1") dns_records.append( @@ -630,33 +689,10 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in } ) - # Current verification / DKIM statuses to compute overall status & per-record statuses --- - verification_status: str = "Unknown" - try: - id_attrs = self.ses_client.get_identity_verification_attributes(Identities=[domain]) - id_for_domain = id_attrs["VerificationAttributes"].get(domain) - if id_for_domain is not None: - verification_status = id_for_domain["VerificationStatus"] - except ClientError: - pass - - dkim_status: str = "Unknown" - try: - dkim_attrs = self.ses_client.get_identity_dkim_attributes(Identities=[domain]) - dkim_for_domain = dkim_attrs["DkimAttributes"].get(domain) - if dkim_for_domain is not None: - dkim_status = dkim_for_domain["DkimVerificationStatus"] - except ClientError: - pass - - mail_from_status: str = "Unknown" - try: - mail_from_attrs = self.ses_client.get_identity_mail_from_domain_attributes(Identities=[domain]) - mail_from_for_domain = mail_from_attrs["MailFromDomainAttributes"].get(domain) - if mail_from_for_domain is not None: - mail_from_status = mail_from_for_domain["MailFromDomainStatus"] - except ClientError: - pass + status = identity_status or self.get_identity_statuses([domain])[domain] + verification_status = status.verification + dkim_status = status.dkim + mail_from_status = status.mail_from_status_for(f"{mail_from_subdomain}.{domain}") # DMARC: check via direct DNS lookup since AWS SES doesn't track it dmarc_status = "Pending" @@ -745,6 +781,51 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in "dnsRecords": dns_records, } + def get_identity_statuses(self, domains: Sequence[str]) -> dict[str, SESIdentityStatus]: + statuses = dict.fromkeys(domains, SESIdentityStatus()) + for batch in batched(dict.fromkeys(domains), IDENTITY_ATTRIBUTES_BATCH_SIZE, strict=False): + self._pace_read("identity_attributes") + statuses.update(self._read_identity_statuses(list(batch))) + return statuses + + def _pace_read(self, operation: str) -> None: + wait_seconds = self._next_read_at.get(operation, 0.0) - monotonic() + if wait_seconds > 0: + sleep(wait_seconds) + self._next_read_at[operation] = monotonic() + self._read_interval_seconds + + def _read_identity_statuses(self, domains: list[str]) -> dict[str, SESIdentityStatus]: + verification: dict[str, str] = {} + try: + id_attrs = self.ses_client.get_identity_verification_attributes(Identities=domains) + verification = {d: a["VerificationStatus"] for d, a in id_attrs["VerificationAttributes"].items()} + except ClientError: + pass + + dkim: dict[str, str] = {} + try: + dkim_attrs = self.ses_client.get_identity_dkim_attributes(Identities=domains) + dkim = {d: a["DkimVerificationStatus"] for d, a in dkim_attrs["DkimAttributes"].items()} + except ClientError: + pass + + mail_from: dict[str, IdentityMailFromDomainAttributesTypeDef] = {} + try: + mail_from_attrs = self.ses_client.get_identity_mail_from_domain_attributes(Identities=domains) + mail_from = dict(mail_from_attrs["MailFromDomainAttributes"]) + except ClientError: + pass + + return { + domain: SESIdentityStatus( + verification=verification.get(domain, "Unknown"), + dkim=dkim.get(domain, "Unknown"), + mail_from=mail_from[domain]["MailFromDomainStatus"] if domain in mail_from else "Unknown", + mail_from_domain=mail_from[domain].get("MailFromDomain") if domain in mail_from else None, + ) + for domain in domains + } + def update_mail_from_subdomain(self, domain: str, mail_from_subdomain: str): """ Update the MAIL FROM subdomain for a given identity diff --git a/products/workflows/backend/services/email_sender_verification.py b/products/workflows/backend/services/email_sender_verification.py new file mode 100644 index 000000000000..bd6ee8b626df --- /dev/null +++ b/products/workflows/backend/services/email_sender_verification.py @@ -0,0 +1,179 @@ +from collections.abc import Callable, Iterable +from datetime import timedelta +from functools import cached_property +from itertools import batched +from time import monotonic + +from django.db.models import Q, QuerySet +from django.utils import timezone + +import structlog +from botocore.exceptions import BotoCoreError + +from posthog.dataclasses import frozen +from posthog.models.integration import ( + EmailIntegration, + Integration, + defer_repository_cache_fields, + mark_verification_requested, +) +from posthog.redis import get_client + +from products.workflows.backend.providers.ses import ( + BACKGROUND_CALL_WORST_CASE_SECONDS, + IDENTITY_ATTRIBUTES_BATCH_SIZE, + READ_MIN_INTERVAL_SECONDS, + SESIdentityStatus, + SESProvider, +) + +logger = structlog.get_logger(__name__) + + +@frozen +class SenderAgeTier: + name: str + min_age: timedelta + max_age: timedelta + work_budget_seconds: int + lock_timeout_seconds: int + + def redis_key(self, purpose: str) -> str: + return f"workflows:email_sender_verification:{self.name}:{purpose}" + + +NEW_SENDERS = SenderAgeTier( + name="new", + min_age=timedelta(), + max_age=timedelta(hours=1), + work_budget_seconds=90, + lock_timeout_seconds=120, +) +SETTLED_SENDERS = SenderAgeTier( + name="settled", + min_age=timedelta(hours=1), + max_age=timedelta(days=7), + work_budget_seconds=240, + lock_timeout_seconds=300, +) + + +class EmailSenderVerification: + SES_CALLS_PER_STEP = 3 + STEP_WORST_CASE_SECONDS = SES_CALLS_PER_STEP * BACKGROUND_CALL_WORST_CASE_SECONDS + READ_MIN_INTERVAL_SECONDS + + def __init__(self, work_budget_seconds: int) -> None: + self.deadline = monotonic() + work_budget_seconds + + @cached_property + def provider(self) -> SESProvider: + return SESProvider.for_background() + + @staticmethod + def refresh_pending(tier: SenderAgeTier) -> None: + lock = get_client().lock(tier.redis_key("lock"), timeout=tier.lock_timeout_seconds) + if not lock.acquire(blocking=False): + return + try: + _TierSweep(tier).run() + finally: + lock.release() + + @staticmethod + def request_domain_check(domain: str) -> None: + mark_verification_requested(_pending_ses_senders().filter(config__domain__iexact=domain)) + + def verify_ready_senders( + self, senders: Iterable[Integration], on_sender: Callable[[Integration], None] = lambda _: None + ) -> bool: + checked: set[tuple[int, str]] = set() + for batch in batched(senders, IDENTITY_ATTRIBUTES_BATCH_SIZE, strict=False): + if not self._has_time_for_step(): + return False + try: + statuses = self.provider.get_identity_statuses([d for sender in batch if (d := _domain(sender))]) + except BotoCoreError: + logger.exception("email_sender_verification_ses_unavailable") + return False + for sender in batch: + ready_status = _ready_status(statuses, sender) + if ready_status and not self._has_time_for_step(): + return False + on_sender(sender) + key = (sender.team_id, _mail_from_domain(sender)) + if ready_status and key not in checked: + checked.add(key) + self._verify(sender, ready_status) + return True + + def _has_time_for_step(self) -> bool: + return monotonic() + self.STEP_WORST_CASE_SECONDS <= self.deadline + + def _verify(self, sender: Integration, status: SESIdentityStatus) -> None: + try: + verification = self.provider.get_email_domain_verification( + sender.config["domain"], _mail_from_subdomain(sender), sender.team_id, identity_status=status + ) + EmailIntegration(sender).apply_verification(verification) + except Exception: + logger.exception("email_sender_verification_failed", team_id=sender.team_id, integration_id=sender.id) + + +class _TierSweep: + def __init__(self, tier: SenderAgeTier) -> None: + self.tier = tier + self.redis = get_client() + + def run(self) -> None: + verification = EmailSenderVerification(self.tier.work_budget_seconds) + if verification.verify_ready_senders(self._senders().iterator(chunk_size=500), on_sender=self._advance): + self.redis.delete(self.tier.redis_key("cursor"), self.tier.redis_key("scan_end")) + + def _in_tier(self) -> Q: + now = timezone.now() + newest, oldest = now - self.tier.min_age, now - self.tier.max_age + requested_in_tier = Q( + config__verification_requested_at__gt=oldest.timestamp(), + config__verification_requested_at__lte=newest.timestamp(), + ) + created_in_tier = Q( + config__verification_requested_at__isnull=True, created_at__gt=oldest, created_at__lte=newest + ) + return requested_in_tier | created_in_tier + + def _advance(self, sender: Integration) -> None: + self.redis.set(self.tier.redis_key("cursor"), sender.id) + + def _senders(self) -> QuerySet[Integration]: + pending = _pending_ses_senders().filter(self._in_tier()) + cursor = int(self.redis.get(self.tier.redis_key("cursor")) or 0) + scan_end = int(self.redis.get(self.tier.redis_key("scan_end")) or 0) + if not scan_end: + scan_end = pending.order_by("-id").values_list("id", flat=True).first() or 0 + self.redis.set(self.tier.redis_key("scan_end"), scan_end) + return pending.filter(id__gt=cursor, id__lte=scan_end).order_by("id") + + +def _pending_ses_senders() -> QuerySet[Integration]: + return defer_repository_cache_fields( + Integration.objects.filter(kind="email", config__provider="ses", config__verified=False) + ) + + +def _ready_status(statuses: dict[str, SESIdentityStatus], sender: Integration) -> SESIdentityStatus | None: + status = statuses.get(_domain(sender) or "") + if status is None or not status.is_ready_for(_mail_from_domain(sender)): + return None + return status + + +def _domain(sender: Integration) -> str | None: + return sender.config.get("domain") + + +def _mail_from_subdomain(sender: Integration) -> str: + return sender.config.get("mail_from_subdomain", "feedback") + + +def _mail_from_domain(sender: Integration) -> str: + return f"{_mail_from_subdomain(sender)}.{_domain(sender)}" diff --git a/products/workflows/backend/services/ses_tenant_events.py b/products/workflows/backend/services/ses_tenant_events.py index 77672ddc31cf..d8c3331087e6 100644 --- a/products/workflows/backend/services/ses_tenant_events.py +++ b/products/workflows/backend/services/ses_tenant_events.py @@ -1,12 +1,13 @@ -"""AWS SES tenant reputation events, delivered via EventBridge → SNS HTTPS subscription. +"""AWS SES tenant reputation and VDM advisor 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, +enqueues a tenant sync, or marks the identity's pending senders as due for the next 1-minute sender +sweep, and both read the authoritative state from the SES API. That makes duplicate, reordered, or partially-shaped events harmless. The SNS message signature and the topic-ARN allowlist are checked by the ingress transport, which reads the allowlist from `WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS`. AWS-side wiring: an EventBridge -rule on `source = aws.ses` for the tenant detail-types, targeting an SNS topic with an HTTPS +rule on `source = aws.ses` for the tenant detail-types and `Advisor Recommendation Status Resolved`, targeting an SNS topic with an HTTPS subscription to this endpoint. """ @@ -21,11 +22,21 @@ from posthog.ingress.contracts import WebhookDelivery from posthog.ingress.verify.sns_signature import is_valid_sns_url +from products.workflows.backend.services.email_sender_verification import EmailSenderVerification 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") +_IDENTITY_ARN_RE = re.compile(r"^arn:aws:ses:[^:]*:[^:]*:identity/([^/]+)$") +ADVISOR_RECOMMENDATION_RESOLVED = "Advisor Recommendation Status Resolved" + + +def _extract_identities(event: dict[str, Any]) -> list[str]: + resources = event.get("resources") + if not isinstance(resources, list): + return [] + return [match.group(1) for r in resources if isinstance(r, str) and (match := _IDENTITY_ARN_RE.match(r))] def _extract_team_id(event: dict[str, Any]) -> int | None: @@ -61,7 +72,7 @@ def _confirm_subscription(message: Mapping[str, Any]) -> None: def _handle_notification(message: Mapping[str, Any]) -> None: - """Enqueue a tenant-state sync for the team the EventBridge event names. + """Queue a sender check for a resolved advisor finding, else enqueue a tenant-state sync for the team the event names. Every unusable payload is dropped rather than raised, because a redelivery would carry the same unusable payload. @@ -73,6 +84,10 @@ def _handle_notification(message: Mapping[str, Any]) -> None: if not isinstance(event, dict) or event.get("source") != "aws.ses": return + if event.get("detail-type") == ADVISOR_RECOMMENDATION_RESOLVED: + _handle_advisor_resolution(event) + return + 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")) @@ -85,8 +100,15 @@ def _handle_notification(message: Mapping[str, Any]) -> None: sync_ses_tenant_state_task.delay(team_id) +def _handle_advisor_resolution(event: dict[str, Any]) -> None: + identities = _extract_identities(event) + logger.info("ses_advisor_resolution_webhook_accepted", identity_count=len(identities)) + for identity in identities: + EmailSenderVerification.request_domain_check(identity) + + def handle_ses_tenant_event(delivery: WebhookDelivery) -> None: - """One SNS delivery: the subscription handshake, or a tenant reputation event.""" + """One SNS delivery: the subscription handshake, a tenant reputation event, or a resolved advisor finding.""" if delivery.event_type == "SubscriptionConfirmation": _confirm_subscription(delivery.payload) elif delivery.event_type == "Notification": diff --git a/products/workflows/backend/tasks/__init__.py b/products/workflows/backend/tasks/__init__.py index 59bcecd8d2fc..57f4902a6dd6 100644 --- a/products/workflows/backend/tasks/__init__.py +++ b/products/workflows/backend/tasks/__init__.py @@ -1,3 +1,17 @@ -from . import data_suggestions, email_sending_tiers, hog_flows, people_import, ses_account_reputation +from . import ( + data_suggestions, + email_sender_verification, + email_sending_tiers, + hog_flows, + people_import, + ses_account_reputation, +) -__all__ = ["data_suggestions", "email_sending_tiers", "hog_flows", "people_import", "ses_account_reputation"] +__all__ = [ + "data_suggestions", + "email_sender_verification", + "email_sending_tiers", + "hog_flows", + "people_import", + "ses_account_reputation", +] diff --git a/products/workflows/backend/tasks/email_sender_verification.py b/products/workflows/backend/tasks/email_sender_verification.py new file mode 100644 index 000000000000..1d9198954eca --- /dev/null +++ b/products/workflows/backend/tasks/email_sender_verification.py @@ -0,0 +1,22 @@ +from celery import shared_task + +from posthog.scoping_audit import skip_team_scope_audit +from posthog.tasks.utils import CeleryQueue + +from products.workflows.backend.services.email_sender_verification import ( + NEW_SENDERS, + SETTLED_SENDERS, + EmailSenderVerification, +) + + +@shared_task(ignore_result=True, queue=CeleryQueue.LONG_RUNNING.value, time_limit=110) +@skip_team_scope_audit +def refresh_new_email_senders() -> None: + EmailSenderVerification.refresh_pending(NEW_SENDERS) + + +@shared_task(ignore_result=True, queue=CeleryQueue.LONG_RUNNING.value, time_limit=270) +@skip_team_scope_audit +def refresh_settled_email_senders() -> None: + EmailSenderVerification.refresh_pending(SETTLED_SENDERS) diff --git a/products/workflows/backend/test/test_email_integration_ses.py b/products/workflows/backend/test/test_email_integration_ses.py index ad8f8c6380c4..a38789497c60 100644 --- a/products/workflows/backend/test/test_email_integration_ses.py +++ b/products/workflows/backend/test/test_email_integration_ses.py @@ -41,7 +41,9 @@ def _set_global_ses_success(self, provider, domain: str) -> None: "DkimAttributes": {domain: {"DkimVerificationStatus": "Success"}} } provider.ses_client.get_identity_mail_from_domain_attributes.return_value = { - "MailFromDomainAttributes": {domain: {"MailFromDomainStatus": "Success"}} + "MailFromDomainAttributes": { + domain: {"MailFromDomainStatus": "Success", "MailFromDomain": f"feedback.{domain}"} + } } @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") diff --git a/products/workflows/backend/test/test_email_sender_verification.py b/products/workflows/backend/test/test_email_sender_verification.py new file mode 100644 index 000000000000..951f643cb118 --- /dev/null +++ b/products/workflows/backend/test/test_email_sender_verification.py @@ -0,0 +1,346 @@ +from collections.abc import Callable +from datetime import timedelta +from types import SimpleNamespace +from typing import Any + +from posthog.test.base import APIBaseTest +from unittest.mock import MagicMock, patch + +from django.utils import timezone + +import dns.resolver +from botocore.exceptions import ClientError, EndpointConnectionError +from parameterized import param, parameterized + +from posthog.models.integration import Integration +from posthog.redis import get_client + +from products.workflows.backend.facade.tasks import refresh_new_email_senders, refresh_settled_email_senders +from products.workflows.backend.services.email_sender_verification import EmailSenderVerification + + +class TestEmailSenderVerification(APIBaseTest): + def setUp(self) -> None: + super().setUp() + redis = get_client() + for key in redis.scan_iter("workflows:email_sender_verification:*"): + redis.delete(key) + self.ses_clock = {"now": 0.0} + for name, fake in [ + ("monotonic", lambda: self.ses_clock["now"]), + ("sleep", lambda seconds: self.ses_clock.update(now=self.ses_clock["now"] + seconds)), + ]: + clock_patch = patch(f"products.workflows.backend.providers.ses.{name}", side_effect=fake) + clock_patch.start() + self.addCleanup(clock_patch.stop) + + def _create_sender(self, domain: str, local_part: str = "sender", **config: Any) -> Integration: + return Integration.objects.create( + team=self.team, + kind="email", + integration_id=f"{local_part}@{domain}", + config={ + "email": f"{local_part}@{domain}", + "domain": domain, + "name": "Example sender", + "provider": "ses", + "mail_from_subdomain": "feedback", + "verified": False, + **config, + }, + ) + + def _mock_ses( + self, + mock_boto_client: MagicMock, + mock_resolver: MagicMock, + *, + dkim_status: str = "Success", + has_dmarc: bool = True, + has_tenant: bool = True, + mail_from_domain: str | None = None, + ) -> MagicMock: + ses = mock_boto_client.return_value + ses.get_identity_verification_attributes.side_effect = lambda Identities: { + "VerificationAttributes": {domain: {"VerificationStatus": "Success"} for domain in Identities} + } + ses.get_identity_dkim_attributes.side_effect = lambda Identities: { + "DkimAttributes": {domain: {"DkimVerificationStatus": dkim_status} for domain in Identities} + } + ses.get_identity_mail_from_domain_attributes.side_effect = lambda Identities: { + "MailFromDomainAttributes": { + domain: {"MailFromDomainStatus": "Success", "MailFromDomain": mail_from_domain or f"feedback.{domain}"} + for domain in Identities + } + } + ses.get_caller_identity.return_value = {"Account": "123456789012"} + ses.list_resource_tenants.return_value = { + "ResourceTenants": [{"TenantName": f"team-{self.team.id}"}] if has_tenant else [] + } + mock_resolver.return_value.resolve.return_value = [SimpleNamespace(strings=[b"v=DMARC1; p=none;"])] + if not has_dmarc: + mock_resolver.return_value.resolve.side_effect = dns.resolver.NXDOMAIN() + ses.verify_domain_identity.side_effect = RuntimeError("Background checks must not create SES identities") + ses.verify_domain_dkim.side_effect = RuntimeError("Background checks must not change DKIM") + ses.set_identity_mail_from_domain.side_effect = RuntimeError("Background checks must not change MAIL FROM") + return ses + + def _timestamp(self, age: timedelta) -> int: + return int((timezone.now() - age).timestamp()) + + def _verified(self, *senders: Integration) -> list[bool]: + return [ + self.client.get(f"/api/projects/{self.team.id}/integrations/{sender.id}/").json()["config"]["verified"] + for sender in senders + ] + + def _default_sender_id(self) -> int | None: + return self.client.get(f"/api/environments/{self.team.id}/").json()["workflows_config"][ + "default_email_integration_id" + ] + + @parameterized.expand( + [ + param("verified", expected_verified=True), + param("dkim_pending", dkim_status="Pending"), + param("dmarc_missing", has_dmarc=False), + param("tenant_missing", has_tenant=False), + param("mail_from_changed", mail_from_domain="other.example.com"), + param("mail_from_differs_in_case", mail_from_domain="Feedback.Example.com.", expected_verified=True), + ] + ) + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_background_verification_updates_sender_status_without_ui( + self, + _name: str, + mock_boto_client: MagicMock, + mock_resolver: MagicMock, + dkim_status: str = "Success", + has_dmarc: bool = True, + has_tenant: bool = True, + mail_from_domain: str | None = None, + expected_verified: bool = False, + ) -> None: + sender = self._create_sender("example.com") + ses = self._mock_ses( + mock_boto_client, + mock_resolver, + dkim_status=dkim_status, + has_dmarc=has_dmarc, + has_tenant=has_tenant, + mail_from_domain=mail_from_domain, + ) + + refresh_new_email_senders() + + assert self._verified(sender) == [expected_verified] + assert self._default_sender_id() == (sender.id if expected_verified else None) + assert [call.kwargs["Identities"] for call in ses.get_identity_verification_attributes.call_args_list] == [ + ["example.com"] + ] + + if expected_verified: + mock_boto_client.reset_mock() + refresh_new_email_senders() + mock_boto_client.assert_not_called() + + @parameterized.expand( + [ + param("new_sender_every_minute", age=timedelta(minutes=10), sweep=refresh_new_email_senders, checked=True), + param( + "new_sender_not_in_five_minute_sweep", age=timedelta(minutes=10), sweep=refresh_settled_email_senders + ), + param( + "settled_sender_every_five_minutes", + age=timedelta(hours=2), + sweep=refresh_settled_email_senders, + checked=True, + ), + param("settled_sender_not_in_minute_sweep", age=timedelta(hours=2), sweep=refresh_new_email_senders), + param("stale_sender", age=timedelta(days=8), sweep=refresh_settled_email_senders), + param( + "old_sender_with_recent_request_every_minute", + age=timedelta(days=30), + requested_age=timedelta(minutes=10), + sweep=refresh_new_email_senders, + checked=True, + ), + param( + "new_sender_with_stale_request", + age=timedelta(minutes=10), + requested_age=timedelta(days=8), + sweep=refresh_new_email_senders, + ), + ] + ) + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_sweeps_check_senders_by_age( + self, + _name: str, + mock_boto_client: MagicMock, + mock_resolver: MagicMock, + age: timedelta, + sweep: Callable[[], None], + requested_age: timedelta | None = None, + checked: bool = False, + ) -> None: + requested_at: dict[str, Any] = ( + {} if requested_age is None else {"verification_requested_at": self._timestamp(requested_age)} + ) + sender = self._create_sender("example.com", **requested_at) + Integration.objects.filter(id=sender.id).update(created_at=timezone.now() - age) + ses = self._mock_ses(mock_boto_client, mock_resolver) + + sweep() + + assert ses.get_identity_verification_attributes.called is checked + assert self._verified(sender) == [checked] + + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_advisor_event_makes_pending_senders_on_its_domain_due_on_the_next_minute_sweep( + self, mock_boto_client: MagicMock, mock_resolver: MagicMock + ) -> None: + sender, other = (self._create_sender(domain) for domain in ["example.com", "other.example.com"]) + Integration.objects.filter(id__in=[sender.id, other.id]).update(created_at=timezone.now() - timedelta(days=30)) + ses = self._mock_ses(mock_boto_client, mock_resolver) + + EmailSenderVerification.request_domain_check("Example.com") + refresh_new_email_senders() + + assert self._verified(sender, other) == [True, False] + assert [call.kwargs["Identities"] for call in ses.get_identity_verification_attributes.call_args_list] == [ + ["example.com"] + ] + + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_advisor_requested_sender_whose_check_fails_is_retried_on_the_next_sweep( + self, mock_boto_client: MagicMock, mock_resolver: MagicMock + ) -> None: + sender = self._create_sender("example.com") + Integration.objects.filter(id=sender.id).update(created_at=timezone.now() - timedelta(days=30)) + ses = self._mock_ses(mock_boto_client, mock_resolver) + ses.list_resource_tenants.side_effect = [ + ClientError({"Error": {"Code": "Throttling", "Message": "Rate exceeded"}}, "ListResourceTenants"), + {"ResourceTenants": [{"TenantName": f"team-{self.team.id}"}]}, + ] + EmailSenderVerification.request_domain_check("example.com") + + refresh_new_email_senders() + assert self._verified(sender) == [False] + + refresh_new_email_senders() + assert self._verified(sender) == [True] + + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_tenant_reads_for_ready_senders_are_spaced_apart( + self, mock_boto_client: MagicMock, mock_resolver: MagicMock + ) -> None: + senders = [self._create_sender(domain) for domain in ["first.example.com", "second.example.com"]] + ses = self._mock_ses(mock_boto_client, mock_resolver) + tenant_read_times: list[float] = [] + + def read_tenants(ResourceArn: str) -> dict[str, list[dict[str, str]]]: + tenant_read_times.append(self.ses_clock["now"]) + return {"ResourceTenants": [{"TenantName": f"team-{self.team.id}"}]} + + ses.list_resource_tenants.side_effect = read_tenants + + refresh_new_email_senders() + + assert self._verified(*senders) == [True, True] + assert tenant_read_times == [0.0, 2.0] + + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_sender_edited_during_check_still_verifies_siblings_on_the_checked_subdomain( + self, mock_boto_client: MagicMock, mock_resolver: MagicMock + ) -> None: + sender = self._create_sender("example.com") + sibling = self._create_sender("example.com", local_part="sibling") + ses = self._mock_ses(mock_boto_client, mock_resolver) + + def edit_sender(**kwargs: str) -> dict[str, list[dict[str, str]]]: + Integration.objects.filter(id=sender.id).update(config={**sender.config, "mail_from_subdomain": "updated"}) + return {"ResourceTenants": [{"TenantName": f"team-{self.team.id}"}]} + + ses.list_resource_tenants.side_effect = edit_sender + + refresh_new_email_senders() + + assert self._verified(sender, sibling) == [False, True] + assert self._default_sender_id() is None + + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_overlapping_sweep_returns_without_checking( + self, mock_boto_client: MagicMock, mock_resolver: MagicMock + ) -> None: + sender = self._create_sender("example.com") + ses = self._mock_ses(mock_boto_client, mock_resolver) + read_statuses = ses.get_identity_verification_attributes.side_effect + + def overlap_sweep(Identities: list[str]) -> dict[str, dict[str, dict[str, str]]]: + ses.get_identity_verification_attributes.side_effect = read_statuses + refresh_new_email_senders() + return read_statuses(Identities=Identities) + + ses.get_identity_verification_attributes.side_effect = overlap_sweep + + refresh_new_email_senders() + + assert ses.get_identity_verification_attributes.call_count == 1 + assert self._verified(sender) == [True] + + @parameterized.expand( + [ + param("provider_failure", fail_first=True, first_sweep=[False, True], second_sweep=[True, True]), + param("budget_exhausted", slow_checks=True, first_sweep=[True, False], second_sweep=[True, True]), + ] + ) + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") + @patch("products.workflows.backend.providers.ses.boto3.client") + def test_interrupted_sweep_does_not_block_other_senders_and_recovers_on_next_sweep( + self, + _name: str, + mock_boto_client: MagicMock, + mock_resolver: MagicMock, + first_sweep: list[bool], + second_sweep: list[bool], + fail_first: bool = False, + slow_checks: bool = False, + ) -> None: + senders = [self._create_sender(domain) for domain in ["first.example.com", "second.example.com"]] + ses = self._mock_ses(mock_boto_client, mock_resolver) + clock = {"now": 0.0} + + def check_tenants(ResourceArn: str) -> dict[str, list[dict[str, str]]]: + if slow_checks: + clock["now"] += 200 + if fail_first and ResourceArn.endswith("/first.example.com"): + raise EndpointConnectionError(endpoint_url="https://ses.example.com") + return {"ResourceTenants": [{"TenantName": f"team-{self.team.id}"}]} + + ses.list_resource_tenants.side_effect = check_tenants + + with patch( + "products.workflows.backend.services.email_sender_verification.monotonic", side_effect=lambda: clock["now"] + ): + refresh_new_email_senders() + assert self._verified(*senders) == first_sweep + assert [call.kwargs["Identities"] for call in ses.get_identity_verification_attributes.call_args_list] == [ + ["first.example.com", "second.example.com"] + ] + + fail_first = False + arrival = self._create_sender("new.example.com") + ses.get_identity_verification_attributes.reset_mock() + refresh_new_email_senders() + assert self._verified(*senders, arrival) == [*second_sweep, not slow_checks] + + refresh_new_email_senders() + assert self._verified(*senders, arrival) == [True, True, True] diff --git a/products/workflows/backend/test/test_ses_provider.py b/products/workflows/backend/test/test_ses_provider.py index 6eab44e15ddf..fcce26202c51 100644 --- a/products/workflows/backend/test/test_ses_provider.py +++ b/products/workflows/backend/test/test_ses_provider.py @@ -261,15 +261,20 @@ def test_verify_email_domain_success(self, mock_resolver_cls): } mock_dkim_attrs.return_value = {"DkimAttributes": {TEST_DOMAIN: {"DkimVerificationStatus": "Success"}}} mock_mail_from_attrs.return_value = { - "MailFromDomainAttributes": {TEST_DOMAIN: {"MailFromDomainStatus": "Success"}} + "MailFromDomainAttributes": { + TEST_DOMAIN: {"MailFromDomainStatus": "Success", "MailFromDomain": f"mail.{TEST_DOMAIN}"} + } } mock_list_tenants.return_value = {"ResourceTenants": [{"TenantName": "team-1"}]} - result = provider.verify_email_domain(TEST_DOMAIN, mail_from_subdomain="mail", team_id=1) + with patch("products.workflows.backend.providers.ses.sleep") as mock_sleep: + result = provider.verify_email_domain(TEST_DOMAIN, mail_from_subdomain="mail", team_id=1) + provider.verify_email_domain(TEST_DOMAIN, mail_from_subdomain="mail", team_id=1) # Should return verified status with DNS records assert result["status"] == "success" assert len(result["dnsRecords"]) > 0 # Records are now always returned + mock_sleep.assert_not_called() @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") def test_verify_email_domain_pending_when_dmarc_missing(self, mock_resolver_cls): @@ -285,7 +290,9 @@ def test_verify_email_domain_pending_when_dmarc_missing(self, mock_resolver_cls) mock_verif_attrs.return_value = {"VerificationAttributes": {TEST_DOMAIN: {"VerificationStatus": "Success"}}} mock_dkim_attrs.return_value = {"DkimAttributes": {TEST_DOMAIN: {"DkimVerificationStatus": "Success"}}} mock_mail_from_attrs.return_value = { - "MailFromDomainAttributes": {TEST_DOMAIN: {"MailFromDomainStatus": "Success"}} + "MailFromDomainAttributes": { + TEST_DOMAIN: {"MailFromDomainStatus": "Success", "MailFromDomain": f"mail.{TEST_DOMAIN}"} + } } result = provider.verify_email_domain(TEST_DOMAIN, mail_from_subdomain="mail", team_id=1) diff --git a/products/workflows/backend/test/test_ses_tenant_events.py b/products/workflows/backend/test/test_ses_tenant_events.py index 8b4f7aa2d9e8..cc8253df34eb 100644 --- a/products/workflows/backend/test/test_ses_tenant_events.py +++ b/products/workflows/backend/test/test_ses_tenant_events.py @@ -106,6 +106,9 @@ def setUp(self) -> None: sync = patch("products.workflows.backend.services.ses_tenant_events.sync_ses_tenant_state_task") self.sync_mock = sync.start() self.addCleanup(sync.stop) + sender_check = patch("products.workflows.backend.services.ses_tenant_events.EmailSenderVerification") + self.sender_check_mock = sender_check.start() + self.addCleanup(sender_check.stop) def test_enqueues_a_sync_for_the_tenant_named_in_the_event(self) -> None: accept_ses_event(_delivery(_notification(_eventbridge_event()))) @@ -124,6 +127,18 @@ def test_ignores_events_from_other_sources(self) -> None: assert not self.sync_mock.delay.called + def test_resolved_advisor_finding_checks_the_senders_on_that_identity(self) -> None: + event = _eventbridge_event( + **{"detail-type": "Advisor Recommendation Status Resolved"}, + detail={"version": "1.0.0", "data": "DKIM record was not found."}, + resources=["arn:aws:ses:us-east-1:123456789012:identity/team-5.example.com"], + ) + + accept_ses_event(_delivery(_notification(event))) + + self.sender_check_mock.request_domain_check.assert_called_once_with("team-5.example.com") + assert not self.sync_mock.delay.called + def test_confirms_subscriptions_by_fetching_the_subscribe_url(self) -> None: with patch("products.workflows.backend.services.ses_tenant_events.requests.get") as get_mock: accept_ses_event(_delivery(_subscription_confirmation())) From 287950a29f1a73deb42c4c1668fd58cf08ce0833 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Mon, 26 Oct 2026 20:38:21 +0000 Subject: [PATCH 2/2] fix(workflows): count a domain as verified once and match the mail from domain in status Co-Authored-By: Claude Opus 5.5 --- posthog/models/integration/email.py | 17 ++++++++++------- products/workflows/backend/providers/ses.py | 2 +- .../test/test_email_domain_connect_api.py | 13 +++++++++---- .../test/test_email_domain_status_api.py | 9 +++++++++ .../test/test_email_sender_verification.py | 18 ++++++++++++++++++ 5 files changed, 47 insertions(+), 12 deletions(-) diff --git a/posthog/models/integration/email.py b/posthog/models/integration/email.py index 3ba3334f73ab..fb2e42cb253a 100644 --- a/posthog/models/integration/email.py +++ b/posthog/models/integration/email.py @@ -175,19 +175,21 @@ def apply_verification(self, verification_result: "EmailDomainVerification") -> self._mark_domain_verified(on_domain_verified=None) return verification_result - def _matching_senders(self) -> "models.QuerySet[model.Integration]": - mail_from_subdomain = self.integration.config.get("mail_from_subdomain", "feedback") - matching_mail_from = Q(config__mail_from_subdomain=mail_from_subdomain) - if mail_from_subdomain == "feedback": - matching_mail_from |= Q(config__mail_from_subdomain__isnull=True) + def _domain_senders(self) -> "models.QuerySet[model.Integration]": return model.Integration.objects.filter( - matching_mail_from, team_id=self.integration.team_id, kind="email", config__domain=self.integration.config.get("domain"), config__provider=self.integration.config.get("provider", "ses"), ) + def _matching_senders(self) -> "models.QuerySet[model.Integration]": + mail_from_subdomain = self.integration.config.get("mail_from_subdomain", "feedback") + matching_mail_from = Q(config__mail_from_subdomain=mail_from_subdomain) + if mail_from_subdomain == "feedback": + matching_mail_from |= Q(config__mail_from_subdomain__isnull=True) + return self._domain_senders().filter(matching_mail_from) + def verify(self, *, on_domain_verified: Callable[[], None] | None = None) -> "EmailDomainVerification": domain = self.integration.config.get("domain") provider = self.integration.config.get("provider", "ses") @@ -290,6 +292,7 @@ def _mark_domain_verified(self, on_domain_verified: Callable[[], None] | None) - senders = list(self._matching_senders().select_for_update().order_by("id")) sender_ids = [sender.id for sender in senders] unverified_ids = [sender.id for sender in senders if not sender.config.get("verified")] + domain_was_verified = self._domain_senders().filter(config__verified=True).exists() model.Integration.objects.filter(id__in=unverified_ids).update(config=_with_config_key("verified", True)) # A concurrent edit can move this sender off the checked MAIL FROM subdomain, so it may not be among them. verifies_this_sender = self.integration.id in sender_ids @@ -306,7 +309,7 @@ def _mark_domain_verified(self, on_domain_verified: Callable[[], None] | None) - return self.integration.config["verified"] = True # A sender added to a domain that is already verified joins it, so only the first verification counts. - if on_domain_verified and len(unverified_ids) == len(senders): + if on_domain_verified and not domain_was_verified: on_domain_verified() # Workers also refresh integrations on their own schedule, so a failed reload only delays sending. diff --git a/products/workflows/backend/providers/ses.py b/products/workflows/backend/providers/ses.py index 3bf5bccc96a7..05c91d6dfd9d 100644 --- a/products/workflows/backend/providers/ses.py +++ b/products/workflows/backend/providers/ses.py @@ -374,7 +374,7 @@ class _SESIdentityState: # SES keeps one MAIL FROM domain per identity, and senders on the same domain can ask for # different ones. The status SES reports belongs to whichever was set last. def for_mail_from_domain(self, mail_from_domain: str) -> "_SESIdentityState": - if self.mail_from_domain == mail_from_domain: + if _same_domain(self.mail_from_domain, mail_from_domain): return self return replace(self, mail_from_status=_UNKNOWN) diff --git a/products/workflows/backend/test/test_email_domain_connect_api.py b/products/workflows/backend/test/test_email_domain_connect_api.py index c7fae9b448e7..9d1434e98e57 100644 --- a/products/workflows/backend/test/test_email_domain_connect_api.py +++ b/products/workflows/backend/test/test_email_domain_connect_api.py @@ -11,6 +11,7 @@ from rest_framework import status from posthog.models.integration import EmailIntegration, Integration +from posthog.models.integration.email import VERIFICATION_REQUESTED_AT from products.workflows.backend.test.email_domain_api_test_case import ( ALL_KINDS, @@ -201,6 +202,10 @@ def test_sender_update_keeps_a_verification_written_after_the_sender_was_loaded( class TestDomainConnectChaos(EmailDomainConnectTestCase): + # Every apply-url call asks SES to verify, which restamps when verification was requested. + def setup_config(self, sender: int) -> dict[str, Any]: + return {key: value for key, value in self.config(sender).items() if key != VERIFICATION_REQUESTED_AT} + @parameterized.expand( [ ("discovery_lookup_times_out", dns_operation(DISCOVERY_RECORD, "TXT"), DnsFault.TIMEOUT), @@ -215,14 +220,14 @@ def test_transient_provider_failure_falls_back_to_manual_setup_and_recovers( self, _name: str, operation: str, fault: DnsFault | HttpFault ) -> None: sender = self.create_sender(f"hello@{DOMAIN}") - config_before = self.config(sender) + config_before = self.setup_config(sender) self.world.faults.inject(operation, fault) assert self.check()["supported"] is False response = self.apply_url(sender) assert (response.status_code, response.json()["detail"]) == (status.HTTP_400_BAD_REQUEST, APPLY_URL_FAILED) - assert self.config(sender) == config_before + assert self.setup_config(sender) == config_before self.world.faults.clear() assert self.check()["supported"] is True @@ -270,7 +275,7 @@ def test_unusable_provider_answers_400_and_leaves_the_sender_unchanged( self, _name: str, arrange: Callable[[EmailDomainConnectTestCase], object], supported: bool, detail: str ) -> None: sender = self.create_sender(f"hello@{DOMAIN}") - config_before = self.config(sender) + config_before = self.setup_config(sender) arrange(self) check = self.check() @@ -278,7 +283,7 @@ def test_unusable_provider_answers_400_and_leaves_the_sender_unchanged( assert (check["supported"], bool(check["available_providers"])) == (supported, not supported) assert (response.status_code, response.json()["detail"]) == (status.HTTP_400_BAD_REQUEST, detail) - assert self.config(sender) == config_before + assert self.setup_config(sender) == config_before @override_settings(SES_REGION="us-east-1") diff --git a/products/workflows/backend/test/test_email_domain_status_api.py b/products/workflows/backend/test/test_email_domain_status_api.py index add5cce7168f..61ae0250fac2 100644 --- a/products/workflows/backend/test/test_email_domain_status_api.py +++ b/products/workflows/backend/test/test_email_domain_status_api.py @@ -111,6 +111,15 @@ def test_status_reads_ses_without_writing_to_it( assert self.world.ses.writes == writes_after_create + def test_status_matches_a_mail_from_domain_that_differs_in_case(self) -> None: + sender = self.create_sender(f"hello@{DOMAIN}") + self.publish(self.records_to_publish(sender), *ALL_KINDS) + identity = self.world.ses.identities[DOMAIN] + assert identity.mail_from_domain is not None + identity.mail_from_domain = f"{identity.mail_from_domain.upper()}." + + assert self.status_body(sender)["status"] == "verified" + def test_status_is_cached_until_refresh(self) -> None: sender = self.create_sender(f"hello@{DOMAIN}") records = self.records_to_publish(sender) diff --git a/products/workflows/backend/test/test_email_sender_verification.py b/products/workflows/backend/test/test_email_sender_verification.py index 951f643cb118..e4e06b4b9f37 100644 --- a/products/workflows/backend/test/test_email_sender_verification.py +++ b/products/workflows/backend/test/test_email_sender_verification.py @@ -13,6 +13,7 @@ from parameterized import param, parameterized from posthog.models.integration import Integration +from posthog.models.integration.email import EmailIntegration from posthog.redis import get_client from products.workflows.backend.facade.tasks import refresh_new_email_senders, refresh_settled_email_senders @@ -275,6 +276,23 @@ def edit_sender(**kwargs: str) -> dict[str, list[dict[str, str]]]: assert self._verified(sender, sibling) == [False, True] assert self._default_sender_id() is None + @patch("posthog.models.integration.email.reload_integrations_on_workers") + @patch( + "products.workflows.backend.facade.api.verify_ses_email_domain", + return_value={"status": "success", "dnsRecords": []}, + ) + def test_a_domain_counts_as_verified_once_across_mail_from_subdomains( + self, _verify: MagicMock, _reload: MagicMock + ) -> None: + self._create_sender("example.com", verified=True) + other_subdomain = self._create_sender("example.com", local_part="other", mail_from_subdomain="bounce") + on_domain_verified = MagicMock() + + EmailIntegration(other_subdomain).verify(on_domain_verified=on_domain_verified) + + assert self._verified(other_subdomain) == [True] + on_domain_verified.assert_not_called() + @patch("products.workflows.backend.providers.ses.dns.resolver.Resolver") @patch("products.workflows.backend.providers.ses.boto3.client") def test_overlapping_sweep_returns_without_checking(