1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146 | (ns ird.p2p.db.gating-test
"Connection gating with :gater, on three real nodes, always on: A lets
only C in. C dials A and they replicate both ways; B — same secret, same
database — dials A and gets no connection, so nothing of B's reaches A;
and the gate is inbound only: when A itself dials B they replicate like
anyone else. If your environment cannot bind loopback tcp, set
IRD_P2P_DB_NO_NODE=1 to skip."
(:require [clojure.string :as str]
[cljs.test :refer [deftest is testing]]
[cljs.core.async :as async :refer [<!]]
[ird.p2p.db.core :as db]
[ird.p2p.db.errors :as err]
[ird.p2p.db.test-util :as tu :refer [deftest-async]]))
(deftest-async gater-refuses-inbound-connections
{:timeout-ms 90000}
(if (tu/skip-nodes?)
(is true "IRD_P2P_DB_NO_NODE set — skipping the real-node gating test")
(let [secret (db/random-secret)
opts {:secret secret :app-salt "test/v1" :name "gated"}
c (<! (db/open opts))
_ (is (not (tu/error? c)) (some-> c ex-message))
b (<! (db/open opts))
_ (is (not (tu/error? b)) (some-> b ex-message))
asked (atom [])
c-suffix (str "/p2p/" (db/peer-id c))
a (<! (db/open (assoc opts :gater (fn [addr] (swap! asked conj addr) (str/ends-with? addr c-suffix)))))
_ (is (not (tu/error? a)) (some-> a ex-message))
a-addrs (<! (db/multiaddrs a))
ca (tu/collector (db/entries-stream a))
cb (tu/collector (db/entries-stream b))
cc (tu/collector (db/entries-stream c))
pa (tu/collector (db/peers-stream a))
pb (tu/collector (db/peers-stream b))
pc (tu/collector (db/peers-stream c))
ra (<! (db/append a {:from "a"}))
rb (<! (db/append b {:from "b"}))]
(is (not (tu/error? ra)) (some-> ra ex-message))
(is (not (tu/error? rb)) (some-> rb ex-message))
(testing "the allowed peer connects and replicates both ways"
(is (true? (<! (db/dial c a-addrs))))
(is (some? (<! (tu/wait-for pa #(and (= :peer-joined (:type %)) (= (db/peer-id c) (:peer %))) 20000))))
(is (some? (<! (tu/wait-for cc #(= "a" (get-in % [:payload :from])) 20000))) "C caught up with A")
(let [rc (<! (db/append c {:from "c"}))]
(is (not (tu/error? rc)) (some-> rc ex-message))
(is (some? (<! (tu/wait-for ca #(= "c" (get-in % [:payload :from])) 20000))) "A received C's record")))
(testing "the refused peer gets no connection"
;; A hangs up right after the noise handshake. libp2p's dialer finishes its
;; side of the upgrade without waiting for the listener, so B's dial may
;; report true for a connection that is already being reset: what counts is
;; that none stays
(let [d (<! (db/dial b a-addrs))]
(is (or (true? d) (err/dial-error? d)) (pr-str d)))
(is (true? (<! (tu/eventually #(not-any? #{(db/peer-id a)} (map :peer ((:connections (:node b))))) 10000)))
"B is left with no connection to A")
(<! (async/timeout 1500))
(is (not-any? #(= (db/peer-id b) (:peer %)) @pa) "A never saw B join")
(is (not-any? #(= (db/peer-id a) (:peer %)) @pb) "…nor B A")
(is (not-any? #{(db/peer-id b)} (map :peer ((:connections (:node a))))) "A holds no connection to B")
(is (not-any? #(= "b" (get-in % [:payload :from])) @ca) "B's record never reached A")
(is (not-any? #(= "a" (get-in % [:payload :from])) @cb) "…nor A's B")
(is (= #{"a" "c"} (set (map #(get-in % [:payload :from]) (<! (db/entries a)))))))
(testing "the gater was asked about inbound connections only, with the caller's peer id"
(is (seq @asked))
(is (every? #(str/starts-with? % "/ip4/127.0.0.1/tcp/") @asked))
(is (= #{(db/peer-id b) (db/peer-id c)} (set (map #(last (str/split % #"/p2p/")) @asked)))))
(testing "outbound is not gated: A dials B and they replicate like anyone else"
(let [n (count @asked)]
(is (true? (<! (db/dial a (<! (db/multiaddrs b))))))
(is (some? (<! (tu/wait-for ca #(= "b" (get-in % [:payload :from])) 20000))) "A received B's record")
(is (some? (<! (tu/wait-for cb #(= "c" (get-in % [:payload :from])) 20000))) "B caught up with A's log")
(is (= n (count @asked)) "the gater was not asked about A's own dial")
;; heads merged in a join are not gossiped on: C, connected to A only, learns
;; of B's record with A's next append, whose links lead to it
(let [ra2 (<! (db/append a {:from "a again"}))]
(is (not (tu/error? ra2)) (some-> ra2 ex-message))
(is (true? (<! (tu/eventually #(= 4 (count @cb) (count @cc)) 20000))) "B and C converge, C through A"))
(let [ea (<! (db/entries a))]
(is (= #{"a" "b" "c" "a again"} (set (map #(get-in % [:payload :from]) ea))))
(is (= ea (<! (db/entries b))))
(is (= ea (<! (db/entries c)))))))
(is (not-any? #(= :sync-error (:type %)) (concat @pa @pb @pc)))
(<! (db/close a))
(<! (db/close b))
(<! (db/close c)))))
(deftest-async a-gater-that-throws-denies
{:timeout-ms 60000}
(if (tu/skip-nodes?)
(is true "IRD_P2P_DB_NO_NODE set — skipping")
(let [secret (db/random-secret)
opts {:secret secret :app-salt "test/v1" :name "gated"}
a (<! (db/open (assoc opts :gater (fn [_] (throw (ex-info "broken gater" {}))))))
_ (is (not (tu/error? a)) (some-> a ex-message))
b (<! (db/open opts))
_ (is (not (tu/error? b)) (some-> b ex-message))]
(<! (db/dial b (<! (db/multiaddrs a))))
(is (true? (<! (tu/eventually #(empty? ((:connections (:node b)))) 10000))) "B is left with no connection")
(is (empty? ((:connections (:node a)))))
(<! (async/timeout 500))
(is (empty? (db/peers a)))
(is (empty? (db/peers b)))
(<! (db/close a))
(<! (db/close b)))))
(deftest-async gater-sees-relayed-connections
{:timeout-ms 90000}
(if (tu/skip-nodes?)
(is true "IRD_P2P_DB_NO_NODE set — skipping")
(let [secret (db/random-secret)
opts {:secret secret :app-salt "test/v1" :name "gated"}
hop (<! (db/open {:secret (db/random-secret) :app-salt "test/v1" :name "relay-only"
:relay-hop? true :listen ["/ip4/127.0.0.1/tcp/0/ws"]}))
_ (is (not (tu/error? hop)) (some-> hop ex-message))
relay-addr (first (<! (db/multiaddrs hop)))
b (<! (db/open (assoc opts :relay relay-addr)))
_ (is (not (tu/error? b)) (some-> b ex-message))
c (<! (db/open (assoc opts :relay relay-addr)))
_ (is (not (tu/error? c)) (some-> c ex-message))
asked (atom [])
b-suffix (str "/p2p/" (db/peer-id b))
a (<! (db/open (assoc opts :relay relay-addr
:gater (fn [addr] (swap! asked conj addr) (not (str/ends-with? addr b-suffix))))))
_ (is (not (tu/error? a)) (some-> a ex-message))
circuit [(str relay-addr "/p2p-circuit/p2p/" (db/peer-id a))]
ca (tu/collector (db/entries-stream a))
pa (tu/collector (db/peers-stream a))
rb (<! (db/append b {:from "b"}))
rc (<! (db/append c {:from "c"}))]
(is (not (tu/error? rb)) (some-> rb ex-message))
(is (not (tu/error? rc)) (some-> rc ex-message))
(<! (db/dial b circuit))
(is (true? (<! (db/dial c circuit))))
(is (some? (<! (tu/wait-for ca #(= "c" (get-in % [:payload :from])) 20000))) "the allowed peer replicates through the relay")
(is (true? (<! (tu/eventually #(not-any? #{(db/peer-id a)} (map :peer ((:connections (:node b))))) 10000)))
"the refused one is left with no connection to A")
(<! (async/timeout 1000))
(is (not-any? #(= (db/peer-id b) (:peer %)) @pa))
(is (= ["c"] (mapv #(get-in % [:payload :from]) @ca)) "B's record never reached A")
(is (= #{(str relay-addr "/p2p-circuit/p2p/" (db/peer-id b)) (str relay-addr "/p2p-circuit/p2p/" (db/peer-id c))}
(set @asked))
"a relayed caller is named by the relay's address, /p2p-circuit and its own peer id")
(<! (db/close a))
(<! (db/close b))
(<! (db/close c))
(<! (db/close hop)))))
|