ALL

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

tunnel/feed_test.go

SHA-256: 806cdb6bd55ab1b38da418b9762e2d9817688b63f608b1e00b194fdedffd56c7
962 líneas · 34338 bytes · Ver en GitHub

1package main
2
3import (
4 "context"
5 "errors"
6 "net"
7 "strings"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12)
13
14// The feed consumer: what the control plane says the node must stop serving, applied within the deadline, once, in order.
15
16// scriptedFeed is a feed the tests write by hand. By default it answers like the control plane's: the events after the cursor, in order, with the head
17// as the next cursor; a test overrides what it answers to feed the node duplicates, disorder and nonsense.
18type scriptedFeed struct {
19 mu sync.Mutex
20 events []FeedEvent
21 head int64
22 fail error
23 custom func(since int64) FeedAnswer
24 cursor []int64 // the cursor of every poll
25 block chan struct{}
26}
27
28func (f *scriptedFeed) push(events ...FeedEvent) {
29 f.mu.Lock()
30 defer f.mu.Unlock()
31 for _, e := range events {
32 if e.Epoch == 0 {
33 f.head++
34 e.Epoch = f.head
35 }
36 f.head = max(f.head, e.Epoch)
37 f.events = append(f.events, e)
38 }
39}
40
41func (f *scriptedFeed) set(custom func(since int64) FeedAnswer) {
42 f.mu.Lock()
43 f.custom = custom
44 f.mu.Unlock()
45}
46
47func (f *scriptedFeed) polls() []int64 {
48 f.mu.Lock()
49 defer f.mu.Unlock()
50 return append([]int64(nil), f.cursor...)
51}
52
53func (f *scriptedFeed) Feed(ctx context.Context, since int64, wait time.Duration) (FeedAnswer, error) {
54 f.mu.Lock()
55 f.cursor = append(f.cursor, since)
56 fail, custom, block := f.fail, f.custom, f.block
57 f.mu.Unlock()
58 if block != nil {
59 select {
60 case <-block:
61 case <-ctx.Done():
62 return FeedAnswer{}, ctx.Err()
63 }
64 }
65 if fail != nil {
66 return FeedAnswer{}, fail
67 }
68 if custom != nil {
69 return custom(since), nil
70 }
71 f.mu.Lock()
72 defer f.mu.Unlock()
73 answer := FeedAnswer{Epoch: f.head, ServerTime: time.Now().Unix()}
74 for _, e := range f.events {
75 if e.Epoch > since {
76 answer.Events = append(answer.Events, e)
77 }
78 }
79 return answer, nil
80}
81
82func (e *testEnv) event(kind string, d *testDevice) FeedEvent {
83 ev := FeedEvent{Type: kind, At: e.now().Unix()}
84 if d != nil {
85 ev.DeviceID, ev.Label = d.id, d.label
86 }
87 return ev
88}
89
90// withFeed is a node on the real clock whose feed is f, polled every 20 ms.
91func withFeed(t *testing.T, f *scriptedFeed) *testEnv {
92 return newTestEnv(t, func(c *Config) {
93 c.Feed = f
94 c.FeedPoll = 20 * time.Millisecond
95 })
96}
97
98// streamOpen opens a client stream of d and sends one request over it, leaving the connection open.
99func streamOpen(t *testing.T, e *testEnv, d *testDevice) *tlsClient {
100 t.Helper()
101 c, err := e.clientTLS(d)
102 if err != nil {
103 t.Fatal(err)
104 }
105 if resp, err := postOver(c, d.host, []byte("hello")); err != nil || string(resp) != "echo:hello" {
106 t.Fatalf("a stream of %s: %q %v", d.host, resp, err)
107 }
108 return &tlsClient{Conn: c}
109}
110
111type tlsClient struct{ net.Conn }
112
113// dead reports whether the node cut the stream within timeout.
114func (c *tlsClient) dead(timeout time.Duration) bool {
115 _ = c.SetReadDeadline(time.Now().Add(timeout))
116 buf := make([]byte, 1)
117 _, err := c.Read(buf)
118 if err == nil {
119 return false
120 }
121 var ne net.Error
122 return !(errors.As(err, &ne) && ne.Timeout())
123}
124
125// ---------------------------------------------------------------------------------------------------------------------------- the deadline
126
127func TestFeed_ARevokeClosesTheControlConnectionAndTheStreamsWithinTheApplyDeadline(t *testing.T) {
128 if policyData.Timing.FeedPollS != 15 || policyData.Timing.FeedApplyDeadlineS != 20 {
129 t.Fatalf("the plan says a poll every 15 s and an apply within 20 s: policy %d and %d", policyData.Timing.FeedPollS, policyData.Timing.FeedApplyDeadlineS)
130 }
131 clk := newFakeClock()
132 e, cp := httpNode(t, clk, nil)
133 d, bystander := e.newDevice(), e.newDevice()
134 dc, bc := e.connectOK(d), e.connectOK(bystander)
135 ds := dc.serve(nil)
136 bs := bc.serve(nil)
137 stream := streamOpen(t, e, d)
138 other := streamOpen(t, e, bystander)
139 defer other.Close()
140
141 cp.addEvent("revoke", d)
142 // Virtual time passes second by second; the node must have acted before the deadline.
143 elapsed := 0
144 for ; elapsed <= policyData.Timing.FeedApplyDeadlineS; elapsed++ {
145 if !e.srv.Online(d.host) {
146 break
147 }
148 clk.Advance(time.Second)
149 time.Sleep(15 * time.Millisecond)
150 }
151 if elapsed > policyData.Timing.FeedApplyDeadlineS {
152 t.Fatalf("the node had not applied the revoke after %d s", elapsed)
153 }
154 eventually(t, "CLOSE revoked", 2*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE revoked") })
155 <-ds.done
156 if !stream.dead(2 * time.Second) {
157 t.Fatal("the stream of a revoked device is still open")
158 }
159 // nothing else was touched
160 if !e.srv.Online(bystander.host) || other.dead(100*time.Millisecond) || len(bs.otherLines()) != 0 {
161 t.Fatalf("another device was affected: lines %q", bs.otherLines())
162 }
163 // and it is not served again: the control plane would refuse its next HELLO
164 e.cp.set(d.id, nil)
165 if _, err := e.clientTLS(d); err == nil {
166 t.Fatal("a client reached a revoked device")
167 }
168}
169
170func TestFeed_ALongPollDeliversARevokeTheMomentItHappens(t *testing.T) {
171 e, cp := httpNode(t, nil, nil)
172 cp.mu.Lock()
173 cp.hold = 10 * time.Second
174 cp.mu.Unlock()
175 d := e.newDevice()
176 dc := e.connectOK(d)
177 ds := dc.serve(nil)
178 stream := streamOpen(t, e, d)
179 eventually(t, "a held poll", 3*time.Second, func() bool {
180 for _, c := range cp.feedCalls() {
181 if c.wait == int64(policyData.Timing.FeedPollS) {
182 return true
183 }
184 }
185 return false
186 })
187 started := time.Now()
188 cp.addEvent("revoke", d)
189 eventually(t, "CLOSE revoked", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE revoked") })
190 if took := time.Since(started); took > 2*time.Second {
191 t.Fatalf("a held poll delivered the revoke after %s", took)
192 }
193 if !stream.dead(2 * time.Second) {
194 t.Fatal("stream alive")
195 }
196 // The poll the node signed: the exact URL, the cursor and the hold.
197 for _, c := range cp.feedCalls() {
198 if !strings.HasPrefix(c.url, "/v1/tunnel/feed?since=") {
199 t.Errorf("feed URL %q", c.url)
200 }
201 }
202 first := cp.feedCalls()[0]
203 if first.since != 0 || first.wait != 0 {
204 t.Fatalf("the catch-up poll is the first, from the start and without a hold: %+v", first)
205 }
206}
207
208func TestFeed_PollsOncePerPeriodAndAdvancesItsCursor(t *testing.T) {
209 clk := newFakeClock()
210 e, cp := httpNode(t, clk, nil)
211 d := e.newDevice()
212 _ = e.connectOK(d)
213 // the catch-up poll, then the first long poll right after it (a control plane that does not hold it answers at once)
214 eventually(t, "the catch-up and the first long poll", 3*time.Second, func() bool { return len(cp.feedCalls()) == 2 })
215 time.Sleep(50 * time.Millisecond)
216 if n := len(cp.feedCalls()); n != 2 {
217 t.Fatalf("the node polled %d times without a period passing", n)
218 }
219 other := &testDevice{id: randomHex(16), label: label20()}
220 cp.addEvent("revoke", other) // nobody here: the cursor still moves
221 step := func(want int) {
222 t.Helper()
223 for i := 0; i < 20 && len(cp.feedCalls()) < want; i++ {
224 clk.Advance(time.Second)
225 time.Sleep(15 * time.Millisecond)
226 }
227 if n := len(cp.feedCalls()); n != want {
228 t.Fatalf("%d polls, want %d", n, want)
229 }
230 }
231 step(4) // the poll that brings the event, and the long poll that follows it at once
232 step(5)
233 calls := cp.feedCalls()
234 wantSince := []int64{0, 0, 0, 1, 1}
235 for i, c := range calls {
236 if c.since != wantSince[i] {
237 t.Errorf("poll %d asked since=%d, want %d", i, c.since, wantSince[i])
238 }
239 wantWait := int64(policyData.Timing.FeedPollS)
240 if i == 0 {
241 wantWait = 0
242 }
243 if c.wait != wantWait {
244 t.Errorf("poll %d asked wait=%d, want %d", i, c.wait, wantWait)
245 }
246 }
247}
248
249// ---------------------------------------------------------------------------------------------------------------------------- what each event does
250
251func TestFeed_RevokeReleaseAndLicenseEndCloseTheConnectionAndKillTheStreams(t *testing.T) {
252 for kind, want := range map[string]string{feedRevoke: "CLOSE revoked", feedRelease: "CLOSE released", feedLicenseEnd: "CLOSE revoked"} {
253 t.Run(kind, func(t *testing.T) {
254 f := &scriptedFeed{}
255 e := withFeed(t, f)
256 d := e.newDevice()
257 dc := e.connectOK(d)
258 ds := dc.serve(nil)
259 stream := streamOpen(t, e, d)
260 f.push(e.event(kind, d))
261 eventually(t, want, 3*time.Second, func() bool { return containsLine(ds.otherLines(), want) })
262 if !stream.dead(2 * time.Second) {
263 t.Fatal("the stream is still open")
264 }
265 eventually(t, "the host unbound", 2*time.Second, func() bool { return !e.srv.Online(d.host) })
266 e.waitIdle()
267 })
268 }
269}
270
271func TestFeed_AStreamOfAHostWhoseConnectionIsAlreadyGoneIsStillKilled(t *testing.T) {
272 f := &scriptedFeed{}
273 e := withFeed(t, f)
274 d := e.newDevice()
275 dc := e.connectOK(d)
276 _ = dc.serve(nil)
277 stream := streamOpen(t, e, d)
278 _ = dc.c.Close() // the device's control connection drops, its data connection and the client's stream are still there
279 eventually(t, "the control connection gone", 2*time.Second, func() bool { return e.srv.routes.controls(d.label) == nil })
280 if stream.dead(100 * time.Millisecond) {
281 t.Fatal("a stream was cut because the control connection dropped: it keeps to its own limits")
282 }
283 f.push(e.event(feedRevoke, d))
284 if !stream.dead(3 * time.Second) {
285 t.Fatal("the revoke did not reach a stream that outlived its control connection")
286 }
287}
288
289func TestFeed_APauseStopsNewStreamsKillsTheOnesInFlightAndForwardsTheSignedStatement(t *testing.T) {
290 const pausedStatement = "eyJhbGciOiJFZERTQSJ9.eyJzdGF0ZSI6InBhdXNlZCJ9.c2lnbmF0dXJl"
291 const resumedStatement = "eyJhbGciOiJFZERTQSJ9.eyJzdGF0ZSI6InJlc3VtZWQifQ.c2lnbmF0dXJl"
292 f := &scriptedFeed{}
293 e := withFeed(t, f)
294 d := e.newDevice()
295 dc := e.connectOK(d)
296 ds := dc.serve(nil)
297 stream := streamOpen(t, e, d)
298
299 answer := func(status, access string) {
300 e.cp.set(d.id, &devicePlan{host: d.host, answer: func(int) AuthorizeResponse {
301 now := e.now()
302 return AuthorizeResponse{DeviceID: d.id, Label: d.label, Host: d.host, Status: status, Plan: "paid", ValidUntil: now.Unix() + 3600, ServerTime: now.Unix(),
303 Policy: okPolicy{20, 45, 1800, 14400, 300}, Nodes: []any{}, Access: access}
304 }})
305 }
306 answer(statusPaused, pausedStatement)
307 f.push(e.event(feedPause, d))
308 eventually(t, "the signed statement of the pause", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "STATEMENT "+pausedStatement) })
309 if !stream.dead(2 * time.Second) {
310 t.Fatal("a stream of a paused device survived the pause")
311 }
312 if e.srv.Online(d.host) {
313 t.Fatal("a paused device is online")
314 }
315 if _, err := e.clientTLS(d); err == nil {
316 t.Fatal("a client got a stream of a paused device")
317 }
318 if !dc.try0() {
319 t.Fatal("the control connection of a paused device must stay (the resume arrives as a statement)")
320 }
321 if n := e.cp.callsFor(d.id, "renew"); n != 1 {
322 t.Fatalf("one refresh was expected, the control plane saw %d", n)
323 }
324
325 answer(statusActive, resumedStatement)
326 f.push(e.event(feedResume, d))
327 eventually(t, "the statement of the resume", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "STATEMENT "+resumedStatement) })
328 eventually(t, "streams again", 2*time.Second, func() bool { return e.srv.Online(d.host) })
329 c, err := e.clientTLS(d)
330 if err != nil {
331 t.Fatalf("a client after the resume: %v", err)
332 }
333 _ = c.Close()
334 e.checkTranscript()
335}
336
337// try0 reports whether the device's connection is still up: nothing closed it.
338func (dc *devConn) try0() bool {
339 select {
340 case _, ok := <-dc.feed:
341 return ok
342 case <-time.After(150 * time.Millisecond):
343 return true
344 }
345}
346
347func TestFeed_APausedConnectionStaysPausedUntilTheControlPlaneAnswers(t *testing.T) {
348 f := &scriptedFeed{}
349 e := withFeed(t, f)
350 d := e.newDevice()
351 dc := e.connectOK(d)
352 ds := dc.serve(nil)
353
354 // the control plane cannot be asked right now
355 e.cp.set(d.id, &devicePlan{host: d.host, err: &ControlError{Status: 503, Code: "end_to_end_unavailable"}})
356 f.push(e.event(feedPause, d))
357 eventually(t, "paused locally, without waiting for an answer", 3*time.Second, func() bool { return !e.srv.Online(d.host) })
358 f.push(e.event(feedResume, d))
359 time.Sleep(300 * time.Millisecond)
360 if e.srv.Online(d.host) {
361 t.Fatal("the node lifted a pause by itself, with no answer from the control plane")
362 }
363 // the answer comes: the pause is lifted
364 e.cp.set(d.id, &devicePlan{host: d.host, validFor: time.Hour})
365 eventually(t, "the refresh after the control plane is back", 6*time.Second, func() bool { return e.srv.Online(d.host) })
366 if e.cp.callsFor(d.id, "renew") < 2 {
367 t.Fatal("the node did not try again")
368 }
369 _ = ds
370}
371
372func TestFeed_AResumeThatTheControlPlaneRefusesCutsTheConnection(t *testing.T) {
373 f := &scriptedFeed{}
374 e := withFeed(t, f)
375 d := e.newDevice()
376 dc := e.connectOK(d)
377 ds := dc.serve(nil)
378 e.cp.set(d.id, &devicePlan{host: d.host, err: &ControlError{Status: 402, Code: codeLicenseInactive}})
379 f.push(e.event(feedResume, d))
380 eventually(t, "CLOSE revoked", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE revoked") })
381}
382
383func TestFeed_TheKillSwitchClosesEverythingAndNobodyIsServedUntilItIsCleared(t *testing.T) {
384 f := &scriptedFeed{}
385 e := withFeed(t, f)
386 a, b := e.newDevice(), e.newDevice()
387 ca, cb := e.connectOK(a), e.connectOK(b)
388 da, db := ca.serve(nil), cb.serve(nil)
389 sa := streamOpen(t, e, a)
390 sb := streamOpen(t, e, b)
391
392 f.push(FeedEvent{Type: feedKill, At: e.now().Unix()})
393 for _, ds := range []*devServer{da, db} {
394 ds := ds
395 eventually(t, "CLOSE killed", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE killed") })
396 }
397 if !sa.dead(2*time.Second) || !sb.dead(2*time.Second) {
398 t.Fatal("a stream survived the kill switch")
399 }
400 calls := len(e.cp.callLog())
401 c := e.connect(a)
402 line := c.hello(func() string { return e.session(a, nil) }, func(s, n string) string { return e.proof(a, s, n, nil) })
403 if line != "ERR control_unavailable" {
404 t.Fatalf("a HELLO while the switch is on: %q", line)
405 }
406 if len(e.cp.callLog()) != calls {
407 t.Fatal("the node asked the control plane about a device while the kill switch is on")
408 }
409
410 f.push(FeedEvent{Type: feedKillClear, At: e.now().Unix()})
411 eventually(t, "served again", 3*time.Second, func() bool {
412 cc := e.connect(a)
413 l := cc.hello(func() string { return e.session(a, nil) }, func(s, n string) string { return e.proof(a, s, n, nil) })
414 return strings.HasPrefix(l, "OK ")
415 })
416}
417
418func TestFeed_TheKillSwitchSeenInAnAuthorizeAnswerIsTheSameAsTheEvent(t *testing.T) {
419 e := newTestEnv(t, nil)
420 a, b := e.newDevice(), e.newDevice()
421 ca := e.connectOK(a)
422 da := ca.serve(nil)
423 e.cp.set(b.id, &devicePlan{host: b.host, answer: func(int) AuthorizeResponse {
424 now := e.now()
425 return AuthorizeResponse{DeviceID: b.id, Label: b.label, Host: b.host, Status: statusActive, Plan: "paid", ValidUntil: now.Unix() + 3600, ServerTime: now.Unix(),
426 Policy: okPolicy{20, 45, 1800, 14400, 300}, Nodes: []any{}, E2EKill: true}
427 }})
428 cb := e.connect(b)
429 if line := cb.hello(func() string { return e.session(b, nil) }, func(s, n string) string { return e.proof(b, s, n, nil) }); line != "ERR control_unavailable" {
430 t.Fatalf("a HELLO answered with the kill flag: %q", line)
431 }
432 eventually(t, "the other device told", 3*time.Second, func() bool { return containsLine(da.otherLines(), "CLOSE killed") })
433}
434
435// ---------------------------------------------------------------------------------------------------------------------------- order, duplicates, stale events
436
437func TestFeed_DuplicatesAreAppliedOnceAndAnEventAtOrBelowTheCursorIsDropped(t *testing.T) {
438 f := &scriptedFeed{}
439 e := withFeed(t, f)
440 d := e.newDevice()
441 dc := e.connectOK(d)
442 ds := dc.serve(nil)
443
444 resume := e.event(feedResume, d)
445 resume.Epoch = 5
446 f.set(func(since int64) FeedAnswer {
447 a := FeedAnswer{Epoch: 5, ServerTime: time.Now().Unix()}
448 // the same event three times in one answer, and again in every answer after it: a feed that does not honor the cursor
449 a.Events = []FeedEvent{resume, resume, resume}
450 return a
451 })
452 eventually(t, "the polls", 3*time.Second, func() bool { return len(f.polls()) >= 6 })
453 if n := e.cp.callsFor(d.id, "renew"); n != 1 {
454 t.Fatalf("the same event was applied %d times", n)
455 }
456 // an older epoch than the cursor changes nothing, even when it is a revoke
457 old := e.event(feedRevoke, d)
458 old.Epoch = 3
459 f.set(func(since int64) FeedAnswer {
460 return FeedAnswer{Epoch: 5, ServerTime: time.Now().Unix(), Events: []FeedEvent{old}}
461 })
462 time.Sleep(200 * time.Millisecond)
463 if !e.srv.Online(d.host) || containsLine(ds.otherLines(), "CLOSE revoked") {
464 t.Fatal("an event below the cursor was applied")
465 }
466 // the cursor moved to the head and never back
467 last := int64(-1)
468 for _, c := range f.polls() {
469 if c < last {
470 t.Fatalf("the cursor went back: %v", f.polls())
471 }
472 last = c
473 }
474}
475
476func TestFeed_AnAnswerOutOfOrderIsAppliedInEpochOrder(t *testing.T) {
477 f := &scriptedFeed{}
478 e := withFeed(t, f)
479 d := e.newDevice()
480 dc := e.connectOK(d)
481 ds := dc.serve(nil)
482
483 // delivered as [resume 8, pause 7]: in order that is a pause and then a resume, so the device ends up served; as delivered it would end paused.
484 pause, resume := e.event(feedPause, d), e.event(feedResume, d)
485 pause.Epoch, resume.Epoch = 7, 8
486 var delivered atomic.Bool
487 f.set(func(since int64) FeedAnswer {
488 if since >= 8 {
489 return FeedAnswer{Epoch: 8, ServerTime: time.Now().Unix()}
490 }
491 delivered.Store(true)
492 return FeedAnswer{Epoch: 8, ServerTime: time.Now().Unix(), Events: []FeedEvent{resume, pause}}
493 })
494 eventually(t, "the answer delivered", 3*time.Second, delivered.Load)
495 // both events produced a refresh and, whatever the interleaving, the control plane's last answer (active) is the state
496 eventually(t, "refreshed", 3*time.Second, func() bool { return e.cp.callsFor(d.id, "renew") >= 1 })
497 eventually(t, "served", 3*time.Second, func() bool { return e.srv.Online(d.host) })
498 if containsLine(ds.otherLines(), "CLOSE revoked") {
499 t.Fatal("closed")
500 }
501 if got := f.polls(); got[len(got)-1] != 8 {
502 t.Fatalf("the cursor must be the highest epoch seen: %v", got)
503 }
504}
505
506func TestFeed_AnEventOlderThanTheLastAnswerForAConnectionIsAlreadyInThatAnswer(t *testing.T) {
507 f := &scriptedFeed{}
508 e := withFeed(t, f)
509 d := e.newDevice()
510 dc := e.connectOK(d)
511 ds := dc.serve(nil)
512 // The device's license was ended and renewed: the revoke is from before the connection's own authorize, and is replayed (a node that restarted
513 // replays the feed from the start). The connection must not be cut by it.
514 stale := e.event(feedRevoke, d)
515 stale.At = e.now().Unix() - 120
516 f.push(stale)
517 time.Sleep(300 * time.Millisecond)
518 if !e.srv.Online(d.host) || containsLine(ds.otherLines(), "CLOSE revoked") {
519 t.Fatal("a revoke older than the connection's last answer cut it")
520 }
521 // a revoke from now is not stale
522 f.push(e.event(feedRevoke, d))
523 eventually(t, "CLOSE revoked", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE revoked") })
524}
525
526func TestFeed_UnreadableEventsAreSkippedAndTheRestOfTheAnswerIsStillApplied(t *testing.T) {
527 d := &testDevice{id: randomHex(16), label: label20()}
528 good := map[string]any{"epoch": 4, "type": "revoke", "device_id": d.id, "label": d.label, "at": 1793000000}
529 doc := map[string]any{"ok": true, "v": 1, "epoch": 4, "server_time": 1793000000, "events": []any{
530 map[string]any{"epoch": 1, "type": "teleport", "device_id": d.id, "label": d.label, "at": 1793000000}, // a type this node does not know
531 map[string]any{"epoch": 2, "type": "revoke", "device_id": "not-an-id", "label": d.label, "at": 1793000000}, // a malformed device id
532 map[string]any{"epoch": 3, "type": "kill", "device_id": d.id, "label": nil, "at": 1793000000}, // a global event that names a device
533 map[string]any{"epoch": 0, "type": "revoke", "device_id": d.id, "label": d.label, "at": 1793000000}, // epoch 0
534 map[string]any{"epoch": 5, "type": "revoke", "device_id": d.id, "label": "UPPERCASELABELUPPERCA", "at": 1}, // a bad label
535 good,
536 }}
537 answer, err := parseFeedAnswer([]byte(canonicalJSON(doc)))
538 if err != nil {
539 t.Fatal(err)
540 }
541 if len(answer.Events) != 1 || answer.Events[0].Epoch != 4 || answer.Skipped != 5 {
542 t.Fatalf("events %+v skipped %d", answer.Events, answer.Skipped)
543 }
544 // the envelope is read strictly: a node does not act on an answer it cannot make sense of
545 for name, bad := range map[string]string{
546 "not ok": `{"ok":false,"v":1,"epoch":1,"server_time":1793000000,"events":[]}`,
547 "another version": `{"ok":true,"v":2,"epoch":1,"server_time":1793000000,"events":[]}`,
548 "no events": `{"ok":true,"v":1,"epoch":1,"server_time":1793000000}`,
549 "a negative head": `{"ok":true,"v":1,"epoch":-1,"server_time":1793000000,"events":[]}`,
550 "a repeated key": `{"ok":true,"ok":true,"v":1,"epoch":1,"server_time":1793000000,"events":[]}`,
551 "reset not bool": `{"ok":true,"v":1,"epoch":1,"server_time":1793000000,"events":[],"reset":"yes"}`,
552 } {
553 if _, err := parseFeedAnswer([]byte(bad)); err == nil {
554 t.Errorf("%s: accepted", name)
555 }
556 }
557}
558
559func TestFeed_AResetClosesEveryTunnelAndTheCursorFollowsTheAnswer(t *testing.T) {
560 f := &scriptedFeed{}
561 e := withFeed(t, f)
562 a, b := e.newDevice(), e.newDevice()
563 ca, cb := e.connectOK(a), e.connectOK(b)
564 da, db := ca.serve(nil), cb.serve(nil)
565 sa := streamOpen(t, e, a)
566 f.set(func(since int64) FeedAnswer {
567 if since < 40 {
568 return FeedAnswer{Epoch: 40, Reset: true, ServerTime: time.Now().Unix()}
569 }
570 return FeedAnswer{Epoch: since, ServerTime: time.Now().Unix()}
571 })
572 for _, ds := range []*devServer{da, db} {
573 ds := ds
574 eventually(t, "CLOSE after a reset", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE expired") })
575 }
576 if !sa.dead(2 * time.Second) {
577 t.Fatal("a stream survived a reset")
578 }
579 eventually(t, "the cursor at the answer's epoch", 3*time.Second, func() bool {
580 p := f.polls()
581 return p[len(p)-1] == 40
582 })
583 // the devices reconnect and are authorized again: nothing is held against them
584 if line := e.connect(a).hello(func() string { return e.session(a, nil) }, func(s, n string) string { return e.proof(a, s, n, nil) }); !strings.HasPrefix(line, "OK ") {
585 t.Fatalf("a device after a reset: %q", line)
586 }
587}
588
589func TestFeed_AnAnswerBehindTheCursorIsNotActedOn(t *testing.T) {
590 f := &scriptedFeed{}
591 e := withFeed(t, f)
592 d := e.newDevice()
593 dc := e.connectOK(d)
594 ds := dc.serve(nil)
595 f.set(func(since int64) FeedAnswer {
596 if since == 0 {
597 return FeedAnswer{Epoch: 10, ServerTime: time.Now().Unix()}
598 }
599 // the control plane (or a cache in front of it) says the head is 3 although the node has seen 10, and brings a revoke with it
600 ev := e.event(feedRevoke, d)
601 ev.Epoch = 3
602 return FeedAnswer{Epoch: 3, ServerTime: time.Now().Unix(), Events: []FeedEvent{ev}}
603 })
604 time.Sleep(300 * time.Millisecond)
605 if !e.srv.Online(d.host) || containsLine(ds.otherLines(), "CLOSE revoked") {
606 t.Fatal("an answer behind the cursor was acted on")
607 }
608 for _, c := range f.polls() {
609 if c != 0 && c != 10 {
610 t.Fatalf("the cursor moved to %d on a bad answer: %v", c, f.polls())
611 }
612 }
613}
614
615// ---------------------------------------------------------------------------------------------------------------------------- the feed is down, the node is starting
616
617func TestFeed_AFeedThatIsDownDoesNotCutAnythingAndIsHeardAgainWithinOnePollPeriod(t *testing.T) {
618 clk := newFakeClock()
619 e, cp := httpNode(t, clk, nil)
620 d := e.newDevice()
621 dc := e.connectOK(d)
622 ds := dc.serve(nil)
623 stream := streamOpen(t, e, d)
624
625 cp.setDown(503)
626 cp.addEvent("revoke", d) // happens while the node cannot hear it
627 for i := 0; i < 20; i++ { // five minutes of failing polls, in 15 s steps
628 clk.Advance(15 * time.Second)
629 time.Sleep(10 * time.Millisecond)
630 }
631 if !e.srv.Online(d.host) || stream.dead(50*time.Millisecond) || containsLine(ds.otherLines(), "CLOSE revoked") {
632 t.Fatal("the node cut a device because its feed is unreachable: what is authenticated goes on until valid_until")
633 }
634 failed := cp.callCount(feedPath)
635 if failed < 5 {
636 t.Fatalf("the node polled %d times in five minutes of failure", failed)
637 }
638 cp.setDown(0)
639 elapsed := 0
640 for ; elapsed <= policyData.Timing.FeedApplyDeadlineS; elapsed++ {
641 if !e.srv.Online(d.host) {
642 break
643 }
644 clk.Advance(time.Second)
645 time.Sleep(15 * time.Millisecond)
646 }
647 if elapsed > policyData.Timing.FeedApplyDeadlineS {
648 t.Fatalf("the revoke was not applied within %d s of the feed coming back", elapsed)
649 }
650}
651
652func TestFeed_TheFirstAnswerIsAppliedBeforeTheNodeAcceptsAnyone(t *testing.T) {
653 f := &scriptedFeed{}
654 f.push(FeedEvent{Type: feedKill, At: time.Now().Unix()}) // the switch was on while the node was down
655 e := withFeed(t, f)
656 a := e.newDevice()
657 c := e.connect(a)
658 if line := c.hello(func() string { return e.session(a, nil) }, func(s, n string) string { return e.proof(a, s, n, nil) }); line != "ERR control_unavailable" {
659 t.Fatalf("a device accepted before the node knew the kill switch was on: %q", line)
660 }
661 if e.cp.callsFor(a.id, "") != 0 {
662 t.Fatal("authorize was asked while the kill switch was on")
663 }
664}
665
666func TestFeed_ALateFeedDoesNotKeepTheNodeFromStarting(t *testing.T) {
667 f := &scriptedFeed{block: make(chan struct{})}
668 started := time.Now()
669 e := newTestEnv(t, func(c *Config) {
670 c.Feed = f
671 c.FeedPoll = 20 * time.Millisecond
672 c.FeedStartWait = 250 * time.Millisecond
673 })
674 if took := time.Since(started); took > 3*time.Second {
675 t.Fatalf("the node took %s to start with a feed that does not answer", took)
676 }
677 d := e.newDevice()
678 _ = e.connectOK(d)
679 close(f.block)
680}
681
682func TestFeed_AFailingFeedIsRetriedWithinThePollPeriodNotTheBackoffCap(t *testing.T) {
683 // The backoff after repeated failures reaches a minute in the policy; the feed must not wait that long to hear a revoke once it is back.
684 if backoff(20) <= secs(policyData.Timing.FeedPollS) {
685 t.Fatal("this test needs a backoff longer than the poll period")
686 }
687 f := &scriptedFeed{fail: errors.New("down")}
688 clk := newFakeClock()
689 e := newEnv(t, clk, func(e *testEnv, c *Config) { c.Feed = f })
690 _ = e
691 for i := 0; i < 25; i++ {
692 clk.Advance(5 * time.Second)
693 time.Sleep(5 * time.Millisecond)
694 }
695 n := len(f.polls())
696 f.mu.Lock()
697 f.fail = nil
698 f.mu.Unlock()
699 for i := 0; i < 5; i++ {
700 clk.Advance(5 * time.Second)
701 time.Sleep(10 * time.Millisecond)
702 }
703 if len(f.polls()) <= n {
704 t.Fatalf("no poll within 25 s after the feed recovered (%d polls)", n)
705 }
706}
707
708func TestFeed_MoreThanAThousandEventsArePagedAtOnceAndTheCursorStopsAtTheLastEventGiven(t *testing.T) {
709 f := &scriptedFeed{}
710 nobody := &testDevice{id: randomHex(16), label: label20()}
711 for i := 0; i < 1500; i++ {
712 f.push(FeedEvent{Type: feedRevoke, DeviceID: nobody.id, Label: nobody.label, At: time.Now().Unix()})
713 }
714 // the control plane's answers: at most 1000 events, and `epoch` is where the cursor goes next (the last event given, not the head)
715 f.set(func(since int64) FeedAnswer {
716 a := FeedAnswer{ServerTime: time.Now().Unix()}
717 for e := since + 1; e <= 1500 && len(a.Events) < 1000; e++ {
718 a.Events = append(a.Events, FeedEvent{Epoch: e, Type: feedRevoke, DeviceID: nobody.id, Label: nobody.label, At: time.Now().Unix()})
719 }
720 a.Epoch = 1500
721 if len(a.Events) == 1000 && since+1000 < 1500 {
722 a.Epoch = since + 1000
723 }
724 return a
725 })
726 clk := newFakeClock()
727 _ = newEnv(t, clk, func(_ *testEnv, c *Config) { c.Feed = f })
728 eventually(t, "every page", 3*time.Second, func() bool { return len(f.polls()) >= 3 })
729 // no clock moved: the pages followed each other at once, and only the empty poll after them waits for a period
730 if got := f.polls(); got[0] != 0 || got[1] != 1000 || got[2] != 1500 || len(got) != 3 {
731 t.Fatalf("the cursor went %v, want [0 1000 1500]", got)
732 }
733}
734
735// ---------------------------------------------------------------------------------------------------------------------------- events during authorize
736
737// gateAnswer makes the control plane's `from`-th answer for d (1 = the first) decide at once and arrive only when release is called: the verdict is
738// from before whatever the feed says in between. Later answers are not held and carry laterStatus.
739func (e *testEnv) gateAnswer(d *testDevice, from int, laterStatus string) (decided <-chan struct{}, release func()) {
740 dec, rel := make(chan struct{}), make(chan struct{})
741 var once sync.Once
742 release = func() { once.Do(func() { close(rel) }) }
743 e.t.Cleanup(release)
744 e.cp.set(d.id, &devicePlan{host: d.host, answer: func(n int) AuthorizeResponse {
745 now := e.now()
746 status := statusActive
747 if n > from {
748 status = laterStatus
749 }
750 resp := AuthorizeResponse{DeviceID: d.id, Label: d.label, Host: d.host, Status: status, Plan: "paid", ValidUntil: now.Unix() + 3600, ServerTime: now.Unix(),
751 Policy: okPolicy{20, 45, 1800, 14400, 300}, Nodes: []any{}}
752 if n == from {
753 close(dec)
754 <-rel
755 }
756 return resp
757 }})
758 return dec, release
759}
760
761// applied waits until the node's feed cursor has passed epoch: the event is applied.
762func (f *scriptedFeed) applied(t *testing.T, epoch int64) {
763 t.Helper()
764 eventually(t, "the event applied", 3*time.Second, func() bool {
765 for _, c := range f.polls() {
766 if c >= epoch {
767 return true
768 }
769 }
770 return false
771 })
772}
773
774func TestFeed_ACutThatIsAppliedWhileAHelloWaitsForTheControlPlaneIsNotLost(t *testing.T) {
775 for kind, want := range map[string]string{feedRevoke: "ERR device_revoked", feedRelease: "ERR device_revoked", feedLicenseEnd: "ERR license_inactive"} {
776 t.Run(kind, func(t *testing.T) {
777 f := &scriptedFeed{}
778 e := withFeed(t, f)
779 d := e.newDevice()
780 decided, release := e.gateAnswer(d, 1, statusActive)
781 hello := e.helloAsync(d)
782 <-decided
783 f.push(e.event(kind, d))
784 f.applied(t, 1)
785 release()
786 select {
787 case line := <-hello:
788 if line != want {
789 t.Fatalf("the HELLO whose answer the %s overtook was answered %q, want %q", kind, line, want)
790 }
791 case <-time.After(5 * time.Second):
792 t.Fatal("no answer")
793 }
794 time.Sleep(100 * time.Millisecond)
795 if e.srv.Online(d.host) {
796 t.Fatal("a device cut while it was authorizing is online")
797 }
798 e.checkTranscript()
799 })
800 }
801}
802
803func TestFeed_AnEventThatDoesNotConcernAHelloWaitingForTheControlPlaneLeavesItAlone(t *testing.T) {
804 f := &scriptedFeed{}
805 e := withFeed(t, f)
806 d, other := e.newDevice(), e.newDevice()
807 decided, release := e.gateAnswer(d, 1, statusActive)
808 hello := e.helloAsync(d)
809 <-decided
810 f.push(e.event(feedRevoke, other)) // another device
811 old := e.event(feedRevoke, d) // this one, but from before the answer: the answer already knew it
812 old.At = e.now().Unix() - 60
813 f.push(old)
814 f.applied(t, 2)
815 release()
816 select {
817 case line := <-hello:
818 if !strings.HasPrefix(line, "OK ") {
819 t.Fatalf("the HELLO was answered %q", line)
820 }
821 case <-time.After(5 * time.Second):
822 t.Fatal("no answer")
823 }
824 if !e.srv.Online(d.host) {
825 t.Fatal("the device is not online")
826 }
827 e.checkTranscript()
828}
829
830func TestFeed_APauseThatIsAppliedWhileAHelloWaitsForTheControlPlaneHoldsAndIsRefreshed(t *testing.T) {
831 f := &scriptedFeed{}
832 e := withFeed(t, f)
833 d := e.newDevice()
834 decided, release := e.gateAnswer(d, 1, statusPaused)
835 hello := e.helloAsync(d)
836 <-decided
837 f.push(e.event(feedPause, d))
838 f.applied(t, 1)
839 release()
840 select {
841 case line := <-hello:
842 if !strings.HasPrefix(line, "OK ") {
843 t.Fatalf("the HELLO was answered %q", line)
844 }
845 case <-time.After(5 * time.Second):
846 t.Fatal("no answer")
847 }
848 if e.srv.Online(d.host) {
849 t.Fatal("a device paused while it was authorizing is served")
850 }
851 eventually(t, "the control plane asked again", 3*time.Second, func() bool { return e.cp.callsFor(d.id, "renew") >= 1 })
852 time.Sleep(100 * time.Millisecond)
853 if e.srv.Online(d.host) {
854 t.Fatal("the refresh, which says paused, lifted the pause")
855 }
856}
857
858func TestFeed_APauseAppliedWhileARenewWaitsForTheControlPlaneIsNotWrittenOverByItsAnswer(t *testing.T) {
859 f := &scriptedFeed{}
860 e := withFeed(t, f)
861 d := e.newDevice()
862 dc := e.connectOK(d)
863 decided, release := e.gateAnswer(d, 2, statusPaused) // the HELLO was the first call; plain plan afterwards
864 e.cp.set(d.id, e.cp.plans[d.id])
865 renewed := make(chan string, 1)
866 go func() {
867 dc.session = e.session(d, nil)
868 dc.send("RENEW " + dc.session)
869 line, err := dc.try(5 * time.Second)
870 if err != nil || !strings.HasPrefix(line, "CHALLENGE ") {
871 renewed <- "no challenge: " + line
872 return
873 }
874 m, _ := parseLine(line, dirN2D)
875 dc.send("PROOF " + e.proof(d, dc.session, m.str("nonce"), nil))
876 line, _ = dc.try(5 * time.Second)
877 renewed <- line
878 }()
879 <-decided
880 f.push(e.event(feedPause, d))
881 f.applied(t, 1)
882 release()
883 select {
884 case line := <-renewed:
885 if !strings.HasPrefix(line, "OK ") {
886 t.Fatalf("the RENEW was answered %q", line)
887 }
888 case <-time.After(5 * time.Second):
889 t.Fatal("no answer")
890 }
891 if e.srv.Online(d.host) {
892 t.Fatal("the RENEW's answer, decided before the pause, lifted it")
893 }
894}
895
896func TestFeed_TheKillSwitchThatGoesOnWhileAHelloWaitsForTheControlPlaneDoesNotServeIt(t *testing.T) {
897 f := &scriptedFeed{}
898 e := withFeed(t, f)
899 d := e.newDevice()
900 decided, release := e.gateAnswer(d, 1, statusActive)
901 hello := e.helloAsync(d)
902 <-decided
903 f.push(e.event(feedKill, nil))
904 f.applied(t, 1)
905 release()
906 select {
907 case line := <-hello:
908 if strings.HasPrefix(line, "OK ") {
909 t.Fatalf("the HELLO was answered %q with the kill switch on", line)
910 }
911 case <-time.After(5 * time.Second):
912 t.Fatal("no answer")
913 }
914 if e.srv.Online(d.host) {
915 t.Fatal("a device is online with the kill switch on")
916 }
917}
918
919func TestFeed_AResetThatIsAppliedWhileAHelloWaitsForTheControlPlaneDoesNotServeItEither(t *testing.T) {
920 f := &scriptedFeed{}
921 e := withFeed(t, f)
922 d := e.newDevice()
923 decided, release := e.gateAnswer(d, 1, statusActive)
924 var reset atomic.Bool
925 f.set(func(since int64) FeedAnswer {
926 if !reset.Swap(true) {
927 return FeedAnswer{Reset: true, Epoch: 7, ServerTime: time.Now().Unix()}
928 }
929 return FeedAnswer{Epoch: 7, ServerTime: time.Now().Unix()}
930 })
931 hello := e.helloAsync(d)
932 <-decided
933 time.Sleep(300 * time.Millisecond) // the reset is applied by the first polls
934 release()
935 select {
936 case line := <-hello:
937 if strings.HasPrefix(line, "OK ") {
938 t.Fatalf("a HELLO overtaken by a reset was answered %q", line)
939 }
940 case <-time.After(5 * time.Second):
941 t.Fatal("no answer")
942 }
943 if e.srv.Online(d.host) {
944 t.Fatal("a device is online after the reset overtook its authorization")
945 }
946}
947
948func TestFeed_ARefreshAfterShutdownStartsNothing(t *testing.T) {
949 f := &scriptedFeed{}
950 e := withFeed(t, f)
951 d := e.newDevice()
952 e.connectOK(d)
953 cc, _ := e.srv.routes.lookup(d.host)
954 e.srv.Shutdown(time.Second)
955 e.srv.refreshAsync(cc)
956 cc.mu.Lock()
957 running := cc.refreshing
958 cc.mu.Unlock()
959 if running {
960 t.Fatal("a refresh was started on a node that has ended")
961 }
962}