1package traffic
2
3import (
4 "context"
5 "net"
6 "net/netip"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/assert"
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/gordon/internal/domain"
15)
16
17func TestUDPPassthroughEcho(t *testing.T) {
18 backend := startUDPEchoServer(t)
19 graph := udpGraph(t, freeUDPAddress(t), backend.address)
20 manager := NewManager()
21 require.NoError(t, manager.Apply(context.Background(), &graph))
22 defer shutdownManager(t, manager)
23
24 conn := dialUDP(t, graph.EntryPoints[0].Address)
25 defer conn.Close()
26 assertUDPRoundTrip(t, conn, "hello udp")
27}
28
29func TestUDPRejectsUntrustedCIDR(t *testing.T) {
30 backend := startUDPEchoServer(t)
31 graph := udpGraph(t, freeUDPAddress(t), backend.address)
32 graph.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
33 manager := NewManager()
34 require.NoError(t, manager.Apply(context.Background(), &graph))
35 defer shutdownManager(t, manager)
36
37 assertUDPRejected(t, manager, graph.EntryPoints[0].Address, 1)
38}
39
40func TestUDPReloadAppliesTrustedCIDRChange(t *testing.T) {
41 backend := startUDPEchoServer(t)
42 graph := udpGraph(t, freeUDPAddress(t), backend.address)
43 graph.EntryPoints[0].TrustedCIDRs = []string{"127.0.0.0/8"}
44 manager := NewManager()
45 require.NoError(t, manager.Apply(context.Background(), &graph))
46 defer shutdownManager(t, manager)
47
48 conn := dialUDP(t, graph.EntryPoints[0].Address)
49 assertUDPRoundTrip(t, conn, "allowed")
50 require.NoError(t, conn.Close())
51
52 updated := graph
53 updated.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
54 require.NoError(t, manager.Apply(context.Background(), &updated))
55 assertUDPRejected(t, manager, graph.EntryPoints[0].Address, 1)
56}
57
58func TestUDPReloadClosesActiveUntrustedSessions(t *testing.T) {
59 backend := startUDPEchoServer(t)
60 graph := udpGraph(t, freeUDPAddress(t), backend.address)
61 graph.EntryPoints[0].TrustedCIDRs = []string{"127.0.0.0/8"}
62 manager := NewManager()
63 require.NoError(t, manager.Apply(context.Background(), &graph))
64 defer shutdownManager(t, manager)
65
66 conn := dialUDP(t, graph.EntryPoints[0].Address)
67 defer conn.Close()
68 assertUDPRoundTrip(t, conn, "before")
69 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 1 }, time.Second, 10*time.Millisecond)
70
71 updated := graph
72 updated.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
73 require.NoError(t, manager.Apply(context.Background(), &updated))
74
75 assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
76 _, err := conn.Write([]byte("after"))
77 require.NoError(t, err)
78 _ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
79 buf := make([]byte, 32)
80 _, err = conn.Read(buf)
81 require.Error(t, err)
82}
83
84func TestUDPReloadReplacesSessionWhenBackendChanges(t *testing.T) {
85 backendA := startUDPEchoServerWithPrefix(t, "a:")
86 backendB := startUDPEchoServerWithPrefix(t, "b:")
87 graph := udpGraph(t, freeUDPAddress(t), backendA.address)
88 manager := NewManager()
89 require.NoError(t, manager.Apply(context.Background(), &graph))
90 defer shutdownManager(t, manager)
91
92 conn := dialUDP(t, graph.EntryPoints[0].Address)
93 defer conn.Close()
94 assertUDPRoundTripWant(t, conn, "first", "a:first")
95
96 updated := graph
97 backend, err := backendFromAddress("echo:udp", backendB.address)
98 require.NoError(t, err)
99 backend.Protocol = domain.NetworkProtocolUDP
100 updated.Services[0].Backends = []domain.TrafficBackend{backend}
101 require.NoError(t, manager.Apply(context.Background(), &updated))
102
103 assertUDPRoundTripWant(t, conn, "second", "a:second")
104 assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
105
106 newConn := dialUDP(t, graph.EntryPoints[0].Address)
107 defer newConn.Close()
108 assertUDPRoundTripWant(t, newConn, "third", "b:third")
109}
110
111func TestUDPStaleRemovalKeepsReplacementSession(t *testing.T) {
112 clientAddr := netip.MustParseAddrPort("127.0.0.1:4100")
113 stale := newTestUDPSession(t, clientAddr)
114 replacement := newTestUDPSession(t, clientAddr)
115 runtime, _ := newTestUDPRuntime(t, freeUDPAddress(t))
116 runtime.mu.Lock()
117 runtime.sessions[clientAddr] = replacement
118 runtime.counters.activeUDPSessions.Store(1)
119 runtime.mu.Unlock()
120
121 // A stale remover carries the pointer of the session that used to own the
122 // client key. Expiry, backend read/write failures, reload and CIDR cleanup
123 // all funnel through removeSession, so none of them may evict the session
124 // that replaced it.
125 runtime.removeSession(stale)
126
127 runtime.mu.Lock()
128 current, ok := runtime.sessions[clientAddr]
129 runtime.mu.Unlock()
130 require.True(t, ok, "replacement session must stay registered")
131 require.Same(t, replacement, current)
132 require.Equal(t, int64(1), runtime.counters.activeUDPSessions.Load(), "stale removal must not decrement the active counter")
133 require.False(t, testUDPSessionClosed(replacement), "replacement backend must stay open")
134
135 runtime.removeSession(replacement)
136
137 runtime.mu.Lock()
138 _, ok = runtime.sessions[clientAddr]
139 runtime.mu.Unlock()
140 require.False(t, ok, "removing the current session must unregister it")
141 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load(), "removing the current session must decrement once")
142 require.True(t, testUDPSessionClosed(replacement))
143
144 // Removing again must be idempotent: no double decrement and no panic.
145 runtime.removeSession(replacement)
146 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
147}
148
149func TestUDPExpiryClosesOnlyRegisteredIdleSessionOnce(t *testing.T) {
150 clientAddr := netip.MustParseAddrPort("127.0.0.1:4200")
151 // stale was replaced under the same key and is no longer registered.
152 stale := newTestUDPSession(t, clientAddr)
153 stale.lastSeen.Store(time.Now().Add(-time.Hour).UnixNano())
154 current := newTestUDPSession(t, clientAddr)
155 runtime, _ := newTestUDPRuntime(t, freeUDPAddress(t))
156 runtime.mu.Lock()
157 runtime.sessions[clientAddr] = current
158 runtime.counters.activeUDPSessions.Store(1)
159 runtime.mu.Unlock()
160
161 runtime.expireIdleSessions(time.Minute)
162 require.False(t, testUDPSessionClosed(current), "fresh registered session must not expire")
163 require.False(t, testUDPSessionClosed(stale), "an unregistered session must never be closed by expiry")
164 require.Equal(t, int64(1), runtime.counters.activeUDPSessions.Load())
165
166 current.lastSeen.Store(time.Now().Add(-time.Hour).UnixNano())
167 runtime.expireIdleSessions(time.Minute)
168
169 runtime.mu.Lock()
170 _, ok := runtime.sessions[clientAddr]
171 runtime.mu.Unlock()
172 require.False(t, ok)
173 require.True(t, testUDPSessionClosed(current))
174 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load(), "expiry must decrement exactly once")
175
176 // The backend loop of the expired session also removes it on return; the
177 // identity check keeps the counter coherent and must not double decrement.
178 runtime.removeSession(current)
179 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
180}
181
182func TestUDPAdmissionGateBlocksPublicationAfterClosingBegins(t *testing.T) {
183 backend := startUDPEchoServer(t)
184 runtime, _ := newTestUDPRuntime(t, backend.address)
185 // Model the instant stop() closes admission but before it cancels the
186 // runtime context: a datagram worker that reaches session() now must not
187 // publish a new session.
188 runtime.mu.Lock()
189 runtime.accepting = false
190 runtime.mu.Unlock()
191 runtime.closed.Store(true)
192
193 session, ok := runtime.session(netip.MustParseAddrPort("127.0.0.1:4300"), effectiveUDPOptions(domain.UDPOptions{}))
194 require.False(t, ok)
195 require.Nil(t, session)
196
197 runtime.mu.Lock()
198 require.Empty(t, runtime.sessions)
199 runtime.mu.Unlock()
200 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
201 requireUDPRuntimeGoroutinesDone(t, runtime)
202}
203
204func TestUDPPublishSessionRejectsCompletedDialAfterAdmissionClosure(t *testing.T) {
205 runtime, _ := newTestUDPRuntime(t, freeUDPAddress(t))
206 clientAddr := netip.MustParseAddrPort("127.0.0.1:4800")
207 dialed := newTestUDPSession(t, clientAddr)
208 runtime.mu.Lock()
209 runtime.accepting = false
210 runtime.mu.Unlock()
211 runtime.closed.Store(true)
212
213 session, ok := runtime.publishSession(dialed, effectiveUDPOptions(domain.UDPOptions{}))
214 require.False(t, ok)
215 require.Nil(t, session)
216 require.True(t, testUDPSessionClosed(dialed), "a rejected dial must have its backend closed")
217 runtime.mu.Lock()
218 require.Empty(t, runtime.sessions)
219 runtime.mu.Unlock()
220 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
221 require.Equal(t, int64(0), runtime.counters.totalAccepted.Load())
222 requireUDPRuntimeGoroutinesDone(t, runtime)
223}
224
225func TestUDPPublishSessionReusesExistingAndTouchesIt(t *testing.T) {
226 runtime, _ := newTestUDPRuntime(t, freeUDPAddress(t))
227 clientAddr := netip.MustParseAddrPort("127.0.0.1:4900")
228 existing := newTestUDPSession(t, clientAddr)
229 existing.lastSeen.Store(time.Now().Add(-time.Hour).UnixNano())
230 dialed := newTestUDPSession(t, clientAddr)
231 runtime.mu.Lock()
232 runtime.sessions[clientAddr] = existing
233 runtime.counters.activeUDPSessions.Store(1)
234 runtime.mu.Unlock()
235
236 session, ok := runtime.publishSession(dialed, effectiveUDPOptions(domain.UDPOptions{}))
237 require.True(t, ok)
238 require.Same(t, existing, session)
239 require.True(t, testUDPSessionClosed(dialed), "the losing dial must have its backend closed")
240 require.False(t, testUDPSessionClosed(existing))
241 require.Greater(t, existing.lastSeen.Load(), time.Now().Add(-time.Minute).UnixNano(), "reuse must touch the existing session")
242 runtime.mu.Lock()
243 current := runtime.sessions[clientAddr]
244 runtime.mu.Unlock()
245 require.Same(t, existing, current)
246 require.Equal(t, int64(1), runtime.counters.activeUDPSessions.Load(), "reuse must not add a session")
247 require.Equal(t, int64(0), runtime.counters.totalAccepted.Load(), "reuse must not count as a new accept")
248 requireUDPRuntimeGoroutinesDone(t, runtime)
249}
250
251func TestUDPRemoveSessionNilIsSafe(t *testing.T) {
252 runtime, _ := newTestUDPRuntime(t, freeUDPAddress(t))
253 require.NotPanics(t, func() { runtime.removeSession(nil) })
254 runtime.mu.Lock()
255 require.Empty(t, runtime.sessions)
256 runtime.mu.Unlock()
257 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
258}
259
260func TestUDPReadLoopRejectsInvalidSourceAddress(t *testing.T) {
261 backend := startUDPEchoServer(t)
262 packetConn := newScriptedPacketConn()
263 runtime, _ := newTestUDPRuntimeWithPacketConn(t, backend.address, packetConn)
264 runtime.start()
265
266 packetConn.packets <- scriptedUDPPacket{payload: []byte("zero"), addr: &net.UDPAddr{}}
267 packetConn.packets <- scriptedUDPPacket{payload: []byte("junk"), addr: testUDPNetAddr{network: "udp", address: "not-an-address"}}
268 require.Eventually(t, func() bool { return runtime.counters.totalRefused.Load() == 2 }, time.Second, 5*time.Millisecond)
269 runtime.mu.Lock()
270 require.Empty(t, runtime.sessions, "invalid source addresses must never open a session")
271 runtime.mu.Unlock()
272 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
273
274 // A valid datagram after the invalid ones proves the read loop keeps serving.
275 packetConn.packets <- scriptedUDPPacket{payload: []byte("ok"), addr: &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 43210}}
276 require.Eventually(t, func() bool { return runtime.counters.activeUDPSessions.Load() == 1 }, time.Second, 5*time.Millisecond)
277 require.Equal(t, int64(1), runtime.counters.totalAccepted.Load())
278}
279
280func TestUDPStopBeforeStartLeavesNoRuntimeGoroutines(t *testing.T) {
281 backend := startUDPEchoServer(t)
282 runtime, _ := newTestUDPRuntime(t, backend.address)
283 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
284 defer cancel()
285
286 runtime.stop(ctx, 50*time.Millisecond)
287 // Starting a runtime whose admission already closed must not launch workers.
288 runtime.start()
289
290 runtime.mu.Lock()
291 require.Empty(t, runtime.sessions)
292 runtime.mu.Unlock()
293 requireUDPRuntimeGoroutinesDone(t, runtime)
294 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
295}
296
297func TestUDPShutdownLeavesNoSessionsOrRuntimeGoroutines(t *testing.T) {
298 backend := startUDPEchoServer(t)
299 graph := udpGraph(t, freeUDPAddress(t), backend.address)
300 graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
301 manager := NewManager()
302 require.NoError(t, manager.Apply(context.Background(), &graph))
303
304 conn := dialUDP(t, graph.EntryPoints[0].Address)
305 defer conn.Close()
306 assertUDPRoundTrip(t, conn, "held")
307 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 1 }, time.Second, 10*time.Millisecond)
308
309 manager.mu.Lock()
310 runtime := manager.udpListeners["udp"]
311 manager.mu.Unlock()
312 require.NotNil(t, runtime)
313
314 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
315 defer cancel()
316 require.NoError(t, manager.Shutdown(ctx))
317
318 runtime.mu.Lock()
319 require.Empty(t, runtime.sessions)
320 runtime.mu.Unlock()
321 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
322 requireUDPRuntimeGoroutinesDone(t, runtime)
323}
324
325func TestUDPConcurrentCreationDuringShutdownLeavesNoSessions(t *testing.T) {
326 for range 5 {
327 backend := startUDPEchoServer(t)
328 runtime, _ := newTestUDPRuntime(t, backend.address)
329 runtime.start()
330
331 var creators sync.WaitGroup
332 for worker := range 8 {
333 creators.Add(1)
334 go func(worker int) {
335 defer creators.Done()
336 for attempt := range 25 {
337 key := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(5000+worker*100+attempt))
338 runtime.session(key, effectiveUDPOptions(domain.UDPOptions{}))
339 }
340 }(worker)
341 }
342
343 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
344 runtime.stop(ctx, 50*time.Millisecond)
345 cancel()
346 creators.Wait()
347
348 runtime.mu.Lock()
349 require.Empty(t, runtime.sessions, "no session may be published after closing began")
350 runtime.mu.Unlock()
351 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
352 requireUDPRuntimeGoroutinesDone(t, runtime)
353 }
354}
355
356// TestUDPInFlightDialDoesNotPublishAfterAdmissionClosure blocks the backend dial
357// on the runtime's private dial seam, closes admission exactly as stop() does,
358// then lets the dial complete: session() must reject the completed dial without
359// publishing a session or a backend goroutine. The seam is instance-local, so no
360// package-global state is touched.
361func TestUDPInFlightDialDoesNotPublishAfterAdmissionClosure(t *testing.T) {
362 runtime, _ := newTestUDPRuntime(t, "127.0.0.1:15353")
363 release := make(chan struct{})
364 dialStarted := make(chan struct{}, 1)
365 liveConn := newLiveUDPConn(t)
366 runtime.dialContext = func(ctx context.Context, _, _ string) (net.Conn, error) {
367 select {
368 case dialStarted <- struct{}{}:
369 default:
370 }
371 select {
372 case <-release:
373 return liveConn, nil
374 case <-ctx.Done():
375 return nil, ctx.Err()
376 }
377 }
378
379 type result struct {
380 session *udpSession
381 ok bool
382 }
383 results := make(chan result, 1)
384 go func() {
385 session, ok := runtime.session(netip.MustParseAddrPort("127.0.0.1:4700"), effectiveUDPOptions(domain.UDPOptions{}))
386 results <- result{session: session, ok: ok}
387 }()
388
389 select {
390 case <-dialStarted:
391 case <-time.After(5 * time.Second):
392 t.Fatal("backend dial never reached the dial seam")
393 }
394
395 runtime.mu.Lock()
396 runtime.accepting = false
397 runtime.mu.Unlock()
398 runtime.closed.Store(true)
399 close(release)
400
401 select {
402 case got := <-results:
403 require.False(t, got.ok, "a dial completing after admission closure must not open a session")
404 require.Nil(t, got.session)
405 case <-time.After(5 * time.Second):
406 t.Fatal("in-flight session creation did not finish")
407 }
408
409 runtime.mu.Lock()
410 require.Empty(t, runtime.sessions)
411 runtime.mu.Unlock()
412 require.Equal(t, int64(0), runtime.counters.activeUDPSessions.Load())
413 require.Equal(t, int64(0), runtime.counters.totalAccepted.Load())
414 requireUDPRuntimeGoroutinesDone(t, runtime)
415}
416
417func TestUDPAddrPortNormalizesIPv4MappedAndPreservesZones(t *testing.T) {
418 plainV4, ok := udpAddrPort(&net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 9000})
419 require.True(t, ok)
420 mappedV4, ok := udpAddrPort(&net.UDPAddr{IP: net.ParseIP("::ffff:127.0.0.1"), Port: 9000})
421 require.True(t, ok)
422 wantV4 := netip.MustParseAddrPort("127.0.0.1:9000")
423 require.Equal(t, wantV4, plainV4)
424 require.Equal(t, wantV4, mappedV4, "IPv4-mapped addresses must share the IPv4 session key")
425
426 zoned, ok := udpAddrPort(&net.UDPAddr{IP: net.ParseIP("fe80::1"), Zone: "eth0", Port: 9001})
427 require.True(t, ok)
428 require.Equal(t, netip.MustParseAddrPort("[fe80::1%eth0]:9001"), zoned)
429 otherZone, ok := udpAddrPort(&net.UDPAddr{IP: net.ParseIP("fe80::1"), Zone: "eth1", Port: 9001})
430 require.True(t, ok)
431 require.NotEqual(t, zoned, otherZone, "IPv6 zones must remain part of the key")
432
433 parsed, ok := udpAddrPort(testUDPNetAddr{network: "udp", address: "[2001:db8::1]:9002"})
434 require.True(t, ok)
435 require.Equal(t, netip.MustParseAddrPort("[2001:db8::1]:9002"), parsed)
436
437 _, ok = udpAddrPort(testUDPNetAddr{network: "udp", address: "not-an-address"})
438 require.False(t, ok)
439}
440
441func TestUDPAddrPortRejectsInvalidAddresses(t *testing.T) {
442 _, ok := udpAddrPort(&net.UDPAddr{})
443 require.False(t, ok, "zero UDPAddr must be rejected")
444 _, ok = udpAddrPort(&net.UDPAddr{IP: nil, Port: 53})
445 require.False(t, ok, "UDPAddr without an IP must be rejected")
446 _, ok = udpAddrPort(testUDPNetAddr{network: "udp", address: "not-an-address"})
447 require.False(t, ok)
448
449 // The trusted-list path must not admit an invalid address either, even when
450 // the list is empty and would otherwise trust every client.
451 require.False(t, trustedAddrPort(nil, netip.AddrPort{}))
452}
453
454func TestUDPSessionPrecomputesImmutableRemoteAddr(t *testing.T) {
455 backend := startUDPEchoServer(t)
456 runtime, _ := newTestUDPRuntime(t, backend.address)
457 options := effectiveUDPOptions(domain.UDPOptions{})
458
459 v4, ok := runtime.session(netip.MustParseAddrPort("127.0.0.1:4600"), options)
460 require.True(t, ok)
461 require.NotNil(t, v4.remoteAddr)
462 require.Equal(t, "127.0.0.1:4600", v4.remoteAddr.String())
463 runtime.removeSession(v4)
464
465 zoned, ok := runtime.session(netip.MustParseAddrPort("[fe80::1%eth0]:4601"), options)
466 require.True(t, ok)
467 require.NotNil(t, zoned.remoteAddr)
468 require.Equal(t, netip.MustParseAddrPort("[fe80::1%eth0]:4601"), zoned.remoteAddr.AddrPort(), "precomputed destination must preserve the IPv6 zone")
469 runtime.removeSession(zoned)
470}
471
472func TestUDPTrustedAddrPortMatchesCIDRs(t *testing.T) {
473 ipv4Trusted, err := parseTrustedCIDRs([]string{"127.0.0.0/8"})
474 require.NoError(t, err)
475 require.True(t, trustedAddrPort(ipv4Trusted, netip.MustParseAddrPort("127.0.0.1:1")))
476 require.True(t, trustedAddrPort(ipv4Trusted, netip.MustParseAddrPort("[::ffff:127.0.0.1]:1")), "v4-mapped client must match IPv4 CIDR")
477 require.False(t, trustedAddrPort(ipv4Trusted, netip.MustParseAddrPort("192.0.2.1:1")))
478 require.False(t, trustedAddrPort(ipv4Trusted, netip.MustParseAddrPort("[2001:db8::1]:1")), "IPv6 client must not match IPv4 CIDR")
479
480 ipv6Trusted, err := parseTrustedCIDRs([]string{"2001:db8::/32"})
481 require.NoError(t, err)
482 require.True(t, trustedAddrPort(ipv6Trusted, netip.MustParseAddrPort("[2001:db8::1]:1")))
483 // Intentional UDP semantic: CIDR trust is evaluated on the IP only. The
484 // zone names the local interface a packet arrived on, not the peer, so a
485 // zoned and zoneless client of the same address are equally trusted. Zones
486 // stay part of the session key; do not "fix" this to compare zones.
487 require.True(t, trustedAddrPort(ipv6Trusted, netip.MustParseAddrPort("[2001:db8::1%eth0]:1")), "zone must not affect CIDR matching")
488 require.True(t, trustedAddrPort(ipv6Trusted, netip.MustParseAddrPort("[2001:db8::1%eth1]:1")), "any zone on a trusted address is trusted")
489 require.False(t, trustedAddrPort(ipv6Trusted, netip.MustParseAddrPort("[2001:dead::2%eth0]:1")), "an untrusted address is untrusted regardless of zone")
490 require.False(t, trustedAddrPort(ipv6Trusted, netip.MustParseAddrPort("127.0.0.1:1")))
491
492 require.True(t, trustedAddrPort(nil, netip.MustParseAddrPort("203.0.113.9:1")), "empty trusted list admits every client")
493 require.False(t, trustedAddrPort(ipv4Trusted, netip.AddrPort{}), "invalid client address is not trusted")
494}
495
496func assertUDPRejected(t *testing.T, manager *Manager, address string, refused int64) {
497 t.Helper()
498 conn := dialUDP(t, address)
499 defer conn.Close()
500 _, err := conn.Write([]byte("blocked"))
501 require.NoError(t, err)
502 _ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
503 buf := make([]byte, 32)
504 _, err = conn.Read(buf)
505 require.Error(t, err)
506 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == refused }, time.Second, 10*time.Millisecond)
507}
508
509func TestUDPTwoClientsGetIsolatedSessions(t *testing.T) {
510 backend := startUDPEchoServer(t)
511 graph := udpGraph(t, freeUDPAddress(t), backend.address)
512 manager := NewManager()
513 require.NoError(t, manager.Apply(context.Background(), &graph))
514 defer shutdownManager(t, manager)
515
516 first := dialUDP(t, graph.EntryPoints[0].Address)
517 defer first.Close()
518 second := dialUDP(t, graph.EntryPoints[0].Address)
519 defer second.Close()
520 assertUDPRoundTrip(t, first, "first")
521 assertUDPRoundTrip(t, second, "second")
522 assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 2 }, time.Second, 10*time.Millisecond)
523}
524
525func TestUDPIdleSessionExpires(t *testing.T) {
526 backend := startUDPEchoServer(t)
527 graph := udpGraph(t, freeUDPAddress(t), backend.address)
528 graph.Options.UDP.IdleTimeout = 40 * time.Millisecond
529 manager := NewManager()
530 require.NoError(t, manager.Apply(context.Background(), &graph))
531 defer shutdownManager(t, manager)
532
533 conn := dialUDP(t, graph.EntryPoints[0].Address)
534 defer conn.Close()
535 assertUDPRoundTrip(t, conn, "expire")
536 assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
537}
538
539func TestUDPMaxSessionsRejectsOverflow(t *testing.T) {
540 backend := startUDPEchoServer(t)
541 graph := udpGraph(t, freeUDPAddress(t), backend.address)
542 graph.Options.UDP.MaxSessions = 1
543 manager := NewManager()
544 require.NoError(t, manager.Apply(context.Background(), &graph))
545 defer shutdownManager(t, manager)
546
547 first := dialUDP(t, graph.EntryPoints[0].Address)
548 defer first.Close()
549 assertUDPRoundTrip(t, first, "first")
550
551 second := dialUDP(t, graph.EntryPoints[0].Address)
552 defer second.Close()
553 _, err := second.Write([]byte("second"))
554 require.NoError(t, err)
555 _ = second.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
556 buf := make([]byte, 32)
557 _, err = second.Read(buf)
558 require.Error(t, err)
559 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
560}
561
562func TestUDPRemovedRouterWithRetainedEntryPointDrainsSession(t *testing.T) {
563 backend := startUDPEchoServer(t)
564 graph := udpGraph(t, freeUDPAddress(t), backend.address)
565 graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
566 manager := NewManager()
567 require.NoError(t, manager.Apply(context.Background(), &graph))
568 defer shutdownManager(t, manager)
569
570 conn := dialUDP(t, graph.EntryPoints[0].Address)
571 defer conn.Close()
572 assertUDPRoundTrip(t, conn, "held")
573
574 next := graph
575 next.Routers = nil
576 next.Services = nil
577 require.NoError(t, manager.Apply(context.Background(), &next))
578 assertUDPRoundTrip(t, conn, "during-drain")
579 assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
580
581 _, err := conn.Write([]byte("after"))
582 require.NoError(t, err)
583 _ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
584 buf := make([]byte, 32)
585 _, err = conn.Read(buf)
586 require.Error(t, err)
587}
588
589func TestUDPBackendChangeDrainsExistingSession(t *testing.T) {
590 oldBackend := startUDPEchoServerWithPrefix(t, "old:")
591 newBackend := startUDPEchoServerWithPrefix(t, "new:")
592 graph := udpGraph(t, freeUDPAddress(t), oldBackend.address)
593 graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
594 manager := NewManager()
595 require.NoError(t, manager.Apply(context.Background(), &graph))
596 defer shutdownManager(t, manager)
597
598 oldConn := dialUDP(t, graph.EntryPoints[0].Address)
599 defer oldConn.Close()
600 assertUDPRoundTripWant(t, oldConn, "held", "old:held")
601
602 next := udpGraph(t, graph.EntryPoints[0].Address, newBackend.address)
603 next.Options = graph.Options
604 require.NoError(t, manager.Apply(context.Background(), &next))
605 assertUDPRoundTripWant(t, oldConn, "during-drain", "old:during-drain")
606 assert.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 0 }, time.Second, 10*time.Millisecond)
607
608 newConn := dialUDP(t, graph.EntryPoints[0].Address)
609 defer newConn.Close()
610 assertUDPRoundTripWant(t, newConn, "after", "new:after")
611}
612
613func TestUDPRemovedRouterDrainsThenClosesSessions(t *testing.T) {
614 backend := startUDPEchoServer(t)
615 graph := udpGraph(t, freeUDPAddress(t), backend.address)
616 graph.Options.UDP.DrainTimeout = 50 * time.Millisecond
617 manager := NewManager()
618 require.NoError(t, manager.Apply(context.Background(), &graph))
619 defer shutdownManager(t, manager)
620
621 conn := dialUDP(t, graph.EntryPoints[0].Address)
622 defer conn.Close()
623 assertUDPRoundTrip(t, conn, "held")
624 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveUDPSessions == 1 }, time.Second, 10*time.Millisecond)
625
626 empty := domain.TrafficGraph{Options: graph.Options}
627 started := time.Now()
628 require.NoError(t, manager.Apply(context.Background(), &empty))
629 assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
630 assert.Equal(t, int64(0), manager.Status().Counters.ActiveUDPSessions)
631}
632
633func TestUDPShutdownClosesSocket(t *testing.T) {
634 backend := startUDPEchoServer(t)
635 graph := udpGraph(t, freeUDPAddress(t), backend.address)
636 manager := NewManager()
637 require.NoError(t, manager.Apply(context.Background(), &graph))
638 require.NoError(t, manager.Shutdown(context.Background()))
639
640 conn := dialUDP(t, graph.EntryPoints[0].Address)
641 defer conn.Close()
642 _, err := conn.Write([]byte("closed"))
643 require.NoError(t, err)
644 _ = conn.SetReadDeadline(time.Now().Add(80 * time.Millisecond))
645 buf := make([]byte, 32)
646 _, err = conn.Read(buf)
647 require.Error(t, err)
648}
649
650func TestUDPStatusCountersTrackDatagrams(t *testing.T) {
651 backend := startUDPEchoServer(t)
652 graph := udpGraph(t, freeUDPAddress(t), backend.address)
653 manager := NewManager()
654 require.NoError(t, manager.Apply(context.Background(), &graph))
655 defer shutdownManager(t, manager)
656
657 conn := dialUDP(t, graph.EntryPoints[0].Address)
658 defer conn.Close()
659 assertUDPRoundTrip(t, conn, "bytes")
660
661 assert.Eventually(t, func() bool {
662 status := manager.Status()
663 return status.Counters.TotalAccepted == 1 &&
664 status.Counters.ActiveUDPSessions == 1 &&
665 status.Counters.BytesIn >= int64(len("bytes")) &&
666 status.Counters.BytesOut >= int64(len("bytes"))
667 }, time.Second, 10*time.Millisecond)
668}
669
670func udpGraph(t *testing.T, listenAddress string, backendAddress string) domain.TrafficGraph {
671 t.Helper()
672 backend, err := backendFromAddress("echo:udp", backendAddress)
673 require.NoError(t, err)
674 backend.Protocol = domain.NetworkProtocolUDP
675 ref := serviceRef("echo", "udp")
676 graph := domain.TrafficGraph{
677 Options: domain.TrafficOptions{UDP: domain.UDPOptions{IdleTimeout: time.Minute, DrainTimeout: 50 * time.Millisecond}},
678 EntryPoints: []domain.EntryPoint{{Name: "udp", Address: listenAddress, Protocol: domain.EntryPointProtocolUDP}},
679 Routers: []domain.TrafficRouter{{Name: "echo", EntryPoint: "udp", Protocol: domain.RouterProtocolUDP, Service: ref}},
680 Services: []domain.TrafficService{{Name: ref, Backends: []domain.TrafficBackend{backend}}},
681 }
682 require.NoError(t, graph.Validate())
683 return graph
684}
685
686func freeUDPAddress(t *testing.T) string {
687 t.Helper()
688 conn, err := net.ListenPacket("udp", "127.0.0.1:0")
689 require.NoError(t, err)
690 address := conn.LocalAddr().String()
691 require.NoError(t, conn.Close())
692 return address
693}
694
695func dialUDP(t *testing.T, address string) *net.UDPConn {
696 t.Helper()
697 udpAddr, err := net.ResolveUDPAddr("udp", address)
698 require.NoError(t, err)
699 conn, err := net.DialUDP("udp", nil, udpAddr)
700 require.NoError(t, err)
701 return conn
702}
703
704func assertUDPRoundTrip(t *testing.T, conn *net.UDPConn, message string) {
705 t.Helper()
706 assertUDPRoundTripWant(t, conn, message, message)
707}
708
709func assertUDPRoundTripWant(t *testing.T, conn *net.UDPConn, message string, want string) {
710 t.Helper()
711 _, err := conn.Write([]byte(message))
712 require.NoError(t, err)
713 _ = conn.SetReadDeadline(time.Now().Add(time.Second))
714 buf := make([]byte, 128)
715 n, err := conn.Read(buf)
716 require.NoError(t, err)
717 assert.Equal(t, want, string(buf[:n]))
718}
719
720// newTestUDPRuntime builds a UDP runtime backed by a valid graph snapshot
721// without binding through Manager.Apply, so tests can drive publication,
722// expiry and shutdown interleavings directly.
723func newTestUDPRuntime(t *testing.T, backendAddress string) (*udpEntryPointRuntime, *Manager) {
724 t.Helper()
725 packetConn, err := net.ListenPacket("udp", "127.0.0.1:0")
726 require.NoError(t, err)
727 return newTestUDPRuntimeWithPacketConn(t, backendAddress, packetConn)
728}
729
730// newTestUDPRuntimeWithPacketConn builds a runtime over a caller-supplied
731// PacketConn so the read loop can be driven deterministically.
732func newTestUDPRuntimeWithPacketConn(t *testing.T, backendAddress string, packetConn net.PacketConn) (*udpEntryPointRuntime, *Manager) {
733 t.Helper()
734 graph := udpGraph(t, freeUDPAddress(t), backendAddress)
735 manager := NewManager()
736 manager.snapshot.Store(&graph)
737 runtime := newUDPEntryPointRuntime(context.Background(), manager, graph.EntryPoints[0], packetConn, nil)
738 t.Cleanup(func() {
739 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
740 defer cancel()
741 runtime.stop(ctx, 50*time.Millisecond)
742 _ = packetConn.Close()
743 })
744 return runtime, manager
745}
746
747// scriptedPacketConn feeds the read loop a fixed sequence of packets, including
748// source addresses that cannot be converted to a session key.
749type scriptedUDPPacket struct {
750 payload []byte
751 addr net.Addr
752}
753
754type scriptedPacketConn struct {
755 packets chan scriptedUDPPacket
756 closed chan struct{}
757 once sync.Once
758}
759
760func newScriptedPacketConn() *scriptedPacketConn {
761 return &scriptedPacketConn{packets: make(chan scriptedUDPPacket, 16), closed: make(chan struct{})}
762}
763
764func (c *scriptedPacketConn) ReadFrom(p []byte) (int, net.Addr, error) {
765 select {
766 case packet := <-c.packets:
767 return copy(p, packet.payload), packet.addr, nil
768 case <-c.closed:
769 return 0, nil, net.ErrClosed
770 }
771}
772
773func (c *scriptedPacketConn) WriteTo(p []byte, _ net.Addr) (int, error) { return len(p), nil }
774
775func (c *scriptedPacketConn) Close() error {
776 c.once.Do(func() { close(c.closed) })
777 return nil
778}
779
780func (c *scriptedPacketConn) LocalAddr() net.Addr { return &net.UDPAddr{IP: net.IPv4zero} }
781func (c *scriptedPacketConn) SetDeadline(time.Time) error { return nil }
782func (c *scriptedPacketConn) SetReadDeadline(time.Time) error { return nil }
783func (c *scriptedPacketConn) SetWriteDeadline(time.Time) error { return nil }
784
785func newTestUDPSession(t *testing.T, clientAddr netip.AddrPort) *udpSession {
786 t.Helper()
787 session := &udpSession{clientAddr: clientAddr, remoteAddr: net.UDPAddrFromAddrPort(clientAddr), backend: newLiveUDPConn(t)}
788 session.touch()
789 return session
790}
791
792// newLiveUDPConn dials a live loopback UDP listener so probe writes never fail
793// with ICMP port-unreachable errors, and tears both down with the test.
794func newLiveUDPConn(t *testing.T) net.Conn {
795 t.Helper()
796 listener, err := net.ListenPacket("udp", "127.0.0.1:0")
797 require.NoError(t, err)
798 t.Cleanup(func() { _ = listener.Close() })
799 go func() {
800 buf := make([]byte, 64)
801 for {
802 if _, _, err := listener.ReadFrom(buf); err != nil {
803 return
804 }
805 }
806 }()
807 conn, err := net.Dial("udp", listener.LocalAddr().String())
808 require.NoError(t, err)
809 t.Cleanup(func() { _ = conn.Close() })
810 return conn
811}
812
813// testUDPSessionClosed probes the backend with a write. Synthetic sessions are
814// connected to a live loopback listener (see newLiveUDPConn), so an open socket
815// always accepts the probe and a closed one always errors, without the ICMP
816// port-unreachable flakiness of writing to an unbound port.
817func testUDPSessionClosed(session *udpSession) bool {
818 _, err := session.backend.Write([]byte("probe"))
819 return err != nil
820}
821
822func requireUDPRuntimeGoroutinesDone(t *testing.T, runtime *udpEntryPointRuntime) {
823 t.Helper()
824 done := make(chan struct{})
825 go func() {
826 runtime.runWG.Wait()
827 close(done)
828 }()
829 select {
830 case <-done:
831 case <-time.After(2 * time.Second):
832 t.Fatal("udp runtime goroutines did not complete")
833 }
834}
835
836type testUDPNetAddr struct {
837 network string
838 address string
839}
840
841func (a testUDPNetAddr) Network() string { return a.network }
842func (a testUDPNetAddr) String() string { return a.address }
843
844type udpEchoServer struct{ address string }
845
846func startUDPEchoServer(t *testing.T) udpEchoServer {
847 t.Helper()
848 return startUDPEchoServerWithPrefix(t, "")
849}
850
851func startUDPEchoServerWithPrefix(t *testing.T, prefix string) udpEchoServer {
852 t.Helper()
853 conn, err := net.ListenPacket("udp", "127.0.0.1:0")
854 require.NoError(t, err)
855 done := make(chan struct{})
856 go func() {
857 defer close(done)
858 buf := make([]byte, udpBufferSize)
859 for {
860 n, addr, err := conn.ReadFrom(buf)
861 if err != nil {
862 return
863 }
864 _, _ = conn.WriteTo(append([]byte(prefix), buf[:n]...), addr)
865 }
866 }()
867 t.Cleanup(func() {
868 _ = conn.Close()
869 <-done
870 })
871 return udpEchoServer{address: conn.LocalAddr().String()}
872}