feature/taildrop: add integration test

Taildrop has never had an end-to-end test since it was introduced.

This adds a basic one.

It caught two recent refactoring bugs & one from 2022 (0f7da5c7dc0).

This is prep for moving the rest of Taildrop out of LocalBackend, so
we can do more refactorings with some confidence.

Updates #15812

Change-Id: I6182e49c5641238af0bfdd9fea1ef0420c112738
Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>

cmd/tailscaled/tailscaled.go

@@ -573,7 +573,7 @@ func getLocalBackend(ctx context.Context, logf logger.Logf, logID logid.PublicID
 		if ms, ok := sys.MagicSock.GetOK(); ok {
 			debugMux.HandleFunc("/debug/magicsock", ms.ServeHTTPDebug)
 		}
-		go runDebugServer(debugMux, args.debug)
+		go runDebugServer(logf, debugMux, args.debug)
 	}
 
 	ns, err := newNetstack(logf, sys)
@@ -819,12 +819,20 @@ func servePrometheusMetrics(w http.ResponseWriter, r *http.Request) {
 	clientmetric.WritePrometheusExpositionFormat(w)
 }
 
-func runDebugServer(mux *http.ServeMux, addr string) {
+func runDebugServer(logf logger.Logf, mux *http.ServeMux, addr string) {
+	ln, err := net.Listen("tcp", addr)
+	if err != nil {
+		log.Fatalf("debug server: %v", err)
+	}
+	if strings.HasSuffix(addr, ":0") {
+		// Log kernel-selected port number so integration tests
+		// can find it portably.
+		logf("DEBUG-ADDR=%v", ln.Addr())
+	}
 	srv := &http.Server{
-		Addr:    addr,
 		Handler: mux,
 	}
-	if err := srv.ListenAndServe(); err != nil {
+	if err := srv.Serve(ln); err != nil {
 		log.Fatal(err)
 	}
 }

feature/taildrop/integration_test.go

@@ -0,0 +1,170 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package taildrop_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"tailscale.com/client/local"
+	"tailscale.com/client/tailscale/apitype"
+	"tailscale.com/tailcfg"
+	"tailscale.com/tstest"
+	"tailscale.com/tstest/integration"
+	"tailscale.com/tstest/integration/testcontrol"
+)
+
+// TODO(bradfitz): add test where control doesn't send tailcfg.CapabilityFileSharing
+// and verify that we get the "file sharing not enabled by Tailscale admin" error.
+
+// TODO(bradfitz): add test between different users with the peercap to permit that?
+
+func TestTaildropIntegration(t *testing.T) {
+	tstest.Parallel(t)
+	controlOpt := integration.ConfigureControl(func(s *testcontrol.Server) {
+		s.AllNodesSameUser = true // required for Taildrop
+	})
+	env := integration.NewTestEnv(t, controlOpt)
+
+	// Create two nodes:
+	n1 := integration.NewTestNode(t, env)
+	d1 := n1.StartDaemon()
+
+	n2 := integration.NewTestNode(t, env)
+	d2 := n2.StartDaemon()
+
+	n1.AwaitListening()
+	t.Logf("n1 is listening")
+	n2.AwaitListening()
+	t.Logf("n2 is listening")
+	n1.MustUp()
+	t.Logf("n1 is up")
+	n2.MustUp()
+	t.Logf("n2 is up")
+	n1.AwaitRunning()
+	t.Logf("n1 is running")
+	n2.AwaitRunning()
+	t.Logf("n2 is running")
+
+	var peerStableID tailcfg.StableNodeID
+
+	if err := tstest.WaitFor(5*time.Second, func() error {
+		st := n1.MustStatus()
+		if len(st.Peer) == 0 {
+			return errors.New("no peers")
+		}
+		if len(st.Peer) > 1 {
+			return fmt.Errorf("got %d peers; want 1", len(st.Peer))
+		}
+		peer := st.Peer[st.Peers()[0]]
+		peerStableID = peer.ID
+		if peer.ID == st.Self.ID {
+			return errors.New("peer is self")
+		}
+
+		if len(st.TailscaleIPs) == 0 {
+			return errors.New("no Tailscale IPs")
+		}
+
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	const timeout = 30 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	c1 := n1.LocalClient()
+	c2 := n2.LocalClient()
+
+	wantNoWaitingFiles := func(c *local.Client) {
+		t.Helper()
+		files, err := c.WaitingFiles(ctx)
+		if err != nil {
+			t.Fatalf("WaitingFiles: %v", err)
+		}
+		if len(files) != 0 {
+			t.Fatalf("WaitingFiles: got %d files; want 0", len(files))
+		}
+	}
+
+	// Verify c2 has no files.
+	wantNoWaitingFiles(c2)
+
+	gotFile := make(chan bool, 1)
+	go func() {
+		v, err := c2.AwaitWaitingFiles(t.Context(), timeout)
+		if err != nil {
+			return
+		}
+		if len(v) != 0 {
+			gotFile <- true
+		}
+	}()
+
+	fileContents := []byte("hello world this is a file")
+
+	n2ID := n2.MustStatus().Self.ID
+	t.Logf("n2 self.ID = %q; n1's peer[0].ID = %q", n2ID, peerStableID)
+	t.Logf("Doing PushFile ...")
+	err := c1.PushFile(ctx, n2.MustStatus().Self.ID, int64(len(fileContents)), "test.txt", bytes.NewReader(fileContents))
+	if err != nil {
+		t.Fatalf("PushFile from n1->n2: %v", err)
+	}
+	t.Logf("PushFile done")
+
+	select {
+	case <-gotFile:
+		t.Logf("n2 saw AwaitWaitingFiles wake up")
+	case <-ctx.Done():
+		t.Fatalf("n2 timeout waiting for AwaitWaitingFiles")
+	}
+
+	files, err := c2.WaitingFiles(ctx)
+	if err != nil {
+		t.Fatalf("c2.WaitingFiles: %v", err)
+	}
+	if len(files) != 1 {
+		t.Fatalf("c2.WaitingFiles: got %d files; want 1", len(files))
+	}
+	got := files[0]
+	want := apitype.WaitingFile{
+		Name: "test.txt",
+		Size: int64(len(fileContents)),
+	}
+	if got != want {
+		t.Fatalf("c2.WaitingFiles: got %+v; want %+v", got, want)
+	}
+
+	// Download the file.
+	rc, size, err := c2.GetWaitingFile(ctx, got.Name)
+	if err != nil {
+		t.Fatalf("c2.GetWaitingFile: %v", err)
+	}
+	if size != int64(len(fileContents)) {
+		t.Fatalf("c2.GetWaitingFile: got size %d; want %d", size, len(fileContents))
+	}
+	gotBytes, err := io.ReadAll(rc)
+	if err != nil {
+		t.Fatalf("c2.GetWaitingFile: %v", err)
+	}
+	if !bytes.Equal(gotBytes, fileContents) {
+		t.Fatalf("c2.GetWaitingFile: got %q; want %q", gotBytes, fileContents)
+	}
+
+	// Now delete it.
+	if err := c2.DeleteWaitingFile(ctx, got.Name); err != nil {
+		t.Fatalf("c2.DeleteWaitingFile: %v", err)
+	}
+	wantNoWaitingFiles(c2)
+
+	d1.MustCleanShutdown(t)
+	d2.MustCleanShutdown(t)
+}

feature/taildrop/localapi.go

@@ -365,6 +365,7 @@ func serveFiles(h *localapi.Handler, w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		ctx := r.Context()
+		var wfs []apitype.WaitingFile
 		if s := r.FormValue("waitsec"); s != "" && s != "0" {
 			d, err := strconv.Atoi(s)
 			if err != nil {
@@ -375,11 +376,18 @@ func serveFiles(h *localapi.Handler, w http.ResponseWriter, r *http.Request) {
 			var cancel context.CancelFunc
 			ctx, cancel = context.WithDeadline(ctx, deadline)
 			defer cancel()
-		}
-		wfs, err := lb.AwaitWaitingFiles(ctx)
-		if err != nil && ctx.Err() == nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
+			wfs, err = lb.AwaitWaitingFiles(ctx)
+			if err != nil && ctx.Err() == nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
+		} else {
+			var err error
+			wfs, err = lb.WaitingFiles()
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusInternalServerError)
+				return
+			}
 		}
 		w.Header().Set("Content-Type", "application/json")
 		json.NewEncoder(w).Encode(wfs)

tstest/integration/integration.go

@@ -33,6 +33,7 @@ import (
 	"time"
 
 	"go4.org/mem"
+	"tailscale.com/client/local"
 	"tailscale.com/derp"
 	"tailscale.com/derp/derphttp"
 	"tailscale.com/ipn"
@@ -436,6 +437,7 @@ func NewTestEnv(t testing.TB, opts ...TestEnvOpt) *TestEnv {
 	derpMap := RunDERPAndSTUN(t, logger.Discard, "127.0.0.1")
 	logc := new(LogCatcher)
 	control := &testcontrol.Server{
+		Logf:    logger.WithPrefix(t.Logf, "testcontrol: "),
 		DERPMap: derpMap,
 	}
 	control.HTTPTestServer = httptest.NewUnstartedServer(control)
@@ -484,6 +486,7 @@ type TestNode struct {
 
 	mu        sync.Mutex
 	onLogLine []func([]byte)
+	lc        *local.Client
 }
 
 // NewTestNode allocates a temp directory for a new test node.
@@ -500,14 +503,18 @@ func NewTestNode(t *testing.T, env *TestEnv) *TestNode {
 		env:       env,
 		dir:       dir,
 		sockFile:  sockFile,
-		stateFile: filepath.Join(dir, "tailscale.state"),
+		stateFile: filepath.Join(dir, "tailscaled.state"), // matches what cmd/tailscaled uses
 	}
 
-	// Look for a data race. Once we see the start marker, start logging the rest.
+	// Look for a data race or panic.
+	// Once we see the start marker, start logging the rest.
 	var sawRace bool
 	var sawPanic bool
 	n.addLogLineHook(func(line []byte) {
 		lineB := mem.B(line)
+		if mem.Contains(lineB, mem.S("DEBUG-ADDR=")) {
+			t.Log(strings.TrimSpace(string(line)))
+		}
 		if mem.Contains(lineB, mem.S("WARNING: DATA RACE")) {
 			sawRace = true
 		}
@@ -522,6 +529,20 @@ func NewTestNode(t *testing.T, env *TestEnv) *TestNode {
 	return n
 }
 
+func (n *TestNode) LocalClient() *local.Client {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	if n.lc == nil {
+		tr := &http.Transport{}
+		n.lc = &local.Client{
+			Socket:        n.sockFile,
+			UseSocketOnly: true,
+		}
+		n.env.t.Cleanup(tr.CloseIdleConnections)
+	}
+	return n.lc
+}
+
 func (n *TestNode) diskPrefs() *ipn.Prefs {
 	t := n.env.t
 	t.Helper()
@@ -668,9 +689,10 @@ func (n *TestNode) StartDaemonAsIPNGOOS(ipnGOOS string) *Daemon {
 	t := n.env.t
 	cmd := exec.Command(n.env.daemon)
 	cmd.Args = append(cmd.Args,
-		"--state="+n.stateFile,
+		"--statedir="+n.dir,
 		"--socket="+n.sockFile,
 		"--socks5-server=localhost:0",
+		"--debug=localhost:0",
 	)
 	if *verboseTailscaled {
 		cmd.Args = append(cmd.Args, "-verbose=2")

tstest/integration/integration_test.go

@@ -278,15 +278,20 @@ func TestOneNodeUpAuth(t *testing.T) {
 	t.Logf("Running up --login-server=%s ...", env.ControlURL())
 
 	cmd := n1.Tailscale("up", "--login-server="+env.ControlURL())
-	var authCountAtomic int32
+	var authCountAtomic atomic.Int32
 	cmd.Stdout = &authURLParserWriter{fn: func(urlStr string) error {
+		t.Logf("saw auth URL %q", urlStr)
 		if env.Control.CompleteAuth(urlStr) {
-			atomic.AddInt32(&authCountAtomic, 1)
+			if authCountAtomic.Add(1) > 1 {
+				err := errors.New("completed multple auth URLs")
+				t.Error(err)
+				return err
+			}
 			t.Logf("completed auth path %s", urlStr)
 			return nil
 		}
 		err := fmt.Errorf("Failed to complete auth path to %q", urlStr)
-		t.Log(err)
+		t.Error(err)
 		return err
 	}}
 	cmd.Stderr = cmd.Stdout
@@ -297,7 +302,7 @@ func TestOneNodeUpAuth(t *testing.T) {
 
 	n1.AwaitRunning()
 
-	if n := atomic.LoadInt32(&authCountAtomic); n != 1 {
+	if n := authCountAtomic.Load(); n != 1 {
 		t.Errorf("Auth URLs completed = %d; want 1", n)
 	}
 

tstest/integration/testcontrol/testcontrol.go

@@ -55,6 +55,10 @@ type Server struct {
 	MagicDNSDomain string
 	HandleC2N      http.Handler // if non-nil, used for /some-c2n-path/ in tests
 
+	// AllNodesSameUser, if true, makes all created nodes
+	// belong to the same user.
+	AllNodesSameUser bool
+
 	// ExplicitBaseURL or HTTPTestServer must be set.
 	ExplicitBaseURL string           // e.g. "http://127.0.0.1:1234" with no trailing URL
 	HTTPTestServer  *httptest.Server // if non-nil, used to get BaseURL
@@ -96,9 +100,9 @@ type Server struct {
 	logins        map[key.NodePublic]*tailcfg.Login
 	updates       map[tailcfg.NodeID]chan updateType
 	authPath      map[string]*AuthPath
-	nodeKeyAuthed map[key.NodePublic]bool // key => true once authenticated
-	msgToSend     map[key.NodePublic]any  // value is *tailcfg.PingRequest or entire *tailcfg.MapResponse
-	allExpired    bool                    // All nodes will be told their node key is expired.
+	nodeKeyAuthed set.Set[key.NodePublic]
+	msgToSend     map[key.NodePublic]any // value is *tailcfg.PingRequest or entire *tailcfg.MapResponse
+	allExpired    bool                   // All nodes will be told their node key is expired.
 }
 
 // BaseURL returns the server's base URL, without trailing slash.
@@ -522,6 +526,10 @@ func (s *Server) getUser(nodeKey key.NodePublic) (*tailcfg.User, *tailcfg.Login)
 		return u, s.logins[nodeKey]
 	}
 	id := tailcfg.UserID(len(s.users) + 1)
+	if s.AllNodesSameUser {
+		id = 123
+	}
+	s.logf("Created user %v for node %s", id, nodeKey)
 	loginName := fmt.Sprintf("user-%d@%s", id, domain)
 	displayName := fmt.Sprintf("User %d", id)
 	login := &tailcfg.Login{
@@ -582,10 +590,8 @@ func (s *Server) CompleteAuth(authPathOrURL string) bool {
 	if ap.nodeKey.IsZero() {
 		panic("zero AuthPath.NodeKey")
 	}
-	if s.nodeKeyAuthed == nil {
-		s.nodeKeyAuthed = map[key.NodePublic]bool{}
-	}
-	s.nodeKeyAuthed[ap.nodeKey] = true
+	s.nodeKeyAuthed.Make()
+	s.nodeKeyAuthed.Add(ap.nodeKey)
 	ap.CompleteSuccessfully()
 	return true
 }
@@ -645,36 +651,40 @@ func (s *Server) serveRegister(w http.ResponseWriter, r *http.Request, mkey key.
 	if s.nodes == nil {
 		s.nodes = map[key.NodePublic]*tailcfg.Node{}
 	}
-
+	_, ok := s.nodes[nk]
 	machineAuthorized := true // TODO: add Server.RequireMachineAuth
+	if !ok {
 
-	v4Prefix := netip.PrefixFrom(netaddr.IPv4(100, 64, uint8(tailcfg.NodeID(user.ID)>>8), uint8(tailcfg.NodeID(user.ID))), 32)
-	v6Prefix := netip.PrefixFrom(tsaddr.Tailscale4To6(v4Prefix.Addr()), 128)
+		nodeID := len(s.nodes) + 1
+		v4Prefix := netip.PrefixFrom(netaddr.IPv4(100, 64, uint8(nodeID>>8), uint8(nodeID)), 32)
+		v6Prefix := netip.PrefixFrom(tsaddr.Tailscale4To6(v4Prefix.Addr()), 128)
 
-	allowedIPs := []netip.Prefix{
-		v4Prefix,
-		v6Prefix,
-	}
-
-	s.nodes[nk] = &tailcfg.Node{
-		ID:                tailcfg.NodeID(user.ID),
-		StableID:          tailcfg.StableNodeID(fmt.Sprintf("TESTCTRL%08x", int(user.ID))),
-		User:              user.ID,
-		Machine:           mkey,
-		Key:               req.NodeKey,
-		MachineAuthorized: machineAuthorized,
-		Addresses:         allowedIPs,
-		AllowedIPs:        allowedIPs,
-		Hostinfo:          req.Hostinfo.View(),
-		Name:              req.Hostinfo.Hostname,
-		Capabilities: []tailcfg.NodeCapability{
-			tailcfg.CapabilityHTTPS,
-			tailcfg.NodeAttrFunnel,
-			tailcfg.CapabilityFunnelPorts + "?ports=8080,443",
-		},
+		allowedIPs := []netip.Prefix{
+			v4Prefix,
+			v6Prefix,
+		}
+		node := &tailcfg.Node{
+			ID:                tailcfg.NodeID(nodeID),
+			StableID:          tailcfg.StableNodeID(fmt.Sprintf("TESTCTRL%08x", int(nodeID))),
+			User:              user.ID,
+			Machine:           mkey,
+			Key:               req.NodeKey,
+			MachineAuthorized: machineAuthorized,
+			Addresses:         allowedIPs,
+			AllowedIPs:        allowedIPs,
+			Hostinfo:          req.Hostinfo.View(),
+			Name:              req.Hostinfo.Hostname,
+			Capabilities: []tailcfg.NodeCapability{
+				tailcfg.CapabilityHTTPS,
+				tailcfg.NodeAttrFunnel,
+				tailcfg.CapabilityFileSharing,
+				tailcfg.CapabilityFunnelPorts + "?ports=8080,443",
+			},
+		}
+		s.nodes[nk] = node
 	}
 	requireAuth := s.RequireAuth
-	if requireAuth && s.nodeKeyAuthed[nk] {
+	if requireAuth && s.nodeKeyAuthed.Contains(nk) {
 		requireAuth = false
 	}
 	allExpired := s.allExpired
@@ -951,7 +961,6 @@ func (s *Server) MapResponse(req *tailcfg.MapRequest) (res *tailcfg.MapResponse,
 	node.CapMap = nodeCapMap
 	node.Capabilities = append(node.Capabilities, tailcfg.NodeAttrDisableUPnP)
 
-	user, _ := s.getUser(nk)
 	t := time.Date(2020, 8, 3, 0, 0, 0, 1, time.UTC)
 	dns := s.DNSConfig
 	if dns != nil && s.MagicDNSDomain != "" {
@@ -1013,7 +1022,7 @@ func (s *Server) MapResponse(req *tailcfg.MapRequest) (res *tailcfg.MapResponse,
 	})
 	res.UserProfiles = s.allUserProfiles()
 
-	v4Prefix := netip.PrefixFrom(netaddr.IPv4(100, 64, uint8(tailcfg.NodeID(user.ID)>>8), uint8(tailcfg.NodeID(user.ID))), 32)
+	v4Prefix := netip.PrefixFrom(netaddr.IPv4(100, 64, uint8(node.ID>>8), uint8(node.ID)), 32)
 	v6Prefix := netip.PrefixFrom(tsaddr.Tailscale4To6(v4Prefix.Addr()), 128)
 
 	res.Node.Addresses = []netip.Prefix{