From becaadd13da0b7671710fdf6b589bbc0e2227309 Mon Sep 17 00:00:00 2001 From: Vlad Date: Wed, 3 Jan 2018 21:16:48 +0200 Subject: [PATCH] whisper: Status message changed, more params added --- whisper/whisperv6/peer.go | 38 +++++++++- whisper/whisperv6/peer_test.go | 116 ++++++++++++++++++++++++------ whisper/whisperv6/whisper.go | 51 ++++++------- whisper/whisperv6/whisper_test.go | 61 ++++++++++++++++ 4 files changed, 214 insertions(+), 52 deletions(-) diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go index b21f07e031..08071c0f77 100644 --- a/whisper/whisperv6/peer.go +++ b/whisper/whisperv6/peer.go @@ -75,8 +75,12 @@ func (p *Peer) handshake() error { // Send the handshake status message asynchronously errc := make(chan error, 1) go func() { - errc <- p2p.Send(p.ws, statusCode, ProtocolVersion) + pow := p.host.MinPow() + powConverted := math.Float64bits(pow) + bloom := p.host.BloomFilter() + errc <- p2p.SendItems(p.ws, statusCode, ProtocolVersion, powConverted, bloom) }() + // Fetch the remote status packet and verify protocol match packet, err := p.ws.ReadMsg() if err != nil { @@ -86,14 +90,42 @@ func (p *Peer) handshake() error { return fmt.Errorf("peer [%x] sent packet %x before status packet", p.ID(), packet.Code) } s := rlp.NewStream(packet.Payload, uint64(packet.Size)) - peerVersion, err := s.Uint() + _, err = s.List() if err != nil { return fmt.Errorf("peer [%x] sent bad status message: %v", p.ID(), err) } + peerVersion, err := s.Uint() + if err != nil { + return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", p.ID(), err) + } if peerVersion != ProtocolVersion { return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", p.ID(), peerVersion, ProtocolVersion) } - // Wait until out own status is consumed too + + // only version is mandatory, subsequent parameters are optional + powRaw, err := s.Uint() + if err == nil { + pow := math.Float64frombits(powRaw) + if math.IsInf(pow, 0) || math.IsNaN(pow) || pow < 0.0 { + return fmt.Errorf("peer [%x] sent bad status message: invalid pow", p.ID()) + } + p.powRequirement = pow + + var bloom []byte + err = s.Decode(&bloom) + if err == nil { + sz := len(bloom) + if sz != bloomFilterSize && sz != 0 { + return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", p.ID(), sz) + } + if isFullNode(bloom) { + p.bloomFilter = nil + } else { + p.bloomFilter = bloom + } + } + } + if err := <-errc; err != nil { return fmt.Errorf("peer [%x] failed to send status packet: %v", p.ID(), err) } diff --git a/whisper/whisperv6/peer_test.go b/whisper/whisperv6/peer_test.go index 599a479be4..b6d0dd02fa 100644 --- a/whisper/whisperv6/peer_test.go +++ b/whisper/whisperv6/peer_test.go @@ -20,6 +20,7 @@ import ( "bytes" "crypto/ecdsa" "fmt" + mrand "math/rand" "net" "sync" "testing" @@ -87,6 +88,9 @@ var nodes [NumNodes]*TestNode var sharedKey []byte = []byte("some arbitrary data here") var sharedTopic TopicType = TopicType{0xF, 0x1, 0x2, 0} var expectedMessage []byte = []byte("per rectum ad astra") +var masterBloomFilter []byte +var masterPow = 0.00000001 +var round int = 1 func TestSimulation(t *testing.T) { // create a chain of whisper nodes, @@ -104,8 +108,13 @@ func TestSimulation(t *testing.T) { // check if each node have received and decrypted exactly one message checkPropagation(t, true) - // send protocol-level messages (powRequirementCode) and check the new PoW requirement values - powReqExchange(t) + // check if Status message was correctly decoded + checkBloomFilterExchange(t) + checkPowExchange(t) + + // send new pow and bloom exchange messages + resetParams(t) + round++ // node #1 sends one expected (decryptable) message sendMsg(t, true, 1) @@ -113,18 +122,65 @@ func TestSimulation(t *testing.T) { // check if each node (except node #0) have received and decrypted exactly one message checkPropagation(t, false) + for i := 1; i < NumNodes; i++ { + time.Sleep(20 * time.Millisecond) + sendMsg(t, true, i) + } + + // check if corresponding protocol-level messages were correctly decoded + checkPowExchangeForNodeZero(t) + checkBloomFilterExchange(t) + stopServers() } +func resetParams(t *testing.T) { + // change pow only for node zero + masterPow = 7777777.0 + nodes[0].shh.SetMinimumPoW(masterPow) + + // change bloom for all nodes + masterBloomFilter = TopicToBloom(sharedTopic) + for i := 0; i < NumNodes; i++ { + nodes[i].shh.SetBloomFilter(masterBloomFilter) + } +} + +func initBloom(t *testing.T) { + masterBloomFilter = make([]byte, bloomFilterSize) + _, err := mrand.Read(masterBloomFilter) + if err != nil { + t.Fatalf("rand failed: %s.", err) + } + + msgBloom := TopicToBloom(sharedTopic) + masterBloomFilter = addBloom(masterBloomFilter, msgBloom) + for i := 0; i < 32; i++ { + masterBloomFilter[i] = 0xFF + } + + if !bloomFilterMatch(masterBloomFilter, msgBloom) { + t.Fatalf("bloom mismatch on initBloom.") + } +} + func initialize(t *testing.T) { + initBloom(t) + var err error ip := net.IPv4(127, 0, 0, 1) port0 := 30303 for i := 0; i < NumNodes; i++ { var node TestNode + b := make([]byte, bloomFilterSize) + copy(b, masterBloomFilter) node.shh = New(&DefaultConfig) - node.shh.SetMinimumPowTest(0.00000001) + node.shh.SetMinimumPoW(masterPow) + node.shh.SetBloomFilter(b) + if !isBloomFilterEqual(node.shh.BloomFilter(), masterBloomFilter) { + t.Fatalf("bloom mismatch on init.") + } node.shh.Start(nil) topics := make([]TopicType, 0) topics = append(topics, sharedTopic) @@ -206,7 +262,7 @@ func checkPropagation(t *testing.T, includingNodeZero bool) { for i := first; i < NumNodes; i++ { f := nodes[i].shh.GetFilter(nodes[i].filerId) if f == nil { - t.Fatalf("failed to get filterId %s from node %d.", nodes[i].filerId, i) + t.Fatalf("failed to get filterId %s from node %d, round %d.", nodes[i].filerId, i, round) } mail := f.Retrieve() @@ -332,34 +388,52 @@ func TestPeerBasic(t *testing.T) { } } -func powReqExchange(t *testing.T) { - for i, node := range nodes { - for peer := range node.shh.peers { - if peer.powRequirement > 1000.0 { - t.Fatalf("node %d: one of the peers' pow requirement is too big (%f).", i, peer.powRequirement) - } - } - } - - const pow float64 = 7777777.0 - nodes[0].shh.SetMinimumPoW(pow) - - // wait until all the messages are delivered - time.Sleep(64 * time.Millisecond) - +func checkPowExchangeForNodeZero(t *testing.T) { cnt := 0 for i, node := range nodes { for peer := range node.shh.peers { if peer.peer.ID() == discover.PubkeyID(&nodes[0].id.PublicKey) { cnt++ - if peer.powRequirement != pow { + if peer.powRequirement != masterPow { t.Fatalf("node %d: failed to set the new pow requirement.", i) } } } } - if cnt == 0 { t.Fatalf("no matching peers found.") } } + +func checkPowExchange(t *testing.T) { + for i, node := range nodes { + for peer := range node.shh.peers { + if peer.peer.ID() != discover.PubkeyID(&nodes[0].id.PublicKey) { + if peer.powRequirement != masterPow { + t.Fatalf("node %d: failed to exchange pow requirement in round %d; expected %f, got %f", + i, round, masterPow, peer.powRequirement) + } + } + } + } +} + +func checkBloomFilterExchange(t *testing.T) { + for i, node := range nodes { + for peer := range node.shh.peers { + if !isBloomFilterEqual(peer.bloomFilter, masterBloomFilter) { + t.Fatalf("node %d: failed to exchange bloom filter requirement in round %d. \n%x expected \n%x got", + i, round, masterBloomFilter, peer.bloomFilter) + } + } + } +} + +func isBloomFilterEqual(a, b []byte) bool { + for i := 0; i < bloomFilterSize; i++ { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index 9ffd7faa24..dd99bc3aae 100644 --- a/whisper/whisperv6/whisper.go +++ b/whisper/whisperv6/whisper.go @@ -130,32 +130,32 @@ func New(cfg *Config) *Whisper { } func (w *Whisper) MinPow() float64 { - val, _ := w.settings.Load(minPowIdx) - if val == nil { + val, exist := w.settings.Load(minPowIdx) + if !exist || val == nil { return DefaultMinimumPoW } return val.(float64) } func (w *Whisper) MinPowTolerance() float64 { - val, _ := w.settings.Load(minPowToleranceIdx) - if val == nil { + val, exist := w.settings.Load(minPowToleranceIdx) + if !exist || val == nil { return DefaultMinimumPoW } return val.(float64) } func (w *Whisper) BloomFilter() []byte { - val, _ := w.settings.Load(bloomFilterIdx) - if val == nil { + val, exist := w.settings.Load(bloomFilterIdx) + if !exist || val == nil { return nil } return val.([]byte) } func (w *Whisper) BloomFilterTolerance() []byte { - val, _ := w.settings.Load(bloomFilterToleranceIdx) - if val == nil { + val, exist := w.settings.Load(bloomFilterToleranceIdx) + if !exist || val == nil { return nil } return val.([]byte) @@ -216,13 +216,16 @@ func (w *Whisper) SetBloomFilter(bloom []byte) error { return fmt.Errorf("invalid bloom filter size: %d", len(bloom)) } - w.settings.Store(bloomFilterIdx, bloom) - w.notifyPeersAboutBloomFilterChange(bloom) + b := make([]byte, bloomFilterSize) + copy(b, bloom) + + w.settings.Store(bloomFilterIdx, b) + w.notifyPeersAboutBloomFilterChange(b) go func() { // allow some time before all the peers have processed the notification time.Sleep(time.Duration(w.syncAllowance) * time.Second) - w.settings.Store(bloomFilterToleranceIdx, bloom) + w.settings.Store(bloomFilterToleranceIdx, b) }() return nil @@ -253,13 +256,6 @@ func (w *Whisper) SetMinimumPowTest(val float64) { w.settings.Store(minPowToleranceIdx, val) } -// SetBloomFilterTest sets the Bloom Filter in test environment -func (w *Whisper) SetBloomFilterTest(bloom []byte) { - w.settings.Store(bloomFilterIdx, bloom) - w.notifyPeersAboutBloomFilterChange(bloom) - w.settings.Store(bloomFilterToleranceIdx, bloom) -} - func (w *Whisper) notifyPeersAboutPowRequirementChange(pow float64) { arr := w.getPeers() for _, p := range arr { @@ -697,7 +693,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { log.Warn("failed to decode bloom filter exchange message, peer will be disconnected", "peer", p.peer.ID(), "err", err) return errors.New("invalid bloom filter exchange message") } - if isFulNode(bloom) { + if isFullNode(bloom) { p.bloomFilter = nil } else { p.bloomFilter = bloom @@ -765,21 +761,20 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) { if envelope.PoW() < wh.MinPow() { // maybe the value was recently changed, and the peers did not adjust yet. - // some tolerance might be still allowed for a short period of adjustment time. + // in this case the previous value is retrieved by MinPowTolerance() + // for a short period of peer synchronization. if envelope.PoW() < wh.MinPowTolerance() { - log.Debug("envelope with low PoW dropped", "PoW", envelope.PoW(), "hash", envelope.Hash().Hex()) - return false, nil // drop envelope without error for now + return false, fmt.Errorf("envelope with low PoW received: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex()) } - - // once the status message includes the PoW requirement, an error should be returned here: - //return false, fmt.Errorf("envelope with low PoW received: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex()) } if !bloomFilterMatch(wh.BloomFilter(), envelope.Bloom()) { // maybe the value was recently changed, and the peers did not adjust yet. - // some tolerance might be still allowed for a short period of adjustment time. + // in this case the previous value is retrieved by BloomFilterTolerance() + // for a short period of peer synchronization. if !bloomFilterMatch(wh.BloomFilterTolerance(), envelope.Bloom()) { - return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v]", envelope.Hash().Hex()) + return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v], bloom: \n%x \n%x \n%x", + envelope.Hash().Hex(), wh.BloomFilter(), envelope.Bloom(), envelope.Topic) } } @@ -1019,7 +1014,7 @@ func GenerateRandomID() (id string, err error) { return id, err } -func isFulNode(bloom []byte) bool { +func isFullNode(bloom []byte) bool { if bloom == nil { return true } diff --git a/whisper/whisperv6/whisper_test.go b/whisper/whisperv6/whisper_test.go index b391a1161b..fa14acb1b1 100644 --- a/whisper/whisperv6/whisper_test.go +++ b/whisper/whisperv6/whisper_test.go @@ -843,3 +843,64 @@ func TestSymmetricSendKeyMismatch(t *testing.T) { t.Fatalf("received a message when keys weren't matching") } } + +func TestBloom(t *testing.T) { + topic := TopicType{0, 0, 255, 6} + b := TopicToBloom(topic) + x := make([]byte, bloomFilterSize) + x[0] = byte(1) + x[32] = byte(1) + x[bloomFilterSize-1] = byte(128) + if !bloomFilterMatch(x, b) || !bloomFilterMatch(b, x) { + t.Fatalf("bloom filter does not match the mask") + } + + _, err := mrand.Read(b) + if err != nil { + t.Fatalf("math rand error") + } + _, err = mrand.Read(x) + if err != nil { + t.Fatalf("math rand error") + } + if !bloomFilterMatch(b, b) { + t.Fatalf("bloom filter does not match self") + } + x = addBloom(x, b) + if !bloomFilterMatch(x, b) { + t.Fatalf("bloom filter does not match combined bloom") + } + if !isFullNode(nil) { + t.Fatalf("isFullNode did not recognize nil as full node") + } + x[17] = 254 + if isFullNode(x) { + t.Fatalf("isFullNode false positive") + } + for i := 0; i < bloomFilterSize; i++ { + b[i] = byte(255) + } + if !isFullNode(b) { + t.Fatalf("isFullNode false negative") + } + if bloomFilterMatch(x, b) { + t.Fatalf("bloomFilterMatch false positive") + } + if !bloomFilterMatch(b, x) { + t.Fatalf("bloomFilterMatch false negative") + } + + w := New(&DefaultConfig) + f := w.BloomFilter() + if f != nil { + t.Fatalf("wrong bloom on creation") + } + err = w.SetBloomFilter(x) + if err != nil { + t.Fatalf("failed to set bloom filter: %s", err) + } + f = w.BloomFilter() + if !bloomFilterMatch(f, x) || !bloomFilterMatch(x, f) { + t.Fatalf("retireved wrong bloom filter") + } +}