path: root/p2p/peer.go

package p2p

import (
    "errors"
    "fmt"
    "io"
    "io/ioutil"
    "net"
    "sort"
    "sync"
    "time"

    "github.com/ethereum/go-ethereum/logger"
    "github.com/ethereum/go-ethereum/p2p/discover"
    "github.com/ethereum/go-ethereum/rlp"
)

const (
    baseProtocolVersion    = 3
    baseProtocolLength     = uint64(16)
    baseProtocolMaxMsgSize = 10 * 1024 * 1024

    disconnectGracePeriod = 2 * time.Second
)

const (
    // devp2p message codes
    handshakeMsg = 0x00
    discMsg      = 0x01
    pingMsg      = 0x02
    pongMsg      = 0x03
    getPeersMsg  = 0x04
    peersMsg     = 0x05
)

// handshake is the RLP structure of the protocol handshake.
type handshake struct {
    Version    uint64
    Name       string
    Caps       []Cap
    ListenPort uint64
    NodeID     discover.NodeID
}

// Peer represents a connected remote node.
type Peer struct {
    // Peers have all the log methods.
    // Use them to display messages related to the peer.
    *logger.Logger

    infoMu sync.Mutex
    name   string
    caps   []Cap

    ourID, remoteID *discover.NodeID
    ourName         string

    rw *frameRW

    // These fields maintain the running protocols.
    protocols []Protocol
    runlock   sync.RWMutex // protects running
    running   map[string]*proto

    // disables protocol handshake, for testing
    noHandshake bool

    protoWG  sync.WaitGroup
    protoErr chan error
    closed   chan struct{}
    disc     chan DiscReason
}

// NewPeer returns a peer for testing purposes.
func NewPeer(id discover.NodeID, name string, caps []Cap) *Peer {
    conn, _ := net.Pipe()
    peer := newPeer(conn, nil, "", nil, &id)
    peer.setHandshakeInfo(name, caps)
    close(peer.closed) // ensures Disconnect doesn't block
    return peer
}

// ID returns the node's public key.
func (p *Peer) ID() discover.NodeID {
    return *p.remoteID
}

// Name returns the node name that the remote node advertised.
func (p *Peer) Name() string {
    // this needs a lock because the information is part of the
    // protocol handshake.
    p.infoMu.Lock()
    name := p.name
    p.infoMu.Unlock()
    return name
}

// Caps returns the capabilities (supported subprotocols) of the remote peer.
func (p *Peer) Caps() []Cap {
    // this needs a lock because the information is part of the
    // protocol handshake.
    p.infoMu.Lock()
    caps := p.caps
    p.infoMu.Unlock()
    return caps
}

// RemoteAddr returns the remote address of the network connection.
func (p *Peer) RemoteAddr() net.Addr {
    return p.rw.RemoteAddr()
}

// LocalAddr returns the local address of the network connection.
func (p *Peer) LocalAddr() net.Addr {
    return p.rw.LocalAddr()
}

// Disconnect terminates the peer connection with the given reason.
// It returns immediately and does not wait until the connection is closed.
func (p *Peer) Disconnect(reason DiscReason) {
    select {
    case p.disc <- reason:
    case <-p.closed:
    }
}

// String implements fmt.Stringer.
func (p *Peer) String() string {
    return fmt.Sprintf("Peer %.8x %v", p.remoteID[:], p.RemoteAddr())
}

func newPeer(conn net.Conn, protocols []Protocol, ourName string, ourID, remoteID *discover.NodeID) *Peer {
    logtag := fmt.Sprintf("Peer %.8x %v", remoteID[:], conn.RemoteAddr())
    return &Peer{
        Logger:    logger.NewLogger(logtag),
        rw:        newFrameRW(conn, msgWriteTimeout),
        ourID:     ourID,
        ourName:   ourName,
        remoteID:  remoteID,
        protocols: protocols,
        running:   make(map[string]*proto),
        disc:      make(chan DiscReason),
        protoErr:  make(chan error),
        closed:    make(chan struct{}),
    }
}

func (p *Peer) setHandshakeInfo(name string, caps []Cap) {
    p.infoMu.Lock()
    p.name = name
    p.caps = caps
    p.infoMu.Unlock()
}

func (p *Peer) run() DiscReason {
    var readErr = make(chan error, 1)
    defer p.closeProtocols()
    defer close(p.closed)

    go func() { readErr <- p.readLoop() }()

    if !p.noHandshake {
        if err := writeProtocolHandshake(p.rw, p.ourName, *p.ourID, p.protocols); err != nil {
            p.DebugDetailf("Protocol handshake error: %v\n", err)
            p.rw.Close()
            return DiscProtocolError
        }
    }

    // Wait for an error or disconnect.
    var reason DiscReason
    select {
    case err := <-readErr:
        // 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()
        return DiscNetworkError

    case err := <-p.protoErr:
        reason = discReasonForError(err)
    case reason = <-p.disc:
    }
    p.politeDisconnect(reason)

    // Wait for readLoop. It will end because conn is now closed.
    <-readErr
    p.Debugf("Disconnected: %v\n", reason)
    return reason
}

func (p *Peer) politeDisconnect(reason DiscReason) {
    done := make(chan struct{})
    go func() {
        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)
        close(done)
    }()
    select {
    case <-done:
    case <-time.After(disconnectGracePeriod):
    }
    p.rw.Close()
}

func (p *Peer) readLoop() error {
    if !p.noHandshake {
        if err := readProtocolHandshake(p, p.rw); err != nil {
            return err
        }
    }
    for {
        msg, err := p.rw.ReadMsg()
        if err != nil {
            return err
        }
        if err = p.handle(msg); err != nil {
            return err
        }
    }
    return nil
}

func (p *Peer) handle(msg Msg) error {
    switch {
    case msg.Code == pingMsg:
        msg.Discard()
        go EncodeMsg(p.rw, pongMsg)
    case msg.Code == discMsg:
        var reason DiscReason
        // no need to discard or for error checking, we'll close the
        // connection after this.
        rlp.Decode(msg.Payload, &reason)
        p.Disconnect(DiscRequested)
        return discRequestedError(reason)
    case msg.Code < baseProtocolLength:
        // ignore other base protocol messages
        return msg.Discard()
    default:
        // it's a subprotocol message
        proto, err := p.getProto(msg.Code)
        if err != nil {
            return fmt.Errorf("msg code out of range: %v", msg.Code)
        }
        proto.in <- msg
    }
    return nil
}

func readProtocolHandshake(p *Peer, rw MsgReadWriter) error {
    // read and handle remote handshake
    msg, err := rw.ReadMsg()
    if err != nil {
        return err
    }
    if msg.Code == discMsg {
        // disconnect before protocol handshake is valid according to the
        // spec and we send it ourself if Server.addPeer fails.
        var reason DiscReason
        rlp.Decode(msg.Payload, &reason)
        return discRequestedError(reason)
    }
    if msg.Code != handshakeMsg {
        return newPeerError(errProtocolBreach, "expected handshake, got %x", msg.Code)
    }
    if msg.Size > baseProtocolMaxMsgSize {
        return newPeerError(errInvalidMsg, "message too big")
    }
    var hs handshake
    if err := msg.Decode(&hs); err != nil {
        return err
    }
    // validate handshake info
    if hs.Version != baseProtocolVersion {
        return newPeerError(errP2PVersionMismatch, "required version %d, received %d\n",
            baseProtocolVersion, hs.Version)
    }
    if hs.NodeID == *p.remoteID {
        return newPeerError(errPubkeyForbidden, "node ID mismatch")
    }
    // TODO: remove Caps with empty name
    p.setHandshakeInfo(hs.Name, hs.Caps)
    p.startSubprotocols(hs.Caps)
    return nil
}

func writeProtocolHandshake(w MsgWriter, name string, id discover.NodeID, ps []Protocol) error {
    var caps []interface{}
    for _, proto := range ps {
        caps = append(caps, proto.cap())
    }
    return EncodeMsg(w, handshakeMsg, baseProtocolVersion, name, caps, 0, id)
}

// startProtocols starts matching named subprotocols.
func (p *Peer) startSubprotocols(caps []Cap) {
    sort.Sort(capsByName(caps))
    p.runlock.Lock()
    defer p.runlock.Unlock()
    offset := baseProtocolLength
outer:
    for _, cap := range caps {
        for _, proto := range p.protocols {
            if proto.Name == cap.Name &&
                proto.Version == cap.Version &&
                p.running[cap.Name] == nil {
                p.running[cap.Name] = p.startProto(offset, proto)
                offset += proto.Length
                continue outer
            }
        }
    }
}

func (p *Peer) startProto(offset uint64, impl Protocol) *proto {
    p.DebugDetailf("Starting protocol %s/%d\n", impl.Name, impl.Version)
    rw := &proto{
        name:    impl.Name,
        in:      make(chan Msg),
        offset:  offset,
        maxcode: impl.Length,
        w:       p.rw,
    }
    p.protoWG.Add(1)
    go func() {
        err := impl.Run(p, rw)
        if err == nil {
            p.DebugDetailf("Protocol %s/%d returned\n", impl.Name, impl.Version)
            err = errors.New("protocol returned")
        } else {
            p.DebugDetailf("Protocol %s/%d error: %v\n", impl.Name, impl.Version, err)
        }
        select {
        case p.protoErr <- err:
        case <-p.closed:
        }
        p.protoWG.Done()
    }()
    return rw
}

// getProto finds the protocol responsible for handling
// the given message code.
func (p *Peer) getProto(code uint64) (*proto, error) {
    p.runlock.RLock()
    defer p.runlock.RUnlock()
    for _, proto := range p.running {
        if code >= proto.offset && code < proto.offset+proto.maxcode {
            return proto, nil
        }
    }
    return nil, newPeerError(errInvalidMsgCode, "%d", code)
}

func (p *Peer) closeProtocols() {
    p.runlock.RLock()
    for _, p := range p.running {
        close(p.in)
    }
    p.runlock.RUnlock()
    p.protoWG.Wait()
}

// writeProtoMsg sends the given message on behalf of the given named protocol.
// this exists because of Server.Broadcast.
func (p *Peer) writeProtoMsg(protoName string, msg Msg) error {
    p.runlock.RLock()
    proto, ok := p.running[protoName]
    p.runlock.RUnlock()
    if !ok {
        return fmt.Errorf("protocol %s not handled by peer", protoName)
    }
    if msg.Code >= proto.maxcode {
        return newPeerError(errInvalidMsgCode, "code %x is out of range for protocol %q", msg.Code, protoName)
    }
    msg.Code += proto.offset
    return p.rw.WriteMsg(msg)
}

type proto struct {
    name            string
    in              chan Msg
    maxcode, offset uint64
    w               MsgWriter
}

func (rw *proto) WriteMsg(msg Msg) error {
    if msg.Code >= rw.maxcode {
        return newPeerError(errInvalidMsgCode, "not handled")
    }
    msg.Code += rw.offset
    return rw.w.WriteMsg(msg)
}

func (rw *proto) ReadMsg() (Msg, error) {
    msg, ok := <-rw.in
    if !ok {
        return msg, io.EOF
    }
    msg.Code -= rw.offset
    return msg, nil
}