whisper/whisperv6: implement pow/bloom exchange protocol (ethereum#15802

)

This is the main feature of v6.

whisper/whisperv6/api.go

@@ -24,15 +24,18 @@ import (
 	"sync"
 	"time"
 
-	"github.com/NiluPlatform/go-nilu/common"
-	"github.com/NiluPlatform/go-nilu/common/hexutil"
-	"github.com/NiluPlatform/go-nilu/crypto"
-	"github.com/NiluPlatform/go-nilu/log"
-	"github.com/NiluPlatform/go-nilu/p2p/discover"
-	"github.com/NiluPlatform/go-nilu/rpc"
+	"github.com/ethereum/go-ethereum/common"
+	"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"
+)
+
+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")
 	ErrInvalidSymmetricKey  = errors.New("invalid symmetric key")
@@ -57,9 +60,32 @@ func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI {
 		w:        w,
 		lastUsed: make(map[string]time.Time),
 	}
+
+	go api.run()
 	return api
 }
 
+// run the api event loop.
+// this loop deletes filter that have not been used within filterTimeout
+func (api *PublicWhisperAPI) run() {
+	timeout := time.NewTicker(2 * time.Minute)
+	for {
+		<-timeout.C
+
+		api.mu.Lock()
+		for id, lastUsed := range api.lastUsed {
+			if time.Since(lastUsed).Seconds() >= filterTimeout {
+				delete(api.lastUsed, id)
+				if err := api.w.Unsubscribe(id); err != nil {
+					log.Error("could not unsubscribe whisper filter", "error", err)
+				}
+				log.Debug("delete whisper filter (timeout)", "id", id)
+			}
+		}
+		api.mu.Unlock()
+	}
+}
+
 // Version returns the Whisper sub-protocol version.
 func (api *PublicWhisperAPI) Version(ctx context.Context) string {
 	return ProtocolVersionStr
@@ -90,7 +116,7 @@ func (api *PublicWhisperAPI) SetMaxMessageSize(ctx context.Context, size uint32)
 	return true, api.w.SetMaxMessageSize(size)
 }
 
-// SetMinPoW sets the minimum PoW, and notifies the peers.
+// SetMinPow sets the minimum PoW, and notifies the peers.
 func (api *PublicWhisperAPI) SetMinPoW(ctx context.Context, pow float64) (bool, error) {
 	return true, api.w.SetMinimumPoW(pow)
 }
@@ -148,7 +174,7 @@ func (api *PublicWhisperAPI) GetPublicKey(ctx context.Context, id string) (hexut
 	return crypto.FromECDSAPub(&key.PublicKey), nil
 }
 
-// GetPrivateKey returns the private key associated with the given key. The key is the hex
+// GetPublicKey returns the private key associated with the given key. The key is the hex
 // encoded representation of a key in the form specified in section 4.3.6 of ANSI X9.62.
 func (api *PublicWhisperAPI) GetPrivateKey(ctx context.Context, id string) (hexutil.Bytes, error) {
 	key, err := api.w.GetPrivateKey(id)
@@ -192,19 +218,6 @@ func (api *PublicWhisperAPI) DeleteSymKey(ctx context.Context, id string) bool {
 	return api.w.DeleteSymKey(id)
 }
 
-// MakeLightClient turns the node into light client, which does not forward
-// any incoming messages, and sends only messages originated in this node.
-func (api *PublicWhisperAPI) MakeLightClient(ctx context.Context) bool {
-	api.w.lightClient = true
-	return api.w.lightClient
-}
-
-// CancelLightClient cancels light client mode.
-func (api *PublicWhisperAPI) CancelLightClient(ctx context.Context) bool {
-	api.w.lightClient = false
-	return !api.w.lightClient
-}
-
 //go:generate gencodec -type NewMessage -field-override newMessageOverride -out gen_newmessage_json.go
 
 // NewMessage represents a new whisper message that is posted through the RPC.
@@ -227,9 +240,8 @@ type newMessageOverride struct {
 	Padding   hexutil.Bytes
 }
 
-// Post posts a message on the Whisper network.
-// returns the hash of the message in case of success.
-func (api *PublicWhisperAPI) Post(ctx context.Context, req NewMessage) (hexutil.Bytes, error) {
+// Post a message on the Whisper network.
+func (api *PublicWhisperAPI) Post(ctx context.Context, req NewMessage) (bool, error) {
 	var (
 		symKeyGiven = len(req.SymKeyID) > 0
 		pubKeyGiven = len(req.PublicKey) > 0
@@ -238,7 +250,7 @@ func (api *PublicWhisperAPI) Post(ctx context.Context, req NewMessage) (hexutil.
 
 	// user must specify either a symmetric or an asymmetric key
 	if (symKeyGiven && pubKeyGiven) || (!symKeyGiven && !pubKeyGiven) {
-		return nil, ErrSymAsym
+		return false, ErrSymAsym
 	}
 
 	params := &MessageParams{
@@ -253,68 +265,57 @@ func (api *PublicWhisperAPI) Post(ctx context.Context, req NewMessage) (hexutil.
 	// Set key that is used to sign the message
 	if len(req.Sig) > 0 {
 		if params.Src, err = api.w.GetPrivateKey(req.Sig); err != nil {
-			return nil, err
+			return false, err
 		}
 	}
 
 	// Set symmetric key that is used to encrypt the message
 	if symKeyGiven {
 		if params.Topic == (TopicType{}) { // topics are mandatory with symmetric encryption
-			return nil, ErrNoTopics
+			return false, ErrNoTopics
 		}
 		if params.KeySym, err = api.w.GetSymKey(req.SymKeyID); err != nil {
-			return nil, err
+			return false, err
 		}
-		if !validateDataIntegrity(params.KeySym, aesKeyLength) {
-			return nil, ErrInvalidSymmetricKey
+		if !validateSymmetricKey(params.KeySym) {
+			return false, ErrInvalidSymmetricKey
 		}
 	}
 
 	// Set asymmetric key that is used to encrypt the message
 	if pubKeyGiven {
 		params.Dst = crypto.ToECDSAPub(req.PublicKey)
 		if !ValidatePublicKey(params.Dst) {
-			return nil, ErrInvalidPublicKey
+			return false, ErrInvalidPublicKey
 		}
 	}
 
 	// encrypt and sent message
 	whisperMsg, err := NewSentMessage(params)
 	if err != nil {
-		return nil, err
+		return false, err
 	}
 
-	var result []byte
 	env, err := whisperMsg.Wrap(params)
 	if err != nil {
-		return nil, err
+		return false, err
 	}
 
 	// send to specific node (skip PoW check)
 	if len(req.TargetPeer) > 0 {
 		n, err := discover.ParseNode(req.TargetPeer)
 		if err != nil {
-			return nil, fmt.Errorf("failed to parse target peer: %s", err)
-		}
-		err = api.w.SendP2PMessage(n.ID[:], env)
-		if err == nil {
-			hash := env.Hash()
-			result = hash[:]
+			return false, fmt.Errorf("failed to parse target peer: %s", err)
 		}
-		return result, err
+		return true, api.w.SendP2PMessage(n.ID[:], env)
 	}
 
 	// ensure that the message PoW meets the node's minimum accepted PoW
 	if req.PowTarget < api.w.MinPow() {
-		return nil, ErrTooLowPoW
+		return false, ErrTooLowPoW
 	}
 
-	err = api.w.Send(env)
-	if err == nil {
-		hash := env.Hash()
-		result = hash[:]
-	}
-	return result, err
+	return true, api.w.Send(env)
 }
 
 //go:generate gencodec -type Criteria -field-override criteriaOverride -out gen_criteria_json.go
@@ -382,7 +383,7 @@ func (api *PublicWhisperAPI) Messages(ctx context.Context, crit Criteria) (*rpc.
 		if err != nil {
 			return nil, err
 		}
-		if !validateDataIntegrity(key, aesKeyLength) {
+		if !validateSymmetricKey(key) {
 			return nil, ErrInvalidSymmetricKey
 		}
 		filter.KeySym = key
@@ -554,7 +555,7 @@ func (api *PublicWhisperAPI) NewMessageFilter(req Criteria) (string, error) {
 		if keySym, err = api.w.GetSymKey(req.SymKeyID); err != nil {
 			return "", err
 		}
-		if !validateDataIntegrity(keySym, aesKeyLength) {
+		if !validateSymmetricKey(keySym) {
 			return "", ErrInvalidSymmetricKey
 		}
 	}
@@ -566,10 +567,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, 1)
+		for _, topic := range req.Topics {
+			topics = append(topics, topic[:])
 		}
 	}
 

whisper/whisperv6/doc.go

@@ -35,7 +35,6 @@ import (
 )
 
 const (
-	EnvelopeVersion    = uint64(0)
 	ProtocolVersion    = uint64(6)
 	ProtocolVersionStr = "6.0"
 	ProtocolName       = "shh"
@@ -52,11 +51,14 @@ const (
 	paddingMask   = byte(3)
 	signatureFlag = byte(4)
 
-	TopicLength     = 4
-	signatureLength = 65
-	aesKeyLength    = 32
-	AESNonceLength  = 12
-	keyIdSize       = 32
+	TopicLength     = 4  // in bytes
+	signatureLength = 65 // in bytes
+	aesKeyLength    = 32 // in bytes
+	AESNonceLength  = 12 // in bytes
+	keyIdSize       = 32 // in bytes
+	bloomFilterSize = 64 // in bytes
+
+	EnvelopeHeaderLength = 20
 
 	MaxMessageSize        = uint32(10 * 1024 * 1024) // maximum accepted size of a message.
 	DefaultMaxMessageSize = uint32(1024 * 1024)
@@ -68,10 +70,8 @@ const (
 	expirationCycle   = time.Second
 	transmissionCycle = 300 * time.Millisecond
 
-	DefaultTTL     = 50 // seconds
-	SynchAllowance = 10 // seconds
-
-	EnvelopeHeaderLength = 20
+	DefaultTTL           = 50 // seconds
+	DefaultSyncAllowance = 10 // seconds
 )
 
 type unknownVersionError uint64

whisper/whisperv6/envelope.go

@@ -42,9 +42,11 @@ type Envelope struct {
 	Data   []byte
 	Nonce  uint64
 
-	pow  float64     // Message-specific PoW as described in the Whisper specification.
-	hash common.Hash // Cached hash of the envelope to avoid rehashing every time.
-	// Don't access hash directly, use Hash() function instead.
+	pow float64 // Message-specific PoW as described in the Whisper specification.
+
+	// the following variables should not be accessed directly, use the corresponding function instead: Hash(), Bloom()
+	hash  common.Hash // Cached hash of the envelope to avoid rehashing every time.
+	bloom []byte
 }
 
 // size returns the size of envelope as it is sent (i.e. public fields only)
@@ -227,3 +229,30 @@ func (e *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) {
 	}
 	return msg
 }
+
+// Bloom maps 4-bytes Topic into 64-byte bloom filter with 3 bits set (at most).
+func (e *Envelope) Bloom() []byte {
+	if e.bloom == nil {
+		e.bloom = TopicToBloom(e.Topic)
+	}
+	return e.bloom
+}
+
+// TopicToBloom converts the topic (4 bytes) to the bloom filter (64 bytes)
+func TopicToBloom(topic TopicType) []byte {
+	b := make([]byte, bloomFilterSize)
+	var index [3]int
+	for j := 0; j < 3; j++ {
+		index[j] = int(topic[j])
+		if (topic[3] & (1 << uint(j))) != 0 {
+			index[j] += 256
+		}
+	}
+
+	for j := 0; j < 3; j++ {
+		byteIndex := index[j] / 8
+		bitIndex := index[j] % 8
+		b[byteIndex] = (1 << uint(bitIndex))
+	}
+	return b
+}