ALL

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

tunnel/limits_test.go

SHA-256: 56e06711db079cd1ef2653de31170cabbc149b6f217b0d542f57a7574fd8baee
991 lines · 36738 bytes · View on GitHub

1package main
2
3import (
4 "context"
5 "crypto/tls"
6 "encoding/json"
7 "fmt"
8 "hash/maphash"
9 "io"
10 "net"
11 "net/netip"
12 "reflect"
13 "strings"
14 "sync"
15 "syscall"
16 "testing"
17 "time"
18)
19
20// The limits of the public listener (limits.go): per source, per device host, global, and the ban for trying names that are not a device.
21
22func newTestLimits(t *testing.T, mutate func(*Config)) (*ipLimits, *fakeClock) {
23 t.Helper()
24 cfg := DefaultConfig()
25 if mutate != nil {
26 mutate(&cfg)
27 }
28 fc := newFakeClock()
29 return newIPLimits(&cfg, newClock(fc.Now, fc.AfterFunc), maphash.MakeSeed()), fc
30}
31
32func TestDefaultConfigHardeningIsThePolicysNumbers(t *testing.T) {
33 cfg := DefaultConfig()
34 for name, c := range map[string]struct{ got, key int }{
35 "PerIPConcurrent": {cfg.PerIPConcurrent, ctPolInt("node_limits.per_ip_concurrent")},
36 "PerIPNewPerS": {cfg.PerIPNewPerS, ctPolInt("node_limits.per_ip_new_per_s")},
37 "IPv6PrefixBits": {cfg.IPv6PrefixBits, ctPolInt("node_limits.ipv6_prefix_bits")},
38 "IPTableMax": {cfg.IPTableMax, ctPolInt("node_limits.ip_table_max")},
39 "OpenAIPerIPFactor": {cfg.OpenAIPerIPFactor, ctPolInt("node_limits.openai_per_ip_factor")},
40 "DevicePerIPFactor": {cfg.DevicePerIPFactor, ctPolInt("node_limits.device_per_ip_factor")},
41 "UnknownSNIPerIP": {cfg.UnknownSNIPerIP, ctPolInt("node_limits.unknown_sni_per_ip")},
42 "LabelAnonSlice": {cfg.LabelAnonSlice, ctPolInt("node_limits.label_anon_slice")},
43 "LabelNewPerS": {cfg.LabelNewPerS, ctPolInt("node_limits.label_new_per_s")},
44 "LabelNewBurst": {cfg.LabelNewBurst, ctPolInt("node_limits.label_new_burst")},
45 "OpenAIReservedPct": {cfg.OpenAIReservedPct, ctPolInt("node_limits.openai_reserved_pct")},
46 "HintsPerLabel": {cfg.HintsPerLabel, ctPolInt("node_limits.hint_addresses_per_label")},
47 "GatedStreamsPerDay": {cfg.GatedStreamsPerDay, ctPolInt("gated.streams_per_day")},
48 } {
49 if c.got != c.key {
50 t.Errorf("%s = %d, policy %d", name, c.got, c.key)
51 }
52 }
53 for name, c := range map[string]struct {
54 got time.Duration
55 key string
56 }{
57 "UnknownSNIWindow": {cfg.UnknownSNIWindow, "node_limits.unknown_sni_window_s"},
58 "UnknownSNIBan": {cfg.UnknownSNIBan, "node_limits.unknown_sni_ban_s"},
59 "HintTTL": {cfg.HintTTL, "node_limits.hint_ttl_s"},
60 "TCPKeepalive": {cfg.TCPKeepalive, "node_limits.tcp_keepalive_s"},
61 "OpenAIPrefixRefresh": {cfg.OpenAIPrefixRefresh, "node_limits.openai_prefix_refresh_s"},
62 "HelloTimeout": {cfg.HelloTimeout, "node_limits.client_hello_deadline_s"},
63 "StreamIdle": {cfg.StreamIdle, "timing.stream_idle_s"},
64 "StreamMaxAge": {cfg.StreamMaxAge, "timing.stream_max_age_s"},
65 } {
66 if c.got != time.Duration(ctPolInt(c.key))*time.Second {
67 t.Errorf("%s = %s, policy %s = %d", name, c.got, c.key, ctPolInt(c.key))
68 }
69 }
70 if int64(cfg.GatedBytesPerDay) != int64(ctPol("gated.bytes_per_day").(float64)) {
71 t.Errorf("GatedBytesPerDay = %d", cfg.GatedBytesPerDay)
72 }
73 if cfg.LabelAnonSlice+ctPolInt("node_limits.label_hint_slice") != cfg.MaxStreamsPerHost {
74 t.Errorf("the anonymous slice %d and the hint slice %d are not the whole host cap %d", cfg.LabelAnonSlice, ctPolInt("node_limits.label_hint_slice"), cfg.MaxStreamsPerHost)
75 }
76}
77
78func TestConfig_HardeningIsValidated(t *testing.T) {
79 for name, mutate := range map[string]func(*Config){
80 "no per-source cap": func(c *Config) { c.PerIPConcurrent = 0 },
81 "no per-source rate": func(c *Config) { c.PerIPNewPerS = 0 },
82 "a prefix of 0 bits": func(c *Config) { c.IPv6PrefixBits = 0 },
83 "a prefix of 129 bits": func(c *Config) { c.IPv6PrefixBits = 129 },
84 "the whole node reserved": func(c *Config) { c.OpenAIReservedPct = 100 },
85 "no ban": func(c *Config) { c.UnknownSNIBan = 0 },
86 "no hint life": func(c *Config) { c.HintTTL = 0 },
87 "no gated budget": func(c *Config) { c.GatedBytesPerDay = 0 },
88 "no keepalive": func(c *Config) { c.TCPKeepalive = 0 },
89 "no device factor": func(c *Config) { c.DevicePerIPFactor = 0 },
90 "metrics on the network": func(c *Config) { c.MetricsAddr = "0.0.0.0:9101" },
91 "metrics on a host name": func(c *Config) { c.MetricsAddr = "example.org:9101" },
92 "metrics without a port": func(c *Config) { c.MetricsAddr = "127.0.0.1" },
93 "prefix list over http": func(c *Config) { c.OpenAIPrefixesURL = "http://example.org/list.json" },
94 "prefix list with a secret": func(c *Config) { c.OpenAIPrefixesURL = "https://user:pass" + "@example.org/list.json" },
95 } {
96 cfg := DefaultConfig()
97 cfg.HTTPSAddr, cfg.HTTPAddr = "", ""
98 cfg.NodeID = "t1"
99 cfg.Keys, cfg.Control = StaticKeys{}, &fakeControl{}
100 mutate(&cfg)
101 if err := cfg.validate(); err == nil {
102 t.Errorf("%s: accepted", name)
103 }
104 }
105 good := DefaultConfig()
106 good.HTTPSAddr, good.HTTPAddr = "", ""
107 good.NodeID = "t1"
108 good.Keys, good.Control = StaticKeys{}, &fakeControl{}
109 good.MetricsAddr, good.OpenAIPrefixesURL = "[::1]:9101", "http://127.0.0.1:8080/list.json"
110 if err := good.validate(); err != nil {
111 t.Errorf("a loopback metrics address and a loopback mirror: %v", err)
112 }
113}
114
115func TestIPLimits_ConcurrentCapAndRateBucketPerSource(t *testing.T) {
116 l, fc := newTestLimits(t, func(c *Config) { c.PerIPConcurrent, c.PerIPNewPerS = 3, 5 })
117 a := netip.MustParseAddr("198.51.100.7")
118 var releases []func()
119 for i := 0; i < 3; i++ {
120 rel, why := l.admit(a, false)
121 if why != "" {
122 t.Fatalf("connection %d: %s", i, why)
123 }
124 releases = append(releases, rel)
125 }
126 if _, why := l.admit(a, false); why != refuseConcurrent {
127 t.Fatalf("a fourth concurrent connection: %q", why)
128 }
129 // another source is untouched
130 if _, why := l.admit(netip.MustParseAddr("198.51.100.8"), false); why != "" {
131 t.Fatalf("another source: %q", why)
132 }
133 releases[0]()
134 releases[0]() // a second release of the same slot gives nothing back
135 if _, why := l.admit(a, false); why != "" {
136 t.Fatalf("after one release: %q", why)
137 }
138 if _, why := l.admit(a, false); why != refuseConcurrent {
139 t.Fatalf("the double release freed a slot that was not its own: %q", why)
140 }
141 // the bucket: 5 new connections in a second, however fast they are released
142 b := netip.MustParseAddr("203.0.113.1")
143 admitted := 0
144 for i := 0; i < 20; i++ {
145 rel, why := l.admit(b, false)
146 if why == "" {
147 admitted++
148 rel()
149 } else if why != refuseRate {
150 t.Fatalf("%q", why)
151 }
152 }
153 if admitted != 5 {
154 t.Fatalf("admitted %d in one instant, want the burst of 5", admitted)
155 }
156 fc.Advance(400 * time.Millisecond)
157 got := 0
158 for i := 0; i < 10; i++ {
159 if rel, why := l.admit(b, false); why == "" {
160 got++
161 rel()
162 }
163 }
164 if got != 2 {
165 t.Fatalf("after 400 ms %d more were admitted, want 2 (5 per second)", got)
166 }
167}
168
169func TestIPLimits_AnIPv6SourceIsItsPrefixAndAnOpenAIAddressHasLargerLimits(t *testing.T) {
170 l, _ := newTestLimits(t, func(c *Config) { c.PerIPConcurrent, c.PerIPNewPerS, c.OpenAIPerIPFactor = 2, 100, 4 })
171 hold := func(addr string, openai bool) string {
172 _, why := l.admit(netip.MustParseAddr(addr), openai)
173 return why
174 }
175 // ten addresses of one /64 are one source
176 for i := 0; i < 2; i++ {
177 if why := hold(fmt.Sprintf("2001:db8:aaaa:1::%x", i+1), false); why != "" {
178 t.Fatalf("%q", why)
179 }
180 }
181 for i := 3; i < 10; i++ {
182 if why := hold(fmt.Sprintf("2001:db8:aaaa:1::%x", i), false); why != refuseConcurrent {
183 t.Fatalf("address %d of the same /64: %q", i, why)
184 }
185 }
186 // the next /64 is another source
187 if why := hold("2001:db8:aaaa:2::1", false); why != "" {
188 t.Fatalf("the next /64: %q", why)
189 }
190 // an address of the OpenAI list holds factor times as many
191 got := 0
192 for i := 0; i < 20; i++ {
193 if hold("192.0.2.9", true) == "" {
194 got++
195 }
196 }
197 if got != 8 {
198 t.Fatalf("an OpenAI address held %d, want 2*4", got)
199 }
200 // an IPv4-mapped address is the IPv4 address
201 if why := hold("::ffff:192.0.2.77", false); why != "" {
202 t.Fatalf("%q", why)
203 }
204 for i := 0; i < 2; i++ {
205 _, why := l.admit(netip.MustParseAddr("192.0.2.77"), false)
206 if i == 0 && why != "" {
207 t.Fatalf("%q", why)
208 }
209 if i == 1 && why != refuseConcurrent {
210 t.Fatalf("the mapped and the plain address are two sources: %q", why)
211 }
212 }
213}
214
215func TestIPLimits_UnknownNamesBanASourceForAWhileAndOnlyDistinctNamesCount(t *testing.T) {
216 l, fc := newTestLimits(t, func(c *Config) {
217 c.UnknownSNIPerIP, c.UnknownSNIWindow, c.UnknownSNIBan = 3, time.Minute, 10*time.Minute
218 })
219 a := netip.MustParseAddr("198.51.100.20")
220 // the same offline device asked for again and again is not a scan
221 for i := 0; i < 50; i++ {
222 if l.strike(a, "offline.example.org", false) {
223 t.Fatal("a repeated name banned the source")
224 }
225 }
226 for _, name := range []string{"b", "c"} {
227 if l.strike(a, name+".example.org", false) {
228 t.Fatalf("banned at %s", name)
229 }
230 }
231 if !l.strike(a, "d.example.org", false) {
232 t.Fatal("the fourth distinct name (budget 3) did not ban")
233 }
234 if _, why := l.admit(a, false); why != refuseBanned {
235 t.Fatalf("a banned source: %q", why)
236 }
237 if _, why := l.admit(netip.MustParseAddr("198.51.100.21"), false); why != "" {
238 t.Fatalf("a ban reached another source: %q", why)
239 }
240 fc.Advance(9 * time.Minute)
241 if _, why := l.admit(a, false); why != refuseBanned {
242 t.Fatalf("still banned at 9 minutes: %q", why)
243 }
244 fc.Advance(2 * time.Minute)
245 if _, why := l.admit(a, false); why != "" {
246 t.Fatalf("the ban outlived its time: %q", why)
247 }
248 // the window: strikes older than it are forgotten
249 b := netip.MustParseAddr("198.51.100.30")
250 l.strike(b, "x1", false)
251 l.strike(b, "x2", false)
252 l.strike(b, "x3", false)
253 fc.Advance(2 * time.Minute)
254 if l.strike(b, "x4", false) {
255 t.Fatal("strikes older than the window counted")
256 }
257 // junk (no TLS, no name) counts every time: there is no name to be the same
258 c := netip.MustParseAddr("198.51.100.40")
259 banned := false
260 for i := 0; i < 4 && !banned; i++ {
261 banned = l.strike(c, "", false)
262 }
263 if !banned {
264 t.Fatal("four junk hellos did not ban")
265 }
266 // an address of the OpenAI list is never struck, whatever it tries
267 o := netip.MustParseAddr("192.0.2.50")
268 for i := 0; i < 100; i++ {
269 if l.strike(o, fmt.Sprintf("n%d.example.org", i), true) {
270 t.Fatal("an OpenAI address was banned")
271 }
272 }
273 if _, why := l.admit(o, true); why != "" {
274 t.Fatalf("%q", why)
275 }
276}
277
278func TestIPLimits_ATableThatIsFullSharesAnOverflowEntryAndShrinksWhenIdle(t *testing.T) {
279 l, fc := newTestLimits(t, func(c *Config) { c.IPTableMax, c.PerIPConcurrent, c.PerIPNewPerS = 4, 1, 1000 })
280 for i := 0; i < 4; i++ {
281 rel, why := l.admit(netip.MustParseAddr(fmt.Sprintf("198.51.100.%d", i+1)), false)
282 if why != "" {
283 t.Fatal(why)
284 }
285 rel()
286 }
287 // the table is full of idle entries: a new source sweeps them and gets its own
288 fc.Advance(2 * time.Second)
289 if _, why := l.admit(netip.MustParseAddr("198.51.100.99"), false); why != "" {
290 t.Fatalf("%q", why)
291 }
292 size, _, _, overflowed := l.snapshot()
293 if size > 4 || overflowed != 0 {
294 t.Fatalf("size %d, overflowed %d", size, overflowed)
295 }
296 // four sources that hold connections cannot be swept: the fifth shares the overflow entry, which is bounded as a crowd (not zero, not unbounded)
297 l2, _ := newTestLimits(t, func(c *Config) { c.IPTableMax, c.PerIPConcurrent, c.PerIPNewPerS = 4, 1, 1000 })
298 for i := 0; i < 4; i++ {
299 if _, why := l2.admit(netip.MustParseAddr(fmt.Sprintf("198.51.100.%d", i+1)), false); why != "" {
300 t.Fatal(why)
301 }
302 }
303 if _, why := l2.admit(netip.MustParseAddr("203.0.113.1"), false); why != "" {
304 t.Fatalf("the first of the overflow: %q", why)
305 }
306 if _, _, _, overflowed := l2.snapshot(); overflowed == 0 {
307 t.Fatal("no overflow was counted")
308 }
309 size, _, _, _ = l2.snapshot()
310 if size != 4 {
311 t.Fatalf("the table grew past its bound: %d", size)
312 }
313 // a later sweep drops what is idle
314 l3, fc3 := newTestLimits(t, func(c *Config) { c.IPTableMax, c.PerIPNewPerS = 100, 1000 })
315 for i := 0; i < 30; i++ {
316 rel, _ := l3.admit(netip.MustParseAddr(fmt.Sprintf("198.51.100.%d", i+1)), false)
317 rel()
318 }
319 fc3.Advance(3 * time.Minute)
320 l3.Sweep()
321 if size, _, _, _ := l3.snapshot(); size != 0 {
322 t.Fatalf("an idle table kept %d entries", size)
323 }
324}
325
326func TestBucketAndLogSampler(t *testing.T) {
327 var b bucket
328 now := time.Now()
329 taken := 0
330 for i := 0; i < 10; i++ {
331 if b.take(now, 2, 3) {
332 taken++
333 }
334 }
335 if taken != 3 {
336 t.Fatalf("a burst of 3 gave %d", taken)
337 }
338 if !b.take(now.Add(time.Second), 2, 3) || !b.take(now.Add(time.Second), 2, 3) || b.take(now.Add(time.Second), 2, 3) {
339 t.Fatal("two tokens a second")
340 }
341 fc := newFakeClock()
342 s := newLogSampler(newClock(fc.Now, fc.AfterFunc))
343 if ok, cut := s.allow("r"); !ok || cut != 0 {
344 t.Fatal("the first line is written")
345 }
346 for i := 0; i < 5; i++ {
347 if ok, _ := s.allow("r"); ok {
348 t.Fatal("a repeat inside a second was written")
349 }
350 }
351 if ok, _ := s.allow("other"); !ok {
352 t.Fatal("another reason has its own second")
353 }
354 fc.Advance(1100 * time.Millisecond)
355 if ok, cut := s.allow("r"); !ok || cut != 5 {
356 t.Fatalf("ok=%v, left out %d, want 5", ok, cut)
357 }
358}
359
360// ------------------------------------------------------------------------------------------------------------------------------------
361// The node: what a source can do to it
362// ------------------------------------------------------------------------------------------------------------------------------------
363
364func TestNode_OneSourceCannotHoldMoreThanThePerIPCapNorFillTheGlobalBudget(t *testing.T) {
365 release := make(chan struct{})
366 e := newVIPEnv(t, func(e *testEnv, c *Config) {
367 c.PerIPConcurrent, c.MaxUnauth, c.HelloTimeout = 4, 16, 3*time.Second
368 })
369 d := e.newDevice()
370 dc := e.connectOK(d)
371 dc.serve(holdHandler(d, release))
372 defer close(release)
373
374 // pre-hello: the attacker opens 40 connections and says nothing. Only 4 are served, and the budget everybody shares is not touched beyond them.
375 var idle []net.Conn
376 for i := 0; i < 40; i++ {
377 c, err := e.vip.dial("198.51.100.66", e.httpsAddr)
378 if err != nil {
379 t.Fatal(err)
380 }
381 defer c.Close()
382 idle = append(idle, c)
383 }
384 eventually(t, "the node counted the refused ones", 3*time.Second, func() bool { return e.rejected(refuseConcurrent) >= 36 })
385 if got := e.srv.unauth.Load(); got > 4 {
386 t.Fatalf("one source holds %d pre-hello slots, cap 4", got)
387 }
388 // somebody else is served through it
389 good, err := e.clientFrom("203.0.113.5", d)
390 if err != nil {
391 t.Fatalf("a client of another source while one floods: %v", err)
392 }
393 _ = good.Close()
394
395 // streams: the same source holds at most 4 streams to the device, the next one is dropped with no answer at all
396 for _, c := range idle {
397 _ = c.Close()
398 }
399 eventually(t, "the slots of the closed connections are back", 3*time.Second, func() bool { return e.srv.unauth.Load() == 0 })
400 var held []*tls.Conn
401 for i := 0; i < 8; i++ {
402 tc, err := e.clientFrom("198.51.100.66", d)
403 if err == nil {
404 held = append(held, tc)
405 }
406 }
407 if len(held) != 4 {
408 t.Fatalf("one source held %d streams, cap 4", len(held))
409 }
410 // (the stream of the other source is still open too: the device holds it until the half-close timeout)
411 eventually(t, "the streams open are the 4 held and the other source's", 3*time.Second, func() bool { return e.metricValue(`all_tunnel_streams_open`) == 5 })
412 e.checkTranscript()
413}
414
415func TestNode_ANewConnectionRateLimitsASourceAndItRecoversWithTime(t *testing.T) {
416 e := newVIPEnv(t, func(e *testEnv, c *Config) { c.PerIPNewPerS, c.PerIPConcurrent = 5, 1000 })
417 dropped := 0
418 for i := 0; i < 15; i++ {
419 answer, closed := e.helloFrom("198.51.100.90", "nothing.devices.all-e2e.test", 300*time.Millisecond)
420 if closed && len(answer) == 0 {
421 dropped++
422 }
423 }
424 if dropped != 10 {
425 t.Fatalf("%d connections dropped without an answer, want 10 of 15 (burst of 5)", dropped)
426 }
427 if e.rejected(refuseRate) != 10 {
428 t.Fatalf("counted %d", e.rejected(refuseRate))
429 }
430 e.offsetNS.Add(int64(3 * time.Second))
431 if answer, _ := e.helloFrom("198.51.100.90", "nothing.devices.all-e2e.test", 300*time.Millisecond); len(answer) == 0 {
432 t.Fatal("still dropped after the bucket refilled")
433 }
434}
435
436func TestNode_UnknownNamesBanTheSourceAtAccept(t *testing.T) {
437 e := newVIPEnv(t, func(e *testEnv, c *Config) {
438 c.UnknownSNIPerIP, c.UnknownSNIBan, c.PerIPNewPerS = 3, 5*time.Minute, 1000
439 })
440 d := e.newDevice()
441 dc := e.connectOK(d)
442 dc.serve(nil)
443 scanner := "198.51.100.101"
444 for i := 0; i < 3; i++ {
445 answer, _ := e.helloFrom(scanner, fmt.Sprintf("%020d.devices.all-e2e.test", i), time.Second)
446 if len(answer) != 7 { // a TLS alert: the node answered
447 t.Fatalf("name %d: %v", i, answer)
448 }
449 }
450 // the fourth different name goes over the budget: it is still answered (the alert), and from then on the source is dropped at accept
451 if answer, _ := e.helloFrom(scanner, "zzzzzzzzzzzzzzzzzzzz.devices.all-e2e.test", time.Second); len(answer) != 7 {
452 t.Fatalf("the name that went over: %v", answer)
453 }
454 answer, closed := e.helloFrom(scanner, d.host, time.Second)
455 if !closed || len(answer) != 0 {
456 t.Fatalf("a banned source got %v (closed %v)", answer, closed)
457 }
458 if e.rejected(refuseBanned) != 1 || e.metricValue(`all_tunnel_ip_bans_total`) != 1 {
459 t.Fatalf("counters: banned %d, bans %d", e.rejected(refuseBanned), e.metricValue(`all_tunnel_ip_bans_total`))
460 }
461 // the device's clients from anywhere else are served; so is ChatGPT retrying an OFFLINE device from one address (one name, however often)
462 if tc, err := e.clientFrom("203.0.113.9", d); err != nil {
463 t.Fatalf("another source: %v", err)
464 } else {
465 _ = tc.Close()
466 }
467 for i := 0; i < 10; i++ {
468 if answer, _ := e.helloFrom("203.0.113.10", "offlineoffline22222.devices.all-e2e.test", time.Second); len(answer) != 7 {
469 t.Fatalf("retry %d of an offline name: %v", i, answer)
470 }
471 }
472 if answer, closed := e.helloFrom("203.0.113.10", d.host, time.Second); closed && len(answer) == 0 {
473 t.Fatal("a source retrying one offline name was banned")
474 }
475 // the ban ends
476 e.offsetNS.Add(int64(6 * time.Minute))
477 if answer, closed := e.helloFrom(scanner, d.host, time.Second); closed && len(answer) == 0 {
478 t.Fatal("still banned after the ban time")
479 }
480}
481
482func TestNode_ClientsThatAreNotTLSAreStruckAndAnOpenAIAddressIsNeverBanned(t *testing.T) {
483 var vipPrefix = "192.0.2.0/24"
484 e := newVIPEnv(t, func(e *testEnv, c *Config) {
485 c.UnknownSNIPerIP, c.PerIPNewPerS, c.PerIPConcurrent = 3, 1000, 1000
486 c.PrefixFetch = func(context.Context) ([]byte, error) {
487 return []byte(`{"prefixes":[{"ipv4Prefix":"` + vipPrefix + `"}]}`), nil
488 }
489 })
490 eventually(t, "the prefix list is in", 3*time.Second, func() bool { return e.srv.openai.size() == 1 })
491 junk := func(ip string) bool {
492 raw, err := e.vip.dial(ip, e.httpsAddr)
493 if err != nil {
494 t.Fatal(err)
495 }
496 defer raw.Close()
497 _, _ = raw.Write([]byte("GET / HTTP/1.1\r\nHost: x\r\n\r\n"))
498 _ = raw.SetReadDeadline(time.Now().Add(time.Second))
499 _, _ = io.ReadAll(raw)
500 return true
501 }
502 for i := 0; i < 8; i++ {
503 junk("198.51.100.150")
504 junk("192.0.2.150") // ChatGPT's address
505 }
506 if answer, closed := e.helloFrom("198.51.100.150", "x.devices.all-e2e.test", time.Second); !closed || len(answer) != 0 {
507 t.Fatalf("a source that sent junk four times was served: %v", answer)
508 }
509 if answer, _ := e.helloFrom("192.0.2.150", "x.devices.all-e2e.test", time.Second); len(answer) != 7 {
510 t.Fatalf("an OpenAI address was dropped: %v", answer)
511 }
512}
513
514func TestNode_SlowlorisBeforeTheClientHelloIsDroppedAtTheDeadlineWithoutAnOpen(t *testing.T) {
515 // The real policy deadline (5 s), not a shortened one.
516 e := newTestEnv(t, func(c *Config) { c.HelloTimeout = DefaultConfig().HelloTimeout })
517 if DefaultConfig().HelloTimeout != 5*time.Second {
518 t.Fatalf("the policy deadline is %s", DefaultConfig().HelloTimeout)
519 }
520 d := e.newDevice()
521 dc := e.connectOK(d)
522 ds := dc.serve(nil)
523 ds.noAttach = true
524 hello := helloOf(d.host, 0, 16000)
525 c, err := net.DialTimeout("tcp", e.httpsAddr, 3*time.Second)
526 if err != nil {
527 t.Fatal(err)
528 }
529 defer c.Close()
530 started := time.Now()
531 done := make(chan time.Duration, 1)
532 go func() {
533 _ = c.SetReadDeadline(time.Now().Add(15 * time.Second))
534 _, _ = io.ReadAll(c)
535 done <- time.Since(started)
536 }()
537 // one byte a second, never the whole hello
538 for i := 0; i < len(hello)/2 && time.Since(started) < 8*time.Second; i++ {
539 if _, err := c.Write(hello[i : i+1]); err != nil {
540 break
541 }
542 time.Sleep(time.Second)
543 }
544 select {
545 case took := <-done:
546 if took < 4500*time.Millisecond || took > 6500*time.Millisecond {
547 t.Fatalf("dropped after %s, want about 5 s", took)
548 }
549 case <-time.After(10 * time.Second):
550 t.Fatal("the slow client was never dropped")
551 }
552 if ds.openCount() != 0 {
553 t.Fatal("an OPEN was sent for a ClientHello that never completed")
554 }
555 if e.logs.Count(`"reason":"hello_timeout"`) < 1 {
556 t.Fatalf("logs:\n%s", e.logs.String())
557 }
558 if e.rejected("hello_timeout") != 1 {
559 t.Fatalf("counted %d", e.rejected("hello_timeout"))
560 }
561}
562
563func TestNode_ManyHeldStreamsFromManySourcesLeaveRoomForASourceTheDeviceMarkedGood(t *testing.T) {
564 release := make(chan struct{})
565 e := newVIPEnv(t, func(e *testEnv, c *Config) { c.PerIPNewPerS = 1000 })
566 d := e.newDevice()
567 dc := e.connectOK(d)
568 ds := dc.serve(holdHandler(d, release))
569 defer close(release)
570
571 // 20 attacker sources, 4 attempts each: 80 attempts to hold 64 streams. The anonymous slice (16) is all they get.
572 held := 0
573 for i := 0; i < 20; i++ {
574 for j := 0; j < 4; j++ {
575 if _, err := e.clientFrom(fmt.Sprintf("198.51.100.%d", 100+i), d); err == nil {
576 held++
577 }
578 }
579 }
580 if held != DefaultConfig().LabelAnonSlice {
581 t.Fatalf("the attackers held %d streams, want the anonymous slice of %d", held, DefaultConfig().LabelAnonSlice)
582 }
583 if e.logs.Count(`"reason":"host_limit"`) < 1 {
584 t.Fatalf("no host_limit in the logs:\n%s", e.logs.String())
585 }
586 // ChatGPT's address has not been marked: it cannot get in either, which is the price of a slice that is anonymous
587 if _, err := e.clientFrom("203.0.113.77", d); err == nil {
588 t.Fatal("an unmarked source got into a full anonymous slice")
589 }
590 // the device marks it (it just completed authenticated calls); now it has the reserved slice
591 dc.send("HINT 203.0.113.77 203.0.113.78 203.0.113.79 203.0.113.80")
592 eventually(t, "the hint is in the table", 3*time.Second, func() bool { return e.srv.hints.good(d.host, netip.MustParseAddr("203.0.113.80")) })
593 tc, err := e.clientFrom("203.0.113.77", d)
594 if err != nil {
595 t.Fatalf("a marked source with the anonymous slice full: %v", err)
596 }
597 defer tc.Close()
598 eventually(t, "the OPEN of the marked source", 3*time.Second, func() bool {
599 for _, o := range ds.openList() {
600 if o.str("peer_ip") == "203.0.113.77" {
601 return o.str("class") == "hint"
602 }
603 }
604 return false
605 })
606 // the marked sources can hold up to the whole host cap (16 held + 48 reserved), and no more
607 more := 1
608 for i := 0; i < 80; i++ {
609 if _, err := e.clientFrom(fmt.Sprintf("203.0.113.%d", 77+i%4), d); err == nil {
610 more++
611 }
612 }
613 if want := ctPolInt("node_limits.label_streams_max") - held; more != want {
614 t.Fatalf("the marked sources held %d more streams, want %d (the whole cap of the host)", more, want)
615 }
616 // the mark lasts hint_ttl_s and no longer
617 e.offsetNS.Add(int64(DefaultConfig().HintTTL + time.Minute))
618 if e.srv.hints.good(d.host, netip.MustParseAddr("203.0.113.77")) {
619 t.Fatal("the mark outlived its TTL")
620 }
621}
622
623func TestNode_AHintMarksAddressesOnlyForTheHostOfTheDeviceThatSentIt(t *testing.T) {
624 e := newVIPEnv(t, nil)
625 a, b := e.newDevice(), e.newDevice()
626 ca, cb := e.connectOK(a), e.connectOK(b)
627 ca.serve(nil)
628 cb.serve(nil)
629 ca.send("HINT 203.0.113.50 2001:db8::50")
630 eventually(t, "the hint", 3*time.Second, func() bool { return e.srv.hints.good(a.host, netip.MustParseAddr("203.0.113.50")) })
631 if e.srv.hints.good(b.host, netip.MustParseAddr("203.0.113.50")) {
632 t.Fatal("a device marked an address for another device's host")
633 }
634 if !e.srv.hints.good(a.host, netip.MustParseAddr("2001:db8::50")) {
635 t.Fatal("the IPv6 address was not marked")
636 }
637 // a HINT before the OK marks nothing: the connection has no host yet
638 c := e.connect(e.newDevice())
639 c.send("HINT 203.0.113.51")
640 time.Sleep(100 * time.Millisecond)
641 if e.srv.hints.size() != 2 {
642 t.Fatalf("%d addresses in the table", e.srv.hints.size())
643 }
644}
645
646func TestNode_PerHostNewStreamRateIsKeptForSourcesNotMarkedGood(t *testing.T) {
647 release := make(chan struct{})
648 e := newVIPEnv(t, func(e *testEnv, c *Config) { c.LabelNewPerS, c.LabelNewBurst = 2, 3 })
649 d := e.newDevice()
650 dc := e.connectOK(d)
651 dc.serve(holdHandler(d, release))
652 defer close(release)
653 ok := 0
654 for i := 0; i < 8; i++ {
655 if _, err := e.clientFrom(fmt.Sprintf("198.51.100.%d", 10+i), d); err == nil {
656 ok++
657 }
658 }
659 if ok != 3 {
660 t.Fatalf("%d streams began in an instant, want the burst of 3", ok)
661 }
662 if e.rejected("label_rate") != 5 {
663 t.Fatalf("counted %d", e.rejected("label_rate"))
664 }
665 // a marked source is not paced by it (the stream cap still applies)
666 dc.send("HINT 203.0.113.1")
667 eventually(t, "the hint", 3*time.Second, func() bool { return e.srv.hints.good(d.host, netip.MustParseAddr("203.0.113.1")) })
668 for i := 0; i < 6; i++ {
669 if _, err := e.clientFrom("203.0.113.1", d); err != nil {
670 t.Fatalf("marked stream %d: %v", i, err)
671 }
672 }
673 // and time refills the bucket
674 e.offsetNS.Add(int64(2 * time.Second))
675 if _, err := e.clientFrom("198.51.100.99", d); err != nil {
676 t.Fatalf("after a second the bucket had tokens again: %v", err)
677 }
678}
679
680func TestNode_TheGlobalBudgetKeepsAShareForOpenAIAndMarkedSources(t *testing.T) {
681 release := make(chan struct{})
682 e := newVIPEnv(t, func(e *testEnv, c *Config) {
683 c.MaxStreams, c.OpenAIReservedPct, c.LabelAnonSlice = 8, 25, 64
684 c.PrefixFetch = func(context.Context) ([]byte, error) {
685 return []byte(`{"prefixes":[{"ipv4Prefix":"192.0.2.0/24"}]}`), nil
686 }
687 })
688 eventually(t, "the prefix list is in", 3*time.Second, func() bool { return e.srv.openai.size() == 1 })
689 d := e.newDevice()
690 dc := e.connectOK(d)
691 ds := dc.serve(holdHandler(d, release))
692 defer close(release)
693 anon := 0
694 for i := 0; i < 12; i++ {
695 if _, err := e.clientFrom(fmt.Sprintf("198.51.100.%d", 10+i), d); err == nil {
696 anon++
697 }
698 }
699 if anon != 6 {
700 t.Fatalf("anonymous sources held %d of 8, want 6 (25%% reserved)", anon)
701 }
702 if e.logs.Count(`"reason":"anon_global_limit"`) < 1 {
703 t.Fatalf("logs:\n%s", e.logs.String())
704 }
705 for i := 0; i < 2; i++ {
706 if _, err := e.clientFrom("192.0.2.9", d); err != nil {
707 t.Fatalf("OpenAI stream %d: %v", i, err)
708 }
709 }
710 if _, err := e.clientFrom("192.0.2.10", d); err == nil {
711 t.Fatal("the budget of 8 was exceeded")
712 }
713 classes := map[string]int{}
714 for _, o := range ds.openList() {
715 classes[o.str("class")]++
716 }
717 if classes["openai"] != 2 || classes["anon"] != 6 {
718 t.Fatalf("classes in the OPENs: %v", classes)
719 }
720}
721
722func TestNode_EverySocketItAcceptsHasTCPKeepalive(t *testing.T) {
723 cfg := DefaultConfig()
724 cfg.TCPKeepalive = 7 * time.Second
725 s := &Server{cfg: cfg}
726 lc := s.listenConfig()
727 ln, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0")
728 if err != nil {
729 t.Fatal(err)
730 }
731 defer ln.Close()
732 accepted := make(chan net.Conn, 1)
733 go func() {
734 c, _ := ln.Accept()
735 accepted <- c
736 }()
737 c, err := net.Dial("tcp", ln.Addr().String())
738 if err != nil {
739 t.Fatal(err)
740 }
741 defer c.Close()
742 srv := <-accepted
743 defer srv.Close()
744 raw, err := srv.(*net.TCPConn).SyscallConn()
745 if err != nil {
746 t.Fatal(err)
747 }
748 var on int
749 var gerr error
750 _ = raw.Control(func(fd uintptr) {
751 on, gerr = syscall.GetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_KEEPALIVE)
752 })
753 if gerr != nil || on == 0 {
754 t.Fatalf("SO_KEEPALIVE = %d, %v", on, gerr)
755 }
756 if got := s.listenConfig().KeepAliveConfig; !got.Enable || got.Idle != 7*time.Second || got.Interval != 7*time.Second || got.Count != keepaliveProbes {
757 t.Fatalf("%+v", got)
758 }
759}
760
761// ------------------------------------------------------------------------------------------------------------------------------------
762// The review's findings: a device's own connections, the overflow entry, the addresses of devices
763// ------------------------------------------------------------------------------------------------------------------------------------
764
765// simultaneousClients opens one stream to d from each address at the same time and returns how many were served and what failed.
766func simultaneousClients(e *testEnv, d *testDevice, addrs []string) (ok int, failed []error) {
767 var wg sync.WaitGroup
768 var mu sync.Mutex
769 for _, ip := range addrs {
770 wg.Add(1)
771 go func(ip string) {
772 defer wg.Done()
773 _, err := e.clientFrom(ip, d)
774 mu.Lock()
775 defer mu.Unlock()
776 if err != nil {
777 failed = append(failed, err)
778 } else {
779 ok++
780 }
781 }(ip)
782 }
783 wg.Wait()
784 return ok, failed
785}
786
787func TestNode_ADevicesOwnConnectionsAreNotChargedToTheAnonymousBucketAtPolicyLimits(t *testing.T) {
788 release := make(chan struct{})
789 e := newVIPEnv(t, nil) // the policy's per-source numbers: 16 held and 10 new a second per anonymous source
790 d := e.newDevice()
791 dc := e.connectOK(d)
792 ds := dc.serve(holdHandler(d, release))
793 defer close(release)
794 // 32 sources the device marked good: only the device's own ATTACH connections (one per stream, all from its one address) are in question here
795 var addrs []string
796 var marked []string
797 for i := 0; i < 32; i++ {
798 addrs = append(addrs, fmt.Sprintf("203.0.113.%d", 1+i))
799 marked = append(marked, addrs[i])
800 }
801 dc.send("HINT " + strings.Join(marked[:16], " "))
802 dc.send("HINT " + strings.Join(marked[16:], " "))
803 eventually(t, "the hints", 3*time.Second, func() bool { return e.srv.hints.good(d.host, netip.MustParseAddr(addrs[31])) })
804
805 ok, failed := simultaneousClients(e, d, addrs)
806 if ok != 32 {
807 t.Fatalf("%d of 32 simultaneous streams were served (%v): the device's ATTACHes were refused as an anonymous source's connections (ip_rate %d, ip_concurrent %d)",
808 ok, failed, e.rejected(refuseRate), e.rejected(refuseConcurrent))
809 }
810 if e.rejected(refuseRate)+e.rejected(refuseConcurrent) != 0 {
811 t.Fatalf("refused at accept: ip_rate %d, ip_concurrent %d", e.rejected(refuseRate), e.rejected(refuseConcurrent))
812 }
813 if ds.openCount() != 32 {
814 t.Fatalf("%d OPENs", ds.openCount())
815 }
816 // and the same device keeps its larger allowance only while it holds an authenticated control connection: when it is gone the address is anonymous again
817 _ = dc.c.Close()
818 eventually(t, "the device is gone", 3*time.Second, func() bool { return e.srv.controlCount() == 0 })
819 eventually(t, "the address is no longer a device's", 3*time.Second, func() bool { return !e.srv.ips.isDevice(netip.MustParseAddr("127.0.0.1")) })
820}
821
822func TestIPLimits_ADeviceAddressHasLargerLimitsAndItsAuthenticatedConnectionsGiveTheirTokenBack(t *testing.T) {
823 l, _ := newTestLimits(t, func(c *Config) { c.PerIPConcurrent, c.PerIPNewPerS, c.DevicePerIPFactor = 100, 3, 4 })
824 // an anonymous source is held to 3 new connections in an instant
825 a := netip.MustParseAddr("198.51.100.7")
826 for i := 0; i < 3; i++ {
827 if _, why := l.admit(a, false); why != "" {
828 t.Fatal(why)
829 }
830 }
831 if _, why := l.admit(a, false); why != refuseRate {
832 t.Fatalf("%q", why)
833 }
834 // the address of a device has 4 times that (12), from the moment its control connection authenticates, though its bucket was made before
835 b := netip.MustParseAddr("198.51.100.8")
836 if _, why := l.admit(b, false); why != "" { // the control connection, admitted as any source's
837 t.Fatal(why)
838 }
839 unmark := l.markDevice(b)
840 if !l.isDevice(b) {
841 t.Fatal("not a device")
842 }
843 var rels []func()
844 for i := 0; i < 12; i++ {
845 rel, why := l.admit(b, false)
846 if why != "" {
847 t.Fatalf("connection %d of the device: %q", i+1, why)
848 }
849 rels = append(rels, rel)
850 }
851 if _, why := l.admit(b, false); why != refuseRate {
852 t.Fatalf("the thirteenth after the mark: %q", why)
853 }
854 // the ones that proved who they are (attached) give their token back: all of them are in the bucket again
855 for _, rel := range rels {
856 l.refund(b, false)
857 rel()
858 }
859 for i := 0; i < 12; i++ {
860 if _, why := l.admit(b, false); why != "" {
861 t.Fatalf("after the refunds, connection %d: %q", i+1, why)
862 }
863 }
864 // a refund never fills the bucket past its burst
865 for i := 0; i < 50; i++ {
866 l.refund(b, false)
867 }
868 if got := l.entries[l.keyOf(b)].tokens; got > 12 {
869 t.Fatalf("%v tokens in a bucket of 12", got)
870 }
871 unmark()
872 unmark() // once
873 if l.isDevice(b) {
874 t.Fatal("still a device after its connection ended")
875 }
876 // two devices behind one address: it stays a device until the last one is gone
877 u1, u2 := l.markDevice(b), l.markDevice(b)
878 u1()
879 if !l.isDevice(b) {
880 t.Fatal("one device left, the address is no longer one")
881 }
882 u2()
883 if l.isDevice(b) || len(l.devices) != 0 {
884 t.Fatal("the table of devices did not empty")
885 }
886}
887
888func TestIPLimits_TheOverflowEntryIsNeverStruckNorBanned(t *testing.T) {
889 l, _ := newTestLimits(t, func(c *Config) { c.IPTableMax, c.PerIPConcurrent, c.PerIPNewPerS, c.UnknownSNIPerIP = 4, 1, 1000, 3 })
890 for i := 0; i < 4; i++ { // four sources that hold connections fill the table and cannot be swept
891 if _, why := l.admit(netip.MustParseAddr(fmt.Sprintf("198.51.100.%d", i+1)), false); why != "" {
892 t.Fatal(why)
893 }
894 }
895 scanner := netip.MustParseAddr("203.0.113.9")
896 for i := 0; i < 10; i++ {
897 if l.strike(scanner, fmt.Sprintf("n%d.example.org", i), false) {
898 t.Fatalf("a scanner inside the overflow banned the whole crowd with name %d", i)
899 }
900 }
901 if _, _, bans, _ := l.snapshot(); bans != 0 {
902 t.Fatalf("%d bans", bans)
903 }
904 // an innocent new source (it too lands in the overflow) is served as before
905 if _, why := l.admit(netip.MustParseAddr("192.0.2.77"), false); why != "" {
906 t.Fatalf("an innocent new source was refused: %q", why)
907 }
908 // a source with an entry of its own is still struck and banned
909 l2, _ := newTestLimits(t, func(c *Config) { c.UnknownSNIPerIP = 3 })
910 banned := false
911 for i := 0; i < 4; i++ {
912 banned = l2.strike(scanner, fmt.Sprintf("n%d.example.org", i), false)
913 }
914 if !banned {
915 t.Fatal("a source of its own was not banned")
916 }
917}
918
919func TestNode_ASourceADeviceMarkedGoodIsNeverBannedForTheNamesItAsks(t *testing.T) {
920 e := newVIPEnv(t, func(e *testEnv, c *Config) { c.UnknownSNIPerIP, c.PerIPNewPerS = 3, 1000 })
921 d := e.newDevice()
922 dc := e.connectOK(d)
923 dc.serve(nil)
924 dc.send("HINT 203.0.113.50")
925 eventually(t, "the hint", 3*time.Second, func() bool { return e.srv.hints.goodAnywhere(netip.MustParseAddr("203.0.113.50")) })
926 // a shared egress address that the fleet knows: many devices are offline at any moment, so it asks for many names
927 for i := 0; i < 10; i++ {
928 if answer, closed := e.helloFrom("203.0.113.50", fmt.Sprintf("off%d.%s", i, testZone), 500*time.Millisecond); closed && len(answer) == 0 {
929 t.Fatalf("name %d: the marked address was dropped without an answer", i)
930 }
931 }
932 if e.metricValue(`all_tunnel_ip_bans_total`) != 0 {
933 t.Fatal("a marked address was banned")
934 }
935 // the same behaviour from an address nobody marked is a scan
936 for i := 0; i < 10; i++ {
937 e.helloFrom("198.51.100.50", fmt.Sprintf("off%d.%s", i, testZone), 500*time.Millisecond)
938 }
939 if e.metricValue(`all_tunnel_ip_bans_total`) != 1 {
940 t.Fatal("an unmarked scanner was not banned")
941 }
942}
943
944func TestNode_AMetricsPortThatIsTakenDoesNotKeepTheNodeFromStarting(t *testing.T) {
945 taken, err := net.Listen("tcp", "127.0.0.1:0")
946 if err != nil {
947 t.Fatal(err)
948 }
949 defer taken.Close()
950 e := newVIPEnv(t, func(e *testEnv, c *Config) { c.MetricsAddr = taken.Addr().String() })
951 if e.srv.MetricsAddr() != "" {
952 t.Fatalf("metrics listen on %s", e.srv.MetricsAddr())
953 }
954 if e.logs.Count(`"msg":"metrics_unavailable"`) != 1 || strings.Contains(e.logs.String(), taken.Addr().String()) {
955 t.Fatalf("logs:\n%s", e.logs.String())
956 }
957 d := e.newDevice()
958 e.connectOK(d).serve(nil)
959 if _, err := e.clientFrom("198.51.100.5", d); err != nil {
960 t.Fatalf("the node does not serve: %v", err)
961 }
962}
963
964// The embedded policy is a PROJECTION of contracts/policy.json once a runtime's file holds only the sections it reads: a section or key the node reads
965// that the projection lacks would read as zero. This holds for every json field of Policy, whatever the projection is.
966func TestPolicyCopyHoldsEveryKeyTheNodeReads(t *testing.T) {
967 var raw map[string]any
968 if err := json.Unmarshal(policyJSON, &raw); err != nil {
969 t.Fatal(err)
970 }
971 var walk func(path string, typ reflect.Type, at any)
972 walk = func(path string, typ reflect.Type, at any) {
973 for i := 0; i < typ.NumField(); i++ {
974 f := typ.Field(i)
975 name, _, _ := strings.Cut(f.Tag.Get("json"), ",")
976 if name == "" || name == "-" {
977 continue
978 }
979 obj, _ := at.(map[string]any)
980 got, ok := obj[name]
981 if !ok {
982 t.Errorf("tunnel/policy.json has no %s.%s, which the node reads (add it to the node's projection in the contract lint table and sync)", path, name)
983 continue
984 }
985 if f.Type.Kind() == reflect.Struct {
986 walk(path+"."+name, f.Type, got)
987 }
988 }
989 }
990 walk("policy", reflect.TypeOf(Policy{}), raw)
991}