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