diff --git a/stack_system.go b/stack_system.go index dd561da..41644bd 100644 --- a/stack_system.go +++ b/stack_system.go @@ -7,6 +7,8 @@ import ( "net/netip" "os" "slices" + "sync" + "sync/atomic" "syscall" "time" @@ -45,17 +47,25 @@ type System struct { icmpTimeout time.Duration tcpListener net.Listener tcpListener6 net.Listener - tcpPort uint16 - tcpPort6 uint16 - tcpNat *TCPNat - udpNat *UDPNat - udpNATOptions UDPNatOptions - dispatcher *ForwardDispatcher - bindInterface bool - interfaceFinder control.InterfaceFinder - frontHeadroom int - txChecksumOffload bool - multiPendingPackets bool + // lx/040: ports are written by acceptLoop on self-heal relisten and read + // concurrently from the tunLoop path (dispatch filter + NAT rewrite) — + // they must be atomic. listenAccess serializes listener replacement + // against Close(); closing marks a deliberate shutdown so acceptLoop can + // tell it apart from the listener dying out from under the stack. + tcpPort atomic.Uint32 + tcpPort6 atomic.Uint32 + closing atomic.Bool + listenAccess sync.Mutex + acceptRecoveries atomic.Uint32 + tcpNat *TCPNat + udpNat *UDPNat + udpNATOptions UDPNatOptions + dispatcher *ForwardDispatcher + bindInterface bool + interfaceFinder control.InterfaceFinder + frontHeadroom int + txChecksumOffload bool + multiPendingPackets bool } type Session struct { @@ -124,10 +134,15 @@ func (s *System) ResetNetwork() { } func (s *System) Close() error { + // lx/040: mark the deliberate shutdown BEFORE closing the listeners so + // acceptLoop exits quietly instead of treating it as a foreign kill. + s.closing.Store(true) s.dispatcher.Close() if s.udpNat != nil { s.udpNat.Close() } + s.listenAccess.Lock() + defer s.listenAccess.Unlock() return common.Close( s.tcpListener, s.tcpListener6, @@ -143,8 +158,10 @@ func (s *System) Start() error { return nil } -func (s *System) start() error { - _ = fixWindowsFirewall() +// lx/040: TCP forwarder bind, shared by start() and the acceptLoop self-heal +// relisten path. isIPv6 selects the address family; the bind-to-interface +// Control and the EADDRNOTAVAIL retry loop match the original start() code. +func (s *System) listenTCP(isIPv6 bool) (net.Listener, error) { var listener net.ListenConfig if s.bindInterface { listener.Control = control.Append(listener.Control, func(network, address string, conn syscall.RawConn) error { @@ -155,37 +172,50 @@ func (s *System) start() error { return nil }) } + network := "tcp4" + address := s.inet4Address + if isIPv6 { + network = "tcp6" + address = s.inet6Address + } + var ( + tcpListener net.Listener + err error + ) + for range 3 { + tcpListener, err = listenNetworkNamespace(s.ctx, s.netNs, listener, network, net.JoinHostPort(address.String(), "0")) + if !retryableListenError(err) { + break + } + time.Sleep(time.Second) + } + if err != nil { + return nil, err + } + return tcpListener, nil +} + +func (s *System) start() error { + _ = fixWindowsFirewall() var tcpListener net.Listener var err error if s.inet4NextAddress.IsValid() { - for range 3 { - tcpListener, err = listenNetworkNamespace(s.ctx, s.netNs, listener, "tcp4", net.JoinHostPort(s.inet4Address.String(), "0")) - if !retryableListenError(err) { - break - } - time.Sleep(time.Second) - } + tcpListener, err = s.listenTCP(false) if err != nil { return err } s.tcpListener = tcpListener - s.tcpPort = M.SocksaddrFromNet(tcpListener.Addr()).Port - go s.acceptLoop(tcpListener) + s.tcpPort.Store(uint32(M.SocksaddrFromNet(tcpListener.Addr()).Port)) + go s.acceptLoop(tcpListener, false) } if s.inet6NextAddress.IsValid() { - for range 3 { - tcpListener, err = listenNetworkNamespace(s.ctx, s.netNs, listener, "tcp6", net.JoinHostPort(s.inet6Address.String(), "0")) - if !retryableListenError(err) { - break - } - time.Sleep(time.Second) - } + tcpListener, err = s.listenTCP(true) if err != nil { return err } s.tcpListener6 = tcpListener - s.tcpPort6 = M.SocksaddrFromNet(tcpListener.Addr()).Port - go s.acceptLoop(tcpListener) + s.tcpPort6.Store(uint32(M.SocksaddrFromNet(tcpListener.Addr()).Port)) + go s.acceptLoop(tcpListener, true) } s.tcpNat = NewNat(s.ctx, s.udpTimeout) udpNATOptions := s.udpNATOptions @@ -367,11 +397,28 @@ func (s *System) processPacket(packet []byte) bool { return writeBack } -func (s *System) acceptLoop(listener net.Listener) { +func (s *System) acceptLoop(listener net.Listener, isIPv6 bool) { for { conn, err := listener.Accept() if err != nil { - return + // lx/040 (SPECS/TASKS/040): upstream silently returns on ANY Accept + // error, leaving the stack alive but every new TCP SYN NAT-rewritten + // onto a dead port (instant RST) until a VPN restart — the LxBox §047 + // "browser dead, QUIC alive" failure. A deliberate System.Close is the + // only quiet exit; anything else means the listener died out from + // under us (e.g. a foreign close on a reused fd number from the + // Java side of the shared Android process) — log it (the errno names + // the killer) and recreate the listener. + if s.closing.Load() { + return + } + newListener, healErr := s.healListener(listener, isIPv6, err) + if healErr != nil { + s.logger.Error("system stack: tcp", ipVersionSuffix(isIPv6), " accept loop died: ", err, "; relisten failed: ", healErr) + return + } + listener = newListener + continue } connPort := M.SocksaddrFromNet(conn.RemoteAddr()).Port session := s.tcpNat.LookupBack(connPort) @@ -383,6 +430,47 @@ func (s *System) acceptLoop(listener net.Listener) { } } +// lx/040: recreate a TCP forwarder listener that died out from under the +// stack. Returns the replacement listener after publishing it (listener field +// + atomic port) under listenAccess, or an error if the stack is closing or +// the bind failed. +func (s *System) healListener(dead net.Listener, isIPv6 bool, cause error) (net.Listener, error) { + port := &s.tcpPort + if isIPv6 { + port = &s.tcpPort6 + } + oldPort := port.Load() + s.logger.Warn("system stack: tcp", ipVersionSuffix(isIPv6), " listener (port ", oldPort, ") accept failed: ", cause, " — recreating listener") + _ = dead.Close() // release netpoll state; harmless if already closed + newListener, err := s.listenTCP(isIPv6) + if err != nil { + return nil, err + } + s.listenAccess.Lock() + defer s.listenAccess.Unlock() + if s.closing.Load() { + _ = newListener.Close() + return nil, net.ErrClosed + } + if isIPv6 { + s.tcpListener6 = newListener + } else { + s.tcpListener = newListener + } + newPort := uint32(M.SocksaddrFromNet(newListener.Addr()).Port) + port.Store(newPort) + recoveries := s.acceptRecoveries.Add(1) + s.logger.Warn("system stack: tcp", ipVersionSuffix(isIPv6), " listener recreated (port ", oldPort, " → ", newPort, ", recoveries: ", recoveries, ")") + return newListener, nil +} + +func ipVersionSuffix(isIPv6 bool) string { + if isIPv6 { + return "6" + } + return "4" +} + func (s *System) dispatchIPv4(ipHdr header.IPv4, destination netip.Addr) bool { switch ipHdr.TransportProtocol() { case header.TCPProtocolNumber: @@ -392,7 +480,7 @@ func (s *System) dispatchIPv4(ipHdr header.IPv4, destination netip.Addr) bool { if ipHdr.SourceAddr() == s.inet4Address && ipHdr.FragmentOffset() == 0 && len(ipHdr.Payload()) >= header.TCPMinimumSize && - header.TCP(ipHdr.Payload()).SourcePort() == s.tcpPort { + header.TCP(ipHdr.Payload()).SourcePort() == uint16(s.tcpPort.Load()) { return false } case header.ICMPv4ProtocolNumber: @@ -411,7 +499,7 @@ func (s *System) dispatchIPv6(ipHdr header.IPv6, destination netip.Addr) bool { } if ipHdr.SourceAddr() == s.inet6Address && len(ipHdr.Payload()) >= header.TCPMinimumSize && - header.TCP(ipHdr.Payload()).SourcePort() == s.tcpPort6 { + header.TCP(ipHdr.Payload()).SourcePort() == uint16(s.tcpPort6.Load()) { return false } case header.ICMPv6ProtocolNumber: @@ -475,7 +563,7 @@ func (s *System) processIPv4TCP(ipHdr header.IPv4, tcpHdr header.TCP) (bool, err destination := netip.AddrPortFrom(ipHdr.DestinationAddr(), tcpHdr.DestinationPort()) if !destination.Addr().IsGlobalUnicast() { return false, nil - } else if source.Addr() == s.inet4Address && source.Port() == s.tcpPort { + } else if source.Addr() == s.inet4Address && source.Port() == uint16(s.tcpPort.Load()) { session := s.tcpNat.LookupBack(destination.Port()) if session == nil { return false, E.New("ipv4: tcp: session not found: ", destination.Port()) @@ -501,7 +589,7 @@ func (s *System) processIPv4TCP(ipHdr header.IPv4, tcpHdr header.TCP) (bool, err } rewriteIPv4TCP(ipHdr, tcpHdr, s.txChecksumOffload, s.inet4NextAddress, natPort, true, - s.inet4Address, s.tcpPort, true) + s.inet4Address, uint16(s.tcpPort.Load()), true) } } return true, nil @@ -512,7 +600,7 @@ func (s *System) processIPv6TCP(ipHdr header.IPv6, tcpHdr header.TCP) (bool, err destination := netip.AddrPortFrom(ipHdr.DestinationAddr(), tcpHdr.DestinationPort()) if !destination.Addr().IsGlobalUnicast() { return false, nil - } else if source.Addr() == s.inet6Address && source.Port() == s.tcpPort6 { + } else if source.Addr() == s.inet6Address && source.Port() == uint16(s.tcpPort6.Load()) { session := s.tcpNat.LookupBack(destination.Port()) if session == nil { return false, E.New("ipv6: tcp: session not found: ", destination.Port()) @@ -538,7 +626,7 @@ func (s *System) processIPv6TCP(ipHdr header.IPv6, tcpHdr header.TCP) (bool, err } rewriteIPv6TCP(ipHdr, tcpHdr, s.txChecksumOffload, s.inet6NextAddress, natPort, true, - s.inet6Address, s.tcpPort6, true) + s.inet6Address, uint16(s.tcpPort6.Load()), true) } } return true, nil diff --git a/stack_system_selfheal_test.go b/stack_system_selfheal_test.go new file mode 100644 index 0000000..ab063c2 --- /dev/null +++ b/stack_system_selfheal_test.go @@ -0,0 +1,117 @@ +package tun + +// lx/040 (SPECS/TASKS/040-SINGTUN_ACCEPTLOOP_SELFHEAL): acceptLoop self-heal. +// +// Red/green против апстрима 2d9b8aed5fe2: там acceptLoop(listener) при любой +// ошибке Accept молча выходит навсегда — восстановления нет, порт не меняется, +// новый connect вечно бьётся в мёртвый сокет. Для red-прогона на чистом +// апстрим-чекауте достаточно адаптировать хелперы ниже (currentTCPPort → +// s.tcpPort, spawnAcceptLoop → go s.acceptLoop(ln)): тест упадёт по таймауту +// ожидания восстановления. + +import ( + "context" + "fmt" + "net" + "net/netip" + "testing" + "time" + + "github.com/sagernet/sing/common/logger" +) + +func newSelfHealTestSystem(t *testing.T) *System { + t.Helper() + s := &System{ + ctx: context.Background(), + logger: logger.NOP(), + inet4Address: netip.MustParseAddr("127.0.0.1"), + udpTimeout: time.Minute, + } + s.tcpNat = NewNat(s.ctx, s.udpTimeout) + ln, err := s.listenTCP(false) + if err != nil { + t.Fatalf("listenTCP: %v", err) + } + s.tcpListener = ln + s.tcpPort.Store(uint32(ln.Addr().(*net.TCPAddr).Port)) + spawnAcceptLoop(s, ln) + return s +} + +func currentTCPPort(s *System) uint32 { + return s.tcpPort.Load() +} + +func spawnAcceptLoop(s *System, ln net.Listener) { + go s.acceptLoop(ln, false) +} + +func dialForwarder(t *testing.T, port uint32) error { + t.Helper() + conn, err := net.DialTimeout("tcp4", fmt.Sprintf("127.0.0.1:%d", port), time.Second) + if err == nil { + _ = conn.Close() + } + return err +} + +// Убийство listener'а мимо System.Close (эмуляция чужого close по +// переиспользованному fd-номеру) должно приводить к пересозданию listener'а +// и продолжению приёма TCP, а не к вечной смерти петли. +func TestSystemAcceptLoopSelfHeal(t *testing.T) { + s := newSelfHealTestSystem(t) + oldPort := currentTCPPort(s) + + if err := dialForwarder(t, oldPort); err != nil { + t.Fatalf("healthy listener refused connect: %v", err) + } + + // Убить listener из-под стека: closing НЕ выставлен. + _ = s.tcpListener.Close() + + deadline := time.Now().Add(5 * time.Second) + healed := false + for time.Now().Before(deadline) { + if s.acceptRecoveries.Load() > 0 { + healed = true + break + } + time.Sleep(10 * time.Millisecond) + } + if !healed { + t.Fatalf("acceptLoop did not recover within 5s (upstream behavior: silent permanent death)") + } + + newPort := currentTCPPort(s) + if newPort == oldPort { + t.Fatalf("recovered port equals dead port %d — relisten did not publish a new port", oldPort) + } + if err := dialForwarder(t, newPort); err != nil { + t.Fatalf("connect to recreated listener (port %d) failed: %v", newPort, err) + } + if got := s.acceptRecoveries.Load(); got != 1 { + t.Fatalf("acceptRecoveries = %d, want 1", got) + } + + s.closing.Store(true) + _ = s.tcpListener.Close() +} + +// Штатное закрытие (closing выставлен, как это делает System.Close) обязано +// оставаться тихим: без пересозданий и без роста счётчика. +func TestSystemAcceptLoopQuietOnClose(t *testing.T) { + s := newSelfHealTestSystem(t) + oldPort := currentTCPPort(s) + + s.closing.Store(true) + _ = s.tcpListener.Close() + + time.Sleep(300 * time.Millisecond) + if got := s.acceptRecoveries.Load(); got != 0 { + t.Fatalf("deliberate close triggered %d recoveries, want 0", got) + } + if port := currentTCPPort(s); port != oldPort { + t.Fatalf("deliberate close changed port %d → %d", oldPort, port) + } +}