def _handle_complaint()

in emails/views.py [0:0]

• 45 lines of code
• 17 McCabe index (conditional complexity)

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)