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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350 | (ns ird.p2p.room-test
"Rooms over the in-process fake relay and the in-memory peer-connection
factory: the hello/verify handshake, identities, strangers, relayed and
direct messaging both ways, chunked big messages, replay suppression,
the WebRTC negotiation and its ICE-restart ladder."
(:require [cljs.test :refer [deftest is testing]]
[cljs.core.async :as async :refer [<!]]
[ird.p2p :as p2p]
[ird.p2p.async :as a]
[ird.p2p.crypto :as crypto]
[ird.p2p.descriptor :as desc]
[ird.p2p.envelope :as env]
[ird.p2p.errors :as err]
[ird.p2p.fake-relay :as fake]
[ird.p2p.identity :as ident]
[ird.p2p.impl.bytes :as bytes]
[ird.p2p.room :as room]
[ird.p2p.test-util :as tu :refer [deftest-async]]
[ird.p2p.transport :as t]
[ird.p2p.transport.ws :as ws]
[ird.p2p.webrtc :as webrtc]))
(def low "00000000000000000000000000000001")
(def high "0f1e2d3c4b5a69788796a5b4c3d2e1f0")
(def third "ffffffffffffffffffffffffffffff00")
(def app-salt "ird-cljs-tests/room")
(defn harness
"Two apps sharing one fake relay, one in-memory WebRTC network and one test clock."
[]
(let [relay (fake/make-relay)
clock (tu/test-clock)
webrtc (webrtc/in-memory-factory)
secret (crypto/random-secret)
descriptor (desc/parse (tu/fixture "descriptor-full.json"))]
{:relay relay :clock clock :webrtc webrtc :secret secret
:app (fn [& [{:keys [direct?] :or {direct? true}}]]
(p2p/make-app {:host "app.example.test" :app-salt app-salt :origin "https://o.example"
:descriptor descriptor
:transport (ws/transport {:socket-fn (fake/socket-fn relay) :sleep-fn (:sleep-fn clock)
:now-fn (:now-fn clock) :jitter-fn (constantly 0.5)})
:pc-factory (when direct? webrtc)
:http-fn (tu/responding 404 {:error "not_found"})
:sleep-fn (:sleep-fn clock) :now-fn (:now-fn clock)}))}))
(defn join
"[room collector] for PEER in the harness."
[h peer & [{:keys [identity direct? secret] :or {direct? true}}]]
(async/go
(let [room (<! (p2p/join ((:app h) {:direct? direct?}) (or secret (:secret h))
{:identity identity :peer-id peer :options {:pacing? false}}))]
(when (tu/error? room) (throw room))
[room (tu/collector (p2p/events room))])))
(defn status= [st] #(and (= :status (:type %)) (= st (:status %))))
(defn ev= [e] #(= e %))
(defn message-from [from pred] #(and (= :message (:type %)) (= from (:from %)) (pred (:message %))))
(deftest-async two-peers-verify-each-other-and-go-direct
(let [h (harness)
[a ca] (<! (join h low))]
(is (= (:room-id a) (<! (crypto/room-id (:secret h) app-salt))))
(is (some? (<! (tu/wait-for ca (status= {:state :joined})))))
(is (some? (<! (tu/wait-for ca (ev= {:type :roster :peers []})))))
(let [[b cb] (<! (join h high))]
(is (some? (<! (tu/wait-for cb (ev= {:type :roster :peers []})))) "b's roster settles before a is verified")
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-joined :peer high})))))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-ready :peer high :via :relayed})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-joined :peer low})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-ready :peer low :via :relayed})))))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-ready :peer high :via :direct})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-ready :peer low :via :direct})))))
(is (= [high] (p2p/members a)))
(is (= :direct (p2p/transport-of b low)))
(is (not (some #(= :peer-identity (:type %)) @ca)) "no identity announced: no identity events")
(<! (p2p/leave b))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-left :peer high})))))
(is (some? (<! (tu/wait-for cb (status= {:state :closed :reason "left"})))))
(is (nil? (<! (p2p/events b))) "the events channel closes after leaving")
(<! (p2p/leave a)))))
(deftest-async identity-is-verified-and-announced
(let [h (harness)
ida (<! (ident/generate {:name "Ana"}))
[a ca] (<! (join h low {:identity ida :direct? false}))
[b cb] (<! (join h high {:direct? false}))
expected {:public-key (:public-key ida) :fingerprint (:fingerprint ida) :name "Ana"}]
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-identity :peer low :identity expected})))))
(is (= expected (p2p/identity-of b low)))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-joined :peer high})))))
(is (nil? (p2p/identity-of a high)))
(<! (p2p/leave a)) (<! (p2p/leave b))))
(defn- intruder
"A raw relay session in the room (knows the room id, not the secret)."
[h room-id peer]
(t/connect (ws/transport {:socket-fn (fake/socket-fn (:relay h)) :sleep-fn (get-in h [:clock :sleep-fn])
:now-fn (get-in h [:clock :now-fn]) :jitter-fn (constantly 0.5)})
{:url "wss://x/ws" :room room-id :peer peer :limits desc/default-limits :pacing? false}))
(defn- raw-member
"A raw relay session that does know the secret (a hand-driven peer, like
another SDK's app): [session room-crypto identity]."
[h room-id peer]
(async/go
(let [s (<! (intruder h room-id peer))
rc (<! (crypto/room-crypto {:secret (:secret h) :app-salt app-salt :peer-id peer}))
id (<! (ident/generate))]
[s rc id])))
(defn- raw-hello!
"Seal a hand-built hello carrying ID-BLOCK verbatim and send it from S (under RC) to TO."
[s rc id-block to]
(async/go
(<! (t/send! s {:t :signal :to to :payload (<! (env/seal-signal rc {:kind "hello" :id id-block} to))}))))
(defn- mentions-hue-or-glyph? [v]
(boolean (some #(and (map? %) (or (contains? % :hue) (contains? % :glyph))) (tree-seq coll? seq v))))
(deftest-async hellos-announcing-hue-and-glyph-interoperate-and-surface-only-name
;; Swift SDK apps announce {pub sig name hue glyph}; the signature covers only
;; sueta-id|v2|salt|room|peer, so such a hello must verify and join, and the
;; unsigned display opinions must be dropped on receipt, never surfaced.
(let [h (harness)
[a ca] (<! (join h low {:direct? false}))
_ (is (some? (<! (tu/wait-for ca (ev= {:type :roster :peers []})))))
room-id (:room-id a)
[s1 rc1 id1] (<! (raw-member h room-id high))
[s2 rc2 id2] (<! (raw-member h room-id third))]
(testing "a Swift-style block: name + hue + glyph over a valid signature"
(let [sig (<! (ident/assert-session id1 app-salt room-id high))
expected {:public-key (:public-key id1) :fingerprint (:fingerprint id1) :name "Ana"}]
(<! (raw-hello! s1 rc1 {:pub (:public-key id1) :sig sig :name "Ana" :hue 200 :glyph "🐢"} low))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-joined :peer high})))) "the hello verified and proved membership")
(is (= {:type :peer-identity :peer high :identity expected}
(<! (tu/wait-for ca #(and (= :peer-identity (:type %)) (= high (:peer %)))))))
(is (= expected (p2p/identity-of a high)))
(is (= #{:public-key :fingerprint :name} (set (keys (p2p/identity-of a high)))))))
(testing "a nameless block: pub and sig only"
(let [sig (<! (ident/assert-session id2 app-salt room-id third))
expected {:public-key (:public-key id2) :fingerprint (:fingerprint id2) :name nil}]
(<! (raw-hello! s2 rc2 {:pub (:public-key id2) :sig sig} low))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-joined :peer third})))))
(is (= {:type :peer-identity :peer third :identity expected}
(<! (tu/wait-for ca #(and (= :peer-identity (:type %)) (= third (:peer %)))))))
(is (= expected (p2p/identity-of a third)))))
(testing "a block whose signature does not verify announces nothing, hue/glyph or not"
(let [bad (<! (ident/assert-session id1 "other/salt" room-id high))]
(<! (raw-hello! s1 rc1 {:pub (:public-key id1) :sig bad :name "Eve" :hue 1 :glyph "🦊"} low))
(<! (async/timeout 30))
(is (= "Ana" (:name (p2p/identity-of a high))))))
(is (= [high third] (p2p/members a)))
(is (not (some mentions-hue-or-glyph? @ca)) "no :hue/:glyph key anywhere in the event stream")
(<! (t/close! s1)) (<! (t/close! s2)) (<! (p2p/leave a))))
(deftest-async strangers-in-the-room-are-ignored
(let [h (harness)
[a ca] (<! (join h low {:direct? false}))]
(is (some? (<! (tu/wait-for ca (ev= {:type :roster :peers []})))))
(let [s (<! (intruder h (:room-id a) third))]
(<! (t/send! s {:t :signal :to low :payload {:v 1 :iv "AAAAAAAAAAAAAAAA" :ct "AAAAAAAAAAAAAAAAAAAAAAAAAAAA"}}))
(<! (t/send! s {:t :signal :to low :payload "junk"}))
(let [[b cb] (<! (join h high {:direct? false}))]
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-joined :peer low})))))
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-joined :peer high})))))
(is (= [high] (p2p/members a)))
(is (not (some (ev= {:type :peer-joined :peer third}) @ca)))
(<! (t/close! s))
(<! (async/timeout 20))
(is (not (some (ev= {:type :peer-left :peer third}) @ca)))
(<! (p2p/leave a)) (<! (p2p/leave b))))))
(deftest-async wrong-secret-never-meets
(let [h (harness)
[a ca] (<! (join h low {:direct? false}))
[b cb] (<! (join h high {:direct? false :secret (crypto/random-secret)}))]
(is (some? (<! (tu/wait-for ca (ev= {:type :roster :peers []})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :roster :peers []})))))
(is (not= (:room-id a) (:room-id b)))
(<! (p2p/leave a)) (<! (p2p/leave b))))
(deftest-async silent-peer-becomes-a-stranger-after-hello-timeout
(let [h (harness)
[a ca] (<! (join h low {:direct? false}))]
(is (some? (<! (tu/wait-for ca (ev= {:type :roster :peers []})))))
(let [s (<! (intruder h (:room-id a) high))]
(is (true? (<! (tu/eventually #(>= (count (fake/signal-frames (:relay h))) 1)))) "a sent its hello")
((get-in h [:clock :advance!]) 11000)
(<! (async/timeout 30))
(is (empty? (p2p/members a)))
(is (not (some (ev= {:type :peer-joined :peer high}) @ca)))
(<! (t/close! s)) (<! (p2p/leave a)))))
(defn- pair [h direct?]
(async/go
(let [[a ca] (<! (join h low {:direct? direct?}))
[b cb] (<! (join h high {:direct? direct?}))
want (if direct? :direct :relayed)]
(is (true? (<! (tu/eventually #(= want (p2p/transport-of a high))))))
(is (true? (<! (tu/eventually #(= want (p2p/transport-of b low))))))
[a ca b cb])))
(deftest-async json-and-binary-both-ways
(doseq [direct? [false true]]
(let [h (harness)
[a ca b cb] (<! (pair h direct?))]
(is (true? (<! (p2p/send a high (p2p/json {:text "salut" :n 1})))))
(is (true? (<! (p2p/send b low (p2p/data "AAEC/w==")))))
(is (true? (<! (p2p/send b low (p2p/json {:t "ack"})))))
(is (some? (<! (tu/wait-for cb (message-from low #(= {:kind :json :value {:text "salut" :n 1}} %))))))
(is (some? (<! (tu/wait-for ca (message-from high #(= {:kind :data :payload "AAEC/w=="} %))))))
(is (some? (<! (tu/wait-for ca (message-from high #(= {:kind :json :value {:t "ack"}} %))))))
(<! (p2p/leave a)) (<! (p2p/leave b)))))
(deftest-async large-messages-are-chunked
{:timeout-ms 30000}
(doseq [direct? [false true]]
(let [h (harness)
[a ca b cb] (<! (pair h direct?))
big-text (apply str (repeat 100000 "ă"))
big-data (bytes/bytes->b64 (map #(mod % 251) (range 100000)))]
(is (true? (<! (p2p/send a high (p2p/json {:text big-text})))))
(is (true? (<! (p2p/send b low (p2p/data big-data)))))
(is (some? (<! (tu/wait-for cb (message-from low #(= big-text (get-in % [:value :text]))) 10000))))
(is (some? (<! (tu/wait-for ca (message-from high #(= big-data (:payload %))) 10000))))
(when-not direct?
(is (every? #(<= (bytes/utf8-byte-length (:text %)) 65536) @(:frames (:relay h))) "every relayed frame stayed under the cap"))
(<! (p2p/leave a)) (<! (p2p/leave b)))))
(deftest-async broadcast-reaches-every-verified-peer
(let [h (harness)
[a _] (<! (join h low {:direct? false}))
[b cb] (<! (join h high {:direct? false}))
[c cc] (<! (join h third {:direct? false}))]
(is (true? (<! (tu/eventually #(= [high third] (p2p/members a))))))
(is (true? (<! (p2p/broadcast a (p2p/json "hi all")))))
(is (some? (<! (tu/wait-for cb (message-from low #(= "hi all" (:value %)))))))
(is (some? (<! (tu/wait-for cc (message-from low #(= "hi all" (:value %)))))))
(<! (p2p/leave a)) (<! (p2p/leave b)) (<! (p2p/leave c))))
(deftest-async replayed-frames-are-dropped-and-unknown-peers-refused
(let [h (harness)
_ (reset! (:duplicate-signals? (:relay h)) true)
[a ca b cb] (<! (pair h false))]
(is (true? (<! (p2p/send a high (p2p/json "once")))))
(is (some? (<! (tu/wait-for cb (message-from low #(= "once" (:value %)))))))
(<! (async/timeout 50))
(is (= 1 (tu/count-of cb (message-from low #(= "once" (:value %))))))
(let [e (<! (p2p/send a third (p2p/json nil)))]
(is (err/peer-unknown? e))
(is (= third (:peer (ex-data e)))))
(is (err/config-error? (<! (p2p/send a high {:kind :data :payload "***"}))))
(<! (p2p/leave a)) (<! (p2p/leave b))
(is (err/closed? (<! (p2p/send a high (p2p/json nil)))))))
(deftest-async peer-loss-while-direct-keeps-the-data-channel
(let [h (harness)
{:keys [relay clock]} h
[a ca b cb] (<! (pair h true))]
;; b's relay socket dies but the data channel is up: no peer-left, messages still flow
(fake/drop-socket! relay high)
(is (some? (<! (tu/wait-for cb #(and (= :status (:type %)) (= :reconnecting (get-in % [:status :state])))))))
(is (true? (<! (p2p/send a high (p2p/json "still here")))))
(is (some? (<! (tu/wait-for cb (message-from low #(= "still here" (:value %)))))))
(is (not (some (ev= {:type :peer-left :peer high}) @ca)))
;; b reconnects (same peer id) after the backoff
(is (true? (<! (tu/eventually #(>= ((:sleepers clock)) 1)))))
((:advance! clock) 15000)
(is (true? (<! (tu/eventually #(>= (tu/count-of cb (status= {:state :joined})) 2)))))
(is (true? (<! (tu/eventually #(= [high] (p2p/members a))))))
(<! (p2p/leave a)) (<! (p2p/leave b))))
(deftest-async lower-peer-offers-and-both-go-direct
(let [h (harness)
[a ca] (<! (join h high))
[b cb] (<! (join h low))]
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-ready :peer low :via :direct})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-ready :peer high :via :direct})))))
(is (seq (filter #(= low (:from %)) (fake/signal-frames (:relay h)))) "the offer came from the lower id")
(<! (p2p/leave a)) (<! (p2p/leave b))))
(deftest-async data-channel-loss-falls-back-then-restarts-via-the-ladder
(let [h (harness)
{:keys [webrtc clock]} h
[a ca] (<! (join h low))
[b cb] (<! (join h high))]
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-ready :peer high :via :direct})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-ready :peer low :via :direct})))))
(is (= 1 (webrtc/connection-attempts webrtc)))
(webrtc/fail-all! webrtc)
(is (true? (<! (tu/eventually #(>= (tu/count-of ca (ev= {:type :peer-ready :peer high :via :relayed})) 2)))))
(is (true? (<! (tu/eventually #(>= (tu/count-of cb (ev= {:type :peer-ready :peer low :via :relayed})) 2)))))
;; relayed messaging works meanwhile
(is (true? (<! (p2p/send a high (p2p/json "via relay")))))
(is (some? (<! (tu/wait-for cb (message-from low #(= "via relay" (:value %)))))))
;; the offerer's 2 s restart sleeper joins the keepalive sleepers
(is (true? (<! (tu/eventually #(>= ((:sleepers clock)) 3)))))
((:advance! clock) 2000)
(is (true? (<! (tu/eventually #(= 2 (webrtc/connection-attempts webrtc))))))
(is (true? (<! (tu/eventually #(>= (tu/count-of ca (ev= {:type :peer-ready :peer high :via :direct})) 2)))))
(is (true? (<! (tu/eventually #(>= (tu/count-of cb (ev= {:type :peer-ready :peer low :via :direct})) 2)))))
(is (true? (<! (p2p/send b low (p2p/json "direct again")))))
(is (some? (<! (tu/wait-for ca (message-from high #(= "direct again" (:value %)))))))
(<! (p2p/leave a)) (<! (p2p/leave b))))
(deftest-async failed-negotiations-walk-the-ladder-then-wait-for-wake
{:timeout-ms 20000}
(let [h (harness)
{:keys [webrtc clock]} h
_ (webrtc/fail-next! webrtc 10)
[a ca] (<! (join h low))
[b cb] (<! (join h high))]
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-ready :peer high :via :relayed})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-ready :peer low :via :relayed})))))
(is (true? (<! (tu/eventually #(= 1 (webrtc/connection-attempts webrtc))))) "the first negotiation failed")
(loop [expected 1 [delay & more] room/ice-restart-ladder-ms]
(when delay
(is (true? (<! (tu/eventually #(>= ((:sleepers clock)) 3)))) "two keepalives + the restart")
((:advance! clock) delay)
(let [want (inc expected)]
(is (true? (<! (tu/eventually #(= want (webrtc/connection-attempts webrtc))))) (str "ladder step " delay))
(recur want more))))
(is (not (some (ev= {:type :peer-ready :peer high :via :direct}) @ca)))
;; out of ladder: nothing more is scheduled until wake
((:advance! clock) 60000)
(<! (async/timeout 50))
(is (= 4 (webrtc/connection-attempts webrtc)))
(webrtc/fail-next! webrtc -10)
(<! (p2p/wake a))
(is (true? (<! (tu/eventually #(and (= :direct (p2p/transport-of a high)) (= :direct (p2p/transport-of b low)))))))
(<! (p2p/leave a)) (<! (p2p/leave b))))
(deftest-async relay-only-rooms-never-negotiate
(let [h (harness)
[a ca] (<! (join h low {:direct? false}))
[b cb] (<! (join h high {:direct? false}))]
(is (some? (<! (tu/wait-for ca (ev= {:type :peer-ready :peer high :via :relayed})))))
(is (some? (<! (tu/wait-for cb (ev= {:type :peer-ready :peer low :via :relayed})))))
(<! (async/timeout 50))
(is (not (some (ev= {:type :peer-ready :peer high :via :direct}) @ca)))
(is (= :relayed (p2p/transport-of a high)))
(<! (p2p/leave a)) (<! (p2p/leave b))))
(deftest-async relay-errors-are-events-and-the-room-stays-joined
(let [h (harness)
[a ca] (<! (join h low {:direct? false}))]
(<! (t/send! (:session a) {:t :bogus}))
(is (some? (<! (tu/wait-for ca (ev= {:type :relay-error :code "bad-message"})))))
(is (= {:state :joined} (room/status a)))
(<! (p2p/leave a))))
|