forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-25 14:16:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

whisper: changed according to the request of @bas-vk

Browse source
This commit is contained in:
Vlad 2016-10-04 13:35:54 +02:00
parent a7351ae20c
commit 54925b8374
7 changed files with 372 additions and 365 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

111
whisper/shhapi/api.go
View file

@ -56,110 +56,115 @@ func APIs() []rpc.API {
}
// Version returns the Whisper version this node offers.
func (self *PublicWhisperAPI) Version() (*rpc.HexNumber, error) {
if self.whisper == nil {
func (api *PublicWhisperAPI) Version() (*rpc.HexNumber, error) {
if api.whisper == nil {
return rpc.NewHexNumber(0), whisperOffLineErr
}
return rpc.NewHexNumber(self.whisper.Version()), nil
return rpc.NewHexNumber(api.whisper.Version()), nil
}
// MarkPeerTrusted marks specific peer trusted, which will allow it
// to send historic (expired) messages.
func (self *PublicWhisperAPI) MarkPeerTrusted(peerID *rpc.HexBytes) error {
if self.whisper == nil {
func (api *PublicWhisperAPI) MarkPeerTrusted(peerID rpc.HexBytes) error {
if api.whisper == nil {
return whisperOffLineErr
}
return self.whisper.MarkPeerTrusted(*peerID)
return api.whisper.MarkPeerTrusted(peerID)
}
// RequestHistoricMessages requests the peer to deliver the old (expired) messages.
// data contains parameters (time frame, payment details, etc.), required
// by the remote email-like server. Whisper is not aware about the data format,
// it will just forward the raw data to the server.
func (self *PublicWhisperAPI) RequestHistoricMessages(peerID *rpc.HexBytes, data *rpc.HexBytes) error {
if self.whisper == nil {
func (api *PublicWhisperAPI) RequestHistoricMessages(peerID rpc.HexBytes, data rpc.HexBytes) error {
if api.whisper == nil {
return whisperOffLineErr
}
return self.whisper.RequestHistoricMessages(*peerID, *data)
return api.whisper.RequestHistoricMessages(peerID, data)
}
// HasIdentity checks if the the whisper node is configured with the private key
// HasIdentity checks if the whisper node is configured with the private key
// of the specified public pair.
func (self *PublicWhisperAPI) HasIdentity(identity string) (bool, error) {
if self.whisper == nil {
func (api *PublicWhisperAPI) HasIdentity(identity string) (bool, error) {
if api.whisper == nil {
return false, whisperOffLineErr
}
return self.whisper.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil
return api.whisper.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil
}
// DeleteIdentity deletes the specifies key if it exists.
func (self *PublicWhisperAPI) DeleteIdentity(identity string) error {
if self.whisper == nil {
func (api *PublicWhisperAPI) DeleteIdentity(identity string) error {
if api.whisper == nil {
return whisperOffLineErr
}
self.whisper.DeleteIdentity(identity)
api.whisper.DeleteIdentity(identity)
return nil
}
// NewIdentity generates a new cryptographic identity for the client, and injects
// it into the known identities for message decryption.
func (self *PublicWhisperAPI) NewIdentity() (string, error) {
if self.whisper == nil {
func (api *PublicWhisperAPI) NewIdentity() (string, error) {
if api.whisper == nil {
return "", whisperOffLineErr
}
identity := self.whisper.NewIdentity()
identity := api.whisper.NewIdentity()
return common.ToHex(crypto.FromECDSAPub(&identity.PublicKey)), nil
}
// GenerateTopicKey generates a random key and stores it under the 'name' id.
// Will be used in the future for session key exchange.
func (self *PublicWhisperAPI) GenerateTopicKey(name string) error {
if self.whisper == nil {
// GenerateTopicKey generates a random symmetric key and stores it under
// the 'name' id. Will be used in the future for session key exchange.
func (api *PublicWhisperAPI) GenerateTopicKey(name string) error {
if api.whisper == nil {
return whisperOffLineErr
}
return self.whisper.GenerateTopicKey(name)
return api.whisper.GenerateTopicKey(name)
}
func (self *PublicWhisperAPI) AddTopicKey(name string, key []byte) error {
if self.whisper == nil {
// AddTopicKey stores the key under the 'name' id.
func (api *PublicWhisperAPI) AddTopicKey(name string, key []byte) error {
if api.whisper == nil {
return whisperOffLineErr
}
return self.whisper.AddTopicKey(name, key)
return api.whisper.AddTopicKey(name, key)
}
func (self *PublicWhisperAPI) HasTopicKey(name string) (bool, error) {
if self.whisper == nil {
// HasTopicKey returns true if there is a key associated with the name string.
// Otherwise returns false.
func (api *PublicWhisperAPI) HasTopicKey(name string) (bool, error) {
if api.whisper == nil {
return false, whisperOffLineErr
}
res := self.whisper.HasTopicKey(name)
res := api.whisper.HasTopicKey(name)
return res, nil
}
func (self *PublicWhisperAPI) DeleteTopicKey(name string) error {
if self.whisper == nil {
// DeleteTopicKey deletes the key associated with the name string if it exists.
func (api *PublicWhisperAPI) DeleteTopicKey(name string) error {
if api.whisper == nil {
return whisperOffLineErr
}
self.whisper.DeleteTopicKey(name)
api.whisper.DeleteTopicKey(name)
return nil
}
// NewWhisperFilter creates and registers a new message filter to watch for inbound whisper messages.
func (self *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (*rpc.HexNumber, error) {
if self.whisper == nil {
// Returns the ID of the newly created Filter.
func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (*rpc.HexNumber, error) {
if api.whisper == nil {
return nil, whisperOffLineErr
}
filter := whisperv5.Filter{
Src: crypto.ToECDSAPub(args.From),
Dst: crypto.ToECDSAPub(args.To),
KeySym: self.whisper.GetTopicKey(args.KeyName),
KeySym: api.whisper.GetTopicKey(args.KeyName),
PoW: args.PoW,
Messages: make(map[common.Hash]*whisperv5.ReceivedMessage),
AcceptP2P: args.AcceptP2P,
}
if len(filter.KeySym) > 0 {
filter.TopicKeyHash = crypto.Keccak256Hash(filter.KeySym)
filter.SymKeyHash = crypto.Keccak256Hash(filter.KeySym)
}
for _, t := range args.Topics {
@ -196,7 +201,7 @@ func (self *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (*rpc.HexNumber,
glog.V(logger.Error).Infof(info)
return nil, errors.New(info)
}
filter.KeyAsym = self.whisper.GetIdentity(filter.Dst)
filter.KeyAsym = api.whisper.GetIdentity(filter.Dst)
if filter.KeyAsym == nil {
info := "NewFilter: non-existent identity provided"
glog.V(logger.Error).Infof(info)
@ -212,18 +217,18 @@ func (self *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (*rpc.HexNumber,
}
}
id := self.whisper.Watch(&filter)
id := api.whisper.Watch(&filter)
return rpc.NewHexNumber(id), nil
}
// UninstallFilter disables and removes an existing filter.
func (self *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) {
self.whisper.Unwatch(filterId.Int())
func (api *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) {
api.whisper.Unwatch(filterId.Int())
}
// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval.
func (self *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage {
f := self.whisper.GetFilter(filterId.Int())
func (api *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage {
f := api.whisper.GetFilter(filterId.Int())
if f != nil {
newMail := f.Retrieve()
return toWhisperMessages(newMail)
@ -232,8 +237,8 @@ func (self *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []Whisper
}
// GetMessages retrieves all the known messages that match a specific filter.
func (self *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage {
all := self.whisper.Messages(filterId.Int())
func (api *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage {
all := api.whisper.Messages(filterId.Int())
return toWhisperMessages(all)
}
@ -246,16 +251,16 @@ func toWhisperMessages(messages []*whisperv5.ReceivedMessage) []WhisperMessage {
return msgs
}
// Post injects a message into the whisper network for distribution.
func (self *PublicWhisperAPI) Post(args PostArgs) error {
if self.whisper == nil {
// Post creates a whisper message and injects it into the network for distribution.
func (api *PublicWhisperAPI) Post(args PostArgs) error {
if api.whisper == nil {
return whisperOffLineErr
}
params := whisperv5.MessageParams{
TTL: args.TTL,
Dst: crypto.ToECDSAPub(args.To),
KeySym: self.whisper.GetTopicKey(args.KeyName),
KeySym: api.whisper.GetTopicKey(args.KeyName),
Topic: args.Topic,
Payload: args.Payload,
Padding: args.Padding,
@ -270,7 +275,7 @@ func (self *PublicWhisperAPI) Post(args PostArgs) error {
glog.V(logger.Error).Infof(info)
return errors.New(info)
}
params.Src = self.whisper.GetIdentity(pub)
params.Src = api.whisper.GetIdentity(pub)
if params.Src == nil {
info := "Post: non-existent identity provided"
glog.V(logger.Error).Infof(info)
@ -278,7 +283,7 @@ func (self *PublicWhisperAPI) Post(args PostArgs) error {
}
}
filter := self.whisper.GetFilter(args.FilterID)
filter := api.whisper.GetFilter(args.FilterID)
if filter == nil && args.FilterID > -1 {
info := fmt.Sprintf("Post: wrong filter id %d", args.FilterID)
glog.V(logger.Error).Infof(info)
@ -358,10 +363,10 @@ func (self *PublicWhisperAPI) Post(args PostArgs) error {
}
if args.PeerID != nil {
return self.whisper.SendP2PMessage(args.PeerID, envelope)
return api.whisper.SendP2PMessage(args.PeerID, envelope)
}
return self.whisper.Send(envelope)
return api.whisper.Send(envelope)
}
type PostArgs struct {

104
whisper/whisperv5/envelope.go
View file

@ -35,12 +35,12 @@ import (
// Envelope represents a clear-text data packet to transmit through the Whisper
// network. Its contents may or may not be encrypted and signed.
type Envelope struct {
Version []byte
Expiry uint32
TTL uint32
Topic TopicType
Salt []byte
AESNonce []byte
Version []byte
Data []byte
EnvNonce uint64
@ -52,13 +52,13 @@ type Envelope struct {
// included into an envelope for network forwarding.
func NewEnvelope(ttl uint32, topic TopicType, salt []byte, aesNonce []byte, msg *SentMessage) *Envelope {
env := Envelope{
Version: make([]byte, 1),
Expiry: uint32(time.Now().Add(time.Second * time.Duration(ttl)).Unix()),
TTL: ttl,
Topic: topic,
Salt: salt,
AESNonce: aesNonce,
Data: msg.Raw,
Version: make([]byte, 1),
EnvNonce: 0,
}
@ -71,31 +71,31 @@ func NewEnvelope(ttl uint32, topic TopicType, salt []byte, aesNonce []byte, msg
return &env
}
func (self *Envelope) IsSymmetric() bool {
return self.AESNonce != nil
func (e *Envelope) IsSymmetric() bool {
return e.AESNonce != nil
}
func (self *Envelope) isAsymmetric() bool {
return !self.IsSymmetric()
func (e *Envelope) isAsymmetric() bool {
return !e.IsSymmetric()
}
func (self *Envelope) Ver() uint64 {
return bytesToIntLittleEndian(self.Version)
func (e *Envelope) Ver() uint64 {
return bytesToIntLittleEndian(e.Version)
}
// Seal closes the envelope by spending the requested amount of time as a proof
// of work on hashing the data.
func (self *Envelope) Seal(options MessageParams) {
func (e *Envelope) Seal(options MessageParams) {
var target int
if options.PoW == 0 {
// adjust for the duration of Seal() execution only if execution time is predefined unconditionally
self.Expiry += options.WorkTime
e.Expiry += options.WorkTime
} else {
target = self.powToFirstBit(options.PoW)
target = e.powToFirstBit(options.PoW)
}
buf := make([]byte, 64)
h := crypto.Keccak256(self.rlpWithoutNonce())
h := crypto.Keccak256(e.rlpWithoutNonce())
copy(buf[:32], h)
finish, bestBit := time.Now().Add(time.Duration(options.WorkTime)*time.Second).UnixNano(), 0
@ -105,7 +105,7 @@ func (self *Envelope) Seal(options MessageParams) {
h = crypto.Keccak256(buf)
firstBit := common.FirstBitSet(common.BigD(h))
if firstBit > bestBit {
self.EnvNonce, bestBit = nonce, firstBit
e.EnvNonce, bestBit = nonce, firstBit
if target > 0 && bestBit >= target {
return
}
@ -115,48 +115,48 @@ func (self *Envelope) Seal(options MessageParams) {
}
}
func (self *Envelope) PoW() float64 {
if self.pow == 0 {
self.calculatePoW(0)
func (e *Envelope) PoW() float64 {
if e.pow == 0 {
e.calculatePoW(0)
}
return self.pow
return e.pow
}
func (self *Envelope) calculatePoW(diff uint32) {
h := self.Hash()
func (e *Envelope) calculatePoW(diff uint32) {
h := e.Hash()
firstBit := common.FirstBitSet(common.BigD(h.Bytes()))
x := math.Pow(2, float64(firstBit))
x /= float64(len(self.Data))
x /= float64(self.TTL + diff)
self.pow = x
x /= float64(len(e.Data))
x /= float64(e.TTL + diff)
e.pow = x
}
func (self *Envelope) powToFirstBit(pow float64) int {
func (e *Envelope) powToFirstBit(pow float64) int {
x := pow
x *= float64(len(self.Data))
x *= float64(self.TTL)
x *= float64(len(e.Data))
x *= float64(e.TTL)
bits := math.Log2(x)
bits = math.Ceil(bits)
return int(bits)
}
// rlpWithoutNonce returns the RLP encoded envelope contents, except the nonce.
func (self *Envelope) rlpWithoutNonce() []byte {
enc, _ := rlp.EncodeToBytes([]interface{}{self.Expiry, self.TTL, self.Topic, self.Salt, self.AESNonce, self.Data})
return enc
func (e *Envelope) rlpWithoutNonce() []byte {
res, _ := rlp.EncodeToBytes([]interface{}{e.Expiry, e.TTL, e.Topic, e.Salt, e.AESNonce, e.Data})
return res
}
// Hash returns the SHA3 hash of the envelope, calculating it if not yet done.
func (self *Envelope) Hash() common.Hash {
if (self.hash == common.Hash{}) {
enc, _ := rlp.EncodeToBytes(self)
self.hash = crypto.Keccak256Hash(enc)
func (e *Envelope) Hash() common.Hash {
if (e.hash == common.Hash{}) {
encoded, _ := rlp.EncodeToBytes(e)
e.hash = crypto.Keccak256Hash(encoded)
}
return self.hash
return e.hash
}
// DecodeRLP decodes an Envelope from an RLP data stream.
func (self *Envelope) DecodeRLP(s *rlp.Stream) error {
func (e *Envelope) DecodeRLP(s *rlp.Stream) error {
raw, err := s.Raw()
if err != nil {
return err
@ -167,16 +167,16 @@ func (self *Envelope) DecodeRLP(s *rlp.Stream) error {
// rlp.Decoder (does not implement DecodeRLP function).
// Only public members will be encoded.
type rlpenv Envelope
if err := rlp.DecodeBytes(raw, (*rlpenv)(self)); err != nil {
if err := rlp.DecodeBytes(raw, (*rlpenv)(e)); err != nil {
return err
}
self.hash = crypto.Keccak256Hash(raw)
e.hash = crypto.Keccak256Hash(raw)
return nil
}
// OpenAsymmetric tries to decrypt an envelope, potentially encrypted with a particular key.
func (self *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*ReceivedMessage, error) {
message := &ReceivedMessage{Raw: self.Data}
func (e *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*ReceivedMessage, error) {
message := &ReceivedMessage{Raw: e.Data}
err := message.decryptAsymmetric(key)
switch err {
case nil:
@ -189,9 +189,9 @@ func (self *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*ReceivedMessage, e
}
// OpenSymmetric tries to decrypt an envelope, potentially encrypted with a particular key.
func (self *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error) {
msg = &ReceivedMessage{Raw: self.Data}
err = msg.decryptSymmetric(key, self.Salt, self.AESNonce)
func (e *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error) {
msg = &ReceivedMessage{Raw: e.Data}
err = msg.decryptSymmetric(key, e.Salt, e.AESNonce)
if err != nil {
msg = nil
}
@ -199,14 +199,14 @@ func (self *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error
}
// Open tries to decrypt an envelope, and populates the message fields in case of success.
func (self *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) {
if self.isAsymmetric() {
msg, _ = self.OpenAsymmetric(watcher.KeyAsym)
func (e *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) {
if e.isAsymmetric() {
msg, _ = e.OpenAsymmetric(watcher.KeyAsym)
if msg != nil {
msg.Dst = watcher.Dst
}
} else if self.IsSymmetric() {
msg, _ = self.OpenSymmetric(watcher.KeySym)
} else if e.IsSymmetric() {
msg, _ = e.OpenSymmetric(watcher.KeySym)
if msg != nil {
msg.TopicKeyHash = crypto.Keccak256Hash(watcher.KeySym)
}
@ -217,12 +217,12 @@ func (self *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) {
if !ok {
return nil
}
msg.Topic = self.Topic
msg.PoW = self.PoW()
msg.TTL = self.TTL
msg.Sent = self.Expiry - self.TTL
msg.EnvelopeHash = self.hash
msg.EnvelopeVersion = self.Ver()
msg.Topic = e.Topic
msg.PoW = e.PoW()
msg.TTL = e.TTL
msg.Sent = e.Expiry - e.TTL
msg.EnvelopeHash = e.hash
msg.EnvelopeVersion = e.Ver()
}
return msg
}

117
whisper/whisperv5/filter.go
View file

@ -24,16 +24,17 @@ import (
)
type Filter struct {
Src *ecdsa.PublicKey // Sender of the message
Dst *ecdsa.PublicKey // Recipient of the message
KeyAsym *ecdsa.PrivateKey // Private Key of recipient
Topics []TopicType // Topics to filter messages with
KeySym []byte // Key associated with the Topic
TopicKeyHash common.Hash // The Keccak256Hash of the symmetric key
PoW float64 // Proof of work as described in the Whisper spec
AcceptP2P bool // Indicates whether this filter is interested in direct peer-to-peer messages
Messages map[common.Hash]*ReceivedMessage
Mutex sync.RWMutex
Src *ecdsa.PublicKey // Sender of the message
Dst *ecdsa.PublicKey // Recipient of the message
KeyAsym *ecdsa.PrivateKey // Private Key of recipient
KeySym []byte // Key associated with the Topic
Topics []TopicType // Topics to filter messages with
PoW float64 // Proof of work as described in the Whisper spec
AcceptP2P bool // Indicates whether this filter is interested in direct peer-to-peer messages
SymKeyHash common.Hash // The Keccak256Hash of the symmetric key, needed for optimization
Messages map[common.Hash]*ReceivedMessage
mutex sync.RWMutex
}
type Filters struct {
@ -50,32 +51,32 @@ func NewFilters(w *Whisper) *Filters {
}
}
func (self *Filters) Install(watcher *Filter) int {
self.mutex.Lock()
defer self.mutex.Unlock()
func (fs *Filters) Install(watcher *Filter) int {
fs.mutex.Lock()
defer fs.mutex.Unlock()
self.watchers[self.id] = watcher
ret := self.id
self.id++
fs.watchers[fs.id] = watcher
ret := fs.id
fs.id++
return ret
}
func (self *Filters) Uninstall(id int) {
self.mutex.Lock()
defer self.mutex.Unlock()
delete(self.watchers, id)
func (fs *Filters) Uninstall(id int) {
fs.mutex.Lock()
defer fs.mutex.Unlock()
delete(fs.watchers, id)
}
func (self *Filters) Get(i int) *Filter {
self.mutex.RLock()
defer self.mutex.RUnlock()
return self.watchers[i]
func (fs *Filters) Get(i int) *Filter {
fs.mutex.RLock()
defer fs.mutex.RUnlock()
return fs.watchers[i]
}
func (self *Filters) NotifyWatchers(env *Envelope, messageCode uint64) {
self.mutex.RLock()
func (fs *Filters) NotifyWatchers(env *Envelope, messageCode uint64) {
fs.mutex.RLock()
var msg *ReceivedMessage
for _, watcher := range self.watchers {
for _, watcher := range fs.watchers {
if messageCode == p2pCode && !watcher.AcceptP2P {
continue
}
@ -94,58 +95,58 @@ func (self *Filters) NotifyWatchers(env *Envelope, messageCode uint64) {
watcher.Trigger(msg)
}
}
self.mutex.RUnlock() // we need to unlock before calling addDecryptedMessage
fs.mutex.RUnlock() // we need to unlock before calling addDecryptedMessage
if msg != nil {
self.whisper.addDecryptedMessage(msg)
fs.whisper.addDecryptedMessage(msg)
}
}
func (self *Filter) expectsAsymmetricEncryption() bool {
return self.KeyAsym != nil
func (f *Filter) expectsAsymmetricEncryption() bool {
return f.KeyAsym != nil
}
func (self *Filter) expectsSymmetricEncryption() bool {
return self.KeySym != nil
func (f *Filter) expectsSymmetricEncryption() bool {
return f.KeySym != nil
}
func (self *Filter) Trigger(msg *ReceivedMessage) {
self.Mutex.Lock()
defer self.Mutex.Unlock()
func (f *Filter) Trigger(msg *ReceivedMessage) {
f.mutex.Lock()
defer f.mutex.Unlock()
if _, exist := self.Messages[msg.EnvelopeHash]; !exist {
self.Messages[msg.EnvelopeHash] = msg
if _, exist := f.Messages[msg.EnvelopeHash]; !exist {
f.Messages[msg.EnvelopeHash] = msg
}
}
func (self *Filter) Retrieve() (all []*ReceivedMessage) {
self.Mutex.Lock()
defer self.Mutex.Unlock()
func (f *Filter) Retrieve() (all []*ReceivedMessage) {
f.mutex.Lock()
defer f.mutex.Unlock()
all = make([]*ReceivedMessage, 0, len(self.Messages))
for _, msg := range self.Messages {
all = make([]*ReceivedMessage, 0, len(f.Messages))
for _, msg := range f.Messages {
all = append(all, msg)
}
self.Messages = make(map[common.Hash]*ReceivedMessage) // delete old messages
f.Messages = make(map[common.Hash]*ReceivedMessage) // delete old messages
return all
}
func (self *Filter) MatchMessage(msg *ReceivedMessage) bool {
if self.PoW > 0 && msg.PoW < self.PoW {
func (f *Filter) MatchMessage(msg *ReceivedMessage) bool {
if f.PoW > 0 && msg.PoW < f.PoW {
return false
}
if self.Src != nil && !isEqual(msg.Src, self.Src) {
if f.Src != nil && !isEqual(msg.Src, f.Src) {
return false
}
if self.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() {
if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() {
// if Dst match, ignore the topic
return isEqual(self.Dst, msg.Dst)
} else if self.expectsSymmetricEncryption() && msg.isSymmetricEncryption() {
return isEqual(f.Dst, msg.Dst)
} else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() {
// check if that both the key and the topic match
if self.TopicKeyHash == msg.TopicKeyHash {
for _, t := range self.Topics {
if f.SymKeyHash == msg.TopicKeyHash {
for _, t := range f.Topics {
if t == msg.Topic {
return true
}
@ -156,23 +157,23 @@ func (self *Filter) MatchMessage(msg *ReceivedMessage) bool {
return false
}
func (self *Filter) MatchEnvelope(envelope *Envelope) bool {
if self.PoW > 0 && envelope.pow < self.PoW {
func (f *Filter) MatchEnvelope(envelope *Envelope) bool {
if f.PoW > 0 && envelope.pow < f.PoW {
return false
}
encryptionMethodMatch := false
if self.expectsAsymmetricEncryption() && envelope.isAsymmetric() {
if f.expectsAsymmetricEncryption() && envelope.isAsymmetric() {
encryptionMethodMatch = true
if self.Topics == nil {
if f.Topics == nil {
return true // wildcard
}
} else if self.expectsSymmetricEncryption() && envelope.IsSymmetric() {
} else if f.expectsSymmetricEncryption() && envelope.IsSymmetric() {
encryptionMethodMatch = true
}
if encryptionMethodMatch {
for _, t := range self.Topics {
for _, t := range f.Topics {
if t == envelope.Topic {
return true
}

105
whisper/whisperv5/message.go
View file

@ -85,12 +85,12 @@ func isMessagePadded(flags byte) bool {
return (flags & paddingMask) != 0
}
func (self *ReceivedMessage) isSymmetricEncryption() bool {
return self.TopicKeyHash != common.Hash{}
func (msg *ReceivedMessage) isSymmetricEncryption() bool {
return msg.TopicKeyHash != common.Hash{}
}
func (self *ReceivedMessage) isAsymmetricEncryption() bool {
return self.Dst != nil
func (msg *ReceivedMessage) isAsymmetricEncryption() bool {
return msg.Dst != nil
}
func DeriveOneTimeKey(key []byte, salt []byte, version uint64) ([]byte, error) {
@ -121,7 +121,7 @@ func NewSentMessage(params *MessageParams) *SentMessage {
// appendPadding appends the pseudorandom padding bytes and sets the padding flag.
// The last byte contains the size of padding (thus, its size must not exceed 256).
func (self *SentMessage) appendPadding(params *MessageParams) {
func (msg *SentMessage) appendPadding(params *MessageParams) {
total := len(params.Payload) + 1
if params.Src != nil {
total += signatureLength
@ -145,44 +145,44 @@ func (self *SentMessage) appendPadding(params *MessageParams) {
if params.Padding != nil {
copy(buf[1:], params.Padding)
}
self.Raw = append(self.Raw, buf...)
self.Raw[0] |= byte(0x1) // number of bytes indicating the padding size
msg.Raw = append(msg.Raw, buf...)
msg.Raw[0] |= byte(0x1) // number of bytes indicating the padding size
}
}
// sign calculates and sets the cryptographic signature for the message,
// also setting the sign flag.
func (self *SentMessage) sign(key *ecdsa.PrivateKey) error {
if isMessageSigned(self.Raw[0]) {
func (msg *SentMessage) sign(key *ecdsa.PrivateKey) error {
if isMessageSigned(msg.Raw[0]) {
// this should not happen, but no reason to panic
glog.V(logger.Error).Infof("Trying to sign a message which was already signed")
return nil
}
hash := crypto.Keccak256(self.Raw)
hash := crypto.Keccak256(msg.Raw)
signature, err := crypto.Sign(hash, key)
if err != nil {
self.Raw = append(self.Raw, signature...)
self.Raw[0] |= signatureFlag
msg.Raw = append(msg.Raw, signature...)
msg.Raw[0] |= signatureFlag
return err
}
return nil
}
// encryptAsymmetric encrypts a message with a public key.
func (self *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error {
func (msg *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error {
if !ValidatePublicKey(key) {
return fmt.Errorf("Invalid public key provided for asymmetric encryption")
}
encrypted, err := crypto.Encrypt(key, self.Raw)
encrypted, err := crypto.Encrypt(key, msg.Raw)
if err == nil {
self.Raw = encrypted
msg.Raw = encrypted
}
return err
}
// encryptSymmetric encrypts a message with a topic key, using AES-GCM-256.
// nonce size should be 12 bytes (see cipher.gcmStandardNonceSize).
func (self *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) {
func (msg *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) {
if !validateSymmetricKey(key) {
return nil, nil, errors.New("invalid key provided for symmetric encryption")
}
@ -217,7 +217,7 @@ func (self *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte
if err != nil {
return nil, nil, err
}
self.Raw = aesgcm.Seal(nil, nonce, self.Raw, nil)
msg.Raw = aesgcm.Seal(nil, nonce, msg.Raw, nil)
return salt, nonce, nil
}
@ -233,24 +233,24 @@ func (self *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte
// - options.From != nil && options.To == nil: signed broadcast (known sender)
// - options.From == nil && options.To != nil: encrypted anonymous message
// - options.From != nil && options.To != nil: encrypted signed message
func (self *SentMessage) Wrap(options MessageParams) (envelope *Envelope, err error) {
func (msg *SentMessage) Wrap(options MessageParams) (envelope *Envelope, err error) {
if options.TTL == 0 {
options.TTL = DefaultTTL
}
if options.Src != nil {
if err = self.sign(options.Src); err != nil {
if err = msg.sign(options.Src); err != nil {
return nil, err
}
}
if len(self.Raw) > MaxMessageLength {
if len(msg.Raw) > MaxMessageLength {
glog.V(logger.Error).Infof("Message size must not exceed %d bytes", MaxMessageLength)
return nil, errors.New("Oversized message")
}
var salt, nonce []byte
if options.Dst != nil {
err = self.encryptAsymmetric(options.Dst)
err = msg.encryptAsymmetric(options.Dst)
} else if options.KeySym != nil {
salt, nonce, err = self.encryptSymmetric(options.KeySym)
salt, nonce, err = msg.encryptSymmetric(options.KeySym)
} else {
err = errors.New("Unable to encrypt the message: neither Dst nor Key")
}
@ -259,15 +259,15 @@ func (self *SentMessage) Wrap(options MessageParams) (envelope *Envelope, err er
return nil, err
}
envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, self)
envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, msg)
envelope.Seal(options)
return envelope, nil
}
// decryptSymmetric decrypts a message with a topic key, using AES-GCM-256.
// nonce size should be 12 bytes (see cipher.gcmStandardNonceSize).
func (self *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []byte) error {
derivedKey, err := DeriveOneTimeKey(key, salt, self.EnvelopeVersion)
func (msg *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []byte) error {
derivedKey, err := DeriveOneTimeKey(key, salt, msg.EnvelopeVersion)
if err != nil {
return err
}
@ -281,76 +281,77 @@ func (self *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []b
return err
}
if len(nonce) != aesgcm.NonceSize() {
glog.V(logger.Error).Infof("AES nonce size must be %d bytes", aesgcm.NonceSize())
return errors.New("Wrong AES nonce size")
info := fmt.Sprintf("Wrong AES nonce size - want: %d, got: %d", len(nonce), aesgcm.NonceSize())
glog.V(logger.Error).Infof(info)
return errors.New(info)
}
decrypted, err := aesgcm.Open(nil, nonce, self.Raw, nil)
decrypted, err := aesgcm.Open(nil, nonce, msg.Raw, nil)
if err != nil {
return err
}
self.Raw = decrypted
msg.Raw = decrypted
return nil
}
// decryptAsymmetric decrypts an encrypted payload with a private key.
func (self *ReceivedMessage) decryptAsymmetric(key *ecdsa.PrivateKey) error {
decrypted, err := crypto.Decrypt(key, self.Raw)
func (msg *ReceivedMessage) decryptAsymmetric(key *ecdsa.PrivateKey) error {
decrypted, err := crypto.Decrypt(key, msg.Raw)
if err == nil {
self.Raw = decrypted
msg.Raw = decrypted
}
return err
}
// Validate checks the validity and extracts the fields in case of success
func (self *ReceivedMessage) Validate() bool {
end := len(self.Raw)
func (msg *ReceivedMessage) Validate() bool {
end := len(msg.Raw)
if end < 1 {
return false
}
if isMessageSigned(self.Raw[0]) {
if isMessageSigned(msg.Raw[0]) {
end -= signatureLength
if end <= 1 {
return false
}
self.Signature = self.Raw[end:]
self.Src = self.Recover()
if self.Src == nil {
msg.Signature = msg.Raw[end:]
msg.Src = msg.Recover()
if msg.Src == nil {
return false
}
}
padSize, ok := self.extractPadding(end)
padSize, ok := msg.extractPadding(end)
if !ok {
return false
}
self.Payload = self.Raw[1+padSize : end]
return self.isSymmetricEncryption() != self.isAsymmetricEncryption()
msg.Payload = msg.Raw[1+padSize : end]
return msg.isSymmetricEncryption() != msg.isAsymmetricEncryption()
}
// extractPadding extracts the padding from raw message.
// although we don't support sending messages with padding size
// exceeding 255 bytes, such messages are perfectly valid, and
// can be successfully decrypted.
func (self *ReceivedMessage) extractPadding(end int) (int, bool) {
func (msg *ReceivedMessage) extractPadding(end int) (int, bool) {
paddingSize := 0
sz := int(self.Raw[0] & paddingMask) // number of bytes containing the entire size of padding, could be zero
sz := int(msg.Raw[0] & paddingMask) // number of bytes containing the entire size of padding, could be zero
if sz != 0 {
paddingSize = int(bytesToIntLittleEndian(self.Raw[1 : 1+sz]))
paddingSize = int(bytesToIntLittleEndian(msg.Raw[1 : 1+sz]))
if paddingSize < sz || paddingSize+1 > end {
return 0, false
}
self.Padding = self.Raw[1+sz : 1+paddingSize]
msg.Padding = msg.Raw[1+sz : 1+paddingSize]
}
return paddingSize, true
}
// Recover retrieves the public key of the message signer.
func (self *ReceivedMessage) Recover() *ecdsa.PublicKey {
func (msg *ReceivedMessage) Recover() *ecdsa.PublicKey {
defer func() { recover() }() // in case of invalid signature
pub, err := crypto.SigToPub(self.hash(), self.Signature)
pub, err := crypto.SigToPub(msg.hash(), msg.Signature)
if err != nil {
glog.V(logger.Error).Infof("Could not get public key from signature: %v", err)
return nil
@ -359,10 +360,10 @@ func (self *ReceivedMessage) Recover() *ecdsa.PublicKey {
}
// hash calculates the SHA3 checksum of the message flags, payload and padding.
func (self *ReceivedMessage) hash() []byte {
if isMessageSigned(self.Raw[0]) {
sz := len(self.Raw) - signatureLength
return crypto.Keccak256(self.Raw[:sz])
func (msg *ReceivedMessage) hash() []byte {
if isMessageSigned(msg.Raw[0]) {
sz := len(msg.Raw) - signatureLength
return crypto.Keccak256(msg.Raw[:sz])
}
return crypto.Keccak256(self.Raw)
return crypto.Keccak256(msg.Raw)
}

56
whisper/whisperv5/peer.go
View file

@ -54,27 +54,27 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
// start initiates the peer updater, periodically broadcasting the whisper packets
// into the network.
func (self *Peer) start() {
go self.update()
glog.V(logger.Debug).Infof("%v: whisper started", self.peer)
func (p *Peer) start() {
go p.update()
glog.V(logger.Debug).Infof("%v: whisper started", p.peer)
}
// stop terminates the peer updater, stopping message forwarding to it.
func (self *Peer) stop() {
close(self.quit)
glog.V(logger.Debug).Infof("%v: whisper stopped", self.peer)
func (p *Peer) stop() {
close(p.quit)
glog.V(logger.Debug).Infof("%v: whisper stopped", p.peer)
}
// handshake sends the protocol initiation status message to the remote peer and
// verifies the remote status too.
func (self *Peer) handshake() error {
func (p *Peer) handshake() error {
// Send the handshake status message asynchronously
errc := make(chan error, 1)
go func() {
errc <- p2p.Send(self.ws, statusCode, ProtocolVersion)
errc <- p2p.Send(p.ws, statusCode, ProtocolVersion)
}()
// Fetch the remote status packet and verify protocol match
packet, err := self.ws.ReadMsg()
packet, err := p.ws.ReadMsg()
if err != nil {
return err
}
@ -101,7 +101,7 @@ func (self *Peer) handshake() error {
// update executes periodic operations on the peer, including message transmission
// and expiration.
func (self *Peer) update() {
func (p *Peer) update() {
// Start the tickers for the updates
expire := time.NewTicker(expirationCycle)
transmit := time.NewTicker(transmissionCycle)
@ -110,41 +110,41 @@ func (self *Peer) update() {
for {
select {
case <-expire.C:
self.expire()
p.expire()
case <-transmit.C:
if err := self.broadcast(); err != nil {
glog.V(logger.Info).Infof("%v: broadcast failed: %v", self.peer, err)
if err := p.broadcast(); err != nil {
glog.V(logger.Info).Infof("%v: broadcast failed: %v", p.peer, err)
return
}
case <-self.quit:
case <-p.quit:
return
}
}
}
// mark marks an envelope known to the peer so that it won't be sent back.
func (self *Peer) mark(envelope *Envelope) {
self.known.Add(envelope.Hash())
func (peer *Peer) mark(envelope *Envelope) {
peer.known.Add(envelope.Hash())
}
// marked checks if an envelope is already known to the remote peer.
func (self *Peer) marked(envelope *Envelope) bool {
return self.known.Has(envelope.Hash())
func (peer *Peer) 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 (self *Peer) expire() {
func (peer *Peer) expire() {
// Assemble the list of available envelopes
available := set.NewNonTS()
for _, envelope := range self.host.Envelopes() {
for _, envelope := range peer.host.Envelopes() {
available.Add(envelope.Hash())
}
// Cross reference availability with known status
unmark := make(map[common.Hash]struct{})
self.known.Each(func(v interface{}) bool {
peer.known.Each(func(v interface{}) bool {
if !available.Has(v.(common.Hash)) {
unmark[v.(common.Hash)] = struct{}{}
}
@ -152,26 +152,26 @@ func (self *Peer) expire() {
})
// Dump all known but unavailable
for hash, _ := range unmark {
self.known.Remove(hash)
peer.known.Remove(hash)
}
}
// broadcast iterates over the collection of envelopes and transmits yet unknown
// ones over the network.
func (self *Peer) broadcast() error {
func (p *Peer) broadcast() error {
// Fetch the envelopes and collect the unknown ones
envelopes := self.host.Envelopes()
envelopes := p.host.Envelopes()
transmit := make([]*Envelope, 0, len(envelopes))
for _, envelope := range envelopes {
if !self.marked(envelope) {
if !p.marked(envelope) {
transmit = append(transmit, envelope)
self.mark(envelope)
p.mark(envelope)
}
}
// Transmit the unknown batch (potentially empty)
if err := p2p.Send(self.ws, messagesCode, transmit); err != nil {
if err := p2p.Send(p.ws, messagesCode, transmit); err != nil {
return err
}
glog.V(logger.Detail).Infoln(self.peer, "broadcasted", len(transmit), "message(s)")
glog.V(logger.Detail).Infoln(p.peer, "broadcasted", len(transmit), "message(s)")
return nil
}

4
whisper/whisperv5/topic.go
View file

@ -50,8 +50,8 @@ func HashToTopic(h common.Hash) (t TopicType) {
}
// String converts a topic byte array to a string representation.
func (self *TopicType) String() string {
return string(common.ToHex(self[:]))
func (topic *TopicType) String() string {
return string(common.ToHex(topic[:]))
}
// UnmarshalJSON parses a hex representation to a topic.

240
whisper/whisperv5/whisper.go
View file

@ -85,23 +85,23 @@ func New(server MailServer) *Whisper {
}
// Protocols returns the whisper sub-protocols ran by this particular client.
func (self *Whisper) Protocols() []p2p.Protocol {
return []p2p.Protocol{self.protocol}
func (w *Whisper) Protocols() []p2p.Protocol {
return []p2p.Protocol{w.protocol}
}
// Version returns the whisper sub-protocols version number.
func (self *Whisper) Version() uint {
return self.protocol.Version
func (w *Whisper) Version() uint {
return w.protocol.Version
}
func (self *Whisper) GetFilter(id int) *Filter {
return self.filters.Get(id)
func (w *Whisper) GetFilter(id int) *Filter {
return w.filters.Get(id)
}
func (self *Whisper) getPeer(peerID []byte) (*Peer, error) {
self.peerMu.Lock()
defer self.peerMu.Unlock()
for p, _ := range self.peers {
func (w *Whisper) getPeer(peerID []byte) (*Peer, error) {
w.peerMu.Lock()
defer w.peerMu.Unlock()
for p, _ := range w.peers {
id := p.peer.ID()
if bytes.Equal(peerID, id[:]) {
return p, nil
@ -112,8 +112,8 @@ func (self *Whisper) getPeer(peerID []byte) (*Peer, error) {
// MarkPeerTrusted marks specific peer trusted, which will allow it
// to send historic (expired) messages.
func (self *Whisper) MarkPeerTrusted(peerID []byte) error {
p, err := self.getPeer(peerID)
func (w *Whisper) MarkPeerTrusted(peerID []byte) error {
p, err := w.getPeer(peerID)
if err != nil {
return err
}
@ -121,26 +121,26 @@ func (self *Whisper) MarkPeerTrusted(peerID []byte) error {
return nil
}
func (self *Whisper) RequestHistoricMessages(peerID []byte, data []byte) error {
wp, err := self.getPeer(peerID)
func (w *Whisper) RequestHistoricMessages(peerID []byte, data []byte) error {
p, err := w.getPeer(peerID)
if err != nil {
return err
}
wp.trusted = true
return p2p.Send(wp.ws, mailRequestCode, data)
p.trusted = true
return p2p.Send(p.ws, mailRequestCode, data)
}
func (self *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error {
wp, err := self.getPeer(peerID)
func (w *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error {
p, err := w.getPeer(peerID)
if err != nil {
return err
}
return p2p.Send(wp.ws, p2pCode, envelope)
return p2p.Send(p.ws, p2pCode, envelope)
}
// NewIdentity generates a new cryptographic identity for the client, and injects
// it into the known identities for message decryption.
func (self *Whisper) NewIdentity() *ecdsa.PrivateKey {
func (w *Whisper) NewIdentity() *ecdsa.PrivateKey {
key, err := crypto.GenerateKey()
if err != nil || !validatePrivateKey(key) {
key, err = crypto.GenerateKey() // retry once
@ -151,36 +151,36 @@ func (self *Whisper) NewIdentity() *ecdsa.PrivateKey {
if !validatePrivateKey(key) {
panic("Failed to generate valid key")
}
self.keyMu.Lock()
defer self.keyMu.Unlock()
self.privateKeys[string(crypto.FromECDSAPub(&key.PublicKey))] = key
w.keyMu.Lock()
defer w.keyMu.Unlock()
w.privateKeys[string(crypto.FromECDSAPub(&key.PublicKey))] = key
return key
}
// DeleteIdentity deletes the specifies key if it exists.
func (self *Whisper) DeleteIdentity(key string) {
self.keyMu.Lock()
defer self.keyMu.Unlock()
delete(self.privateKeys, key)
func (w *Whisper) DeleteIdentity(key string) {
w.keyMu.Lock()
defer w.keyMu.Unlock()
delete(w.privateKeys, key)
}
// HasIdentity checks if the the whisper node is configured with the private key
// of the specified public pair.
func (self *Whisper) HasIdentity(key *ecdsa.PublicKey) bool {
self.keyMu.RLock()
defer self.keyMu.RUnlock()
return self.privateKeys[string(crypto.FromECDSAPub(key))] != nil
func (w *Whisper) HasIdentity(key *ecdsa.PublicKey) bool {
w.keyMu.RLock()
defer w.keyMu.RUnlock()
return w.privateKeys[string(crypto.FromECDSAPub(key))] != nil
}
// GetIdentity retrieves the private key of the specified public identity.
func (self *Whisper) GetIdentity(key *ecdsa.PublicKey) *ecdsa.PrivateKey {
self.keyMu.RLock()
defer self.keyMu.RUnlock()
return self.privateKeys[string(crypto.FromECDSAPub(key))]
func (w *Whisper) GetIdentity(key *ecdsa.PublicKey) *ecdsa.PrivateKey {
w.keyMu.RLock()
defer w.keyMu.RUnlock()
return w.privateKeys[string(crypto.FromECDSAPub(key))]
}
func (self *Whisper) GenerateTopicKey(name string) error {
if self.HasTopicKey(name) {
func (w *Whisper) GenerateTopicKey(name string) error {
if w.HasTopicKey(name) {
return fmt.Errorf("Key with name [%s] already exists", name)
}
@ -192,14 +192,14 @@ func (self *Whisper) GenerateTopicKey(name string) error {
return fmt.Errorf("crypto/rand failed to generate valid key")
}
self.keyMu.Lock()
defer self.keyMu.Unlock()
self.topicKeys[name] = key
w.keyMu.Lock()
defer w.keyMu.Unlock()
w.topicKeys[name] = key
return nil
}
func (self *Whisper) AddTopicKey(name string, key []byte) error {
if self.HasTopicKey(name) {
func (w *Whisper) AddTopicKey(name string, key []byte) error {
if w.HasTopicKey(name) {
return fmt.Errorf("Key with name [%s] already exists", name)
}
@ -208,77 +208,77 @@ func (self *Whisper) AddTopicKey(name string, key []byte) error {
return err
}
self.keyMu.Lock()
defer self.keyMu.Unlock()
self.topicKeys[name] = derived
w.keyMu.Lock()
defer w.keyMu.Unlock()
w.topicKeys[name] = derived
return nil
}
func (self *Whisper) HasTopicKey(name string) bool {
self.keyMu.RLock()
defer self.keyMu.RUnlock()
return self.topicKeys[name] != nil
func (w *Whisper) HasTopicKey(name string) bool {
w.keyMu.RLock()
defer w.keyMu.RUnlock()
return w.topicKeys[name] != nil
}
func (self *Whisper) DeleteTopicKey(name string) {
self.keyMu.Lock()
defer self.keyMu.Unlock()
delete(self.topicKeys, name)
func (w *Whisper) DeleteTopicKey(name string) {
w.keyMu.Lock()
defer w.keyMu.Unlock()
delete(w.topicKeys, name)
}
func (self *Whisper) GetTopicKey(name string) []byte {
self.keyMu.RLock()
defer self.keyMu.RUnlock()
return self.topicKeys[name]
func (w *Whisper) GetTopicKey(name string) []byte {
w.keyMu.RLock()
defer w.keyMu.RUnlock()
return w.topicKeys[name]
}
// Watch installs a new message handler to run in case a matching packet arrives
// from the whisper network.
func (self *Whisper) Watch(f *Filter) int {
return self.filters.Install(f)
func (w *Whisper) Watch(f *Filter) int {
return w.filters.Install(f)
}
// Unwatch removes an installed message handler.
func (self *Whisper) Unwatch(id int) {
self.filters.Uninstall(id)
func (w *Whisper) Unwatch(id int) {
w.filters.Uninstall(id)
}
// Send injects a message into the whisper send queue, to be distributed in the
// network in the coming cycles.
func (self *Whisper) Send(envelope *Envelope) error {
return self.add(envelope)
func (w *Whisper) Send(envelope *Envelope) error {
return w.add(envelope)
}
// Start implements node.Service, starting the background data propagation thread
// of the Whisper protocol.
func (self *Whisper) Start(*p2p.Server) error {
func (w *Whisper) Start(*p2p.Server) error {
glog.V(logger.Info).Infoln("Whisper started")
go self.update()
go w.update()
return nil
}
// Stop implements node.Service, stopping the background data propagation thread
// of the Whisper protocol.
func (self *Whisper) Stop() error {
close(self.quit)
func (w *Whisper) Stop() error {
close(w.quit)
glog.V(logger.Info).Infoln("Whisper stopped")
return nil
}
// handlePeer is called by the underlying P2P layer when the whisper sub-protocol
// connection is negotiated.
func (self *Whisper) HandlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
func (wh *Whisper) HandlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
// Create the new peer and start tracking it
whisperPeer := newPeer(self, peer, rw)
whisperPeer := newPeer(wh, peer, rw)
self.peerMu.Lock()
self.peers[whisperPeer] = struct{}{}
self.peerMu.Unlock()
wh.peerMu.Lock()
wh.peers[whisperPeer] = struct{}{}
wh.peerMu.Unlock()
defer func() {
self.peerMu.Lock()
delete(self.peers, whisperPeer)
self.peerMu.Unlock()
wh.peerMu.Lock()
delete(wh.peers, whisperPeer)
wh.peerMu.Unlock()
}()
// Run the peer handshake and state updates
@ -288,11 +288,11 @@ func (self *Whisper) HandlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
whisperPeer.start()
defer whisperPeer.stop()
return self.runMessageLoop(whisperPeer, rw)
return wh.runMessageLoop(whisperPeer, rw)
}
// runMessageLoop reads and processes inbound messages directly to merge into client-global state.
func (self *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
for {
// fetch the next packet
packet, err := rw.ReadMsg()
@ -313,13 +313,13 @@ func (self *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
}
// inject all envelopes into the internal pool
for _, envelope := range envelopes {
if err := self.add(envelope); err != nil {
if err := wh.add(envelope); err != nil {
glog.V(logger.Warn).Infof("%v: bad envelope received: [%v], peer will be disconnected", p.peer, err)
return fmt.Errorf("invalid envelope")
}
p.mark(envelope)
if self.mailServer != nil {
self.mailServer.Archive(envelope)
if wh.mailServer != nil {
wh.mailServer.Archive(envelope)
}
}
case p2pCode:
@ -334,16 +334,16 @@ func (self *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
return fmt.Errorf("garbage received (directMessage)")
}
for _, envelope := range envelopes {
self.postEvent(envelope, p2pCode)
wh.postEvent(envelope, p2pCode)
}
}
case mailRequestCode:
// Must be processed if mail server is implemented. Otherwise ignore.
if self.mailServer != nil {
if wh.mailServer != nil {
s := rlp.NewStream(packet.Payload, uint64(packet.Size))
data, err := s.Bytes()
if err == nil {
self.mailServer.DeliverMail(p, data)
wh.mailServer.DeliverMail(p, data)
} else {
glog.V(logger.Error).Infof("%v: bad requestHistoricMessages received: [%v]", p.peer, err)
}
@ -360,12 +360,12 @@ func (self *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
// 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 (self *Whisper) add(envelope *Envelope) error {
func (wh *Whisper) add(envelope *Envelope) error {
now := uint32(time.Now().Unix())
sent := envelope.Expiry - envelope.TTL
if sent > now {
if sent+SynchAllowance > now {
if sent-SynchAllowance > now {
return fmt.Errorf("message created in the future")
} else {
// recalculate PoW, adjusted for the time difference, plus one second for latency
@ -392,42 +392,42 @@ func (self *Whisper) add(envelope *Envelope) error {
hash := envelope.Hash()
self.poolMu.Lock()
_, alreadyCached := self.envelopes[hash]
wh.poolMu.Lock()
_, alreadyCached := wh.envelopes[hash]
if !alreadyCached {
self.envelopes[hash] = envelope
if self.expirations[envelope.Expiry] == nil {
self.expirations[envelope.Expiry] = set.NewNonTS()
wh.envelopes[hash] = envelope
if wh.expirations[envelope.Expiry] == nil {
wh.expirations[envelope.Expiry] = set.NewNonTS()
}
if !self.expirations[envelope.Expiry].Has(hash) {
self.expirations[envelope.Expiry].Add(hash)
if !wh.expirations[envelope.Expiry].Has(hash) {
wh.expirations[envelope.Expiry].Add(hash)
}
}
self.poolMu.Unlock()
wh.poolMu.Unlock()
if alreadyCached {
glog.V(logger.Detail).Infof("whisper envelope already cached: %x\n", envelope)
} else {
self.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 %x\n", envelope)
}
return nil
}
// postEvent delivers the message to the watchers.
func (self *Whisper) postEvent(envelope *Envelope, messageCode uint64) {
func (w *Whisper) postEvent(envelope *Envelope, messageCode uint64) {
// if the version of incoming message is higher than
// currently supported version, we can not decrypt it,
// and therefore just ignore this message
if envelope.Ver() <= EnvelopeVersion {
// todo: review if you need an additional thread here
go self.filters.NotifyWatchers(envelope, messageCode)
go w.filters.NotifyWatchers(envelope, messageCode)
}
}
// update loops until the lifetime of the whisper node, updating its internal
// state by expiring stale messages from the pool.
func (self *Whisper) update() {
func (w *Whisper) update() {
// Start a ticker to check for expirations
expire := time.NewTicker(expirationCycle)
@ -435,9 +435,9 @@ func (self *Whisper) update() {
for {
select {
case <-expire.C:
self.expire()
w.expire()
case <-self.quit:
case <-w.quit:
return
}
}
@ -445,46 +445,46 @@ func (self *Whisper) update() {
// expire iterates over all the expiration timestamps, removing all stale
// messages from the pools.
func (self *Whisper) expire() {
self.poolMu.Lock()
defer self.poolMu.Unlock()
func (w *Whisper) expire() {
w.poolMu.Lock()
defer w.poolMu.Unlock()
now := uint32(time.Now().Unix())
for then, hashSet := range self.expirations {
for then, hashSet := range w.expirations {
// Short circuit if a future time
if then > now {
continue
}
// Dump all expired messages and remove timestamp
hashSet.Each(func(v interface{}) bool {
delete(self.envelopes, v.(common.Hash))
delete(self.messages, v.(common.Hash))
delete(w.envelopes, v.(common.Hash))
delete(w.messages, v.(common.Hash))
return true
})
self.expirations[then].Clear()
w.expirations[then].Clear()
}
}
// envelopes retrieves all the messages currently pooled by the node.
func (self *Whisper) Envelopes() []*Envelope {
self.poolMu.RLock()
defer self.poolMu.RUnlock()
func (w *Whisper) Envelopes() []*Envelope {
w.poolMu.RLock()
defer w.poolMu.RUnlock()
all := make([]*Envelope, 0, len(self.envelopes))
for _, envelope := range self.envelopes {
all := make([]*Envelope, 0, len(w.envelopes))
for _, envelope := range w.envelopes {
all = append(all, envelope)
}
return all
}
// Messages retrieves all the currently pooled messages matching a filter id.
func (self *Whisper) Messages(id int) []*ReceivedMessage {
self.poolMu.RLock()
defer self.poolMu.RUnlock()
func (w *Whisper) Messages(id int) []*ReceivedMessage {
w.poolMu.RLock()
defer w.poolMu.RUnlock()
result := make([]*ReceivedMessage, 0)
if filter := self.filters.Get(id); filter != nil {
for _, msg := range self.messages {
if filter := w.filters.Get(id); filter != nil {
for _, msg := range w.messages {
if filter.MatchMessage(msg) {
result = append(result, msg)
}
@ -493,11 +493,11 @@ func (self *Whisper) Messages(id int) []*ReceivedMessage {
return result
}
func (self *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
self.poolMu.Lock()
defer self.poolMu.Unlock()
func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
w.poolMu.Lock()
defer w.poolMu.Unlock()
self.messages[msg.EnvelopeHash] = msg
w.messages[msg.EnvelopeHash] = msg
}
func ValidatePublicKey(k *ecdsa.PublicKey) bool {