From 580d116704eeec6ff557e5bbb81a84906ae8bbae Mon Sep 17 00:00:00 2001 From: lash Date: Thu, 29 Mar 2018 20:55:41 +0200 Subject: [PATCH 1/9] swarm/pss: WIP outbox --- swarm/pss/pss.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 3a91b546d5..feb8d0832d 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -687,6 +687,12 @@ func (self *Pss) forward(msg *PssMsg) { to := make([]byte, addressLength) copy(to[:len(msg.To)], msg.To) + // message hash + digest, err := self.storeMsg(msg) + if err != nil { + log.Warn(fmt.Sprintf("could not store message %v to cache: %v", msg, err)) + } + // send with kademlia // find the closest peer to the recipient and attempt to send sent := 0 @@ -822,3 +828,17 @@ func (self *Pss) isMsgExpired(msg *PssMsg) bool { } return false } + +func (self *Pss) isMsgExpired(msg *PssMsg) bool { + msgexp := time.Unix(int64(msg.Expire), 0) + // if msgexp.Before(time.Now()) { + // log.Trace("pss expired :/ ... dropping") + // return nil + // } else if msgexp.After(time.Now().Add(self.msgTTL)) { + // return errors.New("Invalid TTL") + // } + if msgexp.Before(time.Now()) || msgexp.After(time.Now().Add(self.msgTTL)) { + return true + } + return false +} From 509a22e6cfcdcdcade46703d7eab15aa7803ac6a Mon Sep 17 00:00:00 2001 From: lash Date: Thu, 29 Mar 2018 21:42:46 +0200 Subject: [PATCH 2/9] swarm/pss: WIP deduplication --- swarm/pss/pss.go | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index feb8d0832d..3a6a01f43f 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -114,7 +114,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 + handlersMu sync.Mutex hashPool sync.Pool // process @@ -348,6 +348,14 @@ func (self *Pss) process(pssmsg *PssMsg) bool { var keyid string var keyFunc func(envelope *whisper.Envelope) (*whisper.ReceivedMessage, string, *PssAddress, error) + digest, err := self.digest(pssmsg) + if err != nil { + log.Warn(fmt.Sprintf("could not store message %v to cache: %v", pssmsg, err)) + } + if self.checkFwdCache(nil, digest) { + log.Trace(fmt.Sprintf("pss relay block-cache match (process): FROM %x TO %x", self.Overlay.BaseAddr(), common.ToHex(pssmsg.To))) + return false + } envelope := pssmsg.Payload psstopic := Topic(envelope.Topic) @@ -677,7 +685,6 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [ Payload: envelope, } self.outbox <- pssmsg - return nil } // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct @@ -688,7 +695,7 @@ func (self *Pss) forward(msg *PssMsg) { copy(to[:len(msg.To)], msg.To) // message hash - digest, err := self.storeMsg(msg) + digest, err := self.digest(msg) if err != nil { log.Warn(fmt.Sprintf("could not store message %v to cache: %v", msg, err)) } @@ -831,12 +838,6 @@ func (self *Pss) isMsgExpired(msg *PssMsg) bool { func (self *Pss) isMsgExpired(msg *PssMsg) bool { msgexp := time.Unix(int64(msg.Expire), 0) - // if msgexp.Before(time.Now()) { - // log.Trace("pss expired :/ ... dropping") - // return nil - // } else if msgexp.After(time.Now().Add(self.msgTTL)) { - // return errors.New("Invalid TTL") - // } if msgexp.Before(time.Now()) || msgexp.After(time.Now().Add(self.msgTTL)) { return true } From 48d957d61693b5d26e078ac9c6111ae92f3eb414 Mon Sep 17 00:00:00 2001 From: lash Date: Fri, 30 Mar 2018 05:03:12 +0200 Subject: [PATCH 3/9] swarm/pss: Remove pss DPA + leaner cache --- swarm/pss/protocol_test.go | 4 ++-- swarm/pss/pss.go | 22 ---------------------- 2 files changed, 2 insertions(+), 24 deletions(-) diff --git a/swarm/pss/protocol_test.go b/swarm/pss/protocol_test.go index b30fc0430d..a4079307d7 100644 --- a/swarm/pss/protocol_test.go +++ b/swarm/pss/protocol_test.go @@ -22,9 +22,9 @@ type protoCtrl struct { // simple ping pong protocol test for the pss devp2p emulation func TestProtocol(t *testing.T) { - t.Run("32", testProtocol) + //t.Run("32", testProtocol) t.Run("8", testProtocol) - t.Run("0", testProtocol) + //t.Run("0", testProtocol) } func testProtocol(t *testing.T) { diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 3a6a01f43f..27d28be99f 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -348,14 +348,6 @@ func (self *Pss) process(pssmsg *PssMsg) bool { var keyid string var keyFunc func(envelope *whisper.Envelope) (*whisper.ReceivedMessage, string, *PssAddress, error) - digest, err := self.digest(pssmsg) - if err != nil { - log.Warn(fmt.Sprintf("could not store message %v to cache: %v", pssmsg, err)) - } - if self.checkFwdCache(nil, digest) { - log.Trace(fmt.Sprintf("pss relay block-cache match (process): FROM %x TO %x", self.Overlay.BaseAddr(), common.ToHex(pssmsg.To))) - return false - } envelope := pssmsg.Payload psstopic := Topic(envelope.Topic) @@ -694,12 +686,6 @@ func (self *Pss) forward(msg *PssMsg) { to := make([]byte, addressLength) copy(to[:len(msg.To)], msg.To) - // message hash - digest, err := self.digest(msg) - if err != nil { - log.Warn(fmt.Sprintf("could not store message %v to cache: %v", msg, err)) - } - // send with kademlia // find the closest peer to the recipient and attempt to send sent := 0 @@ -835,11 +821,3 @@ func (self *Pss) isMsgExpired(msg *PssMsg) bool { } return false } - -func (self *Pss) isMsgExpired(msg *PssMsg) bool { - msgexp := time.Unix(int64(msg.Expire), 0) - if msgexp.Before(time.Now()) || msgexp.After(time.Now().Add(self.msgTTL)) { - return true - } - return false -} From c6ab94079c8c0bdfd1769c6551405303d7e92f54 Mon Sep 17 00:00:00 2001 From: lash Date: Fri, 30 Mar 2018 03:17:23 +0200 Subject: [PATCH 4/9] swarm/pss: Enable raw message sending --- swarm/pss/api.go | 7 ++- swarm/pss/protocol_test.go | 2 +- swarm/pss/pss.go | 42 +++++++++++++++-- swarm/pss/pss_test.go | 97 +++++++++++++++++++++++++++++++++++--- swarm/pss/types.go | 1 + 5 files changed, 135 insertions(+), 14 deletions(-) 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 a4079307d7..a9a6b5bdd5 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 27d28be99f..d891c1f68c 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -71,6 +71,7 @@ type PssParams struct { CacheTTL time.Duration privateKey *ecdsa.PrivateKey SymKeyCacheCapacity int + AllowRaw bool } // Sane defaults for Pss @@ -115,6 +116,7 @@ type Pss struct { // message handling handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle() handlersMu sync.Mutex + allowRaw bool hashPool sync.Pool // process @@ -154,6 +156,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)() @@ -359,7 +362,9 @@ 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())) + if self.allowRaw { + self.executeHandlers(rawTopic, envelope.Data, nil, false, "") + } return false } @@ -368,17 +373,25 @@ func (self *Pss) process(pssmsg *PssMsg) bool { self.outbox <- pssmsg }() } - handlers := self.getHandlers(psstopic) + if psstopic == rawTopic { + return false + } + self.executeHandlers(psstopic, recvmsg.Payload, from, asymmetric, keyid) + + return true + +} + +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 +599,25 @@ func (self *Pss) cleanKeys() (count int) { // SECTION: Message sending ///////////////////////////////////////////////////////////////////// +// 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 diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 9eaba00ccc..9af5080ccf 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,6 +419,88 @@ func TestMismatch(t *testing.T) { } +func TestRawSend(t *testing.T) { + t.Run("32", testRawSend) + t.Run("8", testRawSend) + t.Run("0", testRawSend) +} + +func testRawSend(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 TestSymSend(t *testing.T) { t.Run("32", testSymSend) @@ -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) } @@ -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) } @@ -867,7 +949,7 @@ outer: func TestDeduplication(t *testing.T) { var err error - clients, err := setupNetwork(3) + clients, err := setupNetwork(3, false) if err != nil { t.Fatal(err) } @@ -1197,13 +1279,13 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) { } // setup simulated network and connect nodes in circle -func setupNetwork(numnodes int) (clients []*rpc.Client, err error) { +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 +1321,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 +1350,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 From e50144273377e4ddf7f928afe04ae76fc939c6e4 Mon Sep 17 00:00:00 2001 From: lash Date: Fri, 6 Apr 2018 09:23:25 +0200 Subject: [PATCH 5/9] swarm/pss: Rebase after deduplication changes --- swarm/pss/protocol_test.go | 4 ++-- swarm/pss/pss.go | 7 +++++-- swarm/pss/pss_test.go | 6 +++--- 3 files changed, 10 insertions(+), 7 deletions(-) diff --git a/swarm/pss/protocol_test.go b/swarm/pss/protocol_test.go index a9a6b5bdd5..319e3991dd 100644 --- a/swarm/pss/protocol_test.go +++ b/swarm/pss/protocol_test.go @@ -22,9 +22,9 @@ type protoCtrl struct { // simple ping pong protocol test for the pss devp2p emulation func TestProtocol(t *testing.T) { - //t.Run("32", testProtocol) + t.Run("32", testProtocol) t.Run("8", testProtocol) - //t.Run("0", testProtocol) + t.Run("0", testProtocol) } func testProtocol(t *testing.T) { diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index d891c1f68c..66c20184b3 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" @@ -115,7 +116,7 @@ type Pss struct { // message handling handlers map[Topic]map[*Handler]bool // topic and version based pss payload handlers. See pss.Handle() - handlersMu sync.Mutex + handlersMu sync.RWMutex allowRaw bool hashPool sync.Pool @@ -615,7 +616,8 @@ func (self *Pss) SendRaw(msg []byte, address PssAddress) error { }, } self.addFwdCache(pssmsg) - return self.enqueue(pssmsg) + self.outbox <- pssmsg + return nil } // Send a message using symmetric encryption @@ -709,6 +711,7 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [ Payload: envelope, } self.outbox <- pssmsg + return nil } // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 9af5080ccf..1c0084be44 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -792,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", From 221e55bee68736419c3d3616457463e6e152da3a Mon Sep 17 00:00:00 2001 From: lash Date: Tue, 10 Apr 2018 12:12:48 +0200 Subject: [PATCH 6/9] swarm/pss: Add documentation for AllowRaw param --- swarm/pss/pss.go | 2 +- swarm/pss/pss_test.go | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 66c20184b3..b788fd050a 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -72,7 +72,7 @@ type PssParams struct { CacheTTL time.Duration privateKey *ecdsa.PrivateKey SymKeyCacheCapacity int - AllowRaw bool + AllowRaw bool // If true, enables sending and receiving messages without builtin pss encryption } // Sane defaults for Pss diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 1c0084be44..6d06c9f264 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -946,6 +946,8 @@ 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 @@ -1278,7 +1280,9 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) { } } -// setup simulated network and connect nodes in circle +// 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) From 538c31101fe8cfadbe887a8d0c1f2ca48a5a2c0b Mon Sep 17 00:00:00 2001 From: lash Date: Tue, 10 Apr 2018 20:57:07 +0200 Subject: [PATCH 7/9] swarm/pss: Add counter and error on full outbox --- swarm/pss/pss.go | 53 ++++++++++++++++++++++++++++++++++--------- swarm/pss/pss_test.go | 30 ++++++++++++------------ 2 files changed, 57 insertions(+), 26 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index b788fd050a..d2eeff5189 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -33,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 @@ -104,6 +104,8 @@ type Pss struct { paddingByteSize int capstring string outbox chan *PssMsg + outboxCapacity int + outboxCounter int // keys and peers pubKeyPool map[string]map[Topic]*pssPeer // mapping of hex public keys to peer address by topic. @@ -149,7 +151,8 @@ 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), + outboxCapacity: defaultOutboxCapacity, pubKeyPool: make(map[string]map[Topic]*pssPeer), symKeyPool: make(map[string]map[Topic]*pssPeer), @@ -328,12 +331,16 @@ 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 !self.enqueue(pssmsg) { + return errors.New("outbox full!") + } } log.Trace("pss for us, yay! ... let's process!", "pss", common.ToHex(self.BaseAddr())) if !self.process(pssmsg) { - self.outbox <- pssmsg + if !self.enqueue(pssmsg) { + return errors.New("outbox full!") + } } return err } @@ -370,9 +377,10 @@ func (self *Pss) process(pssmsg *PssMsg) bool { } if len(pssmsg.To) < addressLength { - go func() { - self.outbox <- pssmsg - }() + if !self.enqueue(pssmsg) { + log.Error("outbox full!") + return false + } } if psstopic == rawTopic { return false @@ -600,6 +608,17 @@ func (self *Pss) cleanKeys() (count int) { // SECTION: Message sending ///////////////////////////////////////////////////////////////////// +func (self *Pss) enqueue(msg *PssMsg) bool { + self.fwdPoolMu.Lock() + defer self.fwdPoolMu.Unlock() + if self.outboxCounter == self.outboxCapacity { + return false + } + self.outboxCounter++ + self.outbox <- msg + return true +} + // Send a raw message (any encryption is responsibility of calling client) // // Will fail if raw messages are disallowed @@ -616,7 +635,9 @@ func (self *Pss) SendRaw(msg []byte, address PssAddress) error { }, } self.addFwdCache(pssmsg) - self.outbox <- pssmsg + if !self.enqueue(pssmsg) { + return errors.New("outbox full!") + } return nil } @@ -710,14 +731,16 @@ 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 + if !self.enqueue(pssmsg) { + return errors.New("outbox full!") + } 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) @@ -782,11 +805,19 @@ 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 !self.enqueue(msg) { + return errors.New("outbox full!") + } } + // remove from queue + self.fwdPoolMu.Lock() + self.outboxCounter-- + self.fwdPoolMu.Unlock() + // cache the message self.addFwdCache(msg) + return nil } ///////////////////////////////////////////////////////////////////// diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 6d06c9f264..9970881bd6 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -419,13 +419,13 @@ func TestMismatch(t *testing.T) { } -func TestRawSend(t *testing.T) { - t.Run("32", testRawSend) - t.Run("8", testRawSend) - t.Run("0", testRawSend) +func TestSendRaw(t *testing.T) { + t.Run("32", testSendRaw) + t.Run("8", testSendRaw) + t.Run("0", testSendRaw) } -func testRawSend(t *testing.T) { +func testSendRaw(t *testing.T) { var addrsize int64 var err error @@ -502,13 +502,13 @@ func testRawSend(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 TestSendSym(t *testing.T) { + t.Run("32", testSendSym) + t.Run("8", testSendSym) + t.Run("0", testSendSym) } -func testSymSend(t *testing.T) { +func testSendSym(t *testing.T) { // address hint size var addrsize int64 @@ -617,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 From 3e10f30bcafa62422330fbf57accd19a1fac90c1 Mon Sep 17 00:00:00 2001 From: lash Date: Tue, 10 Apr 2018 21:07:04 +0200 Subject: [PATCH 8/9] swarm/pss: Move rawtopic test before decryption attempt --- swarm/pss/pss.go | 25 ++++++++----------------- 1 file changed, 8 insertions(+), 17 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index d2eeff5189..95cacd7da5 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -361,6 +361,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 true + } if len(envelope.AESNonce) > 0 { // detect symkey msg according to whisperv5/envelope.go:OpenSymmetric keyFunc = self.processSym @@ -370,9 +374,6 @@ func (self *Pss) process(pssmsg *PssMsg) bool { } recvmsg, keyid, from, err = keyFunc(envelope) if err != nil { - if self.allowRaw { - self.executeHandlers(rawTopic, envelope.Data, nil, false, "") - } return false } @@ -382,9 +383,6 @@ func (self *Pss) process(pssmsg *PssMsg) bool { return false } } - if psstopic == rawTopic { - return false - } self.executeHandlers(psstopic, recvmsg.Payload, from, asymmetric, keyid) return true @@ -609,13 +607,11 @@ func (self *Pss) cleanKeys() (count int) { ///////////////////////////////////////////////////////////////////// func (self *Pss) enqueue(msg *PssMsg) bool { - self.fwdPoolMu.Lock() - defer self.fwdPoolMu.Unlock() - if self.outboxCounter == self.outboxCapacity { - return false + select { + case self.outbox <- msg: + return true + default: } - self.outboxCounter++ - self.outbox <- msg return true } @@ -810,11 +806,6 @@ func (self *Pss) forward(msg *PssMsg) error { } } - // remove from queue - self.fwdPoolMu.Lock() - self.outboxCounter-- - self.fwdPoolMu.Unlock() - // cache the message self.addFwdCache(msg) return nil From ecde73011d77eeba1433a52cbe9004b08e8fdb02 Mon Sep 17 00:00:00 2001 From: lash Date: Wed, 11 Apr 2018 09:00:14 +0200 Subject: [PATCH 9/9] swarm/pss: Add error to outbox --- swarm/pss/pss.go | 48 ++++++++++++++++++------------------------- swarm/pss/pss_test.go | 4 ++-- 2 files changed, 22 insertions(+), 30 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 95cacd7da5..a2ce5de75a 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -104,8 +104,6 @@ type Pss struct { paddingByteSize int capstring string outbox chan *PssMsg - outboxCapacity int - outboxCounter int // keys and peers pubKeyPool map[string]map[Topic]*pssPeer // mapping of hex public keys to peer address by topic. @@ -152,7 +150,6 @@ func NewPss(k network.Overlay, params *PssParams) *Pss { paddingByteSize: defaultPaddingByteSize, capstring: cap.String(), outbox: make(chan *PssMsg, defaultOutboxCapacity), - outboxCapacity: defaultOutboxCapacity, pubKeyPool: make(map[string]map[Topic]*pssPeer), symKeyPool: make(map[string]map[Topic]*pssPeer), @@ -331,15 +328,16 @@ 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())) - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") + 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 !self.enqueue(pssmsg) { - return errors.New("outbox full!") + if err := self.process(pssmsg); err != nil { + qerr := self.enqueue(pssmsg) + if qerr != nil { + err = fmt.Errorf("%s + %s", err, qerr) } } return err @@ -351,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 @@ -363,7 +361,7 @@ func (self *Pss) process(pssmsg *PssMsg) bool { psstopic := Topic(envelope.Topic) if self.allowRaw && psstopic == rawTopic { self.executeHandlers(rawTopic, envelope.Data, nil, false, "") - return true + return nil } if len(envelope.AESNonce) > 0 { // detect symkey msg according to whisperv5/envelope.go:OpenSymmetric @@ -374,18 +372,17 @@ func (self *Pss) process(pssmsg *PssMsg) bool { } recvmsg, keyid, from, err = keyFunc(envelope) if err != nil { - return false + return errors.New("Decryption failed") } if len(pssmsg.To) < addressLength { - if !self.enqueue(pssmsg) { - log.Error("outbox full!") - return false + if err := self.enqueue(pssmsg); err != nil { + return err } } self.executeHandlers(psstopic, recvmsg.Payload, from, asymmetric, keyid) - return true + return nil } @@ -606,13 +603,14 @@ func (self *Pss) cleanKeys() (count int) { // SECTION: Message sending ///////////////////////////////////////////////////////////////////// -func (self *Pss) enqueue(msg *PssMsg) bool { +func (self *Pss) enqueue(msg *PssMsg) error { select { case self.outbox <- msg: - return true + return nil default: } - return true + + return errors.New("outbox full") } // Send a raw message (any encryption is responsibility of calling client) @@ -631,10 +629,7 @@ func (self *Pss) SendRaw(msg []byte, address PssAddress) error { }, } self.addFwdCache(pssmsg) - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") - } - return nil + return self.enqueue(pssmsg) } // Send a message using symmetric encryption @@ -727,10 +722,7 @@ func (self *Pss) send(to []byte, topic Topic, msg []byte, asymmetric bool, key [ Expire: uint32(time.Now().Add(self.msgTTL).Unix()), Payload: envelope, } - if !self.enqueue(pssmsg) { - return errors.New("outbox full!") - } - return nil + return self.enqueue(pssmsg) } // Forwards a pss message to the peer(s) closest to the to recipient address in the PssMsg struct @@ -801,8 +793,8 @@ func (self *Pss) forward(msg *PssMsg) error { if sent == 0 { log.Debug("unable to forward to any peers") time.Sleep(time.Millisecond) - if !self.enqueue(msg) { - return errors.New("outbox full!") + if err := self.enqueue(msg); err != nil { + return err } } diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 9970881bd6..29961ee3ae 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -1192,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) } } @@ -1274,7 +1274,7 @@ 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) } }