mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-25 22:26:42 +00:00
whisper5: added new files
This commit is contained in:
Vlad
parent
d14292e0ed
commit
baef246417
7 changed files with 1395 additions and 170 deletions
424
whisper5/api.go
Normal file
424
whisper5/api.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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()),
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
141
whisper5/filter.go
Normal file
141
whisper5/filter.go
Normal file
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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]
|
||||
}
|
||||
*/
|
||||
|
|
|
|||
175
whisper5/peer.go
Normal file
175
whisper5/peer.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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
|
||||
}
|
||||
38
whisper5/topic.go
Normal file
38
whisper5/topic.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
// 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[:])
|
||||
}
|
||||
349
whisper5/whisper.go
Normal file
349
whisper5/whisper.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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
|
||||
}
|
||||
*/
|
||||
Loading…
Reference in a new issue