whisper: abstract Peer struct to support libp2p and devp2p

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

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))
+		topics = make([][]byte, 0, len(req.Topics))
-		for i, topic := range req.Topics {
+		for _, topic := range req.Topics {
-			topics[i] = make([]byte, TopicLength)
+			topics = append(topics, topic[:])
-			copy(topics[i], topic[:])
 		}
 	}
 

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)
 }

whisper/whisperv6/peer.go

@@ -29,10 +29,9 @@ import (
 	set "gopkg.in/fatih/set.v0"
 )
 
-// Peer represents a whisper protocol peer connection.
+// PeerBase represents a whisper protocol peer connection.
-type Peer struct {
+type PeerBase struct {
 	host *Whisper
-	peer *p2p.Peer
 	ws   p2p.MsgReadWriter
 
 	trusted        bool
@@ -46,11 +45,42 @@ 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 {
+func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) Peer {
-	return &Peer{
+	return &DevP2PPeer{
+		&PeerBase {
 			host:           host,
-		peer:           remote,
 			ws:             rw,
 			trusted:        false,
 			powRequirement: 0.0,
@@ -58,25 +88,27 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
 			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 {
+func (peer *DevP2PPeer) ID() string {
-	id := peer.peer.ID()
+	return peer.peer.ID().String()
-	return id[:]
 }
 
-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++ {

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.ID() != discover.PubkeyID(&nodes[0].id.PublicKey).String() {
-				if peer.powRequirement != masterPow {
+				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
 				}

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 {
+func (whisper *Whisper) getPeers() []Peer {
-	arr := make([]*Peer, len(whisper.peers))
+	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()
+		id := p.ID()
-		if bytes.Equal(peerID, 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
+	p.setTrusted(true)
-	return p2p.Send(p.ws, p2pRequestCode, envelope)
+	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 {
+func (whisper *Whisper) SendP2PDirect(peer Peer, envelope *Envelope) error {
-	return p2p.Send(peer.ws, p2pMessageCode, envelope)
+	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)

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)
+	peerIDBytes := make([]byte, 64)
-	mrand.Read(peerID)
+	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.")