test: add basic signal IT tests

.github/workflows/golang-test.yml

@@ -18,4 +18,33 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v2
       - name: Test
-        run: go test ./...
+        run: go test ./...
+
+  test_build:
+    strategy:
+      matrix:
+        os: [ windows, linux, darwin ]
+        go-version: [1.16.x]
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Install Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: ${{ matrix.go-version }}
+
+      - name: Cache Go modules
+        uses: actions/cache@v1
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: |
+            ${{ runner.os }}-go-
+
+      - name: Install modules
+        run: go mod tidy
+
+      - name: run build
+        run: GOOS=${{ matrix.os }} go build .

cmd/signal.go

@@ -6,7 +6,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	sig "github.com/wiretrustee/wiretrustee/signal"
-	sProto "github.com/wiretrustee/wiretrustee/signal/proto"
+	sigProto "github.com/wiretrustee/wiretrustee/signal/proto"
 	"google.golang.org/grpc"
 	"net"
 )
@@ -30,7 +30,7 @@ var (
 			}
 			var opts []grpc.ServerOption
 			grpcServer := grpc.NewServer(opts...)
-			sProto.RegisterSignalExchangeServer(grpcServer, sig.NewServer())
+			sigProto.RegisterSignalExchangeServer(grpcServer, sig.NewServer())
 			log.Printf("started server: localhost:%v", port)
 			if err := grpcServer.Serve(lis); err != nil {
 				log.Fatalf("failed to serve: %v", err)

go.mod

@@ -4,8 +4,10 @@ go 1.16
 
 require (
 	github.com/cenkalti/backoff/v4 v4.1.0
-	github.com/golang/protobuf v1.4.3
+	github.com/golang/protobuf v1.5.2
 	github.com/google/nftables v0.0.0-20201230142148-715e31cb3c31
+	github.com/onsi/ginkgo v1.16.4
+	github.com/onsi/gomega v1.13.0
 	github.com/pion/ice/v2 v2.1.7
 	github.com/sirupsen/logrus v1.7.0
 	github.com/spf13/cobra v1.1.3

iface/iface_darwin.go

@@ -18,7 +18,7 @@ func assignAddr(address string, tunDevice tun.Device) error {
 	ip := strings.Split(address, "/")
 	cmd := exec.Command("ifconfig", ifaceName, "inet", address, ip[0])
 	if out, err := cmd.CombinedOutput(); err != nil {
-		log.Infoln("Command: %v failed with output %s and error: ", cmd.String(), out)
+		log.Infof("Command: %v failed with output %s and error: ", cmd.String(), out)
 		return err
 	}
 	_, resolvedNet, err := net.ParseCIDR(address)

signal/README.md

@@ -1,18 +1,23 @@
+# Wiretrustee Signal Server
+
 This is a Wiretrustee signal-exchange server and client library to exchange connection information between Wiretrustee Trusted Device and Wiretrustee Hub
 
 The project uses gRPC library and defines service in protobuf file located in:
  ```proto/signal_exchange.proto```
- 
+
  To build the project you have to do the following things.
 
 Install protobuf version 3 (by default v3 is installed on ubuntu 20.04. On previous versions it is proto 2):
- ```
+
+```bash
+#!/bin/bash
 sudo apt install protoc-gen-go
 sudo apt install golang-goprotobuf-dev
- ```
- 
+```
+
 Generate gRPC code:
- ```
- protoc -I proto/ proto/signalexchange.proto --go_out=plugins=grpc:proto
 
- ```
+```bash
+#!/bin/bash
+protoc -I proto/ proto/signalexchange.proto --go_out=plugins=grpc:proto
+```

signal/client.go

@@ -29,7 +29,7 @@ type Client struct {
 	ctx        context.Context
 	stream     proto.SignalExchange_ConnectStreamClient
 	//waiting group to notify once stream is connected
-	connWg sync.WaitGroup //todo use a channel instead??
+	connWg *sync.WaitGroup //todo use a channel instead??
 }
 
 // Close Closes underlying connections to the Signal Exchange
@@ -55,11 +55,13 @@ func NewClient(ctx context.Context, addr string, key wgtypes.Key) (*Client, erro
 		return nil, err
 	}
 
+	var wg sync.WaitGroup
 	return &Client{
 		realClient: proto.NewSignalExchangeClient(conn),
 		ctx:        ctx,
 		signalConn: conn,
 		key:        key,
+		connWg:     &wg,
 	}, nil
 }
 
@@ -107,15 +109,22 @@ func (c *Client) connect(key string, msgHandler func(msg *proto.Message) error)
 	// add key fingerprint to the request header to be identified on the server side
 	md := metadata.New(map[string]string{proto.HeaderId: key})
 	ctx := metadata.NewOutgoingContext(c.ctx, md)
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
 
 	stream, err := c.realClient.ConnectStream(ctx)
 
 	c.stream = stream
 	if err != nil {
 		return err
 	}
+	// blocks
+	header, err := c.stream.Header()
+	if err != nil {
+		return 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")
+	}
 	//connection established we are good to use the stream
 	c.connWg.Done()
 

signal/proto/constants.go

@@ -2,3 +2,4 @@ package proto
 
 // protocol constants, field names that can be used by both client and server
 const HeaderId = "x-wiretrustee-peer-id"
+const HeaderRegistered = "x-wiretrustee-peer-registered"

signal/signal.go

@@ -12,20 +12,20 @@ import (
 	"io"
 )
 
-// SignalExchangeServer an instance of a Signal server
-type SignalExchangeServer struct {
+// Server an instance of a Signal server
+type Server struct {
 	registry *peer.Registry
 }
 
 // NewServer creates a new Signal server
-func NewServer() *SignalExchangeServer {
-	return &SignalExchangeServer{
+func NewServer() *Server {
+	return &Server{
 		registry: peer.NewRegistry(),
 	}
 }
 
 // Send forwards a message to the signal peer
-func (s *SignalExchangeServer) Send(ctx context.Context, msg *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
+func (s *Server) Send(ctx context.Context, msg *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
 
 	if !s.registry.IsPeerRegistered(msg.Key) {
 		return nil, fmt.Errorf("unknown peer %s", msg.Key)
@@ -46,14 +46,20 @@ func (s *SignalExchangeServer) Send(ctx context.Context, msg *proto.EncryptedMes
 }
 
 // ConnectStream connects to the exchange stream
-func (s *SignalExchangeServer) ConnectStream(stream proto.SignalExchange_ConnectStreamServer) error {
+func (s *Server) ConnectStream(stream proto.SignalExchange_ConnectStreamServer) error {
 	p, err := s.connectPeer(stream)
 	if err != nil {
 		return err
 	}
 
-	log.Infof("peer [%s] has successfully connected", p.Id)
+	//needed to confirm that the peer has been registered so that the client can proceed
+	header := metadata.Pairs(proto.HeaderRegistered, "1")
+	err = stream.SendHeader(header)
+	if err != nil {
+		return err
+	}
 
+	log.Infof("peer [%s] has successfully connected", p.Id)
 	for {
 		msg, err := stream.Recv()
 		if err == io.EOF {
@@ -83,7 +89,7 @@ func (s *SignalExchangeServer) ConnectStream(stream proto.SignalExchange_Connect
 // Handles initial Peer connection.
 // Each connection must provide an ID header.
 // At this moment the connecting Peer will be registered in the peer.Registry
-func (s SignalExchangeServer) connectPeer(stream proto.SignalExchange_ConnectStreamServer) (*peer.Peer, error) {
+func (s Server) connectPeer(stream proto.SignalExchange_ConnectStreamServer) (*peer.Peer, error) {
 	if meta, hasMeta := metadata.FromIncomingContext(stream.Context()); hasMeta {
 		if id, found := meta[proto.HeaderId]; found {
 			p := peer.NewPeer(id[0], stream)

signal/signal_suite_test.go

@@ -0,0 +1,13 @@
+package signal_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"testing"
+)
+
+func TestSignal(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Signal Suite")
+}