diff --git a/Makefile b/Makefile
index b6f9df22644b196178f6fa17222f586149198794..64cdc581e683cdbb63240888c31193b0fe10eb3d 100644
--- a/Makefile
+++ b/Makefile
@@ -20,7 +20,7 @@ update_release:
GOFLAGS="" go get gitlab.com/xx_network/primitives@release
GOFLAGS="" go get gitlab.com/elixxir/primitives@release
GOFLAGS="" go get gitlab.com/xx_network/crypto@release
- GOFLAGS="" go get gitlab.com/elixxir/crypto@release
+ GOFLAGS="" go get gitlab.com/elixxir/crypto@singleUseMultiPartRequest
GOFLAGS="" go get gitlab.com/xx_network/comms@release
GOFLAGS="" go get gitlab.com/elixxir/comms@release
GOFLAGS="" go get gitlab.com/elixxir/ekv@master
diff --git a/go.mod b/go.mod
index 6a99bc114b871f87af2b56acb96e2ff603f2dda9..417152e5eba15cd54e8c92b4b6f92bca92a9e325 100644
--- a/go.mod
+++ b/go.mod
@@ -13,8 +13,8 @@ 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.20220415180223-ec8d560270a1
- gitlab.com/elixxir/ekv v0.1.7
+ gitlab.com/elixxir/crypto v0.0.7-0.20220418163058-a76028e93dd3
+ gitlab.com/elixxir/ekv v0.1.6
gitlab.com/elixxir/primitives v0.0.3-0.20220330212736-cce83b5f948f
gitlab.com/xx_network/comms v0.0.4-0.20220315161313-76acb14429ac
gitlab.com/xx_network/crypto v0.0.5-0.20220317171841-084640957d71
diff --git a/go.sum b/go.sum
index 8b274f94dffe0f1a5eb524a70567efb96e5e82cd..e763825e430744f3ed713446cbe28f812517f615 100644
--- a/go.sum
+++ b/go.sum
@@ -283,7 +283,6 @@ gitlab.com/elixxir/crypto v0.0.3/go.mod h1:ZNgBOblhYToR4m8tj4cMvJ9UsJAUKq+p0gCp0
gitlab.com/elixxir/crypto v0.0.7-0.20220222221347-95c7ae58da6b/go.mod h1:tD6XjtQh87T2nKZL5I/pYPck5M2wLpkZ1Oz7H/LqO10=
gitlab.com/elixxir/crypto v0.0.7-0.20220309234716-1ba339865787 h1:+qmsWov412+Yn7AKUhTbOcDgAydNXlNLPmFpO2W5LwY=
gitlab.com/elixxir/crypto v0.0.7-0.20220309234716-1ba339865787/go.mod h1:tD6XjtQh87T2nKZL5I/pYPck5M2wLpkZ1Oz7H/LqO10=
-gitlab.com/elixxir/crypto v0.0.7-0.20220317172048-3de167bd9406/go.mod h1:tD6XjtQh87T2nKZL5I/pYPck5M2wLpkZ1Oz7H/LqO10=
gitlab.com/elixxir/crypto v0.0.7-0.20220325215559-7489d68d7714 h1:epnov8zyFWod14MUNtGHSbZCVSkZjN4NvoiBs1TgEV8=
gitlab.com/elixxir/crypto v0.0.7-0.20220325215559-7489d68d7714/go.mod h1:tD6XjtQh87T2nKZL5I/pYPck5M2wLpkZ1Oz7H/LqO10=
gitlab.com/elixxir/crypto v0.0.7-0.20220325224306-705ce59288bb h1:WdlmG+KPaM2Pjo1EFiFFPYEVSMV64Di1CitQnXGWBOQ=
@@ -298,14 +297,8 @@ gitlab.com/elixxir/crypto v0.0.7-0.20220331001626-1829e71edf56 h1:1HJHlRwh3dDbvw
gitlab.com/elixxir/crypto v0.0.7-0.20220331001626-1829e71edf56/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
gitlab.com/elixxir/crypto v0.0.7-0.20220406193349-d25222ea3c6e h1:P+E0+AdevTNWBdqf4+covcmTrRfe6rKPLtevFrjbKQA=
gitlab.com/elixxir/crypto v0.0.7-0.20220406193349-d25222ea3c6e/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
-gitlab.com/elixxir/crypto v0.0.7-0.20220414175442-6d2304df43d7 h1:xEE795GeUyQaa4lRAI8IjyH31glm2OvFgzY9eiMEr1M=
-gitlab.com/elixxir/crypto v0.0.7-0.20220414175442-6d2304df43d7/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
-gitlab.com/elixxir/crypto v0.0.7-0.20220414225314-6f3eb9c073a5 h1:yw3G8ZEiWu2eSZWRQmj6nBhiJIYK3Cw2MJzDPkNHYVA=
-gitlab.com/elixxir/crypto v0.0.7-0.20220414225314-6f3eb9c073a5/go.mod h1:tD6XjtQh87T2nKZL5I/pYPck5M2wLpkZ1Oz7H/LqO10=
-gitlab.com/elixxir/crypto v0.0.7-0.20220415172207-7de5e3cdb340 h1:f1JsT60cKFXcHPoaOD1ohIOA22FQd42vbKjF9wrKfNs=
-gitlab.com/elixxir/crypto v0.0.7-0.20220415172207-7de5e3cdb340/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
-gitlab.com/elixxir/crypto v0.0.7-0.20220415180223-ec8d560270a1 h1:stzHgpYxHQu3JgvQu5Vr0hr4nzJUk5CxspZEoNx26eQ=
-gitlab.com/elixxir/crypto v0.0.7-0.20220415180223-ec8d560270a1/go.mod h1:JkByWX/TXCjdu6pRJsx+jwttbBGvlAljYSJMImDmt+4=
+gitlab.com/elixxir/crypto v0.0.7-0.20220418163058-a76028e93dd3 h1:tYr7CjBj3p4tmUmvEmsX5n0m0GsfG8eJOu1YLoBHE2g=
+gitlab.com/elixxir/crypto v0.0.7-0.20220418163058-a76028e93dd3/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=
diff --git a/single/cypher.go b/single/cypher.go
index 44a5c96287528e6ae02b9afca5699f97dc5ebd14..1e51d5ecb8acabebd274b6c8cb4ce06a9b9d6d24 100644
--- a/single/cypher.go
+++ b/single/cypher.go
@@ -18,15 +18,21 @@ import (
"gitlab.com/elixxir/primitives/format"
)
+type newKeyFn func(dhKey *cyclic.Int, keyNum uint64) []byte
+type newFpFn func(dhKey *cyclic.Int, keyNum uint64) format.Fingerprint
+
// makeCyphers generates all fingerprints for a given number of messages.
-func makeCyphers(dhKey *cyclic.Int, messageCount uint8) []cypher {
+func makeCyphers(dhKey *cyclic.Int, messageCount uint8, newKey newKeyFn,
+ newFp newFpFn) []cypher {
cypherList := make([]cypher, messageCount)
for i := uint8(0); i < messageCount; i++ {
cypherList[i] = cypher{
- dhKey: dhKey,
- num: i,
+ dhKey: dhKey,
+ num: i,
+ newKey: newKey,
+ newFp: newFp,
}
}
@@ -34,16 +40,18 @@ func makeCyphers(dhKey *cyclic.Int, messageCount uint8) []cypher {
}
type cypher struct {
- dhKey *cyclic.Int
- num uint8
+ dhKey *cyclic.Int
+ num uint8
+ newKey newKeyFn // Function used to create new key
+ newFp newFpFn // Function used to create new fingerprint
}
func (rk *cypher) getKey() []byte {
- return singleUse.NewResponseKey(rk.dhKey, uint64(rk.num))
+ return rk.newKey(rk.dhKey, uint64(rk.num))
}
func (rk *cypher) GetFingerprint() format.Fingerprint {
- return singleUse.NewResponseFingerprint(rk.dhKey, uint64(rk.num))
+ return rk.newFp(rk.dhKey, uint64(rk.num))
}
func (rk *cypher) Encrypt(rp message.ResponsePart) (
diff --git a/single/listener.go b/single/listener.go
index ce2bffbb019578c310d684dc3190a6bb24a24db5..c41d5fde017e715cc4145e8fa1fb952e061d01cd 100644
--- a/single/listener.go
+++ b/single/listener.go
@@ -14,7 +14,7 @@ import (
)
type Receiver interface {
- Callback(*Request, receptionID.EphemeralIdentity, rounds.Round)
+ Callback(*Request, receptionID.EphemeralIdentity, []rounds.Round)
}
type Listener interface {
@@ -74,7 +74,7 @@ func (l *listener) Process(ecrMsg format.Message,
// Generate DH key and symmetric key
senderPubkey := requestMsg.GetPubKey(l.grp)
dhKey := l.grp.Exp(senderPubkey, l.myPrivKey, l.grp.NewInt(1))
- key := singleUse.NewTransmitKey(dhKey)
+ key := singleUse.NewRequestKey(dhKey)
// Verify the MAC
if !singleUse.VerifyMAC(key, requestMsg.GetPayload(), ecrMsg.GetMac()) {
@@ -95,20 +95,56 @@ func (l *listener) Process(ecrMsg format.Message,
return
}
- used := uint32(0)
-
- r := Request{
- sender: payload.GetRecipientID(requestMsg.GetPubKey(l.grp)),
- senderPubKey: senderPubkey,
- dhKey: dhKey,
- tag: l.tag,
- maxParts: 0,
- used: &used,
- requestPayload: payload.GetContents(),
- net: l.net,
+ cbFunc := func(payloadContents []byte, rounds []rounds.Round) {
+ used := uint32(0)
+
+ r := Request{
+ sender: payload.GetRecipientID(requestMsg.GetPubKey(l.grp)),
+ senderPubKey: senderPubkey,
+ dhKey: dhKey,
+ tag: l.tag,
+ maxParts: payload.GetMaxResponseParts(),
+ used: &used,
+ requestPayload: payloadContents,
+ net: l.net,
+ }
+
+ go l.cb.Callback(&r, receptionID, rounds)
}
- go l.cb.Callback(&r, receptionID, round)
+ if numParts := payload.GetNumParts(); numParts > 1 {
+ c := message.NewCollator(numParts)
+ _, _, err = c.Collate(payload)
+ if err != nil {
+
+ return
+ }
+ cyphers := makeCyphers(dhKey, numParts,
+ singleUse.NewRequestPartKey, singleUse.NewRequestPartFingerprint)
+ ridCollector := newRoundIdCollector(int(numParts))
+ for i, cy := range cyphers {
+ key = singleUse.NewRequestPartKey(dhKey, uint64(i+1))
+ fp = singleUse.NewRequestPartFingerprint(dhKey, uint64(i+1))
+ p := &requestPartProcessor{
+ myId: l.myId,
+ tag: l.tag,
+ cb: cbFunc,
+ c: c,
+ cy: cy,
+ roundIDs: ridCollector,
+ }
+ err = l.net.AddFingerprint(l.myId, fp, p)
+ if err != nil {
+ jww.ERROR.Printf("Failed to add fingerprint for request part "+
+ "%d of %d (%s): %+v", i, numParts, l.tag, err)
+ return
+ }
+ }
+
+ l.net.CheckInProgressMessages()
+ } else {
+ cbFunc(payload.GetContents(), []rounds.Round{round})
+ }
}
func (l *listener) Stop() {
diff --git a/single/message/collator.go b/single/message/collator.go
index d2eff3a366f657e1d21f3c031cabd74a3b6f0850..9e59eb9e95f00b5b4ec8b0550bf403523e130cec 100644
--- a/single/message/collator.go
+++ b/single/message/collator.go
@@ -9,10 +9,9 @@ import (
// Error messages.
const (
// Collate
- errUnmarshalResponsePart = "failed to unmarshal response payload: %+v"
- errMaxParts = "max number of parts reported by payload %d is larger than collator expected (%d)"
- errPartOutOfRange = "payload part number %d greater than max number of expected parts (%d)"
- errPartExists = "a payload for the part number %d already exists in the list"
+ errMaxParts = "max number of parts reported by payload %d is larger than collator expected (%d)"
+ errPartOutOfRange = "payload part number %d greater than max number of expected parts (%d)"
+ errPartExists = "a payload for the part number %d already exists in the list"
)
// Initial value of the Collator maxNum that indicates it has yet to be set.
@@ -26,6 +25,17 @@ type Collator struct {
sync.Mutex
}
+type Part interface {
+ // GetNumParts returns the total number of parts in the message.
+ GetNumParts() uint8
+
+ // GetPartNum returns the index of this part in the message.
+ GetPartNum() uint8
+
+ // GetContents returns the contents of the message part.
+ GetContents() []byte
+}
+
// NewCollator generates an empty list of payloads to fit the max number of
// possible messages. maxNum is set to indicate that it is not yet set.
func NewCollator(messageCount uint8) *Collator {
@@ -38,38 +48,33 @@ func NewCollator(messageCount uint8) *Collator {
// Collate collects message payload parts. Once all parts are received, the full
// collated payload is returned along with true. Otherwise, returns false.
-func (c *Collator) Collate(payloadBytes []byte) ([]byte, bool, error) {
- payload, err := UnmarshalResponsePart(payloadBytes)
- if err != nil {
- return nil, false, errors.Errorf(errUnmarshalResponsePart, err)
- }
-
+func (c *Collator) Collate(part Part) ([]byte, bool, error) {
c.Lock()
defer c.Unlock()
// If this is the first message received, then set the max number of
// messages expected to be received off its max number of parts
if c.maxNum == unsetCollatorMax {
- if int(payload.GetNumParts()) > len(c.payloads) {
+ if int(part.GetNumParts()) > len(c.payloads) {
return nil, false, errors.Errorf(
- errMaxParts, payload.GetNumParts(), len(c.payloads))
+ errMaxParts, part.GetNumParts(), len(c.payloads))
}
- c.maxNum = int(payload.GetNumParts())
+ c.maxNum = int(part.GetNumParts())
}
// Make sure that the part number is within the expected number of parts
- if int(payload.GetPartNum()) >= c.maxNum {
+ if int(part.GetPartNum()) >= c.maxNum {
return nil, false,
- errors.Errorf(errPartOutOfRange, payload.GetPartNum(), c.maxNum)
+ errors.Errorf(errPartOutOfRange, part.GetPartNum(), c.maxNum)
}
// Make sure no payload with the same part number exists
- if c.payloads[payload.GetPartNum()] != nil {
- return nil, false, errors.Errorf(errPartExists, payload.GetPartNum())
+ if c.payloads[part.GetPartNum()] != nil {
+ return nil, false, errors.Errorf(errPartExists, part.GetPartNum())
}
// Add the payload to the list
- c.payloads[payload.GetPartNum()] = payload.GetContents()
+ c.payloads[part.GetPartNum()] = part.GetContents()
c.count++
// Return false if not all messages have been received
diff --git a/single/message/collator_test.go b/single/message/collator_test.go
index 9ad8312525a246fd14e9bd21f67701cd0125af91..617697db480b6dd6f6375757b326952db17021f9 100644
--- a/single/message/collator_test.go
+++ b/single/message/collator_test.go
@@ -4,7 +4,6 @@ import (
"bytes"
"fmt"
"reflect"
- "strings"
"testing"
)
@@ -29,16 +28,17 @@ func TestNewCollator(t *testing.T) {
func TestCollator_Collate(t *testing.T) {
messageCount := 16
msgPayloadSize := 2
- msgParts := map[int]ResponsePart{}
+ msgParts := map[int]mockPart{}
expectedData := make([]byte, messageCount*msgPayloadSize)
copy(expectedData, "This is the expected final data.")
buff := bytes.NewBuffer(expectedData)
for i := 0; i < messageCount; i++ {
- msgParts[i] = NewResponsePart(msgPayloadSize + 5)
- msgParts[i].SetNumParts(uint8(messageCount))
- msgParts[i].SetPartNum(uint8(i))
- msgParts[i].SetContents(buff.Next(msgPayloadSize))
+ msgParts[i] = mockPart{
+ numParts: uint8(messageCount),
+ partNum: uint8(i),
+ contents: buff.Next(msgPayloadSize),
+ }
}
c := NewCollator(uint8(messageCount))
@@ -51,7 +51,7 @@ func TestCollator_Collate(t *testing.T) {
var err error
var collated bool
- fullPayload, collated, err = c.Collate(part.Marshal())
+ fullPayload, collated, err = c.Collate(part)
if err != nil {
t.Errorf("Collate returned an error for part #%d: %+v", j, err)
}
@@ -71,31 +71,12 @@ func TestCollator_Collate(t *testing.T) {
}
}
-// Error path: the byte slice cannot be unmarshaled.
-func TestCollator_collate_UnmarshalError(t *testing.T) {
- payloadBytes := []byte{1}
- c := NewCollator(1)
- payload, collated, err := c.Collate(payloadBytes)
- expectedErr := strings.Split(errUnmarshalResponsePart, "%")[0]
-
- if err == nil || !strings.Contains(err.Error(), expectedErr) {
- t.Errorf("Collate failed to return an error for failing to "+
- "unmarshal the payload.\nexpected: %s\nreceived: %+v",
- expectedErr, err)
- }
-
- if payload != nil || collated {
- t.Errorf("Collate signaled the payload was collated on error."+
- "\npayload: %+v\ncollated: %+v", payload, collated)
- }
-}
-
// Error path: max reported parts by payload larger than set in Collator.
func TestCollator_Collate_MaxPartsError(t *testing.T) {
- payloadBytes := []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF}
+ p := mockPart{0xFF, 0xFF, []byte{0xFF, 0xFF, 0xFF}}
messageCount := uint8(1)
c := NewCollator(messageCount)
- _, _, err := c.Collate(payloadBytes)
+ _, _, err := c.Collate(p)
expectedErr := fmt.Sprintf(errMaxParts, 0xFF, messageCount)
if err == nil || err.Error() != expectedErr {
@@ -107,11 +88,11 @@ func TestCollator_Collate_MaxPartsError(t *testing.T) {
// Error path: the message part number is greater than the max number of parts.
func TestCollator_Collate_PartNumTooLargeError(t *testing.T) {
- payloadBytes := []byte{25, 5, 5, 5, 5}
- partNum := uint8(5)
- c := NewCollator(partNum)
- _, _, err := c.Collate(payloadBytes)
- expectedErr := fmt.Sprintf(errPartOutOfRange, partNum, c.maxNum)
+ p := mockPart{35, 5, []byte{5, 5, 5}}
+ messageCount := uint8(1)
+ c := NewCollator(messageCount)
+ _, _, err := c.Collate(p)
+ expectedErr := fmt.Sprintf(errMaxParts, p.numParts, messageCount)
if err == nil || err.Error() != expectedErr {
t.Errorf("Collate failed to return the expected error when the part "+
@@ -122,17 +103,28 @@ func TestCollator_Collate_PartNumTooLargeError(t *testing.T) {
// Error path: a message with the part number already exists.
func TestCollator_Collate_PartExistsError(t *testing.T) {
- payloadBytes := []byte{0, 1, 5, 0, 1, 20}
+ p := mockPart{5, 1, []byte{5, 0, 1, 20}}
c := NewCollator(5)
- _, _, err := c.Collate(payloadBytes)
+ _, _, err := c.Collate(p)
if err != nil {
t.Fatalf("Collate returned an error: %+v", err)
}
- expectedErr := fmt.Sprintf(errPartExists, payloadBytes[1])
+ expectedErr := fmt.Sprintf(errPartExists, p.partNum)
- _, _, err = c.Collate(payloadBytes)
+ _, _, err = c.Collate(p)
if err == nil || err.Error() != expectedErr {
t.Errorf("Collate failed to return an error when the part number "+
"already exists.\nexpected: %s\nreceived: %+v", expectedErr, err)
}
}
+
+type mockPart struct {
+ numParts uint8
+ partNum uint8
+ contents []byte
+}
+
+func (m mockPart) GetNumParts() uint8 { return m.numParts }
+func (m mockPart) GetPartNum() uint8 { return m.partNum }
+func (m mockPart) GetContents() []byte { return m.contents }
+func (m mockPart) Marshal() []byte { return append([]byte{m.numParts, m.partNum}, m.contents...) }
diff --git a/single/message/request.go b/single/message/request.go
index 75e83510926c7b1d559be6068b0bde1702cae34c..3519902e75b6804df0e4ca0108b396e5af79863b 100644
--- a/single/message/request.go
+++ b/single/message/request.go
@@ -277,11 +277,23 @@ func (mp RequestPayload) GetNumRequestParts() uint8 {
return mp.numRequestParts[0]
}
+// GetNumParts returns the number of messages in the request. This function
+// wraps GetMaxRequestParts so that RequestPayload adheres to the Part
+// interface.
+func (mp RequestPayload) GetNumParts() uint8 {
+ return mp.GetNumRequestParts()
+}
+
// SetNumRequestParts sets the number of messages in the request.
func (mp RequestPayload) SetNumRequestParts(num uint8) {
copy(mp.numRequestParts, []byte{num})
}
+// GetPartNum always returns 0 since it is the first message.
+func (mp RequestPayload) GetPartNum() uint8 {
+ return 0
+}
+
// GetContents returns the payload's contents.
func (mp RequestPayload) GetContents() []byte {
return mp.contents[:binary.BigEndian.Uint16(mp.size)]
diff --git a/single/message/requestPart.go b/single/message/requestPart.go
new file mode 100644
index 0000000000000000000000000000000000000000..11d9f27b663cd15d500d7bf3321427c08f70c5d9
--- /dev/null
+++ b/single/message/requestPart.go
@@ -0,0 +1,138 @@
+///////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC //
+// //
+// Use of this source code is governed by a license that can be found in the //
+// LICENSE file //
+///////////////////////////////////////////////////////////////////////////////
+
+package message
+
+import (
+ "encoding/binary"
+ "github.com/pkg/errors"
+ jww "github.com/spf13/jwalterweatherman"
+)
+
+// Error messages.
+const (
+ // NewRequestPart
+ errReqPartPayloadSize = "[SU] Failed to create new single-use request " +
+ "message part: external payload size (%d) is smaller than the " +
+ "minimum message size for a request part (%d)."
+
+ // UnmarshalRequestPart
+ errReqPartDataSize = "size of data (%d) must be at least %d"
+
+ // RequestPart.SetContents
+ errReqPartContentsSize = "[SU] Failed to set contents of single-use " +
+ "request message part: size of the supplied contents (%d) is larger " +
+ "than the max message size (%d)."
+)
+
+// Sizes of fields.
+const (
+ reqPartPartNumLen = 1
+ reqPartSizeLen = 2
+ reqPartMinSize = reqPartPartNumLen + reqPartSizeLen
+)
+
+/*
++------------------------------+
+| cMix Message Contents |
++---------+---------+----------+
+| partNum | size | contents |
+| 1 byte | 2 bytes | variable |
++---------+---------+----------+
+*/
+
+type RequestPart struct {
+ data []byte // Serial of all contents
+ partNum []byte // Index of message in a series of messages
+ size []byte // Size of the contents
+ contents []byte // The encrypted contents
+}
+
+// NewRequestPart generates a new request message part of the specified size.
+func NewRequestPart(externalPayloadSize int) RequestPart {
+ if externalPayloadSize < reqPartMinSize {
+ jww.FATAL.Panicf(
+ errReqPartPayloadSize, externalPayloadSize, reqPartMinSize)
+ }
+
+ rmp := mapRequestPart(make([]byte, externalPayloadSize))
+ return rmp
+}
+
+// GetRequestPartContentsSize returns the size of the contents for the given
+// external payload size.
+func GetRequestPartContentsSize(externalPayloadSize int) int {
+ return externalPayloadSize - reqPartMinSize
+}
+
+// mapRequestPart builds a message part mapped to the passed in data.
+// It is mapped by reference; a copy is not made.
+func mapRequestPart(data []byte) RequestPart {
+ return RequestPart{
+ data: data,
+ partNum: data[:reqPartPartNumLen],
+ size: data[reqPartPartNumLen:reqPartMinSize],
+ contents: data[reqPartMinSize:],
+ }
+}
+
+// UnmarshalRequestPart converts a byte buffer into a request message part.
+func UnmarshalRequestPart(b []byte) (RequestPart, error) {
+ if len(b) < reqPartMinSize {
+ return RequestPart{}, errors.Errorf(
+ errReqPartDataSize, len(b), reqPartMinSize)
+ }
+ return mapRequestPart(b), nil
+}
+
+// Marshal returns the bytes of the message part.
+func (m RequestPart) Marshal() []byte {
+ return m.data
+}
+
+// GetPartNum returns the index of this part in the message.
+func (m RequestPart) GetPartNum() uint8 {
+ return m.partNum[0]
+}
+
+// SetPartNum sets the part number of the message.
+func (m RequestPart) SetPartNum(num uint8) {
+ copy(m.partNum, []byte{num})
+}
+
+// GetNumParts always returns 0. It is here so that RequestPart adheres to th
+// Part interface.
+func (m RequestPart) GetNumParts() uint8 {
+ return 0
+}
+
+// GetContents returns the contents of the message part.
+func (m RequestPart) GetContents() []byte {
+ return m.contents[:binary.BigEndian.Uint16(m.size)]
+}
+
+// GetContentsSize returns the length of the contents.
+func (m RequestPart) GetContentsSize() int {
+ return int(binary.BigEndian.Uint16(m.size))
+}
+
+// GetMaxContentsSize returns the max capacity of the contents.
+func (m RequestPart) GetMaxContentsSize() int {
+ return len(m.contents)
+}
+
+// SetContents sets the contents of the message part. Does not zero out previous
+// contents.
+func (m RequestPart) SetContents(contents []byte) {
+ if len(contents) > len(m.contents) {
+ jww.FATAL.Panicf(errReqPartContentsSize, len(contents), len(m.contents))
+ }
+
+ binary.BigEndian.PutUint16(m.size, uint16(len(contents)))
+
+ copy(m.contents, contents)
+}
diff --git a/single/message/requestPart_test.go b/single/message/requestPart_test.go
new file mode 100644
index 0000000000000000000000000000000000000000..75090ffd6823c5c6844fbaa60bea64057f380fca
--- /dev/null
+++ b/single/message/requestPart_test.go
@@ -0,0 +1,183 @@
+///////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC //
+// //
+// Use of this source code is governed by a license that can be found in the //
+// LICENSE file //
+///////////////////////////////////////////////////////////////////////////////
+
+package message
+
+import (
+ "bytes"
+ "fmt"
+ "math/rand"
+ "reflect"
+ "testing"
+)
+
+// Happy path.
+func Test_NewRequestPart(t *testing.T) {
+ prng := rand.New(rand.NewSource(42))
+ payloadSize := prng.Intn(2000)
+ expected := RequestPart{
+ data: make([]byte, payloadSize),
+ partNum: make([]byte, reqPartPartNumLen),
+ size: make([]byte, reqPartSizeLen),
+ contents: make([]byte, payloadSize-reqPartMinSize),
+ }
+
+ rmp := NewRequestPart(payloadSize)
+
+ if !reflect.DeepEqual(expected, rmp) {
+ t.Errorf("NewRequestPart did not return the expected "+
+ "RequestPart.\nexpected: %+v\nreceived: %+v", expected, rmp)
+ }
+}
+
+// Error path: provided contents size is not large enough.
+func Test_NewRequestPart_PayloadSizeError(t *testing.T) {
+ externalPayloadSize := 1
+ expectedErr := fmt.Sprintf(
+ errReqPartPayloadSize, externalPayloadSize, reqPartMinSize)
+ defer func() {
+ if r := recover(); r == nil || r != expectedErr {
+ t.Errorf("NewRequestPart did not panic with the expected error "+
+ "when the size of the payload is smaller than the required "+
+ "size.\nexpected: %s\nreceived: %+v", expectedErr, r)
+ }
+ }()
+
+ _ = NewRequestPart(externalPayloadSize)
+}
+
+// Happy path.
+func Test_mapRequestPart(t *testing.T) {
+ prng := rand.New(rand.NewSource(42))
+ expectedPartNum := uint8(prng.Uint32())
+ size := []byte{uint8(prng.Uint64()), uint8(prng.Uint64())}
+ expectedContents := make([]byte, prng.Intn(2000))
+ prng.Read(expectedContents)
+ var data []byte
+ data = append(data, expectedPartNum)
+ data = append(data, size...)
+ data = append(data, expectedContents...)
+
+ rmp := mapRequestPart(data)
+
+ if expectedPartNum != rmp.partNum[0] {
+ t.Errorf("mapRequestPart did not correctly map partNum."+
+ "\nexpected: %d\nreceived: %d", expectedPartNum, rmp.partNum[0])
+ }
+
+ if !bytes.Equal(expectedContents, rmp.contents) {
+ t.Errorf("mapRequestPart did not correctly map contents."+
+ "\nexpected: %+v\nreceived: %+v", expectedContents, rmp.contents)
+ }
+
+ if !bytes.Equal(data, rmp.data) {
+ t.Errorf("mapRequestPart did not save the data correctly."+
+ "\nexpected: %+v\nreceived: %+v", data, rmp.data)
+ }
+}
+
+// Happy path.
+func TestRequestPart_Marshal_UnmarshalRequestPart(t *testing.T) {
+ prng := rand.New(rand.NewSource(42))
+ payload := make([]byte, prng.Intn(2000))
+ prng.Read(payload)
+ rmp := NewRequestPart(prng.Intn(2000))
+
+ data := rmp.Marshal()
+
+ newRmp, err := UnmarshalRequestPart(data)
+ if err != nil {
+ t.Errorf("UnmarshalRequestPart produced an error: %+v", err)
+ }
+
+ if !reflect.DeepEqual(rmp, newRmp) {
+ t.Errorf("Failed to Marshal and unmarshal the RequestPart."+
+ "\nexpected: %+v\nrecieved: %+v", rmp, newRmp)
+ }
+}
+
+// Error path: provided bytes are too small.
+func Test_UnmarshalRequestPart_Error(t *testing.T) {
+ data := []byte{1}
+ expectedErr := fmt.Sprintf(errReqPartDataSize, len(data), reqPartMinSize)
+ _, err := UnmarshalRequestPart([]byte{1})
+ if err == nil || err.Error() != expectedErr {
+ t.Errorf("UnmarshalRequestPart did not produce the expected error "+
+ "when the byte slice is smaller required."+
+ "\nexpected: %s\nreceived: %+v", expectedErr, err)
+ }
+}
+
+// Happy path.
+func TestRequestPart_SetPartNum_GetPartNum(t *testing.T) {
+ prng := rand.New(rand.NewSource(42))
+ expectedPartNum := uint8(prng.Uint32())
+ rmp := NewRequestPart(prng.Intn(2000))
+
+ rmp.SetPartNum(expectedPartNum)
+
+ if expectedPartNum != rmp.GetPartNum() {
+ t.Errorf("GetPartNum failed to return the expected part number."+
+ "\nexpected: %d\nrecieved: %d", expectedPartNum, rmp.GetPartNum())
+ }
+}
+
+// Happy path.
+func TestRequestPart_GetMaxParts(t *testing.T) {
+ prng := rand.New(rand.NewSource(42))
+ expectedMaxParts := uint8(0)
+ rmp := NewRequestPart(prng.Intn(2000))
+
+ if expectedMaxParts != rmp.GetNumParts() {
+ t.Errorf("GetNumParts failed to return the expected max parts."+
+ "\nexpected: %d\nrecieved: %d", expectedMaxParts, rmp.GetNumParts())
+ }
+}
+
+// Happy path.
+func TestRequestPart_SetContents_GetContents_GetContentsSize_GetMaxContentsSize(t *testing.T) {
+ prng := rand.New(rand.NewSource(42))
+ externalPayloadSize := prng.Intn(2000)
+ contentSize := externalPayloadSize - reqPartMinSize - 10
+ expectedContents := make([]byte, contentSize)
+ prng.Read(expectedContents)
+ rmp := NewRequestPart(externalPayloadSize)
+ rmp.SetContents(expectedContents)
+
+ if !bytes.Equal(expectedContents, rmp.GetContents()) {
+ t.Errorf("GetContents failed to return the expected contents."+
+ "\nexpected: %+v\nrecieved: %+v", expectedContents, rmp.GetContents())
+ }
+
+ if contentSize != rmp.GetContentsSize() {
+ t.Errorf("GetContentsSize failed to return the expected contents size."+
+ "\nexpected: %d\nrecieved: %d", contentSize, rmp.GetContentsSize())
+ }
+
+ if externalPayloadSize-reqPartMinSize != rmp.GetMaxContentsSize() {
+ t.Errorf("GetMaxContentsSize failed to return the expected max "+
+ "contents size.\nexpected: %d\nrecieved: %d",
+ externalPayloadSize-reqPartMinSize, rmp.GetMaxContentsSize())
+ }
+}
+
+// Error path: size of supplied contents does not match message contents size.
+func TestRequestPart_SetContents_ContentsSizeError(t *testing.T) {
+ payloadSize, contentsLen := 255, 500
+ expectedErr := fmt.Sprintf(
+ errReqPartContentsSize, contentsLen, payloadSize-reqPartMinSize)
+ defer func() {
+ if r := recover(); r == nil || r != expectedErr {
+ t.Errorf("SetContents did not panic with the expected error when "+
+ "the size of the supplied bytes is larger than the content "+
+ "size.\nexpected: %s\nreceived: %+v", expectedErr, r)
+ }
+ }()
+
+ rmp := NewRequestPart(payloadSize)
+ rmp.SetContents(make([]byte, contentsLen))
+}
diff --git a/single/receivedRequest.go b/single/receivedRequest.go
index b8762674c226da6cf109e2ee223eca7aad1daecb..0cbf4adddfc3595f403c19ba663d053eb390afa8 100644
--- a/single/receivedRequest.go
+++ b/single/receivedRequest.go
@@ -10,6 +10,7 @@ import (
"gitlab.com/elixxir/client/single/message"
ds "gitlab.com/elixxir/comms/network/dataStructures"
"gitlab.com/elixxir/crypto/cyclic"
+ "gitlab.com/elixxir/crypto/e2e/singleUse"
"gitlab.com/elixxir/primitives/states"
"gitlab.com/xx_network/primitives/id"
"sync"
@@ -91,7 +92,8 @@ func (r Request) Respond(payload []byte, cmixParams cmix.CMIXParams,
parts := partitionResponse(payload, r.net.GetMaxMessageLength(), r.maxParts)
// Encrypt and send the partitions
- cyphers := makeCyphers(r.dhKey, uint8(len(parts)))
+ cyphers := makeCyphers(r.dhKey, uint8(len(parts)),
+ singleUse.NewResponseKey, singleUse.NewResponseFingerprint)
rounds := make([]id.Round, len(parts))
sendResults := make(chan ds.EventReturn, len(parts))
diff --git a/single/request.go b/single/request.go
index 646608b1b2cc845006aa0be723c31c834e4b32c8..6198b1ffc46d5b8c0a6d74cec3cb6134c28291f8 100644
--- a/single/request.go
+++ b/single/request.go
@@ -1,8 +1,7 @@
package single
import (
- "encoding/base64"
- "fmt"
+ "bytes"
"github.com/pkg/errors"
jww "github.com/spf13/jwalterweatherman"
"gitlab.com/elixxir/client/cmix"
@@ -26,7 +25,7 @@ import (
// Response interface allows for callbacks to
type Response interface {
Callback(payload []byte, receptionID receptionID.EphemeralIdentity,
- round rounds.Round, err error)
+ rounds []rounds.Round, err error)
}
type RequestParams struct {
@@ -46,11 +45,13 @@ func GetDefaultRequestParams() RequestParams {
// Error messages.
const (
// TransmitRequest
- errNetworkHealth = "cannot send singe-use request when the network is not healthy"
- errMakeDhKeys = "failed to generate DH keys (%s for %s): %+v"
- errMakeIDs = "failed to generate IDs (%s for %s): %+v"
- errAddFingerprint = "failed to add fingerprint %d of %d: %+v (%s for %s)"
- errSendRequest = "failed to send %s request to %s: %+v"
+ errPayloadSize = "size of payload %d exceeds the maximum size of %d (%s for %s)"
+ errNetworkHealth = "cannot send singe-use request when the network is not healthy"
+ errMakeDhKeys = "failed to generate DH keys (%s for %s): %+v"
+ errMakeIDs = "failed to generate IDs (%s for %s): %+v"
+ errAddFingerprint = "failed to add fingerprint %d of %d: %+v (%s for %s)"
+ errSendRequest = "failed to send %s request to %s: %+v"
+ errSendRequestPart = "failed to send request part %d of %d (%s for %s): %+v"
// generateDhKeys
errGenerateInGroup = "failed to generate private key in group: %+v"
@@ -63,11 +64,17 @@ const (
errResponseTimeout = "waiting for response to single-use request timed out after %s"
)
+// Maximum number of request part cMix messages.
+const maxNumRequestParts = 255
+
// GetMaxRequestSize returns the maximum size of a request payload.
func GetMaxRequestSize(net CMix, e2eGrp *cyclic.Group) int {
payloadSize := message.GetRequestPayloadSize(net.GetMaxMessageLength(),
e2eGrp.GetP().ByteLen())
- return message.GetRequestContentsSize(payloadSize)
+ requestSize := message.GetRequestContentsSize(payloadSize)
+ requestPartSize := message.GetRequestPartContentsSize(
+ net.GetMaxMessageLength())
+ return requestSize + (maxNumRequestParts * requestPartSize)
}
/* Single is a system which allows for an end-to-end encrypted anonymous request
@@ -83,16 +90,24 @@ func GetMaxRequestSize(net CMix, e2eGrp *cyclic.Group) int {
// containing the given payload. The request is identified as coming from a new
// user ID and the recipient of the request responds to that address. As a
// result, this request does not reveal the identity of the sender.
-// The current implementation only allows for a single cMix request payload.
-// Because the request payload itself must include negotiation materials, it is
-// limited to just a few thousand bits of payload, and will return an error if
-// the payload is too large. GetMaxRequestSize can be used to get this max size.
+//
+// The current implementation allows for up to maxNumRequestParts cMix request
+// payloads. GetMaxRequestSize can be used to get the max size.
+//
// The network follower must be running and healthy to transmit.
func TransmitRequest(recipient contact.Contact, tag string, payload []byte,
- callback Response, param RequestParams, net CMix, rng csprng.Source,
- e2eGrp *cyclic.Group) (id.Round, receptionID.EphemeralIdentity, error) {
+ callback Response, param RequestParams, net cmix.Client, rng csprng.Source,
+ e2eGrp *cyclic.Group) ([]id.Round, receptionID.EphemeralIdentity, error) {
+
+ if len(payload) > GetMaxRequestSize(net, e2eGrp) {
+ return nil, receptionID.EphemeralIdentity{}, errors.Errorf(
+ errPayloadSize, len(payload), GetMaxRequestSize(net, e2eGrp), tag,
+ recipient)
+ }
+
if !net.IsHealthy() {
- return 0, receptionID.EphemeralIdentity{}, errors.New(errNetworkHealth)
+ return nil, receptionID.EphemeralIdentity{},
+ errors.New(errNetworkHealth)
}
// Get address ID address space size; this blocks until the address space
@@ -103,31 +118,34 @@ func TransmitRequest(recipient contact.Contact, tag string, payload []byte,
// Generate DH key and public key
dhKey, publicKey, err := generateDhKeys(e2eGrp, recipient.DhPubKey, rng)
if err != nil {
- return 0, receptionID.EphemeralIdentity{},
+ return nil, receptionID.EphemeralIdentity{},
errors.Errorf(errMakeDhKeys, tag, recipient, err)
}
// Build the message payload
+ payloadSize := message.GetRequestPayloadSize(net.GetMaxMessageLength(),
+ e2eGrp.GetP().ByteLen())
+ firstPart, parts := partitionPayload(
+ message.GetRequestContentsSize(payloadSize),
+ message.GetRequestPartContentsSize(net.GetMaxMessageLength()),
+ payload)
request := message.NewRequest(
net.GetMaxMessageLength(), e2eGrp.GetP().ByteLen())
requestPayload := message.NewRequestPayload(
- request.GetPayloadSize(), payload, param.MaxResponseMessages)
+ request.GetPayloadSize(), firstPart, param.MaxResponseMessages)
// Generate new user ID and address ID
var sendingID receptionID.EphemeralIdentity
requestPayload, sendingID, err = makeIDs(
requestPayload, publicKey, addressSize, param.Timeout, timeStart, rng)
if err != nil {
- return 0, receptionID.EphemeralIdentity{},
+ return nil, receptionID.EphemeralIdentity{},
errors.Errorf(errMakeIDs, tag, recipient, err)
}
- fmt.Printf("reqPayload: %v\n", base64.
- StdEncoding.EncodeToString(requestPayload.Marshal()))
-
- // Encrypt payload
- fp := singleUse.NewTransmitFingerprint(recipient.DhPubKey)
- key := singleUse.NewTransmitKey(dhKey)
+ // Encrypt and assemble payload
+ fp := singleUse.NewRequestFingerprint(recipient.DhPubKey)
+ key := singleUse.NewRequestKey(dhKey)
encryptedPayload := auth.Crypt(key, fp[:24], requestPayload.Marshal())
fmt.Printf("ecrPayload: %v\n", base64.StdEncoding.EncodeToString(encryptedPayload))
// Generate CMix message MAC
@@ -142,7 +160,7 @@ func TransmitRequest(recipient contact.Contact, tag string, payload []byte,
timeoutKillChan := make(chan bool)
var callbackOnce sync.Once
wrapper := func(payload []byte, receptionID receptionID.EphemeralIdentity,
- round rounds.Round, err error) {
+ rounds []rounds.Round, err error) {
select {
case timeoutKillChan <- true:
default:
@@ -150,12 +168,14 @@ func TransmitRequest(recipient contact.Contact, tag string, payload []byte,
callbackOnce.Do(func() {
net.DeleteClientFingerprints(sendingID.Source)
- go callback.Callback(payload, receptionID, round, err)
+ go callback.Callback(payload, receptionID, rounds, err)
})
}
- cyphers := makeCyphers(dhKey, param.MaxResponseMessages)
+ cyphers := makeCyphers(dhKey, param.MaxResponseMessages,
+ singleUse.NewResponseKey, singleUse.NewResponseFingerprint)
+ roundIds := newRoundIdCollector(len(cyphers))
for i, cy := range cyphers {
processor := responseProcessor{
sendingID: sendingID,
@@ -164,12 +184,13 @@ func TransmitRequest(recipient contact.Contact, tag string, payload []byte,
cy: cy,
tag: tag,
recipient: &recipient,
+ roundIDs: roundIds,
}
err = net.AddFingerprint(
sendingID.Source, processor.cy.GetFingerprint(), &processor)
if err != nil {
- return 0, receptionID.EphemeralIdentity{}, errors.Errorf(
+ return nil, receptionID.EphemeralIdentity{}, errors.Errorf(
errAddFingerprint, i, len(cyphers), tag, recipient, err)
}
}
@@ -189,14 +210,34 @@ func TransmitRequest(recipient contact.Contact, tag string, payload []byte,
rid, _, err := net.Send(recipient.ID, cmixMsg.RandomFingerprint(rng), svc,
request.Marshal(), mac, param.CmixParam)
if err != nil {
- return 0, receptionID.EphemeralIdentity{},
+ return nil, receptionID.EphemeralIdentity{},
errors.Errorf(errSendRequest, tag, recipient, err)
}
+ roundIDs := make([]id.Round, len(parts)+1)
+ roundIDs[0] = rid
+ for i, part := range parts {
+ requestPart := message.NewRequestPart(net.GetMaxMessageLength())
+ requestPart.SetPartNum(uint8(i + 1))
+ requestPart.SetContents(part)
+
+ key = singleUse.NewResponseKey(dhKey, uint64(i+1))
+ fp = singleUse.NewResponseFingerprint(dhKey, uint64(i+1))
+ encryptedPayload = auth.Crypt(key, fp[:24], requestPart.Marshal())
+ mac = singleUse.MakeMAC(key, encryptedPayload)
+
+ roundIDs[i+1], _, err = net.Send(
+ recipient.ID, fp, svc, encryptedPayload, mac, param.CmixParam)
+ if err != nil {
+ return nil, receptionID.EphemeralIdentity{}, errors.Errorf(
+ errSendRequestPart, i, len(part), tag, recipient, err)
+ }
+ }
+
remainingTimeout := param.Timeout - netTime.Since(timeStart)
go waitForTimeout(timeoutKillChan, wrapper, remainingTimeout)
- return rid, sendingID, nil
+ return roundIDs, sendingID, nil
}
// generateDhKeys generates a new public key and DH key.
@@ -276,8 +317,36 @@ func waitForTimeout(kill chan bool, cb callbackWrapper, timeout time.Duration) {
cb(
nil,
receptionID.EphemeralIdentity{},
- rounds.Round{},
+ nil,
errors.Errorf(errResponseTimeout, timeout),
)
}
}
+
+// partitionPayload splits the payload into its parts. The first part is of size
+// firstPartSize and is shorter than the rest since it is sent in the
+// message.Request, which includes extra information. It is also returned on its
+// own so that it can be handled on its own. The rest of the parts are of size
+// partSize and will be sent as part of message.RequestPart.
+func partitionPayload(firstPartSize, partSize int, payload []byte) (
+ firstPart []byte, parts [][]byte) {
+
+ // Return just the first part if it fits in a single message
+ if len(payload) <= firstPartSize {
+ return payload, nil
+ }
+
+ firstPart = payload[:firstPartSize]
+
+ numParts := (len(payload[:firstPartSize]) + partSize - 1) / partSize
+ parts = make([][]byte, 0, numParts)
+ buff := bytes.NewBuffer(payload[firstPartSize:])
+
+ for n := buff.Next(partSize); len(n) > 0; n = buff.Next(partSize) {
+ newPart := make([]byte, partSize)
+ copy(newPart, n)
+ parts = append(parts, newPart)
+ }
+
+ return firstPart, parts
+}
diff --git a/single/requestPartProcessor.go b/single/requestPartProcessor.go
new file mode 100644
index 0000000000000000000000000000000000000000..41aa2bd8c40200b1c1d4637b69dca68089a1680c
--- /dev/null
+++ b/single/requestPartProcessor.go
@@ -0,0 +1,57 @@
+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC //
+// //
+// Use of this source code is governed by a license that can be found in the //
+// LICENSE file //
+////////////////////////////////////////////////////////////////////////////////
+
+package single
+
+import (
+ jww "github.com/spf13/jwalterweatherman"
+ "gitlab.com/elixxir/client/cmix/identity/receptionID"
+ "gitlab.com/elixxir/client/cmix/rounds"
+ "gitlab.com/elixxir/client/single/message"
+ "gitlab.com/elixxir/primitives/format"
+ "gitlab.com/xx_network/primitives/id"
+)
+
+// requestPartProcessor handles the decryption and collation of request parts.
+type requestPartProcessor struct {
+ myId *id.ID
+ tag string
+ cb func(payloadContents []byte, rounds []rounds.Round)
+ c *message.Collator
+ cy cypher
+ roundIDs *roundCollector
+}
+
+func (rpp *requestPartProcessor) Process(msg format.Message,
+ _ receptionID.EphemeralIdentity, round rounds.Round) {
+ decrypted, err := rpp.cy.Decrypt(msg.GetContents(), msg.GetMac())
+ if err != nil {
+ jww.ERROR.Printf("[SU] Failed to decrypt single-use request payload "+
+ "for %s to %s: %+v", rpp.tag, rpp.myId, err)
+ return
+ }
+
+ requestPart, err := message.UnmarshalRequestPart(decrypted)
+ if err != nil {
+ jww.ERROR.Printf("[SU] Failed to unmarshal single-use request part "+
+ "payload for %s to %s: %+v", rpp.tag, rpp.myId, err)
+ return
+ }
+
+ payload, done, err := rpp.c.Collate(requestPart)
+ if err != nil {
+ jww.ERROR.Printf("[SU] Failed to collate single-use request payload "+
+ "for %s to %s: %+v", rpp.tag, rpp.myId, err)
+ return
+ }
+
+ rpp.roundIDs.add(round)
+
+ if done {
+ rpp.cb(payload, rpp.roundIDs.getList())
+ }
+}
diff --git a/single/responseProcessor.go b/single/responseProcessor.go
index c6b503164bbdbbea64bc0f6d644f26e8c4923fc9..0f7134331051900027e3bf699218878bcdfa4f8a 100644
--- a/single/responseProcessor.go
+++ b/single/responseProcessor.go
@@ -12,7 +12,7 @@ import (
)
type callbackWrapper func(payload []byte,
- receptionID receptionID.EphemeralIdentity, round rounds.Round, err error)
+ receptionID receptionID.EphemeralIdentity, rounds []rounds.Round, err error)
// responseProcessor is registered for each potential fingerprint. Adheres to
// the message.Processor interface registered with cmix.Client
@@ -23,6 +23,7 @@ type responseProcessor struct {
cy cypher
tag string
recipient *contact.Contact
+ roundIDs *roundCollector
}
// Process decrypts a response part and adds it to the collator - returning
@@ -42,7 +43,14 @@ func (rsp *responseProcessor) Process(ecrMsg format.Message,
return
}
- payload, done, err := rsp.c.Collate(decrypted)
+ responsePart, err := message.UnmarshalResponsePart(decrypted)
+ if err != nil {
+ jww.ERROR.Printf("[SU] Failed to unmarshal single-use response part "+
+ "payload for %s to %s: %+v", rsp.tag, rsp.recipient.ID, err)
+ return
+ }
+
+ payload, done, err := rsp.c.Collate(responsePart)
if err != nil {
jww.ERROR.Printf("[SU] Failed to collate single-use response payload "+
"for %s to %s, single use may fail: %+v",
@@ -50,7 +58,9 @@ func (rsp *responseProcessor) Process(ecrMsg format.Message,
return
}
+ rsp.roundIDs.add(round)
+
if done {
- rsp.callback(payload, receptionID, round, nil)
+ rsp.callback(payload, receptionID, rsp.roundIDs.getList(), nil)
}
}
diff --git a/single/roundCollector.go b/single/roundCollector.go
new file mode 100644
index 0000000000000000000000000000000000000000..4d48b8e32c0f3e458457502481dd69344c510ee1
--- /dev/null
+++ b/single/roundCollector.go
@@ -0,0 +1,52 @@
+////////////////////////////////////////////////////////////////////////////////
+// Copyright © 2020 xx network SEZC //
+// //
+// Use of this source code is governed by a license that can be found in the //
+// LICENSE file //
+////////////////////////////////////////////////////////////////////////////////
+
+package single
+
+import (
+ "gitlab.com/elixxir/client/cmix/rounds"
+ "gitlab.com/xx_network/primitives/id"
+ "sync"
+)
+
+// roundCollector keeps track of a list of unique rounds. Multiple inserts of
+// the same round are ignored. It is multi-thread safe.
+type roundCollector struct {
+ list map[id.Round]rounds.Round
+ mux sync.Mutex
+}
+
+// newRoundIdCollector initialises a new roundCollector with a list of the
+// given size. Size is not necessary and can be larger or smaller than the real
+// size.
+func newRoundIdCollector(size int) *roundCollector {
+ return &roundCollector{
+ list: make(map[id.Round]rounds.Round, size),
+ }
+}
+
+// add inserts a new round to the list.
+func (rc *roundCollector) add(round rounds.Round) {
+ rc.mux.Lock()
+ defer rc.mux.Unlock()
+
+ rc.list[round.ID] = round
+}
+
+// getList returns the list of round IDs.
+func (rc *roundCollector) getList() []rounds.Round {
+ rc.mux.Lock()
+ defer rc.mux.Unlock()
+
+ list := make([]rounds.Round, 0, len(rc.list))
+
+ for _, round := range rc.list {
+ list = append(list, round)
+ }
+
+ return list
+}