tailcat: don't crash the server when a client resets before it's served

tailcat.go

@@ -711,7 +711,7 @@ func (s *Server) startLocked(ctx context.Context) error {
 			if s.OnTCP == nil {
 				return nil, true // send RST
 			}
-			return s.OnTCP(dst.Port()), true
+			return recoverConnHandler(logf, s.OnTCP(dst.Port())), true
 		}
 		if s.OnTCPForward == nil {
 			return nil, true // send RST
@@ -722,7 +722,7 @@ func (s *Server) startLocked(ctx context.Context) error {
 			copy(a4[:], d6[12:16])
 			dst = netip.AddrPortFrom(netip.AddrFrom4(a4), dst.Port())
 		}
-		return s.OnTCPForward(dst), true
+		return recoverConnHandler(logf, s.OnTCPForward(dst)), true
 	}
 	ns.GetUDPHandlerForFlow = func(src, dst netip.AddrPort) (handler func(nettype.ConnPacketConn), intercept bool) {
 		var h func(ConnPacketConn)
@@ -2433,6 +2433,27 @@ func ProxyConns(a, b net.Conn) {
 
 const proxyConnDrainTimeout = 5 * time.Second
 
+// recoverConnHandler wraps h so that, as with net/http, a panic while
+// serving one connection is logged and closes that connection instead
+// of crashing the whole server. A nil h stays nil, which resets the
+// connection.
+func recoverConnHandler(logf logger.Logf, h func(net.Conn)) func(net.Conn) {
+	if h == nil {
+		return nil
+	}
+	return func(c net.Conn) {
+		defer func() {
+			if err := recover(); err != nil {
+				buf := make([]byte, 64<<10)
+				buf = buf[:runtime.Stack(buf, false)]
+				logf("panic serving TCP connection: %v\n%s", err, buf)
+				c.Close()
+			}
+		}()
+		h(c)
+	}
+}
+
 // closeProxyConn waits up to proxyConnDrainTimeout for a gVisor TCP connection
 // whose peer has already closed to acknowledge our FIN, then closes it.
 // Calling Close immediately after CloseWrite can race gVisor's protocol

tailcat_exec.go

@@ -25,8 +25,16 @@ import (
 // command has exited.
 func (s *Server) ExecConnHandler(argv []string) func(net.Conn) {
 	return func(c net.Conn) {
+		// netstack reports no addresses once a connection is no longer
+		// connected: the client reset it before it got here.
+		local, remote := c.LocalAddr(), c.RemoteAddr()
+		if local == nil || remote == nil {
+			s.lb.logf("exec %v: connection closed before it was served", argv[0])
+			c.Close()
+			return
+		}
 		cmd := exec.Command(argv[0], argv[1:]...)
-		cmd.Env = append(os.Environ(), s.PeerEnv(c.LocalAddr(), c.RemoteAddr())...)
+		cmd.Env = append(os.Environ(), s.PeerEnv(local, remote)...)
 		cmd.Stderr = os.Stderr
 		if err := runConnCommand(c, cmd); err != nil {
 			s.lb.logf("exec %v: %v", argv[0], err)
@@ -44,10 +52,15 @@ func (s *Server) ExecConnHandler(argv []string) func(net.Conn) {
 //     talking to.
 //   - TAILCAT_REMOTE_ADDR: the peer's tailcat IP:port.
 //   - TAILCAT_LOCAL_ADDR: the server's own IP:port the peer connected to.
+//
+// A nil address, as a closed connection reports, is left out.
 func (s *Server) PeerEnv(local, remote net.Addr) []string {
-	env := []string{
-		"TAILCAT_REMOTE_ADDR=" + remote.String(),
-		"TAILCAT_LOCAL_ADDR=" + local.String(),
+	var env []string
+	if remote != nil {
+		env = append(env, "TAILCAT_REMOTE_ADDR="+remote.String())
+	}
+	if local != nil {
+		env = append(env, "TAILCAT_LOCAL_ADDR="+local.String())
 	}
 	if k, ok := s.PeerKey(remote); ok {
 		env = append(env, "TAILCAT_PEER_KEY="+k.String())

tailcat_exec_test.go

@@ -0,0 +1,125 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package tailcat
+
+import (
+	"context"
+	"io"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"gvisor.dev/gvisor/pkg/tcpip"
+	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
+	"gvisor.dev/gvisor/pkg/tcpip/network/ipv4"
+	"tailscale.com/tstest/integration"
+	"tailscale.com/wgengine/filter"
+)
+
+// TestExecConnHandlerResetBeforeServed covers a client that resets its
+// connection between the handshake and the handler starting: gonet
+// then reports nil local and remote addresses.
+func TestExecConnHandlerResetBeforeServed(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	clientAddr := tcpip.AddrFrom4([4]byte{192, 0, 2, 1})
+	serverAddr := tcpip.AddrFrom4([4]byte{192, 0, 2, 2})
+	clientStack, clientLink := newTestTCPStack(t, clientAddr)
+	serverStack, serverLink := newTestTCPStack(t, serverAddr)
+	go relayTestPackets(ctx, clientLink, serverLink, nil)
+	go relayTestPackets(ctx, serverLink, clientLink, nil)
+
+	ln, err := gonet.ListenTCP(serverStack, tcpip.FullAddress{NIC: testNICID, Addr: serverAddr, Port: 1234}, ipv4.ProtocolNumber)
+	if err != nil {
+		t.Fatalf("ListenTCP: %v", err)
+	}
+	defer ln.Close()
+	accepted := make(chan net.Conn, 1)
+	go func() {
+		c, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		accepted <- c
+	}()
+	client, err := gonet.DialTCP(clientStack, tcpip.FullAddress{NIC: testNICID, Addr: serverAddr, Port: 1234}, ipv4.ProtocolNumber)
+	if err != nil {
+		t.Fatalf("DialTCP: %v", err)
+	}
+	server := <-accepted
+	defer server.Close()
+
+	ep, _ := gonetTCPConnInternals(client)
+	ep.(tcpip.Endpoint).Abort() // sends a RST
+	for deadline := time.Now().Add(5 * time.Second); server.RemoteAddr() != nil; time.Sleep(10 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatal("server never saw the client's reset")
+		}
+	}
+
+	ran := filepath.Join(t.TempDir(), "ran")
+	s := &Server{lb: &locoBackend{logf: t.Logf}}
+	s.ExecConnHandler([]string{"/bin/sh", "-c", "touch " + ran})(server)
+	if _, err := os.Stat(ran); err == nil {
+		t.Error("the command ran for a connection that was already reset")
+	}
+}
+
+// TestServerSurvivesHandlerPanic checks that a panic in one
+// connection's handler doesn't take the server down with it.
+func TestServerSurvivesHandlerPanic(t *testing.T) {
+	t.Parallel()
+
+	dm := integration.RunDERPAndSTUN(t, mkLogger(t, "derpstun"), "127.0.0.1")
+	reg := dm.Regions[1]
+	if reg == nil {
+		t.Fatal("no region 1 in derpmap")
+	}
+
+	s := &Server{Logf: mkLogger(t, "server"), Region: reg}
+	t.Cleanup(func() { s.Close() })
+	s.OnTCP = func(port uint16) func(net.Conn) {
+		if port == 81 {
+			return func(net.Conn) { panic("handler bug") }
+		}
+		return func(c net.Conn) {
+			io.WriteString(c, "ok")
+			c.Close()
+		}
+	}
+	s.ServedTCPPorts = []filter.PortRange{{First: 80, Last: 81}}
+	if err := s.Start(); err != nil {
+		t.Fatalf("server Start: %v", err)
+	}
+
+	c := &Client{Server: s.TailcatAddr(), Logf: mkLogger(t, "client")}
+	t.Cleanup(func() { c.Close() })
+	PingForTest(t, s, c)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	bad, err := c.DialTCPPort(ctx, 81)
+	if err != nil {
+		t.Fatalf("DialTCPPort(81): %v", err)
+	}
+	bad.SetReadDeadline(time.Now().Add(5 * time.Second))
+	if _, err := io.ReadAll(bad); err != nil {
+		t.Logf("port 81 read: %v", err)
+	}
+	bad.Close()
+
+	conn, err := c.DialTCPPort(ctx, 80)
+	if err != nil {
+		t.Fatalf("DialTCPPort(80) after the panic: %v", err)
+	}
+	defer conn.Close()
+	conn.SetReadDeadline(time.Now().Add(5 * time.Second))
+	got, err := io.ReadAll(conn)
+	if err != nil || string(got) != "ok" {
+		t.Fatalf("port 80 after the panic = %q, %v; want %q", got, err, "ok")
+	}
+}