diff --git a/posthog/api/test/test_integration.py b/posthog/api/test/test_integration.py index 357c7463157f..bf900f6b0656 100644 --- a/posthog/api/test/test_integration.py +++ b/posthog/api/test/test_integration.py @@ -666,6 +666,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 @@ -724,7 +725,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 @@ -741,6 +744,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") @@ -778,6 +782,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") @@ -802,7 +807,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, @@ -829,10 +840,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 eb0922152a84..ed94e60838d3 100644 --- a/posthog/models/integration/__init__.py +++ b/posthog/models/integration/__init__.py @@ -38,7 +38,7 @@ 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, @@ -276,4 +276,5 @@ "RedshiftIntegration", "SnowflakeIntegrationError", "SnowflakeIntegration", + "mark_verification_requested", ] diff --git a/posthog/models/integration/email.py b/posthog/models/integration/email.py index 132309a4899d..fb2e42cb253a 100644 --- a/posthog/models/integration/email.py +++ b/posthog/models/integration/email.py @@ -8,6 +8,7 @@ from django.conf import settings from django.core.cache import cache 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 @@ -39,6 +40,23 @@ class EmailDomainStatusReport: checked_at: datetime +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 @@ -100,6 +118,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, }, @@ -151,6 +170,26 @@ 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": + self._mark_domain_verified(on_domain_verified=None) + return verification_result + + def _domain_senders(self) -> "models.QuerySet[model.Integration]": + return model.Integration.objects.filter( + 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") @@ -180,6 +219,8 @@ def verify(self, *, on_domain_verified: Callable[[], None] | None = None) -> "Em if verification_result.get("status") == "success": self._mark_domain_verified(on_domain_verified) + else: + mark_verification_requested(model.Integration.objects.filter(id=self.integration.id)) return verification_result @@ -248,39 +289,34 @@ def _check_domain(self) -> "EmailDomainCheck": def _mark_domain_verified(self, on_domain_verified: Callable[[], None] | None) -> None: with transaction.atomic(): - senders = list( - model.Integration.objects.select_for_update() - .filter( - team_id=self.integration.team_id, - kind="email", - config__domain=self.integration.config.get("domain"), - config__provider=self.integration.config.get("provider", "ses"), + 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 + if verifies_this_sender: + from products.workflows.backend.facade.api import ( + set_default_email_sender_if_unset, # noqa: PLC0415 — keeps the workflows facade off the model import path ) - .order_by("id") - ) - unverified_senders = [sender for sender in senders if not sender.config.get("verified")] - for sender in unverified_senders: - sender.config["verified"] = True - sender.save() - self.integration.config["verified"] = True - - 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) + set_default_email_sender_if_unset(self.integration.team_id, self.integration.id) + if sender_ids: + self._reload_senders_on_workers(sender_ids) + if not verifies_this_sender: + 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_senders) == len(senders): + if on_domain_verified and not domain_was_verified: on_domain_verified() - self._reload_senders_on_workers(senders) # Workers also refresh integrations on their own schedule, so a failed reload only delays sending. # Failing the request instead would hide a verification that is already saved. - def _reload_senders_on_workers(self, senders: list[model.Integration]) -> None: + def _reload_senders_on_workers(self, sender_ids: list[int]) -> None: try: - reload_integrations_on_workers(self.integration.team_id, [sender.id for sender in senders]) + reload_integrations_on_workers(self.integration.team_id, sender_ids) except Exception as error: capture_exception(error, {"team_id": self.integration.team_id}) diff --git a/posthog/tasks/scheduled.py b/posthog/tasks/scheduled.py index ce357e1508bf..24ebc961f74b 100644 --- a/posthog/tasks/scheduled.py +++ b/posthog/tasks/scheduled.py @@ -159,6 +159,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, ) @@ -339,6 +341,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 69f264323ab7..37c8845f53a6 100644 --- a/products/workflows/backend/facade/tasks.py +++ b/products/workflows/backend/facade/tasks.py @@ -1,4 +1,8 @@ from products.workflows.backend.tasks.batch_jobs import resume_batch_jobs_waiting_for_realtime_cohort +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 ( @@ -15,5 +19,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 4ecb6490fda4..05c91d6dfd9d 100644 --- a/products/workflows/backend/providers/ses.py +++ b/products/workflows/backend/providers/ses.py @@ -6,7 +6,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, Literal, cast from django.conf import settings @@ -40,6 +40,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 @@ -94,6 +95,42 @@ # The status endpoint is polled and holds a request worker, so its reads fail fast and the next poll retries. STATUS_READ_CONFIG = Config(connect_timeout=2, read_timeout=5, retries={"mode": "standard", "max_attempts": 2}) +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: @@ -334,21 +371,10 @@ class _SESIdentityState: mail_from_status: str mail_from_domain: str | None - @classmethod - def unknown(cls) -> "_SESIdentityState": - return cls( - verification_status=_UNKNOWN, - verification_token=None, - dkim_status=_UNKNOWN, - dkim_tokens=(), - mail_from_status=_UNKNOWN, - mail_from_domain=None, - ) - # 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) @@ -403,26 +429,29 @@ class SESProvider: ses_status_client: "SESClient" ses_v2_status_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=STATUS_READ_CONFIG, + config=client_config or STATUS_READ_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. @@ -459,6 +488,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}" @@ -605,6 +638,7 @@ def _associate_tenant_resource(self, tenant_name: str, resource_arn: str) -> Non raise def _list_identity_tenants(self, domain: str, client: "SESV2Client | None" = None) -> set[str]: + self._pace_read("resource_tenants") try: resp = (client or self.ses_v2_client).list_resource_tenants(ResourceArn=self._identity_arn(domain)) except ClientError as e: @@ -711,6 +745,20 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in 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: # DMARC is in the list too: AWS SES has no method to check its presence, so we do a direct # DNS lookup further below and include the result in the overall status. dns_records = [ @@ -724,10 +772,11 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in ) ] - identity = self._read_identity_state_or_unknown(domain) - verification_status = identity.verification_status - dkim_status = identity.dkim_status - mail_from_status = identity.mail_from_status + mail_from_domain = f"{mail_from_subdomain}.{domain}" + 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(mail_from_domain) # DMARC: check via direct DNS lookup since AWS SES doesn't track it dmarc_status = "Pending" @@ -749,9 +798,9 @@ def verify_email_domain(self, domain: str, mail_from_subdomain: str, team_id: in # Normalize overall status overall: EmailDomainVerificationStatus - if identity.is_ready(dmarc_published=dmarc_status == "Success"): + if status.is_ready_for(mail_from_domain) and dmarc_status == "Success": overall = "success" - elif "Failed" in identity.statuses: + elif "Failed" in (verification_status, dkim_status, mail_from_status): overall = "failed" else: overall = "pending" @@ -852,13 +901,6 @@ def _ses_region(self) -> str: def _is_associated_with_team(self, domain: str, team_id: int, client: "SESV2Client | None" = None) -> bool: return self._tenant_name_for_team(team_id) in self._list_identity_tenants(domain, client) - def _read_identity_state_or_unknown(self, domain: str) -> "_SESIdentityState": - try: - return self._read_identity_state(domain) - except ClientError: - logger.warning("Could not read the SES identity attributes for %s", domain, exc_info=True) - return _SESIdentityState.unknown() - def _read_identity_state(self, domain: str, client: "SESClient | None" = None) -> "_SESIdentityState": ses = client or self.ses_client id_for_domain = ses.get_identity_verification_attributes(Identities=[domain])["VerificationAttributes"].get( @@ -877,6 +919,51 @@ def _read_identity_state(self, domain: str, client: "SESClient | None" = None) - mail_from_domain=mail_from_for_domain.get("MailFromDomain") if mail_from_for_domain else None, ) + 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_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_integration_ses.py b/products/workflows/backend/test/test_email_integration_ses.py index 11d2c9cb1a3b..5b9f71894431 100644 --- a/products/workflows/backend/test/test_email_integration_ses.py +++ b/products/workflows/backend/test/test_email_integration_ses.py @@ -44,7 +44,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..e4e06b4b9f37 --- /dev/null +++ b/products/workflows/backend/test/test_email_sender_verification.py @@ -0,0 +1,364 @@ +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.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 +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("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( + 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 27b63586f39a..2fe3e348f6d2 100644 --- a/products/workflows/backend/test/test_ses_provider.py +++ b/products/workflows/backend/test/test_ses_provider.py @@ -262,15 +262,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): @@ -286,7 +291,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()))