Clojure POP3 server automata definition

automata.clj

(ns sentinel.pop3.automata
  (:require [instaparse.core :as insta]
            [clojure.java.io :as io]
            [clojure.string :as s]
            [sentinel.auto :refer [fsm automate]]
            [taoensso.timbre :as timbre]
            [sentinel.address :as address]
            [sentinel.db.main :as db-main]
            [buddy.hashers :as hashers]
            [amazonica.aws.s3 :as s3])
  (:import (org.h2.jdbc JdbcBlob)))


(def connections-atom (atom {}))


(timbre/refer-timbre)

(insta/set-default-input-format! :abnf)

(def pop3-command-parser
  (insta/parser
    (io/resource "pop3-command.abnf")))

(defn parse-command [input]
  (if (keyword? input) input
                       (vec (insta/transform {:keyword (comp s/upper-case str) :param str} (pop3-command-parser input)))))

;https://tools.ietf.org/html/rfc1939

(defn aquire-lock [domains-map {:keys [user] :as state} password]
  (let [[username domain-name :as u] (address/split user)]
    (when-let [{:keys [domain/db] :as domain} (domains-map domain-name)]
      (let [{:keys [id password_hash]} (db-main/get-account-by-username db domain-name username)]
        (when (and (hashers/check password password_hash)
                   (not (get-in @connections-atom [domain-name id])))
          (swap! connections-atom assoc-in [domain-name id] true))))))

(defn sign-in [{:keys [user] :as state} domains-map]
  (let [[username domain-name] (address/split user)]
    (when-let [{:keys [domain/db] :as domain} (domains-map domain-name)]
      (let [{:keys [id]} (db-main/get-account-by-username db domain-name username)]
        {:user-id id
         :domain domain}))))


(defn long-response [ok coll]
  (cons ok (concat coll '("."))))

(defn *quit [{:keys [messages-list deleted-set user-id]
              {domain-name :domain/name} :domain} input]
  (swap! connections-atom update domain-name dissoc user-id)
  {:emit "+OK Bye"})

(defn *stat [{id :user-id {domain-name :domain/name db :domain/db} :domain} _]
  (let [{:keys [c s]} (db-main/stat db domain-name id)]
    {:emit (str "+OK " ((fnil long 0) c) " " ((fnil long 0) s))}))

(defn *list [{:keys [messages-list deleted-set]
              {domain-name :domain/name db :domain/db} :domain} [_ ^String n]]
  (if-not n
    (let [ok (str "+OK " (count messages-list) " messages (" (reduce + (map :length messages-list)) " octets)")]
      {:emit (long-response ok (->> messages-list
                                    (remove #(deleted-set (:num %)))
                                    (map (fn [{:keys [num length]}] (str num " " length)))))})
    (try
      (let [n (Integer. n)]
        (assert (and (<= 1 n (count messages-list)) (not (contains? deleted-set n))))
        {:emit (let [{:keys [length] :as m} (some #(when (= n (:num %)) %) messages-list)]
                 (str "+OK " n " " length))})
      (catch Throwable _ {:emit "-ERR invalid message number (deleted?)"}))))

(defn *uidl [{:keys [messages-list deleted-set]
              {domain-name :domain/name db :domain/db} :domain} [_ ^String n]]
  (if-not n
    (let [ok "+OK"]
      {:emit (long-response ok (->> messages-list
                                    (remove #(deleted-set (:num %)))
                                    (map (fn [{:keys [num id]}] (str num " " id)))))})
    (try
      (let [n (Integer. n)]
        (assert (and (<= 1 n (count messages-list)) (not (contains? deleted-set n))))
        {:emit (let [{:keys [id] :as m} (some #(when (= n (:num %)) %) messages-list)]
                 (str "+OK " n " " id))})
      (catch Throwable _ {:emit "-ERR invalid message number (deleted?)"}))))

(defn *dele [{:keys [messages-list deleted-set] {domain-name :domain/name db :domain/db} :domain}
             [_ ^String n]]
  (try
    (let [n (Integer. n)]
      (assert (and (<= 1 n (count messages-list)) (not (contains? deleted-set n))))
      {:emit (str "+OK message " n " deleted")
       :deleted-set (conj deleted-set n)})
    (catch Throwable _ {:emit "-ERR invalid message number (already deleted?)"})))

(defn *rset [{:keys [deleted-set]} _]
  {:emit (str "+OK " (count deleted-set) " messages undeleted")
   :deleted-set #{}})




(defn message-stream [{credentials :provider/credentials domain-name :domain/name db :domain/db}
                      message-id]
  (let [{:keys [message path storage]} (db-main/get-message db domain-name message-id)]
    (case storage
      "db" (let [^JdbcBlob message-blob (:message (db-main/get-message db domain-name message-id))]
             (.getBinaryStream message-blob))
      "s3" (let [[bucket-name key] (s/split path #";")] 
             (:input-stream (s3/get-object credentials :bucket-name bucket-name :key key))))))

(defn *retr [{:keys [messages-list deleted-set domain]}
             [_ ^String msg]]
  (try
    (let [msg (Integer. msg)
          {:keys [id length] :as m} (some #(when (= msg (:num %)) %) messages-list)]
      (assert (and (<= 1 msg (count messages-list)) (not (contains? deleted-set msg))))
      (let [is (message-stream domain id)]
        {:emit (long-response "+OK"
                              (map #(if (s/starts-with? % ".") (str "." %) %)
                                   (line-seq (io/reader is))))}))
    (catch Throwable e (do
                         (error e)
                         {:emit "-ERR"}))))

(defn *top [{:keys [messages-list deleted-set domain]} 
            [_ ^String msg ^String n]]
  (try
    (let [msg (Integer. msg)
          n (Integer. n)
          {:keys [id length] :as m} (some #(when (= msg (:num %)) %) messages-list)]
      (assert (and (nat-int? n) (<= 1 msg (count messages-list)) (not (contains? deleted-set msg))))
      (let [is (message-stream domain id)
            lines (map #(if (s/starts-with? % ".") (str "." %) %)
                       (line-seq (io/reader is)))
            [header body] (split-with #(not= % "") lines)]
        {:emit (long-response "+OK"
                              (concat header (take (inc n) body)))}))
    (catch Throwable e (do
                         (error e)
                         {:emit "-ERR"}))))


(defn pop3-automata [domains-map]
  (fsm [state input]
    [:hello
     [_ :client/connect] "+OK POP3 server ready" :authorization]

    [:authorization
     [_ ["USER" user]] (fn [_ _] {:emit (str "+OK hello " user " please send PASS") :user user}) :authorization
     [{:user user} ["PASS" (password :guard (partial aquire-lock domains-map state))]]
     (fn [state _]
       (let [{:keys [user-id domain]} (sign-in state domains-map)
             {domain-name :domain/name db :domain/db} domain
             l (db-main/messages-list db domain-name user-id)]
         {:emit    (str "+OK " user " authorized with id " user-id)
          :user-id user-id
          :domain domain
          :messages-list l
          :deleted-set #{}}))
     :transaction
     [_ ["CAPA"]] ["+OK" "USER" "UIDL" "TOP" "."] :authorization
     [_ _] "-ERR" :authorization]

    [:transaction
     [_ ["STAT"]] *stat :transaction
     [_ ["LIST"]] *list :transaction
     [_ ["LIST" n]] *list :transaction
     [_ ["RETR" n]] *retr :transaction
     [_ ["DELE" n]] *dele :transaction
     [_ ["NOOP"]] "+OK" :transaction
     [_ ["CAPA"]] ["+OK" "USER" "UIDL" "TOP" "."] :transaction
     [_ ["RSET"]] *rset :transaction
     [_ ["TOP" msg n]] *top :transaction
     [_ ["UIDL"]] *uidl :transaction
     [_ ["UIDL" msg]] *uidl :transaction
     [_ _] "-ERR Invalid command" :transaction]

    [[:authorization :transaction]
     [_ ["QUIT"]] *quit :quit]
    [[:authorization :transaction]
     [_ :client/disconnect] *quit :quit]
    [:quit]))