From 736e632215d49dd7bc61126f78dda4bad12768ea Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Fri, 27 Feb 2015 03:06:55 +0000 Subject: [PATCH] p2p: use RLPx frames for messaging --- p2p/handshake.go | 27 ++++++++++----------------- p2p/message.go | 19 +++++++++++++++++++ p2p/peer.go | 21 ++++++++++++--------- p2p/peer_test.go | 36 +++++++++++++++++++----------------- p2p/server.go | 7 ++++--- p2p/server_test.go | 13 +++++++++---- 6 files changed, 73 insertions(+), 50 deletions(-) diff --git a/p2p/handshake.go b/p2p/handshake.go index 17f572dea..10ef970dc 100644 --- a/p2p/handshake.go +++ b/p2p/handshake.go @@ -32,14 +32,10 @@ const ( ) type conn struct { - *frameRW + MsgReadWriter *protoHandshake } -func newConn(fd net.Conn, hs *protoHandshake) *conn { - return &conn{newFrameRW(fd, msgWriteTimeout), hs} -} - // encHandshake contains the state of the encryption handshake. type encHandshake struct { remoteID discover.NodeID @@ -115,17 +111,16 @@ func setupInboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake) ( // Run the protocol handshake using authenticated messages. // TODO: move buffering setup here (out of newFrameRW) - phsrw := newRlpxFrameRW(fd, secrets) - rhs, err := readProtocolHandshake(phsrw, our) + rw := newRlpxFrameRW(fd, secrets) + rhs, err := readProtocolHandshake(rw, our) if err != nil { return nil, err } - if err := writeProtocolHandshake(phsrw, our); err != nil { + // TODO: validate that handshake node ID matches + if err := writeProtocolHandshake(rw, our); err != nil { return nil, fmt.Errorf("protocol write error: %v", err) } - - rw := newFrameRW(fd, msgWriteTimeout) - return &conn{rw, rhs}, nil + return &conn{&lockedRW{wrapped: rw}, rhs}, nil } func setupOutboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, dial *discover.Node) (*conn, error) { @@ -136,20 +131,18 @@ func setupOutboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, // Run the protocol handshake using authenticated messages. // TODO: move buffering setup here (out of newFrameRW) - phsrw := newRlpxFrameRW(fd, secrets) - if err := writeProtocolHandshake(phsrw, our); err != nil { + rw := newRlpxFrameRW(fd, secrets) + if err := writeProtocolHandshake(rw, our); err != nil { return nil, fmt.Errorf("protocol write error: %v", err) } - rhs, err := readProtocolHandshake(phsrw, our) + rhs, err := readProtocolHandshake(rw, our) if err != nil { return nil, fmt.Errorf("protocol handshake read error: %v", err) } if rhs.ID != dial.ID { return nil, errors.New("dialed node id mismatch") } - - rw := newFrameRW(fd, msgWriteTimeout) - return &conn{rw, rhs}, nil + return &conn{&lockedRW{wrapped: rw}, rhs}, nil } // outboundEncHandshake negotiates a session token on conn. diff --git a/p2p/message.go b/p2p/message.go index 7adad4b09..d61faad13 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -119,6 +119,25 @@ func EncodeMsg(w MsgWriter, code uint64, data ...interface{}) error { return w.WriteMsg(NewMsg(code, data...)) } +// lockedRW wraps a MsgReadWriter with locks around +// ReadMsg and WriteMsg. +type lockedRW struct { + rmu, wmu sync.Mutex + wrapped MsgReadWriter +} + +func (rw *lockedRW) ReadMsg() (Msg, error) { + rw.rmu.Lock() + defer rw.rmu.Unlock() + return rw.wrapped.ReadMsg() +} + +func (rw *lockedRW) WriteMsg(msg Msg) error { + rw.wmu.Lock() + defer rw.wmu.Unlock() + return rw.wrapped.WriteMsg(msg) +} + // frameRW is a MsgReadWriter that reads and writes devp2p message frames. // As required by the interface, ReadMsg and WriteMsg can be called from // multiple goroutines. diff --git a/p2p/peer.go b/p2p/peer.go index fb027c834..4982c4612 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -40,6 +40,7 @@ type Peer struct { // Use them to display messages related to the peer. *logger.Logger + conn net.Conn rw *conn running map[string]*protoRW @@ -52,8 +53,9 @@ type Peer struct { // NewPeer returns a peer for testing purposes. func NewPeer(id discover.NodeID, name string, caps []Cap) *Peer { pipe, _ := net.Pipe() - conn := newConn(pipe, &protoHandshake{ID: id, Name: name, Caps: caps}) - peer := newPeer(conn, nil) + msgpipe, _ := MsgPipe() + conn := &conn{msgpipe, &protoHandshake{ID: id, Name: name, Caps: caps}} + peer := newPeer(pipe, conn, nil) close(peer.closed) // ensures Disconnect doesn't block return peer } @@ -76,12 +78,12 @@ func (p *Peer) Caps() []Cap { // RemoteAddr returns the remote address of the network connection. func (p *Peer) RemoteAddr() net.Addr { - return p.rw.RemoteAddr() + return p.conn.RemoteAddr() } // LocalAddr returns the local address of the network connection. func (p *Peer) LocalAddr() net.Addr { - return p.rw.LocalAddr() + return p.conn.LocalAddr() } // Disconnect terminates the peer connection with the given reason. @@ -98,10 +100,11 @@ func (p *Peer) String() string { return fmt.Sprintf("Peer %.8x %v", p.rw.ID[:], p.RemoteAddr()) } -func newPeer(conn *conn, protocols []Protocol) *Peer { - logtag := fmt.Sprintf("Peer %.8x %v", conn.ID[:], conn.RemoteAddr()) +func newPeer(fd net.Conn, conn *conn, protocols []Protocol) *Peer { + logtag := fmt.Sprintf("Peer %.8x %v", conn.ID[:], fd.RemoteAddr()) p := &Peer{ Logger: logger.NewLogger(logtag), + conn: fd, rw: conn, running: matchProtocols(protocols, conn.Caps, conn), disc: make(chan DiscReason), @@ -138,7 +141,7 @@ loop: // We rely on protocols to abort if there is a write error. It // might be more robust to handle them here as well. p.DebugDetailf("Read error: %v\n", err) - p.rw.Close() + p.conn.Close() return DiscNetworkError case err := <-p.protoErr: reason = discReasonForError(err) @@ -161,14 +164,14 @@ func (p *Peer) politeDisconnect(reason DiscReason) { EncodeMsg(p.rw, discMsg, uint(reason)) // Wait for the other side to close the connection. // Discard any data that they send until then. - io.Copy(ioutil.Discard, p.rw) + io.Copy(ioutil.Discard, p.conn) close(done) }() select { case <-done: case <-time.After(disconnectGracePeriod): } - p.rw.Close() + p.conn.Close() } func (p *Peer) readLoop() error { diff --git a/p2p/peer_test.go b/p2p/peer_test.go index a1260adbd..1ba43bed5 100644 --- a/p2p/peer_test.go +++ b/p2p/peer_test.go @@ -3,6 +3,7 @@ package p2p import ( "bytes" "fmt" + "io" "io/ioutil" "net" "reflect" @@ -29,8 +30,8 @@ var discard = Protocol{ }, } -func testPeer(protos []Protocol) (*conn, *Peer, <-chan DiscReason) { - fd1, fd2 := net.Pipe() +func testPeer(protos []Protocol) (io.Closer, *conn, *Peer, <-chan DiscReason) { + fd1, _ := net.Pipe() hs1 := &protoHandshake{ID: randomID(), Version: baseProtocolVersion} hs2 := &protoHandshake{ID: randomID(), Version: baseProtocolVersion} for _, p := range protos { @@ -38,11 +39,12 @@ func testPeer(protos []Protocol) (*conn, *Peer, <-chan DiscReason) { hs2.Caps = append(hs2.Caps, p.cap()) } - peer := newPeer(newConn(fd1, hs1), protos) + p1, p2 := MsgPipe() + peer := newPeer(fd1, &conn{p1, hs1}, protos) errc := make(chan DiscReason, 1) go func() { errc <- peer.run() }() - return newConn(fd2, hs2), peer, errc + return p1, &conn{p2, hs2}, peer, errc } func TestPeerProtoReadMsg(t *testing.T) { @@ -67,8 +69,8 @@ func TestPeerProtoReadMsg(t *testing.T) { }, } - rw, _, errc := testPeer([]Protocol{proto}) - defer rw.Close() + closer, rw, _, errc := testPeer([]Protocol{proto}) + defer closer.Close() EncodeMsg(rw, baseProtocolLength+2, 1) EncodeMsg(rw, baseProtocolLength+3, 2) @@ -105,8 +107,8 @@ func TestPeerProtoReadLargeMsg(t *testing.T) { }, } - rw, _, errc := testPeer([]Protocol{proto}) - defer rw.Close() + closer, rw, _, errc := testPeer([]Protocol{proto}) + defer closer.Close() EncodeMsg(rw, 18, make([]byte, msgsize)) select { @@ -134,8 +136,8 @@ func TestPeerProtoEncodeMsg(t *testing.T) { return nil }, } - rw, _, _ := testPeer([]Protocol{proto}) - defer rw.Close() + closer, rw, _, _ := testPeer([]Protocol{proto}) + defer closer.Close() if err := expectMsg(rw, 17, []string{"foo", "bar"}); err != nil { t.Error(err) @@ -145,8 +147,8 @@ func TestPeerProtoEncodeMsg(t *testing.T) { func TestPeerWriteForBroadcast(t *testing.T) { defer testlog(t).detach() - rw, peer, peerErr := testPeer([]Protocol{discard}) - defer rw.Close() + closer, rw, peer, peerErr := testPeer([]Protocol{discard}) + defer closer.Close() // test write errors if err := peer.writeProtoMsg("b", NewMsg(3)); err == nil { @@ -181,8 +183,8 @@ func TestPeerWriteForBroadcast(t *testing.T) { func TestPeerPing(t *testing.T) { defer testlog(t).detach() - rw, _, _ := testPeer(nil) - defer rw.Close() + closer, rw, _, _ := testPeer(nil) + defer closer.Close() if err := EncodeMsg(rw, pingMsg); err != nil { t.Fatal(err) } @@ -194,15 +196,15 @@ func TestPeerPing(t *testing.T) { func TestPeerDisconnect(t *testing.T) { defer testlog(t).detach() - rw, _, disc := testPeer(nil) - defer rw.Close() + closer, rw, _, disc := testPeer(nil) + defer closer.Close() if err := EncodeMsg(rw, discMsg, DiscQuitting); err != nil { t.Fatal(err) } if err := expectMsg(rw, discMsg, []interface{}{DiscRequested}); err != nil { t.Error(err) } - rw.Close() // make test end faster + closer.Close() // make test end faster if reason := <-disc; reason != DiscRequested { t.Errorf("run returned wrong reason: got %v, want %v", reason, DiscRequested) } diff --git a/p2p/server.go b/p2p/server.go index 3ea2538d1..e53e832aa 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -358,14 +358,15 @@ func (srv *Server) findPeers() { func (srv *Server) startPeer(fd net.Conn, dest *discover.Node) { // TODO: handle/store session token - fd.SetDeadline(time.Now().Add(handshakeTimeout)) + // TODO: reenable deadlines + // fd.SetDeadline(time.Now().Add(handshakeTimeout)) conn, err := srv.setupFunc(fd, srv.PrivateKey, srv.ourHandshake, dest) if err != nil { fd.Close() srvlog.Debugf("Handshake with %v failed: %v", fd.RemoteAddr(), err) return } - p := newPeer(conn, srv.Protocols) + p := newPeer(fd, conn, srv.Protocols) if ok, reason := srv.addPeer(conn.ID, p); !ok { srvlog.DebugDetailf("Not adding %v (%v)\n", p, reason) p.politeDisconnect(reason) @@ -375,7 +376,7 @@ func (srv *Server) startPeer(fd net.Conn, dest *discover.Node) { srvlog.Debugf("Added %v\n", p) srvjslog.LogJson(&logger.P2PConnected{ RemoteId: fmt.Sprintf("%x", conn.ID[:]), - RemoteAddress: conn.RemoteAddr().String(), + RemoteAddress: fd.RemoteAddr().String(), RemoteVersionString: conn.Name, NumConnections: srv.PeerCount(), }) diff --git a/p2p/server_test.go b/p2p/server_test.go index c109fffb9..c348f5a9a 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -11,6 +11,7 @@ import ( "time" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/p2p/discover" ) @@ -23,8 +24,14 @@ func startTestServer(t *testing.T, pf newPeerHook) *Server { newPeerHook: pf, setupFunc: func(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, dial *discover.Node) (*conn, error) { id := randomID() + rw := newRlpxFrameRW(fd, secrets{ + MAC: zero16, + AES: zero16, + IngressMAC: sha3.NewKeccak256(), + EgressMAC: sha3.NewKeccak256(), + }) return &conn{ - frameRW: newFrameRW(fd, msgWriteTimeout), + MsgReadWriter: rw, protoHandshake: &protoHandshake{ID: id, Version: baseProtocolVersion}, }, nil }, @@ -143,9 +150,7 @@ func TestServerBroadcast(t *testing.T) { // broadcast one message srv.Broadcast("discard", 0, "foo") - goldbuf := new(bytes.Buffer) - writeMsg(goldbuf, NewMsg(16, "foo")) - golden := goldbuf.Bytes() + golden := unhex("66e94e166f0a2c3b884cfa59ca34") // check that the message has been written everywhere for i, conn := range conns {