From 2cbc96c0421d8b723a854ff0c813d74513d30d61 Mon Sep 17 00:00:00 2001 From: Guillaume Ballet Date: Wed, 11 Apr 2018 15:43:39 +0200 Subject: [PATCH] whisper: implement @fjl remarks wrt rlp encoding Notes: - There was a request for several timeouts, I still need to see how the system will react when they run out. - It is possible that the decoded palyload is bigger than the encoded payload, some ad-hoc function needs to be made for that. --- whisper/whisperv6/libp2p_glue.go | 92 +++++++++++--------------------- 1 file changed, 30 insertions(+), 62 deletions(-) diff --git a/whisper/whisperv6/libp2p_glue.go b/whisper/whisperv6/libp2p_glue.go index 82d095dad2..5f7fac9b5b 100644 --- a/whisper/whisperv6/libp2p_glue.go +++ b/whisper/whisperv6/libp2p_glue.go @@ -17,15 +17,15 @@ package whisperv6 import ( + "bufio" "bytes" "context" - "encoding/binary" "fmt" "github.com/ethereum/go-ethereum/log" "io" - "math" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" crypto "github.com/libp2p/go-libp2p-crypto" host "github.com/libp2p/go-libp2p-host" inet "github.com/libp2p/go-libp2p-net" @@ -40,85 +40,54 @@ import ( // LibP2PStream is a wrapper used to implement the MsgReadWriter // interface for libp2p's streams. type LibP2PStream struct { - stream inet.Stream + lp2pStream inet.Stream + rlpStream *rlp.Stream } -const ( - codeLength = 8 - payloadSizeLength = 4 -) +func newLibp2pStream(lp2pStream inet.Stream) p2p.MsgReadWriter { + return &LibP2PStream{lp2pStream: lp2pStream, rlpStream: rlp.NewStream(bufio.NewReader(lp2pStream), 0)} +} // ReadMsg implements the MsgReadWriter interface to read messages // from lilbp2p streams. func (stream *LibP2PStream) ReadMsg() (p2p.Msg, error) { - codeBytes := make([]byte, codeLength) - nbytes, err := stream.stream.Read(codeBytes) + msgcode, err := stream.rlpStream.Uint() if err != nil { - return p2p.Msg{}, err - } else if nbytes != len(codeBytes) { - return p2p.Msg{}, fmt.Errorf("Invalid message header length: expected %d, got %d", codeLength, nbytes) + return p2p.Msg{}, fmt.Errorf("can't read message code: %v", err) } - code := binary.LittleEndian.Uint64(codeBytes) - - sizeBytes := make([]byte, payloadSizeLength) - nbytes, err = stream.stream.Read(sizeBytes) + _, size, err := stream.rlpStream.Kind() if err != nil { - return p2p.Msg{}, err - } else if nbytes != len(sizeBytes) { - return p2p.Msg{}, fmt.Errorf("Invalid message size length: expected %d, got %d", len(sizeBytes), nbytes) + return p2p.Msg{}, fmt.Errorf("can't read message size: %v", err) } - size := binary.LittleEndian.Uint32(sizeBytes) - if size > math.MaxInt32 { - return p2p.Msg{}, fmt.Errorf("Invalid message size length: got %d which is above the max of %d", size, math.MaxInt32) + // Only the size of the encoded payload is checked, so theoretically + // the decrypted message could be much bigger. We would need to add + // a function to rlp.Stream to find out what is the raw size. Since + // this is still work in progress, we won't bother with that just yet. + if size > uint64(MaxMessageSize) { + return p2p.Msg{}, fmt.Errorf("message too large") + } + content, err := stream.rlpStream.Raw() + if err != nil { + return p2p.Msg{}, fmt.Errorf("can't read message: %v", err) } - payload := make([]byte, size) - nbytes, err = stream.stream.Read(payload) - if err != nil { - return p2p.Msg{}, err - } else if nbytes != int(size) { - return p2p.Msg{}, fmt.Errorf("Invalid message payload length: expected %d, got %d", size, nbytes) - } - - return p2p.Msg{Code: code, Size: size, Payload: bytes.NewReader(payload)}, nil + return p2p.Msg{Code: msgcode, Size: uint32(len(content)), Payload: bytes.NewReader(content)}, nil } // WriteMsg implements the MsgReadWriter interface to write messages // to lilbp2p streams. func (stream *LibP2PStream) WriteMsg(msg p2p.Msg) error { - // Refuse to write messages with an unsigned size greater than - // a signed 32-bit integer size. This is because len() returns - // an int, forcing a conversion at some locations in the code, - // and on some blatforms that might cause an issue. - if msg.Size > math.MaxInt32 { - return fmt.Errorf("Payload size must be a maximum of %d bytes", math.MaxInt32) - } - data := make([]byte, msg.Size+codeLength+payloadSizeLength) - binary.LittleEndian.PutUint64(data[0:codeLength], msg.Code) - binary.LittleEndian.PutUint32(data[codeLength:codeLength+payloadSizeLength], msg.Size) + stream.lp2pStream.SetWriteDeadline(time.Now().Add(transmissionCycle)) - nbytes, err := msg.Payload.Read(data[codeLength+payloadSizeLength:]) - if nbytes > math.MaxInt32 || uint32(nbytes) != msg.Size { - return fmt.Errorf("Invalid size read in libp2p stream: read %d bytes, was expecting %d bytes", nbytes, msg.Size) - } else if err != nil && err != io.EOF { + if err := rlp.Encode(stream.lp2pStream, msg.Code); err != nil { return err } - - nbytes, err = stream.stream.Write(data) - - if err != nil { + _, err := io.Copy(stream.lp2pStream, msg.Payload) return err } - if nbytes != len(data) { - return fmt.Errorf("Invalid size written in libp2p stream: wrote %d bytes, was expecting %d bytes", nbytes, msg.Size) - } - - return nil -} - // LibP2PPeer implements Peer for libp2p type LibP2PPeer struct { *PeerBase @@ -177,10 +146,9 @@ func (server *LibP2PWhisperServer) connectToPeer(p *LibP2PPeer) error { } // Save the stream - lps := LibP2PStream{ - stream: s, - } - p.connectionStream = &lps + lps := newLibp2pStream(s).(*LibP2PStream) + + p.connectionStream = lps p.ws = p.connectionStream // TODO send my known list of peers @@ -205,7 +173,7 @@ func (server *LibP2PWhisperServer) Start() error { } } - lps := &LibP2PStream{stream} + lps := newLibp2pStream(stream).(*LibP2PStream) // Unknown peer if peer == nil { @@ -234,7 +202,7 @@ func (server *LibP2PWhisperServer) Start() error { func (server *LibP2PWhisperServer) Stop() { for _, p := range server.Peers { // TODO send disconnect message - p.connectionStream.stream.Close() + p.connectionStream.lp2pStream.Close() } server.Host.Close()