From bb233c72b6a00c8e84bc72bca62264a31f79896c Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Thu, 3 Sep 2026 19:22:00 +0900 Subject: [PATCH] [client] Rebuild the overlay listeners when the TUN is renewed (#7397) --- client/internal/dnsfwd/forwarder.go | 125 ++++++++++++++++++++--- client/internal/dnsfwd/forwarder_test.go | 52 ++++++++++ client/internal/dnsfwd/manager.go | 10 ++ client/internal/engine.go | 87 +++++++++++++++- client/internal/engine_ssh.go | 36 ++++++- client/ssh/auth/auth.go | 19 ++++ client/ssh/server/server.go | 33 +++++- 7 files changed, 341 insertions(+), 21 deletions(-) diff --git a/client/internal/dnsfwd/forwarder.go b/client/internal/dnsfwd/forwarder.go index b7e5a10e3..e3cb597be 100644 --- a/client/internal/dnsfwd/forwarder.go +++ b/client/internal/dnsfwd/forwarder.go @@ -54,12 +54,20 @@ type DNSForwarder struct { ttl uint32 statusRecorder *peer.Status - dnsServer *dns.Server - mux *dns.ServeMux - tcpServer *dns.Server - tcpMux *dns.ServeMux + mux *dns.ServeMux + tcpMux *dns.ServeMux - mutex sync.RWMutex + mutex sync.RWMutex + // closed records that Close has run, so a Listen still in flight does not + // go on to serve sockets nobody will shut down. + closed bool + // The sockets are kept alongside the servers because closing them is the + // only stop that always works: a server whose ActivateAndServe has not run + // yet refuses to shut down, and would otherwise start serving afterwards. + udpConn net.PacketConn + tcpLn net.Listener + dnsServer *dns.Server + tcpServer *dns.Server fwdEntries []*ForwarderEntry firewall firewaller resolver resolver @@ -106,7 +114,7 @@ func (f *DNSForwarder) Listen(entries []*ForwarderEntry) error { mux := dns.NewServeMux() f.mux = mux mux.HandleFunc(".", f.handleDNSQueryUDP) - f.dnsServer = &dns.Server{ + dnsServer := &dns.Server{ PacketConn: udpLn, Handler: mux, } @@ -114,22 +122,32 @@ func (f *DNSForwarder) Listen(entries []*ForwarderEntry) error { tcpMux := dns.NewServeMux() f.tcpMux = tcpMux tcpMux.HandleFunc(".", f.handleDNSQueryTCP) - f.tcpServer = &dns.Server{ + tcpServer := &dns.Server{ Listener: tcpLn, Handler: tcpMux, } - f.UpdateDomains(entries) + if !f.publish(udpLn, tcpLn, dnsServer, tcpServer, entries) { + log.Infof("DNS forwarder on %s was closed before it started serving", addrDesc) + if err := udpLn.Close(); err != nil { + log.Debugf("close UDP listener of a closed forwarder: %v", err) + } + if err := tcpLn.Close(); err != nil { + log.Debugf("close TCP listener of a closed forwarder: %v", err) + } + return nil + } + log.Debugf("DNS forwarder serving %d domains", len(entries)) errCh := make(chan error, 2) go func() { log.Infof("DNS UDP listener running on %s", addrDesc) - errCh <- f.dnsServer.ActivateAndServe() + errCh <- dnsServer.ActivateAndServe() }() go func() { log.Infof("DNS TCP listener running on %s", addrDesc) - errCh <- f.tcpServer.ActivateAndServe() + errCh <- tcpServer.ActivateAndServe() }() return <-errCh @@ -151,6 +169,46 @@ func (f *DNSForwarder) createTCPListener(netstackNet *netstack.Net) (net.Listene return net.ListenTCP("tcp", net.TCPAddrFromAddrPort(f.listenAddress)) } +// publish hands the sockets, servers and entries to the forwarder so Close can +// reach them and Domains can report them, and says whether serving may begin. +// Listen runs on its own goroutine, so a Close can arrive before it gets this +// far; false means the caller must close what it created instead of serving on +// it. +// +// The entries go in under the same lock rather than afterwards. Anything that +// reads them in between would otherwise see a forwarder that is listening and +// serves no domain, which for a caller rebuilding one means it comes back +// refusing every routed query. +func (f *DNSForwarder) publish( + udpConn net.PacketConn, + tcpLn net.Listener, + dnsServer, tcpServer *dns.Server, + entries []*ForwarderEntry, +) bool { + f.mutex.Lock() + defer f.mutex.Unlock() + + if f.closed { + return false + } + + f.udpConn = udpConn + f.tcpLn = tcpLn + f.dnsServer = dnsServer + f.tcpServer = tcpServer + f.fwdEntries = entries + return true +} + +// Domains returns the entries currently being served. The slice is replaced +// wholesale by UpdateDomains rather than mutated, so the caller may read it but +// must not write to it. +func (f *DNSForwarder) Domains() []*ForwarderEntry { + f.mutex.RLock() + defer f.mutex.RUnlock() + return f.fwdEntries +} + func (f *DNSForwarder) UpdateDomains(entries []*ForwarderEntry) { f.mutex.Lock() defer f.mutex.Unlock() @@ -189,19 +247,45 @@ func (f *DNSForwarder) removeStaleCacheEntries(oldEntries, newEntries []*Forward } func (f *DNSForwarder) Close(ctx context.Context) error { + // Marked closed under the lock so a Listen that has not published its + // servers yet gives up instead of racing this shutdown. The shutdowns + // themselves block, so they run outside it. + f.mutex.Lock() + f.closed = true + dnsServer, tcpServer := f.dnsServer, f.tcpServer + udpConn, tcpLn := f.udpConn, f.tcpLn + f.mutex.Unlock() + var result *multierror.Error - if f.dnsServer != nil { - if err := f.dnsServer.ShutdownContext(ctx); err != nil { + if dnsServer != nil { + if err := shutdownServer(ctx, dnsServer); err != nil { result = multierror.Append(result, fmt.Errorf("UDP shutdown: %w", err)) } } - if f.tcpServer != nil { - if err := f.tcpServer.ShutdownContext(ctx); err != nil { + if tcpServer != nil { + if err := shutdownServer(ctx, tcpServer); err != nil { result = multierror.Append(result, fmt.Errorf("TCP shutdown: %w", err)) } } + // The sockets are closed even when the shutdowns above reported nothing to + // do. A server that has been published but has not reached + // ActivateAndServe refuses to shut down, and closing what it was about to + // serve on is what stops it: the alternative is a listener still answering + // on an interface that has gone away. A shutdown that did run has already + // closed these, so the second close is expected to fail. + if udpConn != nil { + if err := udpConn.Close(); err != nil { + log.Debugf("close UDP socket of the DNS forwarder: %v", err) + } + } + if tcpLn != nil { + if err := tcpLn.Close(); err != nil { + log.Debugf("close TCP socket of the DNS forwarder: %v", err) + } + } + return nberrors.FormatErrorOrNil(result) } @@ -514,3 +598,16 @@ func attachEDE(resp *dns.Msg, code uint16, text string) { } opt.Option = append(opt.Option, &dns.EDNS0_EDE{InfoCode: code, ExtraText: text}) } + +// shutdownServer shuts a server down gracefully, treating "never started" as +// success. A server that was published but has not reached ActivateAndServe +// has nothing to wind down, and the caller closes its socket regardless, which +// is what actually stops it. dns exports no sentinel for this, so the message +// is all there is to match on. +func shutdownServer(ctx context.Context, server *dns.Server) error { + err := server.ShutdownContext(ctx) + if err == nil || strings.Contains(err.Error(), "server not started") { + return nil + } + return err +} diff --git a/client/internal/dnsfwd/forwarder_test.go b/client/internal/dnsfwd/forwarder_test.go index c69a9166e..a64ba80e7 100644 --- a/client/internal/dnsfwd/forwarder_test.go +++ b/client/internal/dnsfwd/forwarder_test.go @@ -1238,3 +1238,55 @@ func TestDNSForwarder_EmptyQuery(t *testing.T) { assert.Nil(t, mockWriter.GetLastResponse(), "Should not write response for empty query") } + +// TestDNSForwarder_ClosedBeforeItServes covers Listen reaching the point of +// serving after the forwarder has already been closed. Listen runs on its own +// goroutine, so it can get there late, and a socket it starts serving then is +// one nothing will ever close: on Android it keeps answering on an interface +// that has been replaced. The close is sequenced first here rather than raced, +// which pins the same state deterministically. +func TestDNSForwarder_ClosedBeforeItServes(t *testing.T) { + f := NewDNSForwarder(netip.MustParseAddrPort("127.0.0.1:0"), 60, nil, nil, nil) + + require.NoError(t, f.Close(context.Background()), "closing a forwarder that never started") + + done := make(chan error, 1) + go func() { done <- f.Listen(nil) }() + + select { + case err := <-done: + assert.NoError(t, err, "a closed forwarder should give up quietly, not serve") + case <-time.After(5 * time.Second): + t.Fatal("Listen went on to serve after the forwarder was closed") + } +} + +// TestDNSForwarder_CloseStopsUnactivatedServers covers the window between +// Listen publishing its servers and reaching ActivateAndServe. A server that +// has not been activated refuses to shut down, so Close has to close the +// sockets itself or they are left serving. +func TestDNSForwarder_CloseStopsUnactivatedServers(t *testing.T) { + f := NewDNSForwarder(netip.MustParseAddrPort("127.0.0.1:0"), 60, nil, nil, nil) + + udpConn, err := f.createUDPListener(nil) + require.NoError(t, err, "create UDP listener") + tcpLn, err := f.createTCPListener(nil) + require.NoError(t, err, "create TCP listener") + + // Published but deliberately never activated, which is the state Listen is + // in for the moment before it starts serving. + require.True(t, f.publish(udpConn, tcpLn, &dns.Server{PacketConn: udpConn}, &dns.Server{Listener: tcpLn}, nil), + "publishing to an open forwarder") + + tcpAddr := tcpLn.Addr().String() + require.NoError(t, f.Close(context.Background()), "close should report no error for servers it could not shut down") + + _, err = tcpLn.Accept() + assert.Error(t, err, "the TCP socket should be closed after Close") + + conn, err := net.DialTimeout("tcp", tcpAddr, time.Second) + if err == nil { + _ = conn.Close() + t.Fatal("the forwarder is still accepting connections after Close") + } +} diff --git a/client/internal/dnsfwd/manager.go b/client/internal/dnsfwd/manager.go index b8684d177..1c62e908d 100644 --- a/client/internal/dnsfwd/manager.go +++ b/client/internal/dnsfwd/manager.go @@ -117,6 +117,16 @@ func (m *Manager) UpdateDomains(entries []*ForwarderEntry) { m.dnsForwarder.UpdateDomains(entries) } +// Domains returns the entries currently being served, or nil when the +// forwarder is not running. +func (m *Manager) Domains() []*ForwarderEntry { + if m.dnsForwarder == nil { + return nil + } + + return m.dnsForwarder.Domains() +} + func (m *Manager) Stop(ctx context.Context) error { if m.dnsForwarder == nil { return nil diff --git a/client/internal/engine.go b/client/internal/engine.go index 72361c8cb..477fba194 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -94,6 +94,13 @@ const ( // exec, os.Stat); without this bound a single stuck call freezes handleSync, and // thus syncMsgMux, for as long as the call hangs (observed multi-minute freezes). systemInfoTimeout = 15 * time.Second + + // dnsForwarderStopTimeout bounds how long stopping the DNS forwarder waits + // for the queries still in flight. One waiting on an unresponsive upstream + // would otherwise hold the stop for the whole upstream timeout, and the + // stop runs with syncMsgMux held. The sockets are closed either way, so + // giving up costs a query that was already failing. + dnsForwarderStopTimeout = 2 * time.Second ) var ErrResetConnection = fmt.Errorf("reset connection") @@ -321,6 +328,10 @@ type localIpUpdater interface { UpdateLocalIPs() error } +// overlayRebind rebuilds one subsystem's sockets on the current interface. The +// error it returns names its own subsystem, since the caller can only log it. +type overlayRebind func() error + // NewEngine creates a new Connection Engine with probes attached func NewEngine( clientCtx context.Context, @@ -2502,7 +2513,72 @@ func (e *Engine) RenewTun(fd int) error { return fmt.Errorf("wireguard interface not initialized") } - return wgInterface.RenewTun(fd) + if err := wgInterface.RenewTun(fd); err != nil { + return err + } + + e.rebindOverlayListeners() + return nil +} + +// rebindOverlayListeners gives the servers that listen on an overlay address +// sockets on the interface as it is now. +// +// A socket belongs to the interface generation it was created on. Renewing the +// TUN builds a new interface and moves the overlay addresses to it, which +// leaves the old sockets in LISTEN with the uspfilter still logging packets +// arriving for them, while every accept fails with EINVAL for the life of the +// socket: from the outside the server looks alive and answers nothing. On +// Android this happens during a normal startup, where the first TUN is +// established before the routes are known and replaced once they arrive. +// +// Rebinding costs whatever those sockets were carrying, which the renewal has +// already broken. Errors are logged rather than returned: the renewal itself +// succeeded, and failing it would hand the caller a working interface and an +// error. +func (e *Engine) rebindOverlayListeners() { + e.syncMsgMux.Lock() + defer e.syncMsgMux.Unlock() + + for _, rebind := range e.overlayRebinds() { + if err := rebind(); err != nil { + log.Errorf("after TUN renewal: %v", err) + } + } +} + +// overlayRebinds is every subsystem of this engine that holds sockets bound to +// an overlay address, and how to rebuild each one's. +// +// A subsystem that starts listening on an overlay address belongs in this list. +// Leaving it out costs nothing that review would notice and produces a listener +// that stays in LISTEN, is logged as receiving packets, and refuses every +// connection for the life of the process. +func (e *Engine) overlayRebinds() []overlayRebind { + return []overlayRebind{ + e.restartSSHListeners, + e.restartDNSForwarder, + } +} + +// restartDNSForwarder rebuilds the DNS forwarder serving the same domains. +// No-op when it is not running. See Engine.rebindOverlayListeners. +func (e *Engine) restartDNSForwarder() error { + if e.dnsForwardMgr == nil { + return nil + } + // Read from the forwarder before it goes away, so the replacement serves + // the domains in force now rather than a copy kept somewhere else. + entries := e.dnsForwardMgr.Domains() + e.stopDNSForwarder() + // Both halves log their own failures, so the only thing left to report is + // the outcome: a start that failed left the manager nil, and the forwarder + // is now down rather than merely rebound. + e.startDNSForwarder(entries) + if e.dnsForwardMgr == nil { + return errors.New("rebind DNS forwarder: it did not come back up") + } + return nil } // updateDNSForwarder start or stop the DNS forwarder based on the domains and the feature flag @@ -2548,7 +2624,14 @@ func (e *Engine) stopDNSForwarder() { return } - if err := e.dnsForwardMgr.Stop(context.Background()); err != nil { + // Bounded because the shutdown waits for queries still in flight, and one + // waiting on an unresponsive upstream holds it for as long as that lookup + // is allowed to take. This runs with syncMsgMux held, so that wait is one + // the whole engine spends. + ctx, cancel := context.WithTimeout(context.Background(), dnsForwarderStopTimeout) + defer cancel() + + if err := e.dnsForwardMgr.Stop(ctx); err != nil { log.Errorf("failed to stop DNS forward: %v", err) } diff --git a/client/internal/engine_ssh.go b/client/internal/engine_ssh.go index 53d2c1122..60bdfd806 100644 --- a/client/internal/engine_ssh.go +++ b/client/internal/engine_ssh.go @@ -24,6 +24,8 @@ type sshServer interface { Stop() error GetStatus() (bool, []sshserver.SessionInfo) UpdateSSHAuth(config *sshauth.Config) + JWTConfig() *sshserver.JWTConfig + AuthConfig() *sshauth.Config } func (e *Engine) setupSSHPortRedirection() error { @@ -77,7 +79,7 @@ func (e *Engine) updateSSH(sshConf *mgmProto.SSHConfig) error { if e.config.DisableSSHAuth != nil && *e.config.DisableSSHAuth { log.Info("starting SSH server without JWT authentication (authentication disabled by config)") - return e.startSSHServer(nil) + return e.startSSHServer(nil, nil) } if protoJWT := sshConf.GetJwtConfig(); protoJWT != nil { @@ -95,7 +97,7 @@ func (e *Engine) updateSSH(sshConf *mgmProto.SSHConfig) error { MaxTokenAge: protoJWT.GetMaxTokenAge(), } - return e.startSSHServer(jwtConfig) + return e.startSSHServer(jwtConfig, nil) } return errors.New("SSH server requires valid JWT configuration") @@ -231,8 +233,33 @@ func (e *Engine) cleanupSSHConfig() { } } -// startSSHServer initializes and starts the SSH server with proper configuration. -func (e *Engine) startSSHServer(jwtConfig *sshserver.JWTConfig) error { +// restartSSHListeners rebuilds the SSH server so it listens on new sockets, on +// the same terms it was started with. No-op when it is not running. See +// Engine.rebindOverlayListeners for why this is needed. +func (e *Engine) restartSSHListeners() error { + if e.sshServer == nil { + return nil + } + // Read from the server before it goes away. A rebuilt one starts with an + // empty authorizer, which fails closed, so without carrying the + // authorization over every JWT login is refused until the next network map + // happens to bring one. + jwtConfig, authConfig := e.sshServer.JWTConfig(), e.sshServer.AuthConfig() + if err := e.stopSSHServer(); err != nil { + return fmt.Errorf("rebind SSH listeners: %w", err) + } + if err := e.startSSHServer(jwtConfig, authConfig); err != nil { + return fmt.Errorf("rebind SSH listeners: %w", err) + } + return nil +} + +// startSSHServer initializes and starts the SSH server with proper +// configuration. authConfig is the fine-grained authorization to open with, and +// is applied before the server accepts anything: a server that starts listening +// with an empty authorizer refuses the logins that arrive in the meantime. +// Nil leaves it as management has not sent one yet. +func (e *Engine) startSSHServer(jwtConfig *sshserver.JWTConfig, authConfig *sshauth.Config) error { if e.wgInterface == nil { return errors.New("wg interface not initialized") } @@ -240,6 +267,7 @@ func (e *Engine) startSSHServer(jwtConfig *sshserver.JWTConfig) error { serverConfig := &sshserver.Config{ HostKeyPEM: e.config.SSHKey, JWT: jwtConfig, + Auth: authConfig, } server := sshserver.New(serverConfig) diff --git a/client/ssh/auth/auth.go b/client/ssh/auth/auth.go index 079282fdc..92f517fac 100644 --- a/client/ssh/auth/auth.go +++ b/client/ssh/auth/auth.go @@ -3,6 +3,7 @@ package auth import ( "errors" "fmt" + "slices" "sync" log "github.com/sirupsen/logrus" @@ -155,6 +156,24 @@ func (a *Authorizer) GetUserIDClaim() string { return a.userIDClaim } +// Config returns the authorization currently in force. The user list and the +// machine-user map are copies; the originals stay in use here. +func (a *Authorizer) Config() *Config { + a.mu.RLock() + defer a.mu.RUnlock() + + machineUsers := make(map[string][]uint32, len(a.machineUsers)) + for osUser, indexes := range a.machineUsers { + machineUsers[osUser] = slices.Clone(indexes) + } + + return &Config{ + UserIDClaim: a.userIDClaim, + AuthorizedUsers: slices.Clone(a.authorizedUsers), + MachineUsers: machineUsers, + } +} + // findUserIndex finds the index of a hashed user ID in the authorized users list // Returns the index and true if found, 0 and false if not found func (a *Authorizer) findUserIndex(hashedUserID sshuserhash.UserIDHash) (int, bool) { diff --git a/client/ssh/server/server.go b/client/ssh/server/server.go index 6735e0f3b..b32da796e 100644 --- a/client/ssh/server/server.go +++ b/client/ssh/server/server.go @@ -197,6 +197,12 @@ type Config struct { // HostKey is the SSH server host key in PEM format HostKeyPEM []byte + + // Auth is the fine-grained authorization to open with. Nil starts with an + // empty authorizer, which authorizes nobody until UpdateSSHAuth is called. + // Setting it here rather than afterwards means the server never accepts a + // login before it knows who is allowed. + Auth *sshauth.Config } // SessionInfo contains information about an active SSH session @@ -220,7 +226,11 @@ func New(config *Config) *Server { connections: make(map[connKey]*connState), jwtEnabled: config.JWT != nil, jwtConfig: config.JWT, - authorizer: sshauth.NewAuthorizer(), // Initialize with empty config + authorizer: sshauth.NewAuthorizer(), + } + + if config.Auth != nil { + s.authorizer.Update(config.Auth) } return s @@ -461,6 +471,27 @@ func (s *Server) UpdateSSHAuth(config *sshauth.Config) { s.authorizer.Update(config) } +// JWTConfig returns the JWT authentication this server was built with, or nil +// when JWT authentication is disabled. +func (s *Server) JWTConfig() *JWTConfig { + s.mu.RLock() + defer s.mu.RUnlock() + return s.jwtConfig +} + +// AuthConfig returns the fine-grained authorization currently in force, or nil +// when the server has no authorizer. +func (s *Server) AuthConfig() *sshauth.Config { + s.mu.RLock() + authorizer := s.authorizer + s.mu.RUnlock() + + if authorizer == nil { + return nil + } + return authorizer.Config() +} + // ensureJWTValidator initializes the JWT validator and extractor if not already initialized func (s *Server) ensureJWTValidator() error { s.mu.RLock()