server_subs_test.mx raw
1 package server
2
3 import (
4 "bytes"
5 "testing"
6 "time"
7
8 "git.smesh.lol/smesh/pkg/broadcast"
9 "git.smesh.lol/smesh/pkg/nostr/envelope"
10 "git.smesh.lol/smesh/pkg/nostr/event"
11 "git.smesh.lol/smesh/pkg/nostr/signer/p8k"
12 "git.smesh.lol/smesh/pkg/nostr/tag"
13 "git.smesh.lol/smesh/pkg/relay/config"
14 )
15
16 // stSubSrv builds a server with a live broadcast worker (its own spawn domain)
17 // and a fake database. Callers must close the broadcaster.
18 func stSubSrv(c *config.C) (s *Server, b *broadcast.Broadcaster) {
19 s = stServer(c)
20 s.db = stFakeDB()
21 b = broadcast.New()
22 s.bcast = b
23 return
24 }
25
26 // stAuthEvent builds and signs a NIP-42 AUTH event. A nil tag slice omits that
27 // tag, so the missing-tag failure paths are reachable.
28 func stAuthEvent(t *testing.T, sg *p8k.Signer, k uint16, ch, rl []byte) (ev *event.E) {
29 t.Helper()
30 ev = event.New()
31 ev.CreatedAt = time.Now().Unix()
32 ev.Kind = k
33 var chT *tag.T
34 var rlT *tag.T
35 if ch != nil {
36 chT = tag.NewFromBytesSlice([]byte("challenge"), ch)
37 }
38 if rl != nil {
39 rlT = tag.NewFromBytesSlice([]byte("relay"), rl)
40 }
41 if chT != nil && rlT != nil {
42 ev.Tags = tag.NewS(chT, rlT)
43 } else if chT != nil {
44 ev.Tags = tag.NewS(chT)
45 } else if rlT != nil {
46 ev.Tags = tag.NewS(rlT)
47 }
48 if serr := ev.Sign(sg); serr != nil {
49 t.Fatal(serr)
50 return
51 }
52 return
53 }
54
55 func stAuthMsg(ev *event.E) (msg []byte) {
56 ar := &envelope.AuthResponse{Event: ev}
57 return ar.Marshal(nil)
58 }
59
60 func stNewSigner(t *testing.T) (sg *p8k.Signer) {
61 t.Helper()
62 sg = p8k.MustNew()
63 if gerr := sg.Generate(); gerr != nil {
64 t.Fatal(gerr)
65 return
66 }
67 return
68 }
69
70 // --- subscription registry ---
71
72 func TestHandleReqRegistersSubscription(t *testing.T) {
73 s, b := stSubSrv(stCfg())
74 defer b.Close()
75 fd := int32(4)
76 s.conns[fd] = &cstate{subs: map[string]*sub{}}
77
78 req := []byte("[\"REQ\",\"s1\",{}]")
79 s.handleReq(fd, req)
80 cs := s.conns[fd]
81 if len(cs.subs) != 1 {
82 t.Fatalf("subs = %d, want 1", len(cs.subs))
83 return
84 }
85 sb := cs.subs["s1"]
86 if sb == nil {
87 t.Fatal("subscription s1 must be registered")
88 return
89 }
90 if sb.id != "s1" {
91 t.Fatalf("subscription id = %s", sb.id)
92 return
93 }
94 if !bytes.Equal(sb.rawReq, req) {
95 t.Fatal("subscription must retain the raw REQ frame")
96 return
97 }
98 if len(s.dbPending) != 1 {
99 t.Fatalf("dbPending = %d, want a history backfill", len(s.dbPending))
100 return
101 }
102 }
103
104 func TestHandleReqSubscriptionCap(t *testing.T) {
105 c := stCfg()
106 c.MaxSubscriptions = 1
107 s, b := stSubSrv(c)
108 defer b.Close()
109 fd := int32(4)
110 s.conns[fd] = &cstate{subs: map[string]*sub{}}
111
112 s.handleReq(fd, []byte("[\"REQ\",\"s1\",{}]"))
113 s.handleReq(fd, []byte("[\"REQ\",\"s2\",{}]"))
114 if len(s.conns[fd].subs) != 1 {
115 t.Fatalf("cap must hold at 1, got %d", len(s.conns[fd].subs))
116 return
117 }
118 if _, has := s.conns[fd].subs["s2"]; has {
119 t.Fatal("s2 must have been refused")
120 return
121 }
122 // Re-subscribing an existing id is a replace, not a second slot.
123 s.handleReq(fd, []byte("[\"REQ\",\"s1\",{}]"))
124 if len(s.conns[fd].subs) != 1 {
125 t.Fatal("replacing an existing subscription must not grow the set")
126 return
127 }
128 }
129
130 func TestHandleReqAuthRequired(t *testing.T) {
131 c := stCfg()
132 c.AuthRequired = true
133 s, b := stSubSrv(c)
134 defer b.Close()
135 fd := int32(4)
136 cs := &cstate{subs: map[string]*sub{}}
137 s.conns[fd] = cs
138
139 s.handleReq(fd, []byte("[\"REQ\",\"s1\",{}]"))
140 if len(cs.subs) != 0 {
141 t.Fatal("an unauthed REQ must be refused under AuthRequired")
142 return
143 }
144
145 cs.authedPubkey = []byte{0xAB}
146 s.handleReq(fd, []byte("[\"REQ\",\"s1\",{}]"))
147 if len(cs.subs) != 1 {
148 t.Fatal("an authed REQ must be accepted")
149 return
150 }
151 }
152
153 func TestHandleReqMalformedAndTainted(t *testing.T) {
154 s, b := stSubSrv(stCfg())
155 defer b.Close()
156 fd := int32(4)
157 cs := &cstate{subs: map[string]*sub{}}
158 s.conns[fd] = cs
159
160 s.handleReq(99, []byte("[\"REQ\",\"s1\",{}]"))
161
162 s.handleReq(fd, []byte("[\"REQ\",]"))
163 if len(cs.subs) != 0 {
164 t.Fatal("a malformed REQ must not register")
165 return
166 }
167
168 // An undersized hex id taints the filter and jails the connection.
169 s.handleReq(fd, []byte("[\"REQ\",\"s2\",{\"ids\":[\"abcd\"]}]"))
170 if !cs.jailed {
171 t.Fatal("an undersized hex id must jail the connection")
172 return
173 }
174 if len(cs.subs) != 0 {
175 t.Fatal("a tainted REQ must not register")
176 return
177 }
178 }
179
180 func TestHandleClose(t *testing.T) {
181 s, b := stSubSrv(stCfg())
182 defer b.Close()
183 fd := int32(4)
184 cs := &cstate{subs: map[string]*sub{}}
185 s.conns[fd] = cs
186
187 s.handleReq(fd, []byte("[\"REQ\",\"s1\",{}]"))
188 if len(cs.subs) != 1 {
189 t.Fatal("setup: s1 must be registered")
190 return
191 }
192 s.handleClose(fd, []byte("[\"CLOSE\",\"s1\"]"))
193 if len(cs.subs) != 0 {
194 t.Fatal("CLOSE must remove the subscription")
195 return
196 }
197 // Unknown connection and malformed CLOSE are no-ops.
198 s.handleClose(99, []byte("[\"CLOSE\",\"s1\"]"))
199 s.handleClose(fd, []byte("[\"CLOSE\"]"))
200 }
201
202 func TestOnWSClose(t *testing.T) {
203 s, b := stSubSrv(stCfg())
204 defer b.Close()
205 fd := int32(4)
206 cs := &cstate{subs: map[string]*sub{}}
207 cs.subs["s1"] = &sub{id: "s1"}
208 cs.challenge = []byte("c")
209 cs.authedPubkey = []byte("p")
210 s.conns[fd] = cs
211 s.pendingReq[1] = fd
212 s.pendingReq[2] = 99
213 s.dbPending[3] = fd
214 s.dbPending[4] = 99
215
216 s.OnWSClose(fd)
217
218 if s.conns[fd] != nil {
219 t.Fatal("OnWSClose must drop the connection record")
220 return
221 }
222 if _, ok := s.pendingReq[1]; ok {
223 t.Fatal("pendingReq entries for the closed fd must be dropped")
224 return
225 }
226 if _, ok := s.pendingReq[2]; !ok {
227 t.Fatal("pendingReq entries for other fds must survive")
228 return
229 }
230 if _, ok := s.dbPending[3]; ok {
231 t.Fatal("dbPending entries for the closed fd must be dropped")
232 return
233 }
234 if _, ok := s.dbPending[4]; !ok {
235 t.Fatal("dbPending entries for other fds must survive")
236 return
237 }
238 }
239
240 // --- NIP-42 auth ---
241
242 func TestHandleAuthSuccess(t *testing.T) {
243 c := stCfg()
244 c.RelayURL = "wss://relay.example.com"
245 s, b := stSubSrv(c)
246 defer b.Close()
247 fd := int32(4)
248 cs := &cstate{subs: map[string]*sub{}}
249 cs.challenge = []byte("chal-123")
250 s.conns[fd] = cs
251 sg := stNewSigner(t)
252
253 ev := stAuthEvent(t, sg, 22242, []byte("chal-123"), []byte("relay.example.com"))
254 s.handleAuth(fd, stAuthMsg(ev))
255
256 if len(cs.authedPubkey) != 32 {
257 t.Fatalf("authedPubkey length = %d, want 32", len(cs.authedPubkey))
258 return
259 }
260 if !bytes.Equal(cs.authedPubkey, ev.Pubkey) {
261 t.Fatal("authedPubkey must be the auth event's pubkey")
262 return
263 }
264 }
265
266 func TestHandleAuthFailures(t *testing.T) {
267 c := stCfg()
268 c.RelayURL = "wss://relay.example.com"
269 s := stServer(c)
270 fd := int32(4)
271 cs := &cstate{subs: map[string]*sub{}}
272 cs.challenge = []byte("chal-123")
273 s.conns[fd] = cs
274 sg := stNewSigner(t)
275
276 // No connection, then an unparseable AUTH frame.
277 s.handleAuth(99, []byte("[\"AUTH\",]"))
278 s.handleAuth(fd, []byte("[\"AUTH\",]"))
279
280 // Wrong event kind.
281 s.handleAuth(fd, stAuthMsg(stAuthEvent(t, sg, 1, []byte("chal-123"), []byte("relay.example.com"))))
282 // Missing challenge tag.
283 s.handleAuth(fd, stAuthMsg(stAuthEvent(t, sg, 22242, nil, []byte("relay.example.com"))))
284 // Wrong challenge.
285 s.handleAuth(fd, stAuthMsg(stAuthEvent(t, sg, 22242, []byte("nope"), []byte("relay.example.com"))))
286 // Missing relay tag.
287 s.handleAuth(fd, stAuthMsg(stAuthEvent(t, sg, 22242, []byte("chal-123"), nil)))
288 // Relay URL mismatch.
289 s.handleAuth(fd, stAuthMsg(stAuthEvent(t, sg, 22242, []byte("chal-123"), []byte("other.example.com"))))
290
291 // Timestamp out of range.
292 old := stAuthEvent(t, sg, 22242, []byte("chal-123"), []byte("relay.example.com"))
293 old.CreatedAt = time.Now().Unix() - 1000
294 if serr := old.Sign(sg); serr != nil {
295 t.Fatal(serr)
296 return
297 }
298 s.handleAuth(fd, stAuthMsg(old))
299
300 // Invalid signature.
301 bad := stAuthEvent(t, sg, 22242, []byte("chal-123"), []byte("relay.example.com"))
302 bad.Sig[0] ^= 0xFF
303 s.handleAuth(fd, stAuthMsg(bad))
304
305 if len(cs.authedPubkey) != 0 {
306 t.Fatal("no failure path may authenticate the connection")
307 return
308 }
309 }
310
311 // --- broadcast domain ---
312
313 func TestHandleBroadcastFrameNoFrames(t *testing.T) {
314 s, b := stSubSrv(stCfg())
315 defer b.Close()
316 s.handleBroadcastFrame()
317 if !b.Valid() {
318 t.Fatal("the broadcast worker must stay alive")
319 return
320 }
321 }
322
323 func TestSendBroadcastFlags(t *testing.T) {
324 c := stCfg()
325 c.RelayURL = "wss://relay.example"
326 c.PrivilegedOpen = false
327 c.NIP70Enforce = true
328 c.MarmotOpen = true
329 s, b := stSubSrv(c)
330 defer b.Close()
331 s.sendBroadcast([]byte("[\"EVENT\",{}]"), 3)
332 }
333
334 // --- worker lifecycle ---
335
336 func TestWorkerStartupAndRespawn(t *testing.T) {
337 s := stServer(stCfg())
338
339 s.startIngestWorkers(1)
340 if s.workers.Len() != 1 || len(s.workerIn) != 1 {
341 t.Fatal("ingest pool must be started")
342 return
343 }
344 if !workerAlive(s.workerDone[0]) {
345 t.Fatal("ingest worker must be alive")
346 return
347 }
348 oldIngest := s.workerIn[0]
349 s.workers.Busy[0] = true
350 s.respawnIngestWorker(0)
351 if s.workers.Busy[0] {
352 t.Fatal("ingest respawn must clear the busy flag")
353 return
354 }
355 if !workerAlive(s.workerDone[0]) {
356 t.Fatal("respawned ingest worker must be alive")
357 return
358 }
359 close(oldIngest)
360 close(s.workerIn[0])
361
362 s.startProxyWorkers(1)
363 if s.proxyPool.Len() != 1 || len(s.proxyBusyTime) != 1 {
364 t.Fatal("proxy pool must be started")
365 return
366 }
367 oldProxy := s.proxyIn[0]
368 s.proxyPool.Busy[0] = true
369 s.proxyBusyTime[0] = 123
370 s.respawnProxyWorker(0)
371 if s.proxyPool.Busy[0] || s.proxyBusyTime[0] != 0 {
372 t.Fatal("proxy respawn must reset busy state")
373 return
374 }
375 if !workerAlive(s.proxyDone[0]) {
376 t.Fatal("respawned proxy worker must be alive")
377 return
378 }
379 close(oldProxy)
380 close(s.proxyIn[0])
381
382 s.startBlossomWorkers(1)
383 if s.blossomPool.Len() != 1 {
384 t.Fatal("blossom pool must be started")
385 return
386 }
387 oldBlossom := s.blossomIn[0]
388 s.blossomPool.Busy[0] = true
389 s.respawnBlossomWorker(0)
390 if s.blossomPool.Busy[0] {
391 t.Fatal("blossom respawn must clear the busy flag")
392 return
393 }
394 if !workerAlive(s.blossomDone[0]) {
395 t.Fatal("respawned blossom worker must be alive")
396 return
397 }
398 close(oldBlossom)
399 close(s.blossomIn[0])
400 }
401
402 func TestServerCloseIdempotent(t *testing.T) {
403 c := stCfg()
404 s, b := stSubSrv(c)
405 if b == nil {
406 t.Fatal("setup: broadcaster must exist")
407 return
408 }
409 s.Close()
410 s.Close()
411 if !s.closed {
412 t.Fatal("Close must set the closed flag")
413 return
414 }
415 // Close's effect on the broadcast worker is not asserted: in this build a
416 // spawn channel's close() is not observed by the child and the spawn done
417 // channel never signals the child's exit, so Valid() stays true forever.
418 // Reported; the closed flag (idempotence) is what Close can be held to.
419
420 // A server with no database still closes cleanly and idempotently.
421 s2 := stServer(stCfg())
422 s2.bcast = broadcast.New()
423 s2.Close()
424 s2.Close()
425 if !s2.closed {
426 t.Fatal("second server Close must set the closed flag")
427 return
428 }
429 }
430