project init

signal/README.md

@@ -0,0 +1,18 @@
+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):
+ ```
+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
+
+ ```

signal/client.go

@@ -0,0 +1,210 @@
+package signal
+
+import (
+	"context"
+	"fmt"
+	"github.com/cenkalti/backoff/v4"
+	log "github.com/sirupsen/logrus"
+	"github.com/wiretrustee/wiretrustee/signal/proto"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"io"
+	"strings"
+	"sync"
+	"time"
+)
+
+// A set of tools to exchange connection details (Wireguard endpoints) with the remote peer.
+
+// Wraps the Signal Exchange Service gRpc client
+type Client struct {
+	realClient proto.SignalExchangeClient
+	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??
+}
+
+// Closes underlying connections to the Signal Exchange
+func (client *Client) Close() error {
+	return client.signalConn.Close()
+}
+
+func NewClient(addr string, ctx context.Context) (*Client, error) {
+
+	conn, err := grpc.DialContext(
+		ctx,
+		addr,
+		grpc.WithInsecure(),
+		grpc.WithBlock(),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			Time:    3 * time.Second,
+			Timeout: 2 * time.Second,
+		}))
+
+	if err != nil {
+		log.Errorf("failed to connect to the signalling server %v", err)
+		return nil, err
+	}
+
+	return &Client{
+		realClient: proto.NewSignalExchangeClient(conn),
+		ctx:        ctx,
+		signalConn: conn,
+	}, nil
+}
+
+// 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 (client *Client) Receive(key string, msgHandler func(msg *proto.Message) error) {
+	client.connWg.Add(1)
+	go func() {
+
+		var backOff = &backoff.ExponentialBackOff{
+			InitialInterval:     backoff.DefaultInitialInterval,
+			RandomizationFactor: backoff.DefaultRandomizationFactor,
+			Multiplier:          backoff.DefaultMultiplier,
+			MaxInterval:         3 * time.Second,
+			MaxElapsedTime:      time.Duration(0), //never stop
+			Stop:                backoff.Stop,
+			Clock:               backoff.SystemClock,
+		}
+
+		operation := func() error {
+			err := client.connect(key, msgHandler)
+			if err != nil {
+				log.Warnf("disconnected from the Signal Exchange due to an error %s. Retrying ... ", err)
+				client.connWg.Add(1)
+				return err
+			}
+
+			backOff.Reset()
+			return nil
+		}
+
+		err := backoff.Retry(operation, backOff)
+		if err != nil {
+			log.Errorf("error while communicating with the Signal Exchange %s ", err)
+			return
+		}
+	}()
+}
+
+func (client *Client) connect(key string, msgHandler func(msg *proto.Message) error) error {
+	client.stream = nil
+
+	// 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(client.ctx, md)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	stream, err := client.realClient.ConnectStream(ctx)
+
+	client.stream = stream
+	if err != nil {
+		return err
+	}
+	//connection established we are good to use the stream
+	client.connWg.Done()
+
+	log.Infof("connected to the Signal Exchange Stream")
+
+	return client.receive(stream, msgHandler)
+}
+
+// Waits until the client is connected to the message stream
+func (client *Client) WaitConnected() {
+	client.connWg.Wait()
+}
+
+// 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 (client *Client) SendToStream(msg *proto.Message) error {
+
+	if client.stream == nil {
+		return fmt.Errorf("connection to the Signal Exchnage has not been established yet. Please call Client.Receive before sending messages")
+	}
+
+	err := client.stream.Send(msg)
+	if err != nil {
+		log.Errorf("error while sending message to peer [%s] [error: %v]", msg.RemoteKey, err)
+		return err
+	}
+
+	return nil
+}
+
+// Sends a message to the remote Peer through the Signal Exchange.
+func (client *Client) Send(msg *proto.Message) error {
+
+	_, err := client.realClient.Connect(context.TODO(), msg)
+	if err != nil {
+		log.Errorf("error while sending message to peer [%s] [error: %v]", msg.RemoteKey, err)
+		return err
+	}
+
+	return nil
+}
+
+// Receives messages from other peers coming through the Signal Exchange
+func (client *Client) receive(stream proto.SignalExchange_ConnectStreamClient,
+	msgHandler func(msg *proto.Message) error) error {
+
+	for {
+		msg, err := stream.Recv()
+		if s, ok := status.FromError(err); ok && s.Code() == codes.Canceled {
+			log.Warnf("stream canceled (usually indicates shutdown)")
+			return err
+		} else if s.Code() == codes.Unavailable {
+			log.Warnf("server has been stopped")
+			return err
+		} else if err == io.EOF {
+			log.Warnf("stream closed by server")
+			return err
+		} else if err != nil {
+			return err
+		}
+		log.Debugf("received a new message from Peer [fingerprint: %s] [type %s]", msg.Key, msg.Type)
+
+		//todo decrypt
+		err = msgHandler(msg)
+
+		if err != nil {
+			log.Errorf("error while handling message of Peer [key: %s] error: [%s]", msg.Key, err.Error())
+			//todo send something??
+		}
+	}
+}
+
+func UnMarshalCredential(msg *proto.Message) (*Credential, error) {
+	credential := strings.Split(msg.Body, ":")
+	if len(credential) != 2 {
+		return nil, fmt.Errorf("error parsing message body %s", msg.Body)
+	}
+	return &Credential{
+		UFrag: credential[0],
+		Pwd:   credential[1],
+	}, nil
+}
+
+func MarshalCredential(ourKey string, remoteKey string, credential *Credential, t proto.Message_Type) *proto.Message {
+	return &proto.Message{
+		Type:      t,
+		Key:       ourKey,
+		RemoteKey: remoteKey,
+		Body:      fmt.Sprintf("%s:%s", credential.UFrag, credential.Pwd),
+	}
+}
+
+type Credential struct {
+	UFrag string
+	Pwd   string
+}

signal/encryption.go

@@ -0,0 +1,52 @@
+package signal
+
+import (
+	"crypto/rand"
+	"fmt"
+	"golang.org/x/crypto/nacl/box"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+)
+
+// As set of tools to encrypt/decrypt messages being sent through the Signal Exchange Service.
+// We want to make sure that the Connection Candidates and other irrelevant (to the Signal Exchange) information can't be read anywhere else but the Peer the message is being sent to.
+// These tools use Golang crypto package (Curve25519, XSalsa20 and Poly1305 to encrypt and authenticate)
+// Wireguard keys are used for encryption
+
+// Encrypts a message using local Wireguard private key and remote peer's public key.
+func EncryptMessage(msg []byte, privateKey wgtypes.Key, remotePubKey wgtypes.Key) ([]byte, error) {
+	nonce, err := genNonce()
+	if err != nil {
+		return nil, err
+	}
+
+	return box.Seal(nil, msg, nonce, toByte32(remotePubKey), toByte32(privateKey)), nil
+}
+
+// Decrypts a message that has been encrypted by the remote peer using Wireguard private key and remote peer's public key.
+func DecryptMessage(encryptedMsg []byte, privateKey wgtypes.Key, remotePubKey wgtypes.Key) ([]byte, error) {
+	nonce, err := genNonce()
+	if err != nil {
+		return nil, err
+	}
+
+	opened, ok := box.Open(nil, encryptedMsg, nonce, toByte32(remotePubKey), toByte32(privateKey))
+	if !ok {
+		return nil, fmt.Errorf("failed to decrypt message from peer %s", remotePubKey.String())
+	}
+
+	return opened, nil
+}
+
+// Generates nonce of size 24
+func genNonce() (*[24]byte, error) {
+	var nonce [24]byte
+	if _, err := rand.Read(nonce[:]); err != nil {
+		return nil, err
+	}
+	return &nonce, nil
+}
+
+// Converts Wireguard key to byte array of size 32 (a format used by the golang crypto package)
+func toByte32(key wgtypes.Key) *[32]byte {
+	return (*[32]byte)(&key)
+}

signal/fingerprint.go

@@ -0,0 +1,18 @@
+package signal
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
+
+const (
+	HexTable = "0123456789abcdef"
+)
+
+// Generates a SHA256 Fingerprint of the string
+func FingerPrint(key string) string {
+	hasher := sha256.New()
+	hasher.Write([]byte(key))
+	sha := hasher.Sum(nil)
+	return hex.EncodeToString(sha)
+}

signal/peer/peer.go

@@ -0,0 +1,54 @@
+package peer
+
+import (
+	log "github.com/sirupsen/logrus"
+	"github.com/wiretrustee/wiretrustee/signal/proto"
+)
+
+// Representation of a connected Peer
+type Peer struct {
+	// a unique id of the Peer (e.g. sha256 fingerprint of the Wireguard public key)
+	Id string
+
+	//a gRpc connection stream to the Peer
+	Stream proto.SignalExchange_ConnectStreamServer
+}
+
+func NewPeer(id string, stream proto.SignalExchange_ConnectStreamServer) *Peer {
+	return &Peer{
+		Id:     id,
+		Stream: stream,
+	}
+}
+
+// registry that holds all currently connected Peers
+type Registry struct {
+	// Peer.key -> Peer
+	Peers map[string]*Peer
+}
+
+func NewRegistry() *Registry {
+	return &Registry{
+		Peers: make(map[string]*Peer),
+	}
+}
+
+// Registers peer in the registry
+func (reg *Registry) Register(peer *Peer) {
+	if _, exists := reg.Peers[peer.Id]; exists {
+		log.Warnf("peer [%s] has been already registered", peer.Id)
+	} else {
+		log.Printf("registering new peer [%s]", peer.Id)
+	}
+	//replace Peer even if exists
+	//todo should we really replace?
+	reg.Peers[peer.Id] = peer
+}
+
+// Deregister Peer from the Registry (usually once it disconnects)
+func (reg *Registry) DeregisterHub(peer *Peer) {
+	if _, ok := reg.Peers[peer.Id]; ok {
+		delete(reg.Peers, peer.Id)
+		log.Printf("deregistered peer [%s]", peer.Id)
+	}
+}

signal/proto/constants.go

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

signal/proto/signalexchange.pb.go

@@ -0,0 +1,301 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: signalexchange.proto
+
+package proto
+
+import (
+	context "context"
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	_ "github.com/golang/protobuf/protoc-gen-go/descriptor"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+	math "math"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+// Message type
+type Message_Type int32
+
+const (
+	Message_OFFER     Message_Type = 0
+	Message_ANSWER    Message_Type = 1
+	Message_CANDIDATE Message_Type = 2
+)
+
+var Message_Type_name = map[int32]string{
+	0: "OFFER",
+	1: "ANSWER",
+	2: "CANDIDATE",
+}
+
+var Message_Type_value = map[string]int32{
+	"OFFER":     0,
+	"ANSWER":    1,
+	"CANDIDATE": 2,
+}
+
+func (x Message_Type) String() string {
+	return proto.EnumName(Message_Type_name, int32(x))
+}
+
+func (Message_Type) EnumDescriptor() ([]byte, []int) {
+	return fileDescriptor_bf680d70b8e3473f, []int{0, 0}
+}
+
+type Message struct {
+	Type Message_Type `protobuf:"varint,1,opt,name=type,proto3,enum=signalexchange.Message_Type" json:"type,omitempty"`
+	// a sha256 fingerprint of the Wireguard public key
+	Key string `protobuf:"bytes,2,opt,name=key,proto3" json:"key,omitempty"`
+	// a sha256 fingerprint of the Wireguard public key of the remote peer to connect to
+	RemoteKey            string   `protobuf:"bytes,3,opt,name=remoteKey,proto3" json:"remoteKey,omitempty"`
+	Body                 string   `protobuf:"bytes,4,opt,name=body,proto3" json:"body,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *Message) Reset()         { *m = Message{} }
+func (m *Message) String() string { return proto.CompactTextString(m) }
+func (*Message) ProtoMessage()    {}
+func (*Message) Descriptor() ([]byte, []int) {
+	return fileDescriptor_bf680d70b8e3473f, []int{0}
+}
+
+func (m *Message) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_Message.Unmarshal(m, b)
+}
+func (m *Message) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_Message.Marshal(b, m, deterministic)
+}
+func (m *Message) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Message.Merge(m, src)
+}
+func (m *Message) XXX_Size() int {
+	return xxx_messageInfo_Message.Size(m)
+}
+func (m *Message) XXX_DiscardUnknown() {
+	xxx_messageInfo_Message.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Message proto.InternalMessageInfo
+
+func (m *Message) GetType() Message_Type {
+	if m != nil {
+		return m.Type
+	}
+	return Message_OFFER
+}
+
+func (m *Message) GetKey() string {
+	if m != nil {
+		return m.Key
+	}
+	return ""
+}
+
+func (m *Message) GetRemoteKey() string {
+	if m != nil {
+		return m.RemoteKey
+	}
+	return ""
+}
+
+func (m *Message) GetBody() string {
+	if m != nil {
+		return m.Body
+	}
+	return ""
+}
+
+func init() {
+	proto.RegisterEnum("signalexchange.Message_Type", Message_Type_name, Message_Type_value)
+	proto.RegisterType((*Message)(nil), "signalexchange.Message")
+}
+
+func init() { proto.RegisterFile("signalexchange.proto", fileDescriptor_bf680d70b8e3473f) }
+
+var fileDescriptor_bf680d70b8e3473f = []byte{
+	// 272 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x50, 0xcd, 0x4a, 0xf3, 0x40,
+	0x14, 0xed, 0xb4, 0xf9, 0x1a, 0x72, 0xa1, 0x21, 0x5c, 0x3e, 0x30, 0x94, 0x2e, 0x42, 0x56, 0x59,
+	0x48, 0x5a, 0xea, 0x52, 0x5c, 0xc4, 0x36, 0x15, 0x11, 0x2b, 0x24, 0x05, 0xc1, 0x5d, 0x9a, 0x5e,
+	0xc7, 0x62, 0x9b, 0x09, 0x93, 0x11, 0x9c, 0x37, 0xf1, 0x25, 0x7c, 0x47, 0xe9, 0x34, 0x20, 0x0a,
+	0xdd, 0xb8, 0x9a, 0xc3, 0xf9, 0x9b, 0xc3, 0x85, 0xff, 0xcd, 0x96, 0x57, 0xc5, 0x8e, 0xde, 0xcb,
+	0x97, 0xa2, 0xe2, 0x14, 0xd7, 0x52, 0x28, 0x81, 0xee, 0x4f, 0x76, 0x18, 0x70, 0x21, 0xf8, 0x8e,
+	0xc6, 0x46, 0x5d, 0xbf, 0x3d, 0x8f, 0x37, 0xd4, 0x94, 0x72, 0x5b, 0x2b, 0x21, 0x8f, 0x89, 0xf0,
+	0x93, 0x81, 0x7d, 0x4f, 0x4d, 0x53, 0x70, 0xc2, 0x09, 0x58, 0x4a, 0xd7, 0xe4, 0xb3, 0x80, 0x45,
+	0xee, 0x74, 0x14, 0xff, 0xfa, 0xa2, 0xb5, 0xc5, 0x2b, 0x5d, 0x53, 0x66, 0x9c, 0xe8, 0x41, 0xef,
+	0x95, 0xb4, 0xdf, 0x0d, 0x58, 0xe4, 0x64, 0x07, 0x88, 0x23, 0x70, 0x24, 0xed, 0x85, 0xa2, 0x3b,
+	0xd2, 0x7e, 0xcf, 0xf0, 0xdf, 0x04, 0x22, 0x58, 0x6b, 0xb1, 0xd1, 0xbe, 0x65, 0x04, 0x83, 0xc3,
+	0x73, 0xb0, 0x0e, 0x8d, 0xe8, 0xc0, 0xbf, 0x87, 0xc5, 0x22, 0xcd, 0xbc, 0x0e, 0x02, 0xf4, 0x93,
+	0x65, 0xfe, 0x98, 0x66, 0x1e, 0xc3, 0x01, 0x38, 0xb3, 0x64, 0x39, 0xbf, 0x9d, 0x27, 0xab, 0xd4,
+	0xeb, 0x4e, 0x3f, 0x18, 0xb8, 0xb9, 0xd9, 0x95, 0xb6, 0xbb, 0xf0, 0x0a, 0xec, 0x99, 0xa8, 0x2a,
+	0x2a, 0x15, 0x9e, 0x9d, 0xd8, 0x3c, 0x3c, 0x25, 0x84, 0x1d, 0xbc, 0x81, 0x41, 0x1b, 0xcf, 0x95,
+	0xa4, 0x62, 0xff, 0x97, 0x92, 0x88, 0x4d, 0xd8, 0xb5, 0xf3, 0x64, 0xc7, 0x97, 0xc7, 0x4b, 0xf7,
+	0xcd, 0x73, 0xf1, 0x15, 0x00, 0x00, 0xff, 0xff, 0x20, 0x13, 0xc1, 0xe1, 0xa6, 0x01, 0x00, 0x00,
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// SignalExchangeClient is the client API for SignalExchange service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type SignalExchangeClient interface {
+	// Synchronously connect to the Signal Exchange service offering connection candidates and waiting for connection candidates from the other party (remote peer)
+	Connect(ctx context.Context, in *Message, opts ...grpc.CallOption) (*Message, error)
+	// Connect to the Signal Exchange service offering connection candidates and maintain a channel for receiving candidates from the other party (remote peer)
+	ConnectStream(ctx context.Context, opts ...grpc.CallOption) (SignalExchange_ConnectStreamClient, error)
+}
+
+type signalExchangeClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewSignalExchangeClient(cc *grpc.ClientConn) SignalExchangeClient {
+	return &signalExchangeClient{cc}
+}
+
+func (c *signalExchangeClient) Connect(ctx context.Context, in *Message, opts ...grpc.CallOption) (*Message, error) {
+	out := new(Message)
+	err := c.cc.Invoke(ctx, "/signalexchange.SignalExchange/Connect", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *signalExchangeClient) ConnectStream(ctx context.Context, opts ...grpc.CallOption) (SignalExchange_ConnectStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &_SignalExchange_serviceDesc.Streams[0], "/signalexchange.SignalExchange/ConnectStream", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &signalExchangeConnectStreamClient{stream}
+	return x, nil
+}
+
+type SignalExchange_ConnectStreamClient interface {
+	Send(*Message) error
+	Recv() (*Message, error)
+	grpc.ClientStream
+}
+
+type signalExchangeConnectStreamClient struct {
+	grpc.ClientStream
+}
+
+func (x *signalExchangeConnectStreamClient) Send(m *Message) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *signalExchangeConnectStreamClient) Recv() (*Message, error) {
+	m := new(Message)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+// SignalExchangeServer is the server API for SignalExchange service.
+type SignalExchangeServer interface {
+	// Synchronously connect to the Signal Exchange service offering connection candidates and waiting for connection candidates from the other party (remote peer)
+	Connect(context.Context, *Message) (*Message, error)
+	// Connect to the Signal Exchange service offering connection candidates and maintain a channel for receiving candidates from the other party (remote peer)
+	ConnectStream(SignalExchange_ConnectStreamServer) error
+}
+
+// UnimplementedSignalExchangeServer can be embedded to have forward compatible implementations.
+type UnimplementedSignalExchangeServer struct {
+}
+
+func (*UnimplementedSignalExchangeServer) Connect(ctx context.Context, req *Message) (*Message, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Connect not implemented")
+}
+func (*UnimplementedSignalExchangeServer) ConnectStream(srv SignalExchange_ConnectStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method ConnectStream not implemented")
+}
+
+func RegisterSignalExchangeServer(s *grpc.Server, srv SignalExchangeServer) {
+	s.RegisterService(&_SignalExchange_serviceDesc, srv)
+}
+
+func _SignalExchange_Connect_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(Message)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SignalExchangeServer).Connect(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/signalexchange.SignalExchange/Connect",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SignalExchangeServer).Connect(ctx, req.(*Message))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _SignalExchange_ConnectStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(SignalExchangeServer).ConnectStream(&signalExchangeConnectStreamServer{stream})
+}
+
+type SignalExchange_ConnectStreamServer interface {
+	Send(*Message) error
+	Recv() (*Message, error)
+	grpc.ServerStream
+}
+
+type signalExchangeConnectStreamServer struct {
+	grpc.ServerStream
+}
+
+func (x *signalExchangeConnectStreamServer) Send(m *Message) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *signalExchangeConnectStreamServer) Recv() (*Message, error) {
+	m := new(Message)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+var _SignalExchange_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "signalexchange.SignalExchange",
+	HandlerType: (*SignalExchangeServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Connect",
+			Handler:    _SignalExchange_Connect_Handler,
+		},
+	},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "ConnectStream",
+			Handler:       _SignalExchange_ConnectStream_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+	},
+	Metadata: "signalexchange.proto",
+}

signal/proto/signalexchange.proto

@@ -0,0 +1,34 @@
+syntax = "proto3";
+
+import "google/protobuf/descriptor.proto";
+
+option go_package = ".;proto";
+
+package signalexchange;
+
+service SignalExchange {
+  // Synchronously connect to the Signal Exchange service offering connection candidates and waiting for connection candidates from the other party (remote peer)
+  rpc Connect(Message) returns (Message) {}
+  // Connect to the Signal Exchange service offering connection candidates and maintain a channel for receiving candidates from the other party (remote peer)
+  rpc ConnectStream(stream Message) returns (stream Message) {}
+}
+
+message Message {
+
+  // Message type
+  enum Type {
+    OFFER = 0;
+    ANSWER = 1;
+    CANDIDATE = 2;
+  }
+
+  Type type = 1;
+
+  // a sha256 fingerprint of the Wireguard public key
+  string key = 2;
+
+  // a sha256 fingerprint of the Wireguard public key of the remote peer to connect to
+  string remoteKey = 3;
+
+  string body = 4;
+}

signal/signal.go

@@ -0,0 +1,99 @@
+package signal
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	log "github.com/sirupsen/logrus"
+	"github.com/wiretrustee/wiretrustee/signal/peer"
+	"github.com/wiretrustee/wiretrustee/signal/proto"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+	"io"
+)
+
+var (
+	port = flag.Int("port", 10000, "The server port")
+)
+
+type SignalExchangeServer struct {
+	registry *peer.Registry
+}
+
+func NewServer() *SignalExchangeServer {
+	return &SignalExchangeServer{
+		registry: peer.NewRegistry(),
+	}
+}
+
+func (s *SignalExchangeServer) Connect(ctx context.Context, msg *proto.Message) (*proto.Message, error) {
+
+	if _, found := s.registry.Peers[msg.Key]; !found {
+		return nil, fmt.Errorf("unknown peer %s", msg.Key)
+	}
+
+	if dstPeer, found := s.registry.Peers[msg.RemoteKey]; found {
+		//forward the message to the target peer
+		err := dstPeer.Stream.Send(msg)
+		if err != nil {
+			log.Errorf("error while forwarding message from peer [%s] to peer [%s]", msg.Key, msg.RemoteKey)
+			//todo respond to the sender?
+		}
+	} else {
+		log.Warnf("message from peer [%s] can't be forwarded to peer [%s] because destination peer is not connected", msg.Key, msg.RemoteKey)
+		//todo respond to the sender?
+	}
+	return &proto.Message{}, nil
+}
+
+func (s *SignalExchangeServer) 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)
+
+	for {
+		msg, err := stream.Recv()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return err
+		}
+		log.Debugf("received a new message from peer [%s] to peer [%s]", p.Id, msg.RemoteKey)
+		// lookup the target peer where the message is going to
+		if dstPeer, found := s.registry.Peers[msg.RemoteKey]; found {
+			//forward the message to the target peer
+			err := dstPeer.Stream.Send(msg)
+			if err != nil {
+				log.Errorf("error while forwarding message from peer [%s] to peer [%s]", p.Id, msg.RemoteKey)
+				//todo respond to the sender?
+			}
+		} else {
+			log.Warnf("message from peer [%s] can't be forwarded to peer [%s] because destination peer is not connected", p.Id, msg.RemoteKey)
+			//todo respond to the sender?
+		}
+
+	}
+	<-stream.Context().Done()
+	return stream.Context().Err()
+}
+
+// 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) {
+	if meta, hasMeta := metadata.FromIncomingContext(stream.Context()); hasMeta {
+		if id, found := meta[proto.HeaderId]; found {
+			p := peer.NewPeer(id[0], stream)
+			s.registry.Register(p)
+			return p, nil
+		} else {
+			return nil, status.Errorf(codes.FailedPrecondition, "missing connection header: "+proto.HeaderId)
+		}
+	} else {
+		return nil, status.Errorf(codes.FailedPrecondition, "missing connection stream meta")
+	}
+}