2021-06-16 20:19:45 +00:00
// Copyright 2019 The Waku Library Authors.
//
// The Waku library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The Waku library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty off
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the Waku library. If not, see <http://www.gnu.org/licenses/>.
//
// This software uses the go-ethereum library, which is licensed
// under the GNU Lesser General Public Library, version 3 or any later.
package wakuv2
import (
"context"
"crypto/ecdsa"
"crypto/sha256"
"errors"
"fmt"
"net"
"runtime"
"sync"
"time"
"github.com/ethereum/go-ethereum/common/hexutil"
"go.uber.org/zap"
mapset "github.com/deckarep/golang-set"
"golang.org/x/crypto/pbkdf2"
gethcommon "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc"
wakuprotocol "github.com/status-im/go-waku/waku/v2/protocol"
"github.com/status-im/go-waku/waku/v2/protocol/relay"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/wakuv2/common"
node "github.com/status-im/go-waku/waku/v2/node"
"github.com/status-im/go-waku/waku/v2/protocol/pb"
"github.com/status-im/go-waku/waku/v2/protocol/store"
wakurelay "github.com/status-im/go-wakurelay-pubsub"
)
const messageQueueLimit = 1024
type settings struct {
MaxMsgSize uint32 // Maximal message length allowed by the waku node
EnableConfirmations bool // Enable sending message confirmations
SoftBlacklistedPeerIDs map [ string ] bool // SoftBlacklistedPeerIDs is a list of peer ids that we want to keep connected but silently drop any envelope from
}
// Waku represents a dark communication interface through the Ethereum
// network, using its very own P2P communication layer.
type Waku struct {
node * node . WakuNode // reference to a libp2p waku node
filters * common . Filters // Message filters installed with Subscribe function
privateKeys map [ string ] * ecdsa . PrivateKey // Private key storage
symKeys map [ string ] [ ] byte // Symmetric key storage
keyMu sync . RWMutex // Mutex associated with key stores
envelopes map [ gethcommon . Hash ] * common . ReceivedMessage // Pool of envelopes currently tracked by this node
expirations map [ uint32 ] mapset . Set // Message expiration pool
poolMu sync . RWMutex // Mutex to sync the message and expiration pools
msgQueue chan * common . ReceivedMessage // Message queue for waku messages that havent been decoded
quit chan struct { } // Channel used for graceful exit
settings settings // Holds configuration settings that can be dynamically changed
settingsMu sync . RWMutex // Mutex to sync the settings access
envelopeFeed event . Feed
timeSource func ( ) time . Time // source of time for waku
logger * zap . Logger
}
// New creates a WakuV2 client ready to communicate through the LibP2P network.
func New ( nodeKey string , cfg * Config , logger * zap . Logger ) ( * Waku , error ) {
if logger == nil {
logger = zap . NewNop ( )
}
logger . Debug ( "starting wakuv2 with config" , zap . Any ( "config" , cfg ) )
if cfg == nil {
c := DefaultConfig
cfg = & c
}
waku := & Waku {
privateKeys : make ( map [ string ] * ecdsa . PrivateKey ) ,
symKeys : make ( map [ string ] [ ] byte ) ,
envelopes : make ( map [ gethcommon . Hash ] * common . ReceivedMessage ) ,
expirations : make ( map [ uint32 ] mapset . Set ) ,
msgQueue : make ( chan * common . ReceivedMessage , messageQueueLimit ) ,
quit : make ( chan struct { } ) ,
timeSource : time . Now ,
logger : logger ,
}
waku . settings = settings {
MaxMsgSize : cfg . MaxMessageSize ,
SoftBlacklistedPeerIDs : make ( map [ string ] bool ) ,
}
waku . filters = common . NewFilters ( )
var privateKey * ecdsa . PrivateKey
var err error
if nodeKey != "" {
privateKey , err = crypto . HexToECDSA ( nodeKey )
} else {
// If no nodekey is provided, create an ephemeral key
privateKey , err = crypto . GenerateKey ( )
}
if err != nil {
return nil , fmt . Errorf ( "failed to setup the go-waku private key: %v" , err )
}
hostAddr , err := net . ResolveTCPAddr ( "tcp" , fmt . Sprint ( cfg . Host , ":" , cfg . Port ) )
if err != nil {
return nil , fmt . Errorf ( "failed to setup the network interface: %v" , err )
}
waku . node , err = node . New ( context . Background ( ) ,
node . WithPrivateKey ( privateKey ) ,
node . WithHostAddress ( [ ] net . Addr { hostAddr } ) ,
node . WithWakuRelay ( wakurelay . WithMaxMessageSize ( int ( waku . settings . MaxMsgSize ) ) ) ,
node . WithWakuStore ( false ) , // Mounts the store protocol (without storing the messages)
)
if err != nil {
fmt . Println ( err )
return nil , fmt . Errorf ( "failed to start the go-waku node: %v" , err )
}
for _ , bootnode := range cfg . BootNodes {
err := waku . node . DialPeer ( bootnode )
if err != nil {
log . Warn ( "Could not dial peer" , err )
} else {
log . Info ( "Bootnode dialed successfully" , bootnode )
}
}
for _ , storenode := range cfg . StoreNodes {
peerID , err := waku . node . AddStorePeer ( storenode )
if err != nil {
log . Warn ( "Could not add store peer" , err )
} else {
log . Info ( "Storepeeer dialed successfully" , "peerId" , peerID . Pretty ( ) )
}
}
go waku . runMsgLoop ( )
log . Info ( "setup the go-waku node successfully" )
return waku , nil
}
func ( w * Waku ) runMsgLoop ( ) {
sub , err := w . node . Subscribe ( nil )
if err != nil {
fmt . Println ( "Could not subscribe:" , err )
return
}
for env := range sub . C {
envelopeErrors , err := w . OnNewEnvelopes ( env )
// TODO: should these be handled?
_ = envelopeErrors
_ = err
}
}
// MaxMessageSize returns the maximum accepted message size.
func ( w * Waku ) MaxMessageSize ( ) uint32 {
w . settingsMu . RLock ( )
defer w . settingsMu . RUnlock ( )
return w . settings . MaxMsgSize
}
// ConfirmationsEnabled returns true if message confirmations are enabled.
func ( w * Waku ) ConfirmationsEnabled ( ) bool {
w . settingsMu . RLock ( )
defer w . settingsMu . RUnlock ( )
return w . settings . EnableConfirmations
}
// CurrentTime returns current time.
func ( w * Waku ) CurrentTime ( ) time . Time {
return w . timeSource ( )
}
// SetTimeSource assigns a particular source of time to a waku object.
func ( w * Waku ) SetTimeSource ( timesource func ( ) time . Time ) {
w . t imeSource = timesource
}
// APIs returns the RPC descriptors the Waku implementation offers
func ( w * Waku ) APIs ( ) [ ] rpc . API {
return [ ] rpc . API {
{
Namespace : Name ,
Version : VersionStr ,
Service : NewPublicWakuAPI ( w ) ,
Public : false ,
} ,
}
}
// Protocols returns the waku sub-protocols ran by this particular client.
func ( w * Waku ) Protocols ( ) [ ] p2p . Protocol {
return [ ] p2p . Protocol { }
}
func ( w * Waku ) SendEnvelopeEvent ( event common . EnvelopeEvent ) int {
return w . envelopeFeed . Send ( event )
}
// SubscribeEnvelopeEvents subscribes to envelopes feed.
// In order to prevent blocking waku producers events must be amply buffered.
func ( w * Waku ) SubscribeEnvelopeEvents ( events chan <- common . EnvelopeEvent ) event . Subscription {
return w . envelopeFeed . Subscribe ( events )
}
// NewKeyPair generates a new cryptographic identity for the client, and injects
// it into the known identities for message decryption. Returns ID of the new key pair.
func ( w * Waku ) NewKeyPair ( ) ( string , error ) {
key , err := crypto . GenerateKey ( )
if err != nil || ! validatePrivateKey ( key ) {
key , err = crypto . GenerateKey ( ) // retry once
}
if err != nil {
return "" , err
}
if ! validatePrivateKey ( key ) {
return "" , fmt . Errorf ( "failed to generate valid key" )
}
id , err := toDeterministicID ( hexutil . Encode ( crypto . FromECDSAPub ( & key . PublicKey ) ) , common . KeyIDSize )
if err != nil {
return "" , err
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . privateKeys [ id ] != nil {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
w . privateKeys [ id ] = key
return id , nil
}
// DeleteKeyPair deletes the specified key if it exists.
func ( w * Waku ) DeleteKeyPair ( key string ) bool {
deterministicID , err := toDeterministicID ( key , common . KeyIDSize )
if err != nil {
return false
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . privateKeys [ deterministicID ] != nil {
delete ( w . privateKeys , deterministicID )
return true
}
return false
}
// AddKeyPair imports a asymmetric private key and returns it identifier.
func ( w * Waku ) AddKeyPair ( key * ecdsa . PrivateKey ) ( string , error ) {
id , err := makeDeterministicID ( hexutil . Encode ( crypto . FromECDSAPub ( & key . PublicKey ) ) , common . KeyIDSize )
if err != nil {
return "" , err
}
if w . HasKeyPair ( id ) {
return id , nil // no need to re-inject
}
w . keyMu . Lock ( )
w . privateKeys [ id ] = key
w . keyMu . Unlock ( )
return id , nil
}
// SelectKeyPair adds cryptographic identity, and makes sure
// that it is the only private key known to the node.
func ( w * Waku ) SelectKeyPair ( key * ecdsa . PrivateKey ) error {
id , err := makeDeterministicID ( hexutil . Encode ( crypto . FromECDSAPub ( & key . PublicKey ) ) , common . KeyIDSize )
if err != nil {
return err
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
w . privateKeys = make ( map [ string ] * ecdsa . PrivateKey ) // reset key store
w . privateKeys [ id ] = key
return nil
}
// DeleteKeyPairs removes all cryptographic identities known to the node
func ( w * Waku ) DeleteKeyPairs ( ) error {
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
w . privateKeys = make ( map [ string ] * ecdsa . PrivateKey )
return nil
}
// HasKeyPair checks if the waku node is configured with the private key
// of the specified public pair.
func ( w * Waku ) HasKeyPair ( id string ) bool {
deterministicID , err := toDeterministicID ( id , common . KeyIDSize )
if err != nil {
return false
}
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
return w . privateKeys [ deterministicID ] != nil
}
// GetPrivateKey retrieves the private key of the specified identity.
func ( w * Waku ) GetPrivateKey ( id string ) ( * ecdsa . PrivateKey , error ) {
deterministicID , err := toDeterministicID ( id , common . KeyIDSize )
if err != nil {
return nil , err
}
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
key := w . privateKeys [ deterministicID ]
if key == nil {
return nil , fmt . Errorf ( "invalid id" )
}
return key , nil
}
// GenerateSymKey generates a random symmetric key and stores it under id,
// which is then returned. Will be used in the future for session key exchange.
func ( w * Waku ) GenerateSymKey ( ) ( string , error ) {
key , err := common . GenerateSecureRandomData ( common . AESKeyLength )
if err != nil {
return "" , err
} else if ! common . ValidateDataIntegrity ( key , common . AESKeyLength ) {
return "" , fmt . Errorf ( "error in GenerateSymKey: crypto/rand failed to generate random data" )
}
id , err := common . GenerateRandomID ( )
if err != nil {
return "" , fmt . Errorf ( "failed to generate ID: %s" , err )
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ id ] != nil {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
w . symKeys [ id ] = key
return id , nil
}
// AddSymKey stores the key with a given id.
func ( w * Waku ) AddSymKey ( id string , key [ ] byte ) ( string , error ) {
deterministicID , err := toDeterministicID ( id , common . KeyIDSize )
if err != nil {
return "" , err
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ deterministicID ] != nil {
return "" , fmt . Errorf ( "key already exists: %v" , id )
}
w . symKeys [ deterministicID ] = key
return deterministicID , nil
}
// AddSymKeyDirect stores the key, and returns its id.
func ( w * Waku ) AddSymKeyDirect ( key [ ] byte ) ( string , error ) {
if len ( key ) != common . AESKeyLength {
return "" , fmt . Errorf ( "wrong key size: %d" , len ( key ) )
}
id , err := common . GenerateRandomID ( )
if err != nil {
return "" , fmt . Errorf ( "failed to generate ID: %s" , err )
}
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ id ] != nil {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
w . symKeys [ id ] = key
return id , nil
}
// AddSymKeyFromPassword generates the key from password, stores it, and returns its id.
func ( w * Waku ) AddSymKeyFromPassword ( password string ) ( string , error ) {
id , err := common . GenerateRandomID ( )
if err != nil {
return "" , fmt . Errorf ( "failed to generate ID: %s" , err )
}
if w . HasSymKey ( id ) {
return "" , fmt . Errorf ( "failed to generate unique ID" )
}
// kdf should run no less than 0.1 seconds on an average computer,
// because it's an once in a session experience
derived := pbkdf2 . Key ( [ ] byte ( password ) , nil , 65356 , common . AESKeyLength , sha256 . New )
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
// double check is necessary, because deriveKeyMaterial() is very slow
if w . symKeys [ id ] != nil {
return "" , fmt . Errorf ( "critical error: failed to generate unique ID" )
}
w . symKeys [ id ] = derived
return id , nil
}
// HasSymKey returns true if there is a key associated with the given id.
// Otherwise returns false.
func ( w * Waku ) HasSymKey ( id string ) bool {
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
return w . symKeys [ id ] != nil
}
// DeleteSymKey deletes the key associated with the name string if it exists.
func ( w * Waku ) DeleteSymKey ( id string ) bool {
w . keyMu . Lock ( )
defer w . keyMu . Unlock ( )
if w . symKeys [ id ] != nil {
delete ( w . symKeys , id )
return true
}
return false
}
// GetSymKey returns the symmetric key associated with the given id.
func ( w * Waku ) GetSymKey ( id string ) ( [ ] byte , error ) {
w . keyMu . RLock ( )
defer w . keyMu . RUnlock ( )
if w . symKeys [ id ] != nil {
return w . symKeys [ id ] , nil
}
return nil , fmt . Errorf ( "non-existent key ID" )
}
// Subscribe installs a new message handler used for filtering, decrypting
// and subsequent storing of incoming messages.
func ( w * Waku ) Subscribe ( f * common . Filter ) ( string , error ) {
s , err := w . filters . Install ( f )
if err != nil {
return s , err
}
return s , nil
}
// GetFilter returns the filter by id.
func ( w * Waku ) GetFilter ( id string ) * common . Filter {
return w . filters . Get ( id )
}
// Unsubscribe removes an installed message handler.
// TODO: This does not seem to update the bloom filter, but does update
// the topic interest map
func ( w * Waku ) Unsubscribe ( id string ) error {
ok := w . filters . Uninstall ( id )
if ! ok {
return fmt . Errorf ( "failed to unsubscribe: invalid ID '%s'" , id )
}
return nil
}
// Unsubscribe removes an installed message handler.
// TODO: This does not seem to update the bloom filter, but does update
// the topic interest map
func ( w * Waku ) UnsubscribeMany ( ids [ ] string ) error {
for _ , id := range ids {
w . logger . Debug ( "cleaning up filter" , zap . String ( "id" , id ) )
ok := w . filters . Uninstall ( id )
if ! ok {
w . logger . Warn ( "could not remove filter with id" , zap . String ( "id" , id ) )
}
}
return nil
}
// Send injects a message into the waku send queue, to be distributed in the
// network in the coming cycles.
func ( w * Waku ) Send ( msg * pb . WakuMessage ) ( [ ] byte , error ) {
return w . node . Publish ( context . Background ( ) , msg , nil )
}
2021-07-21 19:02:50 +00:00
func ( w * Waku ) Query ( topics [ ] types . TopicType , from uint64 , to uint64 , opts [ ] store . HistoryRequestOption ) ( cursor * pb . Index , err error ) {
2021-06-16 20:19:45 +00:00
strTopics := make ( [ ] string , len ( topics ) )
for i , t := range topics {
strTopics [ i ] = t . String ( )
}
result , err := w . node . Query ( context . Background ( ) , strTopics , float64 ( from ) , float64 ( to ) , opts ... )
for _ , msg := range result . Messages {
2021-07-21 19:02:50 +00:00
envelope := wakuprotocol . NewEnvelope ( msg , string ( relay . DefaultWakuTopic ) )
2021-06-16 20:19:45 +00:00
_ , err = w . OnNewEnvelopes ( envelope )
if err != nil {
2021-07-21 19:02:50 +00:00
return nil , err
2021-06-16 20:19:45 +00:00
}
}
2021-07-21 19:02:50 +00:00
if len ( result . Messages ) != 0 {
cursor = result . PagingInfo . Cursor
}
return
2021-06-16 20:19:45 +00:00
}
// Start implements node.Service, starting the background data propagation thread
// of the Waku protocol.
2021-06-30 11:40:54 +00:00
func ( w * Waku ) Start ( ) error {
2021-06-16 20:19:45 +00:00
numCPU := runtime . NumCPU ( )
for i := 0 ; i < numCPU ; i ++ {
go w . processQueue ( )
}
return nil
}
// Stop implements node.Service, stopping the background data propagation thread
// of the Waku protocol.
func ( w * Waku ) Stop ( ) error {
w . node . Stop ( )
close ( w . quit )
return nil
}
func ( w * Waku ) OnNewEnvelopes ( envelope * wakuprotocol . Envelope ) ( [ ] common . EnvelopeError , error ) {
recvMessage := common . NewReceivedMessage ( envelope )
envelopeErrors := make ( [ ] common . EnvelopeError , 0 )
w . logger . Debug ( "received new envelope" )
trouble := false
_ , err := w . add ( recvMessage )
if err != nil {
w . logger . Info ( "invalid envelope received" , zap . Error ( err ) )
}
common . EnvelopesValidatedCounter . Inc ( )
if trouble {
return envelopeErrors , errors . New ( "received invalid envelope" )
}
return envelopeErrors , nil
}
// addEnvelope adds an envelope to the envelope map, used for sending
func ( w * Waku ) addEnvelope ( envelope * common . ReceivedMessage ) {
hash := envelope . Hash ( )
w . poolMu . Lock ( )
w . envelopes [ hash ] = envelope
w . poolMu . Unlock ( )
}
func ( w * Waku ) add ( recvMessage * common . ReceivedMessage ) ( bool , error ) {
common . EnvelopesReceivedCounter . Inc ( )
hash := recvMessage . Hash ( )
w . poolMu . Lock ( )
_ , alreadyCached := w . envelopes [ hash ]
w . poolMu . Unlock ( )
if ! alreadyCached {
w . addEnvelope ( recvMessage )
}
if alreadyCached {
log . Trace ( "w envelope already cached" , "hash" , recvMessage . Hash ( ) . Hex ( ) )
common . EnvelopesCachedCounter . WithLabelValues ( "hit" ) . Inc ( )
} else {
log . Trace ( "cached w envelope" , "hash" , recvMessage . Hash ( ) . Hex ( ) )
common . EnvelopesCachedCounter . WithLabelValues ( "miss" ) . Inc ( )
common . EnvelopesSizeMeter . Observe ( float64 ( recvMessage . Envelope . Size ( ) ) )
w . postEvent ( recvMessage ) // notify the local node about the new message
}
return true , nil
}
// postEvent queues the message for further processing.
func ( w * Waku ) postEvent ( envelope * common . ReceivedMessage ) {
w . msgQueue <- envelope
}
// processQueue delivers the messages to the watchers during the lifetime of the waku node.
func ( w * Waku ) processQueue ( ) {
for {
select {
case <- w . quit :
return
case e := <- w . msgQueue :
w . filters . NotifyWatchers ( e )
w . envelopeFeed . Send ( common . EnvelopeEvent {
Topic : e . Topic ,
Hash : e . Hash ( ) ,
Event : common . EventEnvelopeAvailable ,
} )
}
}
}
// Envelopes retrieves all the messages currently pooled by the node.
func ( w * Waku ) Envelopes ( ) [ ] * common . ReceivedMessage {
w . poolMu . RLock ( )
defer w . poolMu . RUnlock ( )
all := make ( [ ] * common . ReceivedMessage , 0 , len ( w . envelopes ) )
for _ , envelope := range w . envelopes {
all = append ( all , envelope )
}
return all
}
// GetEnvelope retrieves an envelope from the message queue by its hash.
// It returns nil if the envelope can not be found.
func ( w * Waku ) GetEnvelope ( hash gethcommon . Hash ) * common . ReceivedMessage {
w . poolMu . RLock ( )
defer w . poolMu . RUnlock ( )
return w . envelopes [ hash ]
}
// isEnvelopeCached checks if envelope with specific hash has already been received and cached.
func ( w * Waku ) IsEnvelopeCached ( hash gethcommon . Hash ) bool {
w . poolMu . Lock ( )
defer w . poolMu . Unlock ( )
_ , exist := w . envelopes [ hash ]
return exist
}
// validatePrivateKey checks the format of the given private key.
func validatePrivateKey ( k * ecdsa . PrivateKey ) bool {
if k == nil || k . D == nil || k . D . Sign ( ) == 0 {
return false
}
return common . ValidatePublicKey ( & k . PublicKey )
}
// makeDeterministicID generates a deterministic ID, based on a given input
func makeDeterministicID ( input string , keyLen int ) ( id string , err error ) {
buf := pbkdf2 . Key ( [ ] byte ( input ) , nil , 4096 , keyLen , sha256 . New )
if ! common . ValidateDataIntegrity ( buf , common . KeyIDSize ) {
return "" , fmt . Errorf ( "error in GenerateDeterministicID: failed to generate key" )
}
id = gethcommon . Bytes2Hex ( buf )
return id , err
}
// toDeterministicID reviews incoming id, and transforms it to format
// expected internally be private key store. Originally, public keys
// were used as keys, now random keys are being used. And in order to
// make it easier to consume, we now allow both random IDs and public
// keys to be passed.
func toDeterministicID ( id string , expectedLen int ) ( string , error ) {
if len ( id ) != ( expectedLen * 2 ) { // we received hex key, so number of chars in id is doubled
var err error
id , err = makeDeterministicID ( id , expectedLen )
if err != nil {
return "" , err
}
}
return id , nil
}