Stop services reconnecting dropped nodes; bind handshake signatures to names

README.md

@@ -430,7 +430,10 @@ The layers, bottom up:
   with a status (`:nok` for a name it won't take, `:alive` when it's
   connected to that run of the node already, and of two nodes connecting
   to each other at once only one connection survives) and a challenge, and each end proves it holds
-  the cookie by signing the other's challenge with HMAC-SHA256. `start!`'s
+  the cookie by signing the other's challenge with HMAC-SHA256. The
+  signature covers both node names as well as the challenge, unlike
+  Erlang's, so a node without the cookie can't relay one node's answer to
+  a third and get in under its name. `start!`'s
   `:cookie` sets it (by default nodes in one VM share a cookie drawn at
   startup), or `:auth` takes any `ensemble.node/Auth`. A connection that
   hasn't shaken hands within `:handshake-timeout` is closed, and a node
@@ -482,7 +485,8 @@ The layers, bottom up:
 - **Passivation is an extension.** Erlang's hibernation stays in memory.
 - **Distribution ships only the loopback transport**; a network transport
   is a library implementing `ensemble.transport`. The cookie handshake
-  signs with HMAC-SHA256 rather than Erlang's MD5 digest, and the frames
+  signs with HMAC-SHA256 over the challenge and both names rather than
+  Erlang's MD5 digest of the challenge, and the frames
   are EDN, not the external term format, so ensemble nodes do not talk to
   Erlang nodes. global and pg have the default scope only.
 - There is no hot code loading, and no reductions (jolt preempts fibers on

src/ensemble/dist.clj

@@ -60,8 +60,13 @@
 ;;
 ;;   A -> B  [:name A creation]
 ;;   B -> A  [:status s], and if s is :ok, [:challenge B creation cB]
-;;   A -> B  [:reply cA (sign cB)]
-;;   B -> A  [:ack (sign cA)]
+;;   A -> B  [:reply cA (sign [cB A B])]
+;;   B -> A  [:ack (sign [cA B A])]
+;;
+;; A signature covers the challenge, the node signing and the node it is
+;; meant for.  Erlang's covers the challenge alone, which lets a node
+;; without the cookie that gets one node to dial it pass the challenge of
+;; a third along and answer that third node as the first.
 ;;
 ;; Each step takes this end's state, the frame read, and ctx: {:up-creation
 ;; f :pending? f :fresh c :sign f} -- the creation of the run of a peer
@@ -138,15 +143,15 @@
       (let [[_ nm cr c] frame]
         (if (= nm (:peer hs))
           [:Next (assoc hs :phase :ack :peer-creation cr :mine (:fresh ctx))
-           [[:reply (:fresh ctx) (sign c)]]]
+           [[:reply (:fresh ctx) (sign [c (:self hs) nm])]]]
           [:Fail [:wrong-node nm] []]))
       (and (= :reply phase) (= :reply tag))
       (let [[_ c digest] frame]
-        (if (= digest (sign (:mine hs)))
-          [:Up (assoc hs :phase :up) [[:ack (sign c)]]]
+        (if (= digest (sign [(:mine hs) (:peer hs) (:self hs)]))
+          [:Up (assoc hs :phase :up) [[:ack (sign [c (:self hs) (:peer hs)])]]]
           [:Fail :bad-cookie []]))
       (and (= :ack phase) (= :ack tag))
-      (if (= (second frame) (sign (:mine hs)))
+      (if (= (second frame) (sign [(:mine hs) (:peer hs) (:self hs)]))
         [:Up (assoc hs :phase :up) []]
         [:Fail :bad-cookie []])
       :else [:Fail [:unexpected frame] []])))

src/ensemble/global.clj

@@ -97,7 +97,8 @@
           (let [nm (some (fn [[n [_ r]]] (when (= r mref) n)) (:locks st))]
             [:noreply (if nm (update st :locks dissoc nm) st)])))
       (and (vector? msg) (= :nodeup (first msg)))
-      (do (gs/cast! [:At ::registrar (second msg)] [:sync (vec (table))])
+      ;; the peer may be down again by now: the sync must not reconnect it
+      (do (node/without-connecting (gs/cast! [:At ::registrar (second msg)] [:sync (vec (table))]))
           [:noreply st])
       (and (vector? msg) (= :nodedown (first msg)))
       (let [peer (second msg)]

src/ensemble/node.clj

@@ -21,7 +21,7 @@
     which a Codec (ensemble.codec, EDN by default) turns into a value;
   - a connection is used only once the handshake has named the peer and
     each end has proved it holds the cookie, by signing the other's
-    challenge with HMAC-SHA256;
+    challenge, with its own name and the other's, with HMAC-SHA256;
   - then frames carry the operations of the distribution protocol, each
     an envelope [op hdr body]: hdr the operation's own fields (the pids
     it names, a ref, a name), body the data it carries -- a message, an
@@ -58,7 +58,8 @@
 (defprotocol Auth
   (-sign [a challenge]
     "Sign challenge, a string, as only a node that may connect could.
-    Both ends of a connection must sign alike."))
+    Both ends of a connection must sign alike.  The handshake's challenge
+    string names the node signing and the node it is for as well."))
 
 (defrecord CookieAuth [key]
   Auth
@@ -328,7 +329,8 @@
     {:up-creation (fn [p] (get-in s [:conns p :creation]))
      :pending? (fn [p] (contains? (:connecting s) p))
      :fresh (random-hex 16)
-     :sign (fn [c] (-sign (:auth s) (str c)))}))
+     ;; what is signed is [challenge signer verifier], printed
+     :sign (fn [v] (-sign (:auth s) (pr-str v)))}))
 
 (defn handshake-step!
   "Take a handshake frame on conn c of node n, through dist/handshake: send
@@ -456,12 +458,25 @@
         (finally
           (swap! nodes update-in [n :connecting] dissoc peer))))))
 
+(def ^:dynamic *connect*
+  "Whether a send to a node with no connection opens one.  False within
+  without-connecting."
+  true)
+
+(defmacro without-connecting
+  "Run body with sends to a node that is not connected dropped instead of
+  connecting to it, as erlang:send/3's noconnect: for a service's own
+  traffic, which must not bring back a node that just went down."
+  [& body]
+  `(binding [*connect* false] ~@body))
+
 (defn- conn-to
   "The connection from n to peer, opened on first use as Erlang connects
-  nodes on first contact; nil when peer cannot be reached or refuses."
+  nodes on first contact, unless *connect* is false; nil when peer cannot
+  be reached or refuses."
   [n peer]
   (or (get-in @nodes [n :conns peer :conn])
-      (when (and (state n) (not= n peer)) (open! n peer))))
+      (when (and *connect* (state n) (not= n peer)) (open! n peer))))
 
 (defn- accept!
   "Take a connection opened to node n: it shakes hands before anything

src/ensemble/pg.clj

@@ -55,8 +55,12 @@
           (recur (into (subvec ms 0 i) (subvec ms (inc i))) (rest xs) (conj out x))))
       [ms out])))
 
-(defn- broadcast! [msg]
-  (doseq [n (act/nodes)] (gs/cast! [:At ::scope n] msg)))
+(defn- broadcast!
+  "Cast msg to the scope on every connected node; one that has gone down
+  meanwhile is not connected to again for it."
+  [msg]
+  (node/without-connecting
+    (doseq [n (act/nodes)] (gs/cast! [:At ::scope n] msg))))
 
 (defn- leave-local
   "st with pids leaving local group g, told to monitors and peers."
@@ -131,7 +135,9 @@
           :watcher (let [[g ref] more] [:noreply (update-in st [:watchers g] dissoc ref)])
           [:noreply st]))
       (and (vector? msg) (= :nodeup (first msg)))
-      (do (gs/cast! [:At ::scope (second msg)] [:sync (act/node) (into {} (filter (comp seq val)) (:local st))])
+      ;; the peer may be down again by now: the sync must not reconnect it
+      (do (node/without-connecting
+            (gs/cast! [:At ::scope (second msg)] [:sync (act/node) (into {} (filter (comp seq val)) (:local st))]))
           [:noreply st])
       (and (vector? msg) (= :nodedown (first msg)))
       [:noreply (publish! (set-remote st (second msg) {}))]

test/ensemble/dist_spec.clj

@@ -203,6 +203,28 @@
         [_ _ out-a] (handshake ia (second out-b) (ctx-for "c" :ca))]
     (= [:Fail :bad-cookie []] (handshake hb (first out-a) (ctx-for "d" :cb)))))
 
+;; a node m without the cookie gets a to dial it and hands a the challenge
+;; c sent m: what a signs names m as the node it answers, so c, expecting
+;; itself there, refuses it
+(law a-signature-relayed-to-another-node-is-refused
+  (let [;; c's side: m claims to be a
+        [_ hc out-c] (handshake (accept-handshake :c.vm 3) [:name :a.vm 1] (ctx-for "k" :cc))
+        cc (nth (second out-c) 3)
+        ;; a dialed m, and m passes c's challenge on as its own
+        [_ ia] (handshake (second (open-handshake :a.vm 1 :m.vm)) [:status :ok] (ctx-for "k" :ca))
+        [_ _ out-a] (handshake ia [:challenge :m.vm 9 cc] (ctx-for "k" :ca))]
+    (= [:Fail :bad-cookie []] (handshake hc (first out-a) (ctx-for "k" :cc)))))
+
+;; nor does an end's own challenge, reflected back to it, prove the peer:
+;; b sends a's challenge as its own, and hands a's answer back as the ack
+(law a-reflected-ack-is-refused
+  (let [ctx (ctx-for "k" :ca)
+        [_ ia] (handshake (second (open-handshake :a.vm 1 :b.vm)) [:status :ok] ctx)
+        [_ ia out-a] (handshake ia [:challenge :b.vm 2 :ca] ctx)
+        ;; what a signed for b, sent back to a as b's proof
+        [_ _ sig] (first out-a)]
+    (= [:Fail :bad-cookie []] (handshake ia [:ack sig] ctx))))
+
 (law the-acceptor-proves-itself-too
   ;; the initiator checks the acceptor's answer to its challenge
   (let [ctx {:up-creation (fn [_] nil) :pending? (fn [_] false) :fresh :ca :sign (sign-with "c")}

test/ensemble/services_test.clj

@@ -9,7 +9,8 @@
             [ensemble.node :as node]
             [ensemble.pg :as pg]
             [ensemble.process :as proc]
-            [ensemble.rpc :as rpc]))
+            [ensemble.rpc :as rpc]
+            [ensemble.sys :as sys]))
 
 (defn- sleep [ms] (a/<!! (a/timeout ms)))
 
@@ -268,3 +269,25 @@
                      (deliver r [during (mons)]))
                    (receive [:never nil]))))
     (is (= [(inc before) before] (deref r 2000 nil)))))
+
+(deftest a-service-catching-up-on-nodeup-does-not-reconnect-a-dropped-node
+  ;; the services hear [:nodeup b] only after the test has dropped b, as a
+  ;; slow scheduler may arrange: their sync must not bring b back
+  (let [a (fresh-node) b (fresh-node)
+        services (fn [n] (node/with-node n (into [] (keep act/whereis) [:ensemble.pg/scope :ensemble.global/registrar])))
+        held (concat (services a) (services b))
+        got (atom [])
+        w (node/with-node a (act/spawn (fn [] (node/monitor-nodes! true)
+                                         (loop [] (swap! got conj (receive [m m])) (recur)))))]
+    (is (= 4 (count held)) "pg and global run on both nodes")
+    (doseq [s held] (sys/suspend! s))
+    (sleep 20)
+    (node/with-node a (node/connect! b))
+    (is (eventually #(= [[:nodeup b]] @got)))
+    (node/with-node a (node/disconnect! b))
+    (is (eventually #(= [[:nodeup b] [:nodedown b]] @got)))
+    (doseq [s held] (sys/resume! s))
+    (sleep 200)
+    (is (= [[:nodeup b] [:nodedown b]] @got) "no sync reconnected b")
+    (is (= [] (node/with-node a (node/connected))))
+    (act/exit! w :kill)))