[android] allow selection/deselection of network resources on android peers (#4607)

client/iface/device/device_android.go

@@ -3,6 +3,7 @@
 package device
 
 import (
+	"fmt"
 	"strings"
 
 	log "github.com/sirupsen/logrus"
@@ -19,11 +20,12 @@ import (
 
 // WGTunDevice ignore the WGTunDevice interface on Android because the creation of the tun device is different on this platform
 type WGTunDevice struct {
-	address    wgaddr.Address
-	port       int
-	key        string
-	mtu        uint16
-	iceBind    *bind.ICEBind
+	address wgaddr.Address
+	port    int
+	key     string
+	mtu     uint16
+	iceBind *bind.ICEBind
+	// todo: review if we can eliminate the TunAdapter
 	tunAdapter TunAdapter
 	disableDNS bool
 
@@ -32,17 +34,19 @@ type WGTunDevice struct {
 	filteredDevice *FilteredDevice
 	udpMux         *udpmux.UniversalUDPMuxDefault
 	configurer     WGConfigurer
+	renewableTun   *RenewableTUN
 }
 
 func NewTunDevice(address wgaddr.Address, port int, key string, mtu uint16, iceBind *bind.ICEBind, tunAdapter TunAdapter, disableDNS bool) *WGTunDevice {
 	return &WGTunDevice{
-		address:    address,
-		port:       port,
-		key:        key,
-		mtu:        mtu,
-		iceBind:    iceBind,
-		tunAdapter: tunAdapter,
-		disableDNS: disableDNS,
+		address:      address,
+		port:         port,
+		key:          key,
+		mtu:          mtu,
+		iceBind:      iceBind,
+		tunAdapter:   tunAdapter,
+		disableDNS:   disableDNS,
+		renewableTun: NewRenewableTUN(),
 	}
 }
 
@@ -65,14 +69,17 @@ func (t *WGTunDevice) Create(routes []string, dns string, searchDomains []string
 		return nil, err
 	}
 
-	tunDevice, name, err := tun.CreateUnmonitoredTUNFromFD(fd)
+	unmonitoredTUN, name, err := tun.CreateUnmonitoredTUNFromFD(fd)
 	if err != nil {
 		_ = unix.Close(fd)
 		log.Errorf("failed to create Android interface: %s", err)
 		return nil, err
 	}
+
+	t.renewableTun.AddDevice(unmonitoredTUN)
+
 	t.name = name
-	t.filteredDevice = newDeviceFilter(tunDevice)
+	t.filteredDevice = newDeviceFilter(t.renewableTun)
 
 	log.Debugf("attaching to interface %v", name)
 	t.device = device.NewDevice(t.filteredDevice, t.iceBind, device.NewLogger(wgLogLevel(), "[netbird] "))
@@ -104,6 +111,23 @@ func (t *WGTunDevice) Up() (*udpmux.UniversalUDPMuxDefault, error) {
 	return udpMux, nil
 }
 
+func (t *WGTunDevice) RenewTun(fd int) error {
+	if t.device == nil {
+		return fmt.Errorf("device not initialized")
+	}
+
+	unmonitoredTUN, _, err := tun.CreateUnmonitoredTUNFromFD(fd)
+	if err != nil {
+		_ = unix.Close(fd)
+		log.Errorf("failed to renew Android interface: %s", err)
+		return err
+	}
+
+	t.renewableTun.AddDevice(unmonitoredTUN)
+
+	return nil
+}
+
 func (t *WGTunDevice) UpdateAddr(addr wgaddr.Address) error {
 	// todo implement
 	return nil

client/iface/device/device_netstack_android.go

@@ -2,6 +2,13 @@
 
 package device
 
+import "fmt"
+
 func (t *TunNetstackDevice) Create(routes []string, dns string, searchDomains []string) (WGConfigurer, error) {
 	return t.create()
 }
+
+func (t *TunNetstackDevice) RenewTun(fd int) error {
+	// Doesn't make sense in Android for Netstack.
+	return fmt.Errorf("this function has not been implemented in Netstack for Android")
+}

client/iface/device/renewable_tun.go

@@ -0,0 +1,309 @@
+//go:build android
+
+package device
+
+import (
+	"io"
+	"os"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"golang.zx2c4.com/wireguard/tun"
+)
+
+// closeAwareDevice wraps a tun.Device along with a flag
+// indicating whether its Close method was called.
+//
+// It also redirects tun.Device's Events() to a separate goroutine
+// and closes it when Close is called.
+//
+// The WaitGroup and CloseOnce fields are used to ensure that the
+// goroutine is awaited and closed only once.
+type closeAwareDevice struct {
+	isClosed atomic.Bool
+	tun.Device
+	closeEventCh chan struct{}
+	wg           sync.WaitGroup
+	closeOnce    sync.Once
+}
+
+func newClosableDevice(tunDevice tun.Device) *closeAwareDevice {
+	return &closeAwareDevice{
+		Device:       tunDevice,
+		isClosed:     atomic.Bool{},
+		closeEventCh: make(chan struct{}),
+	}
+}
+
+// redirectEvents redirects the Events() method of the underlying tun.Device
+// to the given channel (RenewableTUN's events channel).
+func (c *closeAwareDevice) redirectEvents(out chan tun.Event) {
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		for {
+			select {
+			case ev, ok := <-c.Device.Events():
+				if !ok {
+					return
+				}
+
+				if ev == tun.EventDown {
+					continue
+				}
+
+				select {
+				case out <- ev:
+				case <-c.closeEventCh:
+					return
+				}
+			case <-c.closeEventCh:
+				return
+			}
+		}
+	}()
+}
+
+// Close calls the underlying Device's Close method
+// after setting isClosed to true.
+func (c *closeAwareDevice) Close() (err error) {
+	c.closeOnce.Do(func() {
+		c.isClosed.Store(true)
+		close(c.closeEventCh)
+		err = c.Device.Close()
+		c.wg.Wait()
+	})
+
+	return err
+}
+
+func (c *closeAwareDevice) IsClosed() bool {
+	return c.isClosed.Load()
+}
+
+type RenewableTUN struct {
+	devices []*closeAwareDevice
+	mu      sync.Mutex
+	cond    *sync.Cond
+	events  chan tun.Event
+	closed  atomic.Bool
+}
+
+func NewRenewableTUN() *RenewableTUN {
+	r := &RenewableTUN{
+		devices: make([]*closeAwareDevice, 0),
+		mu:      sync.Mutex{},
+		events:  make(chan tun.Event, 16),
+	}
+	r.cond = sync.NewCond(&r.mu)
+	return r
+}
+
+func (r *RenewableTUN) File() *os.File {
+	for {
+		dev := r.peekLast()
+		if dev == nil {
+			if !r.waitForDevice() {
+				return nil
+			}
+			continue
+		}
+
+		file := dev.File()
+
+		if dev.IsClosed() {
+			time.Sleep(1 * time.Millisecond)
+			continue
+		}
+
+		return file
+	}
+}
+
+// Read reads from an underlying tun.Device kept in the r.devices slice.
+// If no device is available, it waits for one to be added via AddDevice().
+//
+// On error, it retries reading from the newest device instead of returning the error
+// if the device is closed; if not, it propagates the error.
+func (r *RenewableTUN) Read(bufs [][]byte, sizes []int, offset int) (n int, err error) {
+	for {
+		dev := r.peekLast()
+		if dev == nil {
+			// wait until AddDevice() signals a new device via cond.Broadcast()
+			if !r.waitForDevice() { // returns false if the renewable TUN itself is closed
+				return 0, io.EOF
+			}
+			continue
+		}
+
+		n, err = dev.Read(bufs, sizes, offset)
+		if err == nil {
+			return n, nil
+		}
+
+		// swap in progress; retry on the newest instead of returning the error
+		if dev.IsClosed() {
+			time.Sleep(1 * time.Millisecond)
+			continue
+		}
+		return n, err // propagate non-swap error
+	}
+}
+
+// Write writes to underlying tun.Device kept in the r.devices slice.
+// If no device is available, it waits for one to be added via AddDevice().
+//
+// On error, it retries writing to the newest device instead of returning the error
+// if the device is closed; if not, it propagates the error.
+func (r *RenewableTUN) Write(bufs [][]byte, offset int) (int, error) {
+	for {
+		dev := r.peekLast()
+		if dev == nil {
+			if !r.waitForDevice() {
+				return 0, io.EOF
+			}
+			continue
+		}
+
+		n, err := dev.Write(bufs, offset)
+		if err == nil {
+			return n, nil
+		}
+
+		if dev.IsClosed() {
+			time.Sleep(1 * time.Millisecond)
+			continue
+		}
+
+		return n, err
+	}
+}
+
+func (r *RenewableTUN) MTU() (int, error) {
+	for {
+		dev := r.peekLast()
+		if dev == nil {
+			if !r.waitForDevice() {
+				return 0, io.EOF
+			}
+			continue
+		}
+		mtu, err := dev.MTU()
+		if err == nil {
+			return mtu, nil
+		}
+		if dev.IsClosed() {
+			time.Sleep(1 * time.Millisecond)
+			continue
+		}
+		return 0, err
+	}
+}
+
+func (r *RenewableTUN) Name() (string, error) {
+	for {
+		dev := r.peekLast()
+		if dev == nil {
+			if !r.waitForDevice() {
+				return "", io.EOF
+			}
+			continue
+		}
+		name, err := dev.Name()
+		if err == nil {
+			return name, nil
+		}
+		if dev.IsClosed() {
+			time.Sleep(1 * time.Millisecond)
+			continue
+		}
+		return "", err
+	}
+}
+
+// Events returns a channel that is fed events from the underlying tun.Device's events channel
+// once it is added.
+func (r *RenewableTUN) Events() <-chan tun.Event {
+	return r.events
+}
+
+func (r *RenewableTUN) Close() error {
+	// Attempts to set the RenewableTUN closed flag to true.
+	// If it's already true, returns immediately.
+	if !r.closed.CompareAndSwap(false, true) {
+		return nil // already closed: idempotent
+	}
+	r.mu.Lock()
+	devices := r.devices
+	r.devices = nil
+	r.cond.Broadcast()
+	r.mu.Unlock()
+
+	var lastErr error
+
+	log.Debugf("closing %d devices", len(devices))
+	for _, device := range devices {
+		if err := device.Close(); err != nil {
+			log.Debugf("error closing a device: %v", err)
+			lastErr = err
+		}
+	}
+
+	close(r.events)
+	return lastErr
+}
+
+func (r *RenewableTUN) BatchSize() int {
+	return 1
+}
+
+func (r *RenewableTUN) AddDevice(device tun.Device) {
+	r.mu.Lock()
+	if r.closed.Load() {
+		r.mu.Unlock()
+		_ = device.Close()
+		return
+	}
+
+	var toClose *closeAwareDevice
+	if len(r.devices) > 0 {
+		toClose = r.devices[len(r.devices)-1]
+	}
+
+	cad := newClosableDevice(device)
+	cad.redirectEvents(r.events)
+
+	r.devices = []*closeAwareDevice{cad}
+	r.cond.Broadcast()
+
+	r.mu.Unlock()
+
+	if toClose != nil {
+		if err := toClose.Close(); err != nil {
+			log.Debugf("error closing last device: %v", err)
+		}
+	}
+}
+
+func (r *RenewableTUN) waitForDevice() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	for len(r.devices) == 0 && !r.closed.Load() {
+		r.cond.Wait()
+	}
+	return !r.closed.Load()
+}
+
+func (r *RenewableTUN) peekLast() *closeAwareDevice {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if len(r.devices) == 0 {
+		return nil
+	}
+
+	return r.devices[len(r.devices)-1]
+}

client/iface/device_android.go

@@ -21,5 +21,6 @@ type WGTunDevice interface {
 	FilteredDevice() *device.FilteredDevice
 	Device() *wgdevice.Device
 	GetNet() *netstack.Net
+	RenewTun(fd int) error
 	GetICEBind() device.EndpointManager
 }

client/iface/iface_create.go

@@ -24,3 +24,7 @@ func (w *WGIface) Create() error {
 func (w *WGIface) CreateOnAndroid([]string, string, []string) error {
 	return fmt.Errorf("this function has not implemented on non mobile")
 }
+
+func (w *WGIface) RenewTun(fd int) error {
+	return fmt.Errorf("this function has not been implemented on non-android")
+}

client/iface/iface_create_android.go

@@ -6,6 +6,7 @@ import (
 
 // CreateOnAndroid creates a new Wireguard interface, sets a given IP and brings it up.
 // Will reuse an existing one.
+// todo: review does this function really necessary or can we merge it with iOS
 func (w *WGIface) CreateOnAndroid(routes []string, dns string, searchDomains []string) error {
 	w.mu.Lock()
 	defer w.mu.Unlock()
@@ -22,3 +23,9 @@ func (w *WGIface) CreateOnAndroid(routes []string, dns string, searchDomains []s
 func (w *WGIface) Create() error {
 	return fmt.Errorf("this function has not implemented on this platform")
 }
+
+func (w *WGIface) RenewTun(fd int) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.tun.RenewTun(fd)
+}

client/iface/iface_create_darwin.go

@@ -39,3 +39,7 @@ func (w *WGIface) Create() error {
 func (w *WGIface) CreateOnAndroid([]string, string, []string) error {
 	return fmt.Errorf("this function has not implemented on this platform")
 }
+
+func (w *WGIface) RenewTun(fd int) error {
+	return fmt.Errorf("this function has not been implemented on this platform")
+}