ALL

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

tunnel/controlplane_test.go

SHA-256: ed538062f2ad979ee9b7dd66bd29e453ea47c9fb564e361b7e86647ecf15026f
514 lines · 20099 bytes · View on GitHub

1package main
2
3import (
4 "bytes"
5 "context"
6 "crypto/ed25519"
7 "crypto/rand"
8 "encoding/json"
9 "net/http"
10 "net/http/httptest"
11 "os"
12 "os/exec"
13 "path/filepath"
14 "strings"
15 "sync"
16 "sync/atomic"
17 "testing"
18 "time"
19)
20
21// The real control-plane client against a fake control plane over HTTPS that authenticates the node with the contracts' own verifier.
22
23func TestControlPlane_AuthorizeCarriesOneAssertionBoundToTheRequestAndNothingElse(t *testing.T) {
24 e, cp := httpNode(t, nil, nil)
25 d := e.newDevice()
26 hc := cp.control()
27 resp, err := hc.Authorize(context.Background(), AuthorizeRequest{DeviceID: d.id, SessionJTI: randomHex(12), Reason: "hello", Node: testNodeID})
28 if err != nil {
29 t.Fatal(err)
30 }
31 if resp.DeviceID != d.id || resp.Host != d.host || resp.Status != statusActive || resp.Plan != "paid" || resp.ValidUntil <= e.now().Unix() {
32 t.Fatalf("answer %+v", resp)
33 }
34 cp.mu.Lock()
35 defer cp.mu.Unlock()
36 if len(cp.bodies) != 1 || len(cp.headers) < 1 || cp.refused != 0 {
37 t.Fatalf("authenticated %d requests (%d authorize), refused %d", len(cp.headers), len(cp.bodies), cp.refused)
38 }
39 // The assertion is the node's ONLY credential, on authorize and on the feed alike: no bearer, no cookie, no shared secret, no commerce key.
40 for _, h := range cp.headers {
41 if h.Get("x-all-node-assertion") == "" {
42 t.Error("a request without its assertion was authenticated")
43 }
44 for name := range h {
45 switch strings.ToLower(name) {
46 case "x-all-node-assertion", "content-type", "accept", "user-agent", "content-length", "accept-encoding":
47 default:
48 t.Errorf("a request of the node carries the header %s", name)
49 }
50 }
51 }
52}
53
54func TestControlPlane_AnAssertionServesOneRequestOnly(t *testing.T) {
55 _, cp := httpNode(t, nil, nil)
56 hc := cp.control()
57 body := []byte(`{"v":1}`)
58 token := hc.assertion("POST", "https://"+cp.srv.Listener.Addr().String()+authorizePath, body)
59 cp.mu.Lock()
60 keys := cp.nodeKeys
61 cp.mu.Unlock()
62 url := "https://" + cp.srv.Listener.Addr().String() + authorizePath
63 seen := map[string]bool{}
64 first := ctVerifyNodeAssertion(token, epochSeconds(time.Now()), keys, "POST", url, body, 5, seen)
65 if !first.ok {
66 t.Fatalf("assertion refused: %s", first.err)
67 }
68 seen[first.jti] = true
69 if r := ctVerifyNodeAssertion(token, epochSeconds(time.Now()), keys, "POST", url, body, 5, seen); r.err != "replayed_jti" {
70 t.Fatalf("a replayed assertion: %s", r.err)
71 }
72 // bound to the method, the URL and the exact body bytes
73 if r := ctVerifyNodeAssertion(token, epochSeconds(time.Now()), keys, "GET", url, body, 5, map[string]bool{}); r.err != "wrong_htm" {
74 t.Fatalf("another method: %s", r.err)
75 }
76 if r := ctVerifyNodeAssertion(token, epochSeconds(time.Now()), keys, "POST", url+"?x=1", body, 5, map[string]bool{}); r.err != "wrong_htu" {
77 t.Fatalf("another URL: %s", r.err)
78 }
79 if r := ctVerifyNodeAssertion(token, epochSeconds(time.Now()), keys, "POST", url, []byte(`{"v":2}`), 5, map[string]bool{}); r.err != "wrong_body_hash" {
80 t.Fatalf("another body: %s", r.err)
81 }
82 // 60 s, no more
83 if r := ctVerifyNodeAssertion(token, epochSeconds(time.Now().Add(61*time.Second)), keys, "POST", url, body, 5, map[string]bool{}); r.err != "expired" {
84 t.Fatalf("an old assertion: %s", r.err)
85 }
86 // every call mints a new one (a retry is a new request)
87 if again := hc.assertion("POST", url, body); again == token {
88 t.Fatal("two assertions are the same token")
89 }
90}
91
92func TestControlPlane_ABadNodeSignatureIsRefusedByTheControlPlaneAndTheDeviceIsTurnedAway(t *testing.T) {
93 e, cp := httpNode(t, nil, nil)
94 d := e.newDevice()
95
96 _, rogueKey, _ := ed25519.GenerateKey(rand.Reader)
97 for name, hc := range map[string]*HTTPControl{
98 "a key the control plane does not know": func() *HTTPControl {
99 c, err := NewHTTPControl(HTTPControlConfig{BaseURL: cp.url(), NodeID: testNodeID, KID: cp.nodeKID, Key: rogueKey, Client: cp.srv.Client()})
100 if err != nil {
101 t.Fatal(err)
102 }
103 return c
104 }(),
105 "a kid it does not know": func() *HTTPControl {
106 c, err := NewHTTPControl(HTTPControlConfig{BaseURL: cp.url(), NodeID: testNodeID, KID: "node-t1-revoked", Key: cp.nodeKey, Client: cp.srv.Client()})
107 if err != nil {
108 t.Fatal(err)
109 }
110 return c
111 }(),
112 "another node's id": func() *HTTPControl {
113 c, err := NewHTTPControl(HTTPControlConfig{BaseURL: cp.url(), NodeID: "t2", KID: cp.nodeKID, Key: cp.nodeKey, Client: cp.srv.Client()})
114 if err != nil {
115 t.Fatal(err)
116 }
117 return c
118 }(),
119 "a clock a minute off": func() *HTTPControl {
120 c, err := NewHTTPControl(HTTPControlConfig{BaseURL: cp.url(), NodeID: testNodeID, KID: cp.nodeKID, Key: cp.nodeKey, Client: cp.srv.Client(),
121 Now: func() time.Time { return e.now().Add(-2 * time.Minute) }})
122 if err != nil {
123 t.Fatal(err)
124 }
125 return c
126 }(),
127 } {
128 _, err := hc.Authorize(context.Background(), AuthorizeRequest{DeviceID: d.id, SessionJTI: randomHex(12), Reason: "hello", Node: testNodeID})
129 ce, ok := err.(*ControlError)
130 if !ok || ce.Status != 401 || ce.Code != "unauthorized" {
131 t.Errorf("%s: %v", name, err)
132 }
133 if _, err := hc.Feed(context.Background(), 0, 0); err == nil {
134 t.Errorf("%s: the feed answered", name)
135 }
136 }
137 if e.cp.callsFor(d.id, "") != 0 {
138 t.Fatal("the control plane decided something for a request it could not authenticate")
139 }
140
141 // A node whose key was revoked (its kid removed from TUNNEL_NODE_KEYS) cannot authorize anyone: the device is turned away with a closed word.
142 cp.mu.Lock()
143 delete(cp.nodeKeys, cp.nodeKID)
144 cp.mu.Unlock()
145 dc := e.connect(d)
146 line := dc.hello(func() string { return e.session(d, nil) }, func(s, n string) string { return e.proof(d, s, n, nil) })
147 if line != "ERR control_unavailable" {
148 t.Fatalf("a device behind a node the control plane refuses must be told to try later, not that it is wrong: %q", line)
149 }
150 dc.expectClosed(2 * time.Second)
151 if e.srv.Online(d.host) {
152 t.Fatal("the host is served although nothing authorized it")
153 }
154}
155
156func TestControlPlane_TheNodeHoldsNoSecretButItsOwnKey(t *testing.T) {
157 // The production binary's whole configuration surface: one key file, no shared secret, no token, no commerce key.
158 bin := productionBinary(t)
159 out, _ := exec.Command(bin, "-h").CombinedOutput()
160 help := strings.ToLower(string(out))
161 for _, word := range []string{"secret", "token", "hmac", "commerce", "password", "bearer", "api-key", "apikey"} {
162 if strings.Contains(help, word) {
163 t.Errorf("the node's flags mention %q:\n%s", word, out)
164 }
165 }
166 if !strings.Contains(help, "-node-key-file") {
167 t.Fatalf("no node key flag:\n%s", out)
168 }
169 for _, name := range []string{"TUNNEL_COMMERCE_KEY", "TUNNEL_SECRET", "TUNNEL_TOKEN", "COMMERCE_HMAC_KEY"} {
170 if strings.Contains(string(out), name) {
171 t.Errorf("the node reads %s", name)
172 }
173 }
174}
175
176func TestControlPlane_ErrorsAreClosedWordsAndNeverTextFromTheNetwork(t *testing.T) {
177 e, cp := httpNode(t, nil, nil)
178 d := e.newDevice()
179 hc := cp.control()
180 ask := func() error {
181 _, err := hc.Authorize(context.Background(), AuthorizeRequest{DeviceID: d.id, SessionJTI: randomHex(12), Reason: "renew", Node: testNodeID})
182 return err
183 }
184 code := func(err error) (int, string) {
185 ce, ok := err.(*ControlError)
186 if !ok {
187 return 0, err.Error()
188 }
189 return ce.Status, ce.Code
190 }
191
192 e.cp.set(d.id, &devicePlan{host: d.host, err: &ControlError{Status: 402, Code: codeLicenseInactive}})
193 if st, w := code(ask()); st != 402 || w != codeLicenseInactive {
194 t.Errorf("402: %d %s", st, w)
195 }
196 e.cp.set(d.id, &devicePlan{host: d.host, err: &ControlError{Status: 401, Code: codeDeviceRevoked}})
197 if st, w := code(ask()); st != 401 || w != codeDeviceRevoked {
198 t.Errorf("401: %d %s", st, w)
199 }
200 cp.setDown(503)
201 if st, w := code(ask()); st != 503 || w != "end_to_end_unavailable" || errForControl(ask()) != codeControlUnavailable {
202 t.Errorf("503: %d %s", st, w)
203 }
204 cp.setDown(0)
205
206 // A control plane that answers with something that is not an error body: bad_response, whatever the text says.
207 for name, serve := range map[string]http.HandlerFunc{
208 "html": func(w http.ResponseWriter, r *http.Request) {
209 w.WriteHeader(502)
210 _, _ = w.Write([]byte("<html>Bad gateway: secret-detail</html>"))
211 },
212 "text word": func(w http.ResponseWriter, r *http.Request) {
213 w.WriteHeader(500)
214 _, _ = w.Write([]byte(`{"ok":false,"error":"Not A Word!"}`))
215 },
216 "200 junk": func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte(`{"ok":true}`)) },
217 "200 other device": func(w http.ResponseWriter, r *http.Request) {
218 a := AuthorizeResponse{DeviceID: randomHex(16), Label: label20(), Status: statusActive, Plan: "paid"}
219 a.Host = a.Label + "." + testZone
220 a.ValidUntil, a.ServerTime = e.now().Unix()+3600, e.now().Unix()
221 a.Policy = okPolicy{20, 45, 1800, 14400, 300}
222 _, _ = w.Write([]byte(canonicalJSON(a.toMap())))
223 },
224 "too large": func(w http.ResponseWriter, r *http.Request) {
225 _, _ = w.Write(bytes.Repeat([]byte("x"), maxControlBody+10))
226 },
227 } {
228 srv := httptest.NewTLSServer(serve)
229 other, err := NewHTTPControl(HTTPControlConfig{BaseURL: srv.URL, NodeID: testNodeID, KID: cp.nodeKID, Key: cp.nodeKey, Client: srv.Client()})
230 if err != nil {
231 t.Fatal(err)
232 }
233 _, aerr := other.Authorize(context.Background(), AuthorizeRequest{DeviceID: d.id, SessionJTI: randomHex(12), Reason: "hello", Node: testNodeID})
234 srv.Close()
235 if aerr == nil {
236 // the answer for another device parses; the node's own check (checkAuthorize) is what refuses it, below
237 if name != "200 other device" {
238 t.Errorf("%s: accepted", name)
239 }
240 continue
241 }
242 if ce, ok := aerr.(*ControlError); ok && strings.Contains(ce.Code, "secret") {
243 t.Errorf("%s: text from the network became a word: %q", name, ce.Code)
244 }
245 if errForControl(aerr) != codeControlUnavailable && name != "text word" {
246 t.Errorf("%s: %q", name, errForControl(aerr))
247 }
248 }
249}
250
251func TestControlPlane_ARedirectIsNeverFollowed(t *testing.T) {
252 e, cp := httpNode(t, nil, nil)
253 d := e.newDevice()
254 var landed atomic.Int64
255 elsewhere := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
256 landed.Add(1)
257 _, _ = w.Write([]byte("{}"))
258 }))
259 defer elsewhere.Close()
260 cp.mu.Lock()
261 cp.redirect = elsewhere.URL + authorizePath
262 cp.mu.Unlock()
263 _, err := cp.control().Authorize(context.Background(), AuthorizeRequest{DeviceID: d.id, SessionJTI: randomHex(12), Reason: "hello", Node: testNodeID})
264 if err == nil || landed.Load() != 0 {
265 t.Fatalf("a redirect was followed: err=%v landed=%d", err, landed.Load())
266 }
267}
268
269func TestControlPlane_AnAnswerThatIsNotWhatWasAskedIsNeverRelayed(t *testing.T) {
270 // The control plane answers for another device: the node refuses to relay it (checkAuthorize) and the device is told control_unavailable.
271 e, _ := httpNode(t, nil, nil)
272 d := e.newDevice()
273 e.cp.set(d.id, &devicePlan{host: d.host, answer: func(int) AuthorizeResponse {
274 a := AuthorizeResponse{DeviceID: randomHex(16), Label: label20(), Status: statusActive, Plan: "paid", Nodes: []any{}}
275 a.Host = a.Label + "." + testZone
276 a.ValidUntil, a.ServerTime = e.now().Unix()+3600, e.now().Unix()
277 a.Policy = okPolicy{20, 45, 1800, 14400, 300}
278 return a
279 }})
280 dc := e.connect(d)
281 line := dc.hello(func() string { return e.session(d, nil) }, func(s, n string) string { return e.proof(d, s, n, nil) })
282 if line != "ERR control_unavailable" || e.srv.Online(d.host) {
283 t.Fatalf("an answer for another device: %q", line)
284 }
285}
286
287func TestControlPlane_TheHostNeverComesFromTheRequest(t *testing.T) {
288 e, cp := httpNode(t, nil, nil)
289 d := e.newDevice()
290 if _, err := cp.control().Authorize(context.Background(), AuthorizeRequest{DeviceID: d.id, SessionJTI: randomHex(12), Reason: "hello", Node: testNodeID}); err != nil {
291 t.Fatal(err)
292 }
293 cp.mu.Lock()
294 bodies := append([][]byte(nil), cp.bodies...)
295 cp.mu.Unlock()
296 if len(bodies) != 1 {
297 t.Fatalf("bodies %d", len(bodies))
298 }
299 doc, perr := parseJSON(string(bodies[0]))
300 if perr != nil {
301 t.Fatal(perr)
302 }
303 m := doc.(map[string]any)
304 if len(m) != 5 || m["host"] != nil || m["label"] != nil {
305 t.Fatalf("the authorize body %s", bodies[0])
306 }
307 if errs := ctSchemaSet.validate("authorize-request.schema.json", doc); len(errs) > 0 {
308 t.Fatalf("not an authorize-request: %v", errs)
309 }
310}
311
312func TestControlPlane_ConfigurationIsValidated(t *testing.T) {
313 _, priv, _ := ed25519.GenerateKey(rand.Reader)
314 good := HTTPControlConfig{BaseURL: "https://api.example.test", NodeID: "t1", KID: "k1", Key: priv}
315 if _, err := NewHTTPControl(good); err != nil {
316 t.Fatal(err)
317 }
318 for name, mutate := range map[string]func(*HTTPControlConfig){
319 "plain http": func(c *HTTPControlConfig) { c.BaseURL = "http://api.example.test" },
320 "a path": func(c *HTTPControlConfig) { c.BaseURL = "https://api.example.test/v1" },
321 "credentials": func(c *HTTPControlConfig) { c.BaseURL = "https://user:pw" + "@api.example.test" },
322 "a query": func(c *HTTPControlConfig) { c.BaseURL = "https://api.example.test?x=1" },
323 "no host": func(c *HTTPControlConfig) { c.BaseURL = "https://" },
324 "bad node id": func(c *HTTPControlConfig) { c.NodeID = "T1" },
325 "bad kid": func(c *HTTPControlConfig) { c.KID = "" },
326 "no key": func(c *HTTPControlConfig) { c.Key = nil },
327 "a short key": func(c *HTTPControlConfig) { c.Key = priv[:10] },
328 "garbage": func(c *HTTPControlConfig) { c.BaseURL = "::not a url" },
329 "another scheme": func(c *HTTPControlConfig) { c.BaseURL = "ftp://api.example.test" },
330 "whitespace host": func(c *HTTPControlConfig) { c.BaseURL = "https://api example.test" },
331 } {
332 c := good
333 mutate(&c)
334 if _, err := NewHTTPControl(c); err == nil {
335 t.Errorf("%s: accepted", name)
336 }
337 }
338 // the port 443 is not part of the URL the control plane's runtime sees
339 c := good
340 c.BaseURL = "https://api.example.test:443"
341 hc, err := NewHTTPControl(c)
342 if err != nil || hc.hostHTU != "api.example.test" {
343 t.Fatalf("hostHTU %q err %v", hc.hostHTU, err)
344 }
345 c.BaseURL = "https://api.example.test:8443"
346 if hc, _ = NewHTTPControl(c); hc.hostHTU != "api.example.test:8443" {
347 t.Fatalf("hostHTU %q", hc.hostHTU)
348 }
349}
350
351func TestNodeKey_FileRulesAndTheEntryThatRegistersIt(t *testing.T) {
352 dir := t.TempDir()
353 path := filepath.Join(dir, "node.key")
354 key, err := GenerateNodeKey(path)
355 if err != nil {
356 t.Fatal(err)
357 }
358 st, _ := os.Stat(path)
359 if st.Mode().Perm() != 0o600 {
360 t.Fatalf("mode %v", st.Mode())
361 }
362 if _, err := GenerateNodeKey(path); err == nil {
363 t.Fatal("a key was overwritten")
364 }
365 loaded, err := LoadNodeKey(path)
366 if err != nil || !loaded.Equal(key) {
367 t.Fatalf("load: %v", err)
368 }
369 // a key that others can read is refused, as ssh refuses it
370 if err := os.Chmod(path, 0o644); err != nil {
371 t.Fatal(err)
372 }
373 if _, err := LoadNodeKey(path); err == nil {
374 t.Fatal("a world-readable key was accepted")
375 }
376 // systemd's credential copy is 0440 with an ACL for the unit's user: accepted only inside $CREDENTIALS_DIRECTORY, and never world-readable
377 if err := os.Chmod(path, 0o440); err != nil {
378 t.Fatal(err)
379 }
380 if _, err := LoadNodeKey(path); err == nil {
381 t.Fatal("a group-readable key outside the credentials directory was accepted")
382 }
383 t.Setenv("CREDENTIALS_DIRECTORY", dir)
384 if loaded, err := LoadNodeKey(path); err != nil || !loaded.Equal(key) {
385 t.Fatalf("the systemd credential copy was refused: %v", err)
386 }
387 if err := os.Chmod(path, 0o444); err != nil {
388 t.Fatal(err)
389 }
390 if _, err := LoadNodeKey(path); err == nil {
391 t.Fatal("a world-readable key in the credentials directory was accepted")
392 }
393 t.Setenv("CREDENTIALS_DIRECTORY", filepath.Join(dir, "elsewhere"))
394 if err := os.Chmod(path, 0o440); err != nil {
395 t.Fatal(err)
396 }
397 if _, err := LoadNodeKey(path); err == nil {
398 t.Fatal("a group-readable key outside the credentials directory was accepted")
399 }
400 t.Setenv("CREDENTIALS_DIRECTORY", "")
401 _ = os.Chmod(path, 0o600)
402 for name, content := range map[string]string{"garbage": "not pem", "a certificate": "-----BEGIN CERTIFICATE-----\nAAAA\n-----END CERTIFICATE-----\n"} {
403 bad := filepath.Join(dir, name)
404 if err := os.WriteFile(bad, []byte(content), 0o600); err != nil {
405 t.Fatal(err)
406 }
407 if _, err := LoadNodeKey(bad); err == nil {
408 t.Errorf("%s: accepted", name)
409 }
410 }
411 if _, err := LoadNodeKey(filepath.Join(dir, "missing")); err == nil {
412 t.Fatal("a missing key was accepted")
413 }
414
415 // the entry is a valid TUNNEL_NODE_KEYS document of the contracts, for exactly this key
416 entry := NodeKeysEntry("t1", "node-t1-1", key)
417 doc, perr := parseJSON(entry)
418 if perr != nil {
419 t.Fatal(perr)
420 }
421 keys, word := ctParseNodeKeys(doc)
422 if word != "" || keys["node-t1-1"].node != "t1" || keys["node-t1-1"].x != b64Encode(key.Public().(ed25519.PublicKey)) {
423 t.Fatalf("entry %s -> %v %q", entry, keys, word)
424 }
425 var round map[string]any
426 if json.Unmarshal([]byte(entry), &round) != nil {
427 t.Fatal("entry is not JSON")
428 }
429}
430
431func TestProductionBinary_InitPrintsTheRegistrationEntryAndRefusesWithoutAKey(t *testing.T) {
432 bin := productionBinary(t)
433 dir := t.TempDir()
434 keyPath := filepath.Join(dir, "node.key")
435 run := func(args ...string) (string, int) {
436 cmd := exec.Command(bin, args...)
437 var out, errb bytes.Buffer
438 cmd.Stdout, cmd.Stderr = &out, &errb
439 err := cmd.Run()
440 code := 0
441 if ee, ok := err.(*exec.ExitError); ok {
442 code = ee.ExitCode()
443 }
444 return out.String() + errb.String(), code
445 }
446 out, code := run("-node-id", "t1", "-node-kid", "node-t1-1", "-node-key-file", keyPath, "-node-key-init")
447 if code != 0 {
448 t.Fatalf("init: %d %s", code, out)
449 }
450 doc, err := parseJSON(strings.TrimSpace(out))
451 if err != nil {
452 t.Fatalf("init printed %q", out)
453 }
454 if _, word := ctParseNodeKeys(doc); word != "" {
455 t.Fatalf("init printed an entry the contracts refuse: %s", word)
456 }
457 if out2, code := run("-node-id", "t1", "-node-kid", "node-t1-1", "-node-key-file", keyPath, "-node-key-print"); code != 0 || strings.TrimSpace(out2) != strings.TrimSpace(out) {
458 t.Fatalf("print: %d %q", code, out2)
459 }
460 if _, code := run("-node-id", "t1", "-node-kid", "node-t1-1", "-node-key-file", keyPath, "-node-key-init"); code == 0 {
461 t.Fatal("init overwrote the key")
462 }
463 if out, code := run("-node-id", "t1", "-node-host", testNodeHost); code == 0 || !strings.Contains(out, "signing key") {
464 t.Fatalf("without a key: %d %s", code, out)
465 }
466 if out, code := run("-node-id", "t1", "-node-host", testNodeHost, "-node-key-file", keyPath, "-node-kid", "node-t1-1", "-control-url", "http://plain.example.test"); code == 0 || !strings.Contains(out, "https") {
467 t.Fatalf("plain http control plane: %d %s", code, out)
468 }
469}
470
471// ---------------------------------------------------------------------------------------------------------------------------- authorize concurrency over HTTP
472
473func TestControlPlane_AuthorizeNeverRunsMoreThanFiftyAtOnceOverHTTP(t *testing.T) {
474 if policyData.Authorization.AuthorizeMaxConcur != 50 {
475 t.Fatalf("the plan says 50 concurrent authorizations, the policy says %d", policyData.Authorization.AuthorizeMaxConcur)
476 }
477 e, cp := httpNode(t, nil, func(c *Config) { c.HelloTimeout, c.FirstLineTimeout = 10*time.Second, 20*time.Second })
478 cp.mu.Lock()
479 cp.delay = 80 * time.Millisecond
480 cp.mu.Unlock()
481 const devices = 160
482 var all []*testDevice
483 for i := 0; i < devices; i++ {
484 all = append(all, e.newDevice())
485 }
486 var connected atomic.Int64
487 var wg sync.WaitGroup
488 for _, d := range all {
489 wg.Add(1)
490 go func(d *testDevice) {
491 defer wg.Done()
492 // (a burst this size can overflow the listener's backlog: a device that is reset tries again, as one does)
493 for attempt := 0; attempt < 40; attempt++ {
494 line := helloOnce(e, d)
495 if strings.HasPrefix(line, "OK ") {
496 connected.Add(1)
497 return
498 }
499 if line != "ERR control_unavailable" && !strings.HasPrefix(line, "unreachable") {
500 t.Errorf("a device was refused: %q", line)
501 return
502 }
503 time.Sleep(time.Duration(20+attempt*5) * time.Millisecond)
504 }
505 }(d)
506 }
507 wg.Wait()
508 if connected.Load() != devices {
509 t.Fatalf("%d of %d connected", connected.Load(), devices)
510 }
511 if peak := cp.maxInflight(); peak > 50 || peak < 2 {
512 t.Fatalf("the control plane saw %d authorizations at once; the node's bound is 50", peak)
513 }
514}