diff --git a/swarm/pss/api.go b/swarm/pss/api.go index 6505d33023..1720f9dba4 100644 --- a/swarm/pss/api.go +++ b/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 { diff --git a/swarm/pss/protocol_test.go b/swarm/pss/protocol_test.go index b30fc0430d..319e3991dd 100644 --- a/swarm/pss/protocol_test.go +++ b/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) } diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 3a91b546d5..a2ce5de75a 100644 --- a/swarm/pss/pss.go +++ b/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) { - self.outbox <- pssmsg + if err := self.process(pssmsg); err != nil { + 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 false + return errors.New("Decryption failed") } if len(pssmsg.To) < addressLength { - go func() { - self.outbox <- pssmsg - }() + if err := self.enqueue(pssmsg); err != nil { + 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 nil + return self.enqueue(pssmsg) } // 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 } ///////////////////////////////////////////////////////////////////// diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 9eaba00ccc..29961ee3ae 100644 --- a/swarm/pss/pss_test.go +++ b/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 TestSymSend(t *testing.T) { - t.Run("32", testSymSend) - t.Run("8", testSymSend) - t.Run("0", testSymSend) +func TestSendRaw(t *testing.T) { + t.Run("32", testSendRaw) + t.Run("8", testSendRaw) + t.Run("0", testSendRaw) } -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) { - t.Run("32", testAsymSend) - t.Run("8", testAsymSend) - t.Run("0", testAsymSend) +func TestSendAsym(t *testing.T) { + t.Run("32", testSendAsym) + t.Run("8", testSendAsym) + 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 -func setupNetwork(numnodes int) (clients []*rpc.Client, err error) { +// setup simulated network with bzz/discovery and pss services. +// 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) diff --git a/swarm/pss/types.go b/swarm/pss/types.go index 9c6e49bb8f..dbab753d7a 100644 --- a/swarm/pss/types.go +++ b/swarm/pss/types.go @@ -20,6 +20,7 @@ const ( var ( topicHashMutex = sync.Mutex{} topicHashFunc = storage.MakeHashFunc("SHA256")() + rawTopic = Topic{} ) type Topic whisper.TopicType