import html import json import logging import re import shlex from datetime import UTC, datetime from email import message_from_bytes from email.iterators import _structure from email.message import EmailMessage from email.utils import parseaddr from io import StringIO from json import JSONDecodeError from textwrap import dedent from typing import Any, Literal, NamedTuple, TypedDict, TypeVar from urllib.parse import urlencode from uuid import uuid4 from django.conf import settings from django.contrib.auth.models import User from django.core.exceptions import ObjectDoesNotExist from django.db import transaction from django.db.models import prefetch_related_objects from django.http import HttpRequest, HttpResponse from django.shortcuts import render from django.template.loader import render_to_string from django.utils.html import escape from django.views.decorators.csrf import csrf_exempt from botocore.exceptions import ClientError from codetiming import Timer from decouple import strtobool from markus.utils import generate_tag from sentry_sdk import capture_message from waffle import get_waffle_flag_model, sample_is_active from privaterelay.ftl_bundles import main as ftl_bundle from privaterelay.models import Profile from privaterelay.utils import ( flag_is_active_in_task, get_subplat_upgrade_link_by_language, glean_logger, ) from .exceptions import CannotMakeAddressException from .models import ( DeletedAddress, DomainAddress, RelayAddress, Reply, address_hash, get_domain_numerical, ) from .policy import relay_policy from .sns import SUPPORTED_SNS_TYPES, verify_from_sns from .types import ( AWS_MailJSON, AWS_SNSMessageJSON, EmailForwardingIssues, EmailHeaderIssues, OutgoingHeaders, ) from .utils import ( InvalidFromHeader, _get_bucket_and_key_from_s3_json, b64_lookup_key, count_all_trackers, decrypt_reply_metadata, derive_reply_keys, encode_dict_gza85, encrypt_reply_metadata, generate_from_header, get_domains_from_settings, get_message_content_from_s3, get_message_id_bytes, get_reply_to_address, histogram_if_enabled, incr_if_enabled, parse_email_header, remove_message_from_s3, remove_trackers, ses_send_raw_email, urlize_and_linebreaks, ) logger = logging.getLogger("events") info_logger = logging.getLogger("eventsinfo") class ReplyHeadersNotFound(Exception): def __init__(self, message="No In-Reply-To or References headers."): self.message = message def first_time_user_test(request): """ Demonstrate rendering of the "First time Relay user" email. Settings like language can be given in the querystring, otherwise settings come from a random free profile. """ in_bundle_country = strtobool(request.GET.get("in_bundle_country", "yes")) email_context = { "in_bundle_country": in_bundle_country, "SITE_ORIGIN": settings.SITE_ORIGIN, } if request.GET.get("format", "html") == "text": return render( request, "emails/first_time_user.txt", email_context, "text/plain; charset=utf-8", ) return render(request, "emails/first_time_user.html", email_context) def reply_requires_premium_test(request): """ Demonstrate rendering of the "Reply requires premium" email. Settings like language can be given in the querystring, otherwise settings come from a random free profile. """ email_context = { "sender": "test@example.com", "forwarded": True, "SITE_ORIGIN": settings.SITE_ORIGIN, } for param in request.GET: email_context[param] = request.GET.get(param) if param == "forwarded" and request.GET[param] == "True": email_context[param] = True for param in request.GET: if param == "content-type" and request.GET[param] == "text/plain": return render( request, "emails/reply_requires_premium.txt", email_context, "text/plain; charset=utf-8", ) return render(request, "emails/reply_requires_premium.html", email_context) def disabled_mask_for_spam_test(request): """ Demonstrate rendering of the "Disabled mask for spam" email. Settings like language can be given in the querystring, otherwise settings come from a random free profile. """ mask = "abc123456@mozmail.com" email_context = { "mask": mask, "SITE_ORIGIN": settings.SITE_ORIGIN, } for param in request.GET: email_context[param] = request.GET.get(param) for param in request.GET: if param == "content-type" and request.GET[param] == "text/plain": return render( request, "emails/disabled_mask_for_spam.txt", email_context, "text/plain; charset=utf-8", ) return render(request, "emails/disabled_mask_for_spam.html", email_context) def first_forwarded_email_test(request: HttpRequest) -> HttpResponse: # TO DO: Update with correct context when trigger is created first_forwarded_email_html = render_to_string( "emails/first_forwarded_email.html", { "SITE_ORIGIN": settings.SITE_ORIGIN, }, ) wrapped_email = wrap_html_email( first_forwarded_email_html, "en-us", True, "test@example.com", 0, ) return HttpResponse(wrapped_email) def wrap_html_email( original_html: str, language: str, has_premium: bool, display_email: str, num_level_one_email_trackers_removed: int | None = None, tracker_report_link: str | None = None, ) -> str: """Add Relay banners, surveys, etc. to an HTML email""" subplat_upgrade_link = get_subplat_upgrade_link_by_language(language) email_context = { "original_html": original_html, "language": language, "has_premium": has_premium, "subplat_upgrade_link": subplat_upgrade_link, "display_email": display_email, "tracker_report_link": tracker_report_link, "num_level_one_email_trackers_removed": num_level_one_email_trackers_removed, "SITE_ORIGIN": settings.SITE_ORIGIN, } content = render_to_string("emails/wrapped_email.html", email_context) # Remove empty lines content_lines = [line for line in content.splitlines() if line.strip()] return "\n".join(content_lines) + "\n" def wrapped_email_test(request: HttpRequest) -> HttpResponse: """ Demonstrate rendering of forwarded HTML emails. Settings like language can be given in the querystring, otherwise settings come from a randomly chosen profile. """ if all(key in request.GET for key in ("language", "has_premium")): user_profile = None else: user_profile = Profile.objects.order_by("?").first() if "language" in request.GET: language = request.GET["language"] else: if user_profile is None: raise ValueError("user_profile must not be None") language = user_profile.language if "has_premium" in request.GET: has_premium = strtobool(request.GET["has_premium"]) else: if user_profile is None: raise ValueError("user_profile must not be None") has_premium = user_profile.has_premium if "num_level_one_email_trackers_removed" in request.GET: num_level_one_email_trackers_removed = int( request.GET["num_level_one_email_trackers_removed"] ) else: num_level_one_email_trackers_removed = 0 if "has_tracker_report_link" in request.GET: has_tracker_report_link = strtobool(request.GET["has_tracker_report_link"]) else: has_tracker_report_link = False if has_tracker_report_link: if num_level_one_email_trackers_removed: trackers = { "fake-tracker.example.com": num_level_one_email_trackers_removed } else: trackers = {} tracker_report_link = ( "/tracker-report/#{" '"sender": "sender@example.com", ' '"received_at": 1658434657, ' f'"trackers": { json.dumps(trackers) }' "}" ) else: tracker_report_link = "" path = "/emails/wrapped_email_test" old_query = { "language": language, "has_premium": "Yes" if has_premium else "No", "has_tracker_report_link": "Yes" if has_tracker_report_link else "No", "num_level_one_email_trackers_removed": str( num_level_one_email_trackers_removed ), } def switch_link(key, value): if old_query[key] == value: return str(value) new_query = old_query.copy() new_query[key] = value return f'{value}' html_content = dedent( f"""\

Email rendering Test

Settings: (clear all)

""" ) wrapped_email = wrap_html_email( original_html=html_content, language=language, has_premium=has_premium, tracker_report_link=tracker_report_link, display_email="test@relay.firefox.com", num_level_one_email_trackers_removed=num_level_one_email_trackers_removed, ) return HttpResponse(wrapped_email) def _store_reply_record( mail: AWS_MailJSON, message_id: str, address: RelayAddress | DomainAddress ) -> AWS_MailJSON: # After relaying email, store a Reply record for it reply_metadata = {} for header in mail["headers"]: if header["name"].lower() in ["message-id", "from", "reply-to"]: reply_metadata[header["name"].lower()] = header["value"] message_id_bytes = get_message_id_bytes(message_id) (lookup_key, encryption_key) = derive_reply_keys(message_id_bytes) lookup = b64_lookup_key(lookup_key) encrypted_metadata = encrypt_reply_metadata(encryption_key, reply_metadata) reply_create_args: dict[str, Any] = { "lookup": lookup, "encrypted_metadata": encrypted_metadata, } if isinstance(address, DomainAddress): reply_create_args["domain_address"] = address else: if not isinstance(address, RelayAddress): raise TypeError("address must be type RelayAddress") reply_create_args["relay_address"] = address Reply.objects.create(**reply_create_args) return mail @csrf_exempt def sns_inbound(request): incr_if_enabled("sns_inbound", 1) # First thing we do is verify the signature json_body = json.loads(request.body) verified_json_body = verify_from_sns(json_body) # Validate ARN and message type topic_arn = verified_json_body.get("TopicArn", None) message_type = verified_json_body.get("Type", None) error_details = validate_sns_arn_and_type(topic_arn, message_type) if error_details: logger.error("validate_sns_arn_and_type_error", extra=error_details) return HttpResponse(error_details["error"], status=400) return _sns_inbound_logic(topic_arn, message_type, verified_json_body) def validate_sns_arn_and_type( topic_arn: str | None, message_type: str | None ) -> dict[str, Any] | None: """ Validate Topic ARN and SNS Message Type. If an error is detected, the return is a dictionary of error details. If no error is detected, the return is None. """ if not topic_arn: error = "Received SNS request without Topic ARN." elif topic_arn not in settings.AWS_SNS_TOPIC: error = "Received SNS message for wrong topic." elif not message_type: error = "Received SNS request without Message Type." elif message_type not in SUPPORTED_SNS_TYPES: error = "Received SNS message for unsupported Type." else: error = None if error: return { "error": error, "received_topic_arn": shlex.quote(topic_arn) if topic_arn else topic_arn, "supported_topic_arn": sorted(settings.AWS_SNS_TOPIC), "received_sns_type": ( shlex.quote(message_type) if message_type else message_type ), "supported_sns_types": SUPPORTED_SNS_TYPES, } return None def _sns_inbound_logic(topic_arn, message_type, json_body): if message_type == "SubscriptionConfirmation": info_logger.info( "SNS SubscriptionConfirmation", extra={"SubscribeURL": json_body["SubscribeURL"]}, ) return HttpResponse("Logged SubscribeURL", status=200) if message_type == "Notification": incr_if_enabled("sns_inbound_Notification", 1) return _sns_notification(json_body) logger.error( "SNS message type did not fall under the SNS inbound logic", extra={"message_type": shlex.quote(message_type)}, ) capture_message( "Received SNS message with type not handled in inbound log", level="error", stack=True, ) return HttpResponse( "Received SNS message with type not handled in inbound log", status=400 ) def _sns_notification(json_body): try: message_json = json.loads(json_body["Message"]) except JSONDecodeError: logger.error( "SNS notification has non-JSON message body", extra={"content": shlex.quote(json_body["Message"])}, ) return HttpResponse("Received SNS notification with non-JSON body", status=400) event_type = message_json.get("eventType") notification_type = message_json.get("notificationType") if notification_type not in { "Complaint", "Received", "Bounce", } and event_type not in {"Complaint", "Bounce"}: logger.error( "SNS notification for unsupported type", extra={ "notification_type": shlex.quote(notification_type), "event_type": shlex.quote(event_type), "keys": [shlex.quote(key) for key in message_json.keys()], }, ) return HttpResponse( ( "Received SNS notification for unsupported Type: " f"{html.escape(shlex.quote(notification_type))}" ), status=400, ) response = _sns_message(message_json) bucket, object_key = _get_bucket_and_key_from_s3_json(message_json) if response.status_code < 500: remove_message_from_s3(bucket, object_key) return response def _get_recipient_with_relay_domain(recipients): domains_to_check = get_domains_from_settings().values() for recipient in recipients: for domain in domains_to_check: if domain in recipient: return recipient return None def _get_relay_recipient_from_message_json(message_json): # Go thru all To, Cc, and Bcc fields and # return the one that has a Relay domain # First check common headers for to or cc match headers_to_check = "to", "cc" common_headers = message_json["mail"]["commonHeaders"] for header in headers_to_check: if header in common_headers: recipient = _get_recipient_with_relay_domain(common_headers[header]) if recipient is not None: return parseaddr(recipient)[1] # SES-SNS sends bcc in a different part of the message recipients = message_json["receipt"]["recipients"] return _get_recipient_with_relay_domain(recipients) def _sns_message(message_json: AWS_SNSMessageJSON) -> HttpResponse: incr_if_enabled("sns_inbound_Notification_Received", 1) init_waffle_flags() notification_type = message_json.get("notificationType") event_type = message_json.get("eventType") if notification_type == "Bounce" or event_type == "Bounce": return _handle_bounce(message_json) if notification_type == "Complaint" or event_type == "Complaint": return _handle_complaint(message_json) if notification_type != "Received": raise ValueError('notification_type must be "Received"') if event_type is not None: raise ValueError("event_type must be None") return _handle_received(message_json) # Enumerate the reasons that an email was not forwarded. # This excludes emails dropped due to mask forwarding settings, # such as "block all" and "block promotional". Those are logged # as Glean email_blocked events. EmailDroppedReason = Literal[ "auto_block_spam", # Email identified as spam, user has the auto_block_spam flag "dmarc_reject_failed", # Email failed DMARC check with a reject policy "hard_bounce_pause", # The user recently had a hard bounce "soft_bounce_pause", # The user recently has a soft bounce "abuse_flag", # The user exceeded an abuse limit, like mails forwarded "user_deactivated", # The user account is deactivated "reply_requires_premium", # The email is a reply from a free user "content_missing", # Could not load the email from storage "error_from_header", # Error generating the From: header, retryable "error_storage", # Error fetching the email contents from storage (S3), retryable "error_sending", # Error sending the forwarded email (SES), retryable ] def log_email_dropped( reason: EmailDroppedReason, mask: RelayAddress | DomainAddress, is_reply: bool = False, can_retry: bool = False, ) -> None: """ Log that an email was dropped for a reason other than a mask blocking setting. This mirrors the interface of glean_logger().log_email_blocked(), which records emails dropped due to the mask's blocking setting. """ extra: dict[str, str | int | bool] = {"reason": reason} if mask.user.profile.metrics_enabled: if mask.user.profile.fxa is not None: extra["fxa_id"] = mask.user.profile.fxa.uid extra["mask_id"] = mask.metrics_id extra |= { "is_random_mask": isinstance(mask, RelayAddress), "is_reply": is_reply, "can_retry": can_retry, } info_logger.info("email_dropped", extra=extra) def _handle_received(message_json: AWS_SNSMessageJSON) -> HttpResponse: """ Handle an AWS SES received notification. For more information, see: https://docs.aws.amazon.com/ses/latest/dg/receiving-email-notifications-contents.html https://docs.aws.amazon.com/ses/latest/dg/notification-contents.html Returns (may be incomplete): * 200 if the email was sent, the Relay address is disabled, the Relay user is flagged for abuse, the email is under a bounce pause, the email was suppressed for spam, the list email was blocked, or the noreply address was the recipient. * 400 if commonHeaders entry is missing, the Relay recipient address is malformed, the email failed DMARC with reject policy, or the email is a reply chain to a non-premium user. * 404 if an S3-stored email was not found, no Relay address was found in the "To", "CC", or "BCC" fields, or the Relay address is not in the database. * 503 if the "From" address is malformed, the S3 client returned an error different from "not found", or the SES client fails And many other returns conditions if the email is a reply. The HTTP returns are an artifact from an earlier time when emails were sent to a webhook. Currently, production instead pulls events from a queue. TODO: Return a more appropriate status object TODO: Document the metrics emitted """ mail = message_json["mail"] if "commonHeaders" not in mail: logger.error("SNS message without commonHeaders") return HttpResponse( "Received SNS notification without commonHeaders.", status=400 ) common_headers = mail["commonHeaders"] receipt = message_json["receipt"] _record_receipt_verdicts(receipt, "all") to_address = _get_relay_recipient_from_message_json(message_json) if to_address is None: incr_if_enabled("no_relay_domain_in_recipient_fields", 1) return HttpResponse("Address does not exist", status=404) _record_receipt_verdicts(receipt, "relay_recipient") from_addresses = parse_email_header(common_headers["from"][0]) if not from_addresses: info_logger.error( "_handle_received: no from address", extra={ "source": mail["source"], "common_headers_from": common_headers["from"], }, ) return HttpResponse("Unable to parse From address", status=400) from_address = from_addresses[0][1] try: [to_local_portion, to_domain_portion] = to_address.split("@") except ValueError: # TODO: Add metric return HttpResponse("Malformed to field.", status=400) if to_local_portion.lower() == "noreply": incr_if_enabled("email_for_noreply_address", 1) return HttpResponse("noreply address is not supported.") try: # FIXME: this ambiguous return of either # RelayAddress or DomainAddress types makes the Rustacean in me throw # up a bit. address = _get_address(to_address) prefetch_related_objects([address.user], "socialaccount_set", "profile") user_profile = address.user.profile except ( ObjectDoesNotExist, CannotMakeAddressException, DeletedAddress.MultipleObjectsReturned, ): if to_local_portion.lower() == "replies": response = _handle_reply(from_address, message_json, to_address) else: response = HttpResponse("Address does not exist", status=404) return response _record_receipt_verdicts(receipt, "valid_user") # if this is spam and the user is set to auto-block spam, early return if user_profile.auto_block_spam and _get_verdict(receipt, "spam") == "FAIL": incr_if_enabled("email_auto_suppressed_for_spam", 1) log_email_dropped(reason="auto_block_spam", mask=address) return HttpResponse("Address rejects spam.") if _get_verdict(receipt, "dmarc") == "FAIL": policy = receipt.get("dmarcPolicy", "none") # TODO: determine action on dmarcPolicy "quarantine" if policy == "reject": log_email_dropped(reason="dmarc_reject_failed", mask=address) incr_if_enabled( "email_suppressed_for_dmarc_failure", tags=["dmarcPolicy:reject", "dmarcVerdict:FAIL"], ) return HttpResponse("DMARC failure, policy is reject", status=400) # if this user is over bounce limits, early return bounce_paused, bounce_type = user_profile.check_bounce_pause() if bounce_paused: _record_receipt_verdicts(receipt, "user_bounce_paused") incr_if_enabled(f"email_suppressed_for_{bounce_type}_bounce", 1) reason: Literal["soft_bounce_pause", "hard_bounce_pause"] = ( "soft_bounce_pause" if bounce_type == "soft" else "hard_bounce_pause" ) log_email_dropped(reason=reason, mask=address) return HttpResponse("Address is temporarily disabled.") # check if this is a reply from an external sender to a Relay user try: (lookup_key, _) = _get_keys_from_headers(mail["headers"]) reply_record = _get_reply_record_from_lookup_key(lookup_key) user_address = address address = reply_record.address message_id = _get_message_id_from_headers(mail["headers"]) # make sure the relay user is premium if not _reply_allowed(from_address, to_address, reply_record, message_id): log_email_dropped(reason="reply_requires_premium", mask=user_address) return HttpResponse("Relay replies require a premium account", status=403) except (ReplyHeadersNotFound, Reply.DoesNotExist): # if there's no In-Reply-To header, or the In-Reply-To value doesn't # match a Reply record, continue to treat this as a regular email from # an external sender to a relay user pass # if account flagged for abuse, early return if user_profile.is_flagged: log_email_dropped(reason="abuse_flag", mask=address) return HttpResponse("Address is temporarily disabled.") if not user_profile.user.is_active: log_email_dropped(reason="user_deactivated", mask=address) return HttpResponse("Account is deactivated.") # if address is set to block, early return if not address.enabled: incr_if_enabled("email_for_disabled_address", 1) address.num_blocked += 1 address.save(update_fields=["num_blocked"]) _record_receipt_verdicts(receipt, "disabled_alias") user_profile.last_engagement = datetime.now(UTC) user_profile.save() glean_logger().log_email_blocked(mask=address, reason="block_all") return HttpResponse("Address is temporarily disabled.") _record_receipt_verdicts(receipt, "active_alias") incr_if_enabled("email_for_active_address", 1) # if address is blocking list emails, and email is from list, early return if ( address and address.block_list_emails and user_profile.has_premium and _check_email_from_list(mail["headers"]) ): incr_if_enabled("list_email_for_address_blocking_lists", 1) address.num_blocked += 1 address.save(update_fields=["num_blocked"]) user_profile.last_engagement = datetime.now(UTC) user_profile.save() glean_logger().log_email_blocked(mask=address, reason="block_promotional") return HttpResponse("Address is not accepting list emails.") # Collect new headers subject = common_headers.get("subject", "") destination_address = user_profile.user.email reply_address = get_reply_to_address() try: from_header = generate_from_header(from_address, to_address) except InvalidFromHeader: # TODO: MPP-3407, MPP-3417 - Determine how to handle these header_from = [] for header in mail["headers"]: if header["name"].lower() == "from": header_from.append(header) info_logger.error( "generate_from_header", extra={ "from_address": from_address, "source": mail["source"], "common_headers_from": common_headers["from"], "headers_from": header_from, }, ) log_email_dropped(reason="error_from_header", mask=address, can_retry=True) return HttpResponse("Cannot parse the From address", status=503) # Get incoming email try: (incoming_email_bytes, transport, load_time_s) = _get_email_bytes(message_json) except ClientError as e: if e.response["Error"].get("Code", "") == "NoSuchKey": logger.error("s3_object_does_not_exist", extra=e.response["Error"]) log_email_dropped(reason="content_missing", mask=address) return HttpResponse("Email not in S3", status=404) logger.error("s3_client_error_get_email", extra=e.response["Error"]) log_email_dropped(reason="error_storage", mask=address, can_retry=True) # we are returning a 503 so that SNS can retry the email processing return HttpResponse("Cannot fetch the message content from S3", status=503) # Handle developer overrides, logging dev_action = _get_developer_mode_action(address) if dev_action: if dev_action.new_destination_address: destination_address = dev_action.new_destination_address _log_dev_notification( "_handle_received: developer_mode", dev_action, message_json ) # Convert to new email headers: OutgoingHeaders = { "Subject": subject, "From": from_header, "To": destination_address, "Reply-To": reply_address, "Resent-From": from_address, } sample_trackers = bool(sample_is_active("tracker_sample")) tracker_removal_flag = flag_is_active_in_task("tracker_removal", address.user) remove_level_one_trackers = bool( tracker_removal_flag and user_profile.remove_level_one_email_trackers ) ( forwarded_email, issues, level_one_trackers_removed, has_html, has_text, ) = _convert_to_forwarded_email( incoming_email_bytes=incoming_email_bytes, headers=headers, to_address=to_address, from_address=from_address, language=user_profile.language, has_premium=user_profile.has_premium, sample_trackers=sample_trackers, remove_level_one_trackers=remove_level_one_trackers, ) if has_html: incr_if_enabled("email_with_html_content", 1) if has_text: incr_if_enabled("email_with_text_content", 1) if issues: info_logger.info( "_handle_received: forwarding issues", extra={"issues": issues} ) # Send new email try: ses_response = ses_send_raw_email( source_address=reply_address, destination_address=destination_address, message=forwarded_email, ) except ClientError: # 503 service unavailable response to SNS so it can retry log_email_dropped(reason="error_sending", mask=address, can_retry=True) return HttpResponse("SES client error on Raw Email", status=503) message_id = ses_response["MessageId"] _store_reply_record(mail, message_id, address) user_profile.update_abuse_metric( email_forwarded=True, forwarded_email_size=len(incoming_email_bytes) ) user_profile.last_engagement = datetime.now(UTC) user_profile.save() address.num_forwarded += 1 address.last_used_at = datetime.now(UTC) if level_one_trackers_removed: address.num_level_one_trackers_blocked = ( address.num_level_one_trackers_blocked or 0 ) + level_one_trackers_removed address.save( update_fields=[ "num_forwarded", "last_used_at", "block_list_emails", "num_level_one_trackers_blocked", ] ) glean_logger().log_email_forwarded(mask=address, is_reply=False) return HttpResponse("Sent email to final recipient.", status=200) class DeveloperModeAction(NamedTuple): mask_id: str action: Literal["log", "simulate_complaint"] = "log" new_destination_address: str | None = None def _get_verdict(receipt, verdict_type): return receipt[f"{verdict_type}Verdict"]["status"] def _check_email_from_list(headers): for header in headers: if header["name"].lower().startswith("list-"): return True return False def _record_receipt_verdicts(receipt, state): verdict_tags = [] for key in sorted(receipt.keys()): if key.endswith("Verdict"): value = receipt[key]["status"] verdict_tags.append(f"{key}:{value}") incr_if_enabled(f"relay.emails.verdicts.{key}", 1, [f"state:{state}"]) elif key == "dmarcPolicy": value = receipt[key] verdict_tags.append(f"{key}:{value}") incr_if_enabled(f"relay.emails.state.{state}", 1, verdict_tags) def _get_message_id_from_headers(headers): message_id = None for header in headers: if header["name"].lower() == "message-id": message_id = header["value"] return message_id def _get_keys_from_headers(headers): in_reply_to = None for header in headers: if header["name"].lower() == "in-reply-to": in_reply_to = header["value"] message_id_bytes = get_message_id_bytes(in_reply_to) return derive_reply_keys(message_id_bytes) if header["name"].lower() == "references": message_ids = header["value"] for message_id in message_ids.split(" "): message_id_bytes = get_message_id_bytes(message_id) lookup_key, encryption_key = derive_reply_keys(message_id_bytes) try: # FIXME: calling code is likely to duplicate this query _get_reply_record_from_lookup_key(lookup_key) return lookup_key, encryption_key except Reply.DoesNotExist: pass raise Reply.DoesNotExist incr_if_enabled("mail_to_replies_without_reply_headers", 1) raise ReplyHeadersNotFound def _get_reply_record_from_lookup_key(lookup_key): lookup = b64_lookup_key(lookup_key) return Reply.objects.get(lookup=lookup) def _strip_localpart_tag(address): [localpart, domain] = address.split("@") subaddress_parts = localpart.split("+") return f"{subaddress_parts[0]}@{domain}" _TransportType = Literal["sns", "s3"] def _get_email_bytes( message_json: AWS_SNSMessageJSON, ) -> tuple[bytes, _TransportType, float]: with Timer(logger=None) as load_timer: if "content" in message_json: # email content in sns message message_content = message_json["content"].encode("utf-8") transport: Literal["sns", "s3"] = "sns" else: # assume email content in S3 bucket, object_key = _get_bucket_and_key_from_s3_json(message_json) message_content = get_message_content_from_s3(bucket, object_key) transport = "s3" histogram_if_enabled("relayed_email.size", len(message_content)) load_time_s = round(load_timer.last, 3) return (message_content, transport, load_time_s) def _get_developer_mode_action( mask: RelayAddress | DomainAddress, ) -> DeveloperModeAction | None: """Get the developer mode actions for this mask, if enabled.""" if not ( flag_is_active_in_task("developer_mode", mask.user) and "DEV:" in mask.description ): return None if "DEV:simulate_complaint" in mask.description: action = DeveloperModeAction( mask_id=mask.metrics_id, action="simulate_complaint", new_destination_address=f"complaint+{mask.metrics_id}@simulator.amazonses.com", ) else: action = DeveloperModeAction(mask_id=mask.metrics_id, action="log") return action def _log_dev_notification( log_message: str, dev_action: DeveloperModeAction, notification: dict[str, Any] ) -> None: """ Log notification JSON This will log information beyond our privacy policy, so it should only be used on Relay staff accounts with prior permission. The notification JSON will be compressed, Ascii85-encoded with padding, and broken into 1024-bytes chunks. This will ensure it fits into GCP's log entry, which has a 64KB limit per label value. """ notification_gza85 = encode_dict_gza85(notification) total_parts = notification_gza85.count("\n") + 1 log_group_id = str(uuid4()) for partnum, part in enumerate(notification_gza85.splitlines()): info_logger.info( log_message, extra={ "mask_id": dev_action.mask_id, "dev_action": dev_action.action, "log_group_id": log_group_id, "part": partnum, "parts": total_parts, "notification_gza85": part, }, ) def _convert_to_forwarded_email( incoming_email_bytes: bytes, headers: OutgoingHeaders, to_address: str, from_address: str, language: str, has_premium: bool, sample_trackers: bool, remove_level_one_trackers: bool, now: datetime | None = None, ) -> tuple[EmailMessage, EmailForwardingIssues, int, bool, bool]: """ Convert an email (as bytes) to a forwarded email. Return is a tuple: - email - The forwarded email - issues - Any detected issues in conversion - level_one_trackers_removed (int) - Number of trackers removed - has_html - True if the email has an HTML representation - has_text - True if the email has a plain text representation """ email = message_from_bytes(incoming_email_bytes, policy=relay_policy) # python/typeshed issue 2418 # The Python 3.2 default was Message, 3.6 uses policy.message_factory, and # policy.default.message_factory is EmailMessage if not isinstance(email, EmailMessage): raise TypeError("email must be type EmailMessage") # Replace headers in the original email header_issues = _replace_headers(email, headers) # Find and replace text content text_body = email.get_body("plain") text_content = None has_text = False if text_body: has_text = True if not isinstance(text_body, EmailMessage): raise TypeError("text_body must be type EmailMessage") text_content = text_body.get_content() new_text_content = _convert_text_content(text_content, to_address) text_body.set_content(new_text_content) # Find and replace HTML content html_body = email.get_body("html") level_one_trackers_removed = 0 has_html = False if html_body: has_html = True if not isinstance(html_body, EmailMessage): raise TypeError("html_body must be type EmailMessage") html_content = html_body.get_content() new_content, level_one_trackers_removed = _convert_html_content( html_content, to_address, from_address, language, has_premium, sample_trackers, remove_level_one_trackers, ) html_body.set_content(new_content, subtype="html") elif text_content: # Try to use the text content to generate HTML content html_content = urlize_and_linebreaks(text_content) new_content, level_one_trackers_removed = _convert_html_content( html_content, to_address, from_address, language, has_premium, sample_trackers, remove_level_one_trackers, ) if not isinstance(text_body, EmailMessage): raise TypeError("text_body must be type EmailMessage") try: text_body.add_alternative(new_content, subtype="html") except TypeError as e: out = StringIO() _structure(email, fp=out) info_logger.error( "Adding HTML alternate failed", extra={"exception": str(e), "structure": out.getvalue()}, ) issues: EmailForwardingIssues = {} if header_issues: issues["headers"] = header_issues return (email, issues, level_one_trackers_removed, has_html, has_text) def _replace_headers( email: EmailMessage, headers: OutgoingHeaders ) -> EmailHeaderIssues: """ Replace the headers in email with new headers. This replaces headers in the passed email object, rather than returns an altered copy. The primary reason is that the Python email package can read an email with non-compliant headers or content, but can't write it. A read/write is required to create a copy that we then alter. This code instead alters the passed EmailMessage object, making header-specific changes in try / except statements. The other reason is the object size. An Email can be up to 10 MB, and we hope to support 40 MB emails someday. Modern servers may be OK with this, but it would be nice to handle the non-compliant headers without crashing before we add a source of memory-related crashes. """ # Look for headers to drop to_drop: list[str] = [] replacements: set[str] = {_k.lower() for _k in headers.keys()} issues: EmailHeaderIssues = [] # Detect non-compliant headers in incoming emails for header in email.keys(): try: value = email[header] except Exception as e: issues.append( {"header": header, "direction": "in", "exception_on_read": repr(e)} ) value = None if getattr(value, "defects", None): issues.append( { "header": header, "direction": "in", "defect_count": len(value.defects), "parsed_value": str(value), "raw_value": str(value.as_raw), } ) elif getattr(getattr(value, "_parse_tree", None), "all_defects", []): issues.append( { "header": header, "direction": "in", "defect_count": len(value._parse_tree.all_defects), "parsed_value": str(value), "raw_value": str(value.as_raw), } ) # Collect headers that will not be forwarded for header in email.keys(): header_lower = header.lower() if ( header_lower not in replacements and header_lower != "mime-version" and not header_lower.startswith("content-") ): to_drop.append(header) # Drop headers that should be dropped for header in to_drop: del email[header] # Replace the requested headers for header, value in headers.items(): del email[header] try: email[header] = value.rstrip("\r\n") except Exception as e: issues.append( { "header": header, "direction": "out", "exception_on_write": repr(e), "value": value, } ) continue try: parsed_value = email[header] except Exception as e: issues.append( { "header": header, "direction": "out", "exception_on_write": repr(e), "value": value, } ) continue if parsed_value.defects: issues.append( { "header": header, "direction": "out", "defect_count": len(parsed_value.defects), "parsed_value": str(parsed_value), "raw_value": str(parsed_value.as_raw), }, ) return issues def _convert_html_content( html_content: str, to_address: str, from_address: str, language: str, has_premium: bool, sample_trackers: bool, remove_level_one_trackers: bool, now: datetime | None = None, ) -> tuple[str, int]: # frontend expects a timestamp in milliseconds now = now or datetime.now(UTC) datetime_now_ms = int(now.timestamp() * 1000) # scramble alias so that clients don't recognize it # and apply default link styles display_email = re.sub("([@.:])", r"\1", to_address) # sample tracker numbers if sample_trackers: count_all_trackers(html_content) tracker_report_link = "" removed_count = 0 if remove_level_one_trackers: html_content, tracker_details = remove_trackers( html_content, from_address, datetime_now_ms ) removed_count = tracker_details["tracker_removed"] tracker_report_details = { "sender": from_address, "received_at": datetime_now_ms, "trackers": tracker_details["level_one"]["trackers"], } tracker_report_link = f"{settings.SITE_ORIGIN}/tracker-report/#" + json.dumps( tracker_report_details ) wrapped_html = wrap_html_email( original_html=html_content, language=language, has_premium=has_premium, display_email=display_email, tracker_report_link=tracker_report_link, num_level_one_email_trackers_removed=removed_count, ) return wrapped_html, removed_count def _convert_text_content(text_content: str, to_address: str) -> str: relay_header_text = ( "This email was sent to your alias " f"{to_address}. To stop receiving emails sent to this alias, " "update the forwarding settings in your dashboard.\n" "---Begin Email---\n" ) wrapped_text = relay_header_text + text_content return wrapped_text def _build_reply_requires_premium_email( from_address: str, reply_record: Reply, message_id: str | None, decrypted_metadata: dict[str, Any] | None, ) -> EmailMessage: # If we haven't forwarded a first reply for this user yet, _reply_allowed # will forward. So, tell the user we forwarded it. forwarded = not reply_record.address.user.profile.forwarded_first_reply sender: str | None = "" if decrypted_metadata is not None: sender = decrypted_metadata.get("reply-to") or decrypted_metadata.get("from") ctx = { "sender": sender or "", "forwarded": forwarded, "SITE_ORIGIN": settings.SITE_ORIGIN, } html_body = render_to_string("emails/reply_requires_premium.html", ctx) text_body = render_to_string("emails/reply_requires_premium.txt", ctx) # Create the message msg = EmailMessage() msg["Subject"] = ftl_bundle.format("replies-not-included-in-free-account-header") msg["From"] = get_reply_to_address() msg["To"] = from_address if message_id: msg["In-Reply-To"] = message_id msg["References"] = message_id msg.set_content(text_body) msg.add_alternative(html_body, subtype="html") return msg def _set_forwarded_first_reply(profile): profile.forwarded_first_reply = True profile.save() def _send_reply_requires_premium_email( from_address: str, reply_record: Reply, message_id: str | None, decrypted_metadata: dict[str, Any] | None, ) -> None: msg = _build_reply_requires_premium_email( from_address, reply_record, message_id, decrypted_metadata ) try: ses_send_raw_email( source_address=get_reply_to_address(premium=False), destination_address=from_address, message=msg, ) # If we haven't forwarded a first reply for this user yet, _reply_allowed will. # So, updated the DB. _set_forwarded_first_reply(reply_record.address.user.profile) except ClientError as e: logger.error("reply_not_allowed_ses_client_error", extra=e.response["Error"]) incr_if_enabled("free_user_reply_attempt", 1) def _reply_allowed( from_address, to_address, reply_record, message_id=None, decrypted_metadata=None ): stripped_from_address = _strip_localpart_tag(from_address) reply_record_email = reply_record.address.user.email stripped_reply_record_address = _strip_localpart_tag(reply_record_email) if (from_address == reply_record_email) or ( stripped_from_address == stripped_reply_record_address ): # This is a Relay user replying to an external sender; if not reply_record.profile.user.is_active: return False if reply_record.profile.is_flagged: return False if reply_record.owner_has_premium: return True # if we haven't forwarded a first reply for this user, return True to allow # this first reply allow_first_reply = not reply_record.address.user.profile.forwarded_first_reply _send_reply_requires_premium_email( from_address, reply_record, message_id, decrypted_metadata ) return allow_first_reply else: # The From: is not a Relay user, so make sure this is a reply *TO* a # premium Relay user try: address = _get_address(to_address) if address.user.profile.has_premium: return True except ObjectDoesNotExist: return False incr_if_enabled("free_user_reply_attempt", 1) return False def _handle_reply( from_address: str, message_json: AWS_SNSMessageJSON, to_address: str ) -> HttpResponse: """ Handle a reply from a Relay user to an external email. Returns (may be incomplete): * 200 if the reply was sent * 400 if the In-Reply-To and References headers are missing, none of the References headers are a reply record, or the SES client raises an error * 403 if the Relay user is not allowed to reply * 404 if the S3-stored email is not found, or there is no matching Reply record in the database * 503 if the S3 client returns an error (other than not found), or the SES client returns an error TODO: Return a more appropriate status object (see _handle_received) TODO: Document metrics emitted """ mail = message_json["mail"] try: (lookup_key, encryption_key) = _get_keys_from_headers(mail["headers"]) except ReplyHeadersNotFound: incr_if_enabled("reply_email_header_error", 1, tags=["detail:no-header"]) return HttpResponse("No In-Reply-To header", status=400) try: reply_record = _get_reply_record_from_lookup_key(lookup_key) except Reply.DoesNotExist: incr_if_enabled("reply_email_header_error", 1, tags=["detail:no-reply-record"]) return HttpResponse("Unknown or stale In-Reply-To header", status=404) address = reply_record.address message_id = _get_message_id_from_headers(mail["headers"]) decrypted_metadata = json.loads( decrypt_reply_metadata(encryption_key, reply_record.encrypted_metadata) ) if not _reply_allowed( from_address, to_address, reply_record, message_id, decrypted_metadata ): log_email_dropped(reason="reply_requires_premium", mask=address, is_reply=True) return HttpResponse("Relay replies require a premium account", status=403) outbound_from_address = address.full_address incr_if_enabled("reply_email", 1) subject = mail["commonHeaders"].get("subject", "") to_address = decrypted_metadata.get("reply-to") or decrypted_metadata.get("from") headers: OutgoingHeaders = { "Subject": subject, "From": outbound_from_address, "To": to_address, "Reply-To": outbound_from_address, } try: (email_bytes, transport, load_time_s) = _get_email_bytes(message_json) except ClientError as e: if e.response["Error"].get("Code", "") == "NoSuchKey": logger.error("s3_object_does_not_exist", extra=e.response["Error"]) log_email_dropped(reason="content_missing", mask=address, is_reply=True) return HttpResponse("Email not in S3", status=404) logger.error("s3_client_error_get_email", extra=e.response["Error"]) log_email_dropped( reason="error_storage", mask=address, is_reply=True, can_retry=True ) # we are returning a 500 so that SNS can retry the email processing return HttpResponse("Cannot fetch the message content from S3", status=503) email = message_from_bytes(email_bytes, policy=relay_policy) if not isinstance(email, EmailMessage): raise TypeError("email must be type EmailMessage") # Convert to a reply email # TODO: Issue #1747 - Remove wrapper / prefix in replies _replace_headers(email, headers) try: ses_send_raw_email( source_address=outbound_from_address, destination_address=to_address, message=email, ) except ClientError: log_email_dropped(reason="error_sending", mask=address, is_reply=True) return HttpResponse("SES client error", status=400) reply_record.increment_num_replied() profile = address.user.profile profile.update_abuse_metric(replied=True) profile.last_engagement = datetime.now(UTC) profile.save() glean_logger().log_email_forwarded(mask=address, is_reply=True) return HttpResponse("Sent email to final recipient.", status=200) def _get_domain_address( local_portion: str, domain_portion: str, create: bool = True ) -> DomainAddress: """ Find or create the DomainAddress for the parts of an email address. If the domain_portion is for a valid subdomain, and create=True, a new DomainAddress will be created and returned. If create=False, DomainAddress.DoesNotExist is raised. If the domain_portion is for an unknown domain, ObjectDoesNotExist is raised. If the domain_portion is for an unclaimed subdomain, Profile.DoesNotExist is raised. """ [address_subdomain, address_domain] = domain_portion.split(".", 1) if address_domain != get_domains_from_settings()["MOZMAIL_DOMAIN"]: if create: incr_if_enabled("email_for_not_supported_domain", 1) raise ObjectDoesNotExist("Address does not exist") try: with transaction.atomic(): locked_profile = Profile.objects.select_for_update().get( subdomain=address_subdomain ) domain_numerical = get_domain_numerical(address_domain) # filter DomainAddress because it may not exist # which will throw an error with get() domain_address = DomainAddress.objects.filter( user=locked_profile.user, address=local_portion, domain=domain_numerical ).first() if domain_address is None: if not create: raise DomainAddress.DoesNotExist() # TODO: Consider flows when a user generating alias on a fly # was unable to receive an email due to user no longer being a # premium user as seen in exception thrown on make_domain_address domain_address = DomainAddress.make_domain_address( locked_profile.user, local_portion, True ) glean_logger().log_email_mask_created( mask=domain_address, created_by_api=False, ) domain_address.last_used_at = datetime.now(UTC) domain_address.save() return domain_address except Profile.DoesNotExist as e: if create: incr_if_enabled("email_for_dne_subdomain", 1) raise e def _get_address(address: str, create: bool = True) -> RelayAddress | DomainAddress: """ Find or create the RelayAddress or DomainAddress for an email address. If an unknown email address is for a valid subdomain, and create is True, a new DomainAddress will be created. On failure, raises exception based on Django's ObjectDoesNotExist: * RelayAddress.DoesNotExist - looks like RelayAddress, deleted or does not exist * Profile.DoesNotExist - looks like DomainAddress, no subdomain match * DomainAddress.DoesNotExist - looks like unknown DomainAddress, create is False * ObjectDoesNotExist - Unknown domain """ local_portion, domain_portion = address.split("@") local_address = local_portion.lower() domain = domain_portion.lower() # if the domain is not the site's 'top' relay domain, # it may be for a user's subdomain email_domains = get_domains_from_settings().values() if domain not in email_domains: return _get_domain_address(local_address, domain, create) # the domain is the site's 'top' relay domain, so look up the RelayAddress try: domain_numerical = get_domain_numerical(domain) relay_address = RelayAddress.objects.get( address=local_address, domain=domain_numerical ) return relay_address except RelayAddress.DoesNotExist as e: if not create: raise e try: DeletedAddress.objects.get( address_hash=address_hash(local_address, domain=domain) ) incr_if_enabled("email_for_deleted_address", 1) # TODO: create a hard bounce receipt rule in SES except DeletedAddress.DoesNotExist: incr_if_enabled("email_for_unknown_address", 1) except DeletedAddress.MultipleObjectsReturned: # not sure why this happens on stage but let's handle it incr_if_enabled("email_for_deleted_address_multiple", 1) raise e def _get_address_if_exists(address: str) -> RelayAddress | DomainAddress | None: """Get the matching RelayAddress or DomainAddress, or None if it doesn't exist.""" try: return _get_address(address, create=False) except (RelayAddress.DoesNotExist, Profile.DoesNotExist, ObjectDoesNotExist): return None def _handle_bounce(message_json: AWS_SNSMessageJSON) -> HttpResponse: """ Handle an AWS SES bounce notification. For more information, see: https://docs.aws.amazon.com/ses/latest/dg/notification-contents.html#bounce-object Returns: * 404 response if any email address does not match a user, * 200 response if all match or none are given Emits a counter metric "email_bounce" with these tags: * bounce_type: 'permanent', 'transient', 'undetermined', 'none' if omitted * bounce_subtype: 'undetermined', 'general', etc., 'none' if omitted * user_match: 'found', 'missing', error states 'no_address' and 'no_recipients' * relay_action: 'no_action', 'auto_block_spam', 'hard_bounce', 'soft_bounce' Emits an info log "bounce_notification", same data as metric, plus: * bounce_action: 'action' from bounced recipient data, or None * bounce_status: 'status' from bounced recipient data, or None * bounce_diagnostic: 'diagnosticCode' from bounced recipient data, or None * bounce_extra: Extra data from bounce_recipient data, if any * domain: User's real email address domain, if an address was given * fxa_id - The Mozilla account (previously known as Firefox Account) ID of the user """ bounce = message_json.get("bounce", {}) bounce_type = bounce.get("bounceType", "none") bounce_subtype = bounce.get("bounceSubType", "none") bounced_recipients = bounce.get("bouncedRecipients", []) now = datetime.now(UTC) bounce_data = [] for recipient in bounced_recipients: recipient_address = recipient.pop("emailAddress", None) data = { "bounce_type": bounce_type, "bounce_subtype": bounce_subtype, "bounce_action": recipient.pop("action", ""), "bounce_status": recipient.pop("status", ""), "bounce_diagnostic": recipient.pop("diagnosticCode", ""), "user_match": "no_address", "relay_action": "no_action", } if recipient: data["bounce_extra"] = recipient.copy() bounce_data.append(data) if recipient_address is None: continue recipient_address = parseaddr(recipient_address)[1] recipient_domain = recipient_address.split("@")[1] data["domain"] = recipient_domain try: user = User.objects.get(email=recipient_address) profile = user.profile data["user_match"] = "found" if (fxa := profile.fxa) and profile.metrics_enabled: data["fxa_id"] = fxa.uid else: data["fxa_id"] = "" except User.DoesNotExist: # TODO: handle bounce for a user who no longer exists # add to SES account-wide suppression list? data["user_match"] = "missing" continue action = None if "spam" in data["bounce_diagnostic"].lower(): # if an email bounced as spam, set to auto block spam for this user # and DON'T set them into bounce pause state action = "auto_block_spam" profile.auto_block_spam = True elif bounce_type == "Permanent": # TODO: handle sub-types: 'General', 'NoEmail', etc. action = "hard_bounce" profile.last_hard_bounce = now elif bounce_type == "Transient": # TODO: handle sub-types: 'MessageTooLarge', 'AttachmentRejected', etc. action = "soft_bounce" profile.last_soft_bounce = now if action: data["relay_action"] = action profile.save() if not bounce_data: # Data when there are no identified recipients bounce_data = [{"user_match": "no_recipients", "relay_action": "no_action"}] for data in bounce_data: tags = { "bounce_type": bounce_type, "bounce_subtype": bounce_subtype, "user_match": data["user_match"], "relay_action": data["relay_action"], } incr_if_enabled( "email_bounce", 1, tags=[generate_tag(key, val) for key, val in tags.items()], ) info_logger.info("bounce_notification", extra=data) if any(data["user_match"] == "missing" for data in bounce_data): return HttpResponse("Address does not exist", status=404) return HttpResponse("OK", status=200) def _build_disabled_mask_for_spam_email( mask: RelayAddress | DomainAddress, ) -> EmailMessage: ctx = {"mask": mask.full_address, "SITE_ORIGIN": settings.SITE_ORIGIN} html_body = render_to_string("emails/disabled_mask_for_spam.html", ctx) text_body = render_to_string("emails/disabled_mask_for_spam.txt", ctx) # Create the message msg = EmailMessage() msg["Subject"] = ftl_bundle.format("relay-deactivated-mask-email-subject") msg["From"] = settings.RELAY_FROM_ADDRESS msg["To"] = mask.user.email msg.set_content(text_body) msg.add_alternative(html_body, subtype="html") return msg def _send_disabled_mask_for_spam_email(mask: RelayAddress | DomainAddress) -> None: msg = _build_disabled_mask_for_spam_email(mask) if not settings.RELAY_FROM_ADDRESS: raise ValueError( "Must set settings.RELAY_FROM_ADDRESS to send disabled_mask_for_spam email." ) try: ses_send_raw_email( source_address=settings.RELAY_FROM_ADDRESS, destination_address=mask.user.email, message=msg, ) except ClientError as e: logger.error("send_disabled_mask_ses_client_error", extra=e.response["Error"]) incr_if_enabled("send_disabled_mask_email", 1) def _handle_complaint(message_json: AWS_SNSMessageJSON) -> HttpResponse: """ Handle an AWS SES complaint notification. This looks for Relay users in the complainedRecipients (real email address) and the From: header (mask address). We expect both to match the same Relay user, and return a 200. If one or the other do not match, a 404 is returned, and errors may be logged. The first time a user complains, this sets the user's auto_block_spam flag to True. The second time a user complains, this disables the mask thru which the spam mail was forwarded, and sends an email to the user to notify them the mask is disabled and can be re-enabled on their dashboard. For more information on the complaint notification, see: https://docs.aws.amazon.com/ses/latest/dg/notification-contents.html#complaint-object Returns: * 404 response if any email address does not match a user, * 200 response if all match or none are given Emits a counter metric "email_complaint" with these tags: * complaint_subtype: 'onaccountsuppressionlist', or 'none' if omitted * complaint_feedback - feedback enumeration from ISP (usually 'abuse') or 'none' * user_match: 'found' or 'no_recipients' * relay_action: 'no_action', 'auto_block_spam', or 'disable_mask' Emits an info log "complaint_notification", same data as metric, plus: * complaint_user_agent - identifies the client used to file the complaint * complaint_extra - Extra data from complainedRecipients data, if any * domain - User's domain, if an address was given * found_in - "complained_recipients" (real email), "from_header" (email mask), or "all" (matching records found in both) * fxa_id - The Mozilla account (previously known as Firefox Account) ID of the user * mask_match - "found" if "From" header contains an email mask, or "not_found" """ complaint_data = _get_complaint_data(message_json) complainers, unknown_count = _gather_complainers(complaint_data) # Reduce future complaints from complaining Relay users actions: list[ComplaintAction] = [] for complainer in complainers: action = _reduce_future_complaints(complainer) actions.append(action) if ( flag_is_active_in_task("developer_mode", complainer["user"]) and action.mask_id ): _log_dev_notification( "_handle_complaint: developer_mode", DeveloperModeAction(mask_id=action.mask_id, action="log"), message_json, ) # Log complaint and actions taken if not actions: # Log the complaint but that no action was taken actions.append(ComplaintAction(user_match="no_recipients")) for action in actions: tags = [ generate_tag(key, val) for key, val in { "complaint_subtype": complaint_data.subtype or "none", "complaint_feedback": complaint_data.feedback_type or "none", "user_match": action.user_match, "relay_action": action.relay_action, }.items() ] incr_if_enabled("email_complaint", tags=tags) log_extra = { "complaint_subtype": complaint_data.subtype or None, "complaint_user_agent": complaint_data.user_agent or None, "complaint_feedback": complaint_data.feedback_type or None, } log_extra.update( { key: value for key, value in action._asdict().items() if (value is not None and key != "mask_id") } ) info_logger.info("complaint_notification", extra=log_extra) if unknown_count: return HttpResponse("Address does not exist", status=404) return HttpResponse("OK", status=200) class RawComplaintData(NamedTuple): complained_recipients: list[tuple[str, dict[str, Any]]] from_addresses: list[str] subtype: str user_agent: str feedback_type: str def _get_complaint_data(message_json: AWS_SNSMessageJSON) -> RawComplaintData: """ Extract complaint data from an AWS SES Complaint Notification. This extracts only the data used by _handle_complaint(). It also works on complaint events, which have a similar structure and the same data needed by _handle_complaint. For more information on the complaint notification, see: https://docs.aws.amazon.com/ses/latest/dg/notification-contents.html#complaint-object """ complaint = message_json["complaint"] T = TypeVar("T") def get_or_log( key: str, source: dict[str, T], data_type: type[T] ) -> tuple[T, bool]: """Get a value from a dictionary, or log if not found""" if key in source: return source[key], True logger.error( "_get_complaint_data: Unexpected message format", extra={"missing_key": key, "found_keys": ",".join(sorted(source.keys()))}, ) return data_type(), False raw_recipients, has_cr = get_or_log("complainedRecipients", complaint, list) complained_recipients = [] no_entries = True for entry in raw_recipients: no_entries = False raw_email_address, has_email = get_or_log("emailAddress", entry, str) if has_email: email_address = parseaddr(raw_email_address)[1] extra = { key: value for key, value in entry.items() if key != "emailAddress" } complained_recipients.append((email_address, extra)) if has_cr and no_entries: logger.error("_get_complaint_data: Empty complainedRecipients") mail, has_mail = get_or_log("mail", message_json, dict) if has_mail: commonHeaders, has_ch = get_or_log("commonHeaders", mail, dict) else: commonHeaders, has_ch = {}, False if has_ch: raw_from_addresses, _ = get_or_log("from", commonHeaders, list) else: raw_from_addresses = [] from_addresses = [parseaddr(addr)[1] for addr in raw_from_addresses] # Only present when set feedback_type = complaint.get("complaintFeedbackType", "") # Only present when destination is on account suppression list subtype = complaint.get("complaintSubType", "") # Only present for feedback reports user_agent = complaint.get("userAgent", "") return RawComplaintData( complained_recipients, from_addresses, subtype, user_agent, feedback_type ) class Complainer(TypedDict): user: User found_in: Literal["complained_recipients", "from_header", "all"] domain: str extra: dict[str, Any] | None masks: list[RelayAddress | DomainAddress] def _gather_complainers( complaint_data: RawComplaintData, ) -> tuple[list[Complainer], int]: """ Fetch Relay Users and masks from the complaint data. This matches data from an AWS SES Complaint Notification (as extracted by _get_complaint_data()) to the Relay database, and returns the Users, RelayAddresses, and DomainAddresses, as well as status and extra data. If the complaint came from the AWS SES complaint simulator, detect developer_mode and move forward with the developer's User data. """ users: dict[int, Complainer] = {} unknown_complainer_count = 0 for email_address, extra_data in complaint_data.complained_recipients: local, domain = email_address.split("@", 1) # If the complainer is the AWS SES complaint simulation, assume that # it was send by a user with the developer_mode flag. Look for # a mask that matches the embedded mask metrics_id, and use # the related user's email instead of the AWS simulator address. # See docs/developer_mode.md if domain == "simulator.amazonses.com" and local.startswith("complaint+"): mask_metrics_id = local.removeprefix("complaint+") mask = _get_mask_by_metrics_id(mask_metrics_id) if mask: email_address = mask.user.email domain = mask.user.email.split("@")[1] try: user = User.objects.get(email=email_address) except User.DoesNotExist: logger.error("_gather_complainers: unknown complainedRecipient") unknown_complainer_count += 1 continue if user.id in users: logger.error("_gather_complainers: complainer appears twice") continue users[user.id] = { "user": user, "found_in": "complained_recipients", "domain": domain, "extra": extra_data or None, "masks": [], } # Collect From: addresses and their users unknown_sender_count = 0 for email_address in complaint_data.from_addresses: mask = _get_address_if_exists(email_address) if not mask: logger.error("_gather_complainers: unknown mask, maybe deleted?") unknown_sender_count += 1 continue if mask.user.id not in users: # Add mask-only entry to users users[mask.user.id] = { "user": mask.user, "found_in": "from_header", "domain": mask.user.email.split("@")[1], "extra": None, "masks": [mask], } continue user_data = users[mask.user.id] if mask in user_data["masks"]: logger.error("_gather_complainers: mask appears twice") continue user_data["masks"].append(mask) if user_data["found_in"] in ("all", "complained_recipients"): user_data["found_in"] = "all" else: logger.error("_gather_complainers: no complainer, multi-mask") return (list(users.values()), unknown_complainer_count + unknown_sender_count) def _get_mask_by_metrics_id(metrics_id: str) -> RelayAddress | DomainAddress | None: """Look up a mask by metrics ID, or None if not found.""" if not metrics_id or metrics_id[0] not in ("R", "D"): return None mask_type_id = metrics_id[0] mask_raw_id = metrics_id[1:] try: mask_id = int(mask_raw_id) except ValueError: return None # ID is not an int, do not try to match to Relay mask if mask_type_id == "R": try: return RelayAddress.objects.get(id=mask_id) except RelayAddress.DoesNotExist: return None try: return DomainAddress.objects.get(id=mask_id) except DomainAddress.DoesNotExist: return None class ComplaintAction(NamedTuple): user_match: Literal["found", "no_recipients"] relay_action: Literal["no_action", "auto_block_spam", "disable_mask"] = "no_action" mask_match: Literal["found", "not_found"] = "not_found" mask_id: str | None = None found_in: Literal["complained_recipients", "from_header", "all"] | None = None fxa_id: str | None = None domain: str | None = None complaint_extra: str | None = None def _reduce_future_complaints(complainer: Complainer) -> ComplaintAction: """Take action to reduce future complaints from complaining user.""" user = complainer["user"] mask_match: Literal["found", "not_found"] = "not_found" relay_action: Literal["no_action", "auto_block_spam", "disable_mask"] = "no_action" mask_id = None if not user.profile.auto_block_spam: relay_action = "auto_block_spam" user.profile.auto_block_spam = True user.profile.save() for mask in complainer["masks"]: mask_match = "found" mask_id = mask.metrics_id if ( flag_is_active_in_task("disable_mask_on_complaint", user) and mask.enabled and relay_action != "auto_block_spam" ): relay_action = "disable_mask" mask.enabled = False mask.save() _send_disabled_mask_for_spam_email(mask) return ComplaintAction( user_match="found", relay_action=relay_action, mask_match=mask_match, mask_id=mask_id, fxa_id=user.profile.metrics_fxa_id, domain=complainer["domain"], found_in=complainer["found_in"], complaint_extra=( json.dumps(complainer["extra"]) if complainer["extra"] else None ), ) _WAFFLE_FLAGS_INITIALIZED = False def init_waffle_flags() -> None: """Initialize waffle flags for email tasks""" global _WAFFLE_FLAGS_INITIALIZED if _WAFFLE_FLAGS_INITIALIZED: return flags: list[tuple[str, str]] = [ ( "disable_mask_on_complaint", "MPP-3119: When a Relay user marks an email as spam, disable the mask.", ), ( "developer_mode", "MPP-3932: Enable logging and overrides for Relay developers.", ), ] waffle_flag_table = get_waffle_flag_model().objects for name, note in flags: waffle_flag_table.get_or_create(name=name, defaults={"note": note}) _WAFFLE_FLAGS_INITIALIZED = True