;; This Source Code Form is subject to the terms of the Mozilla Public ;; License, v. 2.0. If a copy of the MPL was not distributed with this ;; file, You can obtain one at http://mozilla.org/MPL/2.0/. ;; ;; Copyright (c) KALEIDOS INC Sucursal en EspaƱa SL (ns app.http.awsns "AWS SNS webhook handler for bounces." (:require [app.common.exceptions :as ex] [app.common.logging :as l] [app.common.pprint :as pp] [app.common.schema :as sm] [app.db :as db] [app.db.sql :as sql] [app.http.client :as http] [app.main :as-alias main] [app.setup :as-alias setup] [app.tokens :as tokens] [clojure.data.json :as j] [cuerdas.core :as str] [integrant.core :as ig] [yetti.request :as yreq] [yetti.response :as-alias yres]) (:import java.net.URI java.security.cert.CertificateFactory java.security.Signature java.util.Base64)) (declare parse-json) (declare handle-request) (declare parse-notification) (declare process-report) (defn- valid-sns-url? "Validates that a URL originates from an SNS endpoint. Only accepts sns..amazonaws.com hosts. See: https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html" [url] (when (string? url) (try (let [uri (URI. url) host (.getHost uri)] (and (= "https" (.getScheme uri)) (boolean (re-matches #"(?i)sns\.[a-z0-9-]+\.amazonaws\.com" host)))) (catch Exception _ false)))) ;; AWS SNS Signature Verification Field Sets ;; See: https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message-verify-message-signature.html ;; ;; V1 and V2 use the SAME field sets (only the hash algorithm differs: SHA1 vs SHA256) ;; The "Signature" field is NEVER part of the string-to-sign (it's the output, not input) ;; "SigningCertURL" and "SignatureVersion" are metadata, not signed ;; Notification: Message, MessageId, Subject (if present), Timestamp, TopicArn, Type (def ^:private notification-fields ["Message" "MessageId" "Subject" "Timestamp" "TopicArn" "Type"]) ;; SubscriptionConfirmation: Message, MessageId, SubscribeURL, Timestamp, Token, TopicArn, Type (def ^:private subscription-fields ["Message" "MessageId" "SubscribeURL" "Timestamp" "Token" "TopicArn" "Type"]) (defn- build-string-to-sign "Builds the string-to-sign for AWS SNS signature verification. V1 and V2 use the same field sets (only hash algorithm differs: SHA1 vs SHA256). See: https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message-verify-message-signature.html" [body] (let [msg-type (get body "Type") fields (if (= "SubscriptionConfirmation" msg-type) subscription-fields notification-fields)] (->> fields (filter #(contains? body %)) (map #(str % "\n" (get body %) "\n")) (apply str)))) (defn- fetch-certificate "Fetches the X.509 certificate from the given URL. Returns an InputStream that must be closed by the caller. See: https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html" [cfg cert-url] (let [response (http/req cfg {:uri cert-url :method :get :timeout 10000} {:sync? true :response-type :input-stream})] (when-not (= 200 (:status response)) (when-let [body (:body response)] (.close ^java.io.Closeable body)) (l/wrn :hint "failed to fetch SNS signing certificate" :action "sns-cert-fetch-failed" :status (:status response) :cert-url cert-url) (ex/raise :type :internal :code :cert-fetch-failed)) (:body response))) (defn- verify-signature "Verifies the RSA signature of the message. See: https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html" [cfg body] (let [cert-url (get body "SigningCertURL") signature (get body "Signature") sig-version (get body "SignatureVersion") algorithm (case sig-version "1" "SHA1withRSA" "2" "SHA256withRSA" nil)] (when-not algorithm (throw (ex-info "Unsupported SNS signature version" {:type :validation :version sig-version}))) (when (and cert-url signature) (try (let [string-sign (build-string-to-sign body)] (with-open [cert-stream (fetch-certificate cfg cert-url)] (let [cf (CertificateFactory/getInstance "X.509") cert (.generateCertificate cf cert-stream) sig (Signature/getInstance algorithm)] (.initVerify sig (.getPublicKey cert)) (.update sig (.getBytes string-sign java.nio.charset.StandardCharsets/UTF_8)) (.verify sig (.decode (Base64/getDecoder) signature))))) (catch clojure.lang.ExceptionInfo e (let [data (ex-data e)] (if (= :validation (:type data)) (throw e) (do (l/wrn :hint "SNS signature verification exception" :action "sns-signature-verification-exception" :cause e) false)))) (catch Exception e (l/wrn :hint "SNS signature verification exception" :action "sns-signature-verification-exception" :cause e) false))))) (defn- verify-sns-message! "Verifies the AWS SNS message signature and URL validity. Throws if verification fails. See: https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html" [cfg body] (let [cert-url (get body "SigningCertURL") subscribe-url (get body "SubscribeURL") mtype (get body "Type")] (when-not (valid-sns-url? cert-url) (l/wrn :hint "SNS certificate URL not from amazonaws.com" :action "sns-invalid-cert-url" :message-type mtype :signing-cert-url cert-url) (ex/raise :type :validation :code :invalid-signing-cert-url :hint "SigningCertURL must be from amazonaws.com")) (when (and (= mtype "SubscriptionConfirmation") (not (valid-sns-url? subscribe-url))) (l/wrn :hint "SNS subscribe URL not from amazonaws.com" :action "sns-invalid-subscribe-url" :message-type mtype :subscribe-url subscribe-url) (ex/raise :type :validation :code :invalid-subscribe-url :hint "SubscribeURL must be from amazonaws.com")) (when-not (verify-signature cfg body) (l/wrn :hint "SNS signature verification failed" :action "sns-signature-verification-failed" :message-type mtype :topic-arn (get body "TopicArn") :signing-cert-url cert-url) (ex/raise :type :authentication :code :invalid-signature :hint "SNS signature verification failed")))) (defmethod ig/assert-key ::routes [_ params] (assert (http/client? (::http/client params)) "expect a valid http client") (assert (sm/valid? ::setup/props (::setup/props params)) "expected valid setup props") (assert (db/pool? (::db/pool params)) "expect valid database pool")) (defmethod ig/init-key ::routes [_ cfg] (letfn [(handler [request] (let [data (-> request yreq/body slurp) result (handle-request cfg data)] {::yres/status (or (:status result) 200)}))] ["/sns" {:handler handler :allowed-methods #{:post}}])) (defn handle-request [cfg data] (try (let [body (parse-json data) mtype (get body "Type")] (when body (verify-sns-message! cfg body)) (cond (= mtype "SubscriptionConfirmation") (let [surl (get body "SubscribeURL") stopic (get body "TopicArn")] (l/info :action "subscription received" :topic stopic :url surl) (http/req cfg {:uri surl :method :post :timeout 10000} {:sync? true}) {:status 200}) (= mtype "Notification") (if-let [message (parse-json (get body "Message"))] (do (let [notification (parse-notification cfg message)] (process-report cfg notification)) {:status 200}) (do (l/wrn :hint "notification with missing or unparseable Message field" :action "sns-missing-message") {:status 400})) :else (do (l/warn :hint "unexpected data received" :report (pr-str body)) {:status 400}))) (catch clojure.lang.ExceptionInfo e (let [data (ex-data e)] (if (#{:validation :authentication} (:type data)) (do (l/wrn :hint "SNS message validation failed" :action "sns-validation-failed" :code (:code data)) {:status 400}) (do (l/error :hint "unexpected exception on awsns" :cause e) {:status 500})))) (catch Throwable cause (l/error :hint "unexpected exception on awsns" :cause cause) {:status 500}))) (defn- parse-bounce [data] {:type "bounce" :kind (str/lower (get data "bounceType")) :category (str/lower (get data "bounceSubType")) :feedback-id (get data "feedbackId") :timestamp (get data "timestamp") :recipients (->> (get data "bouncedRecipients") (mapv (fn [item] {:email (str/lower (get item "emailAddress")) :status (get item "status") :action (get item "action") :dcode (get item "diagnosticCode")})))}) (defn- parse-complaint [data] {:type "complaint" :user-agent (get data "userAgent") :kind (get data "complaintFeedbackType") :category (get data "complaintSubType") :timestamp (get data "arrivalDate") :feedback-id (get data "feedbackId") :recipients (->> (get data "complainedRecipients") (mapv #(get % "emailAddress")) (mapv str/lower))}) (defn- extract-headers [mail] (reduce (fn [acc item] (let [key (get item "name") val (get item "value")] (assoc acc (str/lower key) val))) {} (get mail "headers"))) (defn- extract-identity [cfg headers] (let [tdata (get headers "x-penpot-data")] (when-not (str/empty? tdata) (let [result (tokens/verify cfg {:token tdata :iss :profile-identity})] (:profile-id result))))) (defn- parse-notification [cfg message] (let [type (get message "notificationType") data (case type "Bounce" (parse-bounce (get message "bounce")) "Complaint" (parse-complaint (get message "complaint")) {:type (keyword (str/lower type)) :message message})] (when data (let [mail (get message "mail")] (when-not mail (ex/raise :type :internal :code :incomplete-notification :hint "no email data received, please enable full headers report")) (let [headers (extract-headers mail) mail {:destination (get mail "destination") :source (get mail "source") :timestamp (get mail "timestamp") :subject (get-in mail ["commonHeaders" "subject"]) :headers headers}] (assoc data :mail mail :profile-id (extract-identity cfg headers))))))) (defn- parse-json [v] (try (j/read-str v) (catch Throwable cause (l/wrn :hint "unable to decode request body" :cause cause)))) (defn- register-bounce-for-profile [{:keys [::db/pool]} {:keys [type kind profile-id] :as report}] (when (= kind "permanent") (try (db/insert! pool :profile-complaint-report {:profile-id profile-id :type (name type) :content (db/tjson report)}) (catch Throwable cause (l/warn :hint "unable to persist profile complaint" :cause cause))) (doseq [recipient (:recipients report)] (db/insert! pool :global-complaint-report {:email (:email recipient) :type (name type) :content (db/tjson report)})) (let [profile (db/exec-one! pool (sql/select :profile {:id profile-id}))] (when (some #(= (:email profile) (:email %)) (:recipients report)) ;; If the report matches the profile email, this means that ;; the report is for itself, can be caused when a user ;; registers with an invalid email or the user email is ;; permanently rejecting receiving the email. In this case we ;; have no option to mark the user as muted (and in this case ;; the profile will be also inactive. (l/inf :hint "mark profile: muted" :profile-id (str (:id profile)) :email (:email profile) :reason "bounce report" :report-id (:feedback-id report)) (db/update! pool :profile {:is-muted true} {:id profile-id} {::db/return-keys false}))))) (defn- register-complaint-for-profile [{:keys [::db/pool]} {:keys [type profile-id] :as report}] (try (db/insert! pool :profile-complaint-report {:profile-id profile-id :type (name type) :content (db/tjson report)}) (catch Throwable cause (l/warn :hint "unable to persist profile complaint" :cause cause))) ;; TODO: maybe also try to find profiles by email and if exists ;; register profile reports for them? (doseq [email (:recipients report)] (db/insert! pool :global-complaint-report {:email email :type (name type) :content (db/tjson report)})) (let [profile (db/exec-one! pool (sql/select :profile {:id profile-id}))] (when (some #(= % (:email profile)) (:recipients report)) ;; If the report matches the profile email, this means that ;; the report is for itself, rare case but can happen; In this ;; case just mark profile as muted (very rare case). (l/inf :hint "mark profile: muted" :profile-id (str (:id profile)) :email (:email profile) :reason "complaint report" :report-id (:feedback-id report)) (db/update! pool :profile {:is-muted true} {:id profile-id} {::db/return-keys false})))) (defn- process-report [cfg {:keys [type profile-id] :as report}] (cond ;; In this case we receive a bounce/complaint notification without ;; confirmed identity, we just emit a warning but do nothing about ;; it because this is not a normal case. All notifications should ;; come with profile identity. (nil? profile-id) (l/wrn :hint "not-identified report" ::l/body (pp/pprint-str report {:length 40 :level 6})) (= "bounce" type) (do (l/trc :hint "bounce report" ::l/body (pp/pprint-str report {:length 40 :level 6})) (register-bounce-for-profile cfg report)) (= "complaint" type) (do (l/trc :hint "complaint report" ::l/body (pp/pprint-str report {:length 40 :level 6})) (register-complaint-for-profile cfg report)) :else (l/wrn :hint "unrecognized report" ::l/body (pp/pprint-str report {:length 20 :level 4}))))