ALL

Solo para leer y verificar. No puedes copiar, usar, modificar ni redistribuir este código. Aviso completo en la página principal.

tunnel/helpers_test.go

SHA-256: 9212c296e9f2150b2188f3ed727c89213728da210c5fcee6fdff7d2a3c7395f1
843 líneas · 25887 bytes · Ver en GitHub

1package main
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "crypto/ecdsa"
8 "crypto/ed25519"
9 "crypto/elliptic"
10 "crypto/rand"
11 "crypto/tls"
12 "crypto/x509"
13 "crypto/x509/pkix"
14 "encoding/hex"
15 "encoding/pem"
16 "fmt"
17 "io"
18 "log/slog"
19 "math/big"
20 "net"
21 "net/http"
22 "os"
23 "strings"
24 "sync"
25 "sync/atomic"
26 "testing"
27 "time"
28)
29
30// marker is a recognizable plaintext. If it ever shows up in a buffer the node touched while splicing a TLS stream, the node could read content.
31const marker = "MARKER-9f3a1c-PLAINTEXT-MUST-NOT-LEAK-7b2e"
32
33const (
34 testZone = "devices.all-e2e.test"
35 testNodeHost = "t1.nodes.all-e2e.test"
36 testNodeID = "t1"
37 testKID = "cp-test-1"
38)
39
40type syncBuf struct {
41 mu sync.Mutex
42 b bytes.Buffer
43}
44
45func (s *syncBuf) Write(p []byte) (int, error) {
46 s.mu.Lock()
47 defer s.mu.Unlock()
48 return s.b.Write(p)
49}
50
51func (s *syncBuf) String() string {
52 s.mu.Lock()
53 defer s.mu.Unlock()
54 return s.b.String()
55}
56
57func (s *syncBuf) Count(sub string) int { return strings.Count(s.String(), sub) }
58
59// makeCA returns a CA pool and a leaf certificate for names, signed by it: this is what a DEVICE owns (the node never has the key), and what the
60// node itself holds for its own name.
61func makeCA(t testing.TB, names ...string) (*x509.CertPool, tls.Certificate) {
62 t.Helper()
63 caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
64 if err != nil {
65 t.Fatal(err)
66 }
67 caTmpl := &x509.Certificate{
68 SerialNumber: big.NewInt(1),
69 Subject: pkix.Name{CommonName: "test CA"},
70 NotBefore: time.Now().Add(-time.Hour),
71 NotAfter: time.Now().Add(24 * time.Hour),
72 IsCA: true,
73 KeyUsage: x509.KeyUsageCertSign,
74 BasicConstraintsValid: true,
75 }
76 caDER, err := x509.CreateCertificate(rand.Reader, caTmpl, caTmpl, &caKey.PublicKey, caKey)
77 if err != nil {
78 t.Fatal(err)
79 }
80 caCert, _ := x509.ParseCertificate(caDER)
81 leafKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
82 if err != nil {
83 t.Fatal(err)
84 }
85 leafTmpl := &x509.Certificate{
86 SerialNumber: big.NewInt(2),
87 Subject: pkix.Name{CommonName: names[0]},
88 NotBefore: time.Now().Add(-time.Hour),
89 NotAfter: time.Now().Add(24 * time.Hour),
90 KeyUsage: x509.KeyUsageDigitalSignature,
91 ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
92 DNSNames: names,
93 }
94 leafDER, err := x509.CreateCertificate(rand.Reader, leafTmpl, caCert, &leafKey.PublicKey, caKey)
95 if err != nil {
96 t.Fatal(err)
97 }
98 pool := x509.NewCertPool()
99 pool.AddCert(caCert)
100 return pool, tls.Certificate{Certificate: [][]byte{leafDER}, PrivateKey: leafKey}
101}
102
103// ---------------------------------------------------------------------------------------------------------------------------- control plane fake
104
105// devicePlan is what the fake control plane answers for one device.
106type devicePlan struct {
107 host string
108 status string // active (default), gated or paused
109 plan string // paid (default) or trial
110 validFor time.Duration
111 err error // refuse instead of answering
112 delay time.Duration // answer late
113 answer func(n int) AuthorizeResponse // overrides everything, n is the call number of this device (1 = first)
114}
115
116type fakeControl struct {
117 env *testEnv
118
119 mu sync.Mutex
120 plans map[string]*devicePlan
121 calls []AuthorizeRequest
122 count map[string]int
123
124 inflight, maxInflight int // calls being answered now, and the most there ever were at once
125}
126
127func (f *fakeControl) set(deviceID string, p *devicePlan) {
128 f.mu.Lock()
129 defer f.mu.Unlock()
130 f.plans[deviceID] = p
131}
132
133func (f *fakeControl) callLog() []AuthorizeRequest {
134 f.mu.Lock()
135 defer f.mu.Unlock()
136 return append([]AuthorizeRequest(nil), f.calls...)
137}
138
139func (f *fakeControl) callsFor(deviceID, reason string) int {
140 n := 0
141 for _, c := range f.callLog() {
142 if c.DeviceID == deviceID && (reason == "" || c.Reason == reason) {
143 n++
144 }
145 }
146 return n
147}
148
149func (f *fakeControl) Authorize(ctx context.Context, req AuthorizeRequest) (AuthorizeResponse, error) {
150 f.mu.Lock()
151 f.calls = append(f.calls, req)
152 f.count[req.DeviceID]++
153 n := f.count[req.DeviceID]
154 p := f.plans[req.DeviceID]
155 f.inflight++
156 f.maxInflight = max(f.maxInflight, f.inflight)
157 f.mu.Unlock()
158 defer func() {
159 f.mu.Lock()
160 f.inflight--
161 f.mu.Unlock()
162 }()
163 if p == nil {
164 return AuthorizeResponse{}, &ControlError{Status: 401, Code: codeDeviceRevoked}
165 }
166 if p.delay > 0 {
167 select {
168 case <-time.After(p.delay):
169 case <-ctx.Done():
170 return AuthorizeResponse{}, ctx.Err()
171 }
172 }
173 if p.err != nil {
174 return AuthorizeResponse{}, p.err
175 }
176 if p.answer != nil {
177 return p.answer(n), nil
178 }
179 now := f.env.now()
180 status, plan := p.status, p.plan
181 if status == "" {
182 status = statusActive
183 }
184 if plan == "" {
185 plan = "paid"
186 }
187 label, _, _ := strings.Cut(p.host, ".")
188 t := policyData.Timing
189 return AuthorizeResponse{
190 DeviceID: req.DeviceID, Label: label, Host: p.host, Status: status, Plan: plan,
191 ValidUntil: now.Unix() + int64(p.validFor/time.Second), ServerTime: now.Unix(),
192 Policy: okPolicy{int64(t.PingIntervalS), int64(t.DeadPeerS), int64(t.StreamIdleS), int64(t.StreamMaxAgeS), int64(t.RenewBeforeValidUntilS)},
193 Nodes: []any{},
194 }, nil
195}
196
197// ---------------------------------------------------------------------------------------------------------------------------- the environment
198
199// testDevice is a device the tests play: an id, the Ed25519 key its sessions are bound to, and (for the clients that reach it) the TLS certificate
200// it holds for its host.
201type testDevice struct {
202 id, label, host string
203 key ed25519.PrivateKey
204 pool *x509.CertPool
205 cert tls.Certificate
206}
207
208type testEnv struct {
209 t *testing.T
210 srv *Server
211 logs *syncBuf
212 cp *fakeControl
213 keys *SwapKeys
214 cpKey ed25519.PrivateKey
215 nodePool *x509.CertPool
216
217 httpsAddr, httpAddr string
218
219 vip *virtualIPs // when set, connections made with vip.dial come from the addresses the test chose (hardening tests)
220 offsetNS atomic.Int64 // the node's clock is the real one plus this
221 clk *fakeClock // when set, the node's clock and timers are this one and the tests move it by hand
222
223 tapMu sync.Mutex
224 tapAll bytes.Buffer
225 tapByID map[string]*bytes.Buffer
226
227 traceMu sync.Mutex
228 events []any
229 seq int
230}
231
232func (e *testEnv) now() time.Time {
233 if e.clk != nil {
234 return e.clk.Now()
235 }
236 return time.Now().Add(time.Duration(e.offsetNS.Load()))
237}
238
239// newTestEnv starts a node (loopback, ephemeral ports) with a fake control plane and a key set whose private half the tests hold.
240func newTestEnv(t *testing.T, mutate func(*Config)) *testEnv {
241 return newEnv(t, nil, func(_ *testEnv, cfg *Config) {
242 if mutate != nil {
243 mutate(cfg)
244 }
245 })
246}
247
248// newClockEnv is newTestEnv on a clock the test moves by hand: every lifetime the node decides (valid_until, the grace, the feed's pace, the quiet
249// period of a drain) follows it, so hours pass in an instant.
250func newClockEnv(t *testing.T, mutate func(*Config)) *testEnv {
251 return newEnv(t, newFakeClock(), func(_ *testEnv, cfg *Config) {
252 if mutate != nil {
253 mutate(cfg)
254 }
255 })
256}
257
258// newEnv builds the node; mutate gets the environment (its fake control plane, its keys, its clock) and the configuration before the node starts.
259func newEnv(t *testing.T, clk *fakeClock, mutate func(*testEnv, *Config)) *testEnv {
260 t.Helper()
261 e := &testEnv{t: t, clk: clk, logs: &syncBuf{}, tapByID: map[string]*bytes.Buffer{}}
262 pub, priv, err := ed25519.GenerateKey(rand.Reader)
263 if err != nil {
264 t.Fatal(err)
265 }
266 e.cpKey = priv
267 e.keys = NewSwapKeys(StaticKeys{testKID: pub})
268 e.cp = &fakeControl{env: e, plans: map[string]*devicePlan{}, count: map[string]int{}}
269 pool, nodeCert := makeCA(t, testNodeHost)
270 e.nodePool = pool
271
272 cfg := DefaultConfig()
273 cfg.HTTPSAddr, cfg.HTTPAddr = "127.0.0.1:0", "127.0.0.1:0"
274 cfg.NodeID, cfg.NodeHost = testNodeID, testNodeHost
275 cfg.CertSource = func(*tls.ClientHelloInfo) (*tls.Certificate, error) { return &nodeCert, nil }
276 cfg.Keys, cfg.Control = e.keys, e.cp
277 cfg.Now = e.now
278 if clk != nil {
279 cfg.AfterFunc = clk.AfterFunc
280 }
281 cfg.MinRenewGap = 0 // tests renew in quick succession; the test of the gap sets its own
282 // Every test client and device comes from loopback: one source. The per-source limits (limits_test.go gives its own) would make a crowd of devices
283 // in a test look like one attacker, so they are out of the way here.
284 cfg.PerIPConcurrent, cfg.PerIPNewPerS = 1<<20, 1<<20
285 cfg.UnknownSNIPerIP, cfg.LabelNewPerS, cfg.LabelNewBurst = 1<<20, 1<<20, 1<<20
286 cfg.LabelAnonSlice = cfg.MaxStreamsPerHost // and every test client is anonymous: the slices have their own tests
287 cfg.HelloTimeout = 2 * time.Second
288 cfg.AttachTimeout = 3 * time.Second
289 cfg.FirstLineTimeout = 3 * time.Second
290 cfg.StreamIdle = 20 * time.Second
291 cfg.StreamMaxAge = time.Minute
292 cfg.HalfCloseTimeout = 5 * time.Second
293 cfg.ACMETimeout = 5 * time.Second
294 cfg.Logger = slog.New(slog.NewJSONHandler(e.logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
295 cfg.tap = func(id, dir string, p []byte) {
296 e.tapMu.Lock()
297 defer e.tapMu.Unlock()
298 e.tapAll.Write(p)
299 b := e.tapByID[id]
300 if b == nil {
301 b = &bytes.Buffer{}
302 e.tapByID[id] = b
303 }
304 b.Write(p)
305 }
306 cfg.trace = func(ev map[string]any) {
307 // the checker reads JSON values: numbers are float64, whatever the node used
308 parsed, err := parseJSON(canonicalJSON(ev))
309 if err != nil {
310 panic(err)
311 }
312 e.traceMu.Lock()
313 e.events = append(e.events, parsed)
314 e.traceMu.Unlock()
315 }
316 if mutate != nil {
317 mutate(e, &cfg)
318 }
319 srv, err := NewServer(cfg)
320 if err != nil {
321 t.Fatal(err)
322 }
323 if err := srv.Start(); err != nil {
324 t.Fatal(err)
325 }
326 e.srv = srv
327 e.httpsAddr, e.httpAddr = srv.HTTPSAddr(), srv.HTTPAddr()
328 t.Cleanup(func() { srv.Shutdown(2 * time.Second) })
329 return e
330}
331
332func (e *testEnv) tapped(id string) []byte {
333 e.tapMu.Lock()
334 defer e.tapMu.Unlock()
335 if b := e.tapByID[id]; b != nil {
336 return append([]byte(nil), b.Bytes()...)
337 }
338 return nil
339}
340
341func (e *testEnv) tappedAll() []byte {
342 e.tapMu.Lock()
343 defer e.tapMu.Unlock()
344 return append([]byte(nil), e.tapAll.Bytes()...)
345}
346
347func randomHex(n int) string {
348 b := make([]byte, n)
349 _, _ = rand.Read(b)
350 return hex.EncodeToString(b)
351}
352
353// label20 is a label of the contracts' alphabet.
354func label20() string {
355 const alphabet = "abcdefghijklmnopqrstuvwxyz234567"
356 b := make([]byte, 20)
357 _, _ = rand.Read(b)
358 for i := range b {
359 b[i] = alphabet[int(b[i])%len(alphabet)]
360 }
361 return string(b)
362}
363
364// newDevice makes a device and tells the fake control plane about it (host under the test zone, paid, valid for an hour unless changed).
365func (e *testEnv) newDevice() *testDevice {
366 d := e.newDeviceNoPlan()
367 e.cp.set(d.id, &devicePlan{host: d.host, validFor: time.Hour})
368 return d
369}
370
371// newDeviceNoPlan makes a device the control plane has not been told about.
372func (e *testEnv) newDeviceNoPlan() *testDevice {
373 e.t.Helper()
374 _, priv, err := ed25519.GenerateKey(rand.Reader)
375 if err != nil {
376 e.t.Fatal(err)
377 }
378 d := &testDevice{id: randomHex(16), label: label20(), key: priv}
379 d.host = d.label + "." + testZone
380 d.pool, d.cert = makeCA(e.t, d.host)
381 return d
382}
383
384func (e *testEnv) plan(d *testDevice) *devicePlan {
385 e.cp.mu.Lock()
386 defer e.cp.mu.Unlock()
387 return e.cp.plans[d.id]
388}
389
390// ---------------------------------------------------------------------------------------------------------------------------- tokens
391
392func signJWS(key ed25519.PrivateKey, header, claims map[string]any) string {
393 h := b64Encode([]byte(canonicalJSON(header)))
394 c := b64Encode([]byte(canonicalJSON(claims)))
395 input := h + "." + c
396 return input + "." + b64Encode(ed25519.Sign(key, []byte(input)))
397}
398
399// session mints a device session for d as the control plane would (aud all-tunnel, 300 s). mutate changes the claims or the header before signing.
400func (e *testEnv) session(d *testDevice, mutate func(header, claims map[string]any)) string {
401 return e.sessionSignedBy(e.cpKey, d, mutate)
402}
403
404// sessionSignedBy is session signed with another key (a forger's).
405func (e *testEnv) sessionSignedBy(signer ed25519.PrivateKey, d *testDevice, mutate func(header, claims map[string]any)) string {
406 iat := e.now().Unix()
407 pub := d.key.Public().(ed25519.PublicKey)
408 header := map[string]any{"alg": "EdDSA", "typ": "device-session", "kid": testKID}
409 claims := map[string]any{
410 "typ": "device-session", "ver": 2, "iss": policyData.Identity.Issuer, "aud": policyData.Identity.AudTunnel, "sub": d.id,
411 "iat": iat, "exp": iat + int64(policyData.Authorization.SessionTTLS), "jti": randomHex(12), "generation": 1,
412 "cnf": map[string]any{"jwk": map[string]any{"kty": "OKP", "crv": "Ed25519", "x": b64Encode(pub)}},
413 }
414 if mutate != nil {
415 mutate(header, claims)
416 }
417 return signJWS(signer, header, claims)
418}
419
420// proof mints the DPoP proof of d over session and nonce for the node's URL.
421func (e *testEnv) proof(d *testDevice, session, nonce string, mutate func(header, claims map[string]any)) string {
422 pub := d.key.Public().(ed25519.PublicKey)
423 header := map[string]any{"alg": "EdDSA", "typ": "dpop+jwt", "jwk": map[string]any{"kty": "OKP", "crv": "Ed25519", "x": b64Encode(pub)}}
424 claims := map[string]any{
425 "sub": d.id, "aud": policyData.Identity.AudTunnel, "htm": policyData.Protocol.ProofHTM, "htu": e.srv.nodeURL, "iat": e.now().Unix(),
426 "jti": randomHex(12), "ath": sha256B64([]byte(session)), "nonce": nonce,
427 }
428 if mutate != nil {
429 mutate(header, claims)
430 }
431 return signJWS(d.key, header, claims)
432}
433
434// ---------------------------------------------------------------------------------------------------------------------------- device side
435
436// dialNode connects to the node's own name the way a device does: TLS 1.3, the node's public certificate, ALPN all-tunnel/2.
437func (e *testEnv) dialNode() (*tls.Conn, error) {
438 return tls.DialWithDialer(&net.Dialer{Timeout: 3 * time.Second}, "tcp", e.httpsAddr, &tls.Config{
439 ServerName: testNodeHost, RootCAs: e.nodePool, NextProtos: []string{policyData.Protocol.ALPN}, MinVersion: tls.VersionTLS13,
440 })
441}
442
443// devConn is the control connection of a device, played by a test.
444type devConn struct {
445 t *testing.T
446 e *testEnv
447 d *testDevice
448 c *tls.Conn
449 br *bufio.Reader
450
451 session string
452 ok message // the parsed OK, once the handshake finished
453
454 feed chan string // once a devServer reads the connection, the lines that are not OPEN arrive here (closed at the end of the connection)
455}
456
457func (e *testEnv) connect(d *testDevice) *devConn {
458 e.t.Helper()
459 c, err := e.dialNode()
460 if err != nil {
461 e.t.Fatal(err)
462 }
463 dc := &devConn{t: e.t, e: e, d: d, c: c, br: bufio.NewReaderSize(c, 1<<17)}
464 e.t.Cleanup(func() { _ = c.Close() })
465 return dc
466}
467
468func (dc *devConn) send(line string) {
469 dc.t.Helper()
470 _ = dc.c.SetWriteDeadline(time.Now().Add(3 * time.Second))
471 if _, err := io.WriteString(dc.c, line+"\n"); err != nil {
472 dc.t.Fatalf("send %.40q: %v", line, err)
473 }
474}
475
476// try reads one line within timeout.
477func (dc *devConn) try(timeout time.Duration) (string, error) {
478 if dc.feed != nil {
479 select {
480 case line, ok := <-dc.feed:
481 if !ok {
482 return "", io.EOF
483 }
484 return line, nil
485 case <-time.After(timeout):
486 return "", &net.OpError{Op: "read", Err: os.ErrDeadlineExceeded}
487 }
488 }
489 _ = dc.c.SetReadDeadline(time.Now().Add(timeout))
490 b, err := dc.br.ReadString('\n')
491 if err != nil {
492 return strings.TrimSuffix(b, "\n"), err
493 }
494 return strings.TrimSuffix(b, "\n"), nil
495}
496
497func (dc *devConn) recv() string {
498 dc.t.Helper()
499 line, err := dc.try(5 * time.Second)
500 if err != nil {
501 dc.t.Fatalf("no line from the node: %q %v", line, err)
502 }
503 return line
504}
505
506// expectClosed asserts the node closes the connection within timeout (any lines before the close are returned).
507func (dc *devConn) expectClosed(timeout time.Duration) []string {
508 dc.t.Helper()
509 var lines []string
510 deadline := time.Now().Add(timeout)
511 for {
512 line, err := dc.try(time.Until(deadline))
513 if err != nil {
514 if ne, ok := err.(net.Error); ok && ne.Timeout() {
515 dc.t.Fatalf("connection still open after %s (lines so far %q)", timeout, lines)
516 }
517 return lines
518 }
519 lines = append(lines, line)
520 }
521}
522
523// hello performs HELLO and PROOF and returns the node's answer line (CHALLENGE and OK, or an ERR at any step) without judging it.
524func (dc *devConn) hello(session func() string, proof func(session, nonce string) string) (final string) {
525 dc.t.Helper()
526 dc.session = session()
527 dc.send("HELLO 2 " + dc.session)
528 line := dc.recv()
529 m, err := parseLine(line, dirN2D)
530 if err != nil || m.str("verb") != "CHALLENGE" {
531 return line
532 }
533 dc.send("PROOF " + proof(dc.session, m.str("nonce")))
534 return dc.recv()
535}
536
537// connectOK completes the handshake for d with good tokens and fails the test unless the node says OK.
538func (e *testEnv) connectOK(d *testDevice) *devConn {
539 e.t.Helper()
540 dc := e.connect(d)
541 line := dc.hello(func() string { return e.session(d, nil) }, func(s, n string) string { return e.proof(d, s, n, nil) })
542 m, err := parseLine(line, dirN2D)
543 if err != nil || m.str("verb") != "OK" {
544 e.t.Fatalf("handshake for %s: %q", d.id, line)
545 }
546 dc.ok = m
547 return dc
548}
549
550// helloAsync runs a whole HELLO for d on its own connection in the background and delivers the node's final line (or the error that ended the
551// exchange); it never calls t.Fatal, so it is safe off the test goroutine.
552func (e *testEnv) helloAsync(d *testDevice) <-chan string {
553 out := make(chan string, 1)
554 c, err := e.dialNode()
555 if err != nil {
556 e.t.Fatal(err)
557 }
558 e.t.Cleanup(func() { _ = c.Close() })
559 go func() {
560 br := bufio.NewReader(c)
561 read := func() string {
562 _ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
563 l, err := br.ReadString('\n')
564 if err != nil {
565 return "read error: " + err.Error()
566 }
567 return strings.TrimSuffix(l, "\n")
568 }
569 session := e.session(d, nil)
570 _, _ = io.WriteString(c, "HELLO 2 "+session+"\n")
571 line := read()
572 if nonce, ok := strings.CutPrefix(line, "CHALLENGE "); ok {
573 _, _ = io.WriteString(c, "PROOF "+e.proof(d, session, nonce, nil)+"\n")
574 line = read()
575 }
576 out <- line
577 }()
578 return out
579}
580
581// renew sends RENEW with a fresh session, answers the CHALLENGE and returns the final line.
582func (dc *devConn) renew() string {
583 dc.t.Helper()
584 dc.session = dc.e.session(dc.d, nil)
585 dc.send("RENEW " + dc.session)
586 line := dc.recv()
587 m, err := parseLine(line, dirN2D)
588 if err != nil || m.str("verb") != "CHALLENGE" {
589 return line
590 }
591 dc.send("PROOF " + dc.e.proof(dc.d, dc.session, m.str("nonce"), nil))
592 return dc.recv()
593}
594
595// devServer plays the device's side of the streams: it reads the control connection, attaches every OPEN through a new data connection and
596// serves it with handler. Lines that are not OPEN are kept for the test.
597type devServer struct {
598 dc *devConn
599 handler func(c net.Conn)
600 noAttach bool
601
602 mu sync.Mutex
603 opens []message
604 other []string
605 wg sync.WaitGroup
606 done chan struct{}
607}
608
609func (dc *devConn) serve(handler func(c net.Conn)) *devServer {
610 ds := &devServer{dc: dc, handler: handler, done: make(chan struct{})}
611 dc.feed = make(chan string, 1024)
612 if handler == nil {
613 ds.handler = ds.echo
614 }
615 go ds.loop()
616 return ds
617}
618
619func (ds *devServer) loop() {
620 defer close(ds.done)
621 defer close(ds.dc.feed)
622 for {
623 _ = ds.dc.c.SetReadDeadline(time.Time{})
624 line, err := ds.dc.br.ReadString('\n')
625 if err != nil {
626 return
627 }
628 line = strings.TrimSuffix(line, "\n")
629 m, perr := parseLine(line, dirN2D)
630 if perr == nil && m.str("verb") == "OPEN" {
631 ds.mu.Lock()
632 ds.opens = append(ds.opens, m)
633 ds.mu.Unlock()
634 if !ds.noAttach {
635 ds.wg.Add(1)
636 go ds.attach(m.str("stream_id"), m.str("secret"))
637 }
638 continue
639 }
640 ds.mu.Lock()
641 ds.other = append(ds.other, line)
642 ds.mu.Unlock()
643 select {
644 case ds.dc.feed <- line:
645 default:
646 }
647 }
648}
649
650func (ds *devServer) attach(id, secret string) {
651 defer ds.wg.Done()
652 c, err := ds.dc.e.dialNode()
653 if err != nil {
654 return
655 }
656 defer c.Close()
657 // No reply line: on success the raw stream starts right after ATTACH.
658 fmt.Fprintf(c, "ATTACH %s %s\n", id, secret)
659 ds.handler(c)
660}
661
662func (ds *devServer) openCount() int { ds.mu.Lock(); defer ds.mu.Unlock(); return len(ds.opens) }
663
664func (ds *devServer) openList() []message {
665 ds.mu.Lock()
666 defer ds.mu.Unlock()
667 return append([]message(nil), ds.opens...)
668}
669
670func (ds *devServer) otherLines() []string {
671 ds.mu.Lock()
672 defer ds.mu.Unlock()
673 return append([]string(nil), ds.other...)
674}
675
676// seenRequest is what the device observed after TLS termination (plaintext).
677type seenRequest struct {
678 method, uri, host string
679 header http.Header
680 body []byte
681}
682
683var (
684 seenMu sync.Mutex
685 seenByDS = map[*devServer][]seenRequest{}
686)
687
688func (ds *devServer) seen() []seenRequest {
689 seenMu.Lock()
690 defer seenMu.Unlock()
691 return append([]seenRequest(nil), seenByDS[ds]...)
692}
693
694// echo is the default device behavior: terminate TLS with the device certificate and answer HTTP/1.1 requests with "echo:"+body.
695func (ds *devServer) echo(c net.Conn) {
696 tc := tls.Server(c, &tls.Config{Certificates: []tls.Certificate{ds.dc.d.cert}, NextProtos: []string{"http/1.1"}, MinVersion: tls.VersionTLS12})
697 _ = tc.SetDeadline(time.Now().Add(10 * time.Second))
698 if err := tc.Handshake(); err != nil {
699 return
700 }
701 _ = tc.SetDeadline(time.Time{})
702 br := bufio.NewReader(tc)
703 for {
704 req, err := http.ReadRequest(br)
705 if err != nil {
706 return
707 }
708 body, _ := io.ReadAll(req.Body)
709 seenMu.Lock()
710 seenByDS[ds] = append(seenByDS[ds], seenRequest{method: req.Method, uri: req.RequestURI, host: req.Host, header: req.Header.Clone(), body: body})
711 seenMu.Unlock()
712 out := append([]byte("echo:"), body...)
713 fmt.Fprintf(tc, "HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n", len(out))
714 if _, err := tc.Write(out); err != nil {
715 return
716 }
717 if req.Close {
718 return
719 }
720 }
721}
722
723// ---------------------------------------------------------------------------------------------------------------------------- client side
724
725// clientTLS dials :443 as ChatGPT would: normal TLS verifying the DEVICE's certificate for d's host.
726func (e *testEnv) clientTLS(d *testDevice) (*tls.Conn, error) {
727 return tls.DialWithDialer(&net.Dialer{Timeout: 3 * time.Second}, "tcp", e.httpsAddr, &tls.Config{
728 ServerName: d.host, RootCAs: d.pool, NextProtos: []string{"http/1.1"}, MinVersion: tls.VersionTLS12,
729 })
730}
731
732// postOver sends one POST with the given body over c and returns the response body.
733func postOver(c net.Conn, host string, body []byte) ([]byte, error) {
734 _ = c.SetDeadline(time.Now().Add(20 * time.Second))
735 req, _ := http.NewRequest("POST", "https://"+host+"/mcp", bytes.NewReader(body))
736 req.ContentLength = int64(len(body))
737 if err := req.Write(c); err != nil {
738 return nil, err
739 }
740 resp, err := http.ReadResponse(bufio.NewReader(c), req)
741 if err != nil {
742 return nil, err
743 }
744 defer resp.Body.Close()
745 return io.ReadAll(resp.Body)
746}
747
748// rawExchange sends raw bytes to addr and returns everything read until EOF.
749func rawExchange(t *testing.T, addr, request string) string {
750 t.Helper()
751 c, err := net.DialTimeout("tcp", addr, 3*time.Second)
752 if err != nil {
753 t.Fatal(err)
754 }
755 defer c.Close()
756 _ = c.SetDeadline(time.Now().Add(10 * time.Second))
757 if _, err := io.WriteString(c, request); err != nil {
758 t.Fatal(err)
759 }
760 b, _ := io.ReadAll(c)
761 return string(b)
762}
763
764func eventually(t *testing.T, what string, within time.Duration, cond func() bool) {
765 t.Helper()
766 deadline := time.Now().Add(within)
767 for time.Now().Before(deadline) {
768 if cond() {
769 return
770 }
771 time.Sleep(10 * time.Millisecond)
772 }
773 t.Fatalf("timed out waiting for %s", what)
774}
775
776func (e *testEnv) waitIdle() {
777 e.t.Helper()
778 eventually(e.t, "all streams closed", 5*time.Second, func() bool { return e.srv.streams.Load() == 0 })
779}
780
781// chunkConn writes in tiny pieces with a pause, so the node sees a ClientHello arriving over many TCP reads.
782type chunkConn struct {
783 net.Conn
784 size int
785 delay time.Duration
786}
787
788func (c chunkConn) Write(p []byte) (int, error) {
789 total := 0
790 for len(p) > 0 {
791 n := min(c.size, len(p))
792 w, err := c.Conn.Write(p[:n])
793 total += w
794 if err != nil {
795 return total, err
796 }
797 p = p[n:]
798 time.Sleep(c.delay)
799 }
800 return total, nil
801}
802
803// ---------------------------------------------------------------------------------------------------------------------------- the transcript
804
805// transcriptDoc is the recording of the node in the contracts' format, closed with an end event, ready for ctCheckTranscript.
806func (e *testEnv) transcriptDoc() map[string]any {
807 e.traceMu.Lock()
808 events := append([]any(nil), e.events...)
809 e.traceMu.Unlock()
810 events = append(events, map[string]any{"t": epochSeconds(e.now()), "ev": "end"})
811 keys := map[string]any{}
812 list := []any{}
813 if k, ok := e.keys.Key(testKID); ok {
814 keys = map[string]any{"kty": "OKP", "crv": "Ed25519", "x": b64Encode(k), "kid": testKID, "alg": "EdDSA", "use": "sig", "key_ops": []any{"verify"}}
815 list = append(list, keys)
816 }
817 return map[string]any{
818 "ctx": map[string]any{"node_id": testNodeID, "node_url": e.srv.nodeURL, "skew": float64(e.srv.cfg.ClockSkew / time.Second),
819 "jwks": map[string]any{"keys": list}, "randoms": map[string]any{}},
820 "events": events,
821 }
822}
823
824// checkTranscript runs the contracts' checker over what the node did and fails the test on the first violation.
825func (e *testEnv) checkTranscript() {
826 e.t.Helper()
827 if violation := ctCheckTranscript(e.transcriptDoc()); violation != "" {
828 e.t.Fatalf("the node broke the contract: %s", violation)
829 }
830}
831
832// pemOf is the PEM of a certificate chain and its ECDSA key.
833func pemOf(t *testing.T, cert tls.Certificate) (certPEM, keyPEM []byte) {
834 t.Helper()
835 for _, der := range cert.Certificate {
836 certPEM = append(certPEM, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...)
837 }
838 der, err := x509.MarshalECPrivateKey(cert.PrivateKey.(*ecdsa.PrivateKey))
839 if err != nil {
840 t.Fatal(err)
841 }
842 return certPEM, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: der})
843}