Merge pull request #376 from ethersphere/pss-raw

swarm/pss: Allow transmission of raw messages

swarm/pss/api.go

@@ -2,6 +2,7 @@ package pss
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -122,7 +123,11 @@ func (pssapi *API) GetAsymmetricAddressHint(topic Topic, pubkeyid string) (PssAd
 }
 
 func (pssapi *API) StringToTopic(topicstring string) (Topic, error) {
-	return BytesToTopic([]byte(topicstring)), nil
+	topicbytes := BytesToTopic([]byte(topicstring))
+	if topicbytes == rawTopic {
+		return rawTopic, errors.New("Topic string hashes to 0x00000000 and cannot be used")
+	}
+	return topicbytes, nil
 }
 
 func (pssapi *API) SendAsym(pubkeyhex string, topic Topic, msg hexutil.Bytes) error {

swarm/pss/protocol_test.go

@@ -37,7 +37,7 @@ func testProtocol(t *testing.T) {
 
 	topic := PingTopic.String()
 
-	clients, err := setupNetwork(2)
+	clients, err := setupNetwork(2, false)
 	if err != nil {
 		t.Fatal(err)
 	}

swarm/pss/pss.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"crypto/ecdsa"
 	"crypto/rand"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -32,7 +33,7 @@ const (
 	defaultMaxMsgSize          = 1024 * 1024
 	defaultCleanInterval       = time.Second * 60 * 10
 	defaultDequeueInterval     = time.Millisecond * 10
-	defaultOutboxQueueSize     = 10000
+	defaultOutboxCapacity      = 10000
 	pssProtocolName            = "pss"
 	pssVersion                 = 1
 	hasherCount                = 8
@@ -71,6 +72,7 @@ type PssParams struct {
 	CacheTTL            time.Duration
 	privateKey          *ecdsa.PrivateKey
 	SymKeyCacheCapacity int
+	AllowRaw            bool // If true, enables sending and receiving messages without builtin pss encryption
 }
 
 // Sane defaults for Pss
@@ -115,6 +117,7 @@ type Pss struct {
 	// message handling
 	handlers   map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle()
 	handlersMu sync.RWMutex
+	allowRaw   bool
 	hashPool   sync.Pool
 
 	// process
@@ -146,7 +149,7 @@ func NewPss(k network.Overlay, params *PssParams) *Pss {
 		msgTTL:          params.MsgTTL,
 		paddingByteSize: defaultPaddingByteSize,
 		capstring:       cap.String(),
-		outbox:          make(chan *PssMsg, defaultOutboxQueueSize),
+		outbox:          make(chan *PssMsg, defaultOutboxCapacity),
 
 		pubKeyPool:                 make(map[string]map[Topic]*pssPeer),
 		symKeyPool:                 make(map[string]map[Topic]*pssPeer),
@@ -154,6 +157,7 @@ func NewPss(k network.Overlay, params *PssParams) *Pss {
 		symKeyDecryptCacheCapacity: params.SymKeyCacheCapacity,
 
 		handlers: make(map[Topic]map[*Handler]bool),
+		allowRaw: params.AllowRaw,
 		hashPool: sync.Pool{
 			New: func() interface{} {
 				return storage.MakeHashFunc(storage.SHA3Hash)()
@@ -324,12 +328,17 @@ func (self *Pss) handlePssMsg(msg interface{}) error {
 		var err error
 		if !self.isSelfPossibleRecipient(pssmsg) {
 			log.Trace("pss was for someone else :'( ... forwarding", "pss", common.ToHex(self.BaseAddr()))
-			self.outbox <- pssmsg
+			if err := self.enqueue(pssmsg); err != nil {
+				return err
+			}
 		}
 		log.Trace("pss for us, yay! ... let's process!", "pss", common.ToHex(self.BaseAddr()))
 
-		if !self.process(pssmsg) {
+		if err := self.process(pssmsg); err != nil {
-			self.outbox <- pssmsg
+			qerr := self.enqueue(pssmsg)
+			if qerr != nil {
+				err = fmt.Errorf("%s + %s", err, qerr)
+			}
 		}
 		return err
 	}
@@ -340,7 +349,7 @@ func (self *Pss) handlePssMsg(msg interface{}) error {
 // Entry point to processing a message for which the current node can be the intended recipient.
 // Attempts symmetric and asymmetric decryption with stored keys.
 // Dispatches message to all handlers matching the message topic
-func (self *Pss) process(pssmsg *PssMsg) bool {
+func (self *Pss) process(pssmsg *PssMsg) error {
 	var err error
 	var recvmsg *whisper.ReceivedMessage
 	var from *PssAddress
@@ -350,6 +359,10 @@ func (self *Pss) process(pssmsg *PssMsg) bool {
 
 	envelope := pssmsg.Payload
 	psstopic := Topic(envelope.Topic)
+	if self.allowRaw && psstopic == rawTopic {
+		self.executeHandlers(rawTopic, envelope.Data, nil, false, "")
+		return nil
+	}
 
 	if len(envelope.AESNonce) > 0 { // detect symkey msg according to whisperv5/envelope.go:OpenSymmetric
 		keyFunc = self.processSym
@@ -359,26 +372,30 @@ func (self *Pss) process(pssmsg *PssMsg) bool {
 	}
 	recvmsg, keyid, from, err = keyFunc(envelope)
 	if err != nil {
-		log.Debug("decrypt message fail", "err", err, "asym", asymmetric, "pss", common.ToHex(self.BaseAddr()))
+		return errors.New("Decryption failed")
-		return false
 	}
 
 	if len(pssmsg.To) < addressLength {
-		go func() {
+		if err := self.enqueue(pssmsg); err != nil {
-			self.outbox <- pssmsg
+			return err
-		}()
+		}
 	}
-	handlers := self.getHandlers(psstopic)
+	self.executeHandlers(psstopic, recvmsg.Payload, from, asymmetric, keyid)
+
+	return nil
+
+}
+
+func (self *Pss) executeHandlers(topic Topic, payload []byte, from *PssAddress, asymmetric bool, keyid string) {
+	handlers := self.getHandlers(topic)
 	nid, _ := discover.HexID("0x00") // this hack is needed to satisfy the p2p method
 	p := p2p.NewPeer(nid, fmt.Sprintf("%x", from), []p2p.Cap{})
 	for f := range handlers {
-		err := (*f)(recvmsg.Payload, p, asymmetric, keyid)
+		err := (*f)(payload, p, asymmetric, keyid)
 		if err != nil {
 			log.Warn("Pss handler %p failed: %v", f, err)
 		}
 	}
-	return true
-
 }
 
 // will return false if using partial address
@@ -586,6 +603,35 @@ func (self *Pss) cleanKeys() (count int) {
 // SECTION: Message sending
 /////////////////////////////////////////////////////////////////////
 
+func (self *Pss) enqueue(msg *PssMsg) error {
+	select {
+	case self.outbox <- msg:
+		return nil
+	default:
+	}
+
+	return errors.New("outbox full")
+}
+
+// Send a raw message (any encryption is responsibility of calling client)
+//
+// Will fail if raw messages are disallowed
+func (self *Pss) SendRaw(msg []byte, address PssAddress) error {
+	if !self.allowRaw {
+		return errors.New("Raw messages not enabled")
+	}
+	pssmsg := &PssMsg{
+		To:     address,
+		Expire: uint32(time.Now().Add(self.msgTTL).Unix()),
+		Payload: &whisper.Envelope{
+			Data:  msg,
+			Topic: whisper.TopicType(rawTopic),
+		},
+	}
+	self.addFwdCache(pssmsg)
+	return self.enqueue(pssmsg)
+}
+
 // Send a message using symmetric encryption
 //
 // Fails if the key id does not match any of the stored symmetric keys
@@ -676,14 +722,13 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [
 		Expire:  uint32(time.Now().Add(self.msgTTL).Unix()),
 		Payload: envelope,
 	}
-	self.outbox <- pssmsg
+	return self.enqueue(pssmsg)
-	return nil
 }
 
 // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct
 // The recipient address can be of any length, and the byte slice will be matched to the MSB slice
 // of the peer address of the equivalent length.
-func (self *Pss) forward(msg *PssMsg) {
+func (self *Pss) forward(msg *PssMsg) error {
 	to := make([]byte, addressLength)
 	copy(to[:len(msg.To)], msg.To)
 
@@ -748,11 +793,14 @@ func (self *Pss) forward(msg *PssMsg) {
 	if sent == 0 {
 		log.Debug("unable to forward to any peers")
 		time.Sleep(time.Millisecond)
-		self.outbox <- msg
+		if err := self.enqueue(msg); err != nil {
+			return err
+		}
 	}
 
 	// cache the message
 	self.addFwdCache(msg)
+	return nil
 }
 
 /////////////////////////////////////////////////////////////////////

swarm/pss/pss_test.go

@@ -61,7 +61,7 @@ func init() {
 	flag.Parse()
 	rand.Seed(time.Now().Unix())
 
-	adapters.RegisterServices(newServices())
+	adapters.RegisterServices(newServices(false))
 	initTest()
 }
 
@@ -419,14 +419,96 @@ func TestMismatch(t *testing.T) {
 
 }
 
-// send symmetrically encrypted message between two directly connected peers
+func TestSendRaw(t *testing.T) {
-func TestSymSend(t *testing.T) {
+	t.Run("32", testSendRaw)
-	t.Run("32", testSymSend)
+	t.Run("8", testSendRaw)
-	t.Run("8", testSymSend)
+	t.Run("0", testSendRaw)
-	t.Run("0", testSymSend)
 }
 
-func testSymSend(t *testing.T) {
+func testSendRaw(t *testing.T) {
+
+	var addrsize int64
+	var err error
+
+	paramstring := strings.Split(t.Name(), "/")
+
+	addrsize, _ = strconv.ParseInt(paramstring[1], 10, 0)
+	log.Info("raw send test", "addrsize", addrsize)
+
+	clients, err := setupNetwork(2, true)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	topic := "0x00000000"
+
+	var loaddrhex string
+	err = clients[0].Call(&loaddrhex, "pss_baseAddr")
+	if err != nil {
+		t.Fatalf("rpc get node 1 baseaddr fail: %v", err)
+	}
+	loaddrhex = loaddrhex[:2+(addrsize*2)]
+	var roaddrhex string
+	err = clients[1].Call(&roaddrhex, "pss_baseAddr")
+	if err != nil {
+		t.Fatalf("rpc get node 2 baseaddr fail: %v", err)
+	}
+	roaddrhex = roaddrhex[:2+(addrsize*2)]
+
+	time.Sleep(time.Millisecond * 500)
+
+	// at this point we've verified that symkeys are saved and match on each peer
+	// now try sending symmetrically encrypted message, both directions
+	lmsgC := make(chan APIMsg)
+	lctx, lcancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer lcancel()
+	lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
+	log.Trace("lsub", "id", lsub)
+	defer lsub.Unsubscribe()
+	rmsgC := make(chan APIMsg)
+	rctx, rcancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer rcancel()
+	rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
+	log.Trace("rsub", "id", rsub)
+	defer rsub.Unsubscribe()
+
+	// send and verify delivery
+	lmsg := []byte("plugh")
+	err = clients[1].Call(nil, "pss_sendRaw", lmsg, loaddrhex)
+	if err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case recvmsg := <-lmsgC:
+		if !bytes.Equal(recvmsg.Msg, lmsg) {
+			t.Fatalf("node 1 received payload mismatch: expected %v, got %v", lmsg, recvmsg)
+		}
+	case cerr := <-lctx.Done():
+		t.Fatalf("test message (left) timed out: %v", cerr)
+	}
+	rmsg := []byte("xyzzy")
+	err = clients[0].Call(nil, "pss_sendRaw", rmsg, roaddrhex)
+	if err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case recvmsg := <-rmsgC:
+		if !bytes.Equal(recvmsg.Msg, rmsg) {
+			t.Fatalf("node 2 received payload mismatch: expected %x, got %v", rmsg, recvmsg.Msg)
+		}
+	case cerr := <-rctx.Done():
+		t.Fatalf("test message (right) timed out: %v", cerr)
+	}
+}
+
+// send symmetrically encrypted message between two directly connected peers
+func TestSendSym(t *testing.T) {
+	t.Run("32", testSendSym)
+	t.Run("8", testSendSym)
+	t.Run("0", testSendSym)
+}
+
+func testSendSym(t *testing.T) {
 
 	// address hint size
 	var addrsize int64
@@ -435,7 +517,7 @@ func testSymSend(t *testing.T) {
 	addrsize, _ = strconv.ParseInt(paramstring[1], 10, 0)
 	log.Info("sym send test", "addrsize", addrsize)
 
-	clients, err := setupNetwork(2)
+	clients, err := setupNetwork(2, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -535,13 +617,13 @@ func testSymSend(t *testing.T) {
 }
 
 // send asymmetrically encrypted message between two directly connected peers
-func TestAsymSend(t *testing.T) {
+func TestSendAsym(t *testing.T) {
-	t.Run("32", testAsymSend)
+	t.Run("32", testSendAsym)
-	t.Run("8", testAsymSend)
+	t.Run("8", testSendAsym)
-	t.Run("0", testAsymSend)
+	t.Run("0", testSendAsym)
 }
 
-func testAsymSend(t *testing.T) {
+func testSendAsym(t *testing.T) {
 
 	// address hint size
 	var addrsize int64
@@ -550,7 +632,7 @@ func testAsymSend(t *testing.T) {
 	addrsize, _ = strconv.ParseInt(paramstring[1], 10, 0)
 	log.Info("asym send test", "addrsize", addrsize)
 
-	clients, err := setupNetwork(2)
+	clients, err := setupNetwork(2, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -710,11 +792,11 @@ func testNetwork(t *testing.T) {
 		}
 		a = adapters.NewExecAdapter(dirname)
 	} else if adapter == "sock" {
-		a = adapters.NewSocketAdapter(newServices())
+		a = adapters.NewSocketAdapter(newServices(false))
 	} else if adapter == "tcp" {
-		a = adapters.NewTCPAdapter(newServices())
+		a = adapters.NewTCPAdapter(newServices(false))
 	} else if adapter == "sim" {
-		a = adapters.NewSimAdapter(newServices())
+		a = adapters.NewSimAdapter(newServices(false))
 	}
 	net := simulations.NewNetwork(a, &simulations.NetworkConfig{
 		ID: "0",
@@ -864,10 +946,12 @@ outer:
 
 }
 
+// check that in a network of a -> b -> c -> a
+// a doesn't receive a sent message twice
 func TestDeduplication(t *testing.T) {
 	var err error
 
-	clients, err := setupNetwork(3)
+	clients, err := setupNetwork(3, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1108,7 +1192,7 @@ func benchmarkSymkeyBruteforceChangeaddr(b *testing.B) {
 	}
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		if !ps.process(pssmsgs[len(pssmsgs)-(i%len(pssmsgs))-1]) {
+		if err := ps.process(pssmsgs[len(pssmsgs)-(i%len(pssmsgs))-1]); err != nil {
 			b.Fatalf("pss processing failed: %v", err)
 		}
 	}
@@ -1190,20 +1274,22 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) {
 		Payload: env,
 	}
 	for i := 0; i < b.N; i++ {
-		if !ps.process(pssmsg) {
+		if err := ps.process(pssmsg); err != nil {
 			b.Fatalf("pss processing failed: %v", err)
 		}
 	}
 }
 
-// setup simulated network and connect nodes in circle
+// setup simulated network with bzz/discovery and pss services.
-func setupNetwork(numnodes int) (clients []*rpc.Client, err error) {
+// connects nodes in a circle
+// if allowRaw is set, omission of builtin pss encryption is enabled (see PssParams)
+func setupNetwork(numnodes int, allowRaw bool) (clients []*rpc.Client, err error) {
 	nodes := make([]*simulations.Node, numnodes)
 	clients = make([]*rpc.Client, numnodes)
 	if numnodes < 2 {
 		return nil, fmt.Errorf("Minimum two nodes in network")
 	}
-	adapter := adapters.NewSimAdapter(newServices())
+	adapter := adapters.NewSimAdapter(newServices(allowRaw))
 	net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{
 		ID:             "0",
 		DefaultService: "bzz",
@@ -1239,7 +1325,7 @@ func setupNetwork(numnodes int) (clients []*rpc.Client, err error) {
 	return clients, nil
 }
 
-func newServices() adapters.Services {
+func newServices(allowRaw bool) adapters.Services {
 	stateStore := state.NewMemStore()
 	kademlias := make(map[discover.NodeID]*network.Kademlia)
 	kademlia := func(id discover.NodeID) *network.Kademlia {
@@ -1268,6 +1354,7 @@ func newServices() adapters.Services {
 			privkey, err := w.GetPrivateKey(keys)
 			pssp := NewPssParams(privkey)
 			pssp.MsgTTL = time.Second * 30
+			pssp.AllowRaw = allowRaw
 			pskad := kademlia(ctx.Config.ID)
 			ps := NewPss(pskad, pssp)
 

swarm/pss/types.go

@@ -20,6 +20,7 @@ const (
 var (
 	topicHashMutex = sync.Mutex{}
 	topicHashFunc  = storage.MakeHashFunc("SHA256")()
+	rawTopic       = Topic{}
 )
 
 type Topic whisper.TopicType