Bind test

client/main.go

@@ -1,12 +1,99 @@
 package main
 
 import (
-	"github.com/netbirdio/netbird/client/cmd"
+	"flag"
-	"os"
+	"github.com/netbirdio/netbird/iface"
+	log "github.com/sirupsen/logrus"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"net"
+	"net/http"
+	"time"
+
+	_ "net/http/pprof"
 )
 
+var name = flag.String("name", "wg0", "WireGuard interface name")
+var addr = flag.String("addr", "100.64.0.1/24", "interface WireGuard IP addr")
+var key = flag.String("key", "100.64.0.1/24", "WireGuard private key")
+var port = flag.Int("port", 51820, "WireGuard port")
+
+var remoteKey = flag.String("remote-key", "", "remote WireGuard public key")
+var remoteAddr = flag.String("remote-addr", "100.64.0.2/32", "remote WireGuard IP addr")
+var remoteEndpoint = flag.String("remote-endpoint", "127.0.0.1:51820", "remote WireGuard endpoint")
+
 func main() {
-	if err := cmd.Execute(); err != nil {
+
-		os.Exit(1)
+	flag.Parse()
+
+	go func() {
+		log.Println(http.ListenAndServe("localhost:6060", nil))
+	}()
+
+	myKey, err := wgtypes.ParseKey(*key)
+	if err != nil {
+		log.Error(err)
+		return
 	}
+
+	log.Infof("public key and addr [%s] [%s] ", myKey.PublicKey().String(), *addr)
+
+	wgIFace, err := iface.NewWGIFace(*name, *addr, 1280)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+	defer wgIFace.Close()
+
+	// todo wrap into UDPMux
+	sharedSock, _, err := listenNet("udp4", *port)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+	defer sharedSock.Close()
+
+	//	err = wgIFace.Create()
+	err = wgIFace.CreateNew(sharedSock)
+	if err != nil {
+		log.Errorf("failed to create interface %s %v", *name, err)
+		return
+	}
+
+	err = wgIFace.Configure(*key, *port)
+	if err != nil {
+		log.Errorf("failed to configure interface %s %v", *name, err)
+		return
+	}
+
+	ip, err := net.ResolveUDPAddr("udp4", *remoteEndpoint)
+	if err != nil {
+		// handle error
+	}
+
+	err = wgIFace.UpdatePeer(*remoteKey, *remoteAddr, 20*time.Second, ip, nil)
+	if err != nil {
+		log.Errorf("failed to configure remote peer %s %v", *remoteKey, err)
+		return
+	}
+
+	select {}
+
+}
+
+func listenNet(network string, port int) (*net.UDPConn, int, error) {
+	conn, err := net.ListenUDP(network, &net.UDPAddr{Port: port})
+	if err != nil {
+		return nil, 0, err
+	}
+
+	// Retrieve port.
+	laddr := conn.LocalAddr()
+	uaddr, err := net.ResolveUDPAddr(
+		laddr.Network(),
+		laddr.String(),
+	)
+	if err != nil {
+		return nil, 0, err
+	}
+	return conn, uaddr.Port, nil
 }

go.mod

@@ -39,6 +39,7 @@ require (
 	github.com/rs/xid v1.3.0
 	github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966
 	github.com/stretchr/testify v1.7.1
+	go.uber.org/zap v1.17.0
 	golang.org/x/net v0.0.0-20220513224357-95641704303c
 	golang.org/x/term v0.0.0-20220526004731-065cf7ba2467
 )
@@ -95,6 +96,8 @@ require (
 	github.com/srwiley/rasterx v0.0.0-20200120212402-85cb7272f5e9 // indirect
 	github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df // indirect
 	github.com/yuin/goldmark v1.4.1 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
 	golang.org/x/image v0.0.0-20200430140353-33d19683fad8 // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect

go.sum

@@ -638,8 +638,11 @@ go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
+go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
+go.uber.org/zap v1.17.0 h1:MTjgFu6ZLKvY6Pvaqk97GlxNBuMpV4Hy/3P6tRGlI2U=
 go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

iface/bind.go

@@ -0,0 +1,152 @@
+package iface
+
+import (
+	log "github.com/sirupsen/logrus"
+	"golang.zx2c4.com/wireguard/conn"
+	"net"
+	"net/netip"
+	"sync"
+)
+
+type packets struct {
+	buff []byte
+	addr net.UDPAddr
+}
+
+type ICEBind struct {
+	mu          sync.Mutex
+	packets     chan packets
+	closeSignal chan struct{}
+	conn        *net.UDPConn
+}
+
+func NewICEBind(udpConn *net.UDPConn) *ICEBind {
+	return &ICEBind{
+		conn: udpConn,
+		mu:   sync.Mutex{},
+	}
+}
+
+func (bind *ICEBind) Open(port uint16) ([]conn.ReceiveFunc, uint16, error) {
+
+	bind.mu.Lock()
+	defer bind.mu.Unlock()
+	log.Infof("opening Bind on port %d", port)
+
+	udpConn, p, err := listenNet("udp4", int(port))
+	if err != nil {
+		return nil, 0, err
+	}
+	bind.conn = udpConn
+	return []conn.ReceiveFunc{bind.makeReceiveIPv4(udpConn)}, uint16(p), nil
+
+	/*bind.packets = make(chan packets)
+	bind.closeSignal = make(chan struct{})
+
+	addrPort, err := netip.ParseAddrPort(bind.conn.LocalAddr().String())
+	if err != nil {
+		return nil, 0, err
+	}
+
+	return []conn.ReceiveFunc{bind.makeReceiveIPv4(bind.conn)}, addrPort.Port(), nil*/
+}
+
+func (bind *ICEBind) fakeReceiveIPv4(c *net.UDPConn) conn.ReceiveFunc {
+	return func(buff []byte) (int, conn.Endpoint, error) {
+		return 0, nil, nil
+	}
+}
+
+func (bind *ICEBind) makeReceiveIPv4(c *net.UDPConn) conn.ReceiveFunc {
+	return func(buff []byte) (int, conn.Endpoint, error) {
+		n, endpoint, err := c.ReadFromUDP(buff)
+		if endpoint != nil {
+			endpoint.IP = endpoint.IP.To4()
+		}
+		return n, (*conn.StdNetEndpoint)(endpoint), err
+	}
+}
+
+/*func (bind *ICEBind) receive(buff []byte) (int, conn.Endpoint, error) {
+	n, endpoint, err := bind.conn.ReadFromUDP(buff)
+	if endpoint != nil {
+		endpoint.IP = endpoint.IP.To4()
+	}
+	return n, (*conn.StdNetEndpoint)(endpoint), err
+
+	select {
+	case <-bind.closeSignal:
+		return 0, nil, net.ErrClosed
+	case pkt := <-bind.packets:
+		return copy(buf, pkt.buff), (*conn.StdNetEndpoint)(&pkt.addr), nil
+	}
+}*/
+
+func (bind *ICEBind) Close() error {
+	bind.mu.Lock()
+	defer bind.mu.Unlock()
+
+	err := bind.conn.Close()
+	if err != nil {
+		return err
+	}
+
+	if bind.closeSignal != nil {
+		select {
+		case <-bind.closeSignal:
+		default:
+			close(bind.closeSignal)
+		}
+		bind.packets = nil
+	}
+	return nil
+}
+
+// SetMark sets the mark for each packet sent through this Bind.
+// This mark is passed to the kernel as the socket option SO_MARK.
+func (bind *ICEBind) SetMark(mark uint32) error {
+	return nil
+}
+
+func (bind *ICEBind) Send(buf []byte, endpoint conn.Endpoint) error {
+
+	nend, ok := endpoint.(*conn.StdNetEndpoint)
+	if !ok {
+		return conn.ErrWrongEndpointType
+	}
+
+	_, err := bind.conn.WriteToUDP(buf, (*net.UDPAddr)(nend))
+	return err
+}
+
+// ParseEndpoint creates a new endpoint from a string.
+func (bind *ICEBind) ParseEndpoint(s string) (ep conn.Endpoint, err error) {
+	ap, err := netip.ParseAddrPort(s)
+	if err != nil {
+		return nil, err
+	}
+
+	return (*conn.StdNetEndpoint)(&net.UDPAddr{
+		IP:   ap.Addr().AsSlice(),
+		Port: int(ap.Port()),
+		Zone: ap.Addr().Zone(),
+	}), err
+}
+
+func listenNet(network string, port int) (*net.UDPConn, int, error) {
+	conn, err := net.ListenUDP(network, &net.UDPAddr{Port: port})
+	if err != nil {
+		return nil, 0, err
+	}
+
+	// Retrieve port.
+	laddr := conn.LocalAddr()
+	uaddr, err := net.ResolveUDPAddr(
+		laddr.Network(),
+		laddr.String(),
+	)
+	if err != nil {
+		return nil, 0, err
+	}
+	return conn, uaddr.Port, nil
+}

iface/iface_linux.go

@@ -3,6 +3,7 @@ package iface
 import (
 	"errors"
 	"math"
+	"net"
 	"os"
 	"syscall"
 
@@ -32,6 +33,12 @@ func WireguardModExists() bool {
 
 	return errors.Is(err, syscall.EINVAL)
 }
+func (w *WGIface) CreateNew(sharedSock *net.UDPConn) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	return w.createWithUserspaceNew(sharedSock)
+}
 
 // Create creates a new Wireguard interface, sets a given IP and brings it up.
 // Will reuse an existing one.
@@ -39,13 +46,7 @@ func (w *WGIface) Create() error {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 
-	if WireguardModExists() {
+	return w.createWithUserspace()
-		log.Info("using kernel WireGuard")
-		return w.createWithKernel()
-	} else {
-		log.Info("using userspace WireGuard")
-		return w.createWithUserspace()
-	}
 }
 
 // createWithKernel Creates a new Wireguard interface using kernel Wireguard module.

iface/iface_unix.go

@@ -12,6 +12,48 @@ import (
 	"net"
 )
 
+func (w *WGIface) createWithUserspaceNew(sharedSock *net.UDPConn) error {
+	tunIface, err := tun.CreateTUN(w.Name, w.MTU)
+	if err != nil {
+		return err
+	}
+
+	w.Interface = tunIface
+	bind := &ICEBind{
+		conn: sharedSock,
+	}
+
+	// We need to create a wireguard-go device and listen to configuration requests
+	tunDevice := device.NewDevice(tunIface, bind, device.NewLogger(device.LogLevelSilent, "[wiretrustee] "))
+	err = tunDevice.Up()
+	if err != nil {
+		return err
+	}
+	uapi, err := getUAPI(w.Name)
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		for {
+			uapiConn, uapiErr := uapi.Accept()
+			if uapiErr != nil {
+				log.Traceln("uapi Accept failed with error: ", uapiErr)
+				continue
+			}
+			go tunDevice.IpcHandle(uapiConn)
+		}
+	}()
+
+	log.Debugln("UAPI listener started")
+
+	err = w.assignAddr()
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // createWithUserspace Creates a new Wireguard interface, using wireguard-go userspace implementation
 func (w *WGIface) createWithUserspace() error {
 

management/client/grpc.go

@@ -109,7 +109,7 @@ func (c *GrpcClient) Sync(msgHandler func(msg *proto.SyncResponse) error) error
 			return err
 		}
 
-		stream, err := c.connectToStream(*serverPubKey)
+		cancel, stream, err := c.connectToStream(*serverPubKey)
 		if err != nil {
 			log.Debugf("failed to open Management Service stream: %s", err)
 			if s, ok := gstatus.FromError(err); ok && s.Code() == codes.PermissionDenied {
@@ -117,6 +117,7 @@ func (c *GrpcClient) Sync(msgHandler func(msg *proto.SyncResponse) error) error
 			}
 			return err
 		}
+		defer cancel()
 
 		log.Infof("connected to the Management Service stream")
 
@@ -145,7 +146,7 @@ func (c *GrpcClient) Sync(msgHandler func(msg *proto.SyncResponse) error) error
 	return nil
 }
 
-func (c *GrpcClient) connectToStream(serverPubKey wgtypes.Key) (proto.ManagementService_SyncClient, error) {
+func (c *GrpcClient) connectToStream(serverPubKey wgtypes.Key) (context.CancelFunc, proto.ManagementService_SyncClient, error) {
 	req := &proto.SyncRequest{}
 
 	myPrivateKey := c.key
@@ -154,11 +155,16 @@ func (c *GrpcClient) connectToStream(serverPubKey wgtypes.Key) (proto.Management
 	encryptedReq, err := encryption.EncryptMessage(serverPubKey, myPrivateKey, req)
 	if err != nil {
 		log.Errorf("failed encrypting message: %s", err)
-		return nil, err
+		return nil, nil, err
 	}
-
+	ctx, cancel := context.WithCancel(c.ctx)
 	syncReq := &proto.EncryptedMessage{WgPubKey: myPublicKey.String(), Body: encryptedReq}
-	return c.realClient.Sync(c.ctx, syncReq)
+	sync, err := c.realClient.Sync(ctx, syncReq)
+	if err != nil {
+		cancel()
+		return nil, nil, err
+	}
+	return cancel, sync, nil
 }
 
 func (c *GrpcClient) receiveEvents(stream proto.ManagementService_SyncClient, serverPubKey wgtypes.Key, msgHandler func(msg *proto.SyncResponse) error) error {