From c2eced014d6d8c6170e4bdd8d30ea98862c98048 Mon Sep 17 00:00:00 2001 From: Vlad Date: Tue, 30 Aug 2016 12:29:10 +0200 Subject: [PATCH 1/4] whisper5: initial commit --- whisper5/doc.go | 61 ++++++++ whisper5/envelope.go | 155 ++++++++++++++++++++ whisper5/message.go | 342 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 558 insertions(+) create mode 100644 whisper5/doc.go create mode 100644 whisper5/envelope.go create mode 100644 whisper5/message.go diff --git a/whisper5/doc.go b/whisper5/doc.go new file mode 100644 index 0000000000..ea6e5589e2 --- /dev/null +++ b/whisper5/doc.go @@ -0,0 +1,61 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +/* +Package whisper implements the Whisper PoC-1. + +(https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec) + +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 +matter/energy duality (apologies to physicists for the blatant abuse of a +fundamental and beautiful natural principle). + +Whisper is a pure identity-based messaging system. Whisper provides a low-level +(non-application-specific) but easily-accessible API without being based upon +or prejudiced by the low-level hardware attributes and characteristics, +particularly the notion of singular endpoints. +*/ +package whisper5 + +import "time" + +const ( + statusCode = 0x00 + messagesCode = 0x01 + + protocolVersion uint64 = 5 + protocolVersionStr = "5.0" + protocolName = "shh" + + signatureFlag = byte(1 << 7) + paddingFlag = byte(1 << 6) + + signatureLength = 65 + maxPadLength = 256 // must not exceed 256 + aesKeyLength = 32 + saltLength = 12 + kdfIterations = 4096 + msgMaxLength = 0xFFFF + + expirationCycle = 800 * time.Millisecond + transmissionCycle = 300 * time.Millisecond + + DefaultTTL = 50 * time.Second + DefaultPoW = 50 * time.Millisecond +) + +type TopicType [4]byte diff --git a/whisper5/envelope.go b/whisper5/envelope.go new file mode 100644 index 0000000000..dda1fbd6db --- /dev/null +++ b/whisper5/envelope.go @@ -0,0 +1,155 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +// Contains the Whisper protocol Envelope element. For formal details please see +// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#envelopes. + +package whisper5 + +import ( + "crypto/ecdsa" + "encoding/binary" + "fmt" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/crypto/ecies" + "github.com/ethereum/go-ethereum/rlp" +) + +// 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 { + Expiry uint32 + TTL uint32 + Topic TopicType + Salt []byte + AESNonce []byte + Data []byte + EnvNonce uint64 + + hash common.Hash // Cached hash of the envelope to avoid rehashing every time +} + +// NewEnvelope wraps a Whisper message with expiration and destination data +// included into an envelope for network forwarding. +func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byte, msg *Message) *Envelope { + return &Envelope{ + Expiry: uint32(time.Now().Add(ttl).Unix()), + TTL: uint32(ttl.Seconds()), + Topic: topic, + Salt: salt, + AESNonce: aesNonce, + Data: msg.Raw, + EnvNonce: 0, + } +} + +// Seal closes the envelope by spending the requested amount of time as a proof +// of work on hashing the data. +func (self *Envelope) Seal(pow time.Duration) { + self.Expiry += uint32(pow.Seconds()) // adjust for the duration of Seal() execution + + buf := make([]byte, 64) + h := crypto.Keccak256(self.rlpWithoutNonce()) + copy(buf[:32], h) + + finish, bestBit := time.Now().Add(pow).UnixNano(), 0 + for nonce := uint64(0); time.Now().UnixNano() < finish; { + for i := 0; i < 1024; i++ { + binary.BigEndian.PutUint64(buf[56:], nonce) + h = crypto.Keccak256(buf) + firstBit := common.FirstBitSet(common.BigD(h)) + if firstBit > bestBit { + self.EnvNonce, bestBit = nonce, firstBit + } + nonce++ + } + } + //return bestBit // todo: uncomment? +} + +// 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 +} + +// 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) + } + return self.hash +} + +// DecodeRLP decodes an Envelope from an RLP data stream. +func (self *Envelope) DecodeRLP(s *rlp.Stream) error { + raw, err := s.Raw() + if err != nil { + return err + } + // The decoding of Envelope uses the struct fields but also needs + // to compute the hash of the whole RLP-encoded envelope. This + // type has the same structure as Envelope but is not an + // rlp.Decoder (does not implement DecodeRLP() function). + type rlpenv Envelope + if err := rlp.DecodeBytes(raw, (*rlpenv)(self)); err != nil { + return err + } + self.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) (*Message, error) { + message := &Message{ + Raw: self.Data, + //Sent: time.Unix(int64(self.Expiry-self.TTL), 0), + //TTL: time.Duration(self.TTL) * time.Second, + //Hash: self.Hash(), + } + + err := message.decryptAsymmetric(key) + switch err { + case nil: + return message, nil + + case ecies.ErrInvalidPublicKey: // addressed to somebody else + return nil, err + + default: + return nil, fmt.Errorf("unable to open envelope, decrypt failed: %v", err) + } +} + +// OpenSymmetric tries to decrypt an envelope, potentially encrypted with a particular key. +func (self *Envelope) OpenSymmetric(key []byte) (msg *Message, err error) { + msg = &Message{ + Raw: self.Data, + //Sent: time.Unix(int64(self.Expiry-self.TTL), 0), + //TTL: time.Duration(self.TTL) * time.Second, + //Hash: self.Hash(), + } + + err = msg.decryptSymmetric(key, self.Salt, self.AESNonce) + if err != nil { + msg = nil + } + return +} diff --git a/whisper5/message.go b/whisper5/message.go new file mode 100644 index 0000000000..906f869b8f --- /dev/null +++ b/whisper5/message.go @@ -0,0 +1,342 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +// Contains the Whisper protocol Message element. For formal details please see +// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#messages. +// todo: fix the spec link + +package whisper5 + +import ( + crand "crypto/rand" + "errors" + mrand "math/rand" + "time" + + "crypto/aes" + "crypto/cipher" + "crypto/ecdsa" + "crypto/sha256" + + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "golang.org/x/crypto/pbkdf2" +) + +// Options specifies the exact way a message should be wrapped into an Envelope. +type Options struct { + Topic TopicType + TTL time.Duration + Src *ecdsa.PrivateKey + Dst *ecdsa.PublicKey + Key []byte // must be 32 bytes. todo: review + Salt []byte + Pad []byte +} + +// Message represents an end-user data packet to transmit through the Whisper +// protocol. These are wrapped into Envelopes that need not be understood by +// intermediate nodes, just forwarded. +type Message struct { + //Flags byte // first bit: signature presence, second: padding presence + //Padding []byte // the first byte contains it's size + //Payload []byte // todo: delete all this + //Signature []byte + + Raw []byte + + // todo: following are the fields, extracted from the Raw field of received msg (not transmitted) + //Sent time.Time // Time when the message was posted into the network + //TTL time.Duration // Maximum time to live allowed for the message + // + Dst *ecdsa.PublicKey // Message recipient (identity used to decode the message) + //Hash common.Hash // Message envelope hash to act as a unique id +} + +func (self *Message) flags() byte { + return self.Raw[0] +} + +func (self *Message) isSigned() bool { + return (self.Raw[0] & signatureFlag) != 0 +} + +func (self *Message) isPadded() bool { + return (self.Raw[0] & paddingFlag) != 0 +} + +// Signature returns the signature part of the raw message. +func (self *Message) Signature() []byte { + sz := len(self.Raw) + if self.isSigned() && sz >= signatureLength+1 { + return self.Raw[sz-signatureLength:] + } else { + return nil + } +} + +// Payload returns the payload part of the raw message. +func (self *Message) Payload() []byte { + end := len(self.Raw) + if self.isSigned() { + end -= signatureLength + } + if self.isPadded() { + paddingSize := int(self.Raw[end-1]) + end -= paddingSize + } + if end <= 1 { + return nil + } + return self.Raw[1:end] +} + +// Padding returns the padding part of the raw message +// without the last byte (which only contains the padding size). +func (self *Message) Padding() []byte { + if !self.isPadded() { + return nil + } + end := len(self.Raw) + if self.isSigned() { + end -= signatureLength + } + paddingSize := int(self.Raw[end-1]) + beg := end - paddingSize + if beg <= 1 { + return nil + } + return self.Raw[beg : end-1] +} + +// NewMessage creates and initializes a non-signed, non-encrypted Whisper message. +func NewMessage(payload []byte) *Message { + // Construct an initial flag set: no signature, no padding, other bits random + flags := byte(mrand.Intn(256)) + flags &= ^signatureFlag + flags &= ^paddingFlag + + msg := Message{} //Message{Sent: time.Now()} // todo: review + msg.Raw = make([]byte, 1, len(payload)+signatureLength+maxPadLength) + msg.Raw[0] = flags + msg.Raw = append(msg.Raw, payload...) + return &msg +} + +// 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 *Message) appendPadding(options Options) { + if self.isSigned() { + // this should not happen, but no reason to panic + glog.V(logger.Error).Infof("Trying to pad a message which was already signed") + return + } else if self.isPadded() { + // this should not happen, but no reason to panic + glog.V(logger.Error).Infof("Trying to pad a message which was already padded") + return + } + + total := len(self.Raw) + if options.Src != nil { + total += signatureLength + } + odd := total % maxPadLength + if odd > 0 { + padSize := maxPadLength - odd + buf := make([]byte, padSize) + mrand.Read(buf) + if options.Pad != nil { + copy(buf, options.Pad) + } + buf[padSize-1] = byte(padSize) + self.Raw = append(self.Raw, buf...) + self.Raw[0] |= paddingFlag + } +} + +// sign calculates and sets the cryptographic signature for the message, +// also setting the sign flag. +func (self *Message) sign(key *ecdsa.PrivateKey) (err error) { + if self.isSigned() { + // this should not happen, but no reason to panic + glog.V(logger.Error).Infof("Trying to sign a message which was already signed") + return + } + signature, err := crypto.Sign(self.hash(), key) + if err != nil { + self.Raw = append(self.Raw, signature...) + self.Raw[0] |= signatureFlag + } + return +} + +// Recover retrieves the public key of the message signer. +func (self *Message) Recover() *ecdsa.PublicKey { + defer func() { recover() }() // in case of invalid signature + + signature := self.Signature() + if signature == nil { + return nil + } + pub, err := crypto.SigToPub(self.hash(), signature) + if err != nil { + glog.V(logger.Error).Infof("Could not get public key from signature: %v", err) + return nil + } + return pub +} + +// encryptAsymmetric encrypts a message with a public key. +func (self *Message) encryptAsymmetric(key *ecdsa.PublicKey) error { + encrypted, err := crypto.Encrypt(key, self.Raw) + if err == nil { + self.Raw = encrypted + } + return err +} + +// decryptAsymmetric decrypts an encrypted payload with a private key. +func (self *Message) decryptAsymmetric(key *ecdsa.PrivateKey) error { + decrypted, err := crypto.Decrypt(key, self.Raw) + if err == nil { + self.Raw = decrypted + } + 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 *Message) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) { + // todo: delete this block + // The key argument should be the AES-256 key, 32 bytes + //if len(key) != aesKeyLength { + // glog.V(logger.Error).Infof("AES key size must be %d bytes", aesKeyLength) + // err = errors.New("Wrong size of AES key") + // return + //} + + salt = make([]byte, saltLength) + _, err = crand.Read(salt) + if err != nil { + return + } + + derivedKey := pbkdf2.Key(key, salt, kdfIterations, aesKeyLength, sha256.New) + + block, err := aes.NewCipher(derivedKey) + if err != nil { + return + } + aesgcm, err := cipher.NewGCM(block) + if err != nil { + return + } + + // never use more than 2^32 random nonces with a given key + nonce = make([]byte, aesgcm.NonceSize()) + _, err = crand.Read(nonce) + if err != nil { + return + } + self.Raw = aesgcm.Seal(nil, nonce, self.Raw, nil) + return +} + +// decryptSymmetric decrypts a message with a topic key, using AES-GCM-256. +// nonce size should be 12 bytes (see cipher.gcmStandardNonceSize). +func (self *Message) decryptSymmetric(key []byte, salt []byte, nonce []byte) error { + // todo: delete this block + // The key argument should be the AES-256 key, 32 bytes + //if len(key) != aesKeyLength { + // glog.V(logger.Error).Infof("AES key size must be %d bytes", aesKeyLength) + // return errors.New("Wrong size of AES key") + //} + + derivedKey := pbkdf2.Key(key, salt, kdfIterations, aesKeyLength, sha256.New) + + block, err := aes.NewCipher(derivedKey) + if err != nil { + return err + } + aesgcm, err := cipher.NewGCM(block) + if err != nil { + 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") + } + decrypted, err := aesgcm.Open(nil, nonce, self.Raw, nil) + if err != nil { + return err + } + self.Raw = decrypted + return nil +} + +// hash calculates the SHA3 checksum of the message flags and payload. +func (self *Message) hash() []byte { + if self.isSigned() { + sz := len(self.Raw) - signatureLength + return crypto.Keccak256(self.Raw[:sz]) + } + return crypto.Keccak256(self.Raw) +} + +// Wrap bundles the message into an Envelope to transmit over the network. +// +// pow (Proof Of Work) controls how much time to spend on hashing the message, +// inherently controlling its priority through the network (smaller hash, bigger +// priority). +// +// The user can control the amount of identity, privacy and encryption through +// the options parameter as follows: +// - options.From == nil && options.To == nil: anonymous broadcast +// - 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 *Message) Wrap(pow time.Duration, options Options) (envelope *Envelope, err error) { + if options.TTL == 0 { + options.TTL = DefaultTTL + } + //self.TTL = options.TTL // todo: review + self.appendPadding(options) + if options.Src != nil { + if err = self.sign(options.Src); err != nil { + return + } + } + if len(self.Raw) > msgMaxLength { + glog.V(logger.Error).Infof("Message size must not exceed %d bytes", msgMaxLength) + err = errors.New("Oversized message") + return + } + var salt, nonce []byte + if options.Dst != nil { + err = self.encryptAsymmetric(options.Dst) + } else if options.Key != nil { + salt, nonce, err = self.encryptSymmetric(options.Key) + } else { + err = errors.New("Unable to encrypt the message: neither Dst nor Key") + } + + if err == nil { + envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, self) + envelope.Seal(pow) + } + return +} From d14292e0edd08ffdb86e5400035785a7d5246114 Mon Sep 17 00:00:00 2001 From: Vlad Date: Tue, 30 Aug 2016 16:58:13 +0200 Subject: [PATCH 2/4] whisper5: minor update --- whisper5/doc.go | 3 +++ whisper5/message.go | 8 +++++--- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/whisper5/doc.go b/whisper5/doc.go index ea6e5589e2..351b8e8aa1 100644 --- a/whisper5/doc.go +++ b/whisper5/doc.go @@ -58,4 +58,7 @@ const ( DefaultPoW = 50 * time.Millisecond ) +// Topic represents a cryptographically secure, probabilistic partial +// classifications of a message, determined as the first (left) 4 bytes of the +// SHA3 hash of some arbitrary data given by the original author of the message. type TopicType [4]byte diff --git a/whisper5/message.go b/whisper5/message.go index 906f869b8f..fbec82e3f4 100644 --- a/whisper5/message.go +++ b/whisper5/message.go @@ -23,7 +23,6 @@ package whisper5 import ( crand "crypto/rand" "errors" - mrand "math/rand" "time" "crypto/aes" @@ -126,7 +125,10 @@ func (self *Message) Padding() []byte { // NewMessage creates and initializes a non-signed, non-encrypted Whisper message. func NewMessage(payload []byte) *Message { // Construct an initial flag set: no signature, no padding, other bits random - flags := byte(mrand.Intn(256)) + buf := make([]byte, 1) + crand.Read(buf) + + flags := buf[0] flags &= ^signatureFlag flags &= ^paddingFlag @@ -158,7 +160,7 @@ func (self *Message) appendPadding(options Options) { if odd > 0 { padSize := maxPadLength - odd buf := make([]byte, padSize) - mrand.Read(buf) + crand.Read(buf) if options.Pad != nil { copy(buf, options.Pad) } From baef2464172a797f995e78a72c89faa4b8caf015 Mon Sep 17 00:00:00 2001 From: Vlad Date: Thu, 1 Sep 2016 18:54:25 +0200 Subject: [PATCH 3/4] whisper5: added new files --- whisper5/api.go | 424 +++++++++++++++++++++++++++++++++++++++++++ whisper5/envelope.go | 54 +++--- whisper5/filter.go | 141 ++++++++++++++ whisper5/message.go | 384 ++++++++++++++++++++++++--------------- whisper5/peer.go | 175 ++++++++++++++++++ whisper5/topic.go | 38 ++++ whisper5/whisper.go | 349 +++++++++++++++++++++++++++++++++++ 7 files changed, 1395 insertions(+), 170 deletions(-) create mode 100644 whisper5/api.go create mode 100644 whisper5/filter.go create mode 100644 whisper5/peer.go create mode 100644 whisper5/topic.go create mode 100644 whisper5/whisper.go diff --git a/whisper5/api.go b/whisper5/api.go new file mode 100644 index 0000000000..ddd17ac392 --- /dev/null +++ b/whisper5/api.go @@ -0,0 +1,424 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +package whisper5 + +// todo: this is just a stub, delete this block ASAP +type PublicWhisperAPI struct { + w *Whisper +} + +func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI { + return nil +} + +/* +import ( + "encoding/json" + "fmt" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/rpc" +) + +// PublicWhisperAPI provides the whisper RPC service. +type PublicWhisperAPI struct { + w *Whisper + + messagesMu sync.RWMutex + messages map[int]*whisperFilter +} + +type whisperOfflineError struct{} + +func (e *whisperOfflineError) Error() string { + return "whisper is offline" +} + +// whisperOffLineErr is returned when the node doesn't offer the shh service. +var whisperOffLineErr = new(whisperOfflineError) + +// NewPublicWhisperAPI create a new RPC whisper service. +func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI { + return &PublicWhisperAPI{w: w, messages: make(map[int]*whisperFilter)} +} + +// Version returns the Whisper version this node offers. +func (s *PublicWhisperAPI) Version() (*rpc.HexNumber, error) { + if s.w == nil { + return rpc.NewHexNumber(0), whisperOffLineErr + } + return rpc.NewHexNumber(s.w.Version()), nil +} + +// HasIdentity checks if the the whisper node is configured with the private key +// of the specified public pair. +func (s *PublicWhisperAPI) HasIdentity(identity string) (bool, error) { + if s.w == nil { + return false, whisperOffLineErr + } + return s.w.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil +} + +// NewIdentity generates a new cryptographic identity for the client, and injects +// it into the known identities for message decryption. +func (s *PublicWhisperAPI) NewIdentity() (string, error) { + if s.w == nil { + return "", whisperOffLineErr + } + + identity := s.w.NewIdentity() + return common.ToHex(crypto.FromECDSAPub(&identity.PublicKey)), nil +} + +type NewFilterArgs struct { + To string + From string + Topics [][][]byte +} + +// NewWhisperFilter creates and registers a new message filter to watch for inbound whisper messages. +func (s *PublicWhisperAPI) NewFilter(args NewFilterArgs) (*rpc.HexNumber, error) { + if s.w == nil { + return nil, whisperOffLineErr + } + + var id int + filter := Filter{ + To: crypto.ToECDSAPub(common.FromHex(args.To)), + From: crypto.ToECDSAPub(common.FromHex(args.From)), + Topics: NewFilterTopics(args.Topics...), + Fn: func(message *Message) { + wmsg := NewWhisperMessage(message) + s.messagesMu.RLock() // Only read lock to the filter pool + defer s.messagesMu.RUnlock() + if s.messages[id] != nil { + s.messages[id].insert(wmsg) + } + }, + } + + id = s.w.Watch(filter) + + s.messagesMu.Lock() + s.messages[id] = newWhisperFilter(id, s.w) + s.messagesMu.Unlock() + + return rpc.NewHexNumber(id), nil +} + +// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval. +func (s *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage { + s.messagesMu.RLock() + defer s.messagesMu.RUnlock() + + if s.messages[filterId.Int()] != nil { + if changes := s.messages[filterId.Int()].retrieve(); changes != nil { + return changes + } + } + return returnWhisperMessages(nil) +} + +// UninstallFilter disables and removes an existing filter. +func (s *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) bool { + s.messagesMu.Lock() + defer s.messagesMu.Unlock() + + if _, ok := s.messages[filterId.Int()]; ok { + delete(s.messages, filterId.Int()) + return true + } + return false +} + +// GetMessages retrieves all the known messages that match a specific filter. +func (s *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage { + // Retrieve all the cached messages matching a specific, existing filter + s.messagesMu.RLock() + defer s.messagesMu.RUnlock() + + var messages []*Message + if s.messages[filterId.Int()] != nil { + messages = s.messages[filterId.Int()].messages() + } + + return returnWhisperMessages(messages) +} + +// returnWhisperMessages converts a Whisper message to a RPC whisper message. +func returnWhisperMessages(messages []*Message) []WhisperMessage { + msgs := make([]WhisperMessage, len(messages)) + for i, msg := range messages { + msgs[i] = NewWhisperMessage(msg) + } + return msgs +} + +type PostArgs struct { + From string `json:"from"` + To string `json:"to"` + Topics [][]byte `json:"topics"` + Payload string `json:"payload"` + Priority int64 `json:"priority"` + TTL int64 `json:"ttl"` +} + +// Post injects a message into the whisper network for distribution. +func (s *PublicWhisperAPI) Post(args PostArgs) (bool, error) { + if s.w == nil { + return false, whisperOffLineErr + } + + // construct whisper message with transmission options + message := NewMessage(common.FromHex(args.Payload)) + options := Options{ + To: crypto.ToECDSAPub(common.FromHex(args.To)), + TTL: time.Duration(args.TTL) * time.Second, + Topics: NewTopics(args.Topics...), + } + + // set sender identity + if len(args.From) > 0 { + if key := s.w.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); key != nil { + options.From = key + } else { + return false, fmt.Errorf("unknown identity to send from: %s", args.From) + } + } + + // Wrap and send the message + pow := time.Duration(args.Priority) * time.Millisecond + envelope, err := message.Wrap(pow, options) + if err != nil { + return false, err + } + + return true, s.w.Send(envelope) +} + +// WhisperMessage is the RPC representation of a whisper message to be sent. +type WhisperMessage struct { + ref *Message + + Payload string `json:"payload"` + To string `json:"to"` + From string `json:"from"` + Sent int64 `json:"sent"` + TTL int64 `json:"ttl"` + Hash string `json:"hash"` +} + +func (args *PostArgs) UnmarshalJSON(data []byte) (err error) { + var obj struct { + From string `json:"from"` + To string `json:"to"` + Topics []string `json:"topics"` + Payload string `json:"payload"` + Priority rpc.HexNumber `json:"priority"` + TTL rpc.HexNumber `json:"ttl"` + } + + if err := json.Unmarshal(data, &obj); err != nil { + return err + } + + args.From = obj.From + args.To = obj.To + args.Payload = obj.Payload + args.Priority = obj.Priority.Int64() + args.TTL = obj.TTL.Int64() + + // decode topic strings + args.Topics = make([][]byte, len(obj.Topics)) + for i, topic := range obj.Topics { + args.Topics[i] = common.FromHex(topic) + } + + return nil +} + +// UnmarshalJSON implements the json.Unmarshaler interface, invoked to convert a +// JSON message blob into a WhisperFilterArgs structure. +func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { + // Unmarshal the JSON message and sanity check + var obj struct { + To interface{} `json:"to"` + From interface{} `json:"from"` + Topics interface{} `json:"topics"` + } + if err := json.Unmarshal(b, &obj); err != nil { + return err + } + + // Retrieve the simple data contents of the filter arguments + if obj.To == nil { + args.To = "" + } else { + argstr, ok := obj.To.(string) + if !ok { + return fmt.Errorf("to is not a string") + } + args.To = argstr + } + if obj.From == nil { + args.From = "" + } else { + argstr, ok := obj.From.(string) + if !ok { + return fmt.Errorf("from is not a string") + } + args.From = argstr + } + // Construct the nested topic array + if obj.Topics != nil { + // Make sure we have an actual topic array + list, ok := obj.Topics.([]interface{}) + if !ok { + return fmt.Errorf("topics is not an array") + } + // Iterate over each topic and handle nil, string or array + topics := make([][]string, len(list)) + for idx, field := range list { + switch value := field.(type) { + case nil: + topics[idx] = []string{} + + case string: + topics[idx] = []string{value} + + case []interface{}: + topics[idx] = make([]string, len(value)) + for i, nested := range value { + switch value := nested.(type) { + case nil: + topics[idx][i] = "" + + case string: + topics[idx][i] = value + + default: + return fmt.Errorf("topic[%d][%d] is not a string", idx, i) + } + } + default: + return fmt.Errorf("topic[%d] not a string or array", idx) + } + } + + topicsDecoded := make([][][]byte, len(topics)) + for i, condition := range topics { + topicsDecoded[i] = make([][]byte, len(condition)) + for j, topic := range condition { + topicsDecoded[i][j] = common.FromHex(topic) + } + } + + args.Topics = topicsDecoded + } + return nil +} + +// whisperFilter is the message cache matching a specific filter, accumulating +// inbound messages until the are requested by the client. +type whisperFilter struct { + id int // Filter identifier for old message retrieval + ref *Whisper // Whisper reference for old message retrieval + + cache []WhisperMessage // Cache of messages not yet polled + skip map[common.Hash]struct{} // List of retrieved messages to avoid duplication + update time.Time // Time of the last message query + + lock sync.RWMutex // Lock protecting the filter internals +} + +// messages retrieves all the cached messages from the entire pool matching the +// filter, resetting the filter's change buffer. +func (w *whisperFilter) messages() []*Message { + w.lock.Lock() + defer w.lock.Unlock() + + w.cache = nil + w.update = time.Now() + + w.skip = make(map[common.Hash]struct{}) + messages := w.ref.Messages(w.id) + for _, message := range messages { + w.skip[message.Hash] = struct{}{} + } + return messages +} + +// insert injects a new batch of messages into the filter cache. +func (w *whisperFilter) insert(messages ...WhisperMessage) { + w.lock.Lock() + defer w.lock.Unlock() + + for _, message := range messages { + if _, ok := w.skip[message.ref.Hash]; !ok { + w.cache = append(w.cache, messages...) + } + } +} + +// retrieve fetches all the cached messages from the filter. +func (w *whisperFilter) retrieve() (messages []WhisperMessage) { + w.lock.Lock() + defer w.lock.Unlock() + + messages, w.cache = w.cache, nil + w.update = time.Now() + + return +} + +// activity returns the last time instance when client requests were executed on +// the filter. +func (w *whisperFilter) activity() time.Time { + w.lock.RLock() + defer w.lock.RUnlock() + + return w.update +} + +// newWhisperFilter creates a new serialized, poll based whisper topic filter. +func newWhisperFilter(id int, ref *Whisper) *whisperFilter { + return &whisperFilter{ + id: id, + ref: ref, + + update: time.Now(), + skip: make(map[common.Hash]struct{}), + } +} + +// NewWhisperMessage converts an internal message into an API version. +func NewWhisperMessage(message *Message) WhisperMessage { + return WhisperMessage{ + ref: message, + + Payload: common.ToHex(message.Payload), + From: common.ToHex(crypto.FromECDSAPub(message.Recover())), + To: common.ToHex(crypto.FromECDSAPub(message.To)), + Sent: message.Sent.Unix(), + TTL: int64(message.TTL / time.Second), + Hash: common.ToHex(message.Hash.Bytes()), + } +} +*/ diff --git a/whisper5/envelope.go b/whisper5/envelope.go index dda1fbd6db..b282618e9a 100644 --- a/whisper5/envelope.go +++ b/whisper5/envelope.go @@ -43,11 +43,12 @@ type Envelope struct { EnvNonce uint64 hash common.Hash // Cached hash of the envelope to avoid rehashing every time + pow int // Message-specific PoW as described in the Whisper specification } // NewEnvelope wraps a Whisper message with expiration and destination data // included into an envelope for network forwarding. -func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byte, msg *Message) *Envelope { +func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byte, msg *SentMessage) *Envelope { return &Envelope{ Expiry: uint32(time.Now().Add(ttl).Unix()), TTL: uint32(ttl.Seconds()), @@ -59,16 +60,24 @@ func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byt } } +func (self *Envelope) isSymmetric() bool { + return self.AESNonce != nil +} + +func (self *Envelope) isAsymmetric() bool { + return !self.isSymmetric() +} + // Seal closes the envelope by spending the requested amount of time as a proof // of work on hashing the data. -func (self *Envelope) Seal(pow time.Duration) { - self.Expiry += uint32(pow.Seconds()) // adjust for the duration of Seal() execution +func (self *Envelope) Seal(dur time.Duration) { + self.Expiry += uint32(dur.Seconds()) // adjust for the duration of Seal() execution buf := make([]byte, 64) h := crypto.Keccak256(self.rlpWithoutNonce()) copy(buf[:32], h) - finish, bestBit := time.Now().Add(pow).UnixNano(), 0 + finish, bestBit := time.Now().Add(dur).UnixNano(), 0 for nonce := uint64(0); time.Now().UnixNano() < finish; { for i := 0; i < 1024; i++ { binary.BigEndian.PutUint64(buf[56:], nonce) @@ -107,7 +116,8 @@ func (self *Envelope) DecodeRLP(s *rlp.Stream) error { // The decoding of Envelope uses the struct fields but also needs // to compute the hash of the whole RLP-encoded envelope. This // type has the same structure as Envelope but is not an - // rlp.Decoder (does not implement DecodeRLP() function). + // 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 { return err @@ -117,39 +127,37 @@ func (self *Envelope) DecodeRLP(s *rlp.Stream) error { } // OpenAsymmetric tries to decrypt an envelope, potentially encrypted with a particular key. -func (self *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*Message, error) { - message := &Message{ - Raw: self.Data, - //Sent: time.Unix(int64(self.Expiry-self.TTL), 0), - //TTL: time.Duration(self.TTL) * time.Second, - //Hash: self.Hash(), - } - +func (self *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*ReceivedMessage, error) { + message := &ReceivedMessage{Raw: self.Data} err := message.decryptAsymmetric(key) switch err { case nil: return message, nil - case ecies.ErrInvalidPublicKey: // addressed to somebody else return nil, err - default: return nil, fmt.Errorf("unable to open envelope, decrypt failed: %v", err) } } // OpenSymmetric tries to decrypt an envelope, potentially encrypted with a particular key. -func (self *Envelope) OpenSymmetric(key []byte) (msg *Message, err error) { - msg = &Message{ - Raw: self.Data, - //Sent: time.Unix(int64(self.Expiry-self.TTL), 0), - //TTL: time.Duration(self.TTL) * time.Second, - //Hash: self.Hash(), - } - +func (self *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error) { + msg = &ReceivedMessage{Raw: self.Data} err = msg.decryptSymmetric(key, self.Salt, self.AESNonce) if err != nil { msg = nil } return } + +// Open tries to decrypt an envelope +func (self *Envelope) Open(watcher *Filter) *ReceivedMessage { + if self.isAsymmetric() { + msg, _ := self.OpenAsymmetric(watcher.KeyAsym) + return msg + } else if self.isSymmetric() { + msg, _ := self.OpenSymmetric(watcher.KeySym) + return msg + } + return nil +} diff --git a/whisper5/filter.go b/whisper5/filter.go new file mode 100644 index 0000000000..2d3b7798ce --- /dev/null +++ b/whisper5/filter.go @@ -0,0 +1,141 @@ +package whisper5 + +import ( + "bytes" + "crypto/ecdsa" +) + +var empty = TopicType{0, 0, 0, 0} + +type Filter struct { + Src *ecdsa.PublicKey // Sender of the message + Dst *ecdsa.PublicKey // Recipient of the message + KeyAsym *ecdsa.PrivateKey // Private Key of recipient + Topic TopicType // Topics to filter messages with + KeySym []byte // Key associated with the Topic + TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + PoW int // Proof of work as described in the Whisper spec + Fn func(msg *ReceivedMessage) // Handler in case of a match +} + +type Filters struct { + id int + watchers map[int]*Filter + ch chan Envelope + quit chan struct{} +} + +func NewFilters() *Filters { + return &Filters{ + ch: make(chan Envelope), + watchers: make(map[int]*Filter), + quit: make(chan struct{}), + } +} + +func (self *Filters) Start() { + go self.loop() +} + +func (self *Filters) Stop() { + close(self.quit) +} + +func (self *Filters) Notify(env *Envelope) { + self.ch <- *env +} + +func (self *Filters) Install(watcher *Filter) int { + self.watchers[self.id] = watcher + ret := self.id + self.id++ + return ret +} + +func (self *Filters) Uninstall(id int) { + delete(self.watchers, id) +} + +func (self *Filters) Get(i int) *Filter { + return self.watchers[i] +} + +func (self *Filters) loop() { + for { + select { + case <-self.quit: + return + case envelope := <-self.ch: + self.processEnvelope(&envelope) + } + } +} + +func (self *Filters) processEnvelope(envelope *Envelope) { + var msg *ReceivedMessage + for _, watcher := range self.watchers { + match := false + if msg != nil { + match = watcher.MatchMessage(msg) + } else { + match = watcher.MatchEnvelope(envelope) + if match { + msg = envelope.Open(watcher) // todo: fill all the fields & validate + } + } + + if match && msg != nil { + watcher.Trigger(msg) + } + } +} + +func (self Filter) expectsPublicKeyEncryption() bool { + return self.KeyAsym != nil +} + +func (self Filter) expectsTopicEncryption() bool { + return self.KeySym != nil +} + +func (self Filter) Trigger(msg *ReceivedMessage) { + go self.Fn(msg) // todo: review +} + +func (self Filter) MatchMessage(msg *ReceivedMessage) bool { + if self.PoW > 0 && msg.PoW < self.PoW { + return false + } + + if self.expectsPublicKeyEncryption() && msg.isAsymmetric() { + return self.Dst == msg.Dst + } else if self.expectsTopicEncryption() && msg.isSymmetric() { + // we need to compare the keys (or rather thier hashes), because of + // possible collision (different keys can produce the same topic). + // we also need to compare the topics, because they could be arbitrary (not related to KeySym). + if self.Topic == msg.Topic && bytes.Equal(self.TopicKeyHash, msg.TopicKeyHash) { + return true + } + } + return false +} + +func (self Filter) MatchEnvelope(envelope *Envelope) bool { + if self.PoW > 0 && envelope.pow < self.PoW { + return false + } + + encryptionMethodMatch := false + if self.expectsPublicKeyEncryption() && envelope.isAsymmetric() { + encryptionMethodMatch = true + } else if self.expectsTopicEncryption() && envelope.isSymmetric() { + encryptionMethodMatch = true + } + + if encryptionMethodMatch { + if self.Topic == empty || self.Topic == envelope.Topic { + return true + } + } + return false +} diff --git a/whisper5/message.go b/whisper5/message.go index fbec82e3f4..93c842cddf 100644 --- a/whisper5/message.go +++ b/whisper5/message.go @@ -30,6 +30,7 @@ import ( "crypto/ecdsa" "crypto/sha256" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger/glog" @@ -47,83 +48,50 @@ type Options struct { Pad []byte } -// Message represents an end-user data packet to transmit through the Whisper -// protocol. These are wrapped into Envelopes that need not be understood by -// intermediate nodes, just forwarded. -type Message struct { - //Flags byte // first bit: signature presence, second: padding presence - //Padding []byte // the first byte contains it's size - //Payload []byte // todo: delete all this - //Signature []byte +// SentMessage represents an end-user data packet to transmit through the +// Whisper protocol. These are wrapped into Envelopes that need not be +// understood by intermediate nodes, just forwarded. +type SentMessage struct { + Raw []byte +} +// ReceivedMessage represents a data packet to be received through the +// Whisper protocol. +type ReceivedMessage struct { Raw []byte - // todo: following are the fields, extracted from the Raw field of received msg (not transmitted) - //Sent time.Time // Time when the message was posted into the network - //TTL time.Duration // Maximum time to live allowed for the message - // - Dst *ecdsa.PublicKey // Message recipient (identity used to decode the message) - //Hash common.Hash // Message envelope hash to act as a unique id + Payload []byte + Padding []byte + Signature []byte + + PoW int // Proof of work as described in the Whisper spec + Sent time.Time // Time when the message was posted into the network + TTL time.Duration // Maximum time to live allowed for the message + Src *ecdsa.PublicKey // Message recipient (identity used to decode the message) + Dst *ecdsa.PublicKey // Message recipient (identity used to decode the message) + Topic TopicType + TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + EnvelopeHash common.Hash // Message envelope hash to act as a unique id } -func (self *Message) flags() byte { - return self.Raw[0] +func isMessageSigned(flags byte) bool { + return (flags & signatureFlag) != 0 } -func (self *Message) isSigned() bool { - return (self.Raw[0] & signatureFlag) != 0 +func isMessagePadded(flags byte) bool { + return (flags & paddingFlag) != 0 } -func (self *Message) isPadded() bool { - return (self.Raw[0] & paddingFlag) != 0 +func (self *ReceivedMessage) isSymmetric() bool { + return self.TopicKeyHash != nil } -// Signature returns the signature part of the raw message. -func (self *Message) Signature() []byte { - sz := len(self.Raw) - if self.isSigned() && sz >= signatureLength+1 { - return self.Raw[sz-signatureLength:] - } else { - return nil - } -} - -// Payload returns the payload part of the raw message. -func (self *Message) Payload() []byte { - end := len(self.Raw) - if self.isSigned() { - end -= signatureLength - } - if self.isPadded() { - paddingSize := int(self.Raw[end-1]) - end -= paddingSize - } - if end <= 1 { - return nil - } - return self.Raw[1:end] -} - -// Padding returns the padding part of the raw message -// without the last byte (which only contains the padding size). -func (self *Message) Padding() []byte { - if !self.isPadded() { - return nil - } - end := len(self.Raw) - if self.isSigned() { - end -= signatureLength - } - paddingSize := int(self.Raw[end-1]) - beg := end - paddingSize - if beg <= 1 { - return nil - } - return self.Raw[beg : end-1] +func (self *ReceivedMessage) isAsymmetric() bool { + return self.Dst != nil } // NewMessage creates and initializes a non-signed, non-encrypted Whisper message. -func NewMessage(payload []byte) *Message { +func NewSentMessage(payload []byte) *SentMessage { // Construct an initial flag set: no signature, no padding, other bits random buf := make([]byte, 1) crand.Read(buf) @@ -132,7 +100,7 @@ func NewMessage(payload []byte) *Message { flags &= ^signatureFlag flags &= ^paddingFlag - msg := Message{} //Message{Sent: time.Now()} // todo: review + msg := SentMessage{} //Message{Sent: time.Now()} // todo: review msg.Raw = make([]byte, 1, len(payload)+signatureLength+maxPadLength) msg.Raw[0] = flags msg.Raw = append(msg.Raw, payload...) @@ -141,12 +109,12 @@ func NewMessage(payload []byte) *Message { // 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 *Message) appendPadding(options Options) { - if self.isSigned() { +func (self *SentMessage) appendPadding(options Options) { + if isMessageSigned(self.Raw[0]) { // this should not happen, but no reason to panic glog.V(logger.Error).Infof("Trying to pad a message which was already signed") return - } else if self.isPadded() { + } else if isMessagePadded(self.Raw[0]) { // this should not happen, but no reason to panic glog.V(logger.Error).Infof("Trying to pad a message which was already padded") return @@ -172,13 +140,14 @@ func (self *Message) appendPadding(options Options) { // sign calculates and sets the cryptographic signature for the message, // also setting the sign flag. -func (self *Message) sign(key *ecdsa.PrivateKey) (err error) { - if self.isSigned() { +func (self *SentMessage) sign(key *ecdsa.PrivateKey) (err error) { + if isMessageSigned(self.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 } - signature, err := crypto.Sign(self.hash(), key) + hash := crypto.Keccak256(self.Raw) + signature, err := crypto.Sign(hash, key) if err != nil { self.Raw = append(self.Raw, signature...) self.Raw[0] |= signatureFlag @@ -186,24 +155,8 @@ func (self *Message) sign(key *ecdsa.PrivateKey) (err error) { return } -// Recover retrieves the public key of the message signer. -func (self *Message) Recover() *ecdsa.PublicKey { - defer func() { recover() }() // in case of invalid signature - - signature := self.Signature() - if signature == nil { - return nil - } - pub, err := crypto.SigToPub(self.hash(), signature) - if err != nil { - glog.V(logger.Error).Infof("Could not get public key from signature: %v", err) - return nil - } - return pub -} - // encryptAsymmetric encrypts a message with a public key. -func (self *Message) encryptAsymmetric(key *ecdsa.PublicKey) error { +func (self *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error { encrypted, err := crypto.Encrypt(key, self.Raw) if err == nil { self.Raw = encrypted @@ -211,26 +164,9 @@ func (self *Message) encryptAsymmetric(key *ecdsa.PublicKey) error { return err } -// decryptAsymmetric decrypts an encrypted payload with a private key. -func (self *Message) decryptAsymmetric(key *ecdsa.PrivateKey) error { - decrypted, err := crypto.Decrypt(key, self.Raw) - if err == nil { - self.Raw = decrypted - } - 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 *Message) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) { - // todo: delete this block - // The key argument should be the AES-256 key, 32 bytes - //if len(key) != aesKeyLength { - // glog.V(logger.Error).Infof("AES key size must be %d bytes", aesKeyLength) - // err = errors.New("Wrong size of AES key") - // return - //} - +func (self *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) { salt = make([]byte, saltLength) _, err = crand.Read(salt) if err != nil { @@ -258,47 +194,6 @@ func (self *Message) encryptSymmetric(key []byte) (salt []byte, nonce []byte, er return } -// decryptSymmetric decrypts a message with a topic key, using AES-GCM-256. -// nonce size should be 12 bytes (see cipher.gcmStandardNonceSize). -func (self *Message) decryptSymmetric(key []byte, salt []byte, nonce []byte) error { - // todo: delete this block - // The key argument should be the AES-256 key, 32 bytes - //if len(key) != aesKeyLength { - // glog.V(logger.Error).Infof("AES key size must be %d bytes", aesKeyLength) - // return errors.New("Wrong size of AES key") - //} - - derivedKey := pbkdf2.Key(key, salt, kdfIterations, aesKeyLength, sha256.New) - - block, err := aes.NewCipher(derivedKey) - if err != nil { - return err - } - aesgcm, err := cipher.NewGCM(block) - if err != nil { - 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") - } - decrypted, err := aesgcm.Open(nil, nonce, self.Raw, nil) - if err != nil { - return err - } - self.Raw = decrypted - return nil -} - -// hash calculates the SHA3 checksum of the message flags and payload. -func (self *Message) hash() []byte { - if self.isSigned() { - sz := len(self.Raw) - signatureLength - return crypto.Keccak256(self.Raw[:sz]) - } - return crypto.Keccak256(self.Raw) -} - // Wrap bundles the message into an Envelope to transmit over the network. // // pow (Proof Of Work) controls how much time to spend on hashing the message, @@ -311,7 +206,7 @@ func (self *Message) hash() []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 *Message) Wrap(pow time.Duration, options Options) (envelope *Envelope, err error) { +func (self *SentMessage) Wrap(pow time.Duration, options Options) (envelope *Envelope, err error) { if options.TTL == 0 { options.TTL = DefaultTTL } @@ -342,3 +237,198 @@ func (self *Message) Wrap(pow time.Duration, options Options) (envelope *Envelop } return } + +/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + +// 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 := pbkdf2.Key(key, salt, kdfIterations, aesKeyLength, sha256.New) + + block, err := aes.NewCipher(derivedKey) + if err != nil { + return err + } + aesgcm, err := cipher.NewGCM(block) + if err != nil { + 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") + } + decrypted, err := aesgcm.Open(nil, nonce, self.Raw, nil) + if err != nil { + return err + } + self.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) + if err == nil { + self.Raw = decrypted + } + return err +} + +// Validate checks the validity and extracts the fields in case of success +func (self *ReceivedMessage) Validate() bool { + sz := len(self.Raw) + cur := sz + if sz < 1 { + return false + } + + if isMessageSigned(self.Raw[0]) { + cur -= signatureLength + if cur <= 1 { + return false + } + self.Signature = self.Raw[cur:] + self.Src = self.Recover() + if self.Src == nil { + return false + } + } + + if isMessagePadded(self.Raw[0]) { + paddingSize := int(self.Raw[cur-1]) + beg := cur - paddingSize + if beg <= 1 { + return false + } + self.Padding = self.Raw[beg : cur-1] + cur = beg + } + + self.Payload = self.Raw[1:cur] + if self.isSymmetric() == self.isAsymmetric() { + return false + } + return true +} + +// Recover retrieves the public key of the message signer. +func (self *ReceivedMessage) Recover() *ecdsa.PublicKey { + defer func() { recover() }() // in case of invalid signature + + pub, err := crypto.SigToPub(self.hash(), self.Signature) + if err != nil { + glog.V(logger.Error).Infof("Could not get public key from signature: %v", err) + return nil + } + return pub +} + +// 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]) + } + return crypto.Keccak256(self.Raw) +} + +// todo: delete this stuff +/* +// Signature returns the signature part of the raw message. +func (self *ReceivedMessage) ExtractSignature() { + if self.Signature == nil { + if sz := len(self.Raw); sz >= signatureLength+1 { + if isMessageSigned(self.Raw[0]) { + self.Signature = self.Raw[sz-signatureLength:] + } + } + } +} + +// Payload returns the payload part of the raw message. +func (self *ReceivedMessage) ExtractPayload() { + if self.Payload == nil { + end := len(self.Raw) + if isMessageSigned(self.Raw[0]) { + end -= signatureLength + if end <= 1 { + return + } + } + if isMessagePadded(self.Raw[0]) { + paddingSize := int(self.Raw[end-1]) + end -= paddingSize + if end <= 1 { + return + } + } + self.Payload = self.Raw[1:end] + } +} + +// Padding returns the padding part of the raw message +// without the last byte (which only contains the padding size). +func (self *ReceivedMessage) ExtractPadding() { + if self.Padding == nil { + end := len(self.Raw) + if isMessagePadded(self.Raw[0]) { + if isMessageSigned(self.Raw[0]) { + end -= signatureLength + if end <= 1 { + return + } + } + paddingSize := int(self.Raw[end-1]) + beg := end - paddingSize + if beg > 1 { + self.Padding = self.Raw[beg : end-1] + } + } + } +} +*/ +/* +// Signature returns the signature part of the raw message. +func (self *ReceivedMessage) ExtractSignature() []byte { + sz := len(self.Raw) + if self.isSigned() && sz >= signatureLength+1 { + return self.Raw[sz-signatureLength:] + } else { + return nil + } +} + +// Payload returns the payload part of the raw message. +func (self *ReceivedMessage) ExtractPayload() []byte { + end := len(self.Raw) + if self.isSigned() { + end -= signatureLength + } + if self.isPadded() { + paddingSize := int(self.Raw[end-1]) + end -= paddingSize + } + if end <= 1 { + return nil + } + return self.Raw[1:end] +} + +// Padding returns the padding part of the raw message +// without the last byte (which only contains the padding size). +func (self *ReceivedMessage) ExtractPadding() []byte { + if !self.isPadded() { + return nil + } + end := len(self.Raw) + if self.isSigned() { + end -= signatureLength + } + paddingSize := int(self.Raw[end-1]) + beg := end - paddingSize + if beg <= 1 { + return nil + } + return self.Raw[beg : end-1] +} +*/ diff --git a/whisper5/peer.go b/whisper5/peer.go new file mode 100644 index 0000000000..434ecd62f9 --- /dev/null +++ b/whisper5/peer.go @@ -0,0 +1,175 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +package whisper5 + +import ( + "fmt" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" + set "gopkg.in/fatih/set.v0" +) + +// peer represents a whisper protocol peer connection. +type peer struct { + host *Whisper + peer *p2p.Peer + ws p2p.MsgReadWriter + + known *set.Set // Messages already known by the peer to avoid wasting bandwidth + + quit chan struct{} +} + +// newPeer creates a new whisper peer object, but does not run the handshake itself. +func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *peer { + return &peer{ + host: host, + peer: remote, + ws: rw, + known: set.New(), + quit: make(chan struct{}), + } +} + +// 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) +} + +// 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) +} + +// handshake sends the protocol initiation status message to the remote peer and +// verifies the remote status too. +func (self *peer) handshake() error { + // Send the handshake status message asynchronously + errc := make(chan error, 1) + go func() { + errc <- p2p.SendItems(self.ws, statusCode, protocolVersion) + }() + // Fetch the remote status packet and verify protocol match + packet, err := self.ws.ReadMsg() + if err != nil { + return err + } + if packet.Code != statusCode { + return fmt.Errorf("peer sent %x before status packet", packet.Code) + } + s := rlp.NewStream(packet.Payload, uint64(packet.Size)) + if _, err := s.List(); err != nil { + return fmt.Errorf("bad status message: %v", err) + } + peerVersion, err := s.Uint() + if err != nil { + return fmt.Errorf("bad status message: %v", err) + } + if peerVersion != protocolVersion { + return fmt.Errorf("protocol version mismatch %d != %d", peerVersion, protocolVersion) + } + // Wait until out own status is consumed too + if err := <-errc; err != nil { + return fmt.Errorf("failed to send status packet: %v", err) + } + return nil +} + +// update executes periodic operations on the peer, including message transmission +// and expiration. +func (self *peer) update() { + // Start the tickers for the updates + expire := time.NewTicker(expirationCycle) + transmit := time.NewTicker(transmissionCycle) + + // Loop and transmit until termination is requested + for { + select { + case <-expire.C: + self.expire() + + case <-transmit.C: + if err := self.broadcast(); err != nil { + glog.V(logger.Info).Infof("%v: broadcast failed: %v", self.peer, err) + return + } + + case <-self.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()) +} + +// 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()) +} + +// expire iterates over all the known envelopes in the host and removes all +// expired (unknown) ones from the known list. +func (self *peer) expire() { + // Assemble the list of available envelopes + available := set.NewNonTS() + for _, envelope := range self.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 { + if !available.Has(v.(common.Hash)) { + unmark[v.(common.Hash)] = struct{}{} + } + return true + }) + // Dump all known but unavailable + for hash, _ := range unmark { + self.known.Remove(hash) + } +} + +// broadcast iterates over the collection of envelopes and transmits yet unknown +// ones over the network. +func (self *peer) broadcast() error { + // Fetch the envelopes and collect the unknown ones + envelopes := self.host.Envelopes() + transmit := make([]*Envelope, 0, len(envelopes)) + for _, envelope := range envelopes { + if !self.marked(envelope) { + transmit = append(transmit, envelope) + self.mark(envelope) + } + } + // Transmit the unknown batch (potentially empty) + if err := p2p.Send(self.ws, messagesCode, transmit); err != nil { + return err + } + glog.V(logger.Detail).Infoln(self.peer, "broadcasted", len(transmit), "message(s)") + return nil +} diff --git a/whisper5/topic.go b/whisper5/topic.go new file mode 100644 index 0000000000..94ebbc0c6a --- /dev/null +++ b/whisper5/topic.go @@ -0,0 +1,38 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +// Contains the Whisper protocol Topic element. For formal details please see +// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#topics. + +package whisper5 + +import "github.com/ethereum/go-ethereum/crypto" + +func NewTopic(data []byte) TopicType { + prefix := [4]byte{} + copy(prefix[:], crypto.Keccak256(data)[:4]) + return TopicType(prefix) +} + +// NewTopicFromString creates a topic using the binary data contents of the specified string. +func NewTopicFromString(data string) TopicType { + return NewTopic([]byte(data)) +} + +// String converts a topic byte array to a string representation. +func (self *TopicType) String() string { + return string(self[:]) +} diff --git a/whisper5/whisper.go b/whisper5/whisper.go new file mode 100644 index 0000000000..3a6f1885ba --- /dev/null +++ b/whisper5/whisper.go @@ -0,0 +1,349 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum 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 go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// 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 go-ethereum library. If not, see . + +package whisper5 + +import ( + "crypto/ecdsa" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rpc" + + set "gopkg.in/fatih/set.v0" +) + +//type MessageEvent struct { +// To *ecdsa.PrivateKey +// From *ecdsa.PublicKey +// Message *Message +//} + +// Whisper represents a dark communication interface through the Ethereum +// network, using its very own P2P communication layer. +type Whisper struct { + protocol p2p.Protocol + filters *Filters + + privateKeys map[string]*ecdsa.PrivateKey + topicKeys map[TopicType][]byte // todo: move to the filter. this is not suitable because of possible collisions + + msgs map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages // todo: rename + envelopes map[common.Hash]*Envelope // Pool of messages currently tracked by this node + expirations map[uint32]*set.SetNonTS // Message expiration pool (TODO: something lighter) + poolMu sync.RWMutex // Mutex to sync the message and expiration pools + + peers map[*peer]struct{} // Set of currently active peers + peerMu sync.RWMutex // Mutex to sync the active peer set + + quit chan struct{} +} + +// New creates a Whisper client ready to communicate through the Ethereum P2P network. +func NewWhisper() *Whisper { + whisper := &Whisper{ + filters: NewFilters(), + privateKeys: make(map[string]*ecdsa.PrivateKey), + topicKeys: make(map[TopicType][]byte), + envelopes: make(map[common.Hash]*Envelope), + expirations: make(map[uint32]*set.SetNonTS), + peers: make(map[*peer]struct{}), + quit: make(chan struct{}), + } + whisper.filters.Start() + + // p2p whisper sub protocol handler + whisper.protocol = p2p.Protocol{ + Name: protocolName, + Version: uint(protocolVersion), + Length: 2, + Run: whisper.handlePeer, + } + + return whisper +} + +// APIs returns the RPC descriptors the Whisper implementation offers +func (s *Whisper) APIs() []rpc.API { + return []rpc.API{ + { + Namespace: protocolName, + Version: protocolVersionStr, + Service: NewPublicWhisperAPI(s), + Public: true, + }, + } +} + +// Protocols returns the whisper sub-protocols ran by this particular client. +func (self *Whisper) Protocols() []p2p.Protocol { + return []p2p.Protocol{self.protocol} +} + +// Version returns the whisper sub-protocols version number. +func (self *Whisper) Version() uint { + return self.protocol.Version +} + +// 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 { + // todo: review + key, err := crypto.GenerateKey() + if err != nil { + panic(err) + } + self.privateKeys[string(crypto.FromECDSAPub(&key.PublicKey))] = key + + return 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 { + return self.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 { + return self.privateKeys[string(crypto.FromECDSAPub(key))] +} + +// 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) +} + +// Unwatch removes an installed message handler. +func (self *Whisper) Unwatch(id int) { + self.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) +} + +// Start implements node.Service, starting the background data propagation thread +// of the Whisper protocol. +func (self *Whisper) Start(*p2p.Server) error { + glog.V(logger.Info).Infoln("Whisper started") + go self.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) + 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 { + // Create the new peer and start tracking it + whisperPeer := newPeer(self, peer, rw) + + self.peerMu.Lock() + self.peers[whisperPeer] = struct{}{} + self.peerMu.Unlock() + + defer func() { + self.peerMu.Lock() + delete(self.peers, whisperPeer) + self.peerMu.Unlock() + }() + + // Run the peer handshake and state updates + if err := whisperPeer.handshake(); err != nil { + return err + } + whisperPeer.start() + defer whisperPeer.stop() + + // Read and process inbound messages directly to merge into client-global state + for { + // Fetch the next packet and decode the contained envelopes + packet, err := rw.ReadMsg() + if err != nil { + return err + } + var envelopes []*Envelope + if err := packet.Decode(&envelopes); err != nil { + glog.V(logger.Info).Infof("%v: failed to decode envelope: %v", peer, err) + continue + } + // Inject all envelopes into the internal pool + for _, envelope := range envelopes { + if err := self.add(envelope); err != nil { + // TODO Punish peer here. Invalid envelope. + glog.V(logger.Debug).Infof("%v: failed to pool envelope: %v", peer, err) + } + whisperPeer.mark(envelope) + } + } +} + +// 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. +func (self *Whisper) add(envelope *Envelope) error { + self.poolMu.Lock() + defer self.poolMu.Unlock() + + // short circuit when a received envelope has already expired + if envelope.Expiry < uint32(time.Now().Unix()) { + return nil + } + + // Insert the message into the tracked pool + hash := envelope.Hash() + if _, ok := self.envelopes[hash]; ok { + glog.V(logger.Detail).Infof("whisper envelope already cached: %x\n", envelope) + return nil + } + self.envelopes[hash] = envelope + + // Insert the message into the expiration pool for later removal + if self.expirations[envelope.Expiry] == nil { + self.expirations[envelope.Expiry] = set.NewNonTS() + } + if !self.expirations[envelope.Expiry].Has(hash) { + self.expirations[envelope.Expiry].Add(hash) + + // Notify the local node of a message arrival + go self.postEvent(envelope) + } + 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) { + self.filters.Notify(envelope) +} + +/* +// createFilter creates a message filter to check against installed handlers. +func createFilter(message *Message, topics []TopicType) filter.Filter { + //return Filter{ + // Src: string(crypto.FromECDSAPub(message.Recover())), + // Dst: string(crypto.FromECDSAPub(message.Dst)), + // Topics: topics, + //} + + matcher := make([][]TopicType, len(topics)) + for i, topic := range topics { + matcher[i] = []TopicType{topic} + } + return filterer{ + to: string(crypto.FromECDSAPub(message.To)), + from: string(crypto.FromECDSAPub(message.Recover())), + matcher: newTopicMatcher(matcher...), + } +} +*/ + +// update loops until the lifetime of the whisper node, updating its internal +// state by expiring stale messages from the pool. +func (self *Whisper) update() { + // Start a ticker to check for expirations + expire := time.NewTicker(expirationCycle) + + // Repeat updates until termination is requested + for { + select { + case <-expire.C: + self.expire() + + case <-self.quit: + return + } + } +} + +// 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() + + now := uint32(time.Now().Unix()) + for then, hashSet := range self.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)) + return true + }) + self.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() + + all := make([]*Envelope, 0, len(self.envelopes)) + for _, envelope := range self.envelopes { + all = append(all, envelope) + } + return all +} + +/* +// Messages retrieves all the currently pooled messages matching a filter id. +// todo: review +//func (self *Whisper) Messages(id int) []*Message { +// messages := make([]*Message, 0) +// if filter := self.filters.Get(id); filter != nil { +// for _, envelope := range self.messages { +// if message := self.open(envelope); message != nil { +// if self.filters.Match(filter, createFilter(message, envelope.Topic)) { +// messages = append(messages, message) +// } +// } +// } +// } +// return messages +//} +func (self *Whisper) Messages(id int) []*ReceivedMessage { + messages := make([]*Envelope, 0) + if filter := self.filters.Get(id); filter != nil { + for _, envelope := range self.envelopes { + //if message := self.open(envelope); message != nil { + if self.filters.Match(envelope) { + messages = append(messages, envelope) + } + //} + } + } + return messages +} +*/ \ No newline at end of file From 309b3ba093a868469411ebc230214922cc833966 Mon Sep 17 00:00:00 2001 From: Vlad Date: Tue, 6 Sep 2016 22:58:58 +0200 Subject: [PATCH 4/4] whisper5: first buildable version --- whisper5/api.go | 429 ++++++++++++++++++++++--------------------- whisper5/doc.go | 24 ++- whisper5/envelope.go | 36 +++- whisper5/filter.go | 57 ++++-- whisper5/message.go | 151 +++------------ whisper5/whisper.go | 79 +++----- 6 files changed, 364 insertions(+), 412 deletions(-) diff --git a/whisper5/api.go b/whisper5/api.go index ddd17ac392..78020d5a0d 100644 --- a/whisper5/api.go +++ b/whisper5/api.go @@ -16,31 +16,24 @@ package whisper5 -// todo: this is just a stub, delete this block ASAP -type PublicWhisperAPI struct { - w *Whisper -} - -func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI { - return nil -} - -/* import ( "encoding/json" + "errors" "fmt" + "strconv" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc" ) // PublicWhisperAPI provides the whisper RPC service. type PublicWhisperAPI struct { - w *Whisper - + whisper *Whisper messagesMu sync.RWMutex messages map[int]*whisperFilter } @@ -56,114 +49,127 @@ var whisperOffLineErr = new(whisperOfflineError) // NewPublicWhisperAPI create a new RPC whisper service. func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI { - return &PublicWhisperAPI{w: w, messages: make(map[int]*whisperFilter)} + return &PublicWhisperAPI{whisper: w, messages: make(map[int]*whisperFilter)} } // Version returns the Whisper version this node offers. -func (s *PublicWhisperAPI) Version() (*rpc.HexNumber, error) { - if s.w == nil { +func (self *PublicWhisperAPI) Version() (*rpc.HexNumber, error) { + if self.whisper == nil { return rpc.NewHexNumber(0), whisperOffLineErr } - return rpc.NewHexNumber(s.w.Version()), nil + return rpc.NewHexNumber(self.whisper.Version()), nil } // HasIdentity checks if the the whisper node is configured with the private key // of the specified public pair. -func (s *PublicWhisperAPI) HasIdentity(identity string) (bool, error) { - if s.w == nil { +func (self *PublicWhisperAPI) HasIdentity(identity string) (bool, error) { + if self.whisper == nil { return false, whisperOffLineErr } - return s.w.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil + return self.whisper.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil } // NewIdentity generates a new cryptographic identity for the client, and injects // it into the known identities for message decryption. -func (s *PublicWhisperAPI) NewIdentity() (string, error) { - if s.w == nil { +func (self *PublicWhisperAPI) NewIdentity() (string, error) { + if self.whisper == nil { return "", whisperOffLineErr } - identity := s.w.NewIdentity() + identity := self.whisper.NewIdentity() return common.ToHex(crypto.FromECDSAPub(&identity.PublicKey)), nil } -type NewFilterArgs struct { - To string - From string - Topics [][][]byte -} - // NewWhisperFilter creates and registers a new message filter to watch for inbound whisper messages. -func (s *PublicWhisperAPI) NewFilter(args NewFilterArgs) (*rpc.HexNumber, error) { - if s.w == nil { +func (self *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (*rpc.HexNumber, error) { + if self.whisper == nil { return nil, whisperOffLineErr } var id int filter := Filter{ - To: crypto.ToECDSAPub(common.FromHex(args.To)), - From: crypto.ToECDSAPub(common.FromHex(args.From)), - Topics: NewFilterTopics(args.Topics...), - Fn: func(message *Message) { + Src: crypto.ToECDSAPub(common.FromHex(args.From)), + Dst: crypto.ToECDSAPub(common.FromHex(args.To)), + KeySym: common.FromHex(args.KeySym), + PoW: args.PoW, + Fn: func(message *ReceivedMessage) { wmsg := NewWhisperMessage(message) - s.messagesMu.RLock() // Only read lock to the filter pool - defer s.messagesMu.RUnlock() - if s.messages[id] != nil { - s.messages[id].insert(wmsg) + self.messagesMu.RLock() // Only read lock to the filter pool + defer self.messagesMu.RUnlock() + if self.messages[id] != nil { + self.messages[id].insert(wmsg) } + // todo: review after api is ready }, } - id = s.w.Watch(filter) + if len(args.To) == 0 && len(args.KeySym) == 0 { + info := "Filter must contain at least one key" + glog.V(logger.Error).Infof(info) + return nil, errors.New(info) + } - s.messagesMu.Lock() - s.messages[id] = newWhisperFilter(id, s.w) - s.messagesMu.Unlock() + if len(args.Topics) > 0 { + for _, s := range args.Topics { + t := common.FromHex(s) + filter.Topics = append(filter.Topics, BytesToTopic(t)) + } + } else { + // if Topics are not provided, just use the default derivation function + t := DeriveTopicFromSymmetricKey(common.FromHex(args.KeySym)) + filter.Topics = append(filter.Topics, t) + } + + id = self.whisper.Watch(&filter) + + self.messagesMu.Lock() + self.messages[id] = newWhisperFilter(id, self.whisper) + self.messagesMu.Unlock() return rpc.NewHexNumber(id), nil } -// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval. -func (s *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage { - s.messagesMu.RLock() - defer s.messagesMu.RUnlock() - - if s.messages[filterId.Int()] != nil { - if changes := s.messages[filterId.Int()].retrieve(); changes != nil { - return changes - } - } - return returnWhisperMessages(nil) -} - // UninstallFilter disables and removes an existing filter. -func (s *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) bool { - s.messagesMu.Lock() - defer s.messagesMu.Unlock() +func (self *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) bool { + self.messagesMu.Lock() + defer self.messagesMu.Unlock() - if _, ok := s.messages[filterId.Int()]; ok { - delete(s.messages, filterId.Int()) + if _, ok := self.messages[filterId.Int()]; ok { + delete(self.messages, filterId.Int()) return true } return false } -// GetMessages retrieves all the known messages that match a specific filter. -func (s *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage { - // Retrieve all the cached messages matching a specific, existing filter - s.messagesMu.RLock() - defer s.messagesMu.RUnlock() +// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval. +func (self *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage { + self.messagesMu.RLock() + defer self.messagesMu.RUnlock() - var messages []*Message - if s.messages[filterId.Int()] != nil { - messages = s.messages[filterId.Int()].messages() + if self.messages[filterId.Int()] != nil { + if changes := self.messages[filterId.Int()].retrieve(); changes != nil { + return changes + } } - - return returnWhisperMessages(messages) + return toWhisperMessages(nil) } -// returnWhisperMessages converts a Whisper message to a RPC whisper message. -func returnWhisperMessages(messages []*Message) []WhisperMessage { +// GetMessages retrieves all the known messages that match a specific filter. +func (self *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage { + // Retrieve all the cached messages matching a specific, existing filter + self.messagesMu.RLock() + defer self.messagesMu.RUnlock() + + var messages []*ReceivedMessage + if self.messages[filterId.Int()] != nil { + messages = self.messages[filterId.Int()].messages() + } + + return toWhisperMessages(messages) +} + +// toWhisperMessages converts a Whisper message to a RPC whisper message. +func toWhisperMessages(messages []*ReceivedMessage) []WhisperMessage { msgs := make([]WhisperMessage, len(messages)) for i, msg := range messages { msgs[i] = NewWhisperMessage(msg) @@ -171,96 +177,131 @@ func returnWhisperMessages(messages []*Message) []WhisperMessage { return msgs } -type PostArgs struct { - From string `json:"from"` - To string `json:"to"` - Topics [][]byte `json:"topics"` - Payload string `json:"payload"` - Priority int64 `json:"priority"` - TTL int64 `json:"ttl"` -} - // Post injects a message into the whisper network for distribution. -func (s *PublicWhisperAPI) Post(args PostArgs) (bool, error) { - if s.w == nil { +func (self *PublicWhisperAPI) Post(args PostArgs) (bool, error) { + if self.whisper == nil { return false, whisperOffLineErr } // construct whisper message with transmission options - message := NewMessage(common.FromHex(args.Payload)) + message := NewSentMessage(common.FromHex(args.Payload)) options := Options{ - To: crypto.ToECDSAPub(common.FromHex(args.To)), TTL: time.Duration(args.TTL) * time.Second, - Topics: NewTopics(args.Topics...), + Dst: crypto.ToECDSAPub(common.FromHex(args.To)), + KeySym: common.FromHex(args.Key), + Topic: BytesToTopic(common.FromHex(args.Topic)), + Pad: common.FromHex(args.Padding), } // set sender identity if len(args.From) > 0 { - if key := s.w.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); key != nil { - options.From = key + if privateKey := self.whisper.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); privateKey != nil { + options.Src = privateKey } else { return false, fmt.Errorf("unknown identity to send from: %s", args.From) } } // Wrap and send the message - pow := time.Duration(args.Priority) * time.Millisecond - envelope, err := message.Wrap(pow, options) + options.Work = time.Duration(options.Work) * time.Millisecond + envelope, err := message.Wrap(options) if err != nil { return false, err } - return true, s.w.Send(envelope) + return true, self.whisper.Send(envelope) } -// WhisperMessage is the RPC representation of a whisper message to be sent. -type WhisperMessage struct { - ref *Message - - Payload string `json:"payload"` - To string `json:"to"` - From string `json:"from"` - Sent int64 `json:"sent"` +type PostArgs struct { TTL int64 `json:"ttl"` - Hash string `json:"hash"` + From string `json:"from"` + To string `json:"to"` + Key string `json:"key"` + Topic string `json:"topic"` + Padding string `json:"padding"` + Payload string `json:"payload"` + Work int64 `json:"work"` // todo: review this field usage + PoW int64 `json:"pow"` // todo: review this field usage } func (args *PostArgs) UnmarshalJSON(data []byte) (err error) { var obj struct { - From string `json:"from"` - To string `json:"to"` - Topics []string `json:"topics"` - Payload string `json:"payload"` - Priority rpc.HexNumber `json:"priority"` - TTL rpc.HexNumber `json:"ttl"` + TTL int64 `json:"ttl"` + From string `json:"from"` + To string `json:"to"` + Key string `json:"key"` + Topic string `json:"topic"` + Payload string `json:"payload"` + Padding string `json:"padding"` + Work int64 `json:"work"` + PoW int64 `json:"pow"` } if err := json.Unmarshal(data, &obj); err != nil { return err } + args.TTL = obj.TTL args.From = obj.From args.To = obj.To + args.Key = obj.Key + args.Topic = obj.Topic args.Payload = obj.Payload - args.Priority = obj.Priority.Int64() - args.TTL = obj.TTL.Int64() - - // decode topic strings - args.Topics = make([][]byte, len(obj.Topics)) - for i, topic := range obj.Topics { - args.Topics[i] = common.FromHex(topic) - } + args.Padding = obj.Padding + args.Work = obj.Work + args.PoW = obj.PoW return nil } +// WhisperMessage is the RPC representation of a whisper message. +type WhisperMessage struct { + ref *ReceivedMessage + + Payload string `json:"payload"` + Padding string `json:"padding"` + From string `json:"from"` + To string `json:"to"` + Sent int64 `json:"sent"` + TTL int64 `json:"ttl"` + PoW int64 `json:"pow"` + Hash string `json:"hash"` +} + +// NewWhisperMessage converts an internal message into an API version. +func NewWhisperMessage(message *ReceivedMessage) WhisperMessage { + return WhisperMessage{ + ref: message, + + Payload: common.ToHex(message.Payload), + Padding: common.ToHex(message.Padding), + From: common.ToHex(crypto.FromECDSAPub(message.Recover())), + To: common.ToHex(crypto.FromECDSAPub(message.Dst)), + Sent: int64(message.Sent), // todo: review format + TTL: int64(message.TTL), // todo: review format + PoW: int64(message.PoW), + Hash: common.ToHex(message.EnvelopeHash.Bytes()), + } +} + +type WhisperFilterArgs struct { + // todo: review types + To string + From string + KeySym string + PoW int + Topics []string +} + // UnmarshalJSON implements the json.Unmarshaler interface, invoked to convert a // JSON message blob into a WhisperFilterArgs structure. -func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { +func (args *WhisperFilterArgs) UnmarshalJSON(b []byte) (err error) { // Unmarshal the JSON message and sanity check var obj struct { To interface{} `json:"to"` From interface{} `json:"from"` + Key interface{} `json:"key"` + PoW interface{} `json:"pow"` Topics interface{} `json:"topics"` } if err := json.Unmarshal(b, &obj); err != nil { @@ -273,7 +314,7 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { } else { argstr, ok := obj.To.(string) if !ok { - return fmt.Errorf("to is not a string") + return fmt.Errorf("'to' is not a string") } args.To = argstr } @@ -282,11 +323,33 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { } else { argstr, ok := obj.From.(string) if !ok { - return fmt.Errorf("from is not a string") + return fmt.Errorf("'from' is not a string") } args.From = argstr } - // Construct the nested topic array + if obj.Key == nil { + args.KeySym = "" + } else { + argstr, ok := obj.Key.(string) + if !ok { + return fmt.Errorf("'key' is not a string") + } + args.KeySym = argstr + } + if obj.PoW == nil { + args.PoW = 0 + } else { + argstr, ok := obj.PoW.(string) + if !ok { + return fmt.Errorf("'pow' is not a string") + } + x, err := strconv.Atoi(argstr) + if err != nil { + return fmt.Errorf("'pow' is invalid") + } + args.PoW = x + } + // Construct the topic array if obj.Topics != nil { // Make sure we have an actual topic array list, ok := obj.Topics.([]interface{}) @@ -294,43 +357,25 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { return fmt.Errorf("topics is not an array") } // Iterate over each topic and handle nil, string or array - topics := make([][]string, len(list)) - for idx, field := range list { + topics := make([]string, len(list)) + for i, field := range list { switch value := field.(type) { case nil: - topics[idx] = []string{} - + topics[i] = "" case string: - topics[idx] = []string{value} - - case []interface{}: - topics[idx] = make([]string, len(value)) - for i, nested := range value { - switch value := nested.(type) { - case nil: - topics[idx][i] = "" - - case string: - topics[idx][i] = value - - default: - return fmt.Errorf("topic[%d][%d] is not a string", idx, i) - } - } + topics[i] = value default: - return fmt.Errorf("topic[%d] not a string or array", idx) + return fmt.Errorf("topic[%d] is not a string", i) } } - topicsDecoded := make([][][]byte, len(topics)) - for i, condition := range topics { - topicsDecoded[i] = make([][]byte, len(condition)) - for j, topic := range condition { - topicsDecoded[i][j] = common.FromHex(topic) - } - } - - args.Topics = topicsDecoded + // todo: delete this block + //topicsDecoded := make([][]byte, len(topics)) + //for j, t := range topics { + // topicsDecoded[j] = common.FromHex(t) + //} + //args.Topics = topicsDecoded + args.Topics = topics } return nil } @@ -338,8 +383,8 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { // whisperFilter is the message cache matching a specific filter, accumulating // inbound messages until the are requested by the client. type whisperFilter struct { - id int // Filter identifier for old message retrieval - ref *Whisper // Whisper reference for old message retrieval + id int // Filter identifier for old message retrieval + whisper *Whisper // Whisper reference for old message retrieval cache []WhisperMessage // Cache of messages not yet polled skip map[common.Hash]struct{} // List of retrieved messages to avoid duplication @@ -348,77 +393,49 @@ type whisperFilter struct { lock sync.RWMutex // Lock protecting the filter internals } +// newWhisperFilter creates a new serialized, poll based whisper topic filter. +func newWhisperFilter(id int, w *Whisper) *whisperFilter { + return &whisperFilter{ + id: id, + whisper: w, + update: time.Now(), + skip: make(map[common.Hash]struct{}), + } +} + // messages retrieves all the cached messages from the entire pool matching the // filter, resetting the filter's change buffer. -func (w *whisperFilter) messages() []*Message { - w.lock.Lock() - defer w.lock.Unlock() +func (filter *whisperFilter) messages() []*ReceivedMessage { + filter.lock.Lock() + defer filter.lock.Unlock() - w.cache = nil - w.update = time.Now() + filter.cache = nil + filter.update = time.Now() - w.skip = make(map[common.Hash]struct{}) - messages := w.ref.Messages(w.id) + filter.skip = make(map[common.Hash]struct{}) + messages := filter.whisper.Messages(filter.id) for _, message := range messages { - w.skip[message.Hash] = struct{}{} + filter.skip[message.EnvelopeHash] = struct{}{} } return messages } // insert injects a new batch of messages into the filter cache. -func (w *whisperFilter) insert(messages ...WhisperMessage) { - w.lock.Lock() - defer w.lock.Unlock() +func (filter *whisperFilter) insert(message WhisperMessage) { + filter.lock.Lock() + defer filter.lock.Unlock() - for _, message := range messages { - if _, ok := w.skip[message.ref.Hash]; !ok { - w.cache = append(w.cache, messages...) - } + if _, ok := filter.skip[message.ref.EnvelopeHash]; !ok { + filter.cache = append(filter.cache, message) } } // retrieve fetches all the cached messages from the filter. -func (w *whisperFilter) retrieve() (messages []WhisperMessage) { - w.lock.Lock() - defer w.lock.Unlock() - - messages, w.cache = w.cache, nil - w.update = time.Now() +func (filter *whisperFilter) retrieve() (messages []WhisperMessage) { + filter.lock.Lock() + defer filter.lock.Unlock() + messages, filter.cache = filter.cache, nil + filter.update = time.Now() return } - -// activity returns the last time instance when client requests were executed on -// the filter. -func (w *whisperFilter) activity() time.Time { - w.lock.RLock() - defer w.lock.RUnlock() - - return w.update -} - -// newWhisperFilter creates a new serialized, poll based whisper topic filter. -func newWhisperFilter(id int, ref *Whisper) *whisperFilter { - return &whisperFilter{ - id: id, - ref: ref, - - update: time.Now(), - skip: make(map[common.Hash]struct{}), - } -} - -// NewWhisperMessage converts an internal message into an API version. -func NewWhisperMessage(message *Message) WhisperMessage { - return WhisperMessage{ - ref: message, - - Payload: common.ToHex(message.Payload), - From: common.ToHex(crypto.FromECDSAPub(message.Recover())), - To: common.ToHex(crypto.FromECDSAPub(message.To)), - Sent: message.Sent.Unix(), - TTL: int64(message.TTL / time.Second), - Hash: common.ToHex(message.Hash.Bytes()), - } -} -*/ diff --git a/whisper5/doc.go b/whisper5/doc.go index 351b8e8aa1..a3cefa4e88 100644 --- a/whisper5/doc.go +++ b/whisper5/doc.go @@ -31,7 +31,11 @@ particularly the notion of singular endpoints. */ package whisper5 -import "time" +import ( + "time" + + "github.com/ethereum/go-ethereum/common" +) const ( statusCode = 0x00 @@ -62,3 +66,21 @@ const ( // classifications of a message, determined as the first (left) 4 bytes of the // SHA3 hash of some arbitrary data given by the original author of the message. type TopicType [4]byte + +func BytesToTopic(b []byte) (t TopicType) { + sz := 4 + if x := len(b); x < 4 { + sz = x + } + for i := 0; i < sz; i++ { + t[i] = b[i] + } + return t +} + +func HashToTopic(h common.Hash) (t TopicType) { + for i := 0; i < 4; i++ { + t[i] = h[i] + } + return t +} diff --git a/whisper5/envelope.go b/whisper5/envelope.go index b282618e9a..a35f507618 100644 --- a/whisper5/envelope.go +++ b/whisper5/envelope.go @@ -70,14 +70,14 @@ func (self *Envelope) isAsymmetric() bool { // Seal closes the envelope by spending the requested amount of time as a proof // of work on hashing the data. -func (self *Envelope) Seal(dur time.Duration) { - self.Expiry += uint32(dur.Seconds()) // adjust for the duration of Seal() execution +func (self *Envelope) Seal(work time.Duration) { + self.Expiry += uint32(work.Seconds()) // adjust for the duration of Seal() execution buf := make([]byte, 64) h := crypto.Keccak256(self.rlpWithoutNonce()) copy(buf[:32], h) - finish, bestBit := time.Now().Add(dur).UnixNano(), 0 + finish, bestBit := time.Now().Add(work).UnixNano(), 0 for nonce := uint64(0); time.Now().UnixNano() < finish; { for i := 0; i < 1024; i++ { binary.BigEndian.PutUint64(buf[56:], nonce) @@ -150,14 +150,30 @@ func (self *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error return } -// Open tries to decrypt an envelope -func (self *Envelope) Open(watcher *Filter) *ReceivedMessage { +// 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) - return msg + msg, _ = self.OpenAsymmetric(watcher.KeyAsym) + if msg != nil { + msg.Dst = watcher.Dst + } } else if self.isSymmetric() { - msg, _ := self.OpenSymmetric(watcher.KeySym) - return msg + msg, _ = self.OpenSymmetric(watcher.KeySym) + if msg != nil { + msg.TopicKeyHash = crypto.Keccak256Hash(watcher.KeySym) + } } - return nil + + if msg != nil { + ok := msg.Validate() + if !ok { + return nil + } + msg.Topic = self.Topic + msg.PoW = self.pow + msg.TTL = self.TTL + msg.Sent = self.Expiry - self.TTL // todo: review + msg.EnvelopeHash = self.hash + } + return msg } diff --git a/whisper5/filter.go b/whisper5/filter.go index 2d3b7798ce..1f6289ca5b 100644 --- a/whisper5/filter.go +++ b/whisper5/filter.go @@ -1,19 +1,18 @@ package whisper5 import ( - "bytes" "crypto/ecdsa" -) -var empty = TopicType{0, 0, 0, 0} + "github.com/ethereum/go-ethereum/common" +) type Filter struct { Src *ecdsa.PublicKey // Sender of the message Dst *ecdsa.PublicKey // Recipient of the message KeyAsym *ecdsa.PrivateKey // Private Key of recipient - Topic TopicType // Topics to filter messages with + Topics []TopicType // Topics to filter messages with KeySym []byte // Key associated with the Topic - TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + TopicKeyHash common.Hash // The Keccak256Hash of the symmetric key PoW int // Proof of work as described in the Whisper spec Fn func(msg *ReceivedMessage) // Handler in case of a match } @@ -23,13 +22,15 @@ type Filters struct { watchers map[int]*Filter ch chan Envelope quit chan struct{} + whisper *Whisper } -func NewFilters() *Filters { +func NewFilters(w *Whisper) *Filters { return &Filters{ ch: make(chan Envelope), watchers: make(map[int]*Filter), quit: make(chan struct{}), + whisper: w, } } @@ -80,7 +81,7 @@ func (self *Filters) processEnvelope(envelope *Envelope) { } else { match = watcher.MatchEnvelope(envelope) if match { - msg = envelope.Open(watcher) // todo: fill all the fields & validate + msg = envelope.Open(watcher) } } @@ -88,18 +89,23 @@ func (self *Filters) processEnvelope(envelope *Envelope) { watcher.Trigger(msg) } } + + if msg != nil { + go self.whisper.addDecryptedMessage(msg) + } } -func (self Filter) expectsPublicKeyEncryption() bool { +func (self Filter) expectsAsymmetricEncryption() bool { return self.KeyAsym != nil } -func (self Filter) expectsTopicEncryption() bool { +func (self Filter) expectsSymmetricEncryption() bool { return self.KeySym != nil } func (self Filter) Trigger(msg *ReceivedMessage) { - go self.Fn(msg) // todo: review + // todo: save msg hash in the filter + self.Fn(msg) } func (self Filter) MatchMessage(msg *ReceivedMessage) bool { @@ -107,14 +113,23 @@ func (self Filter) MatchMessage(msg *ReceivedMessage) bool { return false } - if self.expectsPublicKeyEncryption() && msg.isAsymmetric() { + if self.Src != nil && msg.Src != self.Src { + return false + } + + if self.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() { return self.Dst == msg.Dst - } else if self.expectsTopicEncryption() && msg.isSymmetric() { + } else if self.expectsSymmetricEncryption() && msg.isSymmetricEncryption() { // we need to compare the keys (or rather thier hashes), because of // possible collision (different keys can produce the same topic). // we also need to compare the topics, because they could be arbitrary (not related to KeySym). - if self.Topic == msg.Topic && bytes.Equal(self.TopicKeyHash, msg.TopicKeyHash) { - return true + if self.TopicKeyHash == msg.TopicKeyHash { + for _, t := range self.Topics { + if t == msg.Topic { + return true + } + } + return false } } return false @@ -126,16 +141,22 @@ func (self Filter) MatchEnvelope(envelope *Envelope) bool { } encryptionMethodMatch := false - if self.expectsPublicKeyEncryption() && envelope.isAsymmetric() { + if self.expectsAsymmetricEncryption() && envelope.isAsymmetric() { encryptionMethodMatch = true - } else if self.expectsTopicEncryption() && envelope.isSymmetric() { + if self.Topics == nil { + return true // wildcard + } + } else if self.expectsSymmetricEncryption() && envelope.isSymmetric() { encryptionMethodMatch = true } if encryptionMethodMatch { - if self.Topic == empty || self.Topic == envelope.Topic { - return true + for _, t := range self.Topics { + if t == envelope.Topic { + return true + } } + return false } return false } diff --git a/whisper5/message.go b/whisper5/message.go index 93c842cddf..22066b0310 100644 --- a/whisper5/message.go +++ b/whisper5/message.go @@ -16,7 +16,7 @@ // Contains the Whisper protocol Message element. For formal details please see // the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#messages. -// todo: fix the spec link +// todo: fix the spec link, and move it to doc.go package whisper5 @@ -39,13 +39,14 @@ import ( // Options specifies the exact way a message should be wrapped into an Envelope. type Options struct { - Topic TopicType - TTL time.Duration - Src *ecdsa.PrivateKey - Dst *ecdsa.PublicKey - Key []byte // must be 32 bytes. todo: review - Salt []byte - Pad []byte + TTL time.Duration + Src *ecdsa.PrivateKey + Dst *ecdsa.PublicKey + KeySym []byte // must be 32 bytes. todo: review + Topic TopicType + Pad []byte + Work time.Duration + PoW int } // SentMessage represents an end-user data packet to transmit through the @@ -65,15 +66,21 @@ type ReceivedMessage struct { Signature []byte PoW int // Proof of work as described in the Whisper spec - Sent time.Time // Time when the message was posted into the network - TTL time.Duration // Maximum time to live allowed for the message + Sent uint32 // Time when the message was posted into the network + TTL uint32 // Maximum time to live allowed for the message Src *ecdsa.PublicKey // Message recipient (identity used to decode the message) Dst *ecdsa.PublicKey // Message recipient (identity used to decode the message) Topic TopicType - TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + TopicKeyHash common.Hash // The Keccak256Hash of the key, associated with the Topic EnvelopeHash common.Hash // Message envelope hash to act as a unique id } +func DeriveTopicFromSymmetricKey(key []byte) TopicType { + // todo: it is not secure enough, use kdf instead + hash := crypto.Keccak256Hash(key) + return HashToTopic(hash) +} + func isMessageSigned(flags byte) bool { return (flags & signatureFlag) != 0 } @@ -82,11 +89,11 @@ func isMessagePadded(flags byte) bool { return (flags & paddingFlag) != 0 } -func (self *ReceivedMessage) isSymmetric() bool { - return self.TopicKeyHash != nil +func (self *ReceivedMessage) isSymmetricEncryption() bool { + return self.TopicKeyHash != common.Hash{} } -func (self *ReceivedMessage) isAsymmetric() bool { +func (self *ReceivedMessage) isAsymmetricEncryption() bool { return self.Dst != nil } @@ -206,11 +213,10 @@ 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(pow time.Duration, options Options) (envelope *Envelope, err error) { +func (self *SentMessage) Wrap(options Options) (envelope *Envelope, err error) { if options.TTL == 0 { options.TTL = DefaultTTL } - //self.TTL = options.TTL // todo: review self.appendPadding(options) if options.Src != nil { if err = self.sign(options.Src); err != nil { @@ -225,15 +231,19 @@ func (self *SentMessage) Wrap(pow time.Duration, options Options) (envelope *Env var salt, nonce []byte if options.Dst != nil { err = self.encryptAsymmetric(options.Dst) - } else if options.Key != nil { - salt, nonce, err = self.encryptSymmetric(options.Key) + } else if options.KeySym != nil { + salt, nonce, err = self.encryptSymmetric(options.KeySym) } else { err = errors.New("Unable to encrypt the message: neither Dst nor Key") } if err == nil { + if (options.Topic == TopicType{}) { + options.Topic = DeriveTopicFromSymmetricKey(options.KeySym) + } + envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, self) - envelope.Seal(pow) + envelope.Seal(options.Work) // todo: use options.pow, review Seal() as well } return } @@ -305,7 +315,7 @@ func (self *ReceivedMessage) Validate() bool { } self.Payload = self.Raw[1:cur] - if self.isSymmetric() == self.isAsymmetric() { + if self.isSymmetricEncryption() == self.isAsymmetricEncryption() { return false } return true @@ -331,104 +341,3 @@ func (self *ReceivedMessage) hash() []byte { } return crypto.Keccak256(self.Raw) } - -// todo: delete this stuff -/* -// Signature returns the signature part of the raw message. -func (self *ReceivedMessage) ExtractSignature() { - if self.Signature == nil { - if sz := len(self.Raw); sz >= signatureLength+1 { - if isMessageSigned(self.Raw[0]) { - self.Signature = self.Raw[sz-signatureLength:] - } - } - } -} - -// Payload returns the payload part of the raw message. -func (self *ReceivedMessage) ExtractPayload() { - if self.Payload == nil { - end := len(self.Raw) - if isMessageSigned(self.Raw[0]) { - end -= signatureLength - if end <= 1 { - return - } - } - if isMessagePadded(self.Raw[0]) { - paddingSize := int(self.Raw[end-1]) - end -= paddingSize - if end <= 1 { - return - } - } - self.Payload = self.Raw[1:end] - } -} - -// Padding returns the padding part of the raw message -// without the last byte (which only contains the padding size). -func (self *ReceivedMessage) ExtractPadding() { - if self.Padding == nil { - end := len(self.Raw) - if isMessagePadded(self.Raw[0]) { - if isMessageSigned(self.Raw[0]) { - end -= signatureLength - if end <= 1 { - return - } - } - paddingSize := int(self.Raw[end-1]) - beg := end - paddingSize - if beg > 1 { - self.Padding = self.Raw[beg : end-1] - } - } - } -} -*/ -/* -// Signature returns the signature part of the raw message. -func (self *ReceivedMessage) ExtractSignature() []byte { - sz := len(self.Raw) - if self.isSigned() && sz >= signatureLength+1 { - return self.Raw[sz-signatureLength:] - } else { - return nil - } -} - -// Payload returns the payload part of the raw message. -func (self *ReceivedMessage) ExtractPayload() []byte { - end := len(self.Raw) - if self.isSigned() { - end -= signatureLength - } - if self.isPadded() { - paddingSize := int(self.Raw[end-1]) - end -= paddingSize - } - if end <= 1 { - return nil - } - return self.Raw[1:end] -} - -// Padding returns the padding part of the raw message -// without the last byte (which only contains the padding size). -func (self *ReceivedMessage) ExtractPadding() []byte { - if !self.isPadded() { - return nil - } - end := len(self.Raw) - if self.isSigned() { - end -= signatureLength - } - paddingSize := int(self.Raw[end-1]) - beg := end - paddingSize - if beg <= 1 { - return nil - } - return self.Raw[beg : end-1] -} -*/ diff --git a/whisper5/whisper.go b/whisper5/whisper.go index 3a6f1885ba..1e24a3893c 100644 --- a/whisper5/whisper.go +++ b/whisper5/whisper.go @@ -31,12 +31,6 @@ import ( set "gopkg.in/fatih/set.v0" ) -//type MessageEvent struct { -// To *ecdsa.PrivateKey -// From *ecdsa.PublicKey -// Message *Message -//} - // Whisper represents a dark communication interface through the Ethereum // network, using its very own P2P communication layer. type Whisper struct { @@ -44,10 +38,10 @@ type Whisper struct { filters *Filters privateKeys map[string]*ecdsa.PrivateKey - topicKeys map[TopicType][]byte // todo: move to the filter. this is not suitable because of possible collisions + //topicKeys map[TopicType][]byte // todo: move to the filter. this is not suitable because of possible collisions - msgs map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages // todo: rename envelopes map[common.Hash]*Envelope // Pool of messages currently tracked by this node + messages map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages expirations map[uint32]*set.SetNonTS // Message expiration pool (TODO: something lighter) poolMu sync.RWMutex // Mutex to sync the message and expiration pools @@ -60,14 +54,16 @@ type Whisper struct { // New creates a Whisper client ready to communicate through the Ethereum P2P network. func NewWhisper() *Whisper { whisper := &Whisper{ - filters: NewFilters(), + //filters: NewFilters(), privateKeys: make(map[string]*ecdsa.PrivateKey), - topicKeys: make(map[TopicType][]byte), + //topicKeys: make(map[TopicType][]byte), envelopes: make(map[common.Hash]*Envelope), + messages: make(map[common.Hash]*ReceivedMessage), expirations: make(map[uint32]*set.SetNonTS), peers: make(map[*peer]struct{}), quit: make(chan struct{}), } + whisper.filters = NewFilters(whisper) whisper.filters.Start() // p2p whisper sub protocol handler @@ -103,16 +99,15 @@ func (self *Whisper) Version() uint { return self.protocol.Version } +// todo: review. maybe we need to delete these identity-related functions, since the key moved to Filter // 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 { - // todo: review key, err := crypto.GenerateKey() if err != nil { panic(err) } self.privateKeys[string(crypto.FromECDSAPub(&key.PublicKey))] = key - return key } @@ -245,27 +240,6 @@ func (self *Whisper) postEvent(envelope *Envelope) { self.filters.Notify(envelope) } -/* -// createFilter creates a message filter to check against installed handlers. -func createFilter(message *Message, topics []TopicType) filter.Filter { - //return Filter{ - // Src: string(crypto.FromECDSAPub(message.Recover())), - // Dst: string(crypto.FromECDSAPub(message.Dst)), - // Topics: topics, - //} - - matcher := make([][]TopicType, len(topics)) - for i, topic := range topics { - matcher[i] = []TopicType{topic} - } - return filterer{ - to: string(crypto.FromECDSAPub(message.To)), - from: string(crypto.FromECDSAPub(message.Recover())), - matcher: newTopicMatcher(matcher...), - } -} -*/ - // update loops until the lifetime of the whisper node, updating its internal // state by expiring stale messages from the pool. func (self *Whisper) update() { @@ -299,6 +273,7 @@ func (self *Whisper) expire() { // 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)) return true }) self.expirations[then].Clear() @@ -317,33 +292,25 @@ func (self *Whisper) Envelopes() []*Envelope { return all } -/* // Messages retrieves all the currently pooled messages matching a filter id. -// todo: review -//func (self *Whisper) Messages(id int) []*Message { -// messages := make([]*Message, 0) -// if filter := self.filters.Get(id); filter != nil { -// for _, envelope := range self.messages { -// if message := self.open(envelope); message != nil { -// if self.filters.Match(filter, createFilter(message, envelope.Topic)) { -// messages = append(messages, message) -// } -// } -// } -// } -// return messages -//} func (self *Whisper) Messages(id int) []*ReceivedMessage { - messages := make([]*Envelope, 0) + self.poolMu.RLock() + defer self.poolMu.RUnlock() + + result := make([]*ReceivedMessage, 0) if filter := self.filters.Get(id); filter != nil { - for _, envelope := range self.envelopes { - //if message := self.open(envelope); message != nil { - if self.filters.Match(envelope) { - messages = append(messages, envelope) + for _, msg := range self.messages { + if filter.MatchMessage(msg) { + result = append(result, msg) } - //} } } - return messages + return result } -*/ \ No newline at end of file + +func (self *Whisper) addDecryptedMessage(msg *ReceivedMessage) { + self.poolMu.Lock() + defer self.poolMu.Unlock() + + self.messages[msg.EnvelopeHash] = msg +} \ No newline at end of file