From cdc90411e5511a006466534914ce492968006d81 Mon Sep 17 00:00:00 2001 From: = Date: Sun, 9 Mar 2025 21:14:53 +0530 Subject: [PATCH] feat: updated gateway to use dtls --- cli/go.mod | 10 +-- cli/go.sum | 10 +++ cli/packages/gateway/gateway.go | 111 ++++++++++++++++---------------- cli/packages/gateway/relay.go | 80 +++++++++++------------ 4 files changed, 110 insertions(+), 101 deletions(-) diff --git a/cli/go.mod b/cli/go.mod index 53f348807..bdb8839d0 100644 --- a/cli/go.mod +++ b/cli/go.mod @@ -29,9 +29,9 @@ require ( github.com/spf13/cobra v1.6.1 github.com/spf13/viper v1.8.1 github.com/stretchr/testify v1.10.0 - golang.org/x/crypto v0.35.0 - golang.org/x/sys v0.30.0 - golang.org/x/term v0.29.0 + golang.org/x/crypto v0.36.0 + golang.org/x/sys v0.31.0 + golang.org/x/term v0.30.0 gopkg.in/yaml.v2 v2.4.0 ) @@ -115,8 +115,8 @@ require ( golang.org/x/mod v0.23.0 // indirect golang.org/x/net v0.35.0 // indirect golang.org/x/oauth2 v0.21.0 // indirect - golang.org/x/sync v0.11.0 // indirect - golang.org/x/text v0.22.0 // indirect + golang.org/x/sync v0.12.0 // indirect + golang.org/x/text v0.23.0 // indirect golang.org/x/time v0.6.0 // indirect golang.org/x/tools v0.30.0 // indirect google.golang.org/api v0.188.0 // indirect diff --git a/cli/go.sum b/cli/go.sum index d87cc825a..ab46476d7 100644 --- a/cli/go.sum +++ b/cli/go.sum @@ -486,6 +486,8 @@ golang.org/x/crypto v0.0.0-20211215165025-cf75a172585e/go.mod h1:P+XmwS30IXTQdn5 golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.35.0 h1:b15kiHdrGCHrP6LvwaQ3c03kgNhhiMgvlhxHQhmg2Xs= golang.org/x/crypto v0.35.0/go.mod h1:dy7dXNW32cAb/6/PRuTNsix8T+vJAqvuIy5Bli/x0YQ= +golang.org/x/crypto v0.36.0 h1:AnAEvhDddvBdpY+uR+MyHmuZzzNqXSe/GvuDeob5L34= +golang.org/x/crypto v0.36.0/go.mod h1:Y4J0ReaxCR1IMaabaSMugxJES1EpwhBHhv2bDHklZvc= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190306152737-a1d7652674e8/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= golang.org/x/exp v0.0.0-20190510132918-efd6b22b2522/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8= @@ -592,6 +594,8 @@ golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w= golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw= +golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -642,9 +646,13 @@ golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc= golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik= +golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.29.0 h1:L6pJp37ocefwRRtYPKSWOWzOtWSxVajvz2ldH/xi3iU= golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s= +golang.org/x/term v0.30.0 h1:PQ39fJZ+mfadBm0y5WlL4vlM7Sx1Hgf13sMIY2+QS9Y= +golang.org/x/term v0.30.0/go.mod h1:NYYFdzHoI5wRh/h5tDMdMqCqPJZEuNqVR5xJLd/n67g= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -656,6 +664,8 @@ golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM= golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY= +golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= +golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= diff --git a/cli/packages/gateway/gateway.go b/cli/packages/gateway/gateway.go index 248a9dc7b..1c3515d76 100644 --- a/cli/packages/gateway/gateway.go +++ b/cli/packages/gateway/gateway.go @@ -14,6 +14,7 @@ import ( "github.com/Infisical/infisical-merge/packages/api" "github.com/Infisical/infisical-merge/packages/systemd" "github.com/go-resty/resty/v2" + "github.com/pion/dtls/v3" "github.com/pion/logging" "github.com/pion/turn/v4" "github.com/rs/zerolog/log" @@ -54,26 +55,6 @@ func (g *Gateway) ConnectWithRelay() error { return err } relayAddress, relayPort := strings.Split(relayDetails.TurnServerAddress, ":")[0], strings.Split(relayDetails.TurnServerAddress, ":")[1] - var conn net.Conn - - // Dial TURN Server - if relayPort == "5349" { - log.Info().Msgf("Provided relay port %s. Using TLS", relayPort) - conn, err = tls.Dial("tcp", relayDetails.TurnServerAddress, &tls.Config{ - ServerName: relayAddress, - }) - } else { - log.Info().Msgf("Provided relay port %s. Using non TLS connection.", relayPort) - peerAddr, errPeer := net.ResolveTCPAddr("tcp", relayDetails.TurnServerAddress) - if errPeer != nil { - return fmt.Errorf("Failed to parse turn server address: %w", err) - } - conn, err = net.DialTCP("tcp", nil, peerAddr) - } - - if err != nil { - return fmt.Errorf("Failed to connect with relay server: %w", err) - } // Start a new TURN Client and wrap our net.Conn in a STUNConn // This allows us to simulate datagram based communication over a net.Conn @@ -81,17 +62,48 @@ func (g *Gateway) ConnectWithRelay() error { if os.Getenv("LOG_LEVEL") == "debug" { logger.DefaultLogLevel = logging.LogLevelDebug } - cfg := &turn.ClientConfig{ + + turnClientCfg := &turn.ClientConfig{ STUNServerAddr: relayDetails.TurnServerAddress, TURNServerAddr: relayDetails.TurnServerAddress, - Conn: turn.NewSTUNConn(conn), Username: relayDetails.TurnServerUsername, Password: relayDetails.TurnServerPassword, Realm: relayDetails.TurnServerRealm, LoggerFactory: logger, } - client, err := turn.NewClient(cfg) + turnAddr, err := net.ResolveUDPAddr("udp4", relayDetails.TurnServerAddress) + if err != nil { + return fmt.Errorf("Failed to parse turn server address: %w", err) + } + + // Dial TURN Server + if relayPort == "5349" { + caCertPool := x509.NewCertPool() + a, _ := os.ReadFile("/Users/akhilmhdh/Akhi/infisical-gateway-poc/certificates/test/ca-cert.pem") + // caCertPool.AppendCertsFromPEM([]byte(g.config.CertificateChain)) + caCertPool.AppendCertsFromPEM(a) + + log.Info().Msgf("Provided relay port %s. Using TLS", relayPort) + conn, err := dtls.Dial("udp", turnAddr, &dtls.Config{ + ServerName: relayAddress, + RootCAs: caCertPool, + }) + if err != nil { + return fmt.Errorf("Failed to connect with relay server: %w", err) + } + turnClientCfg.Conn = turn.NewSTUNConn(conn) + } else { + log.Info().Msgf("Provided relay port %s. Using non TLS connection.", relayPort) + conn, err := net.ListenPacket("udp4", turnAddr.String()) + if err != nil { + return fmt.Errorf("Failed to connect with relay server: %w", err) + } + + turnClientCfg.Conn = conn + } + + client, err := turn.NewClient(turnClientCfg) if err != nil { return fmt.Errorf("Failed to create relay client: %w", err) } @@ -168,7 +180,6 @@ func (g *Gateway) Listen(ctx context.Context) error { ClientAuth: tls.RequireAndVerifyClientCert, NextProtos: []string{"infisical-gateway"}, } - // Setup QUIC listener on the relayConn quicConfig := &quic.Config{ EnableDatagrams: true, @@ -176,7 +187,6 @@ func (g *Gateway) Listen(ctx context.Context) error { KeepAlivePeriod: 2 * time.Second, } - g.registerRelayIsActive(ctx, errCh) quicListener, err := quic.Listen(relayUdpConnection, tlsConfig, quicConfig) if err != nil { return fmt.Errorf("Failed to listen for QUIC: %w", err) @@ -185,6 +195,8 @@ func (g *Gateway) Listen(ctx context.Context) error { log.Printf("Listener started on %s", quicListener.Addr()) + g.registerRelayIsActive(ctx, quicListener.Addr().String(), errCh) + log.Info().Msg("Gateway started successfully") var wg sync.WaitGroup @@ -320,13 +332,12 @@ func (g *Gateway) createPermissionForStaticIps(staticIps string) error { return nil } -func (g *Gateway) registerRelayIsActive(ctx context.Context, errCh chan error) error { +func (g *Gateway) registerRelayIsActive(ctx context.Context, addr string, errCh chan error) error { ticker := time.NewTicker(15 * time.Second) maxFailures := 3 failures := 0 log.Info().Msg("Starting relay connection health check") - go func() { time.Sleep(5 * time.Second) for { @@ -335,36 +346,26 @@ func (g *Gateway) registerRelayIsActive(ctx context.Context, errCh chan error) e log.Info().Msg("Stopping relay connection health check") return case <-ticker.C: - func() { - log.Debug().Msg("Performing relay connection health check") - - if g.client == nil { - failures++ - log.Warn().Int("failures", failures).Msg("TURN client is nil") - if failures >= maxFailures { - errCh <- fmt.Errorf("relay connection check failed: TURN client is nil") - } + log.Debug().Msg("Performing relay connection health check") + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + // Try to establish a QUIC connection + conn, err := quic.DialAddr(ctx, addr, &tls.Config{ + InsecureSkipVerify: false, // Skip certificate verification + NextProtos: []string{"infisical-gateway"}, + }, nil) + if err != nil && !strings.Contains(err.Error(), "tls:") { + failures++ + log.Warn().Err(err).Int("failures", failures).Msg("Failed to refresh TURN permissions") + if failures >= maxFailures { + errCh <- fmt.Errorf("relay connection check failed: %w", err) return } - - // we try to refresh permissions - this is a lightweight operation - // that will fail immediately if the UDP connection is broken. good for health check - log.Debug().Msg("Refreshing TURN permissions to verify connection") - if err := g.createPermissionForStaticIps(g.config.InfisicalStaticIp); err != nil { - failures++ - log.Warn().Err(err).Int("failures", failures).Msg("Failed to refresh TURN permissions") - if failures >= maxFailures { - errCh <- fmt.Errorf("relay connection check failed: %w", err) - } - return - } - - log.Debug().Msg("Successfully refreshed TURN permissions - connection is healthy") - if failures > 0 { - log.Info().Int("previous_failures", failures).Msg("Relay connection restored") - failures = 0 - } - }() + continue + } + if conn != nil { + defer conn.CloseWithError(0, "All good") + } } } }() diff --git a/cli/packages/gateway/relay.go b/cli/packages/gateway/relay.go index bbd1332a4..08a5eb247 100644 --- a/cli/packages/gateway/relay.go +++ b/cli/packages/gateway/relay.go @@ -4,7 +4,6 @@ package gateway import ( - "context" "crypto/tls" "crypto/x509" "errors" @@ -12,12 +11,13 @@ import ( "net" "os" "os/signal" - "runtime" + + // "runtime" "strconv" "syscall" - udplistener "github.com/Infisical/infisical-merge/packages/gateway/udp_listener" "github.com/Infisical/infisical-merge/packages/systemd" + "github.com/pion/dtls/v3" "github.com/pion/logging" "github.com/pion/turn/v4" "github.com/rs/zerolog/log" @@ -108,7 +108,7 @@ func NewGatewayRelay(configFilePath string) (*GatewayRelay, error) { } func (g *GatewayRelay) Run() error { - addr, err := net.ResolveTCPAddr("tcp", "0.0.0.0:"+strconv.Itoa(g.Config.Port)) + addr, err := net.ResolveUDPAddr("udp", "0.0.0.0:"+strconv.Itoa(g.Config.Port)) if err != nil { return fmt.Errorf("Failed to parse server address: %s", err) } @@ -117,13 +117,6 @@ func (g *GatewayRelay) Run() error { // and process them yourself. logger := logging.NewDefaultLeveledLoggerForScope("lt-creds", logging.LogLevelTrace, os.Stdout) - // Create `numThreads` UDP listeners to pass into pion/turn - // pion/turn itself doesn't allocate any UDP sockets, but lets the user pass them in - // this allows us to add logging, storage or modify inbound/outbound traffic - // UDP listeners share the same local address:port with setting SO_REUSEPORT and the kernel - // will load-balance received packets per the IP 5-tuple - listenerConfig := udplistener.SetupListenerConfig() - publicIP := g.Config.PublicIP relayAddressGenerator := &turn.RelayAddressGeneratorPortRange{ RelayAddress: net.ParseIP(publicIP), // Claim that we are listening on IP passed by user @@ -132,49 +125,54 @@ func (g *GatewayRelay) Run() error { MaxPort: g.Config.RelayMaxPort, } - threadNum := runtime.NumCPU() - listenerConfigs := make([]turn.ListenerConfig, threadNum) - var connAddress string - for i := 0; i < threadNum; i++ { - conn, listErr := listenerConfig.Listen(context.Background(), addr.Network(), addr.String()) - if listErr != nil { - return fmt.Errorf("Failed to allocate TCP listener at %s:%s %s", addr.Network(), addr.String(), listErr) - } - - listenerConfigs[i] = turn.ListenerConfig{ - RelayAddressGenerator: relayAddressGenerator, - } - - if g.Config.isTlsEnabled { - caCertPool := x509.NewCertPool() - caCertPool.AppendCertsFromPEM([]byte(g.Config.tlsCa)) - - listenerConfigs[i].Listener = tls.NewListener(conn, &tls.Config{ - Certificates: []tls.Certificate{g.Config.tls}, - ClientCAs: caCertPool, - }) - } else { - listenerConfigs[i].Listener = conn - } - connAddress = conn.Addr().String() - } - loggerF := logging.NewDefaultLoggerFactory() loggerF.DefaultLogLevel = logging.LogLevelDebug + caCertPool := x509.NewCertPool() + caCertPool.AppendCertsFromPEM([]byte(g.Config.tlsCa)) + + listenerConfigs := make([]turn.ListenerConfig, 0) + packetConfigs := make([]turn.PacketConnConfig, 0) + + if g.Config.isTlsEnabled { + caCertPool := x509.NewCertPool() + caCertPool.AppendCertsFromPEM([]byte(g.Config.tlsCa)) + dtlsServer, err := dtls.Listen("udp", addr, &dtls.Config{ + Certificates: []tls.Certificate{g.Config.tls}, + ClientCAs: caCertPool, + }) + if err != nil { + return fmt.Errorf("Failed to start dtls server: %w", err) + } + listenerConfigs = append(listenerConfigs, turn.ListenerConfig{ + RelayAddressGenerator: relayAddressGenerator, + Listener: dtlsServer, + }) + } else { + udpListener, err := net.ListenPacket("udp4", "0.0.0.0:"+strconv.Itoa(g.Config.Port)) + if err != nil { + return fmt.Errorf("Failed to relay udp listener: %w", err) + } + packetConfigs = append(packetConfigs, turn.PacketConnConfig{ + RelayAddressGenerator: relayAddressGenerator, + PacketConn: udpListener, + }) + } + server, err := turn.NewServer(turn.ServerConfig{ Realm: g.Config.Realm, AuthHandler: turn.LongTermTURNRESTAuthHandler(g.Config.AuthSecret, logger), // PacketConnConfigs is a list of UDP Listeners and the configuration around them - ListenerConfigs: listenerConfigs, - LoggerFactory: loggerF, + ListenerConfigs: listenerConfigs, + PacketConnConfigs: packetConfigs, + LoggerFactory: loggerF, }) if err != nil { return fmt.Errorf("Failed to start server: %w", err) } - log.Info().Msgf("Relay listening on %s\n", connAddress) + log.Info().Msgf("Relay listening on %d\n", g.Config.Port) // make this compatiable with systemd notify mode systemd.SdNotify(false, systemd.SdNotifyReady)