whisper: final polishing

This commit is contained in:
Vlad 2016-12-07 19:42:37 +01:00
parent e47820d164
commit 3cf9b8dc83
7 changed files with 150 additions and 30 deletions

View file

@ -15,9 +15,7 @@
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
/* /*
Package whisper implements the Whisper PoC-1. Package whisper implements the Whisper protocol (version 5).
(https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec)
Whisper combines aspects of both DHTs and datagram messaging systems (e.g. UDP). Whisper combines aspects of both DHTs and datagram messaging systems (e.g. UDP).
As such it may be likened and compared to both, not dissimilar to the As such it may be likened and compared to both, not dissimilar to the
@ -57,11 +55,12 @@ const (
saltLength = 12 saltLength = 12
AESNonceMaxLength = 12 AESNonceMaxLength = 12
MaxMessageLength = 0xFFFF // todo: remove this restriction after testing in morden and analizing stats. this should be regulated by MinimumPoW. MaxMessageLength = 0xFFFF // todo: remove this restriction after testing. this should be regulated by PoW.
MinimumPoW = 10.0 // todo: review MinimumPoW = 10.0 // todo: review after testing.
padSizeLimitLower = 128 // it can not be less - we don't want to reveal the absence of signature padSizeLimitLower = 128 // it can not be less - we don't want to reveal the absence of signature
padSizeLimitUpper = 256 // just an arbitrary number, could be changed without losing compatibility padSizeLimitUpper = 256 // just an arbitrary number, could be changed without losing compatibility
messageQueueLimit = 1024
expirationCycle = time.Second expirationCycle = time.Second
transmissionCycle = 300 * time.Millisecond transmissionCycle = 300 * time.Millisecond

View file

@ -14,14 +14,14 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Contains the Whisper protocol Envelope element. For formal details please see // Contains the Whisper protocol Envelope element.
// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#envelopes.
package whisperv5 package whisperv5
import ( import (
"crypto/ecdsa" "crypto/ecdsa"
"encoding/binary" "encoding/binary"
"errors"
"fmt" "fmt"
"math" "math"
"time" "time"
@ -86,8 +86,8 @@ func (e *Envelope) Ver() uint64 {
// Seal closes the envelope by spending the requested amount of time as a proof // Seal closes the envelope by spending the requested amount of time as a proof
// of work on hashing the data. // of work on hashing the data.
func (e *Envelope) Seal(options *MessageParams) { func (e *Envelope) Seal(options *MessageParams) error {
var target int var target, bestBit int
if options.PoW == 0 { if options.PoW == 0 {
// adjust for the duration of Seal() execution only if execution time is predefined unconditionally // adjust for the duration of Seal() execution only if execution time is predefined unconditionally
e.Expiry += options.WorkTime e.Expiry += options.WorkTime
@ -99,7 +99,7 @@ func (e *Envelope) Seal(options *MessageParams) {
h := crypto.Keccak256(e.rlpWithoutNonce()) h := crypto.Keccak256(e.rlpWithoutNonce())
copy(buf[:32], h) copy(buf[:32], h)
finish, bestBit := time.Now().Add(time.Duration(options.WorkTime)*time.Second).UnixNano(), 0 finish := time.Now().Add(time.Duration(options.WorkTime) * time.Second).UnixNano()
for nonce := uint64(0); time.Now().UnixNano() < finish; { for nonce := uint64(0); time.Now().UnixNano() < finish; {
for i := 0; i < 1024; i++ { for i := 0; i < 1024; i++ {
binary.BigEndian.PutUint64(buf[56:], nonce) binary.BigEndian.PutUint64(buf[56:], nonce)
@ -108,12 +108,18 @@ func (e *Envelope) Seal(options *MessageParams) {
if firstBit > bestBit { if firstBit > bestBit {
e.EnvNonce, bestBit = nonce, firstBit e.EnvNonce, bestBit = nonce, firstBit
if target > 0 && bestBit >= target { if target > 0 && bestBit >= target {
return return nil
} }
} }
nonce++ nonce++
} }
} }
if target > 0 && bestBit < target {
return errors.New("Failed to reach the PoW target")
}
return nil
} }
func (e *Envelope) PoW() float64 { func (e *Envelope) PoW() float64 {

View file

@ -14,9 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Contains the Whisper protocol Message element. For formal details please see // Contains the Whisper protocol Message element.
// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#messages.
// todo: fix the spec link, and move it to doc.go
package whisperv5 package whisperv5
@ -256,7 +254,11 @@ func (msg *SentMessage) Wrap(options *MessageParams) (envelope *Envelope, err er
} }
envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, msg) envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, msg)
envelope.Seal(options) err = envelope.Seal(options)
if err != nil {
return nil, err
}
return envelope, nil return envelope, nil
} }

View file

@ -30,11 +30,15 @@ func copyFromBuf(dst []byte, src []byte, beg int) int {
} }
func generateMessageParams() (*MessageParams, error) { func generateMessageParams() (*MessageParams, error) {
// set all the parameters except p.Dst
buf := make([]byte, 1024) buf := make([]byte, 1024)
randomize(buf) randomize(buf)
sz := rand.Intn(400) sz := rand.Intn(400)
var p MessageParams var p MessageParams
p.PoW = 0.01
p.WorkTime = 1
p.TTL = uint32(rand.Intn(1024)) p.TTL = uint32(rand.Intn(1024))
p.Payload = make([]byte, sz) p.Payload = make([]byte, sz)
p.Padding = make([]byte, padSizeLimitUpper) p.Padding = make([]byte, padSizeLimitUpper)
@ -52,8 +56,6 @@ func generateMessageParams() (*MessageParams, error) {
return nil, err return nil, err
} }
// p.Dst, p.PoW, p.WorkTime are not set
p.PoW = 0.01
return &p, nil return &p, nil
} }
@ -152,6 +154,16 @@ func TestMessageWrap(t *testing.T) {
if pow < target { if pow < target {
t.Fatalf("failed Wrap with seed %d: pow < target (%f vs. %f).", seed, pow, target) t.Fatalf("failed Wrap with seed %d: pow < target (%f vs. %f).", seed, pow, target)
} }
// set PoW target too high, expect error
msg2 := NewSentMessage(params)
params.TTL = 1000000
params.WorkTime = 1
params.PoW = 10000000.0
env, err = msg2.Wrap(params)
if err == nil {
t.Fatalf("unexpectedly reached the PoW target with seed %d.", seed)
}
} }
func TestMessageSeal(t *testing.T) { func TestMessageSeal(t *testing.T) {
@ -274,3 +286,32 @@ func singleEnvelopeOpenTest(t *testing.T, symmetric bool) {
} }
} }
} }
func TestEncryptWithZeroKey(t *testing.T) {
InitSingleTest()
params, err := generateMessageParams()
if err != nil {
t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
}
msg := NewSentMessage(params)
params.KeySym = make([]byte, aesKeyLength)
_, err = msg.Wrap(params)
if err == nil {
t.Fatalf("wrapped with zero key, seed: %d.", seed)
}
params.KeySym = make([]byte, 0)
_, err = msg.Wrap(params)
if err == nil {
t.Fatalf("wrapped with empty key, seed: %d.", seed)
}
params.KeySym = nil
_, err = msg.Wrap(params)
if err == nil {
t.Fatalf("wrapped with nil key, seed: %d.", seed)
}
}

View file

@ -14,8 +14,7 @@
// You should have received a copy of the GNU Lesser General Public License // You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Contains the Whisper protocol Topic element. For formal details please see // Contains the Whisper protocol Topic element.
// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#topics.
package whisperv5 package whisperv5

View file

@ -51,7 +51,7 @@ type Whisper struct {
symKeys map[string][]byte symKeys map[string][]byte
keyMu sync.RWMutex keyMu sync.RWMutex
envelopes map[common.Hash]*Envelope // Pool of messages currently tracked by this node envelopes map[common.Hash]*Envelope // Pool of envelopes currently tracked by this node
messages map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages, which are not expired yet messages map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages, which are not expired yet
expirations map[uint32]*set.SetNonTS // Message expiration pool expirations map[uint32]*set.SetNonTS // Message expiration pool
poolMu sync.RWMutex // Mutex to sync the message and expiration pools poolMu sync.RWMutex // Mutex to sync the message and expiration pools
@ -64,7 +64,8 @@ type Whisper struct {
messageQueue chan QueueItem messageQueue chan QueueItem
quit chan struct{} quit chan struct{}
test bool overflow bool
test bool
} }
// New creates a Whisper client ready to communicate through the Ethereum P2P network. // New creates a Whisper client ready to communicate through the Ethereum P2P network.
@ -78,7 +79,7 @@ func NewWhisper(server MailServer) *Whisper {
expirations: make(map[uint32]*set.SetNonTS), expirations: make(map[uint32]*set.SetNonTS),
peers: make(map[*Peer]struct{}), peers: make(map[*Peer]struct{}),
mailServer: server, mailServer: server,
messageQueue: make(chan QueueItem, 1024), messageQueue: make(chan QueueItem, messageQueueLimit),
quit: make(chan struct{}), quit: make(chan struct{}),
} }
whisper.filters = NewFilters(whisper) whisper.filters = NewFilters(whisper)
@ -397,7 +398,7 @@ func (wh *Whisper) add(envelope *Envelope) error {
if sent > now { if sent > now {
if sent-SynchAllowance > now { if sent-SynchAllowance > now {
return fmt.Errorf("message created in the future") return fmt.Errorf("envelope created in the future [%x]", envelope.Hash())
} else { } else {
// recalculate PoW, adjusted for the time difference, plus one second for latency // recalculate PoW, adjusted for the time difference, plus one second for latency
envelope.calculatePoW(sent - now + 1) envelope.calculatePoW(sent - now + 1)
@ -408,30 +409,31 @@ func (wh *Whisper) add(envelope *Envelope) error {
if envelope.Expiry+SynchAllowance*2 < now { if envelope.Expiry+SynchAllowance*2 < now {
return fmt.Errorf("very old message") return fmt.Errorf("very old message")
} else { } else {
glog.V(logger.Debug).Infof("expired envelope dropped [%x]", envelope.Hash())
return nil // drop envelope without error return nil // drop envelope without error
} }
} }
if len(envelope.Data) > MaxMessageLength { if len(envelope.Data) > MaxMessageLength {
return fmt.Errorf("huge messages are not allowed") return fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash())
} }
if len(envelope.Version) > 4 { if len(envelope.Version) > 4 {
return fmt.Errorf("oversized Version") return fmt.Errorf("oversized version [%x]", envelope.Hash())
} }
if len(envelope.AESNonce) > AESNonceMaxLength { if len(envelope.AESNonce) > AESNonceMaxLength {
// the standard AES GSM nonce size is 12, // the standard AES GSM nonce size is 12,
// but const gcmStandardNonceSize cannot be accessed directly // but const gcmStandardNonceSize cannot be accessed directly
return fmt.Errorf("oversized AESNonce") return fmt.Errorf("oversized AESNonce [%x]", envelope.Hash())
} }
if len(envelope.Salt) > saltLength { if len(envelope.Salt) > saltLength {
return fmt.Errorf("oversized Salt") return fmt.Errorf("oversized salt [%x]", envelope.Hash())
} }
if envelope.PoW() < MinimumPoW && !wh.test { if envelope.PoW() < MinimumPoW && !wh.test {
glog.V(logger.Debug).Infof("envelope with low PoW dropped: %f", envelope.PoW()) glog.V(logger.Debug).Infof("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash())
return nil // drop envelope without error return nil // drop envelope without error
} }
@ -451,10 +453,10 @@ func (wh *Whisper) add(envelope *Envelope) error {
wh.poolMu.Unlock() wh.poolMu.Unlock()
if alreadyCached { if alreadyCached {
glog.V(logger.Detail).Infof("whisper envelope already cached: %x\n", envelope) glog.V(logger.Detail).Infof("whisper envelope already cached [%x]\n", envelope.Hash())
} else { } else {
glog.V(logger.Detail).Infof("cached whisper envelope [%x]: %v\n", envelope.Hash(), envelope)
wh.postEvent(envelope, messagesCode) // notify the local node about the new message wh.postEvent(envelope, messagesCode) // notify the local node about the new message
glog.V(logger.Detail).Infof("cached whisper envelope %v\n", envelope)
} }
return nil return nil
} }
@ -465,11 +467,28 @@ func (w *Whisper) postEvent(envelope *Envelope, messageCode uint64) {
// currently supported version, we can not decrypt it, // currently supported version, we can not decrypt it,
// and therefore just ignore this message // and therefore just ignore this message
if envelope.Ver() <= EnvelopeVersion { if envelope.Ver() <= EnvelopeVersion {
w.checkOverflow()
i := QueueItem{hash: envelope.Hash(), code: messageCode} i := QueueItem{hash: envelope.Hash(), code: messageCode}
w.messageQueue <- i w.messageQueue <- i
} }
} }
// checkOverflow checks if message queue overflow occurs and reports it if necessary.
func (w *Whisper) checkOverflow() {
queueSize := len(w.messageQueue)
if queueSize == messageQueueLimit {
if !w.overflow {
w.overflow = true
glog.V(logger.Warn).Infoln("message queue overflow")
}
} else if queueSize <= messageQueueLimit/2 {
if w.overflow {
w.overflow = false
}
}
}
// processQueue delivers the messages to the watchers during the lifetime of the whisper node. // processQueue delivers the messages to the watchers during the lifetime of the whisper node.
func (w *Whisper) processQueue() { func (w *Whisper) processQueue() {
for { for {

View file

@ -19,6 +19,7 @@ package whisperv5
import ( import (
"bytes" "bytes"
"testing" "testing"
"time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
@ -309,3 +310,56 @@ func TestWhisperSymKeyManagement(t *testing.T) {
t.Fatalf("failed to delete second key: second key is not nil.") t.Fatalf("failed to delete second key: second key is not nil.")
} }
} }
func TestExpiry(t *testing.T) {
InitSingleTest()
w := NewWhisper(nil)
w.test = true
w.Start(nil)
defer w.Stop()
params, err := generateMessageParams()
if err != nil {
t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err)
}
params.TTL = 1
msg := NewSentMessage(params)
env, err := msg.Wrap(params)
if err != nil {
t.Fatalf("failed Wrap with seed %d: %s.", seed, err)
}
err = w.Send(env)
if err != nil {
t.Fatalf("failed to send envelope with seed %d: %s.", seed, err)
}
// wait till received or timeout
var received, expired bool
for j := 0; j < 20; j++ {
time.Sleep(100 * time.Millisecond)
if len(w.Envelopes()) > 0 {
received = true
break
}
}
if !received {
t.Fatalf("did not receive the sent envelope, seed: %d.", seed)
}
// wait till expired or timeout
for j := 0; j < 20; j++ {
time.Sleep(100 * time.Millisecond)
if len(w.Envelopes()) == 0 {
expired = true
break
}
}
if !expired {
t.Fatalf("expire failed, seed: %d.", seed)
}
}