1package traffic
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "crypto/tls"
8 "errors"
9 "fmt"
10 "io"
11 "net"
12 "net/http"
13 "sync/atomic"
14 "testing"
15 "time"
16
17 "github.com/stretchr/testify/assert"
18 "github.com/stretchr/testify/require"
19
20 "github.com/bnema/gordon/internal/domain"
21)
22
23func TestTrafficRuntimeDefaultsApplySafeLimits(t *testing.T) {
24 assert.Equal(t, 1024, effectiveTCPOptions(domain.TCPOptions{}).MaxConnections)
25 assert.Equal(t, 4096, effectiveUDPOptions(domain.UDPOptions{}).MaxSessions)
26}
27
28func TestBackendFromAddressRejectsOutOfRangePort(t *testing.T) {
29 _, err := backendFromAddress("bad:tcp", "127.0.0.1:70000")
30 require.Error(t, err)
31 assert.Contains(t, err.Error(), "outside 1-65535")
32}
33
34func TestSmartTCPDispatch(t *testing.T) {
35 t.Run("cleartext HTTP request reaches HTTP handler", func(t *testing.T) {
36 manager, graph, hits := startSmartTCP(t, nil, nil)
37 defer shutdownManager(t, manager)
38
39 conn := dialTCP(t, graph.EntryPoints[0].Address)
40 defer conn.Close()
41 _, err := conn.Write([]byte("GET /clear HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
42 require.NoError(t, err)
43 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
44 require.NoError(t, err)
45 body, err := io.ReadAll(resp.Body)
46 require.NoError(t, err)
47 require.NoError(t, resp.Body.Close())
48 assert.Contains(t, string(body), "http:/clear")
49 assert.Equal(t, int64(1), hits.httpHits.Load())
50 assert.Equal(t, int64(0), hits.rawHits.Load())
51 })
52
53 t.Run("h2c prior-knowledge reaches HTTP handler", func(t *testing.T) {
54 manager, graph, hits := startSmartTCP(t, nil, nil)
55 defer shutdownManager(t, manager)
56 conn := dialTCP(t, graph.EntryPoints[0].Address)
57 defer conn.Close()
58 _, err := conn.Write([]byte(h2cPreface))
59 require.NoError(t, err)
60 assert.Eventually(t, func() bool { return hits.httpHits.Load() == 1 }, time.Second, 10*time.Millisecond)
61 assert.Equal(t, int64(0), hits.rawHits.Load())
62 })
63
64 t.Run("HTTPS request reaches configured HTTPS fallback", func(t *testing.T) {
65 manager, graph, _ := startSmartTCP(t, nil, nil)
66 defer shutdownManager(t, manager)
67 assertHTTPSBody(t, graph.EntryPoints[0].Address, "app.example.com", "smart https")
68 })
69
70 t.Run("TLS passthrough SNI match wins before HTTPS fallback", func(t *testing.T) {
71 backend := startTLSGreetingServer(t, "raw.example.com", "raw tls\n")
72 manager, graph, _ := startSmartTCP(t, &smartTCPRoute{tlsSNI: "raw.example.com", backendAddress: backend.address}, nil)
73 defer shutdownManager(t, manager)
74 assertTLSGreeting(t, graph.EntryPoints[0].Address, "raw.example.com", "raw tls\n")
75 assertHTTPSBody(t, graph.EntryPoints[0].Address, "app.example.com", "smart https")
76 })
77
78 t.Run("no-SNI TLS reaches HTTPS fallback", func(t *testing.T) {
79 manager, graph, _ := startSmartTCP(t, nil, nil)
80 defer shutdownManager(t, manager)
81 assertHTTPSBodyNoSNI(t, graph.EntryPoints[0].Address, "smart https")
82 })
83
84 t.Run("unknown bytes reach raw fallback immediately when source policy allows", func(t *testing.T) {
85 backend := startTCPEchoServer(t, 0)
86 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, trustedCIDRs: []string{"127.0.0.0/8"}})
87 defer shutdownManager(t, manager)
88 conn := dialTCP(t, graph.EntryPoints[0].Address)
89 defer conn.Close()
90 assertRoundTrip(t, conn, "ssh-raw")
91 })
92
93 t.Run("unknown bytes refused without raw fallback", func(t *testing.T) {
94 manager, graph, _ := startSmartTCP(t, nil, nil)
95 defer shutdownManager(t, manager)
96 assertTCPRejected(t, manager, graph.EntryPoints[0].Address, 1)
97 })
98
99 t.Run("unknown bytes refused when raw fallback source policy denies peer", func(t *testing.T) {
100 backend := startTCPEchoServer(t, 0)
101 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, trustedCIDRs: []string{"192.0.2.0/24"}})
102 defer shutdownManager(t, manager)
103 assertTCPRejected(t, manager, graph.EntryPoints[0].Address, 1)
104 })
105
106 t.Run("malformed HTTP-looking bytes do not reach raw fallback", func(t *testing.T) {
107 backend := startTCPEchoServer(t, 0)
108 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, allowPublic: true})
109 defer shutdownManager(t, manager)
110 conn := dialTCP(t, graph.EntryPoints[0].Address)
111 defer conn.Close()
112 _, err := conn.Write([]byte("GET /broken\r\n"))
113 require.NoError(t, err)
114 _, err = bufio.NewReader(conn).ReadByte()
115 require.Error(t, err)
116 })
117
118 t.Run("malformed TLS-looking bytes do not reach raw fallback", func(t *testing.T) {
119 backend := startTCPEchoServer(t, 0)
120 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, allowPublic: true})
121 defer shutdownManager(t, manager)
122 conn := dialTCP(t, graph.EntryPoints[0].Address)
123 defer conn.Close()
124 _, err := conn.Write([]byte{22, 3, 3, 0, 8, 2, 0, 0, 4, 0, 0, 0, 0})
125 require.NoError(t, err)
126 _, err = bufio.NewReader(conn).ReadByte()
127 require.Error(t, err)
128 })
129
130 t.Run("PROXY v1 and v2 prefixes rejected", func(t *testing.T) {
131 manager, graph, _ := startSmartTCP(t, nil, nil)
132 defer shutdownManager(t, manager)
133 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, []byte("PROXY TCP4 127.0.0.1 127.0.0.1 1 2\r\n"))
134 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, append([]byte(proxyV2Signature), 0x21, 0x11, 0, 0))
135 })
136
137 t.Run("sniff deadline cleared before backend handoff", func(t *testing.T) {
138 backend := startTCPEchoServer(t, 150*time.Millisecond)
139 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, allowPublic: true})
140 defer shutdownManager(t, manager)
141 conn := dialTCP(t, graph.EntryPoints[0].Address)
142 defer conn.Close()
143 assertRoundTrip(t, conn, "deadline-clear")
144 })
145
146 t.Run("raw fallback trusted CIDR reload reuses same listener", func(t *testing.T) {
147 backend := startTCPEchoServer(t, 0)
148 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, trustedCIDRs: []string{"192.0.2.0/24"}})
149 defer shutdownManager(t, manager)
150 assertTCPRejected(t, manager, graph.EntryPoints[0].Address, 1)
151
152 initialRuntime := tcpRuntimeForTest(t, manager, "edge")
153 replacement := graph
154 replacement.EntryPoints = append([]domain.EntryPoint(nil), graph.EntryPoints...)
155 replacement.EntryPoints[0].RawFallbackTrustedCIDRs = []string{"127.0.0.0/8"}
156 replacement.Routers = append([]domain.TrafficRouter(nil), graph.Routers...)
157 replacement.Services = append([]domain.TrafficService(nil), graph.Services...)
158 require.NoError(t, manager.Apply(context.Background(), &replacement))
159 assert.Same(t, initialRuntime, tcpRuntimeForTest(t, manager, "edge"))
160
161 conn := dialTCP(t, replacement.EntryPoints[0].Address)
162 defer conn.Close()
163 assertRoundTrip(t, conn, "raw-reloaded")
164 })
165
166 t.Run("raw fallback connection closes when trusted CIDR no longer allows peer", func(t *testing.T) {
167 backend := startTCPEchoServer(t, 0)
168 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, trustedCIDRs: []string{"127.0.0.0/8"}})
169 defer shutdownManager(t, manager)
170
171 conn := dialTCP(t, graph.EntryPoints[0].Address)
172 defer conn.Close()
173 assertRoundTrip(t, conn, "before")
174
175 updated := graph
176 updated.EntryPoints = append([]domain.EntryPoint(nil), graph.EntryPoints...)
177 updated.EntryPoints[0].RawFallbackTrustedCIDRs = []string{"192.0.2.0/24"}
178 require.NoError(t, manager.Apply(context.Background(), &updated))
179
180 assert.Eventually(t, func() bool {
181 _, writeErr := conn.Write([]byte("after\n"))
182 if writeErr != nil {
183 return true
184 }
185 _ = conn.SetReadDeadline(time.Now().Add(10 * time.Millisecond))
186 _, readErr := bufio.NewReader(conn).ReadByte()
187 return readErr != nil
188 }, time.Second, 10*time.Millisecond)
189 })
190
191 t.Run("TLS ClientHello read after sniff is bounded", func(t *testing.T) {
192 manager, graph, _ := startSmartTCP(t, nil, nil)
193 defer shutdownManager(t, manager)
194 graph.Options.TCP.DialTimeout = 100 * time.Millisecond
195 require.NoError(t, manager.Apply(context.Background(), &graph))
196
197 conn := dialTCP(t, graph.EntryPoints[0].Address)
198 defer conn.Close()
199 start := time.Now()
200 _, err := conn.Write([]byte{22, 3, 3, 0, 42})
201 require.NoError(t, err)
202 require.NoError(t, conn.SetReadDeadline(time.Now().Add(750*time.Millisecond)))
203 _, err = bufio.NewReader(conn).ReadByte()
204 require.Error(t, err)
205 assert.Less(t, time.Since(start), 500*time.Millisecond)
206 })
207}
208
209func TestSetSmartTCPHTTPServerCopiesProtocols(t *testing.T) {
210 manager := NewManager()
211 protocols := new(http.Protocols)
212 protocols.SetHTTP1(true)
213 manager.SetSmartTCPHTTPServer("edge", http.NotFoundHandler(), protocols)
214
215 protocols.SetHTTP1(false)
216 protocols.SetUnencryptedHTTP2(true)
217
218 config, ok := manager.smartHTTPServer("edge")
219 require.True(t, ok)
220 assert.True(t, config.Protocols.HTTP1())
221 assert.False(t, config.Protocols.UnencryptedHTTP2())
222}
223
224func TestSmartTCPStatusCounters(t *testing.T) {
225 t.Run("HTTP accepted", func(t *testing.T) {
226 manager, graph, _ := startSmartTCP(t, nil, nil)
227 defer shutdownManager(t, manager)
228 conn := dialTCP(t, graph.EntryPoints[0].Address)
229 defer conn.Close()
230 _, err := conn.Write([]byte("GET /status HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
231 require.NoError(t, err)
232 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
233 require.NoError(t, err)
234 _, _ = io.ReadAll(resp.Body)
235 _ = resp.Body.Close()
236 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.HTTPAccepted == 1 })
237 })
238
239 t.Run("h2c accepted", func(t *testing.T) {
240 manager, graph, _ := startSmartTCP(t, nil, nil)
241 defer shutdownManager(t, manager)
242 conn := dialTCP(t, graph.EntryPoints[0].Address)
243 defer conn.Close()
244 _, err := conn.Write([]byte(h2cPreface))
245 require.NoError(t, err)
246 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.H2CAccepted == 1 })
247 })
248
249 t.Run("HTTPS fallback accepted", func(t *testing.T) {
250 manager, graph, _ := startSmartTCP(t, nil, nil)
251 defer shutdownManager(t, manager)
252 assertHTTPSBody(t, graph.EntryPoints[0].Address, "app.example.com", "smart https")
253 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.HTTPSFallbackAccepted == 1 })
254 })
255
256 t.Run("TLS passthrough accepted", func(t *testing.T) {
257 backend := startTLSGreetingServer(t, "raw.example.com", "raw tls\n")
258 manager, graph, _ := startSmartTCP(t, &smartTCPRoute{tlsSNI: "raw.example.com", backendAddress: backend.address}, nil)
259 defer shutdownManager(t, manager)
260 assertTLSGreeting(t, graph.EntryPoints[0].Address, "raw.example.com", "raw tls\n")
261 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.TLSPassthroughAccepted == 1 })
262 })
263
264 t.Run("raw fallback accepted", func(t *testing.T) {
265 backend := startTCPEchoServer(t, 0)
266 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, allowPublic: true})
267 defer shutdownManager(t, manager)
268 conn := dialTCP(t, graph.EntryPoints[0].Address)
269 defer conn.Close()
270 assertRoundTrip(t, conn, "raw")
271 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.RawFallbackAccepted == 1 })
272 })
273
274 t.Run("raw fallback backend dial failure is not accepted", func(t *testing.T) {
275 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: freeTCPAddress(t), allowPublic: true})
276 defer shutdownManager(t, manager)
277 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, []byte{0})
278 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.RawFallbackAccepted == 0 })
279 assert.Eventually(t, func() bool {
280 return manager.Status().Counters.TotalAccepted == 0 && manager.Status().Counters.TotalErrors == 1
281 }, time.Second, 10*time.Millisecond)
282 })
283
284 t.Run("refused and rejected classes", func(t *testing.T) {
285 backend := startTCPEchoServer(t, 0)
286 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, trustedCIDRs: []string{"192.0.2.0/24"}})
287 defer shutdownManager(t, manager)
288 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, []byte("PROXY TCP4 127.0.0.1 127.0.0.1 1 2\r\n"))
289 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, []byte{0})
290 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, []byte("GET /broken\r\n"))
291 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, bytes.Repeat([]byte("G"), maxSmartTCPSniffBytes+1))
292 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool {
293 return c.PROXYRefused == 1 && c.RawFallbackCIDRRefused == 1 && c.MalformedRejected == 1 && c.ClientHelloTooLarge == 1
294 })
295 })
296
297 t.Run("unknown without fallback refused", func(t *testing.T) {
298 manager, graph, _ := startSmartTCP(t, nil, nil)
299 defer shutdownManager(t, manager)
300 assertSmartTCPWriteRejected(t, graph.EntryPoints[0].Address, []byte{0})
301 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.UnknownNoFallbackRefused == 1 })
302 })
303
304 t.Run("sniff timeout", func(t *testing.T) {
305 manager, graph, _ := startSmartTCP(t, nil, nil)
306 defer shutdownManager(t, manager)
307 graph.Options.TCP.DialTimeout = 50 * time.Millisecond
308 require.NoError(t, manager.Apply(context.Background(), &graph))
309 conn := dialTCP(t, graph.EntryPoints[0].Address)
310 defer conn.Close()
311 _, err := conn.Write([]byte("P"))
312 require.NoError(t, err)
313 _, _ = bufio.NewReader(conn).ReadByte()
314 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.SniffTimeout == 1 })
315 })
316
317 t.Run("entrypoint CIDR refused", func(t *testing.T) {
318 manager, graph, _ := startSmartTCP(t, nil, nil)
319 defer shutdownManager(t, manager)
320 graph.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
321 require.NoError(t, manager.Apply(context.Background(), &graph))
322 conn := dialTCP(t, graph.EntryPoints[0].Address)
323 _ = conn.Close()
324 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.EntrypointCIDRRefused == 1 })
325 })
326}
327
328func TestManagerStatusAggregatesSmartTCP(t *testing.T) {
329 manager, graph, _ := startSmartTCP(t, nil, nil)
330 defer shutdownManager(t, manager)
331 conn := dialTCP(t, graph.EntryPoints[0].Address)
332 defer conn.Close()
333 _, err := conn.Write([]byte("GET /aggregate HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
334 require.NoError(t, err)
335 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
336 require.NoError(t, err)
337 _, _ = io.ReadAll(resp.Body)
338 _ = resp.Body.Close()
339 assertSmartTCPStatus(t, manager, func(c domain.SmartTCPCounters) bool { return c.HTTPAccepted == 1 })
340}
341
342func assertSmartTCPStatus(t *testing.T, manager *Manager, match func(domain.SmartTCPCounters) bool) {
343 t.Helper()
344 assert.Eventually(t, func() bool {
345 status := manager.Status()
346 if len(status.EntryPoints) == 0 || !match(status.EntryPoints[0].SmartTCP) {
347 return false
348 }
349 return match(status.Counters.SmartTCP)
350 }, time.Second, 10*time.Millisecond)
351}
352
353func TestSmartTCPLifecycleHardening(t *testing.T) {
354 t.Run("internal HTTP and HTTPS fallback servers stop when entrypoint removed", func(t *testing.T) {
355 manager, graph, _ := startSmartTCP(t, nil, nil)
356 initialRuntime := tcpRuntimeForTest(t, manager, "edge")
357 require.NotNil(t, initialRuntime)
358
359 require.NoError(t, manager.Apply(context.Background(), &domain.TrafficGraph{Options: graph.Options}))
360 assert.Eventually(t, func() bool {
361 initialRuntime.mu.Lock()
362 defer initialRuntime.mu.Unlock()
363 return initialRuntime.smartHTTPServer == nil && initialRuntime.smartTLSServer == nil
364 }, time.Second, 10*time.Millisecond)
365 shutdownManager(t, manager)
366 })
367
368 t.Run("HTTP fallback handoff remains counted until hijacked connection closes", func(t *testing.T) {
369 manager := NewManager()
370 graph := smartTCPGraph(t, freeTCPAddress(t), nil, nil)
371 graph.Options.TCP.MaxConnections = 1
372 hijacked := make(chan struct{})
373 hijackedClosed := &atomic.Bool{}
374 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
375 hj, ok := w.(http.Hijacker)
376 require.True(t, ok)
377 conn, rw, err := hj.Hijack()
378 require.NoError(t, err)
379 defer conn.Close()
380 _, _ = rw.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\nready\n")
381 _ = rw.Flush()
382 if hijackedClosed.CompareAndSwap(false, true) {
383 close(hijacked)
384 }
385 _, _ = rw.ReadString('\n')
386 }), nil)
387 manager.SetSmartTCPTLSServer("edge", http.NotFoundHandler(), testTLSConfig(t, "app.example.com"))
388 require.NoError(t, manager.Apply(context.Background(), &graph))
389 defer shutdownManager(t, manager)
390
391 first := dialTCP(t, graph.EntryPoints[0].Address)
392 defer first.Close()
393 firstReader := bufio.NewReader(first)
394 _, err := first.Write([]byte("GET /ws HTTP/1.1\r\nHost: app.example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"))
395 require.NoError(t, err)
396 resp, err := http.ReadResponse(firstReader, nil)
397 require.NoError(t, err)
398 assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
399 ready, err := firstReader.ReadString('\n')
400 require.NoError(t, err)
401 assert.Equal(t, "ready\n", ready)
402 <-hijacked
403 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
404
405 second := dialTCP(t, graph.EntryPoints[0].Address)
406 _, err = second.Write([]byte("GET / HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
407 require.NoError(t, err)
408 _, err = bufio.NewReader(second).ReadByte()
409 _ = second.Close()
410 require.Error(t, err)
411 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
412
413 _, err = first.Write([]byte("close\n"))
414 require.NoError(t, err)
415 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 0 }, time.Second, 10*time.Millisecond)
416 third := dialTCP(t, graph.EntryPoints[0].Address)
417 defer third.Close()
418 _, err = third.Write([]byte("GET /after HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
419 require.NoError(t, err)
420 resp, err = http.ReadResponse(bufio.NewReader(third), nil)
421 require.NoError(t, err)
422 assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
423 })
424
425 t.Run("shutdown force drain closes open HTTP fallback handoff", func(t *testing.T) {
426 manager := NewManager()
427 graph := smartTCPGraph(t, freeTCPAddress(t), nil, nil)
428 graph.Options.TCP.DrainTimeout = 50 * time.Millisecond
429 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
430 hj, ok := w.(http.Hijacker)
431 require.True(t, ok)
432 conn, rw, err := hj.Hijack()
433 require.NoError(t, err)
434 defer conn.Close()
435 _, _ = rw.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\nready\n")
436 _ = rw.Flush()
437 _, _ = rw.ReadString('\n')
438 }), nil)
439 manager.SetSmartTCPTLSServer("edge", http.NotFoundHandler(), testTLSConfig(t, "app.example.com"))
440 require.NoError(t, manager.Apply(context.Background(), &graph))
441
442 conn := dialTCP(t, graph.EntryPoints[0].Address)
443 defer conn.Close()
444 reader := bufio.NewReader(conn)
445 _, err := conn.Write([]byte("GET /ws HTTP/1.1\r\nHost: app.example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"))
446 require.NoError(t, err)
447 resp, err := http.ReadResponse(reader, nil)
448 require.NoError(t, err)
449 assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
450 ready, err := reader.ReadString('\n')
451 require.NoError(t, err)
452 assert.Equal(t, "ready\n", ready)
453
454 started := time.Now()
455 require.NoError(t, manager.Apply(context.Background(), &domain.TrafficGraph{Options: graph.Options}))
456 assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
457 _, err = conn.Write([]byte("after\n"))
458 if err == nil {
459 _, err = reader.ReadByte()
460 }
461 require.Error(t, err)
462 })
463
464 t.Run("trusted CIDR change closes active smart TCP connections no longer allowed", func(t *testing.T) {
465 backend := startTCPEchoServer(t, 0)
466 manager, graph, _ := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: backend.address, allowPublic: true})
467 defer shutdownManager(t, manager)
468 graph.EntryPoints[0].TrustedCIDRs = []string{"127.0.0.0/8"}
469 require.NoError(t, manager.Apply(context.Background(), &graph))
470
471 conn := dialTCP(t, graph.EntryPoints[0].Address)
472 defer conn.Close()
473 assertRoundTrip(t, conn, "before")
474
475 updated := graph
476 updated.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
477 require.NoError(t, manager.Apply(context.Background(), &updated))
478 assert.Eventually(t, func() bool {
479 _, writeErr := conn.Write([]byte("after\n"))
480 if writeErr != nil {
481 return true
482 }
483 _ = conn.SetReadDeadline(time.Now().Add(10 * time.Millisecond))
484 _, readErr := bufio.NewReader(conn).ReadByte()
485 return readErr != nil
486 }, time.Second, 10*time.Millisecond)
487 })
488
489 t.Run("active upgraded connection remains tied to accepted generation", func(t *testing.T) {
490 manager := NewManager()
491 graph := smartTCPGraph(t, freeTCPAddress(t), nil, nil)
492 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
493 hj, ok := w.(http.Hijacker)
494 require.True(t, ok)
495 conn, rw, err := hj.Hijack()
496 require.NoError(t, err)
497 _, _ = rw.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\nold-ready\n")
498 _ = rw.Flush()
499 line, err := rw.ReadString('\n')
500 if err == nil {
501 _, _ = conn.Write([]byte("old:" + line))
502 }
503 _ = conn.Close()
504 }), nil)
505 manager.SetSmartTCPTLSServer("edge", http.NotFoundHandler(), testTLSConfig(t, "app.example.com"))
506 require.NoError(t, manager.Apply(context.Background(), &graph))
507 defer shutdownManager(t, manager)
508
509 conn := dialTCP(t, graph.EntryPoints[0].Address)
510 defer conn.Close()
511 reader := bufio.NewReader(conn)
512 _, err := conn.Write([]byte("GET /ws HTTP/1.1\r\nHost: app.example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"))
513 require.NoError(t, err)
514 resp, err := http.ReadResponse(reader, nil)
515 require.NoError(t, err)
516 assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
517 ready, err := reader.ReadString('\n')
518 require.NoError(t, err)
519 assert.Equal(t, "old-ready\n", ready)
520
521 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("new")) }), nil)
522 _, err = conn.Write([]byte("ping\n"))
523 require.NoError(t, err)
524 got, err := reader.ReadString('\n')
525 require.NoError(t, err)
526 assert.Equal(t, "old:ping\n", got)
527 })
528
529 t.Run("listener is not reused for incompatible address", func(t *testing.T) {
530 manager, graph, _ := startSmartTCP(t, nil, nil)
531 defer shutdownManager(t, manager)
532 initialRuntime := tcpRuntimeForTest(t, manager, "edge")
533
534 updated := graph
535 updated.EntryPoints[0].Address = freeTCPAddress(t)
536 require.NoError(t, manager.Apply(context.Background(), &updated))
537 assert.NotSame(t, initialRuntime, tcpRuntimeForTest(t, manager, "edge"))
538 })
539}
540
541func TestSmartTCPHTTPHandoff(t *testing.T) {
542 t.Run("HTTP Upgrade h2c reaches handler and not raw fallback", func(t *testing.T) {
543 manager, graph, hits := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: startTCPEchoServer(t, 0).address, allowPublic: true})
544 defer shutdownManager(t, manager)
545 conn := dialTCP(t, graph.EntryPoints[0].Address)
546 defer conn.Close()
547 _, err := conn.Write([]byte("GET /upgrade HTTP/1.1\r\nHost: app.example.com\r\nConnection: Upgrade, HTTP2-Settings\r\nUpgrade: h2c\r\nHTTP2-Settings: AAMAAABkAAQAAP__\r\n\r\n"))
548 require.NoError(t, err)
549 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
550 require.NoError(t, err)
551 _, _ = io.ReadAll(resp.Body)
552 _ = resp.Body.Close()
553 assert.Equal(t, int64(1), hits.httpHits.Load())
554 assert.Equal(t, int64(0), hits.rawHits.Load())
555 })
556
557 t.Run("websocket upgrade supports hijacker", func(t *testing.T) {
558 manager := NewManager()
559 graph := smartTCPGraph(t, freeTCPAddress(t), nil, nil)
560 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
561 hj, ok := w.(http.Hijacker)
562 require.True(t, ok)
563 conn, rw, err := hj.Hijack()
564 require.NoError(t, err)
565 defer conn.Close()
566 _, _ = rw.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\nws-ok")
567 _ = rw.Flush()
568 }), nil)
569 manager.SetSmartTCPTLSServer("edge", http.NotFoundHandler(), testTLSConfig(t, "app.example.com"))
570 require.NoError(t, manager.Apply(context.Background(), &graph))
571 defer shutdownManager(t, manager)
572 conn := dialTCP(t, graph.EntryPoints[0].Address)
573 defer conn.Close()
574 _, err := conn.Write([]byte("GET /ws HTTP/1.1\r\nHost: app.example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"))
575 require.NoError(t, err)
576 body, err := io.ReadAll(conn)
577 require.NoError(t, err)
578 assert.Contains(t, string(body), "ws-ok")
579 })
580
581 t.Run("CONNECT reaches HTTP handler and not raw fallback", func(t *testing.T) {
582 manager, graph, hits := startSmartTCP(t, nil, &smartTCPRoute{backendAddress: startTCPEchoServer(t, 0).address, allowPublic: true})
583 defer shutdownManager(t, manager)
584 conn := dialTCP(t, graph.EntryPoints[0].Address)
585 defer conn.Close()
586 _, err := conn.Write([]byte("CONNECT db.example.com:443 HTTP/1.1\r\nHost: db.example.com:443\r\nConnection: close\r\n\r\n"))
587 require.NoError(t, err)
588 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
589 require.NoError(t, err)
590 _, _ = io.ReadAll(resp.Body)
591 _ = resp.Body.Close()
592 assert.Equal(t, int64(1), hits.httpHits.Load())
593 assert.Equal(t, int64(0), hits.rawHits.Load())
594 })
595
596 t.Run("handoff connection has cleared deadlines and useful addresses", func(t *testing.T) {
597 seen := make(chan string, 1)
598 manager := NewManager()
599 graph := smartTCPGraph(t, freeTCPAddress(t), nil, nil)
600 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
601 time.Sleep(150 * time.Millisecond)
602 seen <- r.RemoteAddr + "|" + r.Context().Value(http.LocalAddrContextKey).(net.Addr).String()
603 _, _ = w.Write([]byte("addr-ok"))
604 }), nil)
605 manager.SetSmartTCPTLSServer("edge", http.NotFoundHandler(), testTLSConfig(t, "app.example.com"))
606 require.NoError(t, manager.Apply(context.Background(), &graph))
607 defer shutdownManager(t, manager)
608 conn := dialTCP(t, graph.EntryPoints[0].Address)
609 defer conn.Close()
610 _, err := conn.Write([]byte("GET /addr HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
611 require.NoError(t, err)
612 resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
613 require.NoError(t, err)
614 _, _ = io.ReadAll(resp.Body)
615 _ = resp.Body.Close()
616 got := <-seen
617 assert.Contains(t, got, "127.0.0.1")
618 assert.Contains(t, got, graph.EntryPoints[0].Address)
619 })
620}
621
622func TestRouteToHTTPListenerServeFailureReturnsFalseAndLeavesTrackedForCaller(t *testing.T) {
623 listener, err := net.Listen("tcp", "127.0.0.1:0")
624 require.NoError(t, err)
625 defer listener.Close()
626
627 runtime := newEntryPointRuntime(context.Background(), NewManager(), domain.EntryPoint{Name: "edge"}, listener, nil, nil)
628 httpListener := newTLSHTTPListener(listener.Addr())
629 require.NoError(t, httpListener.Close())
630
631 clientConn, serverConn := net.Pipe()
632 defer clientConn.Close()
633 defer serverConn.Close()
634 tracked := &trackedTCPConn{client: clientConn}
635 runtime.track(tracked)
636
637 accepted := runtime.routeToHTTPListener(tracked, serverConn, httpListener)
638 require.False(t, accepted)
639
640 runtime.mu.Lock()
641 _, stillTracked := runtime.activeConns[tracked]
642 runtime.mu.Unlock()
643 assert.True(t, stillTracked)
644
645 runtime.untrack(tracked)
646}
647
648func TestTCPManagerLifecycle(t *testing.T) {
649 backend := startTCPEchoServer(t, 0)
650 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
651 manager := NewManager()
652
653 require.NoError(t, manager.Apply(context.Background(), &graph))
654 defer shutdownManager(t, manager)
655
656 conn := dialTCP(t, graph.EntryPoints[0].Address)
657 require.NoError(t, conn.Close())
658
659 status := manager.Status()
660 require.Equal(t, reloadStatusOK, status.LastReloadStatus)
661 require.Len(t, status.EntryPoints, 1)
662 assert.True(t, status.EntryPoints[0].Active)
663
664 require.NoError(t, manager.Shutdown(context.Background()))
665 _, err := net.DialTimeout("tcp", graph.EntryPoints[0].Address, 100*time.Millisecond)
666 require.Error(t, err)
667}
668
669func TestTCPManagerRejectsInvalidGraphWithoutReplacingSnapshot(t *testing.T) {
670 backend := startTCPEchoServer(t, 0)
671 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
672 manager := NewManager()
673 require.NoError(t, manager.Apply(context.Background(), &graph))
674 defer shutdownManager(t, manager)
675
676 invalid := graph
677 invalid.EntryPoints = append(invalid.EntryPoints, graph.EntryPoints[0])
678 err := manager.Apply(context.Background(), &invalid)
679 require.Error(t, err)
680
681 status := manager.Status()
682 assert.Equal(t, reloadStatusError, status.LastReloadStatus)
683 assert.Contains(t, status.LastReloadError, "duplicate entrypoint")
684
685 conn := dialTCP(t, graph.EntryPoints[0].Address)
686 defer conn.Close()
687 assertRoundTrip(t, conn, "still works")
688}
689
690func TestTCPManagerReplacesSameAddressEntrypoint(t *testing.T) {
691 backend := startTCPEchoServer(t, 0)
692 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
693 manager := NewManager()
694 require.NoError(t, manager.Apply(context.Background(), &graph))
695 defer shutdownManager(t, manager)
696
697 replacement := graph
698 replacement.EntryPoints = []domain.EntryPoint{{Name: "postgres-tls", Address: graph.EntryPoints[0].Address, Protocol: domain.EntryPointProtocolTLSMux}}
699 replacement.Routers = nil
700 replacement.Services = nil
701 require.NoError(t, manager.Apply(context.Background(), &replacement))
702
703 status := manager.Status()
704 require.Len(t, status.EntryPoints, 1)
705 assert.Equal(t, "postgres-tls", status.EntryPoints[0].Name)
706 assert.Equal(t, domain.EntryPointProtocolTLSMux, status.EntryPoints[0].Protocol)
707}
708
709func TestTCPManagerSameAddressProtocolChangeStartsTLSHTTPServer(t *testing.T) {
710 backend := startTCPEchoServer(t, 0)
711 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
712 manager := NewManager()
713 require.NoError(t, manager.Apply(context.Background(), &graph))
714 defer shutdownManager(t, manager)
715
716 manager.SetTLSHTTPServer("websecure", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
717 _, _ = w.Write([]byte("secure"))
718 }), testTLSConfig(t, "app.example.com"))
719 replacement := domain.TrafficGraph{
720 Options: graph.Options,
721 EntryPoints: []domain.EntryPoint{{Name: "websecure", Address: graph.EntryPoints[0].Address, Protocol: domain.EntryPointProtocolTLSMux}},
722 }
723 require.NoError(t, manager.Apply(context.Background(), &replacement))
724
725 assertHTTPSBody(t, graph.EntryPoints[0].Address, "app.example.com", "secure")
726}
727
728func TestTCPManagerSameAddressTLSMuxRenameRefreshesHTTPSRoute(t *testing.T) {
729 address := freeTCPAddress(t)
730 manager := NewManager()
731 manager.SetTLSHTTPServer("old-secure", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
732 _, _ = w.Write([]byte("old"))
733 }), testTLSConfig(t, "app.example.com"))
734 manager.SetTLSHTTPServer("new-secure", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
735 _, _ = w.Write([]byte("new"))
736 }), testTLSConfig(t, "app.example.com"))
737 graph := domain.TrafficGraph{
738 Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}},
739 EntryPoints: []domain.EntryPoint{{Name: "old-secure", Address: address, Protocol: domain.EntryPointProtocolTLSMux}},
740 }
741 require.NoError(t, manager.Apply(context.Background(), &graph))
742 defer shutdownManager(t, manager)
743 assertHTTPSBody(t, address, "app.example.com", "old")
744
745 renamed := graph
746 renamed.EntryPoints[0].Name = "new-secure"
747 require.NoError(t, manager.Apply(context.Background(), &renamed))
748
749 assertHTTPSBody(t, address, "app.example.com", "new")
750}
751
752func TestTCPRuntimeDoesNotPublishTLSHTTPServerAfterClose(t *testing.T) {
753 listener, err := net.Listen("tcp", "127.0.0.1:0")
754 require.NoError(t, err)
755 defer listener.Close()
756 manager := NewManager()
757 runtime := newEntryPointRuntime(context.Background(), manager, domain.EntryPoint{Name: "websecure", Address: listener.Addr().String(), Protocol: domain.EntryPointProtocolTLSMux}, listener, nil, nil)
758 runtime.closed.Store(true)
759
760 runtime.replaceTLSHTTPServer(runtime.entryPointSnapshot(), TLSHTTPServerConfig{
761 Handler: http.NotFoundHandler(),
762 TLSConfig: testTLSConfig(t, "app.example.com"),
763 })
764
765 runtime.mu.Lock()
766 server := runtime.tlsHTTPServer
767 runtime.mu.Unlock()
768 require.Nil(t, server)
769}
770
771func TestTCPManagerSameAddressProtocolChangeStopsTLSHTTPServer(t *testing.T) {
772 backend := startTCPEchoServer(t, 0)
773 graph := domain.TrafficGraph{
774 Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}},
775 EntryPoints: []domain.EntryPoint{{Name: "websecure", Address: freeTCPAddress(t), Protocol: domain.EntryPointProtocolTLSMux}},
776 }
777 manager := NewManager()
778 manager.SetTLSHTTPServer("websecure", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
779 _, _ = w.Write([]byte("secure"))
780 }), testTLSConfig(t, "app.example.com"))
781 require.NoError(t, manager.Apply(context.Background(), &graph))
782 defer shutdownManager(t, manager)
783 assertHTTPSBody(t, graph.EntryPoints[0].Address, "app.example.com", "secure")
784
785 replacement := tcpGraph(t, graph.EntryPoints[0].Address, backend.address)
786 require.NoError(t, manager.Apply(context.Background(), &replacement))
787
788 _, err := tls.DialWithDialer(&net.Dialer{Timeout: time.Second}, "tcp", graph.EntryPoints[0].Address, &tls.Config{ServerName: "app.example.com", InsecureSkipVerify: true})
789 require.Error(t, err)
790}
791
792func TestTrackedTCPConnClosesLateBackendAfterStaleClose(t *testing.T) {
793 clientA, clientB := net.Pipe()
794 backendA, backendB := net.Pipe()
795 defer clientA.Close()
796 defer backendB.Close()
797
798 conn := &trackedTCPConn{client: clientB}
799 conn.close()
800 conn.setBackend(backendA)
801
802 _, err := backendB.Write([]byte("late backend"))
803 if err == nil {
804 _ = backendA.SetReadDeadline(time.Now().Add(50 * time.Millisecond))
805 _, err = backendA.Read(make([]byte, 1))
806 }
807 require.Error(t, err)
808}
809
810func TestTCPPassthroughRejectsUntrustedCIDR(t *testing.T) {
811 backend := startTCPEchoServer(t, 0)
812 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
813 graph.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
814 manager := NewManager()
815 require.NoError(t, manager.Apply(context.Background(), &graph))
816 defer shutdownManager(t, manager)
817
818 assertTCPRejected(t, manager, graph.EntryPoints[0].Address, 1)
819}
820
821func TestTCPPassthroughReloadAppliesTrustedCIDRChange(t *testing.T) {
822 backend := startTCPEchoServer(t, 0)
823 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
824 graph.EntryPoints[0].TrustedCIDRs = []string{"127.0.0.0/8"}
825 manager := NewManager()
826 require.NoError(t, manager.Apply(context.Background(), &graph))
827 defer shutdownManager(t, manager)
828
829 conn := dialTCP(t, graph.EntryPoints[0].Address)
830 assertRoundTrip(t, conn, "allowed")
831 require.NoError(t, conn.Close())
832
833 updated := graph
834 updated.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
835 require.NoError(t, manager.Apply(context.Background(), &updated))
836 assertTCPRejected(t, manager, graph.EntryPoints[0].Address, 1)
837}
838
839func TestTLSMuxReloadClosesActiveUntrustedHTTPSFallbackConnections(t *testing.T) {
840 graph := domain.TrafficGraph{
841 Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}},
842 EntryPoints: []domain.EntryPoint{{Name: "websecure", Address: freeTCPAddress(t), Protocol: domain.EntryPointProtocolTLSMux, TrustedCIDRs: []string{"127.0.0.0/8"}}},
843 }
844 manager := NewManager()
845 manager.SetTLSHTTPServer("websecure", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
846 body := "secure"
847 w.Header().Set("Content-Length", fmt.Sprintf("%d", len(body)))
848 _, _ = w.Write([]byte(body))
849 }), testTLSConfig(t, "app.example.com"))
850 require.NoError(t, manager.Apply(context.Background(), &graph))
851 defer shutdownManager(t, manager)
852
853 conn, err := tls.DialWithDialer(&net.Dialer{Timeout: time.Second}, "tcp", graph.EntryPoints[0].Address, &tls.Config{ServerName: "app.example.com", InsecureSkipVerify: true})
854 require.NoError(t, err)
855 defer conn.Close()
856 reader := bufio.NewReader(conn)
857 _, err = conn.Write([]byte("GET / HTTP/1.1\r\nHost: app.example.com\r\nConnection: keep-alive\r\n\r\n"))
858 require.NoError(t, err)
859 resp, err := http.ReadResponse(reader, nil)
860 require.NoError(t, err)
861 _, err = io.ReadAll(resp.Body)
862 require.NoError(t, err)
863 require.NoError(t, resp.Body.Close())
864
865 updated := graph
866 updated.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
867 require.NoError(t, manager.Apply(context.Background(), &updated))
868
869 _, err = conn.Write([]byte("GET / HTTP/1.1\r\nHost: app.example.com\r\nConnection: close\r\n\r\n"))
870 if err == nil {
871 _ = conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
872 _, err = http.ReadResponse(reader, nil)
873 }
874 require.Error(t, err)
875}
876
877func TestTCPPassthroughReloadClosesActiveUntrustedConnections(t *testing.T) {
878 backend := startTCPEchoServer(t, 0)
879 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
880 graph.EntryPoints[0].TrustedCIDRs = []string{"127.0.0.0/8"}
881 manager := NewManager()
882 require.NoError(t, manager.Apply(context.Background(), &graph))
883 defer shutdownManager(t, manager)
884
885 conn := dialTCP(t, graph.EntryPoints[0].Address)
886 defer conn.Close()
887 assertRoundTrip(t, conn, "before")
888
889 updated := graph
890 updated.EntryPoints[0].TrustedCIDRs = []string{"192.0.2.0/24"}
891 require.NoError(t, manager.Apply(context.Background(), &updated))
892
893 assert.Eventually(t, func() bool {
894 _, writeErr := conn.Write([]byte("after"))
895 if writeErr != nil {
896 return true
897 }
898 _ = conn.SetReadDeadline(time.Now().Add(10 * time.Millisecond))
899 _, readErr := bufio.NewReader(conn).ReadByte()
900 return readErr != nil
901 }, time.Second, 10*time.Millisecond)
902}
903
904func assertTCPRejected(t *testing.T, manager *Manager, address string, refused int64) {
905 t.Helper()
906 conn := dialTCP(t, address)
907 defer conn.Close()
908 _, writeErr := conn.Write([]byte("blocked"))
909 if writeErr == nil {
910 _, readErr := bufio.NewReader(conn).ReadByte()
911 require.Error(t, readErr)
912 }
913 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == refused }, time.Second, 10*time.Millisecond)
914}
915
916func TestTCPPassthroughEcho(t *testing.T) {
917 backend := startTCPEchoServer(t, 0)
918 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
919 manager := NewManager()
920 require.NoError(t, manager.Apply(context.Background(), &graph))
921 defer shutdownManager(t, manager)
922
923 conn := dialTCP(t, graph.EntryPoints[0].Address)
924 defer conn.Close()
925 assertRoundTrip(t, conn, "hello tcp")
926}
927
928func TestTCPPassthroughUnknownRouterCloses(t *testing.T) {
929 graph := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
930 graph.Routers = nil
931 graph.Services = nil
932 manager := NewManager()
933 require.NoError(t, manager.Apply(context.Background(), &graph))
934 defer shutdownManager(t, manager)
935
936 conn := dialTCP(t, graph.EntryPoints[0].Address)
937 defer conn.Close()
938 _, writeErr := conn.Write([]byte("hello"))
939 if writeErr == nil {
940 _, readErr := bufio.NewReader(conn).ReadByte()
941 require.Error(t, readErr)
942 }
943 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
944}
945
946func TestTCPPassthroughBackendDialFailure(t *testing.T) {
947 graph := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
948 graph.Options.TCP.DialTimeout = 50 * time.Millisecond
949 manager := NewManager()
950 require.NoError(t, manager.Apply(context.Background(), &graph))
951 defer shutdownManager(t, manager)
952
953 conn := dialTCP(t, graph.EntryPoints[0].Address)
954 defer conn.Close()
955 _, writeErr := conn.Write([]byte("hello"))
956 if writeErr == nil {
957 _, readErr := bufio.NewReader(conn).ReadByte()
958 require.Error(t, readErr)
959 }
960 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalErrors == 1 }, time.Second, 10*time.Millisecond)
961}
962
963func TestTCPPassthroughShutdownStopsAccepting(t *testing.T) {
964 backend := startTCPEchoServer(t, 0)
965 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
966 manager := NewManager()
967 require.NoError(t, manager.Apply(context.Background(), &graph))
968
969 require.NoError(t, manager.Shutdown(context.Background()))
970 _, err := net.DialTimeout("tcp", graph.EntryPoints[0].Address, 100*time.Millisecond)
971 require.Error(t, err)
972}
973
974func TestTCPPassthroughIdleTimeoutClosesConnection(t *testing.T) {
975 backend := startTCPEchoServer(t, 0)
976 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
977 graph.Options.TCP.IdleTimeout = 50 * time.Millisecond
978 manager := NewManager()
979 require.NoError(t, manager.Apply(context.Background(), &graph))
980 defer shutdownManager(t, manager)
981
982 conn := dialTCP(t, graph.EntryPoints[0].Address)
983 defer conn.Close()
984 require.NoError(t, conn.SetReadDeadline(time.Now().Add(250*time.Millisecond)))
985 started := time.Now()
986 _, err := bufio.NewReader(conn).ReadByte()
987 require.Error(t, err)
988 var netErr net.Error
989 require.False(t, errors.As(err, &netErr) && netErr.Timeout(), "read hit the client deadline instead of the proxy idle timeout")
990 assert.Less(t, time.Since(started), 250*time.Millisecond)
991}
992
993func TestTCPPassthroughMaxConnectionsRejectsOverflow(t *testing.T) {
994 backend := startTCPEchoServer(t, 0)
995 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
996 graph.Options.TCP.MaxConnections = 1
997 manager := NewManager()
998 require.NoError(t, manager.Apply(context.Background(), &graph))
999 defer shutdownManager(t, manager)
1000
1001 first := dialTCP(t, graph.EntryPoints[0].Address)
1002 defer first.Close()
1003 assertRoundTrip(t, first, "first")
1004 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
1005
1006 second := dialTCP(t, graph.EntryPoints[0].Address)
1007 defer second.Close()
1008 _, err := second.Write([]byte("second"))
1009 require.NoError(t, err)
1010 _, err = bufio.NewReader(second).ReadByte()
1011 require.Error(t, err)
1012 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalRefused == 1 }, time.Second, 10*time.Millisecond)
1013}
1014
1015func TestTCPPassthroughDrainWaitsForActiveConnectionThenTimesOut(t *testing.T) {
1016 backend := startTCPEchoServer(t, 0)
1017 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
1018 graph.Options.TCP.DrainTimeout = 50 * time.Millisecond
1019 manager := NewManager()
1020 require.NoError(t, manager.Apply(context.Background(), &graph))
1021 defer shutdownManager(t, manager)
1022
1023 conn := dialTCP(t, graph.EntryPoints[0].Address)
1024 defer conn.Close()
1025 assertRoundTrip(t, conn, "held")
1026 require.Eventually(t, func() bool { return manager.Status().Counters.ActiveTCPConnections == 1 }, time.Second, 10*time.Millisecond)
1027
1028 empty := domain.TrafficGraph{Options: graph.Options}
1029 started := time.Now()
1030 require.NoError(t, manager.Apply(context.Background(), &empty))
1031 assert.GreaterOrEqual(t, time.Since(started), 40*time.Millisecond)
1032
1033 _, err := conn.Write([]byte("after drain"))
1034 if err == nil {
1035 _, err = bufio.NewReader(conn).ReadByte()
1036 }
1037 require.Error(t, err)
1038}
1039
1040func TestTCPPassthroughStatusCountersTrackAcceptedRefusedErrorsAndBytes(t *testing.T) {
1041 backend := startTCPEchoServer(t, 0)
1042 graph := tcpGraph(t, freeTCPAddress(t), backend.address)
1043 graph.Options.TCP.MaxConnections = 1
1044 manager := NewManager()
1045 require.NoError(t, manager.Apply(context.Background(), &graph))
1046 defer shutdownManager(t, manager)
1047
1048 conn := dialTCP(t, graph.EntryPoints[0].Address)
1049 assertRoundTrip(t, conn, "bytes")
1050 second := dialTCP(t, graph.EntryPoints[0].Address)
1051 _, err := second.Write([]byte("refused"))
1052 require.NoError(t, err)
1053 _, _ = bufio.NewReader(second).ReadByte()
1054 _ = second.Close()
1055 _ = conn.Close()
1056
1057 assert.Eventually(t, func() bool {
1058 status := manager.Status()
1059 return status.Counters.TotalAccepted == 1 &&
1060 status.Counters.TotalRefused == 1 &&
1061 status.Counters.BytesIn >= int64(len("bytes\n")) &&
1062 status.Counters.BytesOut >= int64(len("bytes\n"))
1063 }, time.Second, 10*time.Millisecond)
1064
1065 bad := tcpGraph(t, freeTCPAddress(t), freeTCPAddress(t))
1066 bad.Options.TCP.DialTimeout = 50 * time.Millisecond
1067 require.NoError(t, manager.Apply(context.Background(), &bad))
1068 badConn := dialTCP(t, bad.EntryPoints[0].Address)
1069 _, err = badConn.Write([]byte("boom"))
1070 require.NoError(t, err)
1071 _, _ = bufio.NewReader(badConn).ReadByte()
1072 _ = badConn.Close()
1073 assert.Eventually(t, func() bool { return manager.Status().Counters.TotalErrors == 1 }, time.Second, 10*time.Millisecond)
1074}
1075
1076type smartTCPRoute struct {
1077 tlsSNI string
1078 backendAddress string
1079 trustedCIDRs []string
1080 allowPublic bool
1081}
1082
1083type smartTCPHits struct {
1084 httpHits *atomic.Int64
1085 rawHits *atomic.Int64
1086}
1087
1088func startSmartTCP(t *testing.T, tlsRoute *smartTCPRoute, rawRoute *smartTCPRoute) (*Manager, domain.TrafficGraph, smartTCPHits) {
1089 t.Helper()
1090 manager := NewManager()
1091 graph := smartTCPGraph(t, freeTCPAddress(t), tlsRoute, rawRoute)
1092 httpHits := &atomic.Int64{}
1093 rawHits := &atomic.Int64{}
1094 manager.SetSmartTCPHTTPServer("edge", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
1095 httpHits.Add(1)
1096 _, _ = w.Write([]byte("http:" + r.URL.Path))
1097 }), nil)
1098 manager.SetSmartTCPTLSServer("edge", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
1099 _, _ = w.Write([]byte("smart https"))
1100 }), testTLSConfig(t, "app.example.com"))
1101 require.NoError(t, manager.Apply(context.Background(), &graph))
1102 return manager, graph, smartTCPHits{httpHits: httpHits, rawHits: rawHits}
1103}
1104
1105func smartTCPGraph(t *testing.T, listenAddress string, tlsRoute *smartTCPRoute, rawRoute *smartTCPRoute) domain.TrafficGraph {
1106 t.Helper()
1107 entry := domain.EntryPoint{Name: "edge", Address: listenAddress, Protocol: domain.EntryPointProtocolSmartTCP}
1108 routers := []domain.TrafficRouter{{Name: "route:app", EntryPoint: "edge", Protocol: domain.RouterProtocolHTTP, Rule: domain.TrafficRule{Host: "app.example.com"}, Service: "route:app"}}
1109 services := []domain.TrafficService{{Name: "route:app"}}
1110 if tlsRoute != nil {
1111 backend, err := backendFromAddress("tls:raw", tlsRoute.backendAddress)
1112 require.NoError(t, err)
1113 routers = append(routers, domain.TrafficRouter{Name: "raw-tls", EntryPoint: "edge", Protocol: domain.RouterProtocolTLSPassthrough, Rule: domain.TrafficRule{SNI: tlsRoute.tlsSNI}, Service: "network_service:tls:raw"})
1114 services = append(services, domain.TrafficService{Name: "network_service:tls:raw", Backends: []domain.TrafficBackend{backend}})
1115 }
1116 if rawRoute != nil {
1117 backend, err := backendFromAddress("raw:tcp", rawRoute.backendAddress)
1118 require.NoError(t, err)
1119 entry.RawFallback = "raw-fallback"
1120 entry.RawFallbackTrustedCIDRs = rawRoute.trustedCIDRs
1121 entry.AllowPublicRawFallback = rawRoute.allowPublic
1122 routers = append(routers, domain.TrafficRouter{Name: "raw-fallback", EntryPoint: "edge", Protocol: domain.RouterProtocolTCP, Service: "network_service:raw:tcp"})
1123 services = append(services, domain.TrafficService{Name: "network_service:raw:tcp", Backends: []domain.TrafficBackend{backend}})
1124 }
1125 graph := domain.TrafficGraph{Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}}, EntryPoints: []domain.EntryPoint{entry}, Routers: routers, Services: services}
1126 require.NoError(t, graph.Validate())
1127 return graph
1128}
1129
1130func assertSmartTCPWriteRejected(t *testing.T, address string, payload []byte) {
1131 t.Helper()
1132 conn := dialTCP(t, address)
1133 defer conn.Close()
1134 _, err := conn.Write(payload)
1135 require.NoError(t, err)
1136 _, err = bufio.NewReader(conn).ReadByte()
1137 require.Error(t, err)
1138}
1139
1140func tcpGraph(t *testing.T, listenAddress string, backendAddress string) domain.TrafficGraph {
1141 t.Helper()
1142 backend, err := backendFromAddress("echo:tcp", backendAddress)
1143 require.NoError(t, err)
1144 ref := serviceRef("echo", "tcp")
1145 graph := domain.TrafficGraph{
1146 Options: domain.TrafficOptions{TCP: domain.TCPOptions{DialTimeout: time.Second, IdleTimeout: time.Minute, DrainTimeout: time.Second}},
1147 EntryPoints: []domain.EntryPoint{{Name: "tcp", Address: listenAddress, Protocol: domain.EntryPointProtocolTCP}},
1148 Routers: []domain.TrafficRouter{{Name: "echo", EntryPoint: "tcp", Protocol: domain.RouterProtocolTCP, Service: ref}},
1149 Services: []domain.TrafficService{{Name: ref, Backends: []domain.TrafficBackend{backend}}},
1150 }
1151 require.NoError(t, graph.Validate())
1152 return graph
1153}
1154
1155func freeTCPAddress(t *testing.T) string {
1156 t.Helper()
1157 listener, err := net.Listen("tcp", "127.0.0.1:0")
1158 require.NoError(t, err)
1159 address := listener.Addr().String()
1160 require.NoError(t, listener.Close())
1161 return address
1162}
1163
1164func dialTCP(t *testing.T, address string) net.Conn {
1165 t.Helper()
1166 conn, err := net.DialTimeout("tcp", address, time.Second)
1167 require.NoError(t, err)
1168 return conn
1169}
1170
1171func assertRoundTrip(t *testing.T, conn net.Conn, message string) {
1172 t.Helper()
1173 _, err := conn.Write([]byte(message + "\n"))
1174 require.NoError(t, err)
1175 line, err := bufio.NewReader(conn).ReadString('\n')
1176 require.NoError(t, err)
1177 assert.Equal(t, message+"\n", line)
1178}
1179
1180func shutdownManager(t *testing.T, manager *Manager) {
1181 t.Helper()
1182 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
1183 defer cancel()
1184 require.NoError(t, manager.Shutdown(ctx))
1185}
1186
1187func tcpRuntimeForTest(t *testing.T, manager *Manager, name string) *entryPointRuntime {
1188 t.Helper()
1189 manager.mu.Lock()
1190 defer manager.mu.Unlock()
1191 runtime := manager.listeners[name]
1192 require.NotNil(t, runtime)
1193 return runtime
1194}
1195
1196type tcpEchoServer struct {
1197 address string
1198 close func()
1199}
1200
1201func startTCPEchoServer(t *testing.T, readDelay time.Duration) tcpEchoServer {
1202 t.Helper()
1203 listener, err := net.Listen("tcp", "127.0.0.1:0")
1204 require.NoError(t, err)
1205 done := make(chan struct{})
1206 go func() {
1207 defer close(done)
1208 for {
1209 conn, err := listener.Accept()
1210 if err != nil {
1211 return
1212 }
1213 go func() {
1214 defer conn.Close()
1215 if readDelay > 0 {
1216 time.Sleep(readDelay)
1217 }
1218 _, _ = io.Copy(conn, conn)
1219 }()
1220 }
1221 }()
1222 server := tcpEchoServer{address: listener.Addr().String(), close: func() {
1223 _ = listener.Close()
1224 <-done
1225 }}
1226 t.Cleanup(server.close)
1227 return server
1228}