diff --git a/whisper/mailserver/mailserver.go b/whisper/mailserver/mailserver.go index 57e6505ad1..b63294958a 100644 --- a/whisper/mailserver/mailserver.go +++ b/whisper/mailserver/mailserver.go @@ -101,7 +101,7 @@ func (s *WMailServer) Archive(env *whisper.Envelope) { } } -func (s *WMailServer) DeliverMail(peer *whisper.Peer, request *whisper.Envelope) { +func (s *WMailServer) DeliverMail(peer whisper.Peer, request *whisper.Envelope) { if peer == nil { log.Error("Whisper peer is nil") return @@ -113,7 +113,7 @@ func (s *WMailServer) DeliverMail(peer *whisper.Peer, request *whisper.Envelope) } } -func (s *WMailServer) processRequest(peer *whisper.Peer, lower, upper uint32, bloom []byte) []*whisper.Envelope { +func (s *WMailServer) processRequest(peer whisper.Peer, lower, upper uint32, bloom []byte) []*whisper.Envelope { ret := make([]*whisper.Envelope, 0) var err error var zero common.Hash @@ -169,7 +169,7 @@ func (s *WMailServer) validateRequest(peerID []byte, request *whisper.Envelope) } // if you want to check the signature, you can do it here. e.g.: - // if !bytes.Equal(peerID, src) { + // if peerID == src { if src == nil { log.Warn(fmt.Sprintf("Wrong signature of p2p request")) return false, 0, 0, nil diff --git a/whisper/whisperv6/api.go b/whisper/whisperv6/api.go index c60bc46a13..b2c5522407 100644 --- a/whisper/whisperv6/api.go +++ b/whisper/whisperv6/api.go @@ -32,6 +32,10 @@ import ( "github.com/ethereum/go-ethereum/rpc" ) +const ( + filterTimeout = 300 // filters are considered timeout out after filterTimeout seconds +) + // List of errors var ( ErrSymAsym = errors.New("specify either a symmetric or an asymmetric key") @@ -107,7 +111,7 @@ func (api *PublicWhisperAPI) MarkTrustedPeer(ctx context.Context, enode string) if err != nil { return false, err } - return true, api.w.AllowP2PMessagesFromPeer(n.ID[:]) + return true, api.w.AllowP2PMessagesFromPeer(n.ID.String()) } // NewKeyPair generates a new public and private key pair for message decryption and encryption. @@ -296,7 +300,7 @@ func (api *PublicWhisperAPI) Post(ctx context.Context, req NewMessage) (hexutil. if err != nil { return nil, fmt.Errorf("failed to parse target peer: %s", err) } - err = api.w.SendP2PMessage(n.ID[:], env) + err = api.w.SendP2PMessage(n.ID.String(), env) if err == nil { hash := env.Hash() result = hash[:] @@ -566,10 +570,9 @@ func (api *PublicWhisperAPI) NewMessageFilter(req Criteria) (string, error) { } if len(req.Topics) > 0 { - topics = make([][]byte, len(req.Topics)) - for i, topic := range req.Topics { - topics[i] = make([]byte, TopicLength) - copy(topics[i], topic[:]) + topics = make([][]byte, 0, len(req.Topics)) + for _, topic := range req.Topics { + topics = append(topics, topic[:]) } } diff --git a/whisper/whisperv6/doc.go b/whisper/whisperv6/doc.go index 066a9766d4..900c06c727 100644 --- a/whisper/whisperv6/doc.go +++ b/whisper/whisperv6/doc.go @@ -93,5 +93,5 @@ func (e unknownVersionError) Error() string { // in order to bypass the expiry checks. type MailServer interface { Archive(env *Envelope) - DeliverMail(whisperPeer *Peer, request *Envelope) + DeliverMail(whisperPeer Peer, request *Envelope) } diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go index 2bf1c905b2..81982f80b3 100644 --- a/whisper/whisperv6/peer.go +++ b/whisper/whisperv6/peer.go @@ -29,10 +29,9 @@ import ( set "gopkg.in/fatih/set.v0" ) -// Peer represents a whisper protocol peer connection. -type Peer struct { +// PeerBase represents a whisper protocol peer connection. +type PeerBase struct { host *Whisper - peer *p2p.Peer ws p2p.MsgReadWriter trusted bool @@ -46,37 +45,70 @@ type Peer struct { quit chan struct{} } +// Peer is an abstract representation of a whisper peer. It could +// represent a devp2p peer or a libp2p peer. +type Peer interface { + ID() string + start() + stop() + handshake() error + update() + mark(*Envelope) + marked(*Envelope) bool + expire() + broadcast() error + notifyAboutPowRequirementChange(pow float64) error + notifyAboutBloomFilterChange([]byte) error + bloomMatch(*Envelope) bool + setBloomFilter([]byte) + isTrusted() bool + setTrusted(bool) + setPoWRequirement(float64) + stream() p2p.MsgReadWriter + + // newPeer(*Whisper, p2p.MsgReadWriter) Peer +} + +// DevP2PPeer is the DevP2P implementation of the Peer interface +type DevP2PPeer struct { + *PeerBase + + peer *p2p.Peer +} + // newPeer creates a new whisper peer object, but does not run the handshake itself. -func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer { - return &Peer{ - host: host, - peer: remote, - ws: rw, - trusted: false, - powRequirement: 0.0, - known: set.New(), - quit: make(chan struct{}), - bloomFilter: MakeFullNodeBloom(), - fullNode: true, +func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer { + return &DevP2PPeer{ + &PeerBase { + host: host, + ws: rw, + trusted: false, + powRequirement: 0.0, + known: set.New(), + quit: make(chan struct{}), + bloomFilter: MakeFullNodeBloom(), + fullNode: true, + }, + remote, } } // start initiates the peer updater, periodically broadcasting the whisper packets // into the network. -func (peer *Peer) start() { +func (peer *DevP2PPeer) start() { go peer.update() log.Trace("start", "peer", peer.ID()) } // stop terminates the peer updater, stopping message forwarding to it. -func (peer *Peer) stop() { +func (peer *DevP2PPeer) stop() { close(peer.quit) log.Trace("stop", "peer", peer.ID()) } // handshake sends the protocol initiation status message to the remote peer and // verifies the remote status too. -func (peer *Peer) handshake() error { +func (peer *DevP2PPeer) handshake() error { // Send the handshake status message asynchronously errc := make(chan error, 1) go func() { @@ -135,7 +167,7 @@ func (peer *Peer) handshake() error { // update executes periodic operations on the peer, including message transmission // and expiration. -func (peer *Peer) update() { +func (peer *DevP2PPeer) update() { // Start the tickers for the updates expire := time.NewTicker(expirationCycle) transmit := time.NewTicker(transmissionCycle) @@ -159,18 +191,18 @@ func (peer *Peer) update() { } // mark marks an envelope known to the peer so that it won't be sent back. -func (peer *Peer) mark(envelope *Envelope) { +func (peer *DevP2PPeer) mark(envelope *Envelope) { peer.known.Add(envelope.Hash()) } // marked checks if an envelope is already known to the remote peer. -func (peer *Peer) marked(envelope *Envelope) bool { +func (peer *DevP2PPeer) marked(envelope *Envelope) bool { return peer.known.Has(envelope.Hash()) } // expire iterates over all the known envelopes in the host and removes all // expired (unknown) ones from the known list. -func (peer *Peer) expire() { +func (peer *DevP2PPeer) expire() { unmark := make(map[common.Hash]struct{}) peer.known.Each(func(v interface{}) bool { if !peer.host.isEnvelopeCached(v.(common.Hash)) { @@ -186,7 +218,7 @@ func (peer *Peer) expire() { // broadcast iterates over the collection of envelopes and transmits yet unknown // ones over the network. -func (peer *Peer) broadcast() error { +func (peer *DevP2PPeer) broadcast() error { envelopes := peer.host.Envelopes() bundle := make([]*Envelope, 0, len(envelopes)) for _, envelope := range envelopes { @@ -212,27 +244,26 @@ func (peer *Peer) broadcast() error { } // ID returns a peer's id -func (peer *Peer) ID() []byte { - id := peer.peer.ID() - return id[:] +func (peer *DevP2PPeer) ID() string { + return peer.peer.ID().String() } -func (peer *Peer) notifyAboutPowRequirementChange(pow float64) error { +func (peer *DevP2PPeer) notifyAboutPowRequirementChange(pow float64) error { i := math.Float64bits(pow) return p2p.Send(peer.ws, powRequirementCode, i) } -func (peer *Peer) notifyAboutBloomFilterChange(bloom []byte) error { +func (peer *DevP2PPeer) notifyAboutBloomFilterChange(bloom []byte) error { return p2p.Send(peer.ws, bloomFilterExCode, bloom) } -func (peer *Peer) bloomMatch(env *Envelope) bool { +func (peer *DevP2PPeer) bloomMatch(env *Envelope) bool { peer.bloomMu.Lock() defer peer.bloomMu.Unlock() return peer.fullNode || BloomFilterMatch(peer.bloomFilter, env.Bloom()) } -func (peer *Peer) setBloomFilter(bloom []byte) { +func (peer *DevP2PPeer) setBloomFilter(bloom []byte) { peer.bloomMu.Lock() defer peer.bloomMu.Unlock() peer.bloomFilter = bloom @@ -242,6 +273,22 @@ func (peer *Peer) setBloomFilter(bloom []byte) { } } +func (peer *DevP2PPeer) isTrusted() bool { + return peer.trusted +} + +func (peer *DevP2PPeer) setTrusted(t bool) { + peer.trusted = t +} + +func (peer *DevP2PPeer) setPoWRequirement(r float64) { + peer.powRequirement = r +} + +func (peer *DevP2PPeer) stream() p2p.MsgReadWriter { + return peer.ws +} + func MakeFullNodeBloom() []byte { bloom := make([]byte, BloomFilterSize) for i := 0; i < BloomFilterSize; i++ { diff --git a/whisper/whisperv6/peer_test.go b/whisper/whisperv6/peer_test.go index ec985ae65b..6add42fc95 100644 --- a/whisper/whisperv6/peer_test.go +++ b/whisper/whisperv6/peer_test.go @@ -314,6 +314,7 @@ func validateMail(t *testing.T, index int, mail []*ReceivedMessage) { } if cnt > 1 { t.Fatalf("node %d received %d.", index, cnt) + return false } if cnt == 1 { @@ -437,9 +438,9 @@ func checkPowExchangeForNodeZeroOnce(t *testing.T, mustPass bool) bool { cnt := 0 for i, node := range nodes { for peer := range node.shh.peers { - if peer.peer.ID() == discover.PubkeyID(&nodes[0].id.PublicKey) { + if peer.ID() == discover.PubkeyID(&nodes[0].id.PublicKey).String() { cnt++ - if peer.powRequirement != masterPow { + if peer.(*DevP2PPeer).powRequirement != masterPow { if mustPass { t.Fatalf("node %d: failed to set the new pow requirement for node zero.", i) } else { @@ -458,10 +459,10 @@ func checkPowExchangeForNodeZeroOnce(t *testing.T, mustPass bool) bool { 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 { + if peer.ID() != discover.PubkeyID(&nodes[0].id.PublicKey).String() { + if peer.(*DevP2PPeer).powRequirement != masterPow { t.Fatalf("node %d: failed to exchange pow requirement in round %d; expected %f, got %f", - i, round, masterPow, peer.powRequirement) + i, round, masterPow, peer.(*DevP2PPeer).powRequirement) } } } @@ -477,7 +478,7 @@ func checkBloomFilterExchangeOnce(t *testing.T, mustPass bool) bool { if !equals { if mustPass { t.Fatalf("node %d: failed to exchange bloom filter requirement in round %d. \n%x expected \n%x got", - i, round, masterBloomFilter, peer.bloomFilter) + i, round, masterBloomFilter, peer.(*DevP2PPeer).bloomFilter) } else { return false } diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index 880cced098..5c873ca049 100644 --- a/whisper/whisperv6/whisper.go +++ b/whisper/whisperv6/whisper.go @@ -17,7 +17,6 @@ package whisperv6 import ( - "bytes" "crypto/ecdsa" "crypto/sha256" "fmt" @@ -72,7 +71,7 @@ type Whisper struct { expirations map[uint32]*set.SetNonTS // Message expiration pool peerMu sync.RWMutex // Mutex to sync the active peer set - peers map[*Peer]struct{} // Set of currently active peers + peers map[Peer]struct{} // Set of currently active peers messageQueue chan *Envelope // Message queue for normal whisper messages p2pMsgQueue chan *Envelope // Message queue for peer-to-peer messages (not to be forwarded any further) @@ -101,7 +100,7 @@ func New(cfg *Config) *Whisper { symKeys: make(map[string][]byte), envelopes: make(map[common.Hash]*Envelope), expirations: make(map[uint32]*set.SetNonTS), - peers: make(map[*Peer]struct{}), + peers: make(map[Peer]struct{}), messageQueue: make(chan *Envelope, messageQueueLimit), p2pMsgQueue: make(chan *Envelope, messageQueueLimit), quit: make(chan struct{}), @@ -304,8 +303,8 @@ func (whisper *Whisper) notifyPeersAboutBloomFilterChange(bloom []byte) { } } -func (whisper *Whisper) getPeers() []*Peer { - arr := make([]*Peer, len(whisper.peers)) +func (whisper *Whisper) getPeers() []Peer { + arr := make([]Peer, len(whisper.peers)) i := 0 whisper.peerMu.Lock() for p := range whisper.peers { @@ -317,12 +316,12 @@ func (whisper *Whisper) getPeers() []*Peer { } // getPeer retrieves peer by ID -func (whisper *Whisper) getPeer(peerID []byte) (*Peer, error) { +func (whisper *Whisper) getPeer(peerID string) (Peer, error) { whisper.peerMu.Lock() defer whisper.peerMu.Unlock() for p := range whisper.peers { - id := p.peer.ID() - if bytes.Equal(peerID, id[:]) { + id := p.ID() + if peerID == id { return p, nil } } @@ -331,12 +330,12 @@ func (whisper *Whisper) getPeer(peerID []byte) (*Peer, error) { // AllowP2PMessagesFromPeer marks specific peer trusted, // which will allow it to send historic (expired) messages. -func (whisper *Whisper) AllowP2PMessagesFromPeer(peerID []byte) error { +func (whisper *Whisper) AllowP2PMessagesFromPeer(peerID string) error { p, err := whisper.getPeer(peerID) if err != nil { return err } - p.trusted = true + p.setTrusted(true) return nil } @@ -345,17 +344,17 @@ func (whisper *Whisper) AllowP2PMessagesFromPeer(peerID []byte) error { // request and respond with a number of peer-to-peer messages (possibly expired), // which are not supposed to be forwarded any further. // The whisper protocol is agnostic of the format and contents of envelope. -func (whisper *Whisper) RequestHistoricMessages(peerID []byte, envelope *Envelope) error { +func (whisper *Whisper) RequestHistoricMessages(peerID string, envelope *Envelope) error { p, err := whisper.getPeer(peerID) if err != nil { return err } - p.trusted = true - return p2p.Send(p.ws, p2pRequestCode, envelope) + p.setTrusted(true) + return p2p.Send(p.stream(), p2pRequestCode, envelope) } // SendP2PMessage sends a peer-to-peer message to a specific peer. -func (whisper *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error { +func (whisper *Whisper) SendP2PMessage(peerID string, envelope *Envelope) error { p, err := whisper.getPeer(peerID) if err != nil { return err @@ -364,8 +363,8 @@ func (whisper *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error } // SendP2PDirect sends a peer-to-peer message to a specific peer. -func (whisper *Whisper) SendP2PDirect(peer *Peer, envelope *Envelope) error { - return p2p.Send(peer.ws, p2pMessageCode, envelope) +func (whisper *Whisper) SendP2PDirect(peer Peer, envelope *Envelope) error { + return p2p.Send(peer.stream(), p2pMessageCode, envelope) } // NewKeyPair generates a new cryptographic identity for the client, and injects @@ -645,28 +644,28 @@ func (whisper *Whisper) HandlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error { } // runMessageLoop reads and processes inbound messages directly to merge into client-global state. -func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { +func (whisper *Whisper) runMessageLoop(p Peer, rw p2p.MsgReadWriter) error { for { // fetch the next packet packet, err := rw.ReadMsg() if err != nil { - log.Warn("message loop", "peer", p.peer.ID(), "err", err) + log.Warn("message loop", "peer", p.ID(), "err", err) return err } if packet.Size > whisper.MaxMessageSize() { - log.Warn("oversized message received", "peer", p.peer.ID()) + log.Warn("oversized message received", "peer", p.ID()) return errors.New("oversized message received") } switch packet.Code { case statusCode: // this should not happen, but no need to panic; just ignore this message. - log.Warn("unxepected status message received", "peer", p.peer.ID()) + log.Warn("unxepected status message received", "peer", p.ID()) case messagesCode: // decode the contained envelopes var envelopes []*Envelope if err := packet.Decode(&envelopes); err != nil { - log.Warn("failed to decode envelopes, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Warn("failed to decode envelopes, peer will be disconnected", "peer", p.ID(), "err", err) return errors.New("invalid envelopes") } @@ -675,7 +674,7 @@ func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { cached, err := whisper.add(env, whisper.lightClient) if err != nil { trouble = true - log.Error("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Error("bad envelope received, peer will be disconnected", "peer", p.ID(), "err", err) } if cached { p.mark(env) @@ -689,15 +688,15 @@ func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { s := rlp.NewStream(packet.Payload, uint64(packet.Size)) i, err := s.Uint() if err != nil { - log.Warn("failed to decode powRequirementCode message, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Warn("failed to decode powRequirementCode message, peer will be disconnected", "peer", p.ID(), "err", err) return errors.New("invalid powRequirementCode message") } f := math.Float64frombits(i) if math.IsInf(f, 0) || math.IsNaN(f) || f < 0.0 { - log.Warn("invalid value in powRequirementCode message, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Warn("invalid value in powRequirementCode message, peer will be disconnected", "peer", p.ID(), "err", err) return errors.New("invalid value in powRequirementCode message") } - p.powRequirement = f + p.setPoWRequirement(f) case bloomFilterExCode: var bloom []byte err := packet.Decode(&bloom) @@ -706,7 +705,7 @@ func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { } if err != nil { - log.Warn("failed to decode bloom filter exchange message, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Warn("failed to decode bloom filter exchange message, peer will be disconnected", "peer", p.ID(), "err", err) return errors.New("invalid bloom filter exchange message") } p.setBloomFilter(bloom) @@ -715,10 +714,10 @@ func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { // this message is not supposed to be forwarded to other peers, and // therefore might not satisfy the PoW, expiry and other requirements. // these messages are only accepted from the trusted peer. - if p.trusted { + if p.isTrusted() { var envelope Envelope if err := packet.Decode(&envelope); err != nil { - log.Warn("failed to decode direct message, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Warn("failed to decode direct message, peer will be disconnected", "peer", p.ID(), "err", err) return errors.New("invalid direct message") } whisper.postEvent(&envelope, true) @@ -728,7 +727,7 @@ func (whisper *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { if whisper.mailServer != nil { var request Envelope if err := packet.Decode(&request); err != nil { - log.Warn("failed to decode p2p request message, peer will be disconnected", "peer", p.peer.ID(), "err", err) + log.Warn("failed to decode p2p request message, peer will be disconnected", "peer", p.ID(), "err", err) return errors.New("invalid p2p request") } whisper.mailServer.DeliverMail(p, &request) diff --git a/whisper/whisperv6/whisper_test.go b/whisper/whisperv6/whisper_test.go index 7fe256309e..e46ac92196 100644 --- a/whisper/whisperv6/whisper_test.go +++ b/whisper/whisperv6/whisper_test.go @@ -17,6 +17,7 @@ package whisperv6 import ( + "fmt" "bytes" "crypto/ecdsa" "crypto/sha256" @@ -51,8 +52,12 @@ func TestWhisperBasic(t *testing.T) { t.Fatalf("failed GetFilter.") } - peerID := make([]byte, 64) - mrand.Read(peerID) + peerIDBytes := make([]byte, 64) + mrand.Read(peerIDBytes) + peerID := "" + for b := range peerIDBytes { + peerID = fmt.Sprintf("%s%02x", peerID, b) + } peer, _ := w.getPeer(peerID) if peer != nil { t.Fatal("found peer for random key.")