ALL

For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

tunnel/splice_test.go

SHA-256: 9e9c1d6472cf16ffb20530fe43ac426cbfcbadf12a273caa3d48a375ed6726a0
577 lines · 17475 bytes · View on GitHub

1package main
2
3import (
4 "bufio"
5 "bytes"
6 "crypto/rand"
7 "crypto/sha256"
8 "crypto/tls"
9 "fmt"
10 "io"
11 "net"
12 "net/http"
13 "strings"
14 "sync"
15 "testing"
16 "time"
17)
18
19// ------------------------------------------------------------------------------------------------------------------------------------
20// Blindness: the node forwards ciphertext and never the plaintext.
21// ------------------------------------------------------------------------------------------------------------------------------------
22
23func TestBlindSplice_NodeNeverSeesPlaintext(t *testing.T) {
24 e := newTestEnv(t, nil)
25 d := e.newDevice()
26 dc := e.connectOK(d)
27 ds := dc.serve(nil)
28
29 c, err := e.clientTLS(d)
30 if err != nil {
31 t.Fatal(err)
32 }
33 defer c.Close()
34 clientAddr := c.LocalAddr().String()
35
36 // The client verified the DEVICE's certificate (issued by the device's CA for the device's name). It is not the node's own certificate.
37 st := c.ConnectionState()
38 if fingerprintOf(st.PeerCertificates[0].Raw) != fingerprintOf(d.cert.Certificate[0]) {
39 t.Fatal("client did not reach the device's TLS server")
40 }
41 if st.NegotiatedProtocol != "http/1.1" {
42 t.Fatalf("ALPN = %q, want http/1.1", st.NegotiatedProtocol)
43 }
44
45 body := []byte(strings.Repeat("A", 100) + marker + strings.Repeat("B", 5000))
46 resp, err := postOver(c, d.host, body)
47 if err != nil {
48 t.Fatal(err)
49 }
50 // Teeth: the endpoints really exchanged the marker in the clear.
51 if !bytes.Contains(resp, []byte(marker)) {
52 t.Fatal("client did not read the marker back from the device")
53 }
54 _ = c.Close()
55 e.waitIdle()
56
57 seen := ds.seen()
58 if len(seen) != 1 || !bytes.Contains(seen[0].body, []byte(marker)) {
59 t.Fatalf("the device did not receive the marker in plaintext: %+v", seen)
60 }
61 opens := ds.openList()
62 if len(opens) != 1 {
63 t.Fatalf("expected exactly one stream, got %v", opens)
64 }
65
66 // What the node could see: every byte it forwarded (both directions, including the peeked ClientHello) and every log line.
67 tapped := e.tapped(opens[0].str("stream_id"))
68 if len(tapped) < len(body)+len(resp) {
69 t.Fatalf("tap saw only %d bytes; the stream carried at least %d", len(tapped), len(body)+len(resp))
70 }
71 for name, hay := range map[string][]byte{
72 "forwarded bytes (this stream)": tapped,
73 "forwarded bytes (all)": e.tappedAll(),
74 "logs": []byte(e.logs.String()),
75 } {
76 if bytes.Contains(hay, []byte(marker)) {
77 t.Fatalf("node-visible %s contain the plaintext marker", name)
78 }
79 if bytes.Contains(hay, []byte(strings.Repeat("A", 100))) || bytes.Contains(hay, []byte(strings.Repeat("B", 100))) {
80 t.Fatalf("node-visible %s contain plaintext filler", name)
81 }
82 }
83 // What it saw is TLS: a ClientHello first, then encrypted application data.
84 if !bytes.HasPrefix(tapped, []byte{0x16, 0x03}) {
85 t.Fatalf("first forwarded bytes are not a TLS handshake record: % x", tapped[:8])
86 }
87 if !bytes.Contains(tapped, []byte{0x17, 0x03, 0x03}) {
88 t.Fatal("no TLS application-data record seen")
89 }
90 // Logs carry only metadata (sizes, ids, times, reasons): no name, no address, no identity of a device.
91 for _, want := range []string{`"stream_open"`, `"stream_close"`, `"up_bytes"`, `"down_bytes"`} {
92 if !strings.Contains(e.logs.String(), want) {
93 t.Fatalf("logs lack %s:\n%s", want, e.logs.String())
94 }
95 }
96 for name, secret := range map[string]string{"host": d.host, "label": d.label, "device id": d.id, "client address": clientAddr, "session": dc.session} {
97 if strings.Contains(e.logs.String(), secret) {
98 t.Fatalf("the logs name the %s", name)
99 }
100 }
101 e.checkTranscript()
102}
103
104// ------------------------------------------------------------------------------------------------------------------------------------
105// Splice correctness
106// ------------------------------------------------------------------------------------------------------------------------------------
107
108func TestSplice_LargeBodiesBothWays(t *testing.T) {
109 e := newTestEnv(t, nil)
110 d := e.newDevice()
111 dc := e.connectOK(d)
112 dc.serve(nil)
113 body := make([]byte, 6<<20)
114 if _, err := rand.Read(body); err != nil {
115 t.Fatal(err)
116 }
117 c, err := e.clientTLS(d)
118 if err != nil {
119 t.Fatal(err)
120 }
121 defer c.Close()
122 resp, err := postOver(c, d.host, body)
123 if err != nil {
124 t.Fatal(err)
125 }
126 if len(resp) != len(body)+5 || sha256.Sum256(resp[5:]) != sha256.Sum256(body) {
127 t.Fatalf("body corrupted or truncated: got %d bytes, want %d", len(resp), len(body)+5)
128 }
129 _ = c.Close()
130 e.waitIdle()
131 if n := e.logs.Count(`"reason":"closed"`); n < 1 {
132 t.Fatalf("stream did not close cleanly:\n%s", e.logs.String())
133 }
134}
135
136func TestSplice_ManyConcurrentStreams(t *testing.T) {
137 e := newTestEnv(t, nil)
138 d := e.newDevice()
139 dc := e.connectOK(d)
140 ds := dc.serve(nil)
141 const n = 30
142 var wg sync.WaitGroup
143 errs := make(chan error, n)
144 for i := 0; i < n; i++ {
145 wg.Add(1)
146 go func(i int) {
147 defer wg.Done()
148 c, err := e.clientTLS(d)
149 if err != nil {
150 errs <- err
151 return
152 }
153 defer c.Close()
154 body := append(make([]byte, 20000), fmt.Sprintf("stream-%d", i)...)
155 if _, err := rand.Read(body[:20000]); err != nil {
156 errs <- err
157 return
158 }
159 resp, err := postOver(c, d.host, body)
160 if err != nil {
161 errs <- err
162 return
163 }
164 if !bytes.Equal(resp, append([]byte("echo:"), body...)) {
165 errs <- fmt.Errorf("stream %d: wrong echo", i)
166 }
167 }(i)
168 }
169 wg.Wait()
170 close(errs)
171 for err := range errs {
172 t.Error(err)
173 }
174 e.waitIdle()
175 ids := map[string]bool{}
176 for _, m := range ds.openList() {
177 ids[m.str("stream_id")] = true
178 }
179 if len(ids) != n {
180 t.Fatalf("expected %d distinct stream ids, got %d", n, len(ids))
181 }
182 e.checkTranscript()
183}
184
185func TestSplice_ClientHelloArrivesInTinyPieces(t *testing.T) {
186 e := newTestEnv(t, nil)
187 d := e.newDevice()
188 dc := e.connectOK(d)
189 dc.serve(nil)
190 tcp, err := net.Dial("tcp", e.httpsAddr)
191 if err != nil {
192 t.Fatal(err)
193 }
194 c := tls.Client(chunkConn{Conn: tcp, size: 5, delay: time.Millisecond}, &tls.Config{ServerName: d.host, RootCAs: d.pool, NextProtos: []string{"http/1.1"}})
195 defer c.Close()
196 _ = c.SetDeadline(time.Now().Add(20 * time.Second))
197 if err := c.Handshake(); err != nil {
198 t.Fatal(err)
199 }
200 resp, err := postOver(c, d.host, []byte("tiny-pieces-"+marker))
201 if err != nil || !bytes.HasSuffix(resp, []byte(marker)) {
202 t.Fatalf("resp=%q err=%v", resp, err)
203 }
204}
205
206func TestSplice_ClientHalfClosesButStillReadsResponse(t *testing.T) {
207 e := newTestEnv(t, nil)
208 d := e.newDevice()
209 dc := e.connectOK(d)
210 dc.serve(nil)
211 tcp, err := net.Dial("tcp", e.httpsAddr)
212 if err != nil {
213 t.Fatal(err)
214 }
215 c := tls.Client(tcp, &tls.Config{ServerName: d.host, RootCAs: d.pool, NextProtos: []string{"http/1.1"}})
216 defer c.Close()
217 _ = c.SetDeadline(time.Now().Add(20 * time.Second))
218 body := []byte("half-close-" + marker)
219 req, _ := http.NewRequest("POST", "https://"+d.host+"/mcp", bytes.NewReader(body))
220 req.ContentLength = int64(len(body))
221 if err := req.Write(c); err != nil {
222 t.Fatal(err)
223 }
224 // FIN toward the node right after the request: the device's answer must still make it back.
225 if err := tcp.(*net.TCPConn).CloseWrite(); err != nil {
226 t.Fatal(err)
227 }
228 resp, err := http.ReadResponse(bufio.NewReader(c), req)
229 if err != nil {
230 t.Fatal(err)
231 }
232 got, _ := io.ReadAll(resp.Body)
233 if !bytes.HasSuffix(got, []byte(marker)) {
234 t.Fatalf("response lost after client half-close: %q", got)
235 }
236}
237
238func deviceTLSServer(d *testDevice, c net.Conn) *tls.Conn {
239 return tls.Server(c, &tls.Config{Certificates: []tls.Certificate{d.cert}, NextProtos: []string{"http/1.1"}})
240}
241
242func TestSplice_DeviceResetMidResponseClosesClientPromptly(t *testing.T) {
243 e := newTestEnv(t, nil)
244 d := e.newDevice()
245 dc := e.connectOK(d)
246 started := make(chan struct{}, 1)
247 dc.serve(func(c net.Conn) {
248 tc := deviceTLSServer(d, c)
249 if tc.Handshake() != nil {
250 return
251 }
252 req, err := http.ReadRequest(bufio.NewReader(tc))
253 if err != nil {
254 return
255 }
256 _, _ = io.Copy(io.Discard, req.Body)
257 _, _ = io.WriteString(tc, "HTTP/1.1 200 OK\r\nContent-Length: 100000\r\n\r\nhalf")
258 started <- struct{}{}
259 // Abort the tunnel connection with a TCP reset.
260 raw := c.(*tls.Conn).NetConn().(*net.TCPConn)
261 _ = raw.SetLinger(0)
262 _ = raw.Close()
263 })
264 c, err := e.clientTLS(d)
265 if err != nil {
266 t.Fatal(err)
267 }
268 defer c.Close()
269 if _, err = postOver(c, d.host, []byte("x")); err == nil {
270 t.Fatal("client should see the truncated response as an error")
271 }
272 <-started
273 e.waitIdle()
274 if e.logs.Count(`"reason":"conn_error"`) == 0 {
275 t.Fatalf("expected conn_error:\n%s", e.logs.String())
276 }
277}
278
279func TestSplice_ClientResetClosesDevicePromptly(t *testing.T) {
280 e := newTestEnv(t, nil)
281 d := e.newDevice()
282 dc := e.connectOK(d)
283 handlerDone := make(chan error, 1)
284 dc.serve(func(c net.Conn) {
285 tc := deviceTLSServer(d, c)
286 if err := tc.Handshake(); err != nil {
287 handlerDone <- err
288 return
289 }
290 _ = tc.SetReadDeadline(time.Now().Add(10 * time.Second))
291 _, err := io.Copy(io.Discard, tc) // blocks until the stream dies
292 handlerDone <- err
293 })
294 tcp, err := net.Dial("tcp", e.httpsAddr)
295 if err != nil {
296 t.Fatal(err)
297 }
298 c := tls.Client(tcp, &tls.Config{ServerName: d.host, RootCAs: d.pool})
299 if err := c.Handshake(); err != nil {
300 t.Fatal(err)
301 }
302 _ = tcp.(*net.TCPConn).SetLinger(0)
303 _ = tcp.Close() // RST
304 select {
305 case <-handlerDone:
306 case <-time.After(5 * time.Second):
307 t.Fatal("device side was not closed after the client vanished")
308 }
309 e.waitIdle()
310}
311
312// ------------------------------------------------------------------------------------------------------------------------------------
313// :443 routing decisions
314// ------------------------------------------------------------------------------------------------------------------------------------
315
316func TestSNIRouting_Rejections(t *testing.T) {
317 e := newTestEnv(t, func(c *Config) { c.HelloTimeout = 400 * time.Millisecond })
318 d := e.newDevice()
319 dc := e.connectOK(d)
320 ds := dc.serve(nil)
321
322 dialTLS := func(serverName string) error {
323 tcp, err := net.Dial("tcp", e.httpsAddr)
324 if err != nil {
325 t.Fatal(err)
326 }
327 defer tcp.Close()
328 _ = tcp.SetDeadline(time.Now().Add(5 * time.Second))
329 return tls.Client(tcp, &tls.Config{ServerName: serverName, InsecureSkipVerify: true}).Handshake()
330 }
331
332 t.Run("no SNI is closed with an alert", func(t *testing.T) {
333 err := dialTLS("")
334 if err == nil || !strings.Contains(err.Error(), "unrecognized name") {
335 t.Fatalf("err = %v", err)
336 }
337 if e.logs.Count(`"reason":"no_sni"`) != 1 {
338 t.Fatalf("logs:\n%s", e.logs.String())
339 }
340 })
341 t.Run("unknown host", func(t *testing.T) {
342 err := dialTLS("nobody.example.test")
343 if err == nil || !strings.Contains(err.Error(), "unrecognized name") {
344 t.Fatalf("err = %v", err)
345 }
346 })
347 t.Run("unusable SNI value", func(t *testing.T) {
348 out := rawExchange(t, e.httpsAddr, string(buildHello(sniExt("192.0.2.3"), true)))
349 if out != "\x15\x03\x03\x00\x02\x02\x70" {
350 t.Fatalf("response = %q", out)
351 }
352 if e.logs.Count(`"reason":"bad_sni"`) != 1 {
353 t.Fatalf("logs:\n%s", e.logs.String())
354 }
355 })
356 t.Run("plain HTTP on 443 is just closed", func(t *testing.T) {
357 if out := rawExchange(t, e.httpsAddr, "GET / HTTP/1.1\r\nHost: "+d.host+"\r\n\r\n"); out != "" {
358 t.Fatalf("response = %q", out)
359 }
360 if e.logs.Count(`"reason":"not_tls"`) != 1 {
361 t.Fatalf("logs:\n%s", e.logs.String())
362 }
363 })
364 t.Run("silent client times out", func(t *testing.T) {
365 c, err := net.Dial("tcp", e.httpsAddr)
366 if err != nil {
367 t.Fatal(err)
368 }
369 defer c.Close()
370 _ = c.SetReadDeadline(time.Now().Add(5 * time.Second))
371 start := time.Now()
372 if n, err := c.Read(make([]byte, 1)); n != 0 || err == nil {
373 t.Fatalf("n=%d err=%v", n, err)
374 }
375 if time.Since(start) > 3*time.Second {
376 t.Fatal("hello timeout not enforced")
377 }
378 eventually(t, "hello_timeout log", 2*time.Second, func() bool { return e.logs.Count(`"reason":"hello_timeout"`) == 1 })
379 })
380 if n := ds.openCount(); n != 0 {
381 t.Fatalf("the device was asked to open %d streams for traffic that should never reach it", n)
382 }
383}
384
385func TestSNIRouting_DeviceOffline(t *testing.T) {
386 e := newTestEnv(t, nil)
387 d := e.newDevice()
388 // No device connected yet.
389 if _, err := e.clientTLS(d); err == nil || !strings.Contains(err.Error(), "unrecognized name") {
390 t.Fatalf("err = %v", err)
391 }
392 // Online, works; offline again after the control connection ends.
393 dc := e.connectOK(d)
394 dc.serve(nil)
395 c, err := e.clientTLS(d)
396 if err != nil {
397 t.Fatal(err)
398 }
399 _ = c.Close()
400 _ = dc.c.Close()
401 eventually(t, "device offline", 3*time.Second, func() bool { return !e.srv.Online(d.host) })
402 if _, err := e.clientTLS(d); err == nil {
403 t.Fatal("handshake should fail while the device is offline")
404 }
405 e.checkTranscript()
406}
407
408// ------------------------------------------------------------------------------------------------------------------------------------
409// Limits, timeouts, shutdown
410// ------------------------------------------------------------------------------------------------------------------------------------
411
412func holdHandler(d *testDevice, release <-chan struct{}) func(net.Conn) {
413 return func(c net.Conn) {
414 tc := deviceTLSServer(d, c)
415 if tc.Handshake() != nil {
416 return
417 }
418 select {
419 case <-release:
420 case <-time.After(15 * time.Second):
421 }
422 }
423}
424
425func TestLimits_PerHostAndGlobal(t *testing.T) {
426 t.Run("per host", func(t *testing.T) {
427 release := make(chan struct{})
428 e := newTestEnv(t, func(c *Config) { c.MaxStreamsPerHost = 2 })
429 d := e.newDevice()
430 dc := e.connectOK(d)
431 dc.serve(holdHandler(d, release))
432 c1, err1 := e.clientTLS(d)
433 c2, err2 := e.clientTLS(d)
434 if err1 != nil || err2 != nil {
435 t.Fatalf("first two streams must work: %v %v", err1, err2)
436 }
437 if _, err := e.clientTLS(d); err == nil || !strings.Contains(err.Error(), "internal error") {
438 t.Fatalf("third stream must be refused, got %v", err)
439 }
440 if e.logs.Count(`"reason":"host_limit"`) != 1 {
441 t.Fatalf("logs:\n%s", e.logs.String())
442 }
443 close(release)
444 c1.Close()
445 c2.Close()
446 e.waitIdle()
447 e.srv.routes.mu.Lock()
448 leaked := e.srv.routes.byHost[d.host].streams
449 e.srv.routes.mu.Unlock()
450 if leaked != 0 {
451 t.Fatalf("per-host counter leaked: %d", leaked)
452 }
453 })
454 t.Run("global", func(t *testing.T) {
455 release := make(chan struct{})
456 e := newTestEnv(t, func(c *Config) { c.MaxStreams = 1 })
457 d := e.newDevice()
458 dc := e.connectOK(d)
459 dc.serve(holdHandler(d, release))
460 c1, err := e.clientTLS(d)
461 if err != nil {
462 t.Fatal(err)
463 }
464 if _, err := e.clientTLS(d); err == nil {
465 t.Fatal("second stream must be refused")
466 }
467 close(release)
468 c1.Close()
469 e.waitIdle()
470 if e.logs.Count(`"reason":"global_limit"`) != 1 {
471 t.Fatalf("logs:\n%s", e.logs.String())
472 }
473 })
474}
475
476func TestLimits_StreamByteCap(t *testing.T) {
477 e := newTestEnv(t, func(c *Config) { c.MaxStreamBytes = 100_000 })
478 d := e.newDevice()
479 dc := e.connectOK(d)
480 dc.serve(nil)
481 c, err := e.clientTLS(d)
482 if err != nil {
483 t.Fatal(err)
484 }
485 defer c.Close()
486 if _, err := postOver(c, d.host, make([]byte, 400_000)); err == nil {
487 t.Fatal("a stream above the byte cap must be cut")
488 }
489 e.waitIdle()
490 if e.logs.Count(`"reason":"byte_cap"`) != 1 {
491 t.Fatalf("logs:\n%s", e.logs.String())
492 }
493}
494
495func TestLimits_StreamIdleAndMaxAge(t *testing.T) {
496 for _, tc := range []struct {
497 name string
498 mutate func(*Config)
499 reason string
500 }{
501 {"idle", func(c *Config) { c.StreamIdle = 300 * time.Millisecond }, "idle"},
502 {"max age", func(c *Config) { c.StreamMaxAge = 300 * time.Millisecond }, "max_age"},
503 } {
504 t.Run(tc.name, func(t *testing.T) {
505 release := make(chan struct{})
506 defer close(release)
507 e := newTestEnv(t, tc.mutate)
508 d := e.newDevice()
509 dc := e.connectOK(d)
510 dc.serve(holdHandler(d, release))
511 c, err := e.clientTLS(d)
512 if err != nil {
513 t.Fatal(err)
514 }
515 defer c.Close()
516 _ = c.SetReadDeadline(time.Now().Add(5 * time.Second))
517 if _, err := c.Read(make([]byte, 1)); err == nil {
518 t.Fatal("the stream should have been closed by the node")
519 }
520 e.waitIdle()
521 if e.logs.Count(`"reason":"`+tc.reason+`"`) != 1 {
522 t.Fatalf("logs:\n%s", e.logs.String())
523 }
524 })
525 }
526}
527
528func TestLimits_AttachTimeoutClosesTheClientAndForgetsTheStream(t *testing.T) {
529 e := newTestEnv(t, func(c *Config) { c.AttachTimeout = 300 * time.Millisecond })
530 d := e.newDevice()
531 dc := e.connectOK(d)
532 ds := dc.serve(nil)
533 ds.noAttach = true
534 if _, err := e.clientTLS(d); err == nil || !strings.Contains(err.Error(), "internal error") {
535 t.Fatalf("err = %v", err)
536 }
537 e.waitIdle()
538 e.srv.mu.Lock()
539 left := len(e.srv.pending)
540 e.srv.mu.Unlock()
541 if left != 0 {
542 t.Fatalf("%d pending streams leaked", left)
543 }
544 if e.logs.Count(`"reason":"attach_timeout"`) != 1 {
545 t.Fatalf("logs:\n%s", e.logs.String())
546 }
547}
548
549func TestShutdown_TellsDevicesAndClosesEverythingWithinGrace(t *testing.T) {
550 release := make(chan struct{})
551 defer close(release)
552 e := newTestEnv(t, nil)
553 d := e.newDevice()
554 dc := e.connectOK(d)
555 ds := dc.serve(holdHandler(d, release))
556 c, err := e.clientTLS(d)
557 if err != nil {
558 t.Fatal(err)
559 }
560 defer c.Close()
561 start := time.Now()
562 e.srv.Shutdown(500 * time.Millisecond)
563 if time.Since(start) > 4*time.Second {
564 t.Fatal("shutdown hung")
565 }
566 _ = c.SetReadDeadline(time.Now().Add(3 * time.Second))
567 if _, err := c.Read(make([]byte, 1)); err == nil {
568 t.Fatal("stream survived shutdown")
569 }
570 if _, err := net.DialTimeout("tcp", e.httpsAddr, 500*time.Millisecond); err == nil {
571 t.Fatal("listener still open after shutdown")
572 }
573 <-ds.done
574 if !containsLine(ds.otherLines(), "CLOSE shutdown") {
575 t.Fatalf("the device was not told: %q", ds.otherLines())
576 }
577}