Split client app into cmd and daemon service (#239)

client/internal/connect.go

@@ -0,0 +1,210 @@
+package internal
+
+import (
+	"context"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/wiretrustee/wiretrustee/iface"
+	mgm "github.com/wiretrustee/wiretrustee/management/client"
+	mgmProto "github.com/wiretrustee/wiretrustee/management/proto"
+	signal "github.com/wiretrustee/wiretrustee/signal/client"
+
+	"github.com/cenkalti/backoff/v4"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// RunClient with main logic.
+func RunClient(
+	ctx context.Context, config *Config, stopCh <-chan int, cleanupCh chan<- struct{},
+) error {
+	backOff := &backoff.ExponentialBackOff{
+		InitialInterval:     time.Second,
+		RandomizationFactor: backoff.DefaultRandomizationFactor,
+		Multiplier:          backoff.DefaultMultiplier,
+		MaxInterval:         10 * time.Second,
+		MaxElapsedTime:      24 * 3 * time.Hour, // stop the client after 3 days trying (must be a huge problem, e.g permission denied)
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}
+
+	state := CtxGetState(ctx)
+	defer state.Set(StatusIdle)
+
+	wrapErr := state.Wrap
+	operation := func() error {
+		// if context cancelled we not start new backoff cycle
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+		}
+
+		state.Set(StatusConnecting)
+		// validate our peer's Wireguard PRIVATE key
+		myPrivateKey, err := wgtypes.ParseKey(config.PrivateKey)
+		if err != nil {
+			log.Errorf("failed parsing Wireguard key %s: [%s]", config.PrivateKey, err.Error())
+			return wrapErr(err)
+		}
+
+		var mgmTlsEnabled bool
+		if config.ManagementURL.Scheme == "https" {
+			mgmTlsEnabled = true
+		}
+
+		// connect (just a connection, no stream yet) and login to Management Service to get an initial global Wiretrustee config
+		mgmClient, loginResp, err := connectToManagement(ctx, config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled)
+		if err != nil {
+			log.Warn(err)
+			return wrapErr(err)
+		}
+
+		// with the global Wiretrustee config in hand connect (just a connection, no stream yet) Signal
+		signalClient, err := connectToSignal(ctx, loginResp.GetWiretrusteeConfig(), myPrivateKey)
+		if err != nil {
+			log.Error(err)
+			return wrapErr(err)
+		}
+
+		peerConfig := loginResp.GetPeerConfig()
+
+		engineConfig, err := createEngineConfig(myPrivateKey, config, peerConfig)
+		if err != nil {
+			log.Error(err)
+			return wrapErr(err)
+		}
+
+		ctx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		engine := NewEngine(ctx, cancel, signalClient, mgmClient, engineConfig)
+		err = engine.Start()
+		if err != nil {
+			log.Errorf("error while starting Wiretrustee Connection Engine: %s", err)
+			return wrapErr(err)
+		}
+
+		log.Print("Wiretrustee engine started, my IP is: ", peerConfig.Address)
+		state.Set(StatusConnected)
+
+		select {
+		case <-stopCh:
+		case <-ctx.Done():
+		}
+
+		backOff.Reset()
+
+		err = mgmClient.Close()
+		if err != nil {
+			log.Errorf("failed closing Management Service client %v", err)
+			return wrapErr(err)
+		}
+		err = signalClient.Close()
+		if err != nil {
+			log.Errorf("failed closing Signal Service client %v", err)
+			return wrapErr(err)
+		}
+
+		err = engine.Stop()
+		if err != nil {
+			log.Errorf("failed stopping engine %v", err)
+			return wrapErr(err)
+		}
+
+		go func() {
+			cleanupCh <- struct{}{}
+		}()
+
+		log.Info("stopped Wiretrustee client")
+
+		if _, err := state.Status(); err == ErrResetConnection {
+			return err
+		}
+
+		return nil
+	}
+
+	err := backoff.Retry(operation, backOff)
+	if err != nil {
+		log.Errorf("exiting client retry loop due to unrecoverable error: %s", err)
+		return err
+	}
+	return nil
+}
+
+// createEngineConfig converts configuration received from Management Service to EngineConfig
+func createEngineConfig(key wgtypes.Key, config *Config, peerConfig *mgmProto.PeerConfig) (*EngineConfig, error) {
+	iFaceBlackList := make(map[string]struct{})
+	for i := 0; i < len(config.IFaceBlackList); i += 2 {
+		iFaceBlackList[config.IFaceBlackList[i]] = struct{}{}
+	}
+
+	engineConf := &EngineConfig{
+		WgIfaceName:    config.WgIface,
+		WgAddr:         peerConfig.Address,
+		IFaceBlackList: iFaceBlackList,
+		WgPrivateKey:   key,
+		WgPort:         iface.DefaultWgPort,
+	}
+
+	if config.PreSharedKey != "" {
+		preSharedKey, err := wgtypes.ParseKey(config.PreSharedKey)
+		if err != nil {
+			return nil, err
+		}
+		engineConf.PreSharedKey = &preSharedKey
+	}
+
+	return engineConf, nil
+}
+
+// connectToSignal creates Signal Service client and established a connection
+func connectToSignal(ctx context.Context, wtConfig *mgmProto.WiretrusteeConfig, ourPrivateKey wgtypes.Key) (*signal.GrpcClient, error) {
+	var sigTLSEnabled bool
+	if wtConfig.Signal.Protocol == mgmProto.HostConfig_HTTPS {
+		sigTLSEnabled = true
+	} else {
+		sigTLSEnabled = false
+	}
+
+	signalClient, err := signal.NewClient(ctx, wtConfig.Signal.Uri, ourPrivateKey, sigTLSEnabled)
+	if err != nil {
+		log.Errorf("error while connecting to the Signal Exchange Service %s: %s", wtConfig.Signal.Uri, err)
+		return nil, status.Errorf(codes.FailedPrecondition, "failed connecting to Signal Service : %s", err)
+	}
+
+	return signalClient, nil
+}
+
+// connectToManagement creates Management Services client, establishes a connection, logs-in and gets a global Wiretrustee config (signal, turn, stun hosts, etc)
+func connectToManagement(ctx context.Context, managementAddr string, ourPrivateKey wgtypes.Key, tlsEnabled bool) (*mgm.GrpcClient, *mgmProto.LoginResponse, error) {
+	log.Debugf("connecting to management server %s", managementAddr)
+	client, err := mgm.NewClient(ctx, managementAddr, ourPrivateKey, tlsEnabled)
+	if err != nil {
+		return nil, nil, status.Errorf(codes.FailedPrecondition, "failed connecting to Management Service : %s", err)
+	}
+	log.Debugf("connected to management server %s", managementAddr)
+
+	serverPublicKey, err := client.GetServerPublicKey()
+	if err != nil {
+		return nil, nil, status.Errorf(codes.FailedPrecondition, "failed while getting Management Service public key: %s", err)
+	}
+
+	loginResp, err := client.Login(*serverPublicKey)
+	if err != nil {
+		if s, ok := status.FromError(err); ok && s.Code() == codes.PermissionDenied {
+			log.Error("peer registration required. Please run wiretrustee login command first")
+			return nil, nil, err
+		} else {
+			return nil, nil, err
+		}
+	}
+
+	log.Debugf("peer logged in to Management Service %s", managementAddr)
+
+	return client, loginResp, nil
+}
+

client/internal/engine.go

@@ -31,6 +31,8 @@ const (
 	PeerConnectionTimeoutMin = 30000 // ms
 )
 
+var ErrResetConnection = fmt.Errorf("reset connection")
+
 // EngineConfig is a config for the Engine
 type EngineConfig struct {
 	WgPort      int
@@ -95,10 +97,12 @@ type Peer struct {
 
 // NewEngine creates a new Connection Engine
 func NewEngine(
+	ctx context.Context, cancel context.CancelFunc,
 	signalClient signal.Client, mgmClient mgm.Client, config *EngineConfig,
-	cancel context.CancelFunc, ctx context.Context,
 ) *Engine {
 	return &Engine{
+		ctx:           ctx,
+		cancel:        cancel,
 		signal:        signalClient,
 		mgmClient:     mgmClient,
 		peerConns:     map[string]*peer.Conn{},
@@ -106,8 +110,6 @@ func NewEngine(
 		config:        config,
 		STUNs:         []*ice.URL{},
 		TURNs:         []*ice.URL{},
-		cancel:        cancel,
-		ctx:           ctx,
 		networkSerial: 0,
 	}
 }
@@ -380,6 +382,7 @@ func (e *Engine) receiveManagementEvents() {
 		if err != nil {
 			// happens if management is unavailable for a long time.
 			// We want to cancel the operation of the whole client
+			_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
 			e.cancel()
 			return
 		}
@@ -615,6 +618,7 @@ func (e *Engine) receiveSignalEvents() {
 		if err != nil {
 			// happens if signal is unavailable for a long time.
 			// We want to cancel the operation of the whole client
+			_ = CtxGetState(e.ctx).Wrap(ErrResetConnection)
 			e.cancel()
 			return
 		}

client/internal/engine_test.go

@@ -40,7 +40,6 @@ var (
 )
 
 func TestEngine_UpdateNetworkMap(t *testing.T) {
-
 	// test setup
 	key, err := wgtypes.GeneratePrivateKey()
 	if err != nil {
@@ -51,12 +50,12 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	engine := NewEngine(&signal.MockClient{}, &mgmt.MockClient{}, &EngineConfig{
+	engine := NewEngine(ctx, cancel, &signal.MockClient{}, &mgmt.MockClient{}, &EngineConfig{
 		WgIfaceName:  "utun100",
 		WgAddr:       "100.64.0.1/24",
 		WgPrivateKey: key,
 		WgPort:       33100,
-	}, cancel, ctx)
+	})
 
 	type testCase struct {
 		name       string
@@ -157,7 +156,6 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 	}
 
 	for _, c := range []testCase{case1, case2, case3, case4, case5} {
-
 		t.Run(c.name, func(t *testing.T) {
 			err = engine.updateNetworkMap(c.networkMap)
 			if err != nil {
@@ -179,13 +177,10 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 				}
 			}
 		})
-
 	}
-
 }
 
 func TestEngine_Sync(t *testing.T) {
-
 	key, err := wgtypes.GeneratePrivateKey()
 	if err != nil {
 		t.Fatal(err)
@@ -199,7 +194,6 @@ func TestEngine_Sync(t *testing.T) {
 	updates := make(chan *mgmtProto.SyncResponse)
 	defer close(updates)
 	syncFunc := func(msgHandler func(msg *mgmtProto.SyncResponse) error) error {
-
 		for msg := range updates {
 			err := msgHandler(msg)
 			if err != nil {
@@ -209,12 +203,12 @@ func TestEngine_Sync(t *testing.T) {
 		return nil
 	}
 
-	engine := NewEngine(&signal.MockClient{}, &mgmt.MockClient{SyncFunc: syncFunc}, &EngineConfig{
+	engine := NewEngine(ctx, cancel, &signal.MockClient{}, &mgmt.MockClient{SyncFunc: syncFunc}, &EngineConfig{
 		WgIfaceName:  "utun100",
 		WgAddr:       "100.64.0.1/24",
 		WgPrivateKey: key,
 		WgPort:       33100,
-	}, cancel, ctx)
+	})
 
 	defer func() {
 		err := engine.Stop()
@@ -264,12 +258,10 @@ func TestEngine_Sync(t *testing.T) {
 			break
 		}
 	}
-
 }
 
 func TestEngine_MultiplePeers(t *testing.T) {
-
-	//log.SetLevel(log.DebugLevel)
+	// log.SetLevel(log.DebugLevel)
 
 	dir := t.TempDir()
 
@@ -285,8 +277,9 @@ func TestEngine_MultiplePeers(t *testing.T) {
 		}
 	}()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(CtxInitState(context.Background()))
 	defer cancel()
+
 	sport := 10010
 	sigServer, err := startSignal(sport)
 	if err != nil {
@@ -366,7 +359,6 @@ func TestEngine_MultiplePeers(t *testing.T) {
 }
 
 func createEngine(ctx context.Context, cancel context.CancelFunc, setupKey string, i int, mport int, sport int) (*Engine, error) {
-
 	key, err := wgtypes.GeneratePrivateKey()
 	if err != nil {
 		return nil, err
@@ -406,7 +398,7 @@ func createEngine(ctx context.Context, cancel context.CancelFunc, setupKey strin
 		WgPort:       wgPort,
 	}
 
-	return NewEngine(signalClient, mgmtClient, conf, cancel, ctx), nil
+	return NewEngine(ctx, cancel, signalClient, mgmtClient, conf), nil
 }
 
 func startSignal(port int) (*grpc.Server, error) {
@@ -429,7 +421,6 @@ func startSignal(port int) (*grpc.Server, error) {
 }
 
 func startManagement(port int, dataDir string) (*grpc.Server, error) {
-
 	config := &server.Config{
 		Stuns:      []*server.Host{},
 		TURNConfig: &server.TURNConfig{},

client/internal/login.go

@@ -0,0 +1,118 @@
+package internal
+
+import (
+	"context"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+	"github.com/wiretrustee/wiretrustee/client/system"
+	mgm "github.com/wiretrustee/wiretrustee/management/client"
+	mgmProto "github.com/wiretrustee/wiretrustee/management/proto"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func Login(ctx context.Context, config *Config, setupKey string) error {
+	backOff := &backoff.ExponentialBackOff{
+		InitialInterval:     time.Second,
+		RandomizationFactor: backoff.DefaultRandomizationFactor,
+		Multiplier:          backoff.DefaultMultiplier,
+		MaxInterval:         2 * time.Second,
+		MaxElapsedTime:      time.Second * 10,
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}
+
+	// validate our peer's Wireguard PRIVATE key
+	myPrivateKey, err := wgtypes.ParseKey(config.PrivateKey)
+	if err != nil {
+		log.Errorf("failed parsing Wireguard key %s: [%s]", config.PrivateKey, err.Error())
+		return err
+	}
+
+	var mgmTlsEnabled bool
+	if config.ManagementURL.Scheme == "https" {
+		mgmTlsEnabled = true
+	}
+
+	loginOp := func() error {
+		log.Debugf("connecting to Management Service %s", config.ManagementURL.String())
+		mgmClient, err := mgm.NewClient(ctx, config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled)
+		if err != nil {
+			log.Errorf("failed connecting to Management Service %s %v", config.ManagementURL.String(), err)
+			return err
+		}
+		log.Debugf("connected to management Service %s", config.ManagementURL.String())
+
+		serverKey, err := mgmClient.GetServerPublicKey()
+		if err != nil {
+			log.Errorf("failed while getting Management Service public key: %v", err)
+			return err
+		}
+
+		_, err = loginPeer(*serverKey, mgmClient, setupKey)
+		if err != nil {
+			log.Errorf("failed logging-in peer on Management Service : %v", err)
+			return err
+		}
+
+		err = mgmClient.Close()
+		if err != nil {
+			log.Errorf("failed closing Management Service client: %v", err)
+			return err
+		}
+
+		return nil
+	}
+
+	err = backoff.RetryNotify(loginOp, backOff, func(err error, duration time.Duration) {
+		log.Warnf("retrying Login to the Management service in %v due to error %v", duration, err)
+	})
+	if err != nil {
+		log.Errorf("exiting login retry loop due to unrecoverable error: %v", err)
+		return err
+	}
+
+	return nil
+}
+
+// loginPeer attempts to login to Management Service. If peer wasn't registered, tries the registration flow.
+func loginPeer(serverPublicKey wgtypes.Key, client *mgm.GrpcClient, setupKey string) (*mgmProto.LoginResponse, error) {
+	loginResp, err := client.Login(serverPublicKey)
+	if err != nil {
+		if s, ok := status.FromError(err); ok && s.Code() == codes.PermissionDenied {
+			log.Debugf("peer registration required")
+			return registerPeer(serverPublicKey, client, setupKey)
+		} else {
+			return nil, err
+		}
+	}
+
+	log.Info("peer has successfully logged-in to Management Service")
+
+	return loginResp, nil
+}
+
+// registerPeer checks whether setupKey was provided via cmd line and if not then it prompts user to enter a key.
+// Otherwise tries to register with the provided setupKey via command line.
+func registerPeer(serverPublicKey wgtypes.Key, client *mgm.GrpcClient, setupKey string) (*mgmProto.LoginResponse, error) {
+	validSetupKey, err := uuid.Parse(setupKey)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Debugf("sending peer registration request to Management Service")
+	info := system.GetInfo()
+	loginResp, err := client.Register(serverPublicKey, validSetupKey.String(), info)
+	if err != nil {
+		log.Errorf("failed registering peer %v", err)
+		return nil, err
+	}
+
+	log.Infof("peer has been successfully registered on Management Service")
+
+	return loginResp, nil
+}

client/internal/state.go

@@ -0,0 +1,67 @@
+package internal
+
+import (
+	"context"
+	"sync"
+)
+
+type StatusType string
+
+const (
+	StatusIdle StatusType = "Idle"
+
+	StatusConnecting StatusType = "Connecting"
+	StatusConnected  StatusType = "Connected"
+)
+
+// CtxInitState setup context state into the context tree.
+//
+// This function should be used to initialize context before
+// CtxGetState will be executed.
+func CtxInitState(ctx context.Context) context.Context {
+	return context.WithValue(ctx, stateCtx, &contextState{
+		status: StatusIdle,
+	})
+}
+
+// CtxGetState object to get/update state/errors of process.
+func CtxGetState(ctx context.Context) *contextState {
+	return ctx.Value(stateCtx).(*contextState)
+}
+
+type contextState struct {
+	err    error
+	status StatusType
+	mutex  sync.Mutex
+}
+
+func (c *contextState) Set(update StatusType) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	c.status = update
+	c.err = nil
+}
+
+func (c *contextState) Status() (StatusType, error) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	if c.err != nil {
+		return "", c.err
+	}
+
+	return c.status, nil
+}
+
+func (c *contextState) Wrap(err error) error {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	c.err = err
+	return err
+}
+
+type stateKey int
+
+var stateCtx stateKey