pss: refactor PssMsg to message package

Author: jpeletier

pss/ARCHITECTURE.md

@@ -2,7 +2,7 @@
 
 Pss provides devp2p functionality for swarm nodes without the need for a direct tcp connection between them.
 
-Messages are encapsulated in a devp2p message structure `PssMsg`. These capsules are forwarded from node to node using ordinary tcp devp2p until they reach their destination: The node or nodes who can successfully decrypt the message.
+Messages are encapsulated in a devp2p message structure `message.Message`. These capsules are forwarded from node to node using ordinary tcp devp2p until they reach their destination: The node or nodes who can successfully decrypt the message.
 
 | Layer     | Contents        |
 |-----------|-----------------|

pss/doc.go

@@ -16,7 +16,7 @@
 
 // Pss provides devp2p functionality for swarm nodes without the need for a direct tcp connection between them.
 //
-// Messages are encapsulated in a devp2p message structure `PssMsg`. These capsules are forwarded from node to node using ordinary tcp devp2p until it reaches its destination: The node or nodes who can successfully decrypt the message.
+// Messages are encapsulated in a devp2p message structure `message.Message`. These capsules are forwarded from node to node using ordinary tcp devp2p until it reaches its destination: The node or nodes who can successfully decrypt the message.
 //
 // Routing of messages is done using swarm's own kademlia routing. Optionally routing can be turned off, forcing the message to be sent to all peers, similar to the behavior of the whisper protocol.
 //
@@ -50,11 +50,11 @@
 //
 // Under the hood, pss implements its own MsgReadWriter, which bridges MsgReadWriter.WriteMsg with Pss.SendRaw, and deftly adds an InjectMsg method which pipes incoming messages to appear on the MsgReadWriter.ReadMsg channel.
 //
-// An incoming connection is nothing more than an actual PssMsg appearing with a certain Topic. If a Handler har been registered to that Topic, the message will be passed to it. This constitutes a "new" connection if:
+// An incoming connection is nothing more than an actual message.Message appearing with a certain Topic. If a Handler har been registered to that Topic, the message will be passed to it. This constitutes a "new" connection if:
 //
 // - The pss node never called AddPeer with this combination of remote peer address and topic, and
 //
-// - The pss node never received a PssMsg from this remote peer with this specific Topic before.
+// - The pss node never received a message.Message from this remote peer with this specific Topic before.
 //
 // If it is a "new" connection, the protocol will be "run" on the remote peer, in the same manner as if it was pre-emptively added.
 //

pss/forwarding_test.go

@@ -248,7 +248,7 @@ func testForwardMsg(t *testing.T, ps *Pss, c *testCase) {
 	resultMap := make(map[pot.Address]int)
 
 	defer func() { sendFunc = sendMsg }()
-	sendFunc = func(_ *Pss, sp *network.Peer, _ *PssMsg) bool {
+	sendFunc = func(_ *Pss, sp *network.Peer, _ *message.Message) bool {
 		if tries < nFails {
 			tries++
 			return false
@@ -347,8 +347,8 @@ func newTestDiscoveryPeer(addr pot.Address, kad *network.Kademlia) *network.Peer
 	return network.NewPeer(bp, kad)
 }
 
-func newTestMsg(addr []byte) *PssMsg {
-	msg := newPssMsg(message.Flags{})
+func newTestMsg(addr []byte) *message.Message {
+	msg := message.New(message.Flags{})
 	msg.To = addr[:]
 	msg.Expire = uint32(time.Now().Add(time.Second * 60).Unix())
 	msg.Topic = [4]byte{}

pss/handshake.go

@@ -291,7 +291,7 @@ func (ctl *HandshakeController) getSymKey(symkeyid string) *handshakeKey {
 	return ctl.symKeyIndex[symkeyid]
 }
 
-// Passed as a PssMsg handler for the topic handshake is activated on
+// Passed as a message.Message handler for the topic handshake is activated on
 // Handles incoming key exchange messages and
 // counts message usage by symmetric key (expiry limit control)
 // Only returns error if key handler fails

pss/keystore.go

@@ -147,7 +147,7 @@ func (ks *KeyStore) getPeerAddress(keyid string, topic message.Topic) (PssAddres
 // If successful, returns the payload of the message and the id
 // of the symmetric key used to decrypt the message.
 // It fails if decryption of the message fails or if the message is corrupted/not valid.
-func (ks *KeyStore) processSym(pssMsg *PssMsg) ([]byte, string, PssAddress, error) {
+func (ks *KeyStore) processSym(pssMsg *message.Message) ([]byte, string, PssAddress, error) {
 	metrics.GetOrRegisterCounter("pss.process.sym", nil).Inc(1)
 
 	for i := ks.symKeyDecryptCacheCursor; i > ks.symKeyDecryptCacheCursor-cap(ks.symKeyDecryptCache) && i > 0; i-- {
@@ -185,7 +185,7 @@ func (ks *KeyStore) processSym(pssMsg *PssMsg) ([]byte, string, PssAddress, erro
 // If successful, returns the payload of the message and the hex representation of
 // the public key used to decrypt the message.
 // It fails if decryption of message fails, or if the message is corrupted.
-func (p *Pss) processAsym(pssMsg *PssMsg) ([]byte, string, PssAddress, error) {
+func (p *Pss) processAsym(pssMsg *message.Message) ([]byte, string, PssAddress, error) {
 	metrics.GetOrRegisterCounter("pss.process.asym", nil).Inc(1)
 
 	unwrapParams := &crypto.UnwrapParams{

pss/message/flags_test.go

@@ -10,11 +10,12 @@ import (
 	"github.com/ethersphere/swarm/pss/message"
 )
 
+var bools = []bool{true, false}
+
 func TestFlags(tx *testing.T) {
 	t := ut.BeginTest(tx, false)
 	defer t.FinishTest()
 
-	bools := []bool{true, false}
 	for _, r := range bools {
 		for _, s := range bools {
 			f := message.Flags{

pss/message/message.go

@@ -0,0 +1,46 @@
+package message
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/common"
+	"golang.org/x/crypto/sha3"
+)
+
+// Message encapsulates messages transported over pss.
+type Message struct {
+	To      []byte
+	Flags   Flags
+	Expire  uint32
+	Topic   Topic
+	Payload []byte
+}
+
+const digestLength = 32 // byte length of digest used for pss cache (currently same as swarm chunk hash)
+
+// Digest holds the digest of a message used for caching
+type Digest [digestLength]byte
+
+// New creates a new PSS message
+func New(flags Flags) *Message {
+	return &Message{
+		Flags: flags,
+	}
+}
+
+// Digest computes a message digest for use as a cache key
+func (msg *Message) Digest() Digest {
+	hasher := sha3.NewLegacyKeccak256()
+	hasher.Write(msg.To)
+	hasher.Write(msg.Topic[:])
+	hasher.Write(msg.Payload)
+	key := hasher.Sum(nil)
+	d := Digest{}
+	copy(d[:], key[:digestLength])
+	return d
+}
+
+// String representation of a PSS message
+func (msg *Message) String() string {
+	return fmt.Sprintf("PssMsg: Recipient: %s, Topic: %v", common.ToHex(msg.To), msg.Topic.String())
+}

pss/message/message_test.go

@@ -0,0 +1,51 @@
+package message_test
+
+import (
+	"encoding/hex"
+	"fmt"
+	"testing"
+
+	"github.com/epiclabs-io/ut"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/ethersphere/swarm/pss/message"
+)
+
+func TestMessage(tx *testing.T) {
+	t := ut.BeginTest(tx, false) // set to true to generate test results
+	defer t.FinishTest()
+
+	// generate some test messages deterministically
+	for i, topicString := range someTopics {
+		flags := message.Flags{
+			Raw:       i&0x1 == 0,
+			Symmetric: i&0x3 == 0,
+		}
+
+		msg := message.New(flags)
+		msg.To = ut.RandomArray(i, common.AddressLength)
+		msg.Expire = uint32(i)
+		msg.Topic = message.NewTopic([]byte(topicString))
+		msg.Payload = ut.RandomArray(i*9361, i*10)
+
+		// test digest function:
+		digest := msg.Digest()
+		t.EqualsKey(fmt.Sprintf("msg%d-digest", i), hex.EncodeToString(digest[:]))
+
+		// test stringer:
+		st := msg.String()
+		t.EqualsKey(fmt.Sprintf("msg%d-string", i), st)
+
+		// Test RLP encoding:
+		bytes, err := rlp.EncodeToBytes(&msg)
+		t.Ok(err)
+		t.EqualsKey(fmt.Sprintf("msg%d-rlp", i), hex.EncodeToString(bytes))
+
+		// Test decoding:
+		var msg2 message.Message
+		err = rlp.DecodeBytes(bytes, &msg2)
+		t.Ok(err)
+		t.Equals(msg, &msg2)
+
+	}
+}

pss/message/testdata/TestMessage/results.json

@@ -0,0 +1,17 @@
+{
+	"msg0-digest": "4b34781cfa28a5ad653855567273675eabb8535461e57e4f4bfc81504d0a828d",
+	"msg0-rlp": "dd94fa12f92afbe00f8508d0e83bab9cf8cebf42e25e03808491273d4980",
+	"msg0-string": "PssMsg: Recipient: 0xfa12f92afbe00f8508d0e83bab9cf8cebf42e25e, Topic: 0x91273d49",
+	"msg1-digest": "7f076bc036335b5d587d48c985d1b6ef8cd7015d6e484d0c7a72faddaa2aceaa",
+	"msg1-rlp": "e794210fc7bb818639ac48a4c6afa2f1581a8b9525e2000184ba78973d8aa84f7f80296fda3fd8df",
+	"msg1-string": "PssMsg: Recipient: 0x210fc7bb818639ac48a4c6afa2f1581a8b9525e2, Topic: 0xba78973d",
+	"msg2-digest": "a3cb8298779bef44c33461f072c54391a39c09b7a726e55d60384d7484760559",
+	"msg2-rlp": "f194e2aadcd868ce028477f86e430140149b0300a9a5020284a6b46dd094f4b754a41bd4d5d11330e2924ff403c95bb84fa5",
+	"msg2-string": "PssMsg: Recipient: 0xe2aadcd868ce028477f86e430140149b0300a9a5, Topic: 0xa6b46dd0",
+	"msg3-digest": "a82a894a753dffad41330dc1abbc85e5bc1791c393eba682eaf3cee56e6b0d9a",
+	"msg3-rlp": "f83b9460f9e0fa212bac5db82b22cee5272ee19a067256000384f013aa4b9e2fb3c9afcd593f3c5d3a96fecc1b7672562cc1b8828888269264bb976ed2",
+	"msg3-string": "PssMsg: Recipient: 0x60f9e0fa212bac5db82b22cee5272ee19a067256, Topic: 0xf013aa4b",
+	"msg4-digest": "8ba6836253a10cf02e5031695ab39917e816b9677d53b4e4b2af5e439b05d362",
+	"msg4-rlp": "f845941dd4751f899d743d0780c9644375aae21132781803048426f57386a834dab59240ba3bcec68fd648a62ba94062413e5b5f89c0441b5809fff0a51dd1084e8f06fce30971",
+	"msg4-string": "PssMsg: Recipient: 0x1dd4751f899d743d0780c9644375aae211327818, Topic: 0x26f57386"
+}

pss/message/topic_test.go

@@ -9,12 +9,12 @@ import (
 )
 
 // test that topic conversion functions give predictable results
+var someTopics = []string{"These", "are", "some", "topics", "A topic can be very long as well, longer than TopicLength"}
+
 func TestTopic(tx *testing.T) {
 	t := ut.BeginTest(tx, false) // set to true to generate test results
 	defer t.FinishTest()
 
-	someTopics := []string{"These", "are", "some", "topics", "A topic can be very long as well, longer than TopicLength"}
-
 	for i, topicString := range someTopics {
 		topic := message.NewTopic([]byte(topicString))