tunnel_test.go raw
1 package tunnel
2
3 import (
4 "bytes"
5 "context"
6 "crypto/rand"
7 "fmt"
8 "io"
9 "sync"
10 "testing"
11 "time"
12
13 "git.smesh.lol/nostrino/internal/memrelay"
14 "git.smesh.lol/nostrino/pkg/crypto/secp256k1"
15 "git.smesh.lol/nostrino/pkg/nostr"
16 "git.smesh.lol/nostrino/pkg/relay"
17 )
18
19 func keyWithByte(b byte) [32]byte {
20 var k [32]byte
21 k[31] = b
22 return k
23 }
24
25 func testPair(t *testing.T) (*memrelay.Relay, *Listener, [32]byte, DialConfig) {
26 t.Helper()
27 r := memrelay.New()
28 t.Cleanup(r.Close)
29 gwKey := keyWithByte(7)
30 gwPub, ok := secp256k1.PubKey(gwKey)
31 if !ok {
32 t.Fatal("gateway key")
33 }
34 factory := func(string) (relay.Transport, error) { return r.Dial("") }
35 l, err := Listen(context.Background(), ListenConfig{
36 Relays: []string{"mem://relay"},
37 Key: gwKey,
38 Chains: []string{"regtest"},
39 TransportFactory: factory,
40 IdleTimeout: 10 * time.Second,
41 Heartbeat: time.Second,
42 })
43 if err != nil {
44 t.Fatalf("listen: %v", err)
45 }
46 t.Cleanup(func() { l.Close() })
47 u := &URI{
48 Gateway: gwPub,
49 Relays: []string{"mem://relay"},
50 Chains: []string{"regtest"},
51 Version: 1,
52 }
53 return r, l, gwKey, DialConfig{
54 URI: u,
55 Key: keyWithByte(9),
56 Chain: "regtest",
57 TransportFactory: factory,
58 IdleTimeout: 10 * time.Second,
59 Heartbeat: time.Second,
60 }
61 }
62
63 func acceptEcho(t *testing.T, l *Listener) {
64 t.Helper()
65 ctx := context.Background()
66 go func() {
67 for {
68 c, err := l.Accept(ctx)
69 if err != nil {
70 return
71 }
72 go func() {
73 defer c.Close()
74 buf := make([]byte, 32<<10)
75 total := 0
76 for {
77 n, err := c.Read(buf)
78 if n > 0 {
79 total += n
80 }
81 if n > 0 {
82 if _, werr := c.Write(buf[:n]); werr != nil {
83 return
84 }
85 }
86 if err != nil {
87 return
88 }
89 }
90 }()
91 }
92 }()
93 }
94
95 func dialPair(t *testing.T, l *Listener, cfg DialConfig) *Conn {
96 t.Helper()
97 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
98 defer cancel()
99 c, err := Dial(ctx, cfg)
100 if err != nil {
101 t.Fatalf("dial: %v", err)
102 }
103 t.Cleanup(func() { c.Close() })
104 return c
105 }
106
107 func TestTunnelEcho(t *testing.T) {
108 _, l, _, cfg := testPair(t)
109 acceptEcho(t, l)
110 c := dialPair(t, l, cfg)
111 c.SetDeadline(time.Now().Add(20 * time.Second))
112
113 payload := make([]byte, 1<<20)
114 if _, err := rand.Read(payload); err != nil {
115 t.Fatal(err)
116 }
117 var wg sync.WaitGroup
118 wg.Add(1)
119 var werr error
120 go func() {
121 defer wg.Done()
122 _, werr = c.Write(payload)
123 }()
124 got := make([]byte, len(payload))
125 if _, err := io.ReadFull(c, got); err != nil {
126 t.Fatalf("read: %v", err)
127 }
128 wg.Wait()
129 if werr != nil {
130 t.Fatalf("write: %v", werr)
131 }
132 if !bytes.Equal(got, payload) {
133 t.Fatal("echo mismatch")
134 }
135 if c.PeerHex() == "" || c.Chain() != "regtest" {
136 t.Fatalf("peer=%q chain=%q", c.PeerHex(), c.Chain())
137 }
138 }
139
140 func TestTunnelRetransmitWithDrops(t *testing.T) {
141 r, l, _, cfg := testPair(t)
142 acceptEcho(t, l)
143 c := dialPair(t, l, cfg)
144 // Lose roughly one in nine deliveries in both directions. The stream must
145 // still deliver every byte in order.
146 r.SetDrop(9)
147 defer r.SetDrop(0)
148 c.SetDeadline(time.Now().Add(30 * time.Second))
149
150 payload := make([]byte, 512<<10)
151 if _, err := rand.Read(payload); err != nil {
152 t.Fatal(err)
153 }
154 go c.Write(payload)
155 got := make([]byte, len(payload))
156 if _, err := io.ReadFull(c, got); err != nil {
157 t.Fatalf("read: %v", err)
158 }
159 if !bytes.Equal(got, payload) {
160 t.Fatal("echo mismatch after drops")
161 }
162 }
163
164 func TestTunnelBackpressure(t *testing.T) {
165 _, l, _, cfg := testPair(t)
166 // Accept the stream but never read from it: the receiver's queue fills, its
167 // credit reaches zero, and the sender must stop instead of buffering the
168 // whole transfer in memory. This is the frozen-worker case.
169 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
170 defer cancel()
171 accepted := make(chan *Conn, 1)
172 go func() {
173 c, err := l.Accept(ctx)
174 if err == nil {
175 accepted <- c
176 }
177 }()
178 c := dialPair(t, l, cfg)
179 select {
180 case held := <-accepted:
181 defer held.Close()
182 case <-ctx.Done():
183 t.Fatal("accept timed out")
184 }
185
186 payload := make([]byte, 4<<20)
187 done := make(chan error, 1)
188 go func() {
189 _, err := c.Write(payload)
190 done <- err
191 }()
192 select {
193 case err := <-done:
194 t.Fatalf("write completed without a reader (err=%v); flow control failed", err)
195 case <-time.After(500 * time.Millisecond):
196 }
197 c.Reset()
198 select {
199 case <-done:
200 case <-time.After(5 * time.Second):
201 t.Fatal("write did not unblock after reset")
202 }
203 }
204
205 func TestTunnelMultipleClients(t *testing.T) {
206 _, l, _, cfg := testPair(t)
207 acceptEcho(t, l)
208 var wg sync.WaitGroup
209 for i := 0; i < 3; i++ {
210 i := i
211 wg.Add(1)
212 go func() {
213 defer wg.Done()
214 cfg := cfg
215 cfg.Key = keyWithByte(byte(20 + i))
216 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
217 defer cancel()
218 c, err := Dial(ctx, cfg)
219 if err != nil {
220 t.Errorf("dial %d: %v", i, err)
221 return
222 }
223 defer c.Close()
224 c.SetDeadline(time.Now().Add(20 * time.Second))
225 payload := bytes.Repeat([]byte{byte(i)}, 128<<10)
226 go c.Write(payload)
227 got := make([]byte, len(payload))
228 if _, err := io.ReadFull(c, got); err != nil {
229 t.Errorf("client %d read: %v", i, err)
230 return
231 }
232 if !bytes.Equal(got, payload) {
233 t.Errorf("client %d echo mismatch", i)
234 }
235 }()
236 }
237 wg.Wait()
238 }
239
240 func TestTunnelMultipleRelays(t *testing.T) {
241 r1, r2 := memrelay.New(), memrelay.New()
242 t.Cleanup(r1.Close)
243 t.Cleanup(r2.Close)
244 gwKey := keyWithByte(7)
245 gwPub, _ := secp256k1.PubKey(gwKey)
246 factory := func(url string) (relay.Transport, error) {
247 switch url {
248 case "mem://a":
249 return r1.Dial("")
250 case "mem://b":
251 return r2.Dial("")
252 default:
253 return nil, fmt.Errorf("unknown relay %s", url)
254 }
255 }
256 l, err := Listen(context.Background(), ListenConfig{
257 Relays: []string{"mem://a", "mem://b"},
258 Key: gwKey,
259 Chains: []string{"regtest"},
260 TransportFactory: factory,
261 IdleTimeout: 10 * time.Second,
262 Heartbeat: time.Second,
263 })
264 if err != nil {
265 t.Fatal(err)
266 }
267 t.Cleanup(func() { l.Close() })
268 acceptEcho(t, l)
269 u := &URI{Gateway: gwPub, Relays: []string{"mem://a", "mem://b"}, Version: 1}
270 cfg := DialConfig{
271 URI: u,
272 Key: keyWithByte(9),
273 Chain: "regtest",
274 TransportFactory: factory,
275 IdleTimeout: 10 * time.Second,
276 Heartbeat: time.Second,
277 }
278 c := dialPair(t, l, cfg)
279 c.SetDeadline(time.Now().Add(20 * time.Second))
280
281 round := func(size int, tag byte) {
282 payload := bytes.Repeat([]byte{tag}, size)
283 go c.Write(payload)
284 got := make([]byte, len(payload))
285 if _, err := io.ReadFull(c, got); err != nil {
286 t.Fatalf("read: %v", err)
287 }
288 if !bytes.Equal(got, payload) {
289 t.Fatal("echo mismatch")
290 }
291 }
292 round(128<<10, 'a')
293 // Lose every delivery on the first relay; the stream continues on the
294 // second without a new session.
295 r1.SetDrop(1)
296 round(128<<10, 'b')
297 }
298
299 func TestTunnelRejectsChain(t *testing.T) {
300 _, _, _, cfg := testPair(t)
301 cfg.URI.Chains = []string{"btc"}
302 cfg.Chain = "btc"
303 ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
304 defer cancel()
305 if _, err := Dial(ctx, cfg); err == nil {
306 t.Fatal("dial succeeded for an unserved chain")
307 }
308 }
309
310 func TestTunnelListenerAllowlist(t *testing.T) {
311 r := memrelay.New()
312 t.Cleanup(r.Close)
313 gwKey := keyWithByte(7)
314 gwPub, _ := secp256k1.PubKey(gwKey)
315 clientKey := keyWithByte(9)
316 clientPub, _ := secp256k1.PubKey(clientKey)
317 allowed := keyWithByte(33)
318 allowedPub, _ := secp256k1.PubKey(allowed)
319 factory := func(string) (relay.Transport, error) { return r.Dial("") }
320 l, err := Listen(context.Background(), ListenConfig{
321 Relays: []string{"mem://relay"},
322 Key: gwKey,
323 Allowlist: []string{nostr.Hex(allowedPub[:])},
324 TransportFactory: factory,
325 })
326 if err != nil {
327 t.Fatal(err)
328 }
329 defer l.Close()
330 u := &URI{Gateway: gwPub, Relays: []string{"mem://relay"}, Version: 1}
331 dial := func(key [32]byte) error {
332 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
333 defer cancel()
334 c, err := Dial(ctx, DialConfig{URI: u, Key: key, TransportFactory: factory})
335 if err == nil {
336 c.Close()
337 }
338 return err
339 }
340 if err := dial(clientKey); err == nil {
341 t.Fatal("disallowed client connected")
342 }
343 if err := dial(allowed); err != nil {
344 t.Fatalf("allowed client failed: %v", err)
345 }
346 _ = clientPub
347 }
348
349 func TestURIParse(t *testing.T) {
350 key := keyWithByte(7)
351 pub, _ := secp256k1.PubKey(key)
352 hexPub := nostr.Hex(pub[:])
353 raw := "nostr+chaindata://" + hexPub + "?relay=wss%3A%2F%2Frelay.example&relay=ws%3A%2F%2F127.0.0.1%3A3334&bridge=wss%3A%2F%2Fbridge.example%2Fnostrino&chains=btc%2Cxbt&v=1&name=home"
354 u, err := ParseURI(raw)
355 if err != nil {
356 t.Fatalf("parse: %v", err)
357 }
358 if len(u.Relays) != 2 || u.Relays[0] != "wss://relay.example" || u.Relays[1] != "ws://127.0.0.1:3334" {
359 t.Fatalf("relays: %#v", u.Relays)
360 }
361 if len(u.Chains) != 2 || !u.ChainAllowed("xbt") || u.ChainAllowed("testnet") {
362 t.Fatalf("chains: %#v", u.Chains)
363 }
364 if len(u.Bridges) != 1 || u.Bridges[0] != "wss://bridge.example/nostrino" {
365 t.Fatalf("bridges: %#v", u.Bridges)
366 }
367 round, err := ParseURI(u.String())
368 if err != nil {
369 t.Fatalf("round trip: %v", err)
370 }
371 if round.Gateway != u.Gateway || len(round.Relays) != 2 || len(round.Bridges) != 1 || len(round.Chains) != 2 || round.Name != "home" {
372 t.Fatalf("round trip mismatch: %#v", round)
373 }
374 for _, bad := range []string{
375 "http://" + hexPub + "?relay=ws://x",
376 "nostr+chaindata://" + hexPub,
377 "nostr+chaindata://deadbeef?relay=ws://x",
378 "nostr+chaindata://" + hexPub + "?relay=http://x",
379 } {
380 if _, err := ParseURI(bad); err == nil {
381 t.Errorf("accepted %q", bad)
382 }
383 }
384 }
385