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

whisper: logging fixed

Browse source
This commit is contained in:
Vlad 2017-03-29 20:36:21 +02:00
parent 87e00e508b
commit 1b6c7c8637
5 changed files with 42 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
cmd/wnode/main.go
View file

@ -627,8 +627,7 @@ func requestExpiredMessagesLoop() {
msg := whisper.NewSentMessage(&params) msg := whisper.NewSentMessage(&params)
if msg == nil { if msg == nil {
fmt.Printf("failed to create new message (OS level error)") utils.Fatalf("failed to create new message (OS level error)")
os.Exit(0)
} }
env, err := msg.Wrap(&params) env, err := msg.Wrap(&params)
if err != nil { if err != nil {
@ -648,7 +647,6 @@ func extractIdFromEnode(s string) []byte {
n, err := discover.ParseNode(s) n, err := discover.ParseNode(s)
if err != nil { if err != nil {
utils.Fatalf("Failed to parse enode: %s", err) utils.Fatalf("Failed to parse enode: %s", err)
return nil
} }
return n.ID[:] return n.ID[:]
} }

12
whisper/whisperv5/filter.go
View file

@ -110,15 +110,15 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) {
if match { if match {
msg = env.Open(watcher) msg = env.Open(watcher)
if msg == nil { if msg == nil {
log.Trace(fmt.Sprintf("msg [%x], filter [%d]: failed to open", env.Hash(), i)) log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", i)
} }
} else { } else {
log.Trace(fmt.Sprintf("msg [%x], filter [%d]: does not match", env.Hash(), i)) log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", i)
} }
} }
if match && msg != nil { if match && msg != nil {
log.Trace(fmt.Sprintf("message decrypted [%x]", env.Hash())) log.Trace("processing message: decrypted", "hash", env.Hash().Hex())
watcher.Trigger(msg) watcher.Trigger(msg)
} }
} }
@ -130,10 +130,10 @@ func (f *Filter) processEnvelope(env *Envelope) *ReceivedMessage {
if msg != nil { if msg != nil {
return msg return msg
} else { } else {
log.Trace(fmt.Sprintf("processing msg [%x]: failed to open", env.Hash())) log.Trace("processing envelope: failed to open", "hash", env.Hash().Hex())
} }
} else { } else {
log.Trace(fmt.Sprintf("processing msg [%x]: does not match", env.Hash())) log.Trace("processing envelope: does not match", "hash", env.Hash().Hex())
} }
return nil return nil
} }
@ -229,6 +229,6 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool {
} else if !ValidatePublicKey(b) { } else if !ValidatePublicKey(b) {
return false return false
} }
// the Curve is always the same, just compare the points // the curve is always the same, just compare the points
return a.X.Cmp(b.X) == 0 && a.Y.Cmp(b.Y) == 0 return a.X.Cmp(b.X) == 0 && a.Y.Cmp(b.Y) == 0
} }

14
whisper/whisperv5/message.go
View file

@ -25,7 +25,6 @@ import (
crand "crypto/rand" crand "crypto/rand"
"crypto/sha256" "crypto/sha256"
"errors" "errors"
"fmt"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
@ -103,7 +102,7 @@ func NewSentMessage(params *MessageParams) *SentMessage {
msg.Raw[0] = 0 // set all the flags to zero msg.Raw[0] = 0 // set all the flags to zero
err := msg.appendPadding(params) err := msg.appendPadding(params)
if err != nil { if err != nil {
log.Error(fmt.Sprintf("NewSentMessage: %s", err)) log.Error("failed to create NewSentMessage", "err", err)
return nil return nil
} }
msg.Raw = append(msg.Raw, params.Payload...) msg.Raw = append(msg.Raw, params.Payload...)
@ -150,7 +149,7 @@ func (msg *SentMessage) appendPadding(params *MessageParams) error {
func (msg *SentMessage) sign(key *ecdsa.PrivateKey) error { func (msg *SentMessage) sign(key *ecdsa.PrivateKey) error {
if isMessageSigned(msg.Raw[0]) { if isMessageSigned(msg.Raw[0]) {
// this should not happen, but no reason to panic // this should not happen, but no reason to panic
log.Error(fmt.Sprintf("Trying to sign a message which was already signed")) log.Error("failed to sign the message: already signed")
return nil return nil
} }
@ -168,7 +167,7 @@ func (msg *SentMessage) sign(key *ecdsa.PrivateKey) error {
// encryptAsymmetric encrypts a message with a public key. // encryptAsymmetric encrypts a message with a public key.
func (msg *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error { func (msg *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error {
if !ValidatePublicKey(key) { if !ValidatePublicKey(key) {
return fmt.Errorf("invalid public key provided for asymmetric encryption") return errors.New("invalid public key provided for asymmetric encryption")
} }
encrypted, err := ecies.Encrypt(crand.Reader, ecies.ImportECDSAPublic(key), msg.Raw, nil, nil) encrypted, err := ecies.Encrypt(crand.Reader, ecies.ImportECDSAPublic(key), msg.Raw, nil, nil)
if err == nil { if err == nil {
@ -270,9 +269,8 @@ func (msg *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []by
return err return err
} }
if len(nonce) != aesgcm.NonceSize() { if len(nonce) != aesgcm.NonceSize() {
info := fmt.Sprintf("wrong AES nonce size - want: %d, got: %d", len(nonce), aesgcm.NonceSize()) log.Error("decrypting the message", "AES nonce size", len(nonce))
log.Error(info) return errors.New("wrong AES nonce size")
return errors.New(info)
} }
decrypted, err := aesgcm.Open(nil, nonce, msg.Raw, nil) decrypted, err := aesgcm.Open(nil, nonce, msg.Raw, nil)
if err != nil { if err != nil {
@ -342,7 +340,7 @@ func (msg *ReceivedMessage) SigToPubKey() *ecdsa.PublicKey {
pub, err := crypto.SigToPub(msg.hash(), msg.Signature) pub, err := crypto.SigToPub(msg.hash(), msg.Signature)
if err != nil { if err != nil {
log.Error(fmt.Sprintf("could not get public key from signature: %v", err)) log.Error("failed to recover public key from signature", "err", err)
return nil return nil
} }
return pub return pub

16
whisper/whisperv5/peer.go
View file

@ -55,13 +55,13 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
// into the network. // into the network.
func (p *Peer) start() { func (p *Peer) start() {
go p.update() go p.update()
log.Debug(fmt.Sprintf("%v: whisper started", p.peer)) log.Trace("start", "peer", p.ID())
} }
// stop terminates the peer updater, stopping message forwarding to it. // stop terminates the peer updater, stopping message forwarding to it.
func (p *Peer) stop() { func (p *Peer) stop() {
close(p.quit) close(p.quit)
log.Debug(fmt.Sprintf("%v: whisper stopped", p.peer)) log.Trace("stop", "peer", p.ID())
} }
// handshake sends the protocol initiation status message to the remote peer and // handshake sends the protocol initiation status message to the remote peer and
@ -78,19 +78,19 @@ func (p *Peer) handshake() error {
return err return err
} }
if packet.Code != statusCode { if packet.Code != statusCode {
return fmt.Errorf("peer sent %x before status packet", packet.Code) return fmt.Errorf("peer [%x] sent packet %x before status packet", p.ID(), packet.Code)
} }
s := rlp.NewStream(packet.Payload, uint64(packet.Size)) s := rlp.NewStream(packet.Payload, uint64(packet.Size))
peerVersion, err := s.Uint() peerVersion, err := s.Uint()
if err != nil { if err != nil {
return fmt.Errorf("bad status message: %v", err) return fmt.Errorf("peer [%x] sent bad status message: %v", p.ID(), err)
} }
if peerVersion != ProtocolVersion { if peerVersion != ProtocolVersion {
return fmt.Errorf("protocol version mismatch %d != %d", peerVersion, ProtocolVersion) return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", p.ID(), peerVersion, ProtocolVersion)
} }
// Wait until out own status is consumed too // Wait until out own status is consumed too
if err := <-errc; err != nil { if err := <-errc; err != nil {
return fmt.Errorf("failed to send status packet: %v", err) return fmt.Errorf("peer [%x] failed to send status packet: %v", p.ID(), err)
} }
return nil return nil
} }
@ -110,7 +110,7 @@ func (p *Peer) update() {
case <-transmit.C: case <-transmit.C:
if err := p.broadcast(); err != nil { if err := p.broadcast(); err != nil {
log.Info(fmt.Sprintf("%v: broadcast failed: %v", p.peer, err)) log.Trace("broadcast failed", "reason", err, "peer", p.ID())
return return
} }
@ -165,7 +165,7 @@ func (p *Peer) broadcast() error {
if err := p2p.Send(p.ws, messagesCode, transmit); err != nil { if err := p2p.Send(p.ws, messagesCode, transmit); err != nil {
return err return err
} }
log.Trace(fmt.Sprint(p.peer, "broadcasted", len(transmit), "message(s)")) log.Trace("broadcast", "num. messages", len(transmit))
return nil return nil
} }

39
whisper/whisperv5/whisper.go
View file

@ -31,6 +31,7 @@ import (
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
"github.com/syndtr/goleveldb/leveldb/errors"
"golang.org/x/crypto/pbkdf2" "golang.org/x/crypto/pbkdf2"
set "gopkg.in/fatih/set.v0" set "gopkg.in/fatih/set.v0"
) )
@ -403,7 +404,7 @@ func (w *Whisper) Send(envelope *Envelope) error {
// Start implements node.Service, starting the background data propagation thread // Start implements node.Service, starting the background data propagation thread
// of the Whisper protocol. // of the Whisper protocol.
func (w *Whisper) Start(*p2p.Server) error { func (w *Whisper) Start(*p2p.Server) error {
log.Info(fmt.Sprintf("Whisper v%d started", ProtocolVersion)) log.Info("started whisper v." + ProtocolVersionStr)
go w.update() go w.update()
numCPU := runtime.NumCPU() numCPU := runtime.NumCPU()
@ -418,7 +419,7 @@ func (w *Whisper) Start(*p2p.Server) error {
// of the Whisper protocol. // of the Whisper protocol.
func (w *Whisper) Stop() error { func (w *Whisper) Stop() error {
close(w.quit) close(w.quit)
log.Info(fmt.Sprint("Whisper stopped")) log.Info("whisper stopped")
return nil return nil
} }
@ -454,29 +455,31 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
// fetch the next packet // fetch the next packet
packet, err := rw.ReadMsg() packet, err := rw.ReadMsg()
if err != nil { if err != nil {
log.Warn("message loop", "peer", p.peer.ID(), "err", err)
return err return err
} }
if packet.Size > uint32(wh.maxMsgLength) { if packet.Size > uint32(wh.maxMsgLength) {
return fmt.Errorf("oversized message received") log.Warn("oversized message received", "peer", p.peer.ID())
return errors.New("oversized message received")
} }
switch packet.Code { switch packet.Code {
case statusCode: case statusCode:
// this should not happen, but no need to panic; just ignore this message. // this should not happen, but no need to panic; just ignore this message.
log.Warn(fmt.Sprintf("%v: unxepected status message received", p.peer)) log.Warn("unxepected status message received", "peer", p.peer.ID())
case messagesCode: case messagesCode:
// decode the contained envelopes // decode the contained envelopes
var envelopes []*Envelope var envelopes []*Envelope
if err := packet.Decode(&envelopes); err != nil { if err := packet.Decode(&envelopes); err != nil {
log.Warn(fmt.Sprintf("%v: failed to decode envelope: [%v], peer will be disconnected", p.peer, err)) log.Warn("failed to decode envelope, peer will be disconnected", "peer", p.peer.ID(), "err", err)
return fmt.Errorf("garbage received") return errors.New("invalid envelope")
} }
// inject all envelopes into the internal pool // inject all envelopes into the internal pool
for _, envelope := range envelopes { for _, envelope := range envelopes {
cached, err := wh.add(envelope) cached, err := wh.add(envelope)
if err != nil { if err != nil {
log.Warn(fmt.Sprintf("%v: bad envelope received: [%v], peer will be disconnected", p.peer, err)) log.Warn("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err)
return fmt.Errorf("invalid envelope") return errors.New("invalid envelope")
} }
if cached { if cached {
p.mark(envelope) p.mark(envelope)
@ -490,8 +493,8 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
if p.trusted { if p.trusted {
var envelope Envelope var envelope Envelope
if err := packet.Decode(&envelope); err != nil { if err := packet.Decode(&envelope); err != nil {
log.Warn(fmt.Sprintf("%v: failed to decode direct message: [%v], peer will be disconnected", p.peer, err)) log.Warn("failed to decode direct message, peer will be disconnected", "peer", p.peer.ID(), "err", err)
return fmt.Errorf("garbage received (directMessage)") return errors.New("invalid direct message")
} }
wh.postEvent(&envelope, true) wh.postEvent(&envelope, true)
} }
@ -500,8 +503,8 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
if wh.mailServer != nil { if wh.mailServer != nil {
var request Envelope var request Envelope
if err := packet.Decode(&request); err != nil { if err := packet.Decode(&request); err != nil {
log.Warn(fmt.Sprintf("%v: failed to decode p2p request message: [%v], peer will be disconnected", p.peer, err)) log.Warn("failed to decode p2p request message, peer will be disconnected", "peer", p.peer.ID(), "err", err)
return fmt.Errorf("garbage received (p2p request)") return errors.New("invalid p2p request")
} }
wh.mailServer.DeliverMail(p, &request) wh.mailServer.DeliverMail(p, &request)
} }
@ -534,7 +537,7 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
if envelope.Expiry+SynchAllowance*2 < now { if envelope.Expiry+SynchAllowance*2 < now {
return false, fmt.Errorf("very old message") return false, fmt.Errorf("very old message")
} else { } else {
log.Debug(fmt.Sprintf("expired envelope dropped [%x]", envelope.Hash())) log.Debug("expired envelope dropped", "hash", envelope.Hash().Hex())
return false, nil // drop envelope without error return false, nil // drop envelope without error
} }
} }
@ -558,7 +561,7 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
} }
if envelope.PoW() < wh.minPoW { if envelope.PoW() < wh.minPoW {
log.Debug(fmt.Sprintf("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash())) log.Debug("envelope with low PoW dropped", "PoW", envelope.PoW(), "hash", envelope.Hash().Hex())
return false, nil // drop envelope without error return false, nil // drop envelope without error
} }
@ -578,9 +581,9 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
wh.poolMu.Unlock() wh.poolMu.Unlock()
if alreadyCached { if alreadyCached {
log.Trace(fmt.Sprintf("whisper envelope already cached [%x]\n", envelope.Hash())) log.Trace("whisper envelope already cached", "hash", envelope.Hash().Hex())
} else { } else {
log.Trace(fmt.Sprintf("cached whisper envelope [%x]: %v\n", envelope.Hash(), envelope)) log.Trace("cached whisper envelope", "hash", envelope.Hash().Hex())
wh.stats.memoryUsed += envelope.size() wh.stats.memoryUsed += envelope.size()
wh.postEvent(envelope, false) // notify the local node about the new message wh.postEvent(envelope, false) // notify the local node about the new message
if wh.mailServer != nil { if wh.mailServer != nil {
@ -612,12 +615,12 @@ func (w *Whisper) checkOverflow() {
if queueSize == messageQueueLimit { if queueSize == messageQueueLimit {
if !w.overflow { if !w.overflow {
w.overflow = true w.overflow = true
log.Warn(fmt.Sprint("message queue overflow")) log.Warn("message queue overflow")
} }
} else if queueSize <= messageQueueLimit/2 { } else if queueSize <= messageQueueLimit/2 {
if w.overflow { if w.overflow {
w.overflow = false w.overflow = false
log.Warn(fmt.Sprint("message queue overflow fixed (back to normal)")) log.Warn("message queue overflow fixed (back to normal)")
} }
} }
} }