Add symmetric broadcast

broadcast/interface.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"gitlab.com/elixxir/client/cmix"
+	"gitlab.com/elixxir/client/cmix/identity/receptionID"
+	"gitlab.com/elixxir/client/cmix/rounds"
+	crypto "gitlab.com/elixxir/crypto/broadcast"
+	"gitlab.com/xx_network/primitives/id"
+	"gitlab.com/xx_network/primitives/id/ephemeral"
+)
+
+// ListenerFunc is registered when creating a new symmetric broadcasting channel
+// and receives all new broadcast messages for the channel.
+type ListenerFunc func(payload []byte,
+	receptionID receptionID.EphemeralIdentity, round rounds.Round)
+
+// Symmetric manages the listening and broadcasting of a symmetric broadcast
+// channel.
+type Symmetric interface {
+	// MaxPayloadSize returns the maximum size for a broadcasted payload.
+	MaxPayloadSize() int
+
+	// Get returns the crypto.Symmetric object containing the cryptographic and
+	// identifying information about the channel.
+	Get() crypto.Symmetric
+
+	// Broadcast broadcasts the payload to the channel. The payload size must be
+	// equal to MaxPayloadSize.
+	Broadcast(payload []byte, cMixParams cmix.CMIXParams) (
+		id.Round, ephemeral.Id, error)
+
+	// Stop unregisters the listener callback and stops the channel's identity
+	// from being tracked.
+	Stop()
+}

broadcast/processor.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	jww "github.com/spf13/jwalterweatherman"
+	"gitlab.com/elixxir/client/cmix/identity/receptionID"
+	"gitlab.com/elixxir/client/cmix/rounds"
+	crypto "gitlab.com/elixxir/crypto/broadcast"
+	"gitlab.com/elixxir/primitives/format"
+)
+
+// Error messages.
+const (
+	errDecrypt = "[BCAST] Failed to decrypt payload for broadcast %s (%q): %+v"
+)
+
+// processor manages the reception and decryption of a broadcast message.
+// Adheres to the message.Processor interface.
+type processor struct {
+	s  *crypto.Symmetric
+	cb ListenerFunc
+}
+
+// Process decrypts the broadcast message and sends the results on the callback.
+func (p *processor) Process(msg format.Message,
+	receptionID receptionID.EphemeralIdentity, round rounds.Round) {
+
+	payload, err := p.s.Decrypt(msg.GetContents(), msg.GetMac(), msg.GetKeyFP())
+	if err != nil {
+		jww.ERROR.Printf(errDecrypt, p.s.ReceptionID, p.s.Name, err)
+		return
+	}
+
+	go p.cb(payload, receptionID, round)
+}
+
+// String returns a string identifying the processor for debugging purposes.
+func (p *processor) String() string {
+	return "symmetricChannel-" + p.s.Name
+}

broadcast/processor_test.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"bytes"
+	"gitlab.com/elixxir/client/cmix/identity/receptionID"
+	"gitlab.com/elixxir/client/cmix/rounds"
+	crypto "gitlab.com/elixxir/crypto/broadcast"
+	"gitlab.com/elixxir/crypto/cmix"
+	"gitlab.com/elixxir/primitives/format"
+	"gitlab.com/xx_network/crypto/csprng"
+	"gitlab.com/xx_network/crypto/signature/rsa"
+	"gitlab.com/xx_network/primitives/id"
+	"testing"
+	"time"
+)
+
+// Tests that process.Process properly decrypts the payload and passes it to the
+// callback.
+func Test_processor_Process(t *testing.T) {
+	rng := csprng.NewSystemRNG()
+	rsaPrivKey, err := rsa.GenerateKey(rng, 64)
+	if err != nil {
+		t.Errorf("Failed to generate RSA key: %+v", err)
+	}
+	s := &crypto.Symmetric{
+		ReceptionID: id.NewIdFromString("channel", id.User, t),
+		Name:        "MyChannel",
+		Description: "This is my channel that I channel stuff on.",
+		Salt:        cmix.NewSalt(rng, 32),
+		RsaPubKey:   rsaPrivKey.GetPublic(),
+	}
+
+	cbChan := make(chan []byte)
+	cb := func(payload []byte, _ receptionID.EphemeralIdentity, _ rounds.Round) {
+		cbChan <- payload
+	}
+
+	p := &processor{
+		s:  s,
+		cb: cb,
+	}
+
+	msg := format.NewMessage(4092)
+	payload := make([]byte, msg.ContentsSize())
+	_, _ = rng.Read(payload)
+	encryptedPayload, mac, fp := p.s.Encrypt(payload, rng)
+	msg.SetContents(encryptedPayload)
+	msg.SetMac(mac)
+	msg.SetKeyFP(fp)
+
+	p.Process(msg, receptionID.EphemeralIdentity{}, rounds.Round{})
+
+	select {
+	case r := <-cbChan:
+		if !bytes.Equal(r, payload) {
+			t.Errorf("Did not receive expected payload."+
+				"\nexpected: %v\nreceived: %v", payload, r)
+		}
+	case <-time.After(15 * time.Millisecond):
+		t.Error("Timed out waiting for listener channel to be called.")
+	}
+}

broadcast/sizedBroadcast.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"encoding/binary"
+	"github.com/pkg/errors"
+)
+
+// Message field sizes.
+const (
+	sizeSize              = 2
+	sizedBroadcastMinSize = sizeSize
+)
+
+// Error messages.
+const (
+	// NewSizedBroadcast
+	errNewSizedBroadcastMaxSize = "size of payload and its size %d too large to fit in max payload size %d"
+
+	// DecodeSizedBroadcast
+	errDecodeSizedBroadcastDataLen = "size of data %d must be greater than %d"
+	errDecodeSizedBroadcastSize    = "stated payload size %d larger than provided data %d"
+)
+
+/*
++---------------------------+
+|   cMix Message Contents   |
++---------+-----------------+
+|  Size   |     Payload     |
+| 2 bytes | remaining space |
++---------+-----------------+
+*/
+
+// NewSizedBroadcast creates a new broadcast with its size information embedded.
+// The maxPayloadSize is the maximum size of the resulting payload. Returns an
+// error when the sized broadcast cannot fit in the max payload size.
+func NewSizedBroadcast(maxPayloadSize int, payload []byte) ([]byte, error) {
+	if len(payload)+sizedBroadcastMinSize > maxPayloadSize {
+		return nil, errors.Errorf(errNewSizedBroadcastMaxSize,
+			len(payload)+sizedBroadcastMinSize, maxPayloadSize)
+	}
+
+	b := make([]byte, sizeSize)
+	binary.LittleEndian.PutUint16(b, uint16(len(payload)))
+
+	return append(b, payload...), nil
+}
+
+// DecodeSizedBroadcast  the data into its original payload of the correct size.
+func DecodeSizedBroadcast(data []byte) ([]byte, error) {
+	if len(data) < sizedBroadcastMinSize {
+		return nil, errors.Errorf(
+			errDecodeSizedBroadcastDataLen, len(data), sizedBroadcastMinSize)
+	}
+
+	size := binary.LittleEndian.Uint16(data[:sizeSize])
+	if int(size) > len(data[sizeSize:]) {
+		return nil, errors.Errorf(
+			errDecodeSizedBroadcastSize, size, len(data[sizeSize:]))
+	}
+
+	return data[sizeSize : size+sizeSize], nil
+}
+
+// MaxSizedBroadcastPayloadSize returns the maximum payload size in a sized
+// broadcast for the given out message max payload size.
+func MaxSizedBroadcastPayloadSize(maxPayloadSize int) int {
+	return maxPayloadSize - sizedBroadcastMinSize
+}

broadcast/sizedBroadcast_test.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+)
+
+// Tests that a payload smaller than the max payload size encoded via
+// NewSizedBroadcast and decoded via DecodeSizedBroadcast matches the original.
+func TestNewSizedBroadcast_DecodeSizedBroadcast_SmallPayload(t *testing.T) {
+	const maxPayloadSize = 512
+	payload := []byte("This is my payload message.")
+
+	data, err := NewSizedBroadcast(maxPayloadSize, payload)
+	if err != nil {
+		t.Errorf("NewSizedBroadcast returned an error: %+v", err)
+	}
+
+	decodedPayload, err := DecodeSizedBroadcast(data)
+	if err != nil {
+		t.Errorf("DecodeSizedBroadcast returned an error: %+v", err)
+	}
+
+	if !bytes.Equal(payload, decodedPayload) {
+		t.Errorf("Decoded payload does not match original."+
+			"\nexpected: %q\nreceived: %q", payload, decodedPayload)
+	}
+}
+
+// Tests that a payload the same size as the max payload size encoded via
+// NewSizedBroadcast and decoded via DecodeSizedBroadcast matches the original.
+func TestNewSizedBroadcast_DecodeSizedBroadcast_FullSizesPayload(t *testing.T) {
+	payload := []byte("This is my payload message.")
+	maxPayloadSize := len(payload) + sizeSize
+
+	data, err := NewSizedBroadcast(maxPayloadSize, payload)
+	if err != nil {
+		t.Errorf("NewSizedBroadcast returned an error: %+v", err)
+	}
+
+	decodedPayload, err := DecodeSizedBroadcast(data)
+	if err != nil {
+		t.Errorf("DecodeSizedBroadcast returned an error: %+v", err)
+	}
+
+	if !bytes.Equal(payload, decodedPayload) {
+		t.Errorf("Decoded payload does not match original."+
+			"\nexpected: %q\nreceived: %q", payload, decodedPayload)
+	}
+}
+
+// Error path: tests that NewSizedBroadcast returns an error when the payload is
+// larger than the max payload size.
+func TestNewSizedBroadcast_MaxPayloadSizeError(t *testing.T) {
+	payload := []byte("This is my payload message.")
+	maxPayloadSize := len(payload)
+	expectedErr := fmt.Sprintf(errNewSizedBroadcastMaxSize,
+		len(payload)+sizedBroadcastMinSize, maxPayloadSize)
+
+	_, err := NewSizedBroadcast(maxPayloadSize, payload)
+	if err == nil || err.Error() != expectedErr {
+		t.Errorf("NewSizedBroadcast did not return the expected error when "+
+			"the payload is too large.\nexpected: %s\nreceived: %+v",
+			expectedErr, err)
+	}
+}
+
+// Error path: tests that DecodeSizedBroadcast returns an error when the length
+// of the data is shorter than the minimum length of a sized broadcast.
+func TestDecodeSizedBroadcast_DataTooShortError(t *testing.T) {
+	data := []byte{0}
+	expectedErr := fmt.Sprintf(
+		errDecodeSizedBroadcastDataLen, len(data), sizedBroadcastMinSize)
+
+	_, err := DecodeSizedBroadcast(data)
+	if err == nil || err.Error() != expectedErr {
+		t.Errorf("DecodeSizedBroadcast did not return the expected error "+
+			"when the data is too small.\nexpected: %s\nreceived: %+v",
+			expectedErr, err)
+	}
+}
+
+// Error path: tests that DecodeSizedBroadcast returns an error when the payload
+// size is larger than the actual payload contained in the data.
+func TestDecodeSizedBroadcast_SizeMismatchError(t *testing.T) {
+	data := []byte{255, 0, 10}
+	expectedErr := fmt.Sprintf(
+		errDecodeSizedBroadcastSize, data[0], len(data[sizeSize:]))
+
+	_, err := DecodeSizedBroadcast(data)
+	if err == nil || err.Error() != expectedErr {
+		t.Errorf("DecodeSizedBroadcast did not return the expected error "+
+			"when the size is too large.\nexpected: %s\nreceived: %+v",
+			expectedErr, err)
+	}
+}
+
+// Tests that MaxSizedBroadcastPayloadSize returns the correct max size.
+func TestMaxSizedBroadcastPayloadSize(t *testing.T) {
+	maxPayloadSize := 512
+	expectedSize := maxPayloadSize - sizedBroadcastMinSize
+	receivedSize := MaxSizedBroadcastPayloadSize(maxPayloadSize)
+	if receivedSize != expectedSize {
+		t.Errorf("Incorrect max paylaod size.\nexpected: %d\nreceived: %d",
+			expectedSize, receivedSize)
+	}
+}

broadcast/symmetricClient.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"github.com/pkg/errors"
+	jww "github.com/spf13/jwalterweatherman"
+	"gitlab.com/elixxir/client/cmix"
+	"gitlab.com/elixxir/client/cmix/identity"
+	"gitlab.com/elixxir/client/cmix/message"
+	crypto "gitlab.com/elixxir/crypto/broadcast"
+	"gitlab.com/elixxir/crypto/fastRNG"
+	"gitlab.com/elixxir/primitives/format"
+	"gitlab.com/xx_network/primitives/id"
+	"gitlab.com/xx_network/primitives/id/ephemeral"
+	"time"
+)
+
+// Error messages.
+const (
+	// symmetricClient.Broadcast
+	errNetworkHealth = "cannot send broadcast when the network is not healthy"
+	errPayloadSize   = "size of payload %d must be %d"
+)
+
+// Tags.
+const (
+	cMixSendTag                  = "SymmBcast"
+	symmetricBroadcastServiceTag = "SymmetricBroadcast"
+)
+
+// symmetricClient manages the sending and receiving of symmetric broadcast
+// messages on a given symmetric broadcast channel. Adheres to the Symmetric
+// interface.
+type symmetricClient struct {
+	channel crypto.Symmetric
+	net     Client
+	rng     *fastRNG.StreamGenerator
+}
+
+// Client contains the methods from cmix.Client that are required by
+// symmetricClient.
+type Client interface {
+	GetMaxMessageLength() int
+	Send(recipient *id.ID, fingerprint format.Fingerprint,
+		service message.Service, payload, mac []byte,
+		cMixParams cmix.CMIXParams) (id.Round, ephemeral.Id, error)
+	IsHealthy() bool
+	AddIdentity(id *id.ID, validUntil time.Time, persistent bool)
+	AddService(clientID *id.ID, newService message.Service,
+		response message.Processor)
+	DeleteClientService(clientID *id.ID)
+	RemoveIdentity(id *id.ID)
+}
+
+// NewSymmetricClient generates a new Symmetric for the given channel. It starts
+// listening for new messages on the callback immediately.
+func NewSymmetricClient(channel crypto.Symmetric, listenerCb ListenerFunc,
+	net Client, rng *fastRNG.StreamGenerator) Symmetric {
+	// Add channel's identity
+	net.AddIdentity(channel.ReceptionID, identity.Forever, true)
+
+	// Create new service
+	service := message.Service{
+		Identifier: channel.ReceptionID.Bytes(),
+		Tag:        symmetricBroadcastServiceTag,
+	}
+
+	// Create new message processor
+	p := &processor{
+		s:  &channel,
+		cb: listenerCb,
+	}
+
+	// Add service
+	net.AddService(channel.ReceptionID, service, p)
+
+	jww.INFO.Printf("New symmetric broadcast client created for channel %q (%s)",
+		channel.Name, channel.ReceptionID)
+
+	return &symmetricClient{
+		channel: channel,
+		net:     net,
+		rng:     rng,
+	}
+}
+
+// MaxPayloadSize returns the maximum size for a broadcasted payload.
+func (s *symmetricClient) MaxPayloadSize() int {
+	return s.net.GetMaxMessageLength()
+}
+
+// Get returns the crypto.Symmetric object containing the cryptographic and
+// identifying information about the channel.
+func (s *symmetricClient) Get() crypto.Symmetric {
+	return s.channel
+}
+
+// Broadcast broadcasts the payload to the channel.
+func (s *symmetricClient) Broadcast(payload []byte, cMixParams cmix.CMIXParams) (
+	id.Round, ephemeral.Id, error) {
+	if !s.net.IsHealthy() {
+		return 0, ephemeral.Id{}, errors.New(errNetworkHealth)
+	}
+
+	if len(payload) != s.MaxPayloadSize() {
+		return 0, ephemeral.Id{},
+			errors.Errorf(errPayloadSize, len(payload), s.MaxPayloadSize())
+	}
+
+	// Encrypt payload
+	rng := s.rng.GetStream()
+	encryptedPayload, mac, fp := s.channel.Encrypt(payload, rng)
+	rng.Close()
+
+	// Create service
+	service := message.Service{
+		Identifier: s.channel.ReceptionID.Bytes(),
+		Tag:        symmetricBroadcastServiceTag,
+	}
+
+	if cMixParams.DebugTag == cmix.DefaultDebugTag {
+		cMixParams.DebugTag = cMixSendTag
+	}
+
+	return s.net.Send(
+		s.channel.ReceptionID, fp, service, encryptedPayload, mac, cMixParams)
+}
+
+// Stop unregisters the listener callback and stops the channel's identity
+// from being tracked.
+func (s *symmetricClient) Stop() {
+	// Removes currently tracked identity
+	s.net.RemoveIdentity(s.channel.ReceptionID)
+
+	// Delete all registered services
+	s.net.DeleteClientService(s.channel.ReceptionID)
+}

broadcast/symmetricClient_test.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"bytes"
+	"fmt"
+	"gitlab.com/elixxir/client/cmix"
+	"gitlab.com/elixxir/client/cmix/identity/receptionID"
+	"gitlab.com/elixxir/client/cmix/rounds"
+	crypto "gitlab.com/elixxir/crypto/broadcast"
+	cMixCrypto "gitlab.com/elixxir/crypto/cmix"
+	"gitlab.com/elixxir/crypto/fastRNG"
+	"gitlab.com/xx_network/crypto/csprng"
+	"gitlab.com/xx_network/primitives/id"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+)
+
+// Tests that symmetricClient adheres to the Symmetric interface.
+var _ Symmetric = (*symmetricClient)(nil)
+
+// Tests that symmetricClient adheres to the Symmetric interface.
+var _ Client = (cmix.Client)(nil)
+
+// Tests that all clients listening on a symmetric broadcast channel receive the
+// message that is broadcasted.
+func Test_symmetricClient_Smoke(t *testing.T) {
+	// Initialise objects used by all clients
+	cMixHandler := newMockCmixHandler()
+	rngGen := fastRNG.NewStreamGenerator(1000, 10, csprng.NewSystemRNG)
+	channel := crypto.Symmetric{
+		ReceptionID: id.NewIdFromString("ReceptionID", id.User, t),
+		Name:        "MyChannel",
+		Description: "This is my channel about stuff.",
+		Salt:        cMixCrypto.NewSalt(csprng.NewSystemRNG(), 32),
+		RsaPubKey:   newRsaPubKey(64, t),
+	}
+
+	// Set up callbacks, callback channels, and the symmetric clients
+	const n = 5
+	cbChans := make([]chan []byte, n)
+	clients := make([]Symmetric, n)
+	for i := range clients {
+		cbChan := make(chan []byte, 10)
+		cb := func(payload []byte, _ receptionID.EphemeralIdentity,
+			_ rounds.Round) {
+			cbChan <- payload
+		}
+
+		s := NewSymmetricClient(channel, cb, newMockCmix(cMixHandler), rngGen)
+
+		cbChans[i] = cbChan
+		clients[i] = s
+
+		// Test that Get returns the expected channel
+		if !reflect.DeepEqual(s.Get(), channel) {
+			t.Errorf("Client %d returned wrong channel."+
+				"\nexpected: %+v\nreceived: %+v", i, channel, s.Get())
+		}
+	}
+
+	// Send broadcast from each client
+	for i := range clients {
+		payload := make([]byte, newMockCmix(cMixHandler).GetMaxMessageLength())
+		copy(payload,
+			fmt.Sprintf("Hello from client %d of %d.", i, len(clients)))
+
+		// Start processes that waits for each client to receive broadcast
+		var wg sync.WaitGroup
+		for j := range cbChans {
+			wg.Add(1)
+			go func(i, j int, cbChan chan []byte) {
+				defer wg.Done()
+				select {
+				case r := <-cbChan:
+					if !bytes.Equal(payload, r) {
+						t.Errorf("Client %d failed to receive expected "+
+							"payload from client %d."+
+							"\nexpected: %q\nreceived: %q", j, i, payload, r)
+					}
+				case <-time.After(25 * time.Millisecond):
+					t.Errorf("Client %d timed out waiting for broadcast "+
+						"payload from client %d.", j, i)
+				}
+			}(i, j, cbChans[j])
+		}
+
+		// Broadcast payload
+		_, _, err := clients[i].Broadcast(payload, cmix.GetDefaultCMIXParams())
+		if err != nil {
+			t.Errorf("Client %d failed to send broadcast: %+v", i, err)
+		}
+
+		// Wait for all clients to receive payload or time out
+		wg.Wait()
+	}
+
+	// Stop each client
+	for i := range clients {
+		clients[i].Stop()
+	}
+
+	payload := make([]byte, newMockCmix(cMixHandler).GetMaxMessageLength())
+	copy(payload, "This message should not get through.")
+
+	// Start waiting on channels and error if anything is received
+	var wg sync.WaitGroup
+	for i := range cbChans {
+		wg.Add(1)
+		go func(i int, cbChan chan []byte) {
+			defer wg.Done()
+			select {
+			case r := <-cbChan:
+				t.Errorf("Client %d received message: %q", i, r)
+			case <-time.After(25 * time.Millisecond):
+			}
+		}(i, cbChans[i])
+	}
+
+	// Broadcast payload
+	_, _, err := clients[0].Broadcast(payload, cmix.GetDefaultCMIXParams())
+	if err != nil {
+		t.Errorf("Client 0 failed to send broadcast: %+v", err)
+	}
+
+	wg.Wait()
+}

broadcast/utils_test.go

+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC                                           //
+//                                                                            //
+// Use of this source code is governed by a license that can be found in the  //
+// LICENSE file                                                               //
+////////////////////////////////////////////////////////////////////////////////
+
+package broadcast
+
+import (
+	"gitlab.com/elixxir/client/cmix"
+	"gitlab.com/elixxir/client/cmix/identity/receptionID"
+	"gitlab.com/elixxir/client/cmix/message"
+	"gitlab.com/elixxir/client/cmix/rounds"
+	"gitlab.com/elixxir/primitives/format"
+	"gitlab.com/xx_network/crypto/signature/rsa"
+	"gitlab.com/xx_network/primitives/id"
+	"gitlab.com/xx_network/primitives/id/ephemeral"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+)
+
+// newRsaPubKey generates a new random RSA public key for testing.
+func newRsaPubKey(seed int64, t *testing.T) *rsa.PublicKey {
+	prng := rand.New(rand.NewSource(seed))
+	privKey, err := rsa.GenerateKey(prng, 64)
+	if err != nil {
+		t.Errorf("Failed to generate new RSA key: %+v", err)
+	}
+
+	return privKey.GetPublic()
+}
+
+////////////////////////////////////////////////////////////////////////////////
+// Mock cMix Client                                                           //
+////////////////////////////////////////////////////////////////////////////////
+
+type mockCmixHandler struct {
+	processorMap map[id.ID]map[string][]message.Processor
+	sync.Mutex
+}
+
+func newMockCmixHandler() *mockCmixHandler {
+	return &mockCmixHandler{
+		processorMap: make(map[id.ID]map[string][]message.Processor),
+	}
+}
+
+type mockCmix struct {
+	numPrimeBytes int
+	health        bool
+	handler       *mockCmixHandler
+}
+
+func newMockCmix(handler *mockCmixHandler) *mockCmix {
+	return &mockCmix{
+		numPrimeBytes: 4096,
+		health:        true,
+		handler:       handler,
+	}
+}
+
+func (m *mockCmix) GetMaxMessageLength() int {
+	return format.NewMessage(m.numPrimeBytes).ContentsSize()
+}
+
+func (m *mockCmix) Send(recipient *id.ID, fingerprint format.Fingerprint,
+	service message.Service, payload, mac []byte, _ cmix.CMIXParams) (
+	id.Round, ephemeral.Id, error) {
+	msg := format.NewMessage(m.numPrimeBytes)
+	msg.SetContents(payload)
+	msg.SetMac(mac)
+	msg.SetKeyFP(fingerprint)
+
+	m.handler.Lock()
+	defer m.handler.Unlock()
+	for _, p := range m.handler.processorMap[*recipient][service.Tag] {
+		p.Process(msg, receptionID.EphemeralIdentity{}, rounds.Round{})
+	}
+
+	return 0, ephemeral.Id{}, nil
+}
+
+func (m *mockCmix) IsHealthy() bool {
+	return m.health
+}
+
+func (m *mockCmix) AddIdentity(id *id.ID, _ time.Time, _ bool) {
+	m.handler.Lock()
+	defer m.handler.Unlock()
+
+	if _, exists := m.handler.processorMap[*id]; exists {
+		return
+	}
+
+	m.handler.processorMap[*id] = make(map[string][]message.Processor)
+}
+
+func (m *mockCmix) AddService(clientID *id.ID, newService message.Service,
+	response message.Processor) {
+	m.handler.Lock()
+	defer m.handler.Unlock()
+
+	if _, exists := m.handler.processorMap[*clientID][newService.Tag]; !exists {
+		m.handler.processorMap[*clientID][newService.Tag] =
+			[]message.Processor{response}
+		return
+	}
+
+	m.handler.processorMap[*clientID][newService.Tag] =
+		append(m.handler.processorMap[*clientID][newService.Tag], response)
+}
+
+func (m *mockCmix) DeleteClientService(clientID *id.ID) {
+	m.handler.Lock()
+	defer m.handler.Unlock()
+
+	for tag := range m.handler.processorMap[*clientID] {
+		delete(m.handler.processorMap[*clientID], tag)
+	}
+}
+
+func (m *mockCmix) RemoveIdentity(id *id.ID) {
+	m.handler.Lock()
+	defer m.handler.Unlock()
+
+	delete(m.handler.processorMap, *id)
+}

cmix/identity/tracker.go

@@ -142,7 +142,7 @@ func (t manager) StartProcesses() stoppable.Stoppable {
 func (t *manager) AddIdentity(id *id.ID, validUntil time.Time, persistent bool) {
 	t.newIdentity <- TrackedID{
 		NextGeneration: netTime.Now().Add(-time.Second),
-		LastGeneration: time.Time{},
+		LastGeneration: netTime.Now().Add(-time.Duration(ephemeral.Period)),
 		Source:         id,
 		ValidUntil:     validUntil,
 		Persistent:     persistent,

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/spf13/viper v1.7.1
 	gitlab.com/elixxir/bloomfilter v0.0.0-20200930191214-10e9ac31b228
 	gitlab.com/elixxir/comms v0.0.4-0.20220323190139-9ed75f3a8b2c
-	gitlab.com/elixxir/crypto v0.0.7-0.20220420170330-979607dcc6da
+	gitlab.com/elixxir/crypto v0.0.7-0.20220425192911-a23209a58073
 	gitlab.com/elixxir/ekv v0.1.7
 	gitlab.com/elixxir/primitives v0.0.3-0.20220330212736-cce83b5f948f
 	gitlab.com/xx_network/comms v0.0.4-0.20220315161313-76acb14429ac

go.sum

@@ -303,6 +303,16 @@ gitlab.com/elixxir/crypto v0.0.7-0.20220418163058-a76028e93dd3 h1:tYr7CjBj3p4tmU
 gitlab.com/elixxir/crypto v0.0.7-0.20220418163058-a76028e93dd3/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
 gitlab.com/elixxir/crypto v0.0.7-0.20220420170330-979607dcc6da h1:SZQTa7Gp2vPAwu3yQDhkiKPhFqQXCsiRi0s40e/em/Y=
 gitlab.com/elixxir/crypto v0.0.7-0.20220420170330-979607dcc6da/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
+gitlab.com/elixxir/crypto v0.0.7-0.20220421222239-fb92da398627 h1:vJ+hTCN0W+CHAUquqHseJC9+mWmTMdTViQbha7zJ/Rs=
+gitlab.com/elixxir/crypto v0.0.7-0.20220421222239-fb92da398627/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425183558-d1b1b40a5ba8 h1:HifQbDMGC9jgnsT6pGilE4JKaYbA/X6+YF9qe+FhjxY=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425183558-d1b1b40a5ba8/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425185754-5050a93bc64c h1:jUyeQpETz8iKbxZirZNUAlul66sUwzMxtKwCUJPrhBE=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425185754-5050a93bc64c/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425192754-f9f5ca6ad2eb h1:K10bnRhnZS6XFxkwGkI6wMLAWLN/lTy9Qjx6f6exSsE=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425192754-f9f5ca6ad2eb/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425192911-a23209a58073 h1:T0BK30t1F9M8RCdsuMN1bL34fSLNPqjqC+oufnjPh3I=
+gitlab.com/elixxir/crypto v0.0.7-0.20220425192911-a23209a58073/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
 gitlab.com/elixxir/ekv v0.1.6 h1:M2hUSNhH/ChxDd+s8xBqSEKgoPtmE6hOEBqQ73KbN6A=
 gitlab.com/elixxir/ekv v0.1.6/go.mod h1:e6WPUt97taFZe5PFLPb1Dupk7tqmDCTQu1kkstqJvw4=
 gitlab.com/elixxir/ekv v0.1.7 h1:OW2z+N4QCqqMFzouAwFTWWMKz0Y/PDhyYReN7gQ5NiQ=

storage/utility/messageBuffer.go

@@ -230,7 +230,7 @@ func (mb *MessageBuffer) Add(m interface{}) interface{} {
 	// Save buffer
 	err = mb.save()
 	if err != nil {
-		jww.FATAL.Panicf("Error whilse saving buffer: %v", err)
+		jww.FATAL.Panicf("Error while saving buffer: %v", err)
 	}
 
 	return m