Share kernel Wireguard port with raw socket (#826)

This PR brings support of a shared port between stun (ICE agent) and
the kernel WireGuard

It implements a single port mode for execution with kernel WireGuard
interface using a raw socket listener.

BPF filters ensure that only STUN packets hit the NetBird userspace app

Removed a lot of the proxy logic and direct mode exchange.

Now we are doing an extra hole punch to the remote WireGuard 
port for best-effort cases and support to old client's direct mode.

sharedsock/filter.go

@@ -0,0 +1,11 @@
+package sharedsock
+
+import "golang.org/x/net/bpf"
+
+const magicCookie uint32 = 0x2112A442
+
+// BPFFilter is a generic filter that provides ipv4 and ipv6 BPF instructions
+type BPFFilter interface {
+	// GetInstructions returns raw BPF instructions for ipv4 and ipv6
+	GetInstructions(port uint32) (ipv4 []bpf.RawInstruction, ipv6 []bpf.RawInstruction, err error)
+}

sharedsock/filter_linux.go

@@ -0,0 +1,47 @@
+package sharedsock
+
+import "golang.org/x/net/bpf"
+
+// STUNFilter implements BPFFilter by filtering on STUN packets.
+// Other packets (non STUN) will be forwarded to the process that own the port (e.g., WireGuard).
+type STUNFilter struct {
+}
+
+// NewSTUNFilter creates an instance of a STUNFilter
+func NewSTUNFilter() BPFFilter {
+	return &STUNFilter{}
+}
+
+// GetInstructions returns raw BPF instructions for ipv4 and ipv6 that filter out anything but STUN packets
+func (sf STUNFilter) GetInstructions(port uint32) (raw4 []bpf.RawInstruction, raw6 []bpf.RawInstruction, err error) {
+	raw4, err = rawInstructions(22, 32, port)
+	if err != nil {
+		return nil, nil, err
+	}
+	raw6, err = rawInstructions(2, 12, port)
+	if err != nil {
+		return nil, nil, err
+	}
+	return raw4, raw6, nil
+}
+
+func rawInstructions(portOff, cookieOff, port uint32) ([]bpf.RawInstruction, error) {
+	// UDP raw socket for ipv4 receives the rcvdPacket with IP headers
+	// UDP raw socket for ipv6 receives the rcvdPacket with UDP headers
+	instructions := []bpf.Instruction{
+		// Load the source port from the UDP header (offset 22 for ipv4 and 2 for ipv6)
+		bpf.LoadAbsolute{Off: portOff, Size: 2},
+		// Check if the source port is equal to the specified `port`. If not, skip the next 3 instructions.
+		bpf.JumpIf{Cond: bpf.JumpNotEqual, Val: port, SkipTrue: 3},
+		// Load the 4-byte value (magic cookie) from the UDP payload (offset 32 for ipv4 and 12 for ipv6)
+		bpf.LoadAbsolute{Off: cookieOff, Size: 4},
+		// Check if the loaded value is equal to the `magicCookie`. If not, skip the next instruction.
+		bpf.JumpIf{Cond: bpf.JumpNotEqual, Val: magicCookie, SkipTrue: 1},
+		// If both the port and the magic cookie match, return a positive value (0xffffffff)
+		bpf.RetConstant{Val: 0xffffffff},
+		// If either the port or the magic cookie doesn't match, return 0
+		bpf.RetConstant{Val: 0},
+	}
+
+	return bpf.Assemble(instructions)
+}

sharedsock/filter_nolinux.go

@@ -0,0 +1,8 @@
+//go:build !linux
+
+package sharedsock
+
+// NewSTUNFilter is a noop method just because we do not support BPF filters on other platforms than Linux
+func NewSTUNFilter() BPFFilter {
+	return nil
+}

sharedsock/sock_linux.go

@@ -0,0 +1,327 @@
+//go:build linux && !android
+
+// Inspired by
+// Jason Donenfeld (https://git.zx2c4.com/wireguard-tools/tree/contrib/nat-hole-punching/nat-punch-client.c#n96)
+// and @stv0g in https://github.com/stv0g/cunicu/tree/ebpf-poc/ebpf_poc
+
+package sharedsock
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/routing"
+	"github.com/libp2p/go-netroute"
+	"github.com/mdlayher/socket"
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/sys/unix"
+)
+
+// ErrSharedSockStopped indicates that shared socket has been stopped
+var ErrSharedSockStopped = fmt.Errorf("shared socked stopped")
+
+// SharedSocket is a net.PacketConn that initiates two raw sockets (ipv4 and ipv6) and listens to UDP packets filtered
+// by BPF instructions (e.g., STUNFilter that checks and sends only STUN packets to the listeners (ReadFrom)).
+type SharedSocket struct {
+	ctx         context.Context
+	conn4       *socket.Conn
+	conn6       *socket.Conn
+	port        int
+	routerMux   sync.RWMutex
+	router      routing.Router
+	packetDemux chan rcvdPacket
+	cancel      context.CancelFunc
+}
+
+type rcvdPacket struct {
+	n    int
+	addr unix.Sockaddr
+	buf  []byte
+	err  error
+}
+
+type receiver func(ctx context.Context, p []byte, flags int) (int, unix.Sockaddr, error)
+
+var writeSerializerOptions = gopacket.SerializeOptions{
+	ComputeChecksums: true,
+	FixLengths:       true,
+}
+
+// Listen creates an IPv4 and IPv6 raw sockets, starts a reader and routing table routines
+func Listen(port int, filter BPFFilter) (net.PacketConn, error) {
+	var err error
+	ctx, cancel := context.WithCancel(context.Background())
+	rawSock := &SharedSocket{
+		ctx:         ctx,
+		cancel:      cancel,
+		port:        port,
+		packetDemux: make(chan rcvdPacket),
+	}
+
+	rawSock.router, err = netroute.New()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create router: %rawSock", err)
+	}
+
+	rawSock.conn4, err = socket.Socket(unix.AF_INET, unix.SOCK_RAW, unix.IPPROTO_UDP, "raw_udp4", nil)
+	if err != nil {
+		return nil, fmt.Errorf("socket.Socket for ipv4 failed with: %rawSock", err)
+	}
+
+	rawSock.conn6, err = socket.Socket(unix.AF_INET6, unix.SOCK_RAW, unix.IPPROTO_UDP, "raw_udp6", nil)
+	if err != nil {
+		log.Errorf("socket.Socket for ipv6 failed with: %rawSock", err)
+	}
+
+	ipv4Instructions, ipv6Instructions, err := filter.GetInstructions(uint32(rawSock.port))
+	if err != nil {
+		_ = rawSock.Close()
+		return nil, fmt.Errorf("getBPFInstructions failed with: %rawSock", err)
+	}
+
+	err = rawSock.conn4.SetBPF(ipv4Instructions)
+	if err != nil {
+		_ = rawSock.Close()
+		return nil, fmt.Errorf("socket4.SetBPF failed with: %rawSock", err)
+	}
+	if rawSock.conn6 != nil {
+		err = rawSock.conn6.SetBPF(ipv6Instructions)
+		if err != nil {
+			_ = rawSock.Close()
+			return nil, fmt.Errorf("socket6.SetBPF failed with: %rawSock", err)
+		}
+	}
+
+	go rawSock.read(rawSock.conn4.Recvfrom)
+	if rawSock.conn6 != nil {
+		go rawSock.read(rawSock.conn6.Recvfrom)
+	}
+
+	go rawSock.updateRouter()
+
+	return rawSock, nil
+}
+
+// updateRouter updates the listener routing table client
+// this is needed to avoid outdated information across different client networks
+func (s *SharedSocket) updateRouter() {
+	ticker := time.NewTicker(15 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-ticker.C:
+			router, err := netroute.New()
+			if err != nil {
+				log.Errorf("failed to create and update packet router for stunListener: %s", err)
+				continue
+			}
+			s.routerMux.Lock()
+			s.router = router
+			s.routerMux.Unlock()
+		}
+	}
+}
+
+// LocalAddr returns an IPv4 address using the supplied port
+func (s *SharedSocket) LocalAddr() net.Addr {
+	// todo check impact on ipv6 discovery
+	return &net.UDPAddr{
+		IP:   net.IPv4zero,
+		Port: s.port,
+	}
+}
+
+// SetDeadline sets both the read and write deadlines associated with the ipv4 and ipv6 Conn sockets
+func (s *SharedSocket) SetDeadline(t time.Time) error {
+	err := s.conn4.SetDeadline(t)
+	if err != nil {
+		return fmt.Errorf("s.conn4.SetDeadline error: %s", err)
+	}
+	if s.conn6 == nil {
+		return nil
+	}
+
+	err = s.conn6.SetDeadline(t)
+	if err != nil {
+		return fmt.Errorf("s.conn6.SetDeadline error: %s", err)
+	}
+	return nil
+}
+
+// SetReadDeadline sets the read deadline associated with the ipv4 and ipv6 Conn sockets
+func (s *SharedSocket) SetReadDeadline(t time.Time) error {
+	err := s.conn4.SetReadDeadline(t)
+	if err != nil {
+		return fmt.Errorf("s.conn4.SetReadDeadline error: %s", err)
+	}
+	if s.conn6 == nil {
+		return nil
+	}
+
+	err = s.conn6.SetReadDeadline(t)
+	if err != nil {
+		return fmt.Errorf("s.conn6.SetReadDeadline error: %s", err)
+	}
+	return nil
+}
+
+// SetWriteDeadline sets the write deadline associated with the ipv4 and ipv6 Conn sockets
+func (s *SharedSocket) SetWriteDeadline(t time.Time) error {
+	err := s.conn4.SetWriteDeadline(t)
+	if err != nil {
+		return fmt.Errorf("s.conn4.SetWriteDeadline error: %s", err)
+	}
+	if s.conn6 == nil {
+		return nil
+	}
+
+	err = s.conn6.SetWriteDeadline(t)
+	if err != nil {
+		return fmt.Errorf("s.conn6.SetWriteDeadline error: %s", err)
+	}
+	return nil
+}
+
+// Close closes the underlying ipv4 and ipv6 conn sockets
+func (s *SharedSocket) Close() error {
+	s.cancel()
+	errGrp := errgroup.Group{}
+	if s.conn4 != nil {
+		errGrp.Go(s.conn4.Close)
+	}
+
+	if s.conn6 != nil {
+		errGrp.Go(s.conn6.Close)
+	}
+	return errGrp.Wait()
+}
+
+// read start a read loop for a specific receiver and sends the packet to the packetDemux channel
+func (s *SharedSocket) read(receiver receiver) {
+	for {
+		buf := make([]byte, 1500)
+		n, addr, err := receiver(s.ctx, buf, 0)
+		select {
+		case <-s.ctx.Done():
+			return
+		case s.packetDemux <- rcvdPacket{n, addr, buf[:n], err}:
+		}
+	}
+}
+
+// ReadFrom reads packets received in the packetDemux channel
+func (s *SharedSocket) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
+	var pkt rcvdPacket
+	select {
+	case <-s.ctx.Done():
+		return -1, nil, ErrSharedSockStopped
+	case pkt = <-s.packetDemux:
+	}
+
+	if pkt.err != nil {
+		return -1, nil, pkt.err
+	}
+	var ip4layer layers.IPv4
+	var udp layers.UDP
+	var payload gopacket.Payload
+	var parser *gopacket.DecodingLayerParser
+	var ip net.IP
+
+	if sa, isIPv4 := pkt.addr.(*unix.SockaddrInet4); isIPv4 {
+		ip = sa.Addr[:]
+		parser = gopacket.NewDecodingLayerParser(layers.LayerTypeIPv4, &ip4layer, &udp, &payload)
+	} else if sa, isIPv6 := pkt.addr.(*unix.SockaddrInet6); isIPv6 {
+		ip = sa.Addr[:]
+		parser = gopacket.NewDecodingLayerParser(layers.LayerTypeUDP, &udp, &payload)
+	} else {
+		return -1, nil, fmt.Errorf("received invalid address family")
+	}
+
+	decodedLayers := make([]gopacket.LayerType, 0, 3)
+
+	err = parser.DecodeLayers(pkt.buf[:], &decodedLayers)
+	if err != nil {
+		return 0, nil, err
+	}
+
+	remoteAddr := &net.UDPAddr{
+		IP:   ip,
+		Port: int(udp.SrcPort),
+	}
+
+	copy(b, payload)
+	return int(udp.Length), remoteAddr, nil
+}
+
+// WriteTo builds a UDP packet and writes it using the specific IP version writter
+func (s *SharedSocket) WriteTo(buf []byte, rAddr net.Addr) (n int, err error) {
+	rUDPAddr, ok := rAddr.(*net.UDPAddr)
+	if !ok {
+		return -1, fmt.Errorf("invalid address type")
+	}
+
+	buffer := gopacket.NewSerializeBuffer()
+	payload := gopacket.Payload(buf)
+
+	udp := &layers.UDP{
+		SrcPort: layers.UDPPort(s.port),
+		DstPort: layers.UDPPort(rUDPAddr.Port),
+	}
+
+	s.routerMux.RLock()
+	defer s.routerMux.RUnlock()
+
+	_, _, src, err := s.router.Route(rUDPAddr.IP)
+	if err != nil {
+		return 0, fmt.Errorf("got an error while checking route, err: %s", err)
+	}
+
+	rSockAddr, conn, nwLayer := s.getWriterObjects(src, rUDPAddr.IP)
+
+	if err := udp.SetNetworkLayerForChecksum(nwLayer); err != nil {
+		return -1, fmt.Errorf("failed to set network layer for checksum: %w", err)
+	}
+
+	if err := gopacket.SerializeLayers(buffer, writeSerializerOptions, udp, payload); err != nil {
+		return -1, fmt.Errorf("failed serialize rcvdPacket: %s", err)
+	}
+
+	bufser := buffer.Bytes()
+
+	return 0, conn.Sendto(context.TODO(), bufser, 0, rSockAddr)
+}
+
+// getWriterObjects returns the specific IP version objects that are used to build a packet and send it using the raw socket
+func (s *SharedSocket) getWriterObjects(src, dest net.IP) (sa unix.Sockaddr, conn *socket.Conn, layer gopacket.NetworkLayer) {
+	if dest.To4() == nil {
+		sa = &unix.SockaddrInet6{}
+		copy(sa.(*unix.SockaddrInet6).Addr[:], dest.To16())
+		conn = s.conn6
+
+		layer = &layers.IPv6{
+			SrcIP: src,
+			DstIP: dest,
+		}
+	} else {
+		sa = &unix.SockaddrInet4{}
+		copy(sa.(*unix.SockaddrInet4).Addr[:], dest.To4())
+		conn = s.conn4
+		layer = &layers.IPv4{
+			Version:  4,
+			TTL:      64,
+			Protocol: layers.IPProtocolUDP,
+			SrcIP:    src,
+			DstIP:    dest,
+		}
+	}
+
+	return sa, conn, layer
+}

sharedsock/sock_linux_test.go

@@ -0,0 +1,162 @@
+package sharedsock
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/netip"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pion/stun"
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+func TestShouldReadSTUNOnReadFrom(t *testing.T) {
+
+	// create raw socket on a port
+	testingPort := 51821
+	rawSock, err := Listen(testingPort, NewSTUNFilter())
+	require.NoError(t, err, "received an error while creating STUN listener, error: %s", err)
+	err = rawSock.SetReadDeadline(time.Now().Add(3 * time.Second))
+	require.NoError(t, err, "unable to set deadline, error: %s", err)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+
+	// when reading from the raw socket
+	buf := make([]byte, 1500)
+	rcvMSG := &stun.Message{
+		Raw: buf,
+	}
+	ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
+	defer cancel()
+
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			_, _, err := rawSock.ReadFrom(buf)
+			if err != nil {
+				log.Errorf("error while reading packet %s", err)
+				return
+			}
+
+			err = rcvMSG.Decode()
+			if err != nil {
+				log.Warnf("error while parsing STUN message. The packet doesn't seem to be a STUN packet: %s", err)
+				return
+			}
+			wg.Done()
+		}
+
+	}()
+
+	// and sending STUN packet to the shared port, the packet has to be handled
+	udpListener, err := net.ListenUDP("udp", &net.UDPAddr{Port: 12345, IP: net.ParseIP("127.0.0.1")})
+	require.NoError(t, err, "received an error while creating regular listener, error: %s", err)
+	defer udpListener.Close()
+	stunMSG, err := stun.Build(stun.NewType(stun.MethodBinding, stun.ClassRequest), stun.TransactionID,
+		stun.Fingerprint,
+	)
+	require.NoError(t, err, "unable to build stun msg, error: %s", err)
+	_, err = udpListener.WriteTo(stunMSG.Raw, net.UDPAddrFromAddrPort(netip.MustParseAddrPort(fmt.Sprintf("127.0.0.1:%d", testingPort))))
+	require.NoError(t, err, "received an error while writing the stun listener, error: %s", err)
+
+	// the packet has to be handled and be a STUN packet
+	wg.Wait()
+	require.EqualValues(t, stunMSG.TransactionID, rcvMSG.TransactionID, "transaction id values did't match")
+}
+
+func TestShouldNotReadNonSTUNPackets(t *testing.T) {
+	testingPort := 39439
+	rawSock, err := Listen(testingPort, NewSTUNFilter())
+	require.NoError(t, err, "received an error while creating STUN listener, error: %s", err)
+	defer rawSock.Close()
+
+	buf := make([]byte, 1500)
+	err = rawSock.SetReadDeadline(time.Now().Add(time.Second))
+	require.NoError(t, err, "unable to set deadline, error: %s", err)
+
+	errGrp := errgroup.Group{}
+	errGrp.Go(func() error {
+		_, _, err := rawSock.ReadFrom(buf)
+		return err
+	})
+	nonStun := []byte("netbird")
+	udpListener, err := net.ListenUDP("udp", &net.UDPAddr{Port: 0, IP: net.ParseIP("127.0.0.1")})
+	require.NoError(t, err, "received an error while creating regular listener, error: %s", err)
+	defer udpListener.Close()
+	remote := net.UDPAddrFromAddrPort(netip.MustParseAddrPort(fmt.Sprintf("127.0.0.1:%d", testingPort)))
+	_, err = udpListener.WriteTo(nonStun, remote)
+	require.NoError(t, err, "received an error while writing the stun listener, error: %s", err)
+
+	err = errGrp.Wait()
+	require.Error(t, err, "should receive an error")
+	if !errors.Is(err, os.ErrDeadlineExceeded) {
+		t.Errorf("error should be I/O timeout, got: %s", err)
+	}
+}
+
+func TestWriteTo(t *testing.T) {
+	udpListener, err := net.ListenUDP("udp4", &net.UDPAddr{Port: 0, IP: net.ParseIP("127.0.0.1")})
+	require.NoError(t, err, "received an error while creating regular listener, error: %s", err)
+	defer udpListener.Close()
+
+	testingPort := 39440
+	rawSock, err := Listen(testingPort, NewSTUNFilter())
+	require.NoError(t, err, "received an error while creating STUN listener, error: %s", err)
+	defer rawSock.Close()
+
+	buf := make([]byte, 1500)
+	err = udpListener.SetReadDeadline(time.Now().Add(3 * time.Second))
+	require.NoError(t, err, "unable to set deadline, error: %s", err)
+
+	errGrp := errgroup.Group{}
+	var remoteAdr net.Addr
+	var rcvBytes int
+	errGrp.Go(func() error {
+		n, a, err := udpListener.ReadFrom(buf)
+		remoteAdr = a
+		rcvBytes = n
+		return err
+	})
+
+	msg := []byte("netbird")
+	_, err = rawSock.WriteTo(msg, udpListener.LocalAddr())
+	require.NoError(t, err, "received an error while writing the stun listener, error: %s", err)
+
+	err = errGrp.Wait()
+	require.NoError(t, err, "received an error while reading the packet, error: %s", err)
+
+	require.EqualValues(t, string(msg), string(buf[:rcvBytes]), "received message should match")
+
+	udpRcv, ok := remoteAdr.(*net.UDPAddr)
+	require.True(t, ok, "udp address conversion didn't work")
+
+	require.EqualValues(t, testingPort, udpRcv.Port, "received address port didn't match")
+}
+
+func TestSharedSocket_Close(t *testing.T) {
+	rawSock, err := Listen(39440, NewSTUNFilter())
+	require.NoError(t, err, "received an error while creating STUN listener, error: %s", err)
+
+	errGrp := errgroup.Group{}
+
+	errGrp.Go(func() error {
+		buf := make([]byte, 1500)
+		_, _, err := rawSock.ReadFrom(buf)
+		return err
+	})
+	_ = rawSock.Close()
+	err = errGrp.Wait()
+	if err != ErrSharedSockStopped {
+		t.Errorf("invalid error response: %s", err)
+	}
+}

sharedsock/sock_nolinux.go

@@ -0,0 +1,14 @@
+//go:build !linux
+
+package sharedsock
+
+import (
+	"fmt"
+	"net"
+	"runtime"
+)
+
+// Listen is not supported on other platforms
+func Listen(port int, filter BPFFilter) (net.PacketConn, error) {
+	return nil, fmt.Errorf(fmt.Sprintf("Not supported OS %s. SharedSocket is only supported on Linux", runtime.GOOS))
+}