all: implement pcap streaming for datapath debugging

Updates: tailscale/corp#8470

Signed-off-by: Tom DNetto <tom@tailscale.com>

wgengine/capture/capture.go

@@ -0,0 +1,201 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Package capture formats packet logging into a debug pcap stream.
+package capture
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"io"
+	"net/http"
+	"sync"
+	"time"
+
+	_ "embed"
+
+	"tailscale.com/util/set"
+)
+
+//go:embed ts-dissector.lua
+var DissectorLua string
+
+var bufferPool = sync.Pool{
+	New: func() any {
+		return new(bytes.Buffer)
+	},
+}
+
+const flushPeriod = 100 * time.Millisecond
+
+func writePcapHeader(w io.Writer) {
+	binary.Write(w, binary.LittleEndian, uint32(0xA1B2C3D4)) // pcap magic number
+	binary.Write(w, binary.LittleEndian, uint16(2))          // version major
+	binary.Write(w, binary.LittleEndian, uint16(4))          // version minor
+	binary.Write(w, binary.LittleEndian, uint32(0))          // this zone
+	binary.Write(w, binary.LittleEndian, uint32(0))          // zone significant figures
+	binary.Write(w, binary.LittleEndian, uint32(65535))      // max packet len
+	binary.Write(w, binary.LittleEndian, uint32(147))        // link-layer ID - USER0
+}
+
+func writePktHeader(w *bytes.Buffer, when time.Time, length int) {
+	s := when.Unix()
+	us := when.UnixMicro() - (s * 1000000)
+
+	binary.Write(w, binary.LittleEndian, uint32(s))      // timestamp in seconds
+	binary.Write(w, binary.LittleEndian, uint32(us))     // timestamp microseconds
+	binary.Write(w, binary.LittleEndian, uint32(length)) // length present
+	binary.Write(w, binary.LittleEndian, uint32(length)) // total length
+}
+
+// Path describes where in the data path the packet was captured.
+type Path uint8
+
+// Valid Path values.
+const (
+	// FromLocal indicates the packet was logged as it traversed the FromLocal path:
+	// i.e.: A packet from the local system into the TUN.
+	FromLocal Path = 0
+	// FromPeer indicates the packet was logged upon reception from a remote peer.
+	FromPeer Path = 1
+	// SynthesizedToLocal indicates the packet was generated from within tailscaled,
+	// and is being routed to the local machine's network stack.
+	SynthesizedToLocal Path = 2
+	// SynthesizedToPeer indicates the packet was generated from within tailscaled,
+	// and is being routed to a remote Wireguard peer.
+	SynthesizedToPeer Path = 3
+)
+
+// New creates a new capture sink.
+func New() *Sink {
+	ctx, c := context.WithCancel(context.Background())
+	return &Sink{
+		ctx:       ctx,
+		ctxCancel: c,
+	}
+}
+
+// Type Sink handles callbacks with packets to be logged,
+// formatting them into a pcap stream which is mirrored to
+// all registered outputs.
+type Sink struct {
+	ctx       context.Context
+	ctxCancel context.CancelFunc
+
+	mu         sync.Mutex
+	outputs    set.HandleSet[io.Writer]
+	flushTimer *time.Timer // or nil if none running
+}
+
+// RegisterOutput connects an output to this sink, which
+// will be written to with a pcap stream as packets are logged.
+// A function is returned which unregisters the output when
+// called.
+//
+// If w implements io.Closer, it will be closed upon error
+// or when the sink is closed. If w implements http.Flusher,
+// it will be flushed periodically.
+func (s *Sink) RegisterOutput(w io.Writer) (unregister func()) {
+	select {
+	case <-s.ctx.Done():
+		return func() {}
+	default:
+	}
+
+	writePcapHeader(w)
+	s.mu.Lock()
+	hnd := s.outputs.Add(w)
+	s.mu.Unlock()
+
+	return func() {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		delete(s.outputs, hnd)
+	}
+}
+
+// NumOutputs returns the number of outputs registered with the sink.
+func (s *Sink) NumOutputs() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.outputs)
+}
+
+// Close shuts down the sink. Future calls to LogPacket
+// are ignored, and any registered output that implements
+// io.Closer is closed.
+func (s *Sink) Close() error {
+	s.ctxCancel()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.flushTimer != nil {
+		s.flushTimer.Stop()
+		s.flushTimer = nil
+	}
+
+	for _, o := range s.outputs {
+		if o, ok := o.(io.Closer); ok {
+			o.Close()
+		}
+	}
+	s.outputs = nil
+	return nil
+}
+
+// WaitCh returns a channel which blocks untill
+// the sink is closed.
+func (s *Sink) WaitCh() <-chan struct{} {
+	return s.ctx.Done()
+}
+
+// LogPacket is called to insert a packet into the capture.
+//
+// This function does not take ownership of the provided data slice.
+func (s *Sink) LogPacket(path Path, when time.Time, data []byte) {
+	select {
+	case <-s.ctx.Done():
+		return
+	default:
+	}
+
+	b := bufferPool.Get().(*bytes.Buffer)
+	b.Reset()
+	b.Grow(16 + 2 + len(data)) // 16b pcap header + 2b custom data + len
+	defer bufferPool.Put(b)
+
+	writePktHeader(b, when, len(data)+2)
+	// Custom tailscale debugging data
+	binary.Write(b, binary.LittleEndian, uint16(path))
+	b.Write(data)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var hadError []set.Handle
+	for hnd, o := range s.outputs {
+		if _, err := o.Write(b.Bytes()); err != nil {
+			hadError = append(hadError, hnd)
+			continue
+		}
+	}
+	for _, hnd := range hadError {
+		if o, ok := s.outputs[hnd].(io.Closer); ok {
+			o.Close()
+		}
+		delete(s.outputs, hnd)
+	}
+
+	if s.flushTimer == nil {
+		s.flushTimer = time.AfterFunc(flushPeriod, func() {
+			s.mu.Lock()
+			defer s.mu.Unlock()
+			for _, o := range s.outputs {
+				if f, ok := o.(http.Flusher); ok {
+					f.Flush()
+				}
+			}
+			s.flushTimer = nil
+		})
+	}
+}

wgengine/capture/ts-dissector.lua

@@ -0,0 +1,27 @@
+tsdebug_ll = Proto("tsdebug", "Tailscale debug")
+PATH = ProtoField.string("tsdebug.PATH","PATH", base.ASCII)
+tsdebug_ll.fields = {PATH}
+
+function tsdebug_ll.dissector(buffer, pinfo, tree)
+    pinfo.cols.protocol = tsdebug_ll.name
+    packet_length = buffer:len()
+    local offset = 0
+    local subtree = tree:add(tsdebug_ll, buffer(), "Tailscale packet")
+
+    -- -- Get path UINT16
+    local path_id = buffer:range(offset, 2):le_uint()
+    if     path_id == 0 then subtree:add(PATH, "FromLocal")
+    elseif path_id == 1 then subtree:add(PATH, "FromPeer")
+    elseif path_id == 2 then subtree:add(PATH, "Synthesized (Inbound / ToLocal)")
+    elseif path_id == 3 then subtree:add(PATH, "Synthesized (Outbound / ToPeer)")
+    end
+    offset = offset + 2
+
+    -- -- Handover rest of data to ip dissector
+    local data_buffer = buffer:range(offset, packet_length-offset):tvb()
+    Dissector.get("ip"):call(data_buffer, pinfo, tree)
+end
+
+-- Install the dissector on link-layer ID 147 (User-defined protocol 0)
+local eth_table = DissectorTable.get("wtap_encap")
+eth_table:add(wtap.USER0, tsdebug_ll)

wgengine/userspace.go

@@ -1579,3 +1579,7 @@ var (
 	metricNumMajorChanges = clientmetric.NewCounter("wgengine_major_changes")
 	metricNumMinorChanges = clientmetric.NewCounter("wgengine_minor_changes")
 )
+
+func (e *userspaceEngine) InstallCaptureHook(cb CaptureCallback) {
+	e.tundev.InstallCaptureHook(tstun.CaptureFunc(cb))
+}

wgengine/watchdog.go

@@ -200,3 +200,7 @@ func (e *watchdogEngine) PeerForIP(ip netip.Addr) (ret PeerForIP, ok bool) {
 func (e *watchdogEngine) Wait() {
 	e.wrap.Wait()
 }
+
+func (e *watchdogEngine) InstallCaptureHook(cb CaptureCallback) {
+	e.wrap.InstallCaptureHook(cb)
+}

wgengine/wgengine.go

@@ -13,6 +13,7 @@ import (
 	"tailscale.com/tailcfg"
 	"tailscale.com/types/key"
 	"tailscale.com/types/netmap"
+	"tailscale.com/wgengine/capture"
 	"tailscale.com/wgengine/filter"
 	"tailscale.com/wgengine/monitor"
 	"tailscale.com/wgengine/router"
@@ -42,6 +43,12 @@ type NetInfoCallback func(*tailcfg.NetInfo)
 // into network map updates.
 type NetworkMapCallback func(*netmap.NetworkMap)
 
+// CaptureCallback is the type used to record packets when
+// debugging packet-capture. This function must not take
+// ownership of the provided data slice: it may only copy
+// out of it within the lifetime of the function.
+type CaptureCallback func(capture.Path, time.Time, []byte)
+
 // someHandle is allocated so its pointer address acts as a unique
 // map key handle. (It needs to have non-zero size for Go to guarantee
 // the pointer is unique.)
@@ -171,4 +178,9 @@ type Engine interface {
 	// WhoIsIPPort looks up an IP:port in the temporary registrations,
 	// and returns a matching Tailscale IP, if it exists.
 	WhoIsIPPort(netip.AddrPort) (netip.Addr, bool)
+
+	// InstallCaptureHook registers a function to be called to capture
+	// packets traversing the data path. The hook can be uninstalled by
+	// calling this function with a nil value.
+	InstallCaptureHook(CaptureCallback)
 }