ebpf test

* updage flag values from environment variables

* add log and removing unused constants

* removing unused code

* Docker build client

* fix indentation

* Documentation with docker command

* use docker volume

.goreleaser.yaml

@@ -37,6 +37,7 @@ builds:
     goarch:
       - amd64
       - arm64
+      - arm
     ldflags:
       - -s -w -X main.version={{.Version}} -X main.commit={{.Commit}} -X main.date={{.CommitDate}} -X main.builtBy=goreleaser
     mod_timestamp: '{{ .CommitTimestamp }}'
@@ -50,6 +51,7 @@ builds:
     goarch:
       - amd64
       - arm64
+      - arm
     ldflags:
       - -s -w -X main.version={{.Version}} -X main.commit={{.Commit}} -X main.date={{.CommitDate}} -X main.builtBy=goreleaser
     mod_timestamp: '{{ .CommitTimestamp }}'
@@ -83,6 +85,52 @@ nfpms:
       postinstall: "release_files/post_install.sh"
       preremove: "release_files/pre_remove.sh"
 dockers:
+  - image_templates:
+      - wiretrustee/wiretrustee:{{ .Version }}-amd64
+    ids:
+      - wiretrustee
+    goarch: amd64
+    use: buildx
+    dockerfile: client/Dockerfile
+    build_flag_templates:
+      - "--platform=linux/amd64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.title={{.ProjectName}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=maintainer=wiretrustee@wiretrustee.com"
+  - image_templates:
+      - wiretrustee/wiretrustee:{{ .Version }}-arm64v8
+    ids:
+      - wiretrustee
+    goarch: arm64
+    use: buildx
+    dockerfile: client/Dockerfile
+    build_flag_templates:
+      - "--platform=linux/arm64"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.title={{.ProjectName}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=maintainer=wiretrustee@wiretrustee.com"
+  - image_templates:
+      - wiretrustee/wiretrustee:{{ .Version }}-arm
+    ids:
+      - wiretrustee
+    goarch: arm
+    goarm: 6
+    use: buildx
+    dockerfile: client/Dockerfile
+    build_flag_templates:
+      - "--platform=linux/arm"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.title={{.ProjectName}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=maintainer=wiretrustee@wiretrustee.com"
   - image_templates:
       - wiretrustee/signal:{{ .Version }}-amd64
     ids:
@@ -113,6 +161,22 @@ dockers:
       - "--label=org.opencontainers.image.revision={{.FullCommit}}"
       - "--label=org.opencontainers.image.version={{.Version}}"
       - "--label=maintainer=wiretrustee@wiretrustee.com"
+  - image_templates:
+      - wiretrustee/signal:{{ .Version }}-arm
+    ids:
+      - wiretrustee-signal
+    goarch: arm
+    goarm: 6
+    use: buildx
+    dockerfile: signal/Dockerfile
+    build_flag_templates:
+      - "--platform=linux/arm"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.title={{.ProjectName}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=maintainer=wiretrustee@wiretrustee.com"
   - image_templates:
       - wiretrustee/management:{{ .Version }}-amd64
     ids:
@@ -143,6 +207,22 @@ dockers:
       - "--label=org.opencontainers.image.revision={{.FullCommit}}"
       - "--label=org.opencontainers.image.version={{.Version}}"
       - "--label=maintainer=wiretrustee@wiretrustee.com"
+  - image_templates:
+      - wiretrustee/management:{{ .Version }}-arm
+    ids:
+      - wiretrustee-mgmt
+    goarch: arm
+    goarm: 6
+    use: buildx
+    dockerfile: management/Dockerfile
+    build_flag_templates:
+      - "--platform=linux/arm"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.title={{.ProjectName}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=maintainer=wiretrustee@wiretrustee.com"
   - image_templates:
       - wiretrustee/management:{{ .Version }}-debug-amd64
     ids:
@@ -174,30 +254,63 @@ dockers:
       - "--label=org.opencontainers.image.version={{.Version}}"
       - "--label=maintainer=wiretrustee@wiretrustee.com"
 
+  - image_templates:
+      - wiretrustee/management:{{ .Version }}-debug-arm
+    ids:
+      - wiretrustee-mgmt
+    goarch: arm
+    goarm: 6
+    use: buildx
+    dockerfile: management/Dockerfile.debug
+    build_flag_templates:
+      - "--platform=linux/arm"
+      - "--label=org.opencontainers.image.created={{.Date}}"
+      - "--label=org.opencontainers.image.title={{.ProjectName}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=org.opencontainers.image.revision={{.FullCommit}}"
+      - "--label=org.opencontainers.image.version={{.Version}}"
+      - "--label=maintainer=wiretrustee@wiretrustee.com"
 docker_manifests:
+  - name_template: wiretrustee/wiretrustee:{{ .Version }}
+    image_templates:
+      - wiretrustee/wiretrustee:{{ .Version }}-arm64v8
+      - wiretrustee/wiretrustee:{{ .Version }}-arm
+      - wiretrustee/wiretrustee:{{ .Version }}-amd64
+
+  - name_template: wiretrustee/wiretrustee:latest
+    image_templates:
+      - wiretrustee/wiretrustee:{{ .Version }}-arm64v8
+      - wiretrustee/wiretrustee:{{ .Version }}-arm
+      - wiretrustee/wiretrustee:{{ .Version }}-amd64
+
   - name_template: wiretrustee/signal:{{ .Version }}
     image_templates:
       - wiretrustee/signal:{{ .Version }}-arm64v8
+      - wiretrustee/signal:{{ .Version }}-arm
       - wiretrustee/signal:{{ .Version }}-amd64
 
   - name_template: wiretrustee/signal:latest
     image_templates:
       - wiretrustee/signal:{{ .Version }}-arm64v8
+      - wiretrustee/signal:{{ .Version }}-arm
       - wiretrustee/signal:{{ .Version }}-amd64
 
   - name_template: wiretrustee/management:{{ .Version }}
     image_templates:
       - wiretrustee/management:{{ .Version }}-arm64v8
+      - wiretrustee/management:{{ .Version }}-arm
       - wiretrustee/management:{{ .Version }}-amd64
 
   - name_template: wiretrustee/management:latest
     image_templates:
       - wiretrustee/management:{{ .Version }}-arm64v8
+      - wiretrustee/management:{{ .Version }}-arm
       - wiretrustee/management:{{ .Version }}-amd64
 
   - name_template: wiretrustee/management:debug-latest
     image_templates:
       - wiretrustee/management:{{ .Version }}-debug-arm64v8
+      - wiretrustee/management:{{ .Version }}-debug-arm
       - wiretrustee/management:{{ .Version }}-debug-amd64
 
 brews:
@@ -221,7 +334,7 @@ uploads:
     ids:
     - deb
     mode: archive
-    target: https://pkgs.wiretrustee.com/debian/pool/{{ .ArtifactName }};deb.distribution=stable;deb.component=main;deb.architecture={{ .Arch }}
+    target: https://pkgs.wiretrustee.com/debian/pool/{{ .ArtifactName }};deb.distribution=stable;deb.component=main;deb.architecture={{ if .Arm }}armhf{{ else }}{{ .Arch }}{{ end }};deb.package=
     username: dev@wiretrustee.com
     method: PUT
   - name: yum
@@ -230,4 +343,4 @@ uploads:
     mode: archive
     target: https://pkgs.wiretrustee.com/yum/{{ .Arch }}{{ if .Arm }}{{ .Arm }}{{ end }}
     username: dev@wiretrustee.com
-    method: PUT
+    method: PUT

README.md

@@ -15,7 +15,7 @@
 <strong>
   Start using Wiretrustee at <a href="https://app.wiretrustee.com/">app.wiretrustee.com</a>
   <br/>
-  See <a href="docs/README.md">Documentation</a>
+  See <a href="https://docs.wiretrustee.com">Documentation</a>
   <br/>
    Join our <a href="https://join.slack.com/t/wiretrustee/shared_invite/zt-vrahf41g-ik1v7fV8du6t0RwxSrJ96A">Slack channel</a>
   <br/>
@@ -29,10 +29,9 @@
 
 It requires zero configuration effort leaving behind the hassle of opening ports, complex firewall rules, vpn gateways, and so forth.
 
-There is no centralized VPN server with Wiretrustee - your computers, devices, machines, and servers connect to each other directly over a fast encrypted tunnel.
-
 **Wiretrustee automates Wireguard-based networks, offering a management layer with:**
-* Centralized Peer IP management with a neat UI dashboard.
+* Centralized Peer IP management with a UI dashboard.
+* Encrypted peer-to-peet connections without a centralized VPN gateway.
 * Automatic Peer discovery and configuration.
 * UDP hole punching to establish peer-to-peer connections behind NAT, firewall, and without a public static IP.
 * Connection relay fallback in case a peer-to-peer connection is not possible.
@@ -40,6 +39,7 @@ There is no centralized VPN server with Wiretrustee - your computers, devices, m
 * Client application SSO with MFA (coming soon).
 * Access Controls (coming soon).
 * Activity Monitoring (coming soon).
+* Private DNS (coming baoon)
 
 ### Secure peer-to-peer VPN in minutes
 <p float="left" align="middle">
@@ -145,6 +145,11 @@ For  **Windows** systems, start powershell as administrator and:
   ```shell
   wiretrustee up --setup-key <SETUP KEY>
    ```
+For **Docker**, you can run with the following command:
+```shell
+docker run --network host --privileged --rm -d -e WT_SETUP_KEY=<SETUP KEY> -v wiretrustee-client:/etc/wiretrustee wiretrustee/wiretrustee:<TAG>
+```
+> TAG > 0.3.0 version
 
 Alternatively, if you are hosting your own Management Service provide `--management-url` property pointing to your Management Service:
   ```shell

client/Dockerfile

@@ -0,0 +1,4 @@
+FROM gcr.io/distroless/base:debug
+ENV WT_LOG_FILE=console
+ENTRYPOINT [ "/go/bin/wiretrustee","up"]
+COPY wiretrustee /go/bin/wiretrustee

client/cmd/login.go

@@ -18,22 +18,21 @@ import (
 )
 
 var (
-	setupKey string
-
 	loginCmd = &cobra.Command{
 		Use:   "login",
 		Short: "login to the Wiretrustee Management Service (first run)",
 		RunE: func(cmd *cobra.Command, args []string) error {
+			SetFlagsFromEnvVars()
+
 			err := util.InitLog(logLevel, logFile)
 			if err != nil {
 				log.Errorf("failed initializing log %v", err)
 				return err
 			}
 
-			config, err := internal.GetConfig(managementURL, configPath)
+			config, err := internal.GetConfig(managementURL, configPath, preSharedKey)
 			if err != nil {
 				log.Errorf("failed getting config %s %v", configPath, err)
-				//os.Exit(ExitSetupFailed)
 				return err
 			}
 
@@ -41,7 +40,6 @@ var (
 			myPrivateKey, err := wgtypes.ParseKey(config.PrivateKey)
 			if err != nil {
 				log.Errorf("failed parsing Wireguard key %s: [%s]", config.PrivateKey, err.Error())
-				//os.Exit(ExitSetupFailed)
 				return err
 			}
 
@@ -56,7 +54,6 @@ var (
 			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)
-				//os.Exit(ExitSetupFailed)
 				return err
 			}
 			log.Debugf("connected to anagement Service %s", config.ManagementURL.String())
@@ -64,21 +61,18 @@ var (
 			serverKey, err := mgmClient.GetServerPublicKey()
 			if err != nil {
 				log.Errorf("failed while getting Management Service public key: %v", err)
-				//os.Exit(ExitSetupFailed)
 				return err
 			}
 
 			_, err = loginPeer(*serverKey, mgmClient, setupKey)
 			if err != nil {
 				log.Errorf("failed logging-in peer on Management Service : %v", err)
-				//os.Exit(ExitSetupFailed)
 				return err
 			}
 
 			err = mgmClient.Close()
 			if err != nil {
 				log.Errorf("failed closing Management Service client: %v", err)
-				//os.Exit(ExitSetupFailed)
 				return err
 			}
 
@@ -151,6 +145,3 @@ func promptPeerSetupKey() (string, error) {
 
 	return "", s.Err()
 }
-
-//func init() {
-//}

client/cmd/root.go

@@ -4,19 +4,15 @@ import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 	"github.com/wiretrustee/wiretrustee/client/internal"
 	"os"
 	"os/signal"
 	"runtime"
+	"strings"
 	"syscall"
 )
 
-const (
-	// ExitSetupFailed defines exit code
-	ExitSetupFailed   = 1
-	DefaultConfigPath = ""
-)
-
 var (
 	configPath        string
 	defaultConfigPath string
@@ -24,6 +20,8 @@ var (
 	defaultLogFile    string
 	logFile           string
 	managementURL     string
+	setupKey          string
+	preSharedKey      string
 	rootCmd           = &cobra.Command{
 		Use:   "wiretrustee",
 		Short: "",
@@ -56,6 +54,7 @@ func init() {
 	rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "info", "sets Wiretrustee log level")
 	rootCmd.PersistentFlags().StringVar(&logFile, "log-file", defaultLogFile, "sets Wiretrustee log path. If console is specified the the log will be output to stdout")
 	rootCmd.PersistentFlags().StringVar(&setupKey, "setup-key", "", "Setup key obtained from the Management Service Dashboard (used to register peer)")
+	rootCmd.PersistentFlags().StringVar(&preSharedKey, "preshared-key", "", "Sets Wireguard PreSharedKey property. If set, then only peers that have the same key can communicate.")
 	rootCmd.AddCommand(serviceCmd)
 	rootCmd.AddCommand(upCmd)
 	rootCmd.AddCommand(loginCmd)
@@ -75,3 +74,28 @@ func SetupCloseHandler() {
 		}
 	}()
 }
+
+// SetFlagsFromEnvVars reads and updates flag values from environment variables with prefix WT_
+func SetFlagsFromEnvVars() {
+	flags := rootCmd.PersistentFlags()
+	flags.VisitAll(func(f *pflag.Flag) {
+
+		envVar := FlagNameToEnvVar(f.Name)
+
+		if value, present := os.LookupEnv(envVar); present {
+			err := flags.Set(f.Name, value)
+			if err != nil {
+				log.Infof("unable to configure flag %s using variable %s, err: %v", f.Name, envVar, err)
+			}
+		}
+	})
+}
+
+// FlagNameToEnvVar converts flag name to environment var name adding a prefix,
+// replacing dashes and making all uppercase (e.g. setup-keys is converted to WT_SETUP_KEYS)
+func FlagNameToEnvVar(f string) string {
+	prefix := "WT_"
+	parsed := strings.ReplaceAll(f, "-", "_")
+	upper := strings.ToUpper(parsed)
+	return prefix + upper
+}

client/cmd/service.go

@@ -34,6 +34,3 @@ var (
 		Short: "manages wiretrustee service",
 	}
 )
-
-func init() {
-}

client/cmd/service_controller.go

@@ -1,7 +1,6 @@
 package cmd
 
 import (
-	"github.com/cenkalti/backoff/v4"
 	"github.com/kardianos/service"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -9,40 +8,21 @@ import (
 	"time"
 )
 
-func (p *program) Start(s service.Service) error {
-
-	var backOff = &backoff.ExponentialBackOff{
-		InitialInterval:     time.Second,
-		RandomizationFactor: backoff.DefaultRandomizationFactor,
-		Multiplier:          backoff.DefaultMultiplier,
-		MaxInterval:         30 * time.Second,
-		MaxElapsedTime:      24 * 3 * time.Hour, //stop after 3 days trying
-		Stop:                backoff.Stop,
-		Clock:               backoff.SystemClock,
-	}
+func (p *program) Start(service.Service) error {
 
 	// Start should not block. Do the actual work async.
 	log.Info("starting service") //nolint
 	go func() {
-		operation := func() error {
-			err := runClient()
-			if err != nil {
-				log.Warnf("retrying Wiretrustee client app due to error: %v", err)
-				return err
-			}
-			return nil
-		}
-
-		err := backoff.Retry(operation, backOff)
+		err := runClient()
 		if err != nil {
-			log.Errorf("exiting client retry loop due to unrecoverable error: %s", err)
+			log.Errorf("stopped Wiretrustee client app due to error: %v", err)
 			return
 		}
 	}()
 	return nil
 }
 
-func (p *program) Stop(s service.Service) error {
+func (p *program) Stop(service.Service) error {
 	go func() {
 		stopCh <- 1
 	}()
@@ -61,6 +41,7 @@ var (
 		Use:   "run",
 		Short: "runs wiretrustee as service",
 		Run: func(cmd *cobra.Command, args []string) {
+			SetFlagsFromEnvVars()
 
 			err := util.InitLog(logLevel, logFile)
 			if err != nil {
@@ -95,6 +76,8 @@ var (
 		Use:   "start",
 		Short: "starts wiretrustee service",
 		RunE: func(cmd *cobra.Command, args []string) error {
+			SetFlagsFromEnvVars()
+
 			err := util.InitLog(logLevel, logFile)
 			if err != nil {
 				log.Errorf("failed initializing log %v", err)
@@ -121,6 +104,8 @@ var (
 		Use:   "stop",
 		Short: "stops wiretrustee service",
 		Run: func(cmd *cobra.Command, args []string) {
+			SetFlagsFromEnvVars()
+
 			err := util.InitLog(logLevel, logFile)
 			if err != nil {
 				log.Errorf("failed initializing log %v", err)
@@ -145,6 +130,8 @@ var (
 		Use:   "restart",
 		Short: "restarts wiretrustee service",
 		Run: func(cmd *cobra.Command, args []string) {
+			SetFlagsFromEnvVars()
+
 			err := util.InitLog(logLevel, logFile)
 			if err != nil {
 				log.Errorf("failed initializing log %v", err)
@@ -163,6 +150,3 @@ var (
 		},
 	}
 )
-
-func init() {
-}

client/cmd/service_installer.go

@@ -10,6 +10,7 @@ var (
 		Use:   "install",
 		Short: "installs wiretrustee service",
 		RunE: func(cmd *cobra.Command, args []string) error {
+			SetFlagsFromEnvVars()
 
 			svcConfig := newSVCConfig()
 
@@ -49,6 +50,7 @@ var (
 		Use:   "uninstall",
 		Short: "uninstalls wiretrustee service from system",
 		Run: func(cmd *cobra.Command, args []string) {
+			SetFlagsFromEnvVars()
 
 			s, err := newSVC(&program{}, newSVCConfig())
 			if err != nil {
@@ -65,6 +67,3 @@ var (
 		},
 	}
 )
-
-func init() {
-}

client/cmd/up.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/kardianos/service"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -12,6 +13,7 @@ import (
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"time"
 )
 
 var (
@@ -19,7 +21,7 @@ var (
 		Use:   "up",
 		Short: "install, login and start wiretrustee client",
 		RunE: func(cmd *cobra.Command, args []string) error {
-
+			SetFlagsFromEnvVars()
 			err := loginCmd.RunE(cmd, args)
 			if err != nil {
 				return err
@@ -61,12 +63,22 @@ func createEngineConfig(key wgtypes.Key, config *internal.Config, peerConfig *mg
 		iFaceBlackList[config.IFaceBlackList[i]] = struct{}{}
 	}
 
-	return &internal.EngineConfig{
+	engineConf := &internal.EngineConfig{
 		WgIface:        config.WgIface,
 		WgAddr:         peerConfig.Address,
 		IFaceBlackList: iFaceBlackList,
 		WgPrivateKey:   key,
-	}, nil
+	}
+
+	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
@@ -117,86 +129,107 @@ func connectToManagement(ctx context.Context, managementAddr string, ourPrivateK
 }
 
 func runClient() error {
-	config, err := internal.ReadConfig(managementURL, configPath)
+	var 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,
+	}
+
+	operation := func() error {
+
+		config, err := internal.ReadConfig(managementURL, configPath)
+		if err != nil {
+			log.Errorf("failed reading config %s %v", configPath, err)
+			return err
+		}
+
+		//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
+		}
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		mgmTlsEnabled := false
+		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 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 err
+		}
+
+		peerConfig := loginResp.GetPeerConfig()
+
+		engineConfig, err := createEngineConfig(myPrivateKey, config, peerConfig)
+		if err != nil {
+			log.Error(err)
+			return err
+		}
+
+		// create start the Wiretrustee Engine that will connect to the Signal and Management streams and manage connections to remote peers.
+		engine := internal.NewEngine(signalClient, mgmClient, engineConfig, cancel, ctx)
+		err = engine.Start()
+		if err != nil {
+			log.Errorf("error while starting Wiretrustee Connection Engine: %s", err)
+			return err
+		}
+
+		log.Print("Wiretrustee engine started, my IP is: ", peerConfig.Address)
+
+		select {
+		case <-stopCh:
+		case <-ctx.Done():
+		}
+
+		backOff.Reset()
+
+		err = mgmClient.Close()
+		if err != nil {
+			log.Errorf("failed closing Management Service client %v", err)
+			return err
+		}
+		err = signalClient.Close()
+		if err != nil {
+			log.Errorf("failed closing Signal Service client %v", err)
+			return err
+		}
+
+		err = engine.Stop()
+		if err != nil {
+			log.Errorf("failed stopping engine %v", err)
+			return err
+		}
+
+		go func() {
+			cleanupCh <- struct{}{}
+		}()
+
+		log.Info("stopped Wiretrustee client")
+
+		return ctx.Err()
+	}
+
+	err := backoff.Retry(operation, backOff)
 	if err != nil {
-		log.Errorf("failed reading config %s %v", configPath, err)
+		log.Errorf("exiting client retry loop due to unrecoverable error: %s", err)
 		return err
 	}
-
-	//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
-	}
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	mgmTlsEnabled := false
-	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 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 err
-	}
-
-	peerConfig := loginResp.GetPeerConfig()
-
-	engineConfig, err := createEngineConfig(myPrivateKey, config, peerConfig)
-	if err != nil {
-		log.Error(err)
-		return err
-	}
-
-	// create start the Wiretrustee Engine that will connect to the Signal and Management streams and manage connections to remote peers.
-	engine := internal.NewEngine(signalClient, mgmClient, engineConfig, cancel, ctx)
-	err = engine.Start()
-	if err != nil {
-		log.Errorf("error while starting Wiretrustee Connection Engine: %s", err)
-		return err
-	}
-
-	log.Print("Wiretrustee engine started, my IP is: ", peerConfig.Address)
-
-	select {
-	case <-stopCh:
-	case <-ctx.Done():
-	}
-
-	err = mgmClient.Close()
-	if err != nil {
-		log.Errorf("failed closing Management Service client %v", err)
-		return err
-	}
-	err = signalClient.Close()
-	if err != nil {
-		log.Errorf("failed closing Signal Service client %v", err)
-		return err
-	}
-
-	err = engine.Stop()
-	if err != nil {
-		log.Errorf("failed stopping engine %v", err)
-		return err
-	}
-
-	go func() {
-		cleanupCh <- struct{}{}
-	}()
-
-	log.Info("stopped Wiretrustee client")
-
-	return ctx.Err()
-
+	return nil
 }

client/installer.nsis

@@ -106,6 +106,7 @@ SectionEnd
 Section Uninstall
 ${INSTALL_TYPE}
 
+Exec '"$INSTDIR\${MAIN_APP_EXE}" service stop'
 Exec '"$INSTDIR\${MAIN_APP_EXE}" service uninstall'
 # wait the service uninstall take unblock the executable
 Sleep 3000

client/internal/config.go

@@ -28,13 +28,14 @@ func init() {
 type Config struct {
 	// Wireguard private key of local peer
 	PrivateKey     string
+	PreSharedKey   string
 	ManagementURL  *url.URL
 	WgIface        string
 	IFaceBlackList []string
 }
 
 //createNewConfig creates a new config generating a new Wireguard key and saving to file
-func createNewConfig(managementURL string, configPath string) (*Config, error) {
+func createNewConfig(managementURL string, configPath string, preSharedKey string) (*Config, error) {
 	wgKey := generateKey()
 	config := &Config{PrivateKey: wgKey, WgIface: iface.WgInterfaceDefault, IFaceBlackList: []string{}}
 	if managementURL != "" {
@@ -47,6 +48,10 @@ func createNewConfig(managementURL string, configPath string) (*Config, error) {
 		config.ManagementURL = managementURLDefault
 	}
 
+	if preSharedKey != "" {
+		config.PreSharedKey = preSharedKey
+	}
+
 	config.IFaceBlackList = []string{iface.WgInterfaceDefault, "tun0"}
 
 	err := util.WriteJson(configPath, config)
@@ -93,11 +98,11 @@ func ReadConfig(managementURL string, configPath string) (*Config, error) {
 }
 
 // GetConfig reads existing config or generates a new one
-func GetConfig(managementURL string, configPath string) (*Config, error) {
+func GetConfig(managementURL string, configPath string, preSharedKey string) (*Config, error) {
 
 	if _, err := os.Stat(configPath); os.IsNotExist(err) {
 		log.Infof("generating new config %s", configPath)
-		return createNewConfig(managementURL, configPath)
+		return createNewConfig(managementURL, configPath, preSharedKey)
 	} else {
 		return ReadConfig(managementURL, configPath)
 	}

client/internal/connection.go

@@ -60,6 +60,8 @@ type ConnConfig struct {
 	// Remote Wireguard public key
 	RemoteWgKey wgtypes.Key
 
+	PreSharedKey *wgtypes.Key
+
 	StunTurnURLS []*ice.URL
 
 	iFaceBlackList map[string]struct{}
@@ -115,7 +117,7 @@ func NewConnection(config ConnConfig,
 		closeCond:         NewCond(),
 		connected:         NewCond(),
 		agent:             nil,
-		wgProxy:           NewWgProxy(config.WgIface, config.RemoteWgKey.String(), config.WgAllowedIPs, config.WgListenAddr),
+		wgProxy:           NewWgProxy(config.WgIface, config.RemoteWgKey.String(), config.WgAllowedIPs, config.WgListenAddr, config.PreSharedKey),
 		Status:            StatusDisconnected,
 	}
 }
@@ -138,12 +140,18 @@ func (conn *Connection) Open(timeout time.Duration) error {
 			return !ok
 		},
 	})
-	conn.agent = a
-
 	if err != nil {
 		return err
 	}
 
+	conn.agent = a
+	defer func() {
+		err := conn.agent.Close()
+		if err != nil {
+			return
+		}
+	}()
+
 	err = conn.listenOnLocalCandidates()
 	if err != nil {
 		return err

client/internal/engine.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/cenkalti/backoff/v4"
-	ice "github.com/pion/ice/v2"
+	"github.com/pion/ice/v2"
 	log "github.com/sirupsen/logrus"
 	"github.com/wiretrustee/wiretrustee/iface"
 	mgm "github.com/wiretrustee/wiretrustee/management/client"
@@ -30,6 +30,8 @@ type EngineConfig struct {
 	WgPrivateKey wgtypes.Key
 	// IFaceBlackList is a list of network interfaces to ignore when discovering connection candidates (ICE related)
 	IFaceBlackList map[string]struct{}
+
+	PreSharedKey *wgtypes.Key
 }
 
 // Engine is a mechanism responsible for reacting on Signal and Management stream events and managing connections to the remote peers.
@@ -142,12 +144,17 @@ func (e *Engine) initializePeer(peer Peer) {
 		RandomizationFactor: backoff.DefaultRandomizationFactor,
 		Multiplier:          backoff.DefaultMultiplier,
 		MaxInterval:         5 * time.Second,
-		MaxElapsedTime:      time.Duration(0), //never stop
+		MaxElapsedTime:      0, //never stop
 		Stop:                backoff.Stop,
 		Clock:               backoff.SystemClock,
 	}, e.ctx)
 
 	operation := func() error {
+
+		if e.signal.GetStatus() != signal.StreamConnected {
+			return fmt.Errorf("not opening connection to peer because Signal is unavailable")
+		}
+
 		_, err := e.openPeerConnection(e.wgPort, e.config.WgPrivateKey, peer)
 		e.peerMux.Lock()
 		defer e.peerMux.Unlock()
@@ -157,7 +164,6 @@ func (e *Engine) initializePeer(peer Peer) {
 		}
 
 		if err != nil {
-			log.Warnln(err)
 			log.Debugf("retrying connection because of error: %s", err.Error())
 			return err
 		}
@@ -234,6 +240,7 @@ func (e *Engine) openPeerConnection(wgPort int, myKey wgtypes.Key, peer Peer) (*
 		RemoteWgKey:    remoteKey,
 		StunTurnURLS:   append(e.STUNs, e.TURNs...),
 		iFaceBlackList: e.config.IFaceBlackList,
+		PreSharedKey:   e.config.PreSharedKey,
 	}
 
 	signalOffer := func(uFrag string, pwd string) error {
@@ -332,6 +339,8 @@ func (e *Engine) receiveManagementEvents() {
 			return nil
 		})
 		if err != nil {
+			// happens if management is unavailable for a long time.
+			// We want to cancel the operation of the whole client
 			e.cancel()
 			return
 		}
@@ -414,68 +423,77 @@ func (e *Engine) updatePeers(remotePeers []*mgmProto.RemotePeerConfig) error {
 
 // receiveSignalEvents connects to the Signal Service event stream to negotiate connection with remote peers
 func (e *Engine) receiveSignalEvents() {
-	// connect to a stream of messages coming from the signal server
-	e.signal.Receive(func(msg *sProto.Message) error {
 
-		e.syncMsgMux.Lock()
-		defer e.syncMsgMux.Unlock()
+	go func() {
+		// connect to a stream of messages coming from the signal server
+		err := e.signal.Receive(func(msg *sProto.Message) error {
 
-		conn := e.conns[msg.Key]
-		if conn == nil {
-			return fmt.Errorf("wrongly addressed message %s", msg.Key)
-		}
+			e.syncMsgMux.Lock()
+			defer e.syncMsgMux.Unlock()
 
-		if conn.Config.RemoteWgKey.String() != msg.Key {
-			return fmt.Errorf("unknown peer %s", msg.Key)
-		}
-
-		switch msg.GetBody().Type {
-		case sProto.Body_OFFER:
-			remoteCred, err := signal.UnMarshalCredential(msg)
-			if err != nil {
-				return err
+			conn := e.conns[msg.Key]
+			if conn == nil {
+				return fmt.Errorf("wrongly addressed message %s", msg.Key)
 			}
-			err = conn.OnOffer(IceCredentials{
-				uFrag: remoteCred.UFrag,
-				pwd:   remoteCred.Pwd,
-			})
 
-			if err != nil {
-				return err
+			if conn.Config.RemoteWgKey.String() != msg.Key {
+				return fmt.Errorf("unknown peer %s", msg.Key)
+			}
+
+			switch msg.GetBody().Type {
+			case sProto.Body_OFFER:
+				remoteCred, err := signal.UnMarshalCredential(msg)
+				if err != nil {
+					return err
+				}
+				err = conn.OnOffer(IceCredentials{
+					uFrag: remoteCred.UFrag,
+					pwd:   remoteCred.Pwd,
+				})
+
+				if err != nil {
+					return err
+				}
+
+				return nil
+			case sProto.Body_ANSWER:
+				remoteCred, err := signal.UnMarshalCredential(msg)
+				if err != nil {
+					return err
+				}
+				err = conn.OnAnswer(IceCredentials{
+					uFrag: remoteCred.UFrag,
+					pwd:   remoteCred.Pwd,
+				})
+
+				if err != nil {
+					return err
+				}
+
+			case sProto.Body_CANDIDATE:
+
+				candidate, err := ice.UnmarshalCandidate(msg.GetBody().Payload)
+				if err != nil {
+					log.Errorf("failed on parsing remote candidate %s -> %s", candidate, err)
+					return err
+				}
+
+				err = conn.OnRemoteCandidate(candidate)
+				if err != nil {
+					log.Errorf("error handling CANDIATE from %s", msg.Key)
+					return err
+				}
 			}
 
 			return nil
-		case sProto.Body_ANSWER:
-			remoteCred, err := signal.UnMarshalCredential(msg)
-			if err != nil {
-				return err
-			}
-			err = conn.OnAnswer(IceCredentials{
-				uFrag: remoteCred.UFrag,
-				pwd:   remoteCred.Pwd,
-			})
-
-			if err != nil {
-				return err
-			}
-
-		case sProto.Body_CANDIDATE:
-
-			candidate, err := ice.UnmarshalCandidate(msg.GetBody().Payload)
-			if err != nil {
-				log.Errorf("failed on parsing remote candidate %s -> %s", candidate, err)
-				return err
-			}
-
-			err = conn.OnRemoteCandidate(candidate)
-			if err != nil {
-				log.Errorf("error handling CANDIATE from %s", msg.Key)
-				return err
-			}
+		})
+		if err != nil {
+			// happens if signal is unavailable for a long time.
+			// We want to cancel the operation of the whole client
+			e.cancel()
+			return
 		}
+	}()
 
-		return nil
-	})
-
-	e.signal.WaitConnected()
+	e.signal.WaitStreamConnected()
 }

client/internal/wgproxy.go

@@ -4,27 +4,30 @@ import (
 	ice "github.com/pion/ice/v2"
 	log "github.com/sirupsen/logrus"
 	"github.com/wiretrustee/wiretrustee/iface"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"net"
 )
 
 // WgProxy an instance of an instance of the Connection Wireguard Proxy
 type WgProxy struct {
-	iface      string
-	remoteKey  string
-	allowedIps string
-	wgAddr     string
-	close      chan struct{}
-	wgConn     net.Conn
+	iface        string
+	remoteKey    string
+	allowedIps   string
+	wgAddr       string
+	close        chan struct{}
+	wgConn       net.Conn
+	preSharedKey *wgtypes.Key
 }
 
 // NewWgProxy creates a new Connection Wireguard Proxy
-func NewWgProxy(iface string, remoteKey string, allowedIps string, wgAddr string) *WgProxy {
+func NewWgProxy(iface string, remoteKey string, allowedIps string, wgAddr string, preSharedKey *wgtypes.Key) *WgProxy {
 	return &WgProxy{
-		iface:      iface,
-		remoteKey:  remoteKey,
-		allowedIps: allowedIps,
-		wgAddr:     wgAddr,
-		close:      make(chan struct{}),
+		iface:        iface,
+		remoteKey:    remoteKey,
+		allowedIps:   allowedIps,
+		wgAddr:       wgAddr,
+		close:        make(chan struct{}),
+		preSharedKey: preSharedKey,
 	}
 }
 
@@ -48,7 +51,7 @@ func (p *WgProxy) Close() error {
 
 // 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)
+	err := iface.UpdatePeer(p.iface, p.remoteKey, p.allowedIps, DefaultWgKeepAlive, host, p.preSharedKey)
 	if err != nil {
 		log.Errorf("error while configuring Wireguard peer [%s] %s", p.remoteKey, err.Error())
 		return err
@@ -67,7 +70,7 @@ func (p *WgProxy) Start(remoteConn *ice.Conn) error {
 	p.wgConn = wgConn
 	// add local proxy connection as a Wireguard peer
 	err = iface.UpdatePeer(p.iface, p.remoteKey, p.allowedIps, DefaultWgKeepAlive,
-		wgConn.LocalAddr().String())
+		wgConn.LocalAddr().String(), p.preSharedKey)
 	if err != nil {
 		log.Errorf("error while configuring Wireguard peer [%s] %s", p.remoteKey, err.Error())
 		return err
@@ -92,13 +95,11 @@ func (p *WgProxy) proxyToRemotePeer(remoteConn *ice.Conn) {
 		default:
 			n, err := p.wgConn.Read(buf)
 			if err != nil {
-				//log.Warnln("failed reading from peer: ", err.Error())
 				continue
 			}
 
 			_, err = remoteConn.Write(buf[:n])
 			if err != nil {
-				//log.Warnln("failed writing to remote peer: ", err.Error())
 				continue
 			}
 		}
@@ -118,13 +119,11 @@ func (p *WgProxy) proxyToLocalWireguard(remoteConn *ice.Conn) {
 		default:
 			n, err := remoteConn.Read(buf)
 			if err != nil {
-				//log.Errorf("failed reading from remote connection %s", err)
 				continue
 			}
 
 			_, err = p.wgConn.Write(buf[:n])
 			if err != nil {
-				//log.Errorf("failed writing to local Wireguard instance %s", err)
 				continue
 			}
 		}

ebpf/filter.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"github.com/cilium/ebpf"
+	"net"
+
+	"syscall"
+)
+
+// Filter represents a classic BPF filter program that can be applied to a socket
+type Filter struct {
+	*ebpf.ProgramSpec
+}
+
+// ApplyTo applies the current filter onto the provided UDPConn
+func (filter Filter) ApplyTo(conn *net.UDPConn) error {
+
+	file, err := conn.File()
+	if err != nil {
+		return err
+	}
+
+	p, err := ebpf.NewProgramWithOptions(filter.ProgramSpec, ebpf.ProgramOptions{
+		LogLevel: 6,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	if err := syscall.SetsockoptInt(int(file.Fd()), syscall.SOL_SOCKET, SO_ATTACH_BPF, p.FD()); err != nil {
+		return err
+	}
+
+	return nil
+}

ebpf/main.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"encoding/hex"
+	"fmt"
+	"github.com/cilium/ebpf"
+	"github.com/cilium/ebpf/asm"
+	"net"
+)
+
+const (
+	SO_ATTACH_BPF   int    = 50
+	StunMagicCookie uint32 = 0x2112A442
+)
+
+func main() {
+
+	// open a raw socket
+	/*fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_UDP)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Print(fd)*/
+	addr := net.UDPAddr{
+		IP:   net.IPv4zero,
+		Port: 12345,
+	}
+	conn, err := net.ListenUDP("udp4", &addr)
+	if err != nil {
+		return
+	}
+
+	filter := &Filter{
+		ProgramSpec: &ebpf.ProgramSpec{
+			Type:    ebpf.SocketFilter,
+			License: "GPL",
+			Instructions: asm.Instructions{
+				asm.Mov.Reg(asm.R6, asm.R1), // LDABS requires ctx in R6
+				asm.LoadAbs(-0x100000+22, asm.Half),
+				asm.JNE.Imm(asm.R0, int32(addr.Port), "skip"),
+				/*				asm.LoadAbs(-0x100000+32, asm.Word),
+								asm.JNE.Imm(asm.R0, int32(StunMagicCookie), "skip"),
+								asm.Mov.Imm(asm.R0, -1).Sym("exit"),*/
+				/*asm.Return(),*/
+				asm.Mov.Imm(asm.R0, 0).Sym("skip"),
+				asm.Return(),
+			},
+		},
+	}
+
+	err = filter.ApplyTo(conn)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Printf("start")
+	buf := make([]byte, 1024)
+	for {
+		n, ra, err := conn.ReadFrom(buf)
+		if err != nil {
+			panic(err)
+		}
+
+		fmt.Printf("Bytes read: %d\n", n)
+		fmt.Printf("Remote Addr: %+v\n", ra)
+		fmt.Printf("Bytes HEX: %s\n", hex.EncodeToString(buf[:n]))
+		fmt.Printf("Bytes String: %s\n", string(buf[:n]))
+		fmt.Println()
+	}
+
+}

ebpf/test/main.go

@@ -0,0 +1,186 @@
+package main
+
+// This code is derived from https://github.com/cloudflare/cloudflare-blog/tree/master/2018-03-ebpf
+//
+// Copyright (c) 2015-2017 Cloudflare, Inc. All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//    * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//    * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//    * Neither the name of the Cloudflare, Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import (
+	"fmt"
+	"net"
+	"syscall"
+
+	"github.com/cilium/ebpf"
+	"github.com/cilium/ebpf/asm"
+)
+
+// ExampleExtractDistance shows how to attach an eBPF socket filter to
+// extract the network distance of an IP host.
+func main() {
+	filter, TTLs, err := newDistanceFilter()
+	if err != nil {
+		panic(err)
+	}
+	defer filter.Close()
+	defer TTLs.Close()
+
+	// Attach filter before the call to connect()
+	dialer := net.Dialer{
+		Control: func(network, address string, c syscall.RawConn) (err error) {
+			const SO_ATTACH_BPF = 50
+
+			err = c.Control(func(fd uintptr) {
+				err = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, SO_ATTACH_BPF, filter.FD())
+			})
+			return err
+		},
+	}
+
+	conn, err := dialer.Dial("tcp", "1.1.1.1:53")
+	if err != nil {
+		panic(err)
+	}
+	conn.Close()
+
+	minDist, err := minDistance(TTLs)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("1.1.1.1:53 is", minDist, "hops away")
+}
+
+func newDistanceFilter() (*ebpf.Program, *ebpf.Map, error) {
+	const ETH_P_IPV6 uint16 = 0x86DD
+
+	ttls, err := ebpf.NewMap(&ebpf.MapSpec{
+		Type:       ebpf.Hash,
+		KeySize:    4,
+		ValueSize:  8,
+		MaxEntries: 4,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+
+	insns := asm.Instructions{
+		// r1 has ctx
+		// r0 = ctx[16] (aka protocol)
+		asm.LoadMem(asm.R0, asm.R1, 16, asm.Word),
+
+		// Perhaps ipv6
+		asm.LoadImm(asm.R2, int64(ETH_P_IPV6), asm.DWord),
+		asm.HostTo(asm.BE, asm.R2, asm.Half),
+		asm.JEq.Reg(asm.R0, asm.R2, "ipv6"),
+
+		// otherwise assume ipv4
+		// 8th byte in IPv4 is TTL
+		// LDABS requires ctx in R6
+		asm.Mov.Reg(asm.R6, asm.R1),
+		asm.LoadAbs(-0x100000+8, asm.Byte),
+		asm.Ja.Label("store-ttl"),
+
+		// 7th byte in IPv6 is Hop count
+		// LDABS requires ctx in R6
+		asm.Mov.Reg(asm.R6, asm.R1).Sym("ipv6"),
+		asm.LoadAbs(-0x100000+7, asm.Byte),
+
+		// stash the load result into FP[-4]
+		asm.StoreMem(asm.RFP, -4, asm.R0, asm.Word).Sym("store-ttl"),
+		// stash the &FP[-4] into r2
+		asm.Mov.Reg(asm.R2, asm.RFP),
+		asm.Add.Imm(asm.R2, -4),
+
+		// r1 must point to map
+		asm.LoadMapPtr(asm.R1, ttls.FD()),
+		asm.FnMapLookupElem.Call(),
+
+		// load ok? inc. Otherwise? jmp to mapupdate
+		asm.JEq.Imm(asm.R0, 0, "update-map"),
+		asm.Mov.Imm(asm.R1, 1),
+		asm.StoreXAdd(asm.R0, asm.R1, asm.DWord),
+		asm.Ja.Label("exit"),
+
+		// MapUpdate
+		// r1 has map ptr
+		asm.LoadMapPtr(asm.R1, ttls.FD()).Sym("update-map"),
+		// r2 has key -> &FP[-4]
+		asm.Mov.Reg(asm.R2, asm.RFP),
+		asm.Add.Imm(asm.R2, -4),
+		// r3 has value -> &FP[-16] , aka 1
+		asm.StoreImm(asm.RFP, -16, 1, asm.DWord),
+		asm.Mov.Reg(asm.R3, asm.RFP),
+		asm.Add.Imm(asm.R3, -16),
+		// r4 has flags, 0
+		asm.Mov.Imm(asm.R4, 0),
+		asm.FnMapUpdateElem.Call(),
+
+		// set exit code to -1, don't trunc packet
+		asm.Mov.Imm(asm.R0, -1).Sym("exit"),
+		asm.Return(),
+	}
+
+	prog, err := ebpf.NewProgram(&ebpf.ProgramSpec{
+		Name:         "distance_filter",
+		Type:         ebpf.SocketFilter,
+		License:      "GPL",
+		Instructions: insns,
+	})
+	if err != nil {
+		ttls.Close()
+		return nil, nil, err
+	}
+
+	return prog, ttls, nil
+}
+
+func minDistance(TTLs *ebpf.Map) (int, error) {
+	var (
+		entries = TTLs.Iterate()
+		ttl     uint32
+		minDist uint32 = 255
+		count   uint64
+	)
+	for entries.Next(&ttl, &count) {
+		var dist uint32
+		switch {
+		case ttl > 128:
+			dist = 255 - ttl
+		case ttl > 64:
+			dist = 128 - ttl
+		case ttl > 32:
+			dist = 64 - ttl
+		default:
+			dist = 32 - ttl
+		}
+		if minDist > dist {
+			minDist = dist
+		}
+	}
+	return int(minDist), entries.Err()
+}

go.mod

@@ -4,8 +4,10 @@ go 1.16
 
 require (
 	github.com/cenkalti/backoff/v4 v4.1.0
+	github.com/cilium/ebpf v0.7.0
 	github.com/golang-jwt/jwt v3.2.2+incompatible
 	github.com/golang/protobuf v1.5.2
+	github.com/google/gopacket v1.1.19
 	github.com/google/uuid v1.2.0
 	github.com/gorilla/mux v1.8.0
 	github.com/kardianos/service v1.2.1-0.20210728001519-a323c3813bc7
@@ -15,9 +17,11 @@ require (
 	github.com/rs/cors v1.8.0
 	github.com/sirupsen/logrus v1.7.0
 	github.com/spf13/cobra v1.1.3
+	github.com/spf13/pflag v1.0.5
 	github.com/vishvananda/netlink v1.1.0
 	golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97
-	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c
+	golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985
+	golang.org/x/sys v0.0.0-20210906170528-6f6e22806c34
 	golang.zx2c4.com/wireguard v0.0.0-20210805125648-3957e9b9dd19
 	golang.zx2c4.com/wireguard/wgctrl v0.0.0-20210803171230-4253848d036c
 	golang.zx2c4.com/wireguard/windows v0.4.5

go.sum

@@ -28,6 +28,8 @@ github.com/cenkalti/backoff/v4 v4.1.0 h1:c8LkOFQTzuO0WBM/ae5HdGQuZPfPxp7lqBRwQRm
 github.com/cenkalti/backoff/v4 v4.1.0/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
+github.com/cilium/ebpf v0.7.0 h1:1k/q3ATgxSXRdrmPfH8d7YK0GfqVsEKZAX9dQZvs56k=
+github.com/cilium/ebpf v0.7.0/go.mod h1:/oI2+1shJiTGAMgl6/RgJr36Eo1jzrRcAWbcXO2usCA=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
@@ -45,6 +47,7 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/frankban/quicktest v1.11.3/go.mod h1:wRf/ReqHper53s+kmmSZizM8NamnL3IM0I9ntUbOk+k=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
@@ -92,6 +95,8 @@ github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
+github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/google/pprof v0.0.0-20190515194954-54271f7e092f/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
@@ -154,6 +159,7 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxv
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
@@ -320,10 +326,12 @@ golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHl
 golang.org/x/lint v0.0.0-20190409202823-959b441ac422/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/lint v0.0.0-20190909230951-414d861bb4ac/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
 golang.org/x/mobile v0.0.0-20190719004257-d2bd2a29d028/go.mod h1:E/iHnbuqvinMTCcRqshq8CkpyQDoeVncDDYHnLhea+o=
 golang.org/x/mod v0.0.0-20190513183733-4bf6d317e70e/go.mod h1:mXi4GBBbnImb6dmsKGUJ2LatrhH/nqhxcFungHvyanc=
 golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -412,6 +420,8 @@ golang.org/x/sys v0.0.0-20210503173754-0981d6026fa6/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c h1:F1jZWGFhYfh0Ci55sIpILtKKK8p3i2/krTr0H1rg74I=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210906170528-6f6e22806c34 h1:GkvMjFtXUmahfDtashnc1mnrCtuBVcwse5QV2lUk/tI=
+golang.org/x/sys v0.0.0-20210906170528-6f6e22806c34/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -442,6 +452,7 @@ golang.org/x/tools v0.0.0-20190911174233-4f2ddba30aff/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

iface/iface.go

@@ -146,7 +146,7 @@ func GetListenPort(iface string) (*int, error) {
 
 // UpdatePeer updates existing Wireguard Peer or creates a new one if doesn't exist
 // Endpoint is optional
-func UpdatePeer(iface string, peerKey string, allowedIps string, keepAlive time.Duration, endpoint string) error {
+func UpdatePeer(iface string, peerKey string, allowedIps string, keepAlive time.Duration, endpoint string, preSharedKey *wgtypes.Key) error {
 
 	log.Debugf("updating interface %s peer %s: endpoint %s ", iface, peerKey, endpoint)
 
@@ -165,6 +165,7 @@ func UpdatePeer(iface string, peerKey string, allowedIps string, keepAlive time.
 		ReplaceAllowedIPs:           true,
 		AllowedIPs:                  []net.IPNet{*ipNet},
 		PersistentKeepaliveInterval: &keepAlive,
+		PresharedKey:                preSharedKey,
 	}
 
 	config := wgtypes.Config{

iface/iface_test.go

@@ -111,7 +111,7 @@ func Test_UpdatePeer(t *testing.T) {
 	keepAlive := 15 * time.Second
 	allowedIP := "10.99.99.2/32"
 	endpoint := "127.0.0.1:9900"
-	err = UpdatePeer(ifaceName, peerPubKey, allowedIP, keepAlive, endpoint)
+	err = UpdatePeer(ifaceName, peerPubKey, allowedIP, keepAlive, endpoint, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -163,7 +163,7 @@ func Test_UpdatePeerEndpoint(t *testing.T) {
 	keepAlive := 15 * time.Second
 	allowedIP := "10.99.99.2/32"
 	endpoint := "127.0.0.1:9900"
-	err = UpdatePeer(ifaceName, peerPubKey, allowedIP, keepAlive, endpoint)
+	err = UpdatePeer(ifaceName, peerPubKey, allowedIP, keepAlive, endpoint, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -204,7 +204,7 @@ func Test_RemovePeer(t *testing.T) {
 	keepAlive := 15 * time.Second
 	allowedIP := "10.99.99.2/32"
 	endpoint := "127.0.0.1:9900"
-	err = UpdatePeer(ifaceName, peerPubKey, allowedIP, keepAlive, endpoint)
+	err = UpdatePeer(ifaceName, peerPubKey, allowedIP, keepAlive, endpoint, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

management/client/client.go

@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"github.com/cenkalti/backoff/v4"
 	log "github.com/sirupsen/logrus"
 	"github.com/wiretrustee/wiretrustee/client/system"
@@ -10,6 +11,7 @@ import (
 	"github.com/wiretrustee/wiretrustee/management/proto"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
 	"io"
@@ -70,13 +72,19 @@ func defaultBackoff(ctx context.Context) backoff.BackOff {
 		InitialInterval:     800 * time.Millisecond,
 		RandomizationFactor: backoff.DefaultRandomizationFactor,
 		Multiplier:          backoff.DefaultMultiplier,
-		MaxInterval:         30 * time.Second,
-		MaxElapsedTime:      24 * 3 * time.Hour, //stop after 3 days trying
+		MaxInterval:         10 * time.Second,
+		MaxElapsedTime:      12 * time.Hour, //stop after 12 hours of trying, the error will be propagated to the general retry of the client
 		Stop:                backoff.Stop,
 		Clock:               backoff.SystemClock,
 	}, ctx)
 }
 
+// ready indicates whether the client is okay and ready to be used
+// for now it just checks whether gRPC connection to the service is ready
+func (c *Client) ready() bool {
+	return c.conn.GetState() == connectivity.Ready
+}
+
 // Sync wraps the real client's Sync endpoint call and takes care of retries and encryption/decryption of messages
 // Blocking request. The result will be sent via msgHandler callback function
 func (c *Client) Sync(msgHandler func(msg *proto.SyncResponse) error) error {
@@ -85,6 +93,12 @@ func (c *Client) Sync(msgHandler func(msg *proto.SyncResponse) error) error {
 
 	operation := func() error {
 
+		log.Debugf("management connection state %v", c.conn.GetState())
+
+		if !c.ready() {
+			return fmt.Errorf("no connection to management")
+		}
+
 		// todo we already have it since we did the Login, maybe cache it locally?
 		serverPubKey, err := c.GetServerPublicKey()
 		if err != nil {
@@ -98,17 +112,15 @@ func (c *Client) Sync(msgHandler func(msg *proto.SyncResponse) error) error {
 			return err
 		}
 
-		log.Infof("connected to the Management Service Stream")
+		log.Infof("connected to the Management Service stream")
 
 		// blocking until error
 		err = c.receiveEvents(stream, *serverPubKey, msgHandler)
 		if err != nil {
-			/*if errStatus, ok := status.FromError(err); ok && errStatus.Code() == codes.PermissionDenied {
-				//todo handle differently??
-			}*/
+			backOff.Reset()
 			return err
 		}
-		backOff.Reset()
+
 		return nil
 	}
 
@@ -141,7 +153,7 @@ func (c *Client) receiveEvents(stream proto.ManagementService_SyncClient, server
 	for {
 		update, err := stream.Recv()
 		if err == io.EOF {
-			log.Errorf("managment stream was closed: %s", err)
+			log.Errorf("Management stream has been closed by server: %s", err)
 			return err
 		}
 		if err != nil {
@@ -167,6 +179,10 @@ func (c *Client) receiveEvents(stream proto.ManagementService_SyncClient, server
 
 // GetServerPublicKey returns server Wireguard public key (used later for encrypting messages sent to the server)
 func (c *Client) GetServerPublicKey() (*wgtypes.Key, error) {
+	if !c.ready() {
+		return nil, fmt.Errorf("no connection to management")
+	}
+
 	mgmCtx, cancel := context.WithTimeout(c.ctx, 5*time.Second) //todo make a general setting
 	defer cancel()
 	resp, err := c.realClient.GetServerKey(mgmCtx, &proto.Empty{})
@@ -183,6 +199,9 @@ func (c *Client) GetServerPublicKey() (*wgtypes.Key, error) {
 }
 
 func (c *Client) login(serverKey wgtypes.Key, req *proto.LoginRequest) (*proto.LoginResponse, error) {
+	if !c.ready() {
+		return nil, fmt.Errorf("no connection to management")
+	}
 	loginReq, err := encryption.EncryptMessage(serverKey, c.key, req)
 	if err != nil {
 		log.Errorf("failed to encrypt message: %s", err)

signal/client/client.go

@@ -11,6 +11,7 @@ import (
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
@@ -23,6 +24,12 @@ import (
 
 // A set of tools to exchange connection details (Wireguard endpoints) with the remote peer.
 
+// Status is the status of the client
+type Status string
+
+const StreamConnected Status = "Connected"
+const StreamDisconnected Status = "Disconnected"
+
 // Client Wraps the Signal Exchange Service gRpc client
 type Client struct {
 	key        wgtypes.Key
@@ -30,8 +37,15 @@ type Client struct {
 	signalConn *grpc.ClientConn
 	ctx        context.Context
 	stream     proto.SignalExchange_ConnectStreamClient
-	//waiting group to notify once stream is connected
-	connWg *sync.WaitGroup //todo use a channel instead??
+	// connectedCh used to notify goroutines waiting for the connection to the Signal stream
+	connectedCh chan struct{}
+	mux         sync.Mutex
+	// StreamConnected indicates whether this client is StreamConnected to the Signal stream
+	status Status
+}
+
+func (c *Client) GetStatus() Status {
+	return c.status
 }
 
 // Close Closes underlying connections to the Signal Exchange
@@ -65,13 +79,13 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key, tlsEnabled boo
 		return nil, err
 	}
 
-	var wg sync.WaitGroup
 	return &Client{
 		realClient: proto.NewSignalExchangeClient(conn),
 		ctx:        ctx,
 		signalConn: conn,
 		key:        key,
-		connWg:     &wg,
+		mux:        sync.Mutex{},
+		status:     StreamDisconnected,
 	}, nil
 }
 
@@ -81,8 +95,8 @@ func defaultBackoff(ctx context.Context) backoff.BackOff {
 		InitialInterval:     800 * time.Millisecond,
 		RandomizationFactor: backoff.DefaultRandomizationFactor,
 		Multiplier:          backoff.DefaultMultiplier,
-		MaxInterval:         30 * time.Second,
-		MaxElapsedTime:      24 * 3 * time.Hour, //stop after 3 days trying
+		MaxInterval:         10 * time.Second,
+		MaxElapsedTime:      12 * time.Hour, //stop after 12 hours of trying, the error will be propagated to the general retry of the client
 		Stop:                backoff.Stop,
 		Clock:               backoff.SystemClock,
 	}, ctx)
@@ -91,36 +105,79 @@ func defaultBackoff(ctx context.Context) backoff.BackOff {
 
 // Receive Connects to the Signal Exchange message stream and starts receiving messages.
 // The messages will be handled by msgHandler function provided.
-// This function runs a goroutine underneath and reconnects to the Signal Exchange if errors occur (e.g. Exchange restart)
-// The key is the identifier of our Peer (could be Wireguard public key)
-func (c *Client) Receive(msgHandler func(msg *proto.Message) error) {
-	c.connWg.Add(1)
-	go func() {
+// This function is blocking and reconnects to the Signal Exchange if errors occur (e.g. Exchange restart)
+// The connection retry logic will try to reconnect for 30 min and if wasn't successful will propagate the error to the function caller.
+func (c *Client) Receive(msgHandler func(msg *proto.Message) error) error {
 
-		var backOff = defaultBackoff(c.ctx)
+	var backOff = defaultBackoff(c.ctx)
 
-		operation := func() error {
+	operation := func() error {
 
-			err := c.connect(c.key.PublicKey().String(), msgHandler)
-			if err != nil {
-				log.Warnf("disconnected from the Signal Exchange due to an error: %v", err)
-				c.connWg.Add(1)
-				return err
-			}
+		c.notifyStreamDisconnected()
 
-			backOff.Reset()
-			return nil
+		log.Debugf("signal connection state %v", c.signalConn.GetState())
+		if !c.ready() {
+			return fmt.Errorf("no connection to signal")
 		}
 
-		err := backoff.Retry(operation, backOff)
+		// connect to Signal stream identifying ourselves with a public Wireguard key
+		// todo once the key rotation logic has been implemented, consider changing to some other identifier (received from management)
+		stream, err := c.connect(c.key.PublicKey().String())
 		if err != nil {
-			log.Errorf("exiting Signal Service connection retry loop due to unrecoverable error: %s", err)
-			return
+			log.Warnf("disconnected from the Signal Exchange due to an error: %v", err)
+			return err
 		}
-	}()
+
+		c.notifyStreamConnected()
+
+		log.Infof("connected to the Signal Service stream")
+
+		// start receiving messages from the Signal stream (from other peers through signal)
+		err = c.receive(stream, msgHandler)
+		if err != nil {
+			log.Warnf("disconnected from the Signal Exchange due to an error: %v", err)
+			backOff.Reset()
+			return err
+		}
+
+		return nil
+	}
+
+	err := backoff.Retry(operation, backOff)
+	if err != nil {
+		log.Errorf("exiting Signal Service connection retry loop due to unrecoverable error: %s", err)
+		return err
+	}
+
+	return nil
+}
+func (c *Client) notifyStreamDisconnected() {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+	c.status = StreamDisconnected
 }
 
-func (c *Client) connect(key string, msgHandler func(msg *proto.Message) error) error {
+func (c *Client) notifyStreamConnected() {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+	c.status = StreamConnected
+	if c.connectedCh != nil {
+		// there are goroutines waiting on this channel -> release them
+		close(c.connectedCh)
+		c.connectedCh = nil
+	}
+}
+
+func (c *Client) getStreamStatusChan() <-chan struct{} {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+	if c.connectedCh == nil {
+		c.connectedCh = make(chan struct{})
+	}
+	return c.connectedCh
+}
+
+func (c *Client) connect(key string) (proto.SignalExchange_ConnectStreamClient, error) {
 	c.stream = nil
 
 	// add key fingerprint to the request header to be identified on the server side
@@ -131,35 +188,48 @@ func (c *Client) connect(key string, msgHandler func(msg *proto.Message) error)
 
 	c.stream = stream
 	if err != nil {
-		return err
+		return nil, err
 	}
 	// blocks
 	header, err := c.stream.Header()
 	if err != nil {
-		return err
+		return nil, err
 	}
 	registered := header.Get(proto.HeaderRegistered)
 	if len(registered) == 0 {
-		return fmt.Errorf("didn't receive a registration header from the Signal server whille connecting to the streams")
+		return nil, fmt.Errorf("didn't receive a registration header from the Signal server whille connecting to the streams")
 	}
-	//connection established we are good to use the stream
-	c.connWg.Done()
 
-	log.Infof("connected to the Signal Exchange Stream")
-
-	return c.receive(stream, msgHandler)
+	return stream, nil
 }
 
-// WaitConnected waits until the client is connected to the message stream
-func (c *Client) WaitConnected() {
-	c.connWg.Wait()
+// ready indicates whether the client is okay and ready to be used
+// for now it just checks whether gRPC connection to the service is in state Ready
+func (c *Client) ready() bool {
+	return c.signalConn.GetState() == connectivity.Ready
+}
+
+// WaitStreamConnected waits until the client is connected to the Signal stream
+func (c *Client) WaitStreamConnected() {
+
+	if c.status == StreamConnected {
+		return
+	}
+
+	ch := c.getStreamStatusChan()
+	select {
+	case <-c.ctx.Done():
+	case <-ch:
+	}
 }
 
 // SendToStream sends a message to the remote Peer through the Signal Exchange using established stream connection to the Signal Server
 // The Client.Receive method must be called before sending messages to establish initial connection to the Signal Exchange
 // Client.connWg can be used to wait
 func (c *Client) SendToStream(msg *proto.EncryptedMessage) error {
-
+	if !c.ready() {
+		return fmt.Errorf("no connection to signal")
+	}
 	if c.stream == nil {
 		return fmt.Errorf("connection to the Signal Exchnage has not been established yet. Please call Client.Receive before sending messages")
 	}
@@ -216,13 +286,17 @@ func (c *Client) encryptMessage(msg *proto.Message) (*proto.EncryptedMessage, er
 // Send sends a message to the remote Peer through the Signal Exchange.
 func (c *Client) Send(msg *proto.Message) error {
 
+	if !c.ready() {
+		return fmt.Errorf("no connection to signal")
+	}
+
 	encryptedMessage, err := c.encryptMessage(msg)
 	if err != nil {
 		return err
 	}
 	_, err = c.realClient.Send(context.TODO(), encryptedMessage)
 	if err != nil {
-		log.Errorf("error while sending message to peer [%s] [error: %v]", msg.RemoteKey, err)
+		//log.Errorf("error while sending message to peer [%s] [error: %v]", msg.RemoteKey, err)
 		return err
 	}
 
@@ -239,10 +313,10 @@ func (c *Client) receive(stream proto.SignalExchange_ConnectStreamClient,
 			log.Warnf("stream canceled (usually indicates shutdown)")
 			return err
 		} else if s.Code() == codes.Unavailable {
-			log.Warnf("server has been stopped")
+			log.Warnf("Signal Service is unavailable")
 			return err
 		} else if err == io.EOF {
-			log.Warnf("stream closed by server")
+			log.Warnf("Signal Service stream closed by server")
 			return err
 		} else if err != nil {
 			return err

signal/client/client_test.go

@@ -48,30 +48,42 @@ var _ = Describe("Client", func() {
 				// connect PeerA to Signal
 				keyA, _ := wgtypes.GenerateKey()
 				clientA := createSignalClient(addr, keyA)
-				clientA.Receive(func(msg *sigProto.Message) error {
-					receivedOnA = msg.GetBody().GetPayload()
-					msgReceived.Done()
-					return nil
-				})
-				clientA.WaitConnected()
+				go func() {
+					err := clientA.Receive(func(msg *sigProto.Message) error {
+						receivedOnA = msg.GetBody().GetPayload()
+						msgReceived.Done()
+						return nil
+					})
+					if err != nil {
+						return
+					}
+				}()
+				clientA.WaitStreamConnected()
 
 				// connect PeerB to Signal
 				keyB, _ := wgtypes.GenerateKey()
 				clientB := createSignalClient(addr, keyB)
-				clientB.Receive(func(msg *sigProto.Message) error {
-					receivedOnB = msg.GetBody().GetPayload()
-					err := clientB.Send(&sigProto.Message{
-						Key:       keyB.PublicKey().String(),
-						RemoteKey: keyA.PublicKey().String(),
-						Body:      &sigProto.Body{Payload: "pong"},
+
+				go func() {
+					err := clientB.Receive(func(msg *sigProto.Message) error {
+						receivedOnB = msg.GetBody().GetPayload()
+						err := clientB.Send(&sigProto.Message{
+							Key:       keyB.PublicKey().String(),
+							RemoteKey: keyA.PublicKey().String(),
+							Body:      &sigProto.Body{Payload: "pong"},
+						})
+						if err != nil {
+							Fail("failed sending a message to PeerA")
+						}
+						msgReceived.Done()
+						return nil
 					})
 					if err != nil {
-						Fail("failed sending a message to PeerA")
+						return
 					}
-					msgReceived.Done()
-					return nil
-				})
-				clientB.WaitConnected()
+				}()
+
+				clientB.WaitStreamConnected()
 
 				// PeerA initiates ping-pong
 				err := clientA.Send(&sigProto.Message{
@@ -100,11 +112,15 @@ var _ = Describe("Client", func() {
 
 				key, _ := wgtypes.GenerateKey()
 				client := createSignalClient(addr, key)
-				client.Receive(func(msg *sigProto.Message) error {
-					return nil
-				})
-				client.WaitConnected()
-
+				go func() {
+					err := client.Receive(func(msg *sigProto.Message) error {
+						return nil
+					})
+					if err != nil {
+						return
+					}
+				}()
+				client.WaitStreamConnected()
 				Expect(client).NotTo(BeNil())
 			})
 		})