diff --git a/cmd/wnode/main.go b/cmd/wnode/main.go index 05e6b29086..75768e7fbe 100644 --- a/cmd/wnode/main.go +++ b/cmd/wnode/main.go @@ -21,6 +21,7 @@ package main import ( "bufio" + "context" "crypto/ecdsa" "crypto/sha512" "encoding/binary" @@ -39,24 +40,38 @@ import ( "github.com/ethereum/go-ethereum/console" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" - "github.com/ethereum/go-ethereum/p2p/nat" "github.com/ethereum/go-ethereum/whisper/mailserver" whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" + "github.com/ethereum/go-ethereum/whisper/whisperv6" "golang.org/x/crypto/pbkdf2" + + libp2pCrypto "github.com/libp2p/go-libp2p-crypto" + libp2pHost "github.com/libp2p/go-libp2p-host" + peer "github.com/libp2p/go-libp2p-peer" + ma "github.com/multiformats/go-multiaddr" + // inet "gx/ipfs/QmU4vCDZTPLDqSDKguWbHCiUe46mZUtmM2g2suBZ9NE8ko/go-libp2p-net" + // swarm "gx/ipfs/QmUhvp4VoQ9cKDVLqAxciEKdm8ymBx2Syx4C1Tv6SmSTPa/go-libp2p-swarm" + // peer "gx/ipfs/QmWNY7dV54ZDYmTA1ykVdwNCqC11mpU4zSUp6XDpLTH9eG/go-libp2p-peer" + + // ma "gx/ipfs/QmW8s4zTsUoX1Q6CeYxVKPyqSKbF7H1YDUyTostBtZ8DaG/go-multiaddr" + + inet "github.com/libp2p/go-libp2p-net" ) const quitCommand = "~Q" // singletons var ( - server *p2p.Server + network *inet.Network + host libp2pHost.Host shh *whisper.Whisper done chan struct{} mailServer mailserver.WMailServer input = bufio.NewReader(os.Stdin) + + bootnode peer.ID ) // encryption @@ -64,7 +79,7 @@ var ( symKey []byte pub *ecdsa.PublicKey asymKey *ecdsa.PrivateKey - nodeid *ecdsa.PrivateKey + nodeid libp2pCrypto.PrivKey topic whisper.TopicType asymKeyID string filterID string @@ -91,7 +106,6 @@ var ( argPoW = flag.Float64("pow", whisper.DefaultMinimumPoW, "PoW for normal messages in float format (e.g. 2.7)") argServerPoW = flag.Float64("mspow", whisper.DefaultMinimumPoW, "PoW requirement for Mail Server request") - argIP = flag.String("ip", "", "IP address and port of this node (e.g. 127.0.0.1:30303)") argPub = flag.String("pub", "", "public key for asymmetric encryption") argDBPath = flag.String("dbpath", "", "path to the server's DB directory") argIDFile = flag.String("idfile", "", "file name with node id (private key)") @@ -111,16 +125,14 @@ func processArgs() { if len(*argIDFile) > 0 { var err error - nodeid, err = crypto.LoadECDSA(*argIDFile) + keyData, err := ioutil.ReadFile(*argIDFile) if err != nil { utils.Fatalf("Failed to load file [%s]: %s.", *argIDFile, err) } - } - const enodePrefix = "enode://" - if len(*argEnode) > 0 { - if (*argEnode)[:len(enodePrefix)] != enodePrefix { - *argEnode = enodePrefix + *argEnode + nodeid, err = libp2pCrypto.UnmarshalPrivateKey(keyData) + if err != nil { + utils.Fatalf("Failed to load file [%s]: %s.", *argIDFile, err) } } @@ -157,7 +169,7 @@ func echo() { fmt.Printf("workTime = %d \n", *argWorkTime) fmt.Printf("pow = %f \n", *argPoW) fmt.Printf("mspow = %f \n", *argServerPoW) - fmt.Printf("ip = %s \n", *argIP) + fmt.Printf("ip = %v \n", host.Addrs()[0].Protocols()) fmt.Printf("pub = %s \n", common.ToHex(crypto.FromECDSAPub(pub))) fmt.Printf("idfile = %s \n", *argIDFile) fmt.Printf("dbpath = %s \n", *argDBPath) @@ -168,7 +180,6 @@ func initialize() { log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*argVerbosity), log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) done = make(chan struct{}) - var peers []*discover.Node var err error if *generateKey { @@ -186,16 +197,23 @@ func initialize() { msPassword = "wwww" } - if *bootstrapMode { - if len(*argIP) == 0 { - argIP = scanLineA("Please enter your IP and port (e.g. 127.0.0.1:30348): ") - } - } else { + if !*bootstrapMode { if len(*argEnode) == 0 { - argEnode = scanLineA("Please enter the peer's enode: ") + argEnode = scanLineA("Please enter the peer's address: ") } - peer := discover.MustParseNode(*argEnode) - peers = append(peers, peer) + peeraddr, err := ma.NewMultiaddr(*argEnode) + if err != nil { + utils.Fatalf("Invalid address: %s", err) + } + pid, err := peeraddr.ValueForProtocol(ma.P_IPFS) + if err != nil { + utils.Fatalf("Could not find peer id: %s", err) + } + peerid, err := peer.IDB58Decode(pid) + if err != nil { + utils.Fatalf("Could not decode peer id: %s", err) + } + bootnode = peerid } cfg := &whisper.Config{ @@ -242,56 +260,30 @@ func initialize() { utils.Fatalf("Failed to retrieve a new key pair: %s", err) } - if nodeid == nil { - tmpID, err := shh.NewKeyPair() - if err != nil { - utils.Fatalf("Failed to generate a new key pair: %s", err) - } - - nodeid, err = shh.GetPrivateKey(tmpID) + if nodeid != nil { + nodeid, _, err = libp2pCrypto.GenerateKeyPair(libp2pCrypto.Ed25519, 384) if err != nil { utils.Fatalf("Failed to retrieve a new key pair: %s", err) } } - - maxPeers := 80 - if *bootstrapMode { - maxPeers = 800 - } - - server = &p2p.Server{ - Config: p2p.Config{ - PrivateKey: nodeid, - MaxPeers: maxPeers, - Name: common.MakeName("wnode", "5.0"), - Protocols: shh.Protocols(), - ListenAddr: *argIP, - NAT: nat.Any(), - BootstrapNodes: peers, - StaticNodes: peers, - TrustedNodes: peers, - }, - } } func startServer() { - err := server.Start() - if err != nil { - utils.Fatalf("Failed to start Whisper peer: %s.", err) - } - fmt.Printf("my public key: %s \n", common.ToHex(crypto.FromECDSAPub(&asymKey.PublicKey))) - fmt.Println(server.NodeInfo().Enode) + fmt.Println(host.ID()) if *bootstrapMode { - configureNode() - fmt.Println("Bootstrap Whisper node started") + fmt.Print("Bootstrap ") } else { - fmt.Println("Whisper node started") - // first see if we can establish connection, then ask for user input - waitForConnection(true) - configureNode() + // first see if we can establish connection + _, err := host.NewStream(context.Background(), bootnode, whisperv6.WhisperProtocolString) + if err != nil { + utils.Fatalf("Error connecting to bootstrap node: %s", err) + } + } + fmt.Println("Whisper node started") + configureNode() if !*forwarderMode { fmt.Printf("Please type the message. To quit type: '%s'\n", quitCommand) @@ -382,27 +374,10 @@ func generateTopic(password []byte) { } } -func waitForConnection(timeout bool) { - var cnt int - var connected bool - for !connected { - time.Sleep(time.Millisecond * 50) - connected = server.PeerCount() > 0 - if timeout { - cnt++ - if cnt > 1000 { - utils.Fatalf("Timeout expired, failed to connect") - } - } - } - - fmt.Println("Connected to peer.") -} - func run() { defer mailServer.Close() startServer() - defer server.Stop() + defer host.Close() shh.Start(nil) defer shh.Stop() @@ -601,7 +576,7 @@ func requestExpiredMessagesLoop() { if err != nil { utils.Fatalf("Failed to save symmetric key for mail request: %s", err) } - peerID = extractIdFromEnode(*argEnode) + peerID = extractIDFromEnode(*argEnode) shh.AllowP2PMessagesFromPeer(peerID) for { @@ -627,11 +602,19 @@ func requestExpiredMessagesLoop() { data = data[:8] } + nodeiddata, err := nodeid.Bytes() + if err != nil { + utils.Fatalf("Error converting libp2p key to geth key: %s", err) + } + nodeidkey, err := crypto.ToECDSA(nodeiddata) + if err != nil { + utils.Fatalf("Error converting libp2p key to geth key: %s", err) + } var params whisper.MessageParams params.PoW = *argServerPoW params.Payload = data params.KeySym = key - params.Src = nodeid + params.Src = nodeidkey params.WorkTime = 5 msg, err := whisper.NewSentMessage(¶ms) @@ -652,7 +635,7 @@ func requestExpiredMessagesLoop() { } } -func extractIdFromEnode(s string) []byte { +func extractIDFromEnode(s string) []byte { n, err := discover.ParseNode(s) if err != nil { utils.Fatalf("Failed to parse enode: %s", err) diff --git a/whisper/whisperv6/api.go b/whisper/whisperv6/api.go index 2f6f671e9e..974149dc1d 100644 --- a/whisper/whisperv6/api.go +++ b/whisper/whisperv6/api.go @@ -21,6 +21,7 @@ import ( "crypto/ecdsa" "errors" "fmt" + "strings" "sync" "time" @@ -28,8 +29,9 @@ import ( "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rpc" + + peer "github.com/libp2p/go-libp2p-peer" ) const ( @@ -126,14 +128,32 @@ func (api *PublicWhisperAPI) SetBloomFilter(ctx context.Context, bloom hexutil.B return true, api.w.SetBloomFilter(bloom) } +// This is a helper for turning an address into a peer.ID. It +// is temporary as there seems to be an official way to to it +// with `Multiaddres`. +func peerIDFromAddress(addr string) (peer.ID, error) { + comps := strings.Split(addr, "/") + var pid peer.ID + var err error + if comps[len(comps)-1] == "" && len(comps) > 1 { + pid, err = peer.IDFromString(comps[len(comps)-2]) + } else { + pid, err = peer.IDFromString(comps[len(comps)-1]) + } + + if err != nil { + log.Error(fmt.Sprintf("Error getting id from enode: %s", err)) + } + + return pid, err +} + // MarkTrustedPeer marks a peer trusted, which will allow it to send historic (expired) messages. // Note: This function is not adding new nodes, the node needs to exists as a peer. -func (api *PublicWhisperAPI) MarkTrustedPeer(ctx context.Context, enode string) (bool, error) { - n, err := discover.ParseNode(enode) - if err != nil { - return false, err - } - return true, api.w.AllowP2PMessagesFromPeer(n.ID[:]) +func (api *PublicWhisperAPI) MarkTrustedPeer(ctx context.Context, url string) (bool, error) { + pid, err := peerIDFromAddress(url) + + return err == nil, api.w.AllowP2PMessagesFromPeer(pid) } // NewKeyPair generates a new public and private key pair for message decryption and encryption. @@ -303,11 +323,11 @@ func (api *PublicWhisperAPI) Post(ctx context.Context, req NewMessage) (bool, er // send to specific node (skip PoW check) if len(req.TargetPeer) > 0 { - n, err := discover.ParseNode(req.TargetPeer) + pid, err := peerIDFromAddress(req.TargetPeer) if err != nil { return false, fmt.Errorf("failed to parse target peer: %s", err) } - return true, api.w.SendP2PMessage(n.ID[:], env) + return true, api.w.SendP2PMessage(pid, env) } // ensure that the message PoW meets the node's minimum accepted PoW diff --git a/whisper/whisperv6/doc.go b/whisper/whisperv6/doc.go index da1b4ee5ba..09af5d811a 100644 --- a/whisper/whisperv6/doc.go +++ b/whisper/whisperv6/doc.go @@ -72,6 +72,9 @@ const ( DefaultTTL = 50 // seconds DefaultSyncAllowance = 10 // seconds + + WhisperPort = 534848 + WhisperProtocolString = "/whisper/6.0" ) type unknownVersionError uint64 diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go index 08071c0f77..918041393a 100644 --- a/whisper/whisperv6/peer.go +++ b/whisper/whisperv6/peer.go @@ -26,12 +26,14 @@ import ( "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" set "gopkg.in/fatih/set.v0" + + libp2pPeer "github.com/libp2p/go-libp2p-peer" ) // peer represents a whisper protocol peer connection. type Peer struct { host *Whisper - peer *p2p.Peer + pid libp2pPeer.ID ws p2p.MsgReadWriter trusted bool @@ -44,10 +46,10 @@ type Peer struct { } // 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 *libp2pPeer.ID, rw p2p.MsgReadWriter) *Peer { return &Peer{ host: host, - peer: remote, + pid: *remote, ws: rw, trusted: false, powRequirement: 0.0, @@ -60,13 +62,13 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer { // into the network. func (p *Peer) start() { go p.update() - log.Trace("start", "peer", p.ID()) + log.Trace("start", "peer", p.pid) } // stop terminates the peer updater, stopping message forwarding to it. func (p *Peer) stop() { close(p.quit) - log.Trace("stop", "peer", p.ID()) + log.Trace("stop", "peer", p.pid) } // handshake sends the protocol initiation status message to the remote peer and @@ -87,19 +89,19 @@ func (p *Peer) handshake() error { return err } if packet.Code != statusCode { - return fmt.Errorf("peer [%x] sent packet %x before status packet", p.ID(), packet.Code) + return fmt.Errorf("peer [%x] sent packet %x before status packet", p.pid, packet.Code) } s := rlp.NewStream(packet.Payload, uint64(packet.Size)) _, err = s.List() if err != nil { - return fmt.Errorf("peer [%x] sent bad status message: %v", p.ID(), err) + return fmt.Errorf("peer [%x] sent bad status message: %v", p.pid, 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) + return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", p.pid, err) } if peerVersion != ProtocolVersion { - return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", p.ID(), peerVersion, ProtocolVersion) + return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", p.pid, peerVersion, ProtocolVersion) } // only version is mandatory, subsequent parameters are optional @@ -107,7 +109,7 @@ func (p *Peer) handshake() error { 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()) + return fmt.Errorf("peer [%x] sent bad status message: invalid pow", p.pid) } p.powRequirement = pow @@ -116,7 +118,7 @@ func (p *Peer) handshake() error { 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) + return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", p.pid, sz) } if isFullNode(bloom) { p.bloomFilter = nil @@ -127,7 +129,7 @@ func (p *Peer) handshake() error { } if err := <-errc; err != nil { - return fmt.Errorf("peer [%x] failed to send status packet: %v", p.ID(), err) + return fmt.Errorf("peer [%x] failed to send status packet: %v", p.pid, err) } return nil } @@ -147,7 +149,7 @@ func (p *Peer) update() { case <-transmit.C: if err := p.broadcast(); err != nil { - log.Trace("broadcast failed", "reason", err, "peer", p.ID()) + log.Trace("broadcast failed", "reason", err, "peer", p.pid) return } @@ -210,11 +212,6 @@ func (p *Peer) broadcast() error { return nil } -func (p *Peer) ID() []byte { - id := p.peer.ID() - return id[:] -} - func (p *Peer) notifyAboutPowRequirementChange(pow float64) error { i := math.Float64bits(pow) return p2p.Send(p.ws, powRequirementCode, i) diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index bc89aadccd..f8bfd72e9c 100644 --- a/whisper/whisperv6/whisper.go +++ b/whisper/whisperv6/whisper.go @@ -17,11 +17,12 @@ package whisperv6 import ( - "bytes" + "context" "crypto/ecdsa" crand "crypto/rand" "crypto/sha256" "fmt" + "io/ioutil" "math" "runtime" "sync" @@ -33,6 +34,12 @@ import ( "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" + libp2p "github.com/libp2p/go-libp2p" + libp2pCrypto "github.com/libp2p/go-libp2p-crypto" + host "github.com/libp2p/go-libp2p-host" + inet "github.com/libp2p/go-libp2p-net" + peer "github.com/libp2p/go-libp2p-peer" + protocol "github.com/libp2p/go-libp2p-protocol" "github.com/syndtr/goleveldb/leveldb/errors" "golang.org/x/crypto/pbkdf2" "golang.org/x/sync/syncmap" @@ -59,8 +66,9 @@ const ( // Whisper represents a dark communication interface through the Ethereum // network, using its very own P2P communication layer. type Whisper struct { - protocol p2p.Protocol // Protocol description and parameters - filters *Filters // Message filters installed with Subscribe function + protocol protocol.ID // Protocol description and parameters + host host.Host + filters *Filters // Message filters installed with Subscribe function privateKeys map[string]*ecdsa.PrivateKey // Private key storage symKeys map[string][]byte // Symmetric key storage @@ -93,6 +101,28 @@ func New(cfg *Config) *Whisper { cfg = &DefaultConfig } + priv, _, err := libp2pCrypto.GenerateKeyPair(libp2pCrypto.Ed25519, 384) + // TODO check if this doesn't reveal my private key, which would + // be very bad. Alternatively, use the crypto stuff from libp2p, + // hopefully this can be converted back and forth with the golang + // crypto primitives. + // pid, _ := peer.IDFromString(fmt.Sprintf("%v", nodeid.D)) + // n, _ := swarm.NewNetwork(context.Background(), []ma.Multiaddr{}, pid, nil, nil) + // fmt.Println(n) + opts := []libp2p.Option{ + libp2p.ListenAddrStrings(fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", WhisperPort)), + libp2p.Identity(priv), + } + + h, err := libp2p.New(context.Background(), opts...) + if err != nil { + // TODO return error too + log.Error("Error setting up the libp2p network: %s", err) + return nil + } + + log.Info("Host address is at %s", h.Addrs()[0].String()) + whisper := &Whisper{ privateKeys: make(map[string]*ecdsa.PrivateKey), symKeys: make(map[string][]byte), @@ -103,29 +133,45 @@ func New(cfg *Config) *Whisper { p2pMsgQueue: make(chan *Envelope, messageQueueLimit), quit: make(chan struct{}), syncAllowance: DefaultSyncAllowance, + host: h, + protocol: WhisperProtocolString, } + h.SetStreamHandler(whisper.protocol, func(stream inet.Stream) { + defer stream.Close() + + data, err := ioutil.ReadAll(stream) + if err != nil { + log.Error("Error reading stream data %s", err) + return + } + if len(data) < 2 { + log.Error("Invalid data received: it has to be at least two bytes") + return + } + + messageType := data[0] + var envelope Envelope + err = rlp.DecodeBytes(data[1:], envelope) + if err != nil { + log.Error(fmt.Sprintf("Error decoding payload: %s", err)) + return + } + + p, err := whisper.getPeer(stream.Conn().RemotePeer()) + if err != nil { + log.Error(fmt.Sprintf("Could not find peer: %s", err)) + return + } + whisper.ProcessIncomingMessage(messageType, data[1:], p) + }) + whisper.filters = NewFilters(whisper) whisper.settings.Store(minPowIdx, cfg.MinimumAcceptedPOW) whisper.settings.Store(maxMsgSizeIdx, cfg.MaxMessageSize) whisper.settings.Store(overflowIdx, false) - // p2p whisper sub protocol handler - whisper.protocol = p2p.Protocol{ - Name: ProtocolName, - Version: uint(ProtocolVersion), - Length: NumberOfMessageCodes, - Run: whisper.HandlePeer, - NodeInfo: func() interface{} { - return map[string]interface{}{ - "version": ProtocolVersionStr, - "maxMessageSize": whisper.MaxMessageSize(), - "minimumPoW": whisper.MinPow(), - } - }, - } - return whisper } @@ -208,16 +254,6 @@ func (w *Whisper) RegisterServer(server MailServer) { w.mailServer = server } -// Protocols returns the whisper sub-protocols ran by this particular client. -func (w *Whisper) Protocols() []p2p.Protocol { - return []p2p.Protocol{w.protocol} -} - -// Version returns the whisper sub-protocols version number. -func (w *Whisper) Version() uint { - return w.protocol.Version -} - // SetMaxMessageSize sets the maximal message size allowed by this node func (w *Whisper) SetMaxMessageSize(size uint32) error { if size > MaxMessageSize { @@ -282,7 +318,7 @@ func (w *Whisper) notifyPeersAboutPowRequirementChange(pow float64) { err = p.notifyAboutPowRequirementChange(pow) } if err != nil { - log.Warn("failed to notify peer about new pow requirement", "peer", p.ID(), "error", err) + log.Warn("failed to notify peer about new pow requirement", "peer", p.pid, "error", err) } } } @@ -296,7 +332,7 @@ func (w *Whisper) notifyPeersAboutBloomFilterChange(bloom []byte) { err = p.notifyAboutBloomFilterChange(bloom) } if err != nil { - log.Warn("failed to notify peer about new bloom filter", "peer", p.ID(), "error", err) + log.Warn("failed to notify peer about new bloom filter", "peer", p.pid, "error", err) } } } @@ -314,12 +350,11 @@ func (w *Whisper) getPeers() []*Peer { } // getPeer retrieves peer by ID -func (w *Whisper) getPeer(peerID []byte) (*Peer, error) { +func (w *Whisper) getPeer(peerID peer.ID) (*Peer, error) { w.peerMu.Lock() defer w.peerMu.Unlock() for p := range w.peers { - id := p.peer.ID() - if bytes.Equal(peerID, id[:]) { + if peerID == p.pid { return p, nil } } @@ -328,7 +363,7 @@ func (w *Whisper) getPeer(peerID []byte) (*Peer, error) { // AllowP2PMessagesFromPeer marks specific peer trusted, // which will allow it to send historic (expired) messages. -func (w *Whisper) AllowP2PMessagesFromPeer(peerID []byte) error { +func (w *Whisper) AllowP2PMessagesFromPeer(peerID peer.ID) error { p, err := w.getPeer(peerID) if err != nil { return err @@ -342,7 +377,7 @@ func (w *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 (w *Whisper) RequestHistoricMessages(peerID []byte, envelope *Envelope) error { +func (w *Whisper) RequestHistoricMessages(peerID peer.ID, envelope *Envelope) error { p, err := w.getPeer(peerID) if err != nil { return err @@ -352,7 +387,7 @@ func (w *Whisper) RequestHistoricMessages(peerID []byte, envelope *Envelope) err } // SendP2PMessage sends a peer-to-peer message to a specific peer. -func (w *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error { +func (w *Whisper) SendP2PMessage(peerID peer.ID, envelope *Envelope) error { p, err := w.getPeer(peerID) if err != nil { return err @@ -362,7 +397,19 @@ func (w *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error { // SendP2PDirect sends a peer-to-peer message to a specific peer. func (w *Whisper) SendP2PDirect(peer *Peer, envelope *Envelope) error { - return p2p.Send(peer.ws, p2pMessageCode, envelope) + stream, err := w.host.NewStream(context.Background(), peer.pid, w.protocol) + if err != nil { + return err + } + defer stream.Close() + + b, err := rlp.EncodeToBytes(envelope) + if err != nil { + return err + } + + stream.Write(append([]byte{p2pMessageCode}, b...)) + return nil } // NewKeyPair generates a new cryptographic identity for the client, and injects @@ -621,18 +668,18 @@ func (w *Whisper) Stop() error { // HandlePeer is called by the underlying P2P layer when the whisper sub-protocol // connection is negotiated. -func (wh *Whisper) HandlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error { +func (w *Whisper) HandlePeer(peer *peer.ID, rw p2p.MsgReadWriter) error { // Create the new peer and start tracking it - whisperPeer := newPeer(wh, peer, rw) + whisperPeer := newPeer(w, peer, rw) - wh.peerMu.Lock() - wh.peers[whisperPeer] = struct{}{} - wh.peerMu.Unlock() + w.peerMu.Lock() + w.peers[whisperPeer] = struct{}{} + w.peerMu.Unlock() defer func() { - wh.peerMu.Lock() - delete(wh.peers, whisperPeer) - wh.peerMu.Unlock() + w.peerMu.Lock() + delete(w.peers, whisperPeer) + w.peerMu.Unlock() }() // Run the peer handshake and state updates @@ -642,184 +689,183 @@ func (wh *Whisper) HandlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error { whisperPeer.start() defer whisperPeer.stop() - return wh.runMessageLoop(whisperPeer, rw) + return w.runMessageLoop(whisperPeer, rw) +} + +// ProcessIncomingMessage gets the payload of a p2p message and decodes it +// depending on its type. +func (w *Whisper) ProcessIncomingMessage(code byte, data []byte, p *Peer) error { + switch code { + case statusCode: + // this should not happen, but no need to panic; just ignore this message. + log.Warn("unxepected status message received", "peer", p.pid) + case messagesCode: + // decode the contained envelopes + var envelopes []*Envelope + err := rlp.DecodeBytes(data, envelopes) + if err != nil { + log.Warn("failed to decode envelopes, peer will be disconnected", "peer", p.pid, "err", err) + return errors.New("invalid envelopes") + } + + trouble := false + for _, env := range envelopes { + cached, err := w.add(env) + if err != nil { + trouble = true + log.Error("bad envelope received, peer will be disconnected", "peer", p.pid, "err", err) + } + if cached { + p.mark(env) + } + } + + if trouble { + return errors.New("invalid envelope") + } + case powRequirementCode: + var i uint64 + err := rlp.DecodeBytes(data, i) + if err != nil { + log.Warn("failed to decode powRequirementCode message, peer will be disconnected", "peer", p.pid, "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.pid, "err", err) + return errors.New("invalid value in powRequirementCode message") + } + p.powRequirement = f + case bloomFilterExCode: + // to be implemented + case p2pMessageCode: + // peer-to-peer message, sent directly to peer bypassing PoW checks, etc. + // 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 { + var envelope Envelope + err := rlp.DecodeBytes(data, envelope) + if err != nil { + log.Warn("failed to decode direct message, peer will be disconnected", "peer", p.pid, "err", err) + return errors.New("invalid direct message") + } + w.postEvent(&envelope, true) + } + case p2pRequestCode: + // Must be processed if mail server is implemented. Otherwise ignore. + if w.mailServer != nil { + var request Envelope + err := rlp.DecodeBytes(data, request) + if err != nil { + log.Warn("failed to decode p2p request message, peer will be disconnected", "peer", p.pid, "err", err) + return errors.New("invalid p2p request") + } + w.mailServer.DeliverMail(p, &request) + } + default: + // New message types might be implemented in the future versions of Whisper. + // For forward compatibility, just ignore. + } + + return nil } // runMessageLoop reads and processes inbound messages directly to merge into client-global state. -func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { +func (w *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.pid, "err", err) return err } - if packet.Size > wh.MaxMessageSize() { - log.Warn("oversized message received", "peer", p.peer.ID()) + if packet.Size > w.MaxMessageSize() { + log.Warn("oversized message received", "peer", p.pid) 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()) - 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) - return errors.New("invalid envelopes") - } + data, err := ioutil.ReadAll(packet.Payload) + err = w.ProcessIncomingMessage(byte(packet.Code), data, p) - trouble := false - for _, env := range envelopes { - cached, err := wh.add(env) - if err != nil { - trouble = true - log.Error("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err) - } - if cached { - p.mark(env) - } - } - - if trouble { - return errors.New("invalid envelope") - } - case powRequirementCode: - 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) - 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) - return errors.New("invalid value in powRequirementCode message") - } - p.powRequirement = f - case bloomFilterExCode: - var bloom []byte - err := packet.Decode(&bloom) - if err == nil && len(bloom) != bloomFilterSize { - err = fmt.Errorf("wrong bloom filter size %d", len(bloom)) - } - - if err != nil { - 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 isFullNode(bloom) { - p.bloomFilter = nil - } else { - p.bloomFilter = bloom - } - case p2pMessageCode: - // peer-to-peer message, sent directly to peer bypassing PoW checks, etc. - // 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 { - 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) - return errors.New("invalid direct message") - } - wh.postEvent(&envelope, true) - } - case p2pRequestCode: - // Must be processed if mail server is implemented. Otherwise ignore. - if wh.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) - return errors.New("invalid p2p request") - } - wh.mailServer.DeliverMail(p, &request) - } - default: - // New message types might be implemented in the future versions of Whisper. - // For forward compatibility, just ignore. + if err != nil { + return err } - - packet.Discard() } } // add inserts a new envelope into the message pool to be distributed within the // whisper network. It also inserts the envelope into the expiration pool at the // appropriate time-stamp. In case of error, connection should be dropped. -func (wh *Whisper) add(envelope *Envelope) (bool, error) { +func (w *Whisper) add(envelope *Envelope) (bool, error) { now := uint32(time.Now().Unix()) sent := envelope.Expiry - envelope.TTL if sent > now { if sent-DefaultSyncAllowance > now { return false, fmt.Errorf("envelope created in the future [%x]", envelope.Hash()) - } else { - // recalculate PoW, adjusted for the time difference, plus one second for latency - envelope.calculatePoW(sent - now + 1) } + + // recalculate PoW, adjusted for the time difference, plus one second for latency + envelope.calculatePoW(sent - now + 1) } if envelope.Expiry < now { if envelope.Expiry+DefaultSyncAllowance*2 < now { return false, fmt.Errorf("very old message") - } else { - log.Debug("expired envelope dropped", "hash", envelope.Hash().Hex()) - return false, nil // drop envelope without error } + log.Debug("expired envelope dropped", "hash", envelope.Hash().Hex()) + return false, nil // drop envelope without error } - if uint32(envelope.size()) > wh.MaxMessageSize() { + if uint32(envelope.size()) > w.MaxMessageSize() { return false, fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash()) } - if envelope.PoW() < wh.MinPow() { + if envelope.PoW() < w.MinPow() { // maybe the value was recently changed, and the peers did not adjust yet. // in this case the previous value is retrieved by MinPowTolerance() // for a short period of peer synchronization. - if envelope.PoW() < wh.MinPowTolerance() { + if envelope.PoW() < w.MinPowTolerance() { return false, fmt.Errorf("envelope with low PoW received: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex()) } } - if !bloomFilterMatch(wh.BloomFilter(), envelope.Bloom()) { + if !bloomFilterMatch(w.BloomFilter(), envelope.Bloom()) { // maybe the value was recently changed, and the peers did not adjust yet. // in this case the previous value is retrieved by BloomFilterTolerance() // for a short period of peer synchronization. - if !bloomFilterMatch(wh.BloomFilterTolerance(), envelope.Bloom()) { + if !bloomFilterMatch(w.BloomFilterTolerance(), envelope.Bloom()) { 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) + envelope.Hash().Hex(), w.BloomFilter(), envelope.Bloom(), envelope.Topic) } } hash := envelope.Hash() - wh.poolMu.Lock() - _, alreadyCached := wh.envelopes[hash] + w.poolMu.Lock() + _, alreadyCached := w.envelopes[hash] if !alreadyCached { - wh.envelopes[hash] = envelope - if wh.expirations[envelope.Expiry] == nil { - wh.expirations[envelope.Expiry] = set.NewNonTS() + w.envelopes[hash] = envelope + if w.expirations[envelope.Expiry] == nil { + w.expirations[envelope.Expiry] = set.NewNonTS() } - if !wh.expirations[envelope.Expiry].Has(hash) { - wh.expirations[envelope.Expiry].Add(hash) + if !w.expirations[envelope.Expiry].Has(hash) { + w.expirations[envelope.Expiry].Add(hash) } } - wh.poolMu.Unlock() + w.poolMu.Unlock() if alreadyCached { log.Trace("whisper envelope already cached", "hash", envelope.Hash().Hex()) } else { log.Trace("cached whisper envelope", "hash", envelope.Hash().Hex()) - wh.statsMu.Lock() - wh.stats.memoryUsed += envelope.size() - wh.statsMu.Unlock() - wh.postEvent(envelope, false) // notify the local node about the new message - if wh.mailServer != nil { - wh.mailServer.Archive(envelope) + w.statsMu.Lock() + w.stats.memoryUsed += envelope.size() + w.statsMu.Unlock() + w.postEvent(envelope, false) // notify the local node about the new message + if w.mailServer != nil { + w.mailServer.Archive(envelope) } } return true, nil