feature: Support live peer list update (#51)

* created InitializePeer and ClosePeerConnection functions

* feature: simplify peer stopping

* chore: remove unused code

* feature: basic management service implementation (#44)

* feat: basic management service implementation [FAILING TESTS]

* test: fix healthcheck test

* test: #39 add peer registration endpoint test

* feat: #39 add setup key handling

* feat: #39 add peer management store persistence

* refactor: extract config read/write to the utility package

* refactor: move file contents copy to the utility package

* refactor: use Accounts instead of Users in the Store

* feature: add management server Docker file

* refactor: introduce datadir instead of config

* chore: use filepath.Join to concat filepaths instead of string concat

* refactor: move stop channel to the root

* refactor: move stop channel to the root

* review: fix PR review notes

Co-authored-by: braginini <hello@wiretrustee.com>

* Handle read config file errors

* feature: add letsencrypt support to the management service

* fix: lint warnings

* chore: change default datadir

* refactor: set default flags in code not Dockerfile

* chore: remove unused code

* Added RemovePeer and centralized configureDevice code

* remove peer from the wg interface when closing proxy

* remove config file

* add iface tests

* fix tests, validate if file exists before removing it

* removed unused functions UpdateListenPort and ConfigureWithKeyGen

* Ensure we don't wait for timeout when closing

* Rename ClosePeerConnection to RemovePeerConnection

* Avoid returning on uapi Accept failures

* Added engine tests

* Remove extra add address code

* Adding iface.Close

* Ensure Close the interface and disable parallel test execution

* check err var when listing interfaces

* chore: add synchronisation to peer management

* chore: add connection status to track peer connection

* refactor: remove unused code

Co-authored-by: braginini <hello@wiretrustee.com>
Co-authored-by: Mikhail Bragin <bangvalo@gmail.com>

connection/connection.go

@@ -16,6 +16,14 @@ var (
 	DefaultWgKeepAlive = 20 * time.Second
 )
 
+type Status string
+
+const (
+	StatusConnected    Status = "Connected"
+	StatusConnecting   Status = "Connecting"
+	StatusDisconnected Status = "Disconnected"
+)
+
 // ConnConfig Connection configuration struct
 type ConnConfig struct {
 	// Local Wireguard listening address  e.g. 127.0.0.1:51820
@@ -66,6 +74,8 @@ type Connection struct {
 	closeCond *Cond
 
 	remoteAuthCond sync.Once
+
+	Status Status
 }
 
 // NewConnection Creates a new connection and sets handling functions for signal protocol
@@ -85,6 +95,7 @@ func NewConnection(config ConnConfig,
 		connected:         NewCond(),
 		agent:             nil,
 		wgProxy:           NewWgProxy(config.WgIface, config.RemoteWgKey.String(), config.WgAllowedIPs, config.WgListenAddr),
+		Status:            StatusDisconnected,
 	}
 }
 
@@ -126,6 +137,7 @@ func (conn *Connection) Open(timeout time.Duration) error {
 		return err
 	}
 
+	conn.Status = StatusConnecting
 	log.Infof("trying to connect to peer %s", conn.Config.RemoteWgKey.String())
 
 	// wait until credentials have been sent from the remote peer (will arrive via a signal server)
@@ -164,17 +176,23 @@ func (conn *Connection) Open(timeout time.Duration) error {
 			}
 		}
 
+		conn.Status = StatusConnected
 		log.Infof("opened connection to peer %s", conn.Config.RemoteWgKey.String())
+	case <-conn.closeCond.C:
+		conn.Status = StatusDisconnected
+		return fmt.Errorf("connection to peer %s has been closed", conn.Config.RemoteWgKey.String())
 	case <-time.After(timeout):
 		err := conn.Close()
 		if err != nil {
 			log.Warnf("error while closing connection to peer %s -> %s", conn.Config.RemoteWgKey.String(), err.Error())
 		}
+		conn.Status = StatusDisconnected
 		return fmt.Errorf("timeout of %vs exceeded while waiting for the remote peer %s", timeout.Seconds(), conn.Config.RemoteWgKey.String())
 	}
 
 	// wait until connection has been closed
 	<-conn.closeCond.C
+	conn.Status = StatusDisconnected
 	return fmt.Errorf("connection to peer %s has been closed", conn.Config.RemoteWgKey.String())
 }
 

connection/engine.go

@@ -9,9 +9,14 @@ import (
 	"github.com/wiretrustee/wiretrustee/signal"
 	sProto "github.com/wiretrustee/wiretrustee/signal/proto"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"sync"
 	"time"
 )
 
+// PeerConnectionTimeout is a timeout of an initial connection attempt to a remote peer.
+// E.g. this peer will wait PeerConnectionTimeout for the remote peer to respond, if not successful then it will retry the connection attempt.
+const PeerConnectionTimeout = 60 * time.Second
+
 // Engine is an instance of the Connection Engine
 type Engine struct {
 	// a list of STUN and TURN servers
@@ -26,6 +31,8 @@ type Engine struct {
 	wgIP string
 	// Network Interfaces to ignore
 	iFaceBlackList map[string]struct{}
+	// PeerMux is used to sync peer operations (e.g. open connection, peer removal)
+	PeerMux *sync.Mutex
 }
 
 // Peer is an instance of the Connection Peer
@@ -44,6 +51,7 @@ func NewEngine(signal *signal.Client, stunsTurns []*ice.URL, wgIface string, wgA
 		wgIP:           wgAddr,
 		conns:          map[string]*Connection{},
 		iFaceBlackList: iFaceBlackList,
+		PeerMux:        &sync.Mutex{},
 	}
 }
 
@@ -71,42 +79,76 @@ func (e *Engine) Start(myKey wgtypes.Key, peers []Peer) error {
 
 	e.receiveSignal()
 
-	// initialize peer agents
 	for _, peer := range peers {
-
 		peer := peer
-		go func() {
-			var backOff = &backoff.ExponentialBackOff{
-				InitialInterval:     backoff.DefaultInitialInterval,
-				RandomizationFactor: backoff.DefaultRandomizationFactor,
-				Multiplier:          backoff.DefaultMultiplier,
-				MaxInterval:         5 * time.Second,
-				MaxElapsedTime:      time.Duration(0), //never stop
-				Stop:                backoff.Stop,
-				Clock:               backoff.SystemClock,
-			}
-			operation := func() error {
-				_, err := e.openPeerConnection(*wgPort, myKey, peer)
-				if err != nil {
-					log.Warnln("retrying connection because of error: ", err.Error())
-					e.conns[peer.WgPubKey] = nil
-					return err
-				}
-				backOff.Reset()
-				return nil
-			}
-
-			err = backoff.Retry(operation, backOff)
-			if err != nil {
-				// should actually never happen
-				panic(err)
-			}
-		}()
+		go e.InitializePeer(*wgPort, myKey, peer)
 	}
 	return nil
 }
 
+// InitializePeer peer agent attempt to open connection
+func (e *Engine) InitializePeer(wgPort int, myKey wgtypes.Key, peer Peer) {
+	var backOff = &backoff.ExponentialBackOff{
+		InitialInterval:     backoff.DefaultInitialInterval,
+		RandomizationFactor: backoff.DefaultRandomizationFactor,
+		Multiplier:          backoff.DefaultMultiplier,
+		MaxInterval:         5 * time.Second,
+		MaxElapsedTime:      time.Duration(0), //never stop
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}
+	operation := func() error {
+		_, err := e.openPeerConnection(wgPort, myKey, peer)
+		e.PeerMux.Lock()
+		defer e.PeerMux.Unlock()
+		if _, ok := e.conns[peer.WgPubKey]; !ok {
+			log.Infof("removing connection attempt with Peer: %v, not retrying", peer.WgPubKey)
+			return nil
+		}
+
+		if err != nil {
+			log.Warnln(err)
+			log.Warnln("retrying connection because of error: ", err.Error())
+			return err
+		}
+		return nil
+	}
+
+	err := backoff.Retry(operation, backOff)
+	if err != nil {
+		// should actually never happen
+		panic(err)
+	}
+}
+
+// RemovePeerConnection closes existing peer connection and removes peer
+func (e *Engine) RemovePeerConnection(peer Peer) error {
+	e.PeerMux.Lock()
+	defer e.PeerMux.Unlock()
+	conn, exists := e.conns[peer.WgPubKey]
+	if exists && conn != nil {
+		delete(e.conns, peer.WgPubKey)
+		return conn.Close()
+	}
+	return nil
+}
+
+// GetPeerConnectionStatus returns a connection Status or nil if peer connection wasn't found
+func (e *Engine) GetPeerConnectionStatus(peerKey string) *Status {
+	e.PeerMux.Lock()
+	defer e.PeerMux.Unlock()
+
+	conn, exists := e.conns[peerKey]
+	if exists && conn != nil {
+		return &conn.Status
+	}
+
+	return nil
+}
+
+// opens a new peer connection
 func (e *Engine) openPeerConnection(wgPort int, myKey wgtypes.Key, peer Peer) (*Connection, error) {
+	e.PeerMux.Lock()
 
 	remoteKey, _ := wgtypes.ParseKey(peer.WgPubKey)
 	connConfig := &ConnConfig{
@@ -130,11 +172,12 @@ func (e *Engine) openPeerConnection(wgPort int, myKey wgtypes.Key, peer Peer) (*
 	signalCandidate := func(candidate ice.Candidate) error {
 		return signalCandidate(candidate, myKey, remoteKey, e.signal)
 	}
-
 	conn := NewConnection(*connConfig, signalCandidate, signalOffer, signalAnswer)
 	e.conns[remoteKey.String()] = conn
+	e.PeerMux.Unlock()
+
 	// blocks until the connection is open (or timeout)
-	err := conn.Open(60 * time.Second)
+	err := conn.Open(PeerConnectionTimeout)
 	if err != nil {
 		return nil, err
 	}

connection/engine_test.go

@@ -0,0 +1,163 @@
+package connection
+
+import (
+	"context"
+	"fmt"
+	ice "github.com/pion/ice/v2"
+	log "github.com/sirupsen/logrus"
+	"github.com/wiretrustee/wiretrustee/iface"
+	sig "github.com/wiretrustee/wiretrustee/signal"
+	"golang.zx2c4.com/wireguard/wgctrl"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"testing"
+	"time"
+)
+
+var engine *Engine
+var testKey wgtypes.Key
+var testPeer Peer
+
+const ifaceName = "utun9991"
+
+func Test_Start(t *testing.T) {
+	level, _ := log.ParseLevel("Debug")
+	log.SetLevel(level)
+
+	var err error
+	testKey, err = wgtypes.GenerateKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx := context.Background()
+
+	iceUrl, err := ice.ParseURL("stun:stun.wiretrustee.com:3468")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var stunURLs = []*ice.URL{iceUrl}
+
+	iFaceBlackList := make(map[string]struct{})
+
+	signalClient, err := sig.NewClient(ctx, "signal.wiretrustee.com:10000", testKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	engine = NewEngine(signalClient, stunURLs, ifaceName, "10.99.91.1/24", iFaceBlackList)
+
+	var emptyPeer []Peer
+	err = engine.Start(testKey, emptyPeer)
+	if err != nil {
+		t.Fatal(err)
+	}
+	wg, err := wgctrl.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer wg.Close()
+
+	_, err = wg.Device(ifaceName)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestEngine_InitializePeerWithoutRemote(t *testing.T) {
+	tmpKey, err := wgtypes.GenerateKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+	testPeer = Peer{
+		tmpKey.PublicKey().String(),
+		"10.99.91.2/32",
+	}
+	go engine.InitializePeer(iface.WgPort, testKey, testPeer)
+	// Let the connections initialize
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	for {
+		status := engine.GetPeerConnectionStatus(testPeer.WgPubKey)
+		err = ctx.Err()
+		if (status != nil && *status == StatusConnecting) || err != nil {
+			if err != nil {
+				t.Fatal(err)
+			}
+			//success
+			break
+		}
+	}
+}
+
+func TestEngine_Initialize2PeersWithoutRemote(t *testing.T) {
+	tmpKey1, err := wgtypes.GenerateKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+	tmpKey2, err := wgtypes.GenerateKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+	testPeer1 := Peer{
+		tmpKey1.PublicKey().String(),
+		"10.99.91.2/32",
+	}
+	testPeer2 := Peer{
+		tmpKey2.PublicKey().String(),
+		"10.99.91.3/32",
+	}
+	go engine.InitializePeer(iface.WgPort, testKey, testPeer1)
+	go engine.InitializePeer(iface.WgPort, testKey, testPeer2)
+	// Let the connections initialize
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	for {
+		status1 := engine.GetPeerConnectionStatus(testPeer1.WgPubKey)
+		status2 := engine.GetPeerConnectionStatus(testPeer2.WgPubKey)
+		err = ctx.Err()
+		if (status1 != nil && status2 != nil) || err != nil {
+			if err != nil {
+				t.Fatal(err)
+			}
+			if *status1 == StatusConnecting && *status2 == StatusConnecting {
+				//success
+				break
+			}
+		}
+	}
+}
+
+func TestEngine_RemovePeerConnectionWithoutRemote(t *testing.T) {
+
+	// Let the connections initialize
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	for {
+		status := engine.GetPeerConnectionStatus(testPeer.WgPubKey)
+		err := ctx.Err()
+		if (status != nil && *status == StatusConnecting) || err != nil {
+			if err != nil {
+				t.Fatal(err)
+			}
+			break
+		}
+	}
+
+	// Let the connections close
+	err := engine.RemovePeerConnection(testPeer)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	status := engine.GetPeerConnectionStatus(testPeer.WgPubKey)
+	if status != nil {
+		t.Fatal(fmt.Errorf("wrong status %v", status))
+	}
+}
+
+func Test_CloseInterface(t *testing.T) {
+	err := iface.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}

connection/wgproxy.go

@@ -38,10 +38,15 @@ func (p *WgProxy) Close() error {
 			return err
 		}
 	}
+	err := iface.RemovePeer(p.iface, p.remoteKey)
+	if err != nil {
+		return err
+	}
 
 	return nil
 }
 
+// StartLocal configure the interface with a peer using a direct IP:Port endpoint to the remote host
 func (p *WgProxy) StartLocal(host string) error {
 	err := iface.UpdatePeer(p.iface, p.remoteKey, p.allowedIps, DefaultWgKeepAlive, host)
 	if err != nil {