WIP tailpipe

Updates #nnn

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

client/tailscale/localclient.go

@@ -19,6 +19,7 @@ import (
 	"net/http"
 	"net/http/httptrace"
 	"net/netip"
+	"net/textproto"
 	"net/url"
 	"os/exec"
 	"runtime"
@@ -558,6 +559,15 @@ func (lc *LocalClient) SetDNS(ctx context.Context, name, value string) error {
 //
 // The ctx is only used for the duration of the call, not the lifetime of the net.Conn.
 func (lc *LocalClient) DialTCP(ctx context.Context, host string, port uint16) (net.Conn, error) {
+	return lc.dialViaLocalAPI(ctx, host, fmt.Sprint(port))
+}
+
+// DialTCPNamedPort is like DialTCP but takes a named port rather than an integer.
+func (lc *LocalClient) DialTCPNamedPort(ctx context.Context, host, portName string) (net.Conn, error) {
+	return lc.dialViaLocalAPI(ctx, host, portName)
+}
+
+func (lc *LocalClient) dialViaLocalAPI(ctx context.Context, host, port string) (net.Conn, error) {
 	connCh := make(chan net.Conn, 1)
 	trace := httptrace.ClientTrace{
 		GotConn: func(info httptrace.GotConnInfo) {
@@ -573,7 +583,7 @@ func (lc *LocalClient) DialTCP(ctx context.Context, host string, port uint16) (n
 		"Upgrade":    []string{"ts-dial"},
 		"Connection": []string{"upgrade"},
 		"Dial-Host":  []string{host},
-		"Dial-Port":  []string{fmt.Sprint(port)},
+		"Dial-Port":  []string{port},
 	}
 	res, err := lc.DoLocalRequest(req)
 	if err != nil {
@@ -605,6 +615,59 @@ func (lc *LocalClient) DialTCP(ctx context.Context, host string, port uint16) (n
 	return netutil.NewAltReadWriteCloserConn(rwc, switchedConn), nil
 }
 
+// ListenNewRandomPortName...
+func (lc *LocalClient) ListenNewRandomPortName(ctx context.Context) (portName string, accept func(context.Context) (net.Conn, error), err error) {
+	connCh := make(chan net.Conn, 1)
+	portNameCh := make(chan string, 1)
+	trace := httptrace.ClientTrace{
+		GotConn: func(info httptrace.GotConnInfo) {
+			connCh <- info.Conn
+		},
+		Got1xxResponse: func(code int, header textproto.MIMEHeader) error {
+			portNameCh <- fmt.Sprintf("Got %v, %v", code, header)
+			return nil
+		},
+	}
+	ctx = httptrace.WithClientTrace(ctx, &trace)
+	req, err := http.NewRequestWithContext(ctx, "POST", "http://local-tailscaled.sock/localapi/v0/open-bidi-pipe", nil)
+	if err != nil {
+		return "", nil, err
+	}
+	req.Header = http.Header{
+		"Upgrade":    []string{"ts-open-bidi-pipe"},
+		"Connection": []string{"upgrade"},
+	}
+
+	doErrc := make(chan error, 1)
+
+	go func() {
+		res, err := lc.DoLocalRequest(req)
+		if err != nil {
+			doErrc <- err
+			return
+		}
+		_ = res
+	}()
+
+	accept = func(ctx context.Context) (net.Conn, error) {
+		panic("TODO")
+	}
+
+	select {
+	case name := <-portNameCh:
+		return name, accept, nil
+	case err := <-doErrc:
+		return "", nil, err
+	}
+
+	// if res.StatusCode != http.StatusSwitchingProtocols {
+	// 	body, _ := io.ReadAll(res.Body)
+	// 	res.Body.Close()
+	// 	return nil, fmt.Errorf("unexpected HTTP response: %s, %s", res.Status, body)
+	// }
+	panic("TODO")
+}
+
 // CurrentDERPMap returns the current DERPMap that is being used by the local tailscaled.
 // It is intended to be used with netcheck to see availability of DERPs.
 func (lc *LocalClient) CurrentDERPMap(ctx context.Context) (*tailcfg.DERPMap, error) {

cmd/tailscale/cli/cli.go

@@ -34,6 +34,7 @@ import (
 
 var Stderr io.Writer = os.Stderr
 var Stdout io.Writer = os.Stdout
+var Stdin io.Reader = os.Stdin
 
 func printf(format string, a ...any) {
 	fmt.Fprintf(Stdout, format, a...)

cmd/tailscale/cli/nc.go

@@ -7,46 +7,65 @@ package cli
 import (
 	"context"
 	"errors"
+	"flag"
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"strconv"
+	"strings"
 
 	"github.com/peterbourgon/ff/v3/ffcli"
+	"tailscale.com/ipn/ipnstate"
 )
 
 var ncCmd = &ffcli.Command{
 	Name:       "nc",
-	ShortUsage: "nc <hostname-or-IP> <port>",
+	ShortUsage: "nc <hostname-or-IP> <port>\n  nc -l",
 	ShortHelp:  "Connect to a port on a host, connected to stdin/stdout",
 	Exec:       runNC,
+	FlagSet: func() *flag.FlagSet {
+		fs := flag.NewFlagSet("nc", flag.ExitOnError)
+		fs.BoolVar(&ncArgs.listen, "l", false, "whether to listen for incoming connections (\"Tailpipe\")")
+		return fs
+	}(),
+}
+
+var ncArgs struct {
+	listen bool
 }
 
 func runNC(ctx context.Context, args []string) error {
-	st, err := localClient.Status(ctx)
-	if err != nil {
-		return fixTailscaledConnectError(err)
-	}
-	description, ok := isRunningOrStarting(st)
-	if !ok {
-		printf("%s\n", description)
-		os.Exit(1)
+	if ncArgs.listen {
+		if len(args) != 0 {
+			return errors.New("no arguments supported with -l")
+		}
+		return runNCListen(ctx)
 	}
 
 	if len(args) != 2 {
 		return errors.New("usage: nc <hostname-or-IP> <port>")
 	}
 
-	hostOrIP, portStr := args[0], args[1]
-	port, err := strconv.ParseUint(portStr, 10, 16)
-	if err != nil {
-		return fmt.Errorf("invalid port number %q", portStr)
+	if _, err := checkRunning(ctx); err != nil {
+		return err
 	}
+	hostOrIP, portStr := args[0], args[1]
 
-	// TODO(bradfitz): also add UDP too, via flag?
-	c, err := localClient.DialTCP(ctx, hostOrIP, uint16(port))
+	var c net.Conn
+	var err error
+	if strings.HasPrefix(portStr, "tailpipe-") {
+		c, err = localClient.DialTCPNamedPort(ctx, hostOrIP, portStr)
+	} else {
+		port, err := strconv.ParseUint(portStr, 10, 16)
+		if err != nil {
+			return fmt.Errorf("invalid port number %q", portStr)
+		}
+		// TODO(bradfitz): also add UDP too, via flag?
+		c, err = localClient.DialTCP(ctx, hostOrIP, uint16(port))
+	}
 	if err != nil {
-		return fmt.Errorf("Dial(%q, %v): %w", hostOrIP, port, err)
+		return fmt.Errorf("Dial(%q, %v): %w", hostOrIP, portStr, err)
 	}
 	defer c.Close()
 	errc := make(chan error, 1)
@@ -60,3 +79,43 @@ func runNC(ctx context.Context, args []string) error {
 	}()
 	return <-errc
 }
+
+func checkRunning(ctx context.Context) (*ipnstate.Status, error) {
+	st, err := localClient.Status(ctx)
+	if err != nil {
+		return nil, fixTailscaledConnectError(err)
+	}
+	description, ok := isRunningOrStarting(st)
+	if !ok {
+		printf("%s\n", description)
+		os.Exit(1)
+	}
+	return st, err
+}
+
+// runNCLIsten opens a tailpipe.
+func runNCListen(ctx context.Context) error {
+	st, err := checkRunning(ctx)
+	if err != nil {
+		return err
+	}
+	portName, accept, err := localClient.ListenNewRandomPortName(ctx)
+	if err != nil {
+		return err
+	}
+	fmt.Fprintf(Stderr, "Port opened. Connect with: nc %v %v\n", st.Self.Addrs[0], portName)
+	c, err := accept(ctx)
+	if err != nil {
+		return err
+	}
+	errc := make(chan error, 1)
+	go func() {
+		_, err := io.Copy(Stdout, c)
+		errc <- err
+	}()
+	go func() {
+		_, err := io.Copy(c, Stdin)
+		errc <- err
+	}()
+	return <-errc
+}

ipn/ipnlocal/tailpipe.go

@@ -0,0 +1,104 @@
+// Copyright (c) 2022 Tailscale Inc & AUTHORS All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package ipnlocal
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptrace"
+	"net/netip"
+
+	"golang.org/x/exp/slices"
+	"tailscale.com/ipn"
+	"tailscale.com/net/netutil"
+	"tailscale.com/net/tsaddr"
+	"tailscale.com/tailcfg"
+)
+
+// PipeDialPeerAPIURL ....
+func (b *LocalBackend) PipeDialPeerAPIURL(ip netip.Addr) (peerAPIURL string, err error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	nm := b.netMap
+	if b.state != ipn.Running || nm == nil {
+		return "", errors.New("not connected to the tailnet")
+	}
+	ipa := netip.PrefixFrom(ip, ip.BitLen())
+	for _, p := range nm.Peers {
+		if slices.Contains(p.Addresses, ipa) {
+			if p.User == nm.User ||
+				len(p.Addresses) > 0 && b.peerHasCapLocked(p.Addresses[0].Addr(), tailcfg.CapabilityTailpipeTarget) {
+				peerAPI := peerAPIBase(b.netMap, p)
+				if peerAPI == "" {
+					continue
+				}
+			}
+			return "", errors.New("invalid target")
+		}
+	}
+	return "", errors.New("target not found")
+}
+
+func (b *LocalBackend) DialTailpipe(ctx context.Context, tailscaleIPStr, portName string) (net.Conn, error) {
+	ip, err := netip.ParseAddr(tailscaleIPStr)
+	if err != nil || !tsaddr.IsTailscaleIP(ip) {
+		return nil, fmt.Errorf("host must be a Tailscale IP for now, not %q", tailscaleIPStr)
+	}
+
+	hc := b.dialer.PeerAPIHTTPClient()
+
+	peerAPIBase, err := b.PipeDialPeerAPIURL(ip)
+	if err != nil {
+		return nil, err
+	}
+	connCh := make(chan net.Conn, 1)
+	trace := httptrace.ClientTrace{
+		GotConn: func(info httptrace.GotConnInfo) {
+			connCh <- info.Conn
+		},
+	}
+	ctx = httptrace.WithClientTrace(ctx, &trace)
+	req, err := http.NewRequestWithContext(ctx, "POST", peerAPIBase+"/localapi/v0/connect-to-open-tailpipe", nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header = http.Header{
+		"Upgrade":    []string{"tailpipe"},
+		"Connection": []string{"upgrade"},
+		"Port-Name":  []string{portName},
+	}
+	res, err := hc.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if res.StatusCode != http.StatusSwitchingProtocols {
+		body, _ := io.ReadAll(res.Body)
+		res.Body.Close()
+		return nil, fmt.Errorf("unexpected HTTP response: %s, %s", res.Status, body)
+	}
+	// From here on, the underlying net.Conn is ours to use, but there
+	// is still a read buffer attached to it within resp.Body. So, we
+	// must direct I/O through resp.Body, but we can still use the
+	// underlying net.Conn for stuff like deadlines.
+	var switchedConn net.Conn
+	select {
+	case switchedConn = <-connCh:
+	default:
+	}
+	if switchedConn == nil {
+		res.Body.Close()
+		return nil, fmt.Errorf("httptrace didn't provide a connection")
+	}
+	rwc, ok := res.Body.(io.ReadWriteCloser)
+	if !ok {
+		res.Body.Close()
+		return nil, errors.New("http Transport did not provide a writable body")
+	}
+	return netutil.NewAltReadWriteCloserConn(rwc, switchedConn), nil
+}

ipn/localapi/localapi.go

@@ -67,6 +67,7 @@ var handler = map[string]localAPIHandler{
 	"login-interactive":       (*Handler).serveLoginInteractive,
 	"logout":                  (*Handler).serveLogout,
 	"metrics":                 (*Handler).serveMetrics,
+	"open-bidi-pipe":          (*Handler).serveOpenBidiPipe,
 	"ping":                    (*Handler).servePing,
 	"prefs":                   (*Handler).servePrefs,
 	"profile":                 (*Handler).serveProfile,
@@ -764,8 +765,16 @@ func (h *Handler) serveDial(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	addr := net.JoinHostPort(hostStr, portStr)
-	outConn, err := h.b.Dialer().UserDial(r.Context(), "tcp", addr)
+	dial := func() (net.Conn, error) {
+		if strings.HasPrefix(portStr, "tailpipe-") {
+			// hostStr is expected to be a Tailscale IP at this point.
+			return h.b.DialTailpipe(r.Context(), hostStr, portStr)
+		}
+		addr := net.JoinHostPort(hostStr, portStr)
+		return h.b.Dialer().UserDial(r.Context(), "tcp", addr)
+	}
+
+	outConn, err := dial()
 	if err != nil {
 		http.Error(w, "dial failure: "+err.Error(), http.StatusBadGateway)
 		return
@@ -933,6 +942,29 @@ func (h *Handler) serveTKAModify(w http.ResponseWriter, r *http.Request) {
 	w.Write(j)
 }
 
+func (h *Handler) serveOpenBidiPipe(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "use POST", http.StatusMethodNotAllowed)
+		return
+	}
+	name := r.FormValue("name")
+	if name != "" && !h.PermitWrite {
+		http.Error(w, "naming a bidi pipe requires write access", http.StatusForbidden)
+		return
+	}
+
+	w.Header().Set("Foo", "bar")
+	w.WriteHeader(http.StatusProcessing) // informational code 102
+
+	time.Sleep(time.Second)
+
+	w.Header().Set("Foo", "baz")
+	w.WriteHeader(http.StatusProcessing) // informational code 102
+
+	w.Header()["Foo"] = nil
+	io.WriteString(w, "the body")
+}
+
 func defBool(a string, def bool) bool {
 	if a == "" {
 		return def

tailcfg/tailcfg.go

@@ -1647,6 +1647,8 @@ const (
 	CapabilityDebugPeer = "https://tailscale.com/cap/debug-peer"
 	// CapabilityWakeOnLAN grants the ability to send a Wake-On-LAN packet.
 	CapabilityWakeOnLAN = "https://tailscale.com/cap/wake-on-lan"
+
+	CapabilityTailpipeTarget = "https://tailscale.com/cap/tailpipe-target"
 )
 
 // SetDNSRequest is a request to add a DNS record.