1package traffic
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "net"
9 "net/http"
10 "strconv"
11 "strings"
12 "sync"
13 "sync/atomic"
14 "time"
15
16 "github.com/bnema/gordon/internal/domain"
17)
18
19type entryPointRuntime struct {
20 manager *Manager
21 entryPoint domain.EntryPoint
22 listener net.Listener
23 counters trafficCounters
24 trusted []*net.IPNet
25
26 started atomic.Bool
27 closed atomic.Bool
28
29 ctx context.Context
30 cancel context.CancelFunc
31 acceptDone chan struct{}
32 acceptDoneOnce sync.Once
33 activeWG sync.WaitGroup
34
35 mu sync.Mutex
36 activeConns map[*trackedTCPConn]struct{}
37
38 tlsHTTPReplaceMu sync.Mutex
39 tlsHTTPListener *tlsHTTPListener
40 tlsHTTPServer *http.Server
41 tlsHTTPDone chan struct{}
42
43 smartHTTPReplaceMu sync.Mutex
44 smartHTTPListener *tlsHTTPListener
45 smartHTTPServer *http.Server
46 smartHTTPDone chan struct{}
47
48 smartTLSReplaceMu sync.Mutex
49 smartTLSListener *tlsHTTPListener
50 smartTLSServer *http.Server
51 smartTLSDone chan struct{}
52
53 rawFallbackTrusted []*net.IPNet
54}
55
56type trackedTCPConn struct {
57 mu sync.Mutex
58 client net.Conn
59 backend net.Conn
60 closing bool
61 onDone func()
62 doneOnce sync.Once
63 rawFallbackName string
64}
65
66func (c *trackedTCPConn) setBackend(backend net.Conn) {
67 c.mu.Lock()
68 if c.closing {
69 c.mu.Unlock()
70 _ = backend.Close()
71 return
72 }
73 c.backend = backend
74 c.mu.Unlock()
75}
76
77func (c *trackedTCPConn) close() {
78 c.mu.Lock()
79 c.closing = true
80 client := c.client
81 backend := c.backend
82 c.mu.Unlock()
83 _ = client.Close()
84 if backend != nil {
85 _ = backend.Close()
86 }
87}
88
89func (c *trackedTCPConn) complete() {
90 c.doneOnce.Do(func() {
91 if c.onDone != nil {
92 c.onDone()
93 }
94 })
95}
96
97func (c *trackedTCPConn) markRawFallback(name string) {
98 c.mu.Lock()
99 c.rawFallbackName = name
100 c.mu.Unlock()
101}
102
103func (c *trackedTCPConn) rawFallback() (string, bool) {
104 c.mu.Lock()
105 defer c.mu.Unlock()
106 return c.rawFallbackName, c.rawFallbackName != ""
107}
108
109func newEntryPointRuntime(parentCtx context.Context, manager *Manager, entryPoint domain.EntryPoint, listener net.Listener, trusted []*net.IPNet, rawTrusted []*net.IPNet) *entryPointRuntime {
110 ctx, cancel := context.WithCancel(context.WithoutCancel(parentCtx))
111 return &entryPointRuntime{
112 manager: manager,
113 entryPoint: entryPoint,
114 listener: listener,
115 trusted: trusted,
116 rawFallbackTrusted: rawTrusted,
117 ctx: ctx,
118 cancel: cancel,
119 acceptDone: make(chan struct{}),
120 activeConns: map[*trackedTCPConn]struct{}{},
121 }
122}
123
124func (r *entryPointRuntime) start() {
125 if !r.started.CompareAndSwap(false, true) {
126 return
127 }
128 entryPoint := r.entryPointSnapshot()
129 r.startTLSHTTPServer(entryPoint)
130 r.startSmartTCPHTTPServers(entryPoint)
131 trafficInfo(r.ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Str("protocol", string(entryPoint.Protocol)).Msg("started tcp traffic entrypoint")
132 go r.acceptLoop()
133}
134
135func (r *entryPointRuntime) startTLSHTTPServer(entryPoint domain.EntryPoint) {
136 if entryPoint.Protocol != domain.EntryPointProtocolTLSMux {
137 return
138 }
139 config, ok := r.manager.tlsHTTPServer(entryPoint.Name)
140 if !ok || config.Handler == nil || config.TLSConfig == nil {
141 return
142 }
143 r.mu.Lock()
144 alreadyStarted := r.tlsHTTPServer != nil
145 r.mu.Unlock()
146 if alreadyStarted {
147 return
148 }
149 r.replaceTLSHTTPServer(entryPoint, config)
150}
151
152func (r *entryPointRuntime) refreshTLSHTTPServer(entryPointName string) {
153 entryPoint := r.entryPointSnapshot()
154 if entryPoint.Name != entryPointName || entryPoint.Protocol != domain.EntryPointProtocolTLSMux {
155 return
156 }
157 config, ok := r.manager.tlsHTTPServer(entryPointName)
158 if !ok || config.Handler == nil || config.TLSConfig == nil {
159 r.stopTLSHTTPServer(r.ctx, 0)
160 return
161 }
162 r.replaceTLSHTTPServer(entryPoint, config)
163}
164
165func (r *entryPointRuntime) replaceTLSHTTPServer(entryPoint domain.EntryPoint, config TLSHTTPServerConfig) {
166 r.tlsHTTPReplaceMu.Lock()
167 defer r.tlsHTTPReplaceMu.Unlock()
168 if r.isClosed() {
169 return
170 }
171 listener := newTLSHTTPListener(r.listener.Addr())
172 server := newEntryPointHTTPServer(config.Handler, r.manager.httpOptions())
173 server.TLSConfig = config.TLSConfig.Clone()
174 done := make(chan struct{})
175
176 r.mu.Lock()
177 oldListener := r.tlsHTTPListener
178 oldServer := r.tlsHTTPServer
179 oldDone := r.tlsHTTPDone
180 if r.isClosed() {
181 r.mu.Unlock()
182 _ = listener.Close()
183 return
184 }
185 r.tlsHTTPListener = listener
186 r.tlsHTTPServer = server
187 r.tlsHTTPDone = done
188 r.mu.Unlock()
189
190 go func() {
191 defer close(done)
192 if err := server.ServeTLS(listener, "", ""); err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) {
193 trafficWarn(r.ctx).Err(err).Str("entrypoint", entryPoint.Name).Msg("tls mux https server stopped with error")
194 }
195 }()
196 stopTLSHTTPServer(r.ctx, entryPoint.Name, oldListener, oldServer, oldDone, 0)
197}
198
199func (r *entryPointRuntime) routeToHTTPS(tracked *trackedTCPConn, conn net.Conn) bool {
200 r.mu.Lock()
201 listener := r.tlsHTTPListener
202 r.mu.Unlock()
203 return r.routeToHTTPListener(tracked, conn, listener)
204}
205
206func (r *entryPointRuntime) routeToSmartHTTP(tracked *trackedTCPConn, conn net.Conn) bool {
207 r.mu.Lock()
208 listener := r.smartHTTPListener
209 r.mu.Unlock()
210 return r.routeToHTTPListener(tracked, conn, listener)
211}
212
213func (r *entryPointRuntime) routeToSmartHTTPS(tracked *trackedTCPConn, conn net.Conn) bool {
214 r.mu.Lock()
215 listener := r.smartTLSListener
216 r.mu.Unlock()
217 return r.routeToHTTPListener(tracked, conn, listener)
218}
219
220func (r *entryPointRuntime) routeToHTTPListener(tracked *trackedTCPConn, conn net.Conn, listener *tlsHTTPListener) bool {
221 if listener == nil {
222 return false
223 }
224 return listener.serve(&trackedHTTPConn{Conn: conn, onClose: tracked.complete})
225}
226
227type trackedHTTPConn struct {
228 net.Conn
229 once sync.Once
230 onClose func()
231}
232
233func (c *trackedHTTPConn) Close() error {
234 err := c.Conn.Close()
235 c.once.Do(c.onClose)
236 return err
237}
238
239func (r *entryPointRuntime) acceptLoop() {
240 defer r.closeAcceptDone()
241 for {
242 conn, err := r.listener.Accept()
243 if err != nil {
244 if r.isClosed() || errors.Is(err, net.ErrClosed) {
245 return
246 }
247 r.counters.totalErrors.Add(1)
248 var netErr net.Error
249 if errors.As(err, &netErr) && netErr.Timeout() {
250 continue
251 }
252 trafficWarn(r.ctx).Err(err).Str("entrypoint", r.entryPointSnapshot().Name).Msg("tcp traffic accept failed; retrying")
253 select {
254 case <-time.After(100 * time.Millisecond):
255 case <-r.ctx.Done():
256 return
257 }
258 continue
259 }
260 if !trustedRemoteAddr(r.trustedSnapshot(), conn.RemoteAddr()) {
261 r.counters.totalRefused.Add(1)
262 if r.entryPointSnapshot().Protocol == domain.EntryPointProtocolSmartTCP {
263 r.counters.smartTCP.entrypointCIDRRefused.Add(1)
264 }
265 _ = conn.Close()
266 continue
267 }
268 r.activeWG.Add(1)
269 go r.handleTCPConn(conn)
270 }
271}
272
273func (r *entryPointRuntime) handleTCPConn(client net.Conn) {
274 options := effectiveTCPOptions(snapshotTCPOptions(r.manager.snapshot.Load()))
275 if !r.reserveConnection(options.MaxConnections) {
276 r.counters.totalRefused.Add(1)
277 _ = client.Close()
278 r.activeWG.Done()
279 return
280 }
281
282 tracked := &trackedTCPConn{client: client}
283 tracked.onDone = func() {
284 r.untrack(tracked)
285 r.releaseConnection()
286 r.activeWG.Done()
287 }
288 if !r.track(tracked) {
289 r.releaseConnection()
290 _ = client.Close()
291 r.activeWG.Done()
292 return
293 }
294 completeOnReturn := true
295 defer func() {
296 if completeOnReturn {
297 tracked.complete()
298 }
299 }()
300
301 switch r.entryPointSnapshot().Protocol {
302 case domain.EntryPointProtocolTCP:
303 r.handlePlainTCP(tracked, options)
304 case domain.EntryPointProtocolTLSMux:
305 if r.handleTLSMux(tracked, options) {
306 completeOnReturn = false
307 }
308 case domain.EntryPointProtocolSmartTCP:
309 if r.handleSmartTCP(tracked, options) {
310 completeOnReturn = false
311 }
312 default:
313 r.counters.totalRefused.Add(1)
314 _ = tracked.client.Close()
315 }
316}
317
318func (r *entryPointRuntime) handlePlainTCP(tracked *trackedTCPConn, options domain.TCPOptions) {
319 _, backend, ok := r.resolveTCPBackend()
320 if !ok {
321 r.counters.totalRefused.Add(1)
322 _ = tracked.client.Close()
323 return
324 }
325 r.proxyToBackend(tracked, tracked.client, backend, options)
326}
327
328func (r *entryPointRuntime) handleSmartTCP(tracked *trackedTCPConn, options domain.TCPOptions) bool {
329 result, err := sniffSmartTCP(tracked.client, clientHelloTimeout(options))
330 if err != nil {
331 r.counters.totalErrors.Add(1)
332 _ = tracked.client.Close()
333 return false
334 }
335 switch result.kind {
336 case dispatchHTTP, dispatchH2C:
337 return r.handleSmartTCPHTTP(tracked, result)
338 case dispatchTLS:
339 return r.handleSmartTCPTLS(tracked, result.conn, options)
340 case dispatchUnknown:
341 return r.handleSmartTCPUnknown(tracked, result.conn, options)
342 case dispatchRejectPROXY:
343 return r.refuseSmartTCP(result.conn, r.counters.smartTCP.proxyRefused.Add)
344 case dispatchSniffTimeout:
345 return r.refuseSmartTCP(result.conn, r.counters.smartTCP.sniffTimeout.Add)
346 case dispatchReject:
347 return r.refuseSmartTCP(result.conn, r.counters.smartTCP.malformedRejected.Add)
348 case dispatchRejectLarge:
349 return r.refuseSmartTCP(result.conn, r.counters.smartTCP.clientHelloTooLarge.Add)
350 default:
351 return r.refuseSmartTCP(result.conn, nil)
352 }
353}
354
355func (r *entryPointRuntime) handleSmartTCPHTTP(tracked *trackedTCPConn, result smartTCPSniffResult) bool {
356 if !r.routeToSmartHTTP(tracked, result.conn) {
357 return r.refuseSmartTCP(result.conn, nil)
358 }
359 r.counters.totalAccepted.Add(1)
360 if result.kind == dispatchH2C {
361 r.counters.smartTCP.h2cAccepted.Add(1)
362 } else {
363 r.counters.smartTCP.httpAccepted.Add(1)
364 }
365 return true
366}
367
368func (r *entryPointRuntime) handleSmartTCPTLS(tracked *trackedTCPConn, conn net.Conn, options domain.TCPOptions) bool {
369 peeked, err := r.peekTLSClientHello(conn, options)
370 if err != nil {
371 return r.handleSmartTCPTLSPeekError(conn, err)
372 }
373 if backend, ok := r.resolveTLSBackend(peeked.sni); ok {
374 r.proxyToBackendAfterDial(tracked, peeked.conn, backend, options, func() {
375 r.counters.smartTCP.tlsPassthroughAccepted.Add(1)
376 })
377 return false
378 }
379 if r.routeToSmartHTTPS(tracked, peeked.conn) {
380 r.counters.totalAccepted.Add(1)
381 r.counters.smartTCP.httpsFallbackAccepted.Add(1)
382 return true
383 }
384 return r.refuseSmartTCP(peeked.conn, nil)
385}
386
387func (r *entryPointRuntime) handleSmartTCPTLSPeekError(conn net.Conn, err error) bool {
388 if isClientHelloTooLargeError(err) {
389 r.counters.totalRefused.Add(1)
390 r.counters.smartTCP.clientHelloTooLarge.Add(1)
391 } else {
392 r.counters.totalErrors.Add(1)
393 }
394 _ = conn.Close()
395 return false
396}
397
398func (r *entryPointRuntime) handleSmartTCPUnknown(tracked *trackedTCPConn, conn net.Conn, options domain.TCPOptions) bool {
399 backend, rawFallbackName, ok, rawCIDRRefused := r.resolveRawFallbackBackendDetailed(conn.RemoteAddr())
400 if ok {
401 tracked.markRawFallback(rawFallbackName)
402 r.proxyToBackendAfterDial(tracked, conn, backend, options, func() {
403 r.counters.smartTCP.rawFallbackAccepted.Add(1)
404 })
405 return false
406 }
407 if rawCIDRRefused {
408 return r.refuseSmartTCP(conn, r.counters.smartTCP.rawFallbackCIDRRefused.Add)
409 }
410 return r.refuseSmartTCP(conn, r.counters.smartTCP.unknownNoFallbackRefused.Add)
411}
412
413func (r *entryPointRuntime) refuseSmartTCP(conn net.Conn, increment func(int64) int64) bool {
414 r.counters.totalRefused.Add(1)
415 if increment != nil {
416 increment(1)
417 }
418 _ = conn.Close()
419 return false
420}
421
422func (r *entryPointRuntime) handleTLSMux(tracked *trackedTCPConn, options domain.TCPOptions) bool {
423 peeked, err := r.peekTLSClientHello(tracked.client, options)
424 if err != nil {
425 r.counters.totalErrors.Add(1)
426 _ = tracked.client.Close()
427 return false
428 }
429 if backend, ok := r.resolveTLSBackend(peeked.sni); ok {
430 r.proxyToBackend(tracked, peeked.conn, backend, options)
431 return false
432 }
433 if r.routeToHTTPS(tracked, peeked.conn) {
434 r.counters.totalAccepted.Add(1)
435 return true
436 }
437 r.counters.totalRefused.Add(1)
438 _ = peeked.conn.Close()
439 return false
440}
441
442func (r *entryPointRuntime) peekTLSClientHello(client net.Conn, options domain.TCPOptions) (peekedTLSConn, error) {
443 if err := client.SetReadDeadline(time.Now().Add(clientHelloTimeout(options))); err != nil {
444 return peekedTLSConn{}, fmt.Errorf("set client hello read deadline for %s: %w", client.RemoteAddr(), err)
445 }
446 sni, replayed, err := peekClientHelloSNI(client)
447 if clearErr := client.SetReadDeadline(time.Time{}); err == nil && clearErr != nil {
448 return peekedTLSConn{}, fmt.Errorf("clear client hello read deadline for %s: %w", client.RemoteAddr(), clearErr)
449 }
450 if err != nil {
451 return peekedTLSConn{}, fmt.Errorf("peek client hello from %s: %w", client.RemoteAddr(), err)
452 }
453 return peekedTLSConn{sni: sni, conn: replayed}, nil
454}
455
456func isClientHelloTooLargeError(err error) bool {
457 return errors.Is(err, domain.ErrClientHelloTooLarge)
458}
459
460func clientHelloTimeout(options domain.TCPOptions) time.Duration {
461 const maxTimeout = 5 * time.Second
462 if options.DialTimeout > 0 && options.DialTimeout < maxTimeout {
463 return options.DialTimeout
464 }
465 return maxTimeout
466}
467
468type peekedTLSConn struct {
469 sni string
470 conn net.Conn
471}
472
473func (r *entryPointRuntime) proxyToBackend(tracked *trackedTCPConn, client net.Conn, backend domain.TrafficBackend, options domain.TCPOptions) bool {
474 return r.proxyToBackendAfterDial(tracked, client, backend, options, nil)
475}
476
477func (r *entryPointRuntime) proxyToBackendAfterDial(tracked *trackedTCPConn, client net.Conn, backend domain.TrafficBackend, options domain.TCPOptions, afterDial func()) bool {
478 dialCtx, cancel := context.WithTimeout(r.ctx, options.DialTimeout)
479 defer cancel()
480 backendConn, err := (&net.Dialer{}).DialContext(dialCtx, "tcp", net.JoinHostPort(backend.Host, strconv.Itoa(backend.Port)))
481 if err != nil {
482 r.counters.totalErrors.Add(1)
483 _ = client.Close()
484 return false
485 }
486
487 r.setBackend(tracked, backendConn)
488 r.counters.totalAccepted.Add(1)
489 if afterDial != nil {
490 afterDial()
491 }
492 r.proxyTCP(client, backendConn, options.IdleTimeout)
493 return true
494}
495
496func (r *entryPointRuntime) resolveRawFallbackBackendDetailed(remote net.Addr) (domain.TrafficBackend, string, bool, bool) {
497 graph := r.manager.snapshot.Load()
498 if graph == nil {
499 return domain.TrafficBackend{}, "", false, false
500 }
501 entryPoint := r.entryPointSnapshot()
502 if entryPoint.RawFallback == "" {
503 return domain.TrafficBackend{}, "", false, false
504 }
505 if !entryPoint.AllowPublicRawFallback {
506 trusted := r.rawFallbackTrustedSnapshot()
507 if len(trusted) == 0 || !trustedRemoteAddr(trusted, remote) {
508 return domain.TrafficBackend{}, "", false, true
509 }
510 }
511 for _, router := range graph.Routers {
512 if router.Name == entryPoint.RawFallback && router.EntryPoint == entryPoint.Name && router.Protocol == domain.RouterProtocolTCP {
513 backend, ok := r.backendForRouter(graph, router)
514 return backend, entryPoint.RawFallback, ok, false
515 }
516 }
517 return domain.TrafficBackend{}, "", false, false
518}
519
520func (r *entryPointRuntime) resolveTCPBackend() (domain.TrafficRouter, domain.TrafficBackend, bool) {
521 graph := r.manager.snapshot.Load()
522 if graph == nil {
523 return domain.TrafficRouter{}, domain.TrafficBackend{}, false
524 }
525 entryPoint := r.entryPointSnapshot()
526 var router domain.TrafficRouter
527 for _, candidate := range graph.Routers {
528 if candidate.EntryPoint == entryPoint.Name && candidate.Protocol == domain.RouterProtocolTCP {
529 if router.Name != "" {
530 return domain.TrafficRouter{}, domain.TrafficBackend{}, false
531 }
532 router = candidate
533 }
534 }
535 if router.Name == "" {
536 return domain.TrafficRouter{}, domain.TrafficBackend{}, false
537 }
538 for _, service := range graph.Services {
539 if service.Name != router.Service || len(service.Backends) != 1 {
540 continue
541 }
542 backend := service.Backends[0]
543 if backend.Protocol != domain.NetworkProtocolTCP {
544 return domain.TrafficRouter{}, domain.TrafficBackend{}, false
545 }
546 return router, backend, true
547 }
548 return domain.TrafficRouter{}, domain.TrafficBackend{}, false
549}
550
551func (r *entryPointRuntime) resolveTLSBackend(sni string) (domain.TrafficBackend, bool) {
552 graph := r.manager.snapshot.Load()
553 sni = normalizeTLSName(sni)
554 if graph == nil || sni == "" {
555 return domain.TrafficBackend{}, false
556 }
557 if router, ok := r.findExactTLSRouter(graph, sni); ok {
558 return r.backendForRouter(graph, router)
559 }
560 if router, ok := r.findWildcardTLSRouter(graph, sni); ok {
561 return r.backendForRouter(graph, router)
562 }
563 return domain.TrafficBackend{}, false
564}
565
566func (r *entryPointRuntime) findExactTLSRouter(graph *domain.TrafficGraph, sni string) (domain.TrafficRouter, bool) {
567 entryPoint := r.entryPointSnapshot()
568 for _, router := range graph.Routers {
569 if router.EntryPoint == entryPoint.Name && router.Protocol == domain.RouterProtocolTLSPassthrough && normalizeTLSName(router.Rule.SNI) == sni {
570 return router, true
571 }
572 }
573 return domain.TrafficRouter{}, false
574}
575
576func (r *entryPointRuntime) findWildcardTLSRouter(graph *domain.TrafficGraph, sni string) (domain.TrafficRouter, bool) {
577 entryPoint := r.entryPointSnapshot()
578 for _, router := range graph.Routers {
579 wildcard := normalizeTLSName(router.Rule.SNI)
580 if router.EntryPoint == entryPoint.Name && router.Protocol == domain.RouterProtocolTLSPassthrough && hostMatchesTLSWildcard(sni, wildcard) {
581 return router, true
582 }
583 }
584 return domain.TrafficRouter{}, false
585}
586
587func (r *entryPointRuntime) backendForRouter(graph *domain.TrafficGraph, router domain.TrafficRouter) (domain.TrafficBackend, bool) {
588 for _, service := range graph.Services {
589 if service.Name == router.Service && len(service.Backends) == 1 && service.Backends[0].Protocol == domain.NetworkProtocolTCP {
590 return service.Backends[0], true
591 }
592 }
593 return domain.TrafficBackend{}, false
594}
595
596func hostMatchesTLSWildcard(host string, wildcard string) bool {
597 suffix, ok := strings.CutPrefix(wildcard, "*.")
598 if !ok || host == suffix {
599 return false
600 }
601 prefix, ok := strings.CutSuffix(host, "."+suffix)
602 return ok && prefix != "" && !strings.Contains(prefix, ".")
603}
604
605func (r *entryPointRuntime) proxyTCP(client net.Conn, backend net.Conn, idleTimeout time.Duration) {
606 defer client.Close()
607 defer backend.Close()
608
609 var wg sync.WaitGroup
610 wg.Add(2)
611 go func() {
612 defer wg.Done()
613 n, err := copyWithIdleTimeout(backend, client, idleTimeout)
614 r.counters.bytesIn.Add(n)
615 if err != nil && !isExpectedCopyError(err) {
616 r.counters.totalErrors.Add(1)
617 }
618 closeWrite(backend)
619 }()
620 go func() {
621 defer wg.Done()
622 n, err := copyWithIdleTimeout(client, backend, idleTimeout)
623 r.counters.bytesOut.Add(n)
624 if err != nil && !isExpectedCopyError(err) {
625 r.counters.totalErrors.Add(1)
626 }
627 closeWrite(client)
628 }()
629 wg.Wait()
630}
631
632func copyWithIdleTimeout(dst net.Conn, src net.Conn, idleTimeout time.Duration) (int64, error) {
633 if idleTimeout <= 0 {
634 return io.Copy(dst, src)
635 }
636 return io.Copy(deadlineWriter{Conn: dst, idleTimeout: idleTimeout}, idleConn{Conn: src, idleTimeout: idleTimeout})
637}
638
639type idleConn struct {
640 net.Conn
641 idleTimeout time.Duration
642}
643
644func (c idleConn) Read(p []byte) (int, error) {
645 if err := c.SetReadDeadline(time.Now().Add(c.idleTimeout)); err != nil {
646 return 0, err
647 }
648 return c.Conn.Read(p)
649}
650
651type deadlineWriter struct {
652 net.Conn
653 idleTimeout time.Duration
654}
655
656func (c deadlineWriter) Write(p []byte) (int, error) {
657 if err := c.SetWriteDeadline(time.Now().Add(c.idleTimeout)); err != nil {
658 return 0, err
659 }
660 return c.Conn.Write(p)
661}
662
663func closeWrite(conn net.Conn) {
664 if writeCloser, ok := conn.(interface{ CloseWrite() error }); ok {
665 _ = writeCloser.CloseWrite()
666 }
667}
668
669func isExpectedCopyError(err error) bool {
670 if err == nil || errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
671 return true
672 }
673 var netErr net.Error
674 if errors.As(err, &netErr) && netErr.Timeout() {
675 return true
676 }
677 return false
678}
679
680func (r *entryPointRuntime) reserveConnection(maxConnections int) bool {
681 active := r.counters.activeTCPConnections.Add(1)
682 if maxConnections > 0 && active > int64(maxConnections) {
683 r.counters.activeTCPConnections.Add(-1)
684 return false
685 }
686 return true
687}
688
689func (r *entryPointRuntime) releaseConnection() {
690 r.counters.activeTCPConnections.Add(-1)
691}
692
693func (r *entryPointRuntime) track(conn *trackedTCPConn) bool {
694 r.mu.Lock()
695 defer r.mu.Unlock()
696 if r.isClosed() {
697 return false
698 }
699 r.activeConns[conn] = struct{}{}
700 return true
701}
702
703func (r *entryPointRuntime) setBackend(conn *trackedTCPConn, backend net.Conn) {
704 r.mu.Lock()
705 _, tracked := r.activeConns[conn]
706 r.mu.Unlock()
707 if !tracked {
708 _ = backend.Close()
709 return
710 }
711 conn.setBackend(backend)
712}
713
714func (r *entryPointRuntime) untrack(conn *trackedTCPConn) {
715 r.mu.Lock()
716 delete(r.activeConns, conn)
717 r.mu.Unlock()
718}
719
720func (r *entryPointRuntime) stop(ctx context.Context, drainTimeout time.Duration) {
721 if r.closed.CompareAndSwap(false, true) {
722 entryPoint := r.entryPointSnapshot()
723 trafficInfo(ctx).Str("entrypoint", entryPoint.Name).Str("address", entryPoint.Address).Msg("stopping tcp traffic entrypoint")
724 r.cancel()
725 _ = r.listener.Close()
726 if !r.started.Load() {
727 r.closeAcceptDone()
728 }
729 }
730 select {
731 case <-r.acceptDone:
732 case <-ctx.Done():
733 return
734 }
735 r.stopTLSHTTPServer(ctx, drainTimeout)
736 r.stopSmartTCPHTTPServer(ctx, drainTimeout)
737 r.stopSmartTCPTLSServer(ctx, drainTimeout)
738 if r.waitActive(ctx, drainTimeout) {
739 trafficInfo(ctx).Str("entrypoint", r.entryPointSnapshot().Name).Msg("stopped tcp traffic entrypoint")
740 return
741 }
742 trafficDebug(ctx).Str("entrypoint", r.entryPointSnapshot().Name).Dur("drain_timeout", drainTimeout).Msg("forcing tcp traffic entrypoint drain")
743 r.closeActiveConns()
744 select {
745 case <-r.activeDone():
746 trafficInfo(ctx).Str("entrypoint", r.entryPointSnapshot().Name).Msg("stopped tcp traffic entrypoint")
747 case <-ctx.Done():
748 }
749}
750
751func (r *entryPointRuntime) stopTLSHTTPServer(ctx context.Context, drainTimeout time.Duration) {
752 r.tlsHTTPReplaceMu.Lock()
753 defer r.tlsHTTPReplaceMu.Unlock()
754 r.stopTLSHTTPServerLocked(ctx, drainTimeout)
755}
756
757func (r *entryPointRuntime) stopTLSHTTPServerLocked(ctx context.Context, drainTimeout time.Duration) {
758 r.mu.Lock()
759 listener := r.tlsHTTPListener
760 server := r.tlsHTTPServer
761 done := r.tlsHTTPDone
762 r.tlsHTTPListener = nil
763 r.tlsHTTPServer = nil
764 r.tlsHTTPDone = nil
765 r.mu.Unlock()
766
767 stopTLSHTTPServer(ctx, r.entryPointSnapshot().Name, listener, server, done, drainTimeout)
768}
769
770func stopTLSHTTPServer(ctx context.Context, entryPointName string, listener *tlsHTTPListener, server *http.Server, done chan struct{}, drainTimeout time.Duration) {
771 if listener == nil || server == nil || done == nil {
772 return
773 }
774 _ = listener.Close()
775 if drainTimeout <= 0 {
776 drainTimeout = defaultTCPOptions().DrainTimeout
777 }
778 shutdownCtx, cancel := context.WithTimeout(ctx, drainTimeout)
779 defer cancel()
780 if err := server.Shutdown(shutdownCtx); err != nil {
781 trafficDebug(ctx).Err(err).Str("entrypoint", entryPointName).Msg("forcing tls mux https server close")
782 _ = server.Close()
783 }
784 select {
785 case <-done:
786 case <-ctx.Done():
787 }
788}
789
790func (r *entryPointRuntime) waitActive(ctx context.Context, drainTimeout time.Duration) bool {
791 if drainTimeout <= 0 {
792 drainTimeout = defaultTCPOptions().DrainTimeout
793 }
794 drainCtx, cancel := context.WithTimeout(ctx, drainTimeout)
795 defer cancel()
796 select {
797 case <-r.activeDone():
798 return true
799 case <-drainCtx.Done():
800 return false
801 }
802}
803
804func (r *entryPointRuntime) closeAcceptDone() {
805 r.acceptDoneOnce.Do(func() { close(r.acceptDone) })
806}
807
808func (r *entryPointRuntime) activeDone() <-chan struct{} {
809 done := make(chan struct{})
810 go func() {
811 r.activeWG.Wait()
812 close(done)
813 }()
814 return done
815}
816
817func (r *entryPointRuntime) closeActiveConns() {
818 r.mu.Lock()
819 defer r.mu.Unlock()
820 for conn := range r.activeConns {
821 conn.close()
822 }
823}
824
825func (r *entryPointRuntime) matches(entryPoint domain.EntryPoint) bool {
826 current := r.entryPointSnapshot()
827 return current.Name == entryPoint.Name && current.Address == entryPoint.Address && current.Protocol == entryPoint.Protocol && trustedCIDRsEqual(current.TrustedCIDRs, entryPoint.TrustedCIDRs) && current.RawFallback == entryPoint.RawFallback && trustedCIDRsEqual(current.RawFallbackTrustedCIDRs, entryPoint.RawFallbackTrustedCIDRs) && current.AllowPublicRawFallback == entryPoint.AllowPublicRawFallback
828}
829
830func (r *entryPointRuntime) sameAddress(entryPoint domain.EntryPoint) bool {
831 return r.entryPointSnapshot().Address == entryPoint.Address
832}
833
834func (r *entryPointRuntime) updateEntryPoint(entryPoint domain.EntryPoint, trusted []*net.IPNet, rawTrusted []*net.IPNet) {
835 r.mu.Lock()
836 previousName := r.entryPoint.Name
837 previousProtocol := r.entryPoint.Protocol
838 r.entryPoint = entryPoint
839 r.trusted = trusted
840 r.rawFallbackTrusted = rawTrusted
841 stale := make([]*trackedTCPConn, 0)
842 for conn := range r.activeConns {
843 if !trustedRemoteAddr(trusted, conn.client.RemoteAddr()) || rawFallbackConnStale(conn, entryPoint, rawTrusted) {
844 stale = append(stale, conn)
845 }
846 }
847 r.mu.Unlock()
848 for _, conn := range stale {
849 conn.close()
850 }
851 if previousProtocol == domain.EntryPointProtocolTLSMux && entryPoint.Protocol != domain.EntryPointProtocolTLSMux {
852 r.stopTLSHTTPServer(r.ctx, 0)
853 }
854 if previousProtocol == domain.EntryPointProtocolSmartTCP && entryPoint.Protocol != domain.EntryPointProtocolSmartTCP {
855 r.stopSmartTCPHTTPServer(r.ctx, 0)
856 r.stopSmartTCPTLSServer(r.ctx, 0)
857 }
858 if entryPoint.Protocol == domain.EntryPointProtocolSmartTCP {
859 r.refreshSmartTCPHTTPServer(entryPoint.Name)
860 r.refreshSmartTCPTLSServer(entryPoint.Name)
861 }
862 if entryPoint.Protocol == domain.EntryPointProtocolTLSMux {
863 if previousProtocol == domain.EntryPointProtocolTLSMux && previousName != entryPoint.Name {
864 r.refreshTLSHTTPServer(entryPoint.Name)
865 return
866 }
867 r.startTLSHTTPServer(entryPoint)
868 }
869}
870
871func rawFallbackConnStale(conn *trackedTCPConn, entryPoint domain.EntryPoint, rawTrusted []*net.IPNet) bool {
872 rawFallbackName, ok := conn.rawFallback()
873 if !ok {
874 return false
875 }
876 if entryPoint.Protocol != domain.EntryPointProtocolSmartTCP || entryPoint.RawFallback == "" || entryPoint.RawFallback != rawFallbackName {
877 return true
878 }
879 if entryPoint.AllowPublicRawFallback {
880 return false
881 }
882 return len(rawTrusted) == 0 || !trustedRemoteAddr(rawTrusted, conn.client.RemoteAddr())
883}
884
885func (r *entryPointRuntime) entryPointSnapshot() domain.EntryPoint {
886 r.mu.Lock()
887 defer r.mu.Unlock()
888 return r.entryPoint
889}
890
891func (r *entryPointRuntime) trustedSnapshot() []*net.IPNet {
892 r.mu.Lock()
893 defer r.mu.Unlock()
894 return r.trusted
895}
896
897func (r *entryPointRuntime) rawFallbackTrustedSnapshot() []*net.IPNet {
898 r.mu.Lock()
899 defer r.mu.Unlock()
900 return r.rawFallbackTrusted
901}
902
903func (r *entryPointRuntime) isClosed() bool {
904 return r.closed.Load()
905}
906
907func backendFromAddress(name string, address string) (domain.TrafficBackend, error) {
908 host, portValue, err := net.SplitHostPort(address)
909 if err != nil {
910 return domain.TrafficBackend{}, fmt.Errorf("split backend address %q: %w", address, err)
911 }
912 port, err := strconv.Atoi(portValue)
913 if err != nil {
914 return domain.TrafficBackend{}, fmt.Errorf("parse backend port %q from address %q: %w", portValue, address, err)
915 }
916 if port < 1 || port > 65535 {
917 return domain.TrafficBackend{}, fmt.Errorf("backend port %q from address %q is outside 1-65535", portValue, address)
918 }
919 return domain.TrafficBackend{Name: name, Host: host, Port: port, Protocol: domain.NetworkProtocolTCP}, nil
920}
921
922func serviceRef(name string, portName string) string {
923 return fmt.Sprintf("network_service:%s:%s", name, portName)
924}