defmodule PhoenixKitWeb.Controllers.EmailWebhookController do @moduledoc """ Secure webhook controller for AWS SNS email tracking events. Handles incoming webhook notifications from AWS Simple Notification Service (SNS) for email events like bounces, complaints, deliveries, opens, and clicks. ## Security Features - **SNS Signature Verification**: Validates authentic AWS requests - **IP Whitelist**: Restricts access to AWS IP ranges - **Rate Limiting**: Prevents abuse with configurable limits - **Replay Attack Protection**: Timestamp verification (max 5 minutes) - **Request Size Limits**: Prevents oversized payloads - **Automatic Subscription Confirmation**: Handles SNS subscription setup ## Supported Event Types - **Bounce**: Hard and soft bounces with detailed reasons - **Complaint**: Spam complaints and feedback loops - **Delivery**: Successful delivery confirmations - **Send**: Send confirmations from SES - **Open**: Email open tracking (pixel-based) - **Click**: Link click tracking ## Configuration Set these environment variables for security: # Enable/disable specific security features PHOENIX_KIT_WEBHOOK_VERIFY_SNS_SIGNATURE=true PHOENIX_KIT_WEBHOOK_CHECK_AWS_IP=true PHOENIX_KIT_WEBHOOK_RATE_LIMIT_ENABLED=true PHOENIX_KIT_WEBHOOK_MAX_AGE_SECONDS=300 ## Usage Add to your router: # Public webhook endpoint (no authentication) post "/phoenix_kit/webhooks/email", PhoenixKitWeb.Controllers.EmailWebhookController, :handle ## AWS SNS Setup 1. Create SNS topic for SES events 2. Subscribe this endpoint to the topic 3. Configure SES to publish events to the topic 4. The controller will automatically confirm subscriptions ## Example Webhook Payload %{ "Type" => "Notification", "Message" => Jason.encode!(%{ "eventType" => "bounce", "mail" => %{"messageId" => "abc123"}, "bounce" => %{ "bounceType" => "Permanent", "bouncedRecipients" => [%{"emailAddress" => "user@example.com"}] } }) } """ use PhoenixKitWeb, :controller import Bitwise require Logger alias PhoenixKit.EmailTracking # Rate limiting configuration (commented out for future use) # @default_rate_limit %{max_requests: 100, window_seconds: 60} # 5 minutes @max_request_age_seconds 300 # 50KB @max_payload_size 50_000 # AWS IP ranges for SNS (these should be updated periodically) @aws_sns_ip_ranges [ # US East (N. Virginia) "54.240.217.0/24", "54.240.218.0/23", "54.240.220.0/22", # EU (Ireland) "176.34.159.192/26", "176.34.185.0/24", # Asia Pacific (Sydney) "54.240.197.0/24", "54.240.198.0/24" ] ## --- Main Handler --- @doc """ Main webhook handler for AWS SNS notifications. Processes all incoming webhook requests with full security validation. """ def handle(conn, params) do start_time = System.monotonic_time(:microsecond) with :ok <- check_request_size(conn), :ok <- check_rate_limit(conn), :ok <- verify_aws_ip(conn), :ok <- verify_request_age(params), {:ok, sns_message} <- parse_sns_message(params), :ok <- verify_sns_signature(sns_message), {:ok, result} <- process_sns_message(sns_message) do # Log successful processing processing_time = System.monotonic_time(:microsecond) - start_time Logger.info("Webhook processed successfully", %{ message_type: sns_message["Type"], processing_time_ms: div(processing_time, 1000), result: result }) conn |> put_resp_content_type("text/plain") |> send_resp(200, "OK") else {:error, :request_too_large} -> Logger.warning("Webhook rejected: request too large") conn |> put_resp_content_type("text/plain") |> send_resp(413, "Request Entity Too Large") {:error, :invalid_aws_ip} -> Logger.warning("Webhook rejected: invalid AWS IP", %{ remote_ip: get_remote_ip(conn) }) conn |> put_resp_content_type("text/plain") |> send_resp(403, "Forbidden") {:error, :request_too_old} -> Logger.warning("Webhook rejected: request too old") conn |> put_resp_content_type("text/plain") |> send_resp(400, "Request Too Old") {:error, :invalid_signature} -> Logger.warning("Webhook rejected: invalid SNS signature") conn |> put_resp_content_type("text/plain") |> send_resp(403, "Invalid Signature") {:error, reason} -> Logger.error("Webhook processing failed", %{ reason: inspect(reason), remote_ip: get_remote_ip(conn) }) conn |> put_resp_content_type("text/plain") |> send_resp(400, "Bad Request") end end ## --- Security Functions --- # Check request payload size defp check_request_size(conn) do content_length = case Plug.Conn.get_req_header(conn, "content-length") do [length_str] -> case Integer.parse(length_str) do {length, _} -> length _ -> 0 end _ -> 0 end if content_length <= @max_payload_size do :ok else {:error, :request_too_large} end end # Check rate limiting per IP defp check_rate_limit(conn) do if rate_limiting_enabled?() do remote_ip = get_remote_ip(conn) check_ip_rate_limit(remote_ip) else :ok end end # Verify request comes from AWS IP ranges defp verify_aws_ip(conn) do if aws_ip_check_enabled?() do remote_ip = get_remote_ip(conn) if ip_in_aws_ranges?(remote_ip) do :ok else {:error, :invalid_aws_ip} end else :ok end end # Verify request timestamp is recent defp verify_request_age(params) do timestamp = params["Timestamp"] || params["timestamp"] case timestamp do nil -> {:error, :missing_timestamp} timestamp_str -> case parse_timestamp(timestamp_str) do {:ok, request_time} -> age_seconds = DateTime.diff(DateTime.utc_now(), request_time, :second) if age_seconds <= @max_request_age_seconds do :ok else {:error, :request_too_old} end {:error, _} -> {:error, :invalid_timestamp} end end end # Parse and validate SNS message structure defp parse_sns_message(params) do required_fields = ["Type"] if Enum.all?(required_fields, &Map.has_key?(params, &1)) do {:ok, params} else {:error, :invalid_sns_message} end end # Verify SNS message signature defp verify_sns_signature(sns_message) do if signature_verification_enabled?() do case verify_aws_sns_signature(sns_message) do :ok -> :ok :error -> {:error, :invalid_signature} end else :ok end end ## --- Message Processing --- # Process different types of SNS messages defp process_sns_message(%{"Type" => "SubscriptionConfirmation"} = message) do # Automatically confirm SNS subscription confirm_subscription(message) end defp process_sns_message(%{"Type" => "UnsubscribeConfirmation"} = message) do # Log unsubscription Logger.info("SNS topic unsubscribed", %{ topic_arn: message["TopicArn"], subscription_arn: message["SubscriptionArn"] }) {:ok, :unsubscribed} end defp process_sns_message(%{"Type" => "Notification"} = message) do # Process email event notification process_email_event_notification(message) end defp process_sns_message(%{"Type" => type}) do Logger.warning("Unknown SNS message type", %{type: type}) {:ok, :unknown_type} end # Automatically confirm SNS subscription defp confirm_subscription(%{"SubscribeURL" => subscribe_url}) do # Log subscription URL for manual confirmation if needed Logger.info("SNS subscription confirmation requested", %{ subscribe_url: subscribe_url }) # For now, just log the URL - implement actual HTTP client based on your needs # You can add :httpc (built into Erlang) or :req if needed: # # case :httpc.request(:get, {subscribe_url, []}, [{:timeout, 10_000}], []) do # {:ok, {{_, 200, _}, _headers, _body}} -> # Logger.info("SNS subscription confirmed") # {:ok, :subscription_confirmed} # {:ok, {{_, status_code, _}, _headers, _body}} -> # Logger.error("Failed to confirm SNS subscription", %{status_code: status_code}) # {:error, :confirmation_failed} # {:error, reason} -> # Logger.error("HTTP error confirming SNS subscription", %{reason: inspect(reason)}) # {:error, :http_error} # end # For now, return success and log for manual confirmation {:ok, :subscription_logged} end defp confirm_subscription(_message) do {:error, :missing_subscribe_url} end # Process email event notification defp process_email_event_notification(%{"Message" => message_json}) do case Jason.decode(message_json) do {:ok, event_data} -> process_email_event(event_data) {:error, reason} -> Logger.error("Failed to parse SNS message JSON", %{ reason: inspect(reason), message: String.slice(message_json, 0, 200) }) {:error, :invalid_json} end end defp process_email_event_notification(_message) do {:error, :missing_message} end # Process individual email event defp process_email_event(event_data) do if EmailTracking.enabled?() and EmailTracking.ses_events_enabled?() do case EmailTracking.process_webhook_event(event_data) do {:ok, :skipped} -> {:ok, :event_skipped} {:ok, _event} -> {:ok, :event_processed} {:error, :message_id_not_found} -> Logger.info("Email log not found for webhook event", %{ event_type: event_data["eventType"], message_id: get_in(event_data, ["mail", "messageId"]) }) {:ok, :log_not_found} {:error, reason} -> Logger.error("Failed to process email event", %{ reason: inspect(reason), event_type: event_data["eventType"] }) {:error, :processing_failed} end else Logger.debug("Email tracking disabled, skipping event") {:ok, :tracking_disabled} end end ## --- Helper Functions --- # Get remote IP address from connection defp get_remote_ip(conn) do case Plug.Conn.get_req_header(conn, "x-forwarded-for") do [forwarded_ips] -> # Take first IP from forwarded chain forwarded_ips |> String.split(",") |> List.first() |> String.trim() [] -> # Direct connection conn.remote_ip |> :inet.ntoa() |> to_string() end end # Check if IP is in AWS SNS ranges defp ip_in_aws_ranges?(ip_string) do case :inet.parse_address(String.to_charlist(ip_string)) do {:ok, ip_tuple} -> Enum.any?(@aws_sns_ip_ranges, fn range -> ip_in_cidr_range?(ip_tuple, range) end) {:error, _} -> false end end # Check if IP is in CIDR range defp ip_in_cidr_range?(ip_tuple, cidr_range) when is_binary(cidr_range) do case String.split(cidr_range, "/") do [network_str, prefix_len_str] -> case { :inet.parse_address(String.to_charlist(network_str)), Integer.parse(prefix_len_str) } do {{:ok, network_tuple}, {prefix_len, _}} -> ip_in_cidr_range?(ip_tuple, network_tuple, prefix_len) _ -> false end _ -> false end end # Actual CIDR range check defp ip_in_cidr_range?({a, b, c, d}, {na, nb, nc, nd}, prefix_len) when prefix_len <= 32 do ip_int = (a <<< 24) + (b <<< 16) + (c <<< 8) + d network_int = (na <<< 24) + (nb <<< 16) + (nc <<< 8) + nd mask = bnot((1 <<< (32 - prefix_len)) - 1) (ip_int &&& mask) == (network_int &&& mask) end defp ip_in_cidr_range?(_, _, _), do: false # Rate limiting implementation (simple in-memory) defp check_ip_rate_limit(_ip) do # This is a simplified implementation # In production, use a proper rate limiting solution like Hammer # rate_limit_config = @default_rate_limit # cache_key = "webhook_rate_limit:#{ip}" # current_time = System.system_time(:second) # window_start = current_time - rate_limit_config.window_seconds # For now, always allow (implement proper rate limiting based on your needs) :ok end # Parse ISO8601 timestamp defp parse_timestamp(timestamp_str) do case DateTime.from_iso8601(timestamp_str) do {:ok, datetime, _offset} -> {:ok, datetime} {:error, reason} -> {:error, reason} end end # Verify AWS SNS signature (simplified implementation) defp verify_aws_sns_signature(sns_message) do # This is a simplified implementation # In production, implement full SNS signature verification: # https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html signature = sns_message["Signature"] signing_cert_url = sns_message["SigningCertURL"] case {signature, signing_cert_url} do {sig, cert_url} when is_binary(sig) and is_binary(cert_url) -> # TODO: Implement full signature verification # For now, just check that required fields are present :ok _ -> :error end end # Configuration helpers defp signature_verification_enabled? do System.get_env("PHOENIX_KIT_WEBHOOK_VERIFY_SNS_SIGNATURE", "true") == "true" end defp aws_ip_check_enabled? do System.get_env("PHOENIX_KIT_WEBHOOK_CHECK_AWS_IP", "true") == "true" end defp rate_limiting_enabled? do System.get_env("PHOENIX_KIT_WEBHOOK_RATE_LIMIT_ENABLED", "true") == "true" end end