finished send unsafe and send e2e

context/stoppable/single.go

@@ -32,7 +32,7 @@ func (s *Single) IsRunning() bool {
 }
 
 // Quit returns the read only channel it will send the stop signal on.
-func (s *Single) Quit() chan<- struct{} {
+func (s *Single) Quit() <-chan struct{} {
 	return s.quit
 }
 

network/manager.go

@@ -14,8 +14,10 @@ import (
 	"gitlab.com/elixxir/client/context"
 	"gitlab.com/elixxir/client/context/stoppable"
 	"gitlab.com/elixxir/client/network/health"
+	"gitlab.com/elixxir/client/network/parse"
 	"gitlab.com/elixxir/comms/client"
 	"gitlab.com/elixxir/comms/network"
+	"gitlab.com/elixxir/primitives/format"
 	"gitlab.com/xx_network/crypto/signature/rsa"
 	"gitlab.com/xx_network/primitives/id"
 	//	"gitlab.com/xx_network/primitives/ndf"
@@ -41,6 +43,9 @@ type Manager struct {
 	//contains the network instance
 	instance *network.Instance
 
+	//Partitioner
+	partitioner parse.Partitioner
+
 	//channels
 	nodeRegistration chan network.NodeGateway
 
@@ -74,6 +79,8 @@ func NewManager(ctx *context.Context) (*Manager, error) {
 			" client network manager")
 	}
 
+	msgSize := format.NewMessage(ctx.Session.Cmix().GetGroup().GetP().ByteLen()).ContentsSize()
+
 	cm := &Manager{
 		Comms:       comms,
 		Context:     ctx,
@@ -81,6 +88,7 @@ func NewManager(ctx *context.Context) (*Manager, error) {
 		health:      health.Init(ctx, 5*time.Second),
 		instance:    instance,
 		uid:         cryptoUser.GetUserID(),
+		partitioner: parse.NewPartitioner(msgSize, ctx),
 	}
 
 	return cm, nil

network/parse/partition.go

@@ -17,10 +17,10 @@ type Partitioner struct {
 	partContentsSize  int
 	deltaFirstPart    int
 	maxSize           int
-	ctx               context.Context
+	ctx               *context.Context
 }
 
-func NewPartitioner(messageSize int, ctx context.Context) Partitioner {
+func NewPartitioner(messageSize int, ctx *context.Context) Partitioner {
 	p := Partitioner{
 		baseMessageSize:   messageSize,
 		firstContentsSize: messageSize - firstHeaderLen,

network/send.go → network/sendE2E.go

@@ -10,13 +10,16 @@ import (
 	"github.com/pkg/errors"
 	"gitlab.com/elixxir/client/context/message"
 	"gitlab.com/elixxir/client/context/params"
+	"gitlab.com/elixxir/primitives/format"
 	"gitlab.com/xx_network/primitives/id"
+	"sync"
+	"time"
 )
 
 // SendE2E sends an end-to-end payload to the provided recipient with
 // the provided msgType. Returns the list of rounds in which parts of
 // the message were sent or an error if it fails.
-func (m *Manager) SendE2E(msg message.Send, e2eP params.E2E, cmixP params.CMIX) (
+func (m *Manager) SendE2E(msg message.Send, e2eP params.E2E) (
 	[]id.Round, error) {
 
 	if !m.health.IsRunning() {
@@ -24,15 +27,70 @@ func (m *Manager) SendE2E(msg message.Send, e2eP params.E2E, cmixP params.CMIX)
 			"network is not healthy")
 	}
 
-	return nil, nil
+	return m.sendE2E(msg, e2eP)
 }
 
-// SendUnsafe sends an unencrypted payload to the provided recipient
-// with the provided msgType. Returns the list of rounds in which parts
-// of the message were sent or an error if it fails.
-// NOTE: Do not use this function unless you know what you are doing.
-// This function always produces an error message in client logging.
-func (m *Manager) SendUnsafe(msg message.Send) ([]id.Round, error) {
-	return nil, nil
+func (m *Manager) sendE2E(msg message.Send, param params.E2E) ([]id.Round, error) {
+
+	//timestamp the message
+	ts := time.Now()
+
+	//partition the message
+	partitions, err := m.partitioner.Partition(msg.Recipient, msg.MessageType, ts,
+		msg.Payload)
+	if err != nil {
+		return nil, errors.WithMessage(err, "failed to send unsafe message")
+	}
+
+	//encrypt then send the partitions over cmix
+	roundIds := make([]id.Round, len(partitions))
+	errCh := make(chan error, len(partitions))
+
+	// get the key manager for the partner
+	partner, err := m.Context.Session.E2e().GetPartner(msg.Recipient)
+	if err != nil {
+		return nil, errors.WithMessagef(err, "Could not send End to End encrypted "+
+			"message, no relationship found with %s", partner)
+	}
+
+	wg := sync.WaitGroup{}
+
+	for i, p := range partitions {
+		//create the cmix message
+		msgCmix := format.NewMessage(m.Context.Session.Cmix().GetGroup().GetP().ByteLen())
+		msgCmix.SetContents(p)
+
+		//get a key to end to end encrypt
+		key, err := partner.GetKeyForSending(param.Type)
+		if err != nil {
+			return nil, errors.WithMessagef(err, "Failed to get key "+
+				"for end to end encryption")
 		}
 
+		//end to end encrypt the cmix message
+		msgEnc := key.Encrypt(msgCmix)
+
+		//send the cmix message, each partition in its own thread
+		wg.Add(1)
+		go func(i int) {
+			var err error
+			roundIds[i], err = m.sendCMIX(msgEnc, param.CMIX)
+			if err != nil {
+				errCh <- err
+			}
+			wg.Done()
+		}(i)
+	}
+
+	wg.Wait()
+
+	//see if any parts failed to send
+	numFail, errRtn := getSendErrors(errCh)
+	if numFail > 0 {
+		return nil, errors.Errorf("Failed to E2E send %v/%v sub payloads:"+
+			" %s", numFail, len(partitions), errRtn)
+	}
+
+	//return the rounds if everything send successfully
+	return roundIds, nil
+}

network/sendUnsafe.go

 package network
+
+import (
+	"github.com/pkg/errors"
+	jww "github.com/spf13/jwalterweatherman"
+	"gitlab.com/elixxir/client/context/message"
+	"gitlab.com/elixxir/client/context/params"
+	"gitlab.com/elixxir/crypto/e2e"
+	"gitlab.com/elixxir/primitives/format"
+	"gitlab.com/xx_network/primitives/id"
+	"sync"
+	"time"
+)
+
+// SendUnsafe sends an unencrypted payload to the provided recipient
+// with the provided msgType. Returns the list of rounds in which parts
+// of the message were sent or an error if it fails.
+// NOTE: Do not use this function unless you know what you are doing.
+// This function always produces an error message in client logging.
+func (m *Manager) SendUnsafe(msg message.Send, param params.Unsafe) ([]id.Round, error) {
+	if !m.health.IsRunning() {
+		return nil, errors.New("cannot send unsafe message when the " +
+			"network is not healthy")
+	}
+
+	jww.WARN.Println("Sending unsafe message. Unsafe payloads have no end" +
+		" to end encryption, they have limited security and privacy " +
+		"preserving properties")
+
+	return m.sendUnsafe(msg, param)
+}
+
+func (m *Manager) sendUnsafe(msg message.Send, param params.Unsafe) ([]id.Round, error) {
+
+	//timestamp the message
+	ts := time.Now()
+
+	//partition the message
+	partitions, err := m.partitioner.Partition(msg.Recipient, msg.MessageType, ts,
+		msg.Payload)
+
+	if err != nil {
+		return nil, errors.WithMessage(err, "failed to send unsafe message")
+	}
+
+	//send the partitions over cmix
+	roundIds := make([]id.Round, len(partitions))
+	errCh := make(chan error, len(partitions))
+
+	wg := sync.WaitGroup{}
+
+	for i, p := range partitions {
+		msgCmix := format.NewMessage(m.Context.Session.Cmix().GetGroup().GetP().ByteLen())
+		msgCmix.SetContents(p)
+		e2e.SetUnencrypted(msgCmix, msg.Recipient)
+		wg.Add(1)
+		go func(i int) {
+			var err error
+			roundIds[i], err = m.sendCMIX(msgCmix, param.CMIX)
+			if err != nil {
+				errCh <- err
+			}
+			wg.Done()
+		}(i)
+	}
+
+	//see if any parts failed to send
+	numFail, errRtn := getSendErrors(errCh)
+	if numFail > 0 {
+		return nil, errors.Errorf("Failed to send %v/%v sub payloads:"+
+			" %s", numFail, len(partitions), errRtn)
+	}
+
+	//return the rounds if everything send successfully
+	return roundIds, nil
+}
+
+//returns any errors on the error channel
+func getSendErrors(c chan error) (int, string) {
+	var errRtn string
+	numFail := 0
+	done := false
+	for !done {
+		select {
+		case err := <-c:
+			errRtn += err.Error()
+			numFail++
+		default:
+			done = true
+		}
+	}
+	return numFail, errRtn
+}

storage/e2e/manager.go

@@ -9,12 +9,11 @@ package e2e
 import (
 	"fmt"
 	"github.com/pkg/errors"
-	jww "github.com/spf13/jwalterweatherman"
 	"gitlab.com/elixxir/client/context/params"
 	"gitlab.com/elixxir/client/storage/versioned"
 	"gitlab.com/elixxir/crypto/cyclic"
-	"gitlab.com/xx_network/primitives/id"
 	dh "gitlab.com/elixxir/crypto/diffieHellman"
+	"gitlab.com/xx_network/primitives/id"
 )
 
 const managerPrefix = "Manager{partner:%s}"
@@ -121,7 +120,7 @@ func (m *Manager) NewReceiveSession(partnerPubKey *cyclic.Int, params SessionPar
 // none is passed
 func (m *Manager) NewSendSession(myPrivKey *cyclic.Int, params SessionParams) *Session {
 	//find the latest public key from the other party
-	sourceSession := m.receive.GetNewestRekeyableSession()
+	sourceSession := m.receive.getNewestRekeyableSession()
 
 	//create the session
 	session := newSession(m, myPrivKey, sourceSession.partnerPubKey, nil,
@@ -134,20 +133,20 @@ func (m *Manager) NewSendSession(myPrivKey *cyclic.Int, params SessionParams) *S
 }
 
 // gets the correct session to send with depending on the type of send
-func (m *Manager) GetSessionForSending(st params.SendType) *Session {
+func (m *Manager) GetKeyForSending(st params.SendType) (*Key, error) {
 	switch st {
 	case params.Standard:
-		return m.send.GetSessionForSending()
+		return m.send.getKeyForSending()
 	case params.KeyExchange:
-		return m.send.GetNewestRekeyableSession()
+		return m.send.getKeyForRekey()
 	default:
-		jww.ERROR.Printf("Cannot get session for invalid Send Type: %s",
-			st)
 	}
 
-	return nil
+	return nil, errors.Errorf("Cannot get session for invalid "+
+		"Send Type: %s", st)
 }
 
+
 // gets the send session of the passed ID. Returns nil if no session is found
 func (m *Manager) GetSendSession(sessionID SessionID) *Session {
 	return m.send.GetByID(sessionID)

storage/e2e/sessionBuff.go

@@ -30,6 +30,7 @@ type sessionBuff struct {
 	key string
 
 	mux     sync.RWMutex
+	sendMux sync.Mutex
 }
 
 func NewSessionBuff(manager *Manager, key string) *sessionBuff {
@@ -151,14 +152,23 @@ func (sb *sessionBuff) GetNewest() *Session {
 	return sb.sessions[0]
 }
 
-// returns the session which is most likely to be successful for sending
-func (sb *sessionBuff) GetSessionForSending() *Session {
-	//dont need to take the lock due to the use of a copy of the buffer
-	sessions := sb.getInternalBufferShallowCopy()
-	if len(sessions) == 0 {
-		return nil
+// returns the key  which is most likely to be successful for sending
+func (sb *sessionBuff) getKeyForSending() (*Key, error) {
+	sb.sendMux.Lock()
+	defer sb.sendMux.Unlock()
+	s := sb.getSessionForSending()
+	if s == nil {
+		return nil, errors.New("Failed to find a session for sending")
+	}
+
+	return s.PopKey()
 }
 
+
+// returns the session which is most likely to be successful for sending
+func (sb *sessionBuff) getSessionForSending() *Session {
+	sessions := sb.sessions
+
 	var confirmedRekey []*Session
 	var unconfirmedActive []*Session
 	var unconfirmedRekey []*Session
@@ -204,8 +214,20 @@ func (sb *sessionBuff) TriggerNegotiation() []*Session {
 	return instructions
 }
 
+// returns the key  which is most likely to be successful for sending
+func (sb *sessionBuff) getKeyForRekey() (*Key, error) {
+	sb.sendMux.Lock()
+	defer sb.sendMux.Unlock()
+	s := sb.getNewestRekeyableSession()
+	if s == nil {
+		return nil, errors.New("Failed to find a session for rekeying")
+	}
+
+	return s.PopReKey()
+}
+
 // returns the newest session which can be used to start a key negotiation
-func (sb *sessionBuff) GetNewestRekeyableSession() *Session {
+func (sb *sessionBuff) getNewestRekeyableSession() *Session {
 	//dont need to take the lock due to the use of a copy of the buffer
 	sessions := sb.getInternalBufferShallowCopy()