// 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 . // Package p2p implements the Ethereum p2p network protocols. package p2p import ( "bytes" "crypto/ecdsa" "encoding/hex" "errors" "net" "sort" "sync" "sync/atomic" "time" "github.com/dexon-foundation/dexon/common" "github.com/dexon-foundation/dexon/common/mclock" "github.com/dexon-foundation/dexon/crypto" "github.com/dexon-foundation/dexon/event" "github.com/dexon-foundation/dexon/log" "github.com/dexon-foundation/dexon/p2p/discover" "github.com/dexon-foundation/dexon/p2p/discv5" "github.com/dexon-foundation/dexon/p2p/enode" "github.com/dexon-foundation/dexon/p2p/enr" "github.com/dexon-foundation/dexon/p2p/nat" "github.com/dexon-foundation/dexon/p2p/netutil" "github.com/dexon-foundation/dexon/rlp" ) const ( defaultDialTimeout = 15 * time.Second // Connectivity defaults. maxActiveDialTasks = 16 defaultMaxPendingPeers = 50 defaultDialRatio = 3 // Maximum time allowed for reading a complete message. // This is effectively the amount of time a connection can be idle. frameReadTimeout = 30 * time.Second // Maximum amount of time allowed for writing a complete message. frameWriteTimeout = 20 * time.Second ) var errServerStopped = errors.New("server stopped") // Config holds Server options. type Config struct { // This field must be set to a valid secp256k1 private key. PrivateKey *ecdsa.PrivateKey `toml:"-"` // MaxPeers is the maximum number of peers that can be // connected. It must be greater than zero. MaxPeers int // MaxPendingPeers is the maximum number of peers that can be pending in the // handshake phase, counted separately for inbound and outbound connections. // Zero defaults to preset values. MaxPendingPeers int `toml:",omitempty"` // DialRatio controls the ratio of inbound to dialed connections. // Example: a DialRatio of 2 allows 1/2 of connections to be dialed. // Setting DialRatio to zero defaults it to 3. DialRatio int `toml:",omitempty"` // NoDiscovery can be used to disable the peer discovery mechanism. // Disabling is useful for protocol debugging (manual topology). NoDiscovery bool // DiscoveryV5 specifies whether the new topic-discovery based V5 discovery // protocol should be started or not. DiscoveryV5 bool `toml:",omitempty"` // Name sets the node name of this server. // Use common.MakeName to create a name that follows existing conventions. Name string `toml:"-"` // BootstrapNodes are used to establish connectivity // with the rest of the network. BootstrapNodes []*enode.Node // BootstrapNodesV5 are used to establish connectivity // with the rest of the network using the V5 discovery // protocol. BootstrapNodesV5 []*discv5.Node `toml:",omitempty"` // Static nodes are used as pre-configured connections which are always // maintained and re-connected on disconnects. StaticNodes []*enode.Node // Trusted nodes are used as pre-configured connections which are always // allowed to connect, even above the peer limit. TrustedNodes []*enode.Node // Connectivity can be restricted to certain IP networks. // If this option is set to a non-nil value, only hosts which match one of the // IP networks contained in the list are considered. NetRestrict *netutil.Netlist `toml:",omitempty"` // NodeDatabase is the path to the database containing the previously seen // live nodes in the network. NodeDatabase string `toml:",omitempty"` // Protocols should contain the protocols supported // by the server. Matching protocols are launched for // each peer. Protocols []Protocol `toml:"-"` // If ListenAddr is set to a non-nil address, the server // will listen for incoming connections. // // If the port is zero, the operating system will pick a port. The // ListenAddr field will be updated with the actual address when // the server is started. ListenAddr string // If set to a non-nil value, the given NAT port mapper // is used to make the listening port available to the // Internet. NAT nat.Interface `toml:",omitempty"` // If Dialer is set to a non-nil value, the given Dialer // is used to dial outbound peer connections. Dialer NodeDialer `toml:"-"` // If NoDial is true, the server will not dial any peers. NoDial bool `toml:",omitempty"` // If EnableMsgEvents is set then the server will emit PeerEvents // whenever a message is sent to or received from a peer EnableMsgEvents bool // Logger is a custom logger to use with the p2p.Server. Logger log.Logger `toml:",omitempty"` } // Server manages all peer connections. type Server struct { // Config fields may not be modified while the server is running. Config // Hooks for testing. These are useful because we can inhibit // the whole protocol stack. newTransport func(net.Conn) transport newPeerHook func(*Peer) lock sync.Mutex // protects running running bool nodedb *enode.DB localnode *enode.LocalNode ntab discoverTable listener net.Listener ourHandshake *protoHandshake lastLookup time.Time DiscV5 *discv5.Network // These are for Peers, PeerCount (and nothing else). peerOp chan peerOpFunc peerOpDone chan struct{} quit chan struct{} addstatic chan *enode.Node removestatic chan *enode.Node adddirect chan *enode.Node removedirect chan *enode.Node addtrusted chan *enode.Node removetrusted chan *enode.Node addgroup chan *dialGroup removegroup chan *dialGroup posthandshake chan *conn addpeer chan *conn delpeer chan peerDrop loopWG sync.WaitGroup // loop, listenLoop peerFeed event.Feed log log.Logger } type peerOpFunc func(map[enode.ID]*Peer) type peerDrop struct { *Peer err error requested bool // true if signaled by the peer } type connFlag int32 const ( dynDialedConn connFlag = 1 << iota staticDialedConn directDialedConn groupDialedConn inboundConn trustedConn ) // conn wraps a network connection with information gathered // during the two handshakes. type conn struct { fd net.Conn transport node *enode.Node flags connFlag cont chan error // The run loop uses cont to signal errors to SetupConn. caps []Cap // valid after the protocol handshake name string // valid after the protocol handshake } type transport interface { // The two handshakes. doEncHandshake(prv *ecdsa.PrivateKey, dialDest *ecdsa.PublicKey) (*ecdsa.PublicKey, error) doProtoHandshake(our *protoHandshake) (*protoHandshake, error) // The MsgReadWriter can only be used after the encryption // handshake has completed. The code uses conn.id to track this // by setting it to a non-nil value after the encryption handshake. MsgReadWriter // transports must provide Close because we use MsgPipe in some of // the tests. Closing the actual network connection doesn't do // anything in those tests because MsgPipe doesn't use it. close(err error) } func (c *conn) String() string { s := c.flags.String() if (c.node.ID() != enode.ID{}) { s += " " + c.node.ID().String() } s += " " + c.fd.RemoteAddr().String() return s } func (f connFlag) String() string { s := "" if f&trustedConn != 0 { s += "-trusted" } if f&dynDialedConn != 0 { s += "-dyndial" } if f&staticDialedConn != 0 { s += "-staticdial" } if f&directDialedConn != 0 { s += "-directdial" } if f&groupDialedConn != 0 { s += "-groupdial" } if f&inboundConn != 0 { s += "-inbound" } if s != "" { s = s[1:] } return s } func (c *conn) is(f connFlag) bool { flags := connFlag(atomic.LoadInt32((*int32)(&c.flags))) return flags&f != 0 } func (c *conn) set(f connFlag, val bool) { for { oldFlags := connFlag(atomic.LoadInt32((*int32)(&c.flags))) flags := oldFlags if val { flags |= f } else { flags &= ^f } if atomic.CompareAndSwapInt32((*int32)(&c.flags), int32(oldFlags), int32(flags)) { return } } } // Peers returns all connected peers. func (srv *Server) Peers() []*Peer { var ps []*Peer select { // Note: We'd love to put this function into a variable but // that seems to cause a weird compiler error in some // environments. case srv.peerOp <- func(peers map[enode.ID]*Peer) { for _, p := range peers { ps = append(ps, p) } }: <-srv.peerOpDone case <-srv.quit: } return ps } // PeerCount returns the number of connected peers. func (srv *Server) PeerCount() int { var count int select { case srv.peerOp <- func(ps map[enode.ID]*Peer) { count = len(ps) }: <-srv.peerOpDone case <-srv.quit: } return count } // AddPeer connects to the given node and maintains the connection until the // server is shut down. If the connection fails for any reason, the server will // attempt to reconnect the peer. func (srv *Server) AddPeer(node *enode.Node) { select { case srv.addstatic <- node: case <-srv.quit: } } // RemovePeer disconnects from the given node func (srv *Server) RemovePeer(node *enode.Node) { select { case srv.removestatic <- node: case <-srv.quit: } } // AddDirectPeer connects to the given node and maintains the connection until the // server is shut down. If the connection fails for any reason, the server will // attempt to reconnect the peer. func (srv *Server) AddDirectPeer(node *enode.Node) { select { case srv.adddirect <- node: case <-srv.quit: } } // RemoveDirectPeer disconnects from the given node func (srv *Server) RemoveDirectPeer(node *enode.Node) { select { case srv.removedirect <- node: case <-srv.quit: } } func (srv *Server) AddGroup(name string, nodes []*enode.Node, num uint64) { m := map[enode.ID]*enode.Node{} for _, node := range nodes { m[node.ID()] = node } g := &dialGroup{name: name, nodes: m, num: num} select { case srv.addgroup <- g: case <-srv.quit: } } func (srv *Server) RemoveGroup(name string) { g := &dialGroup{name: name} select { case srv.removegroup <- g: case <-srv.quit: } } // AddTrustedPeer adds the given node to a reserved whitelist which allows the // node to always connect, even if the slot are full. func (srv *Server) AddTrustedPeer(node *enode.Node) { select { case srv.addtrusted <- node: case <-srv.quit: } } // RemoveTrustedPeer removes the given node from the trusted peer set. func (srv *Server) RemoveTrustedPeer(node *enode.Node) { select { case srv.removetrusted <- node: case <-srv.quit: } } // SubscribePeers subscribes the given channel to peer events func (srv *Server) SubscribeEvents(ch chan *PeerEvent) event.Subscription { return srv.peerFeed.Subscribe(ch) } // Self returns the local node's endpoint information. func (srv *Server) Self() *enode.Node { srv.lock.Lock() ln := srv.localnode srv.lock.Unlock() if ln == nil { return enode.NewV4(&srv.PrivateKey.PublicKey, net.ParseIP("0.0.0.0"), 0, 0) } return ln.Node() } func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *enode.Node { // If the node is running but discovery is off, manually assemble the node infos. if ntab == nil { addr := srv.tcpAddr(listener) return enode.NewV4(&srv.PrivateKey.PublicKey, addr.IP, addr.Port, 0) } // Otherwise return the discovery node. return ntab.Self() } func (srv *Server) tcpAddr(listener net.Listener) net.TCPAddr { addr := net.TCPAddr{IP: net.IP{0, 0, 0, 0}} if listener == nil { return addr // Inbound connections disabled, use zero address. } // Otherwise inject the listener address too. if a, ok := listener.Addr().(*net.TCPAddr); ok { addr = *a } if srv.NAT != nil { if ip, err := srv.NAT.ExternalIP(); err == nil { addr.IP = ip } } if addr.IP.IsUnspecified() { addr.IP = net.IP{127, 0, 0, 1} } return addr } func (srv *Server) GetPrivateKey() *ecdsa.PrivateKey { return srv.PrivateKey } // Stop terminates the server and all active peer connections. // It blocks until all active connections have been closed. func (srv *Server) Stop() { srv.lock.Lock() if !srv.running { srv.lock.Unlock() return } srv.running = false if srv.listener != nil { // this unblocks listener Accept srv.listener.Close() } close(srv.quit) srv.lock.Unlock() srv.loopWG.Wait() } // sharedUDPConn implements a shared connection. Write sends messages to the underlying connection while read returns // messages that were found unprocessable and sent to the unhandled channel by the primary listener. type sharedUDPConn struct { *net.UDPConn unhandled chan discover.ReadPacket } // ReadFromUDP implements discv5.conn func (s *sharedUDPConn) ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error) { packet, ok := <-s.unhandled if !ok { return 0, nil, errors.New("Connection was closed") } l := len(packet.Data) if l > len(b) { l = len(b) } copy(b[:l], packet.Data[:l]) return l, packet.Addr, nil } // Close implements discv5.conn func (s *sharedUDPConn) Close() error { return nil } // Start starts running the server. // Servers can not be re-used after stopping. func (srv *Server) Start() (err error) { srv.lock.Lock() defer srv.lock.Unlock() if srv.running { return errors.New("server already running") } srv.running = true srv.log = srv.Config.Logger if srv.log == nil { srv.log = log.New() } if srv.NoDial && srv.ListenAddr == "" { srv.log.Warn("P2P server will be useless, neither dialing nor listening") } // static fields if srv.PrivateKey == nil { return errors.New("Server.PrivateKey must be set to a non-nil key") } if srv.newTransport == nil { srv.newTransport = newRLPX } if srv.Dialer == nil { srv.Dialer = TCPDialer{&net.Dialer{Timeout: defaultDialTimeout}} } srv.quit = make(chan struct{}) srv.addpeer = make(chan *conn) srv.delpeer = make(chan peerDrop) srv.posthandshake = make(chan *conn) srv.addstatic = make(chan *enode.Node) srv.removestatic = make(chan *enode.Node) srv.adddirect = make(chan *enode.Node) srv.removedirect = make(chan *enode.Node) srv.addgroup = make(chan *dialGroup) srv.removegroup = make(chan *dialGroup) srv.addtrusted = make(chan *enode.Node) srv.removetrusted = make(chan *enode.Node) srv.peerOp = make(chan peerOpFunc) srv.peerOpDone = make(chan struct{}) if err := srv.setupLocalNode(); err != nil { return err } if srv.ListenAddr != "" { if err := srv.setupListening(); err != nil { return err } } if err := srv.setupDiscovery(); err != nil { return err } dynPeers := srv.maxDialedConns() dialer := newDialState(srv.localnode.ID(), srv.StaticNodes, srv.BootstrapNodes, srv.ntab, dynPeers, srv.NetRestrict) srv.loopWG.Add(1) go srv.run(dialer) return nil } func (srv *Server) setupLocalNode() error { // Create the devp2p handshake. pubkey := crypto.FromECDSAPub(&srv.PrivateKey.PublicKey) srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: pubkey[1:]} for _, p := range srv.Protocols { srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap()) } sort.Sort(capsByNameAndVersion(srv.ourHandshake.Caps)) // Create the local node. db, err := enode.OpenDB(srv.Config.NodeDatabase) if err != nil { return err } srv.nodedb = db srv.localnode = enode.NewLocalNode(db, srv.PrivateKey) srv.localnode.SetFallbackIP(net.IP{127, 0, 0, 1}) srv.localnode.Set(capsByNameAndVersion(srv.ourHandshake.Caps)) // TODO: check conflicts for _, p := range srv.Protocols { for _, e := range p.Attributes { srv.localnode.Set(e) } } switch srv.NAT.(type) { case nil: // No NAT interface, do nothing. case nat.ExtIP: // ExtIP doesn't block, set the IP right away. ip, _ := srv.NAT.ExternalIP() srv.localnode.SetStaticIP(ip) default: // Ask the router about the IP. This takes a while and blocks startup, // do it in the background. srv.loopWG.Add(1) go func() { defer srv.loopWG.Done() if ip, err := srv.NAT.ExternalIP(); err == nil { srv.localnode.SetStaticIP(ip) } }() } return nil } func (srv *Server) setupDiscovery() error { if srv.NoDiscovery && !srv.DiscoveryV5 { return nil } addr, err := net.ResolveUDPAddr("udp", srv.ListenAddr) if err != nil { return err } conn, err := net.ListenUDP("udp", addr) if err != nil { return err } realaddr := conn.LocalAddr().(*net.UDPAddr) srv.log.Debug("UDP listener up", "addr", realaddr) if srv.NAT != nil { if !realaddr.IP.IsLoopback() { go nat.Map(srv.NAT, srv.quit, "udp", realaddr.Port, realaddr.Port, "ethereum discovery") } } srv.localnode.SetFallbackUDP(realaddr.Port) // Discovery V4 var unhandled chan discover.ReadPacket var sconn *sharedUDPConn if !srv.NoDiscovery { if srv.DiscoveryV5 { unhandled = make(chan discover.ReadPacket, 100) sconn = &sharedUDPConn{conn, unhandled} } cfg := discover.Config{ PrivateKey: srv.PrivateKey, NetRestrict: srv.NetRestrict, Bootnodes: srv.BootstrapNodes, Unhandled: unhandled, } ntab, err := discover.ListenUDP(conn, srv.localnode, cfg) if err != nil { return err } srv.ntab = ntab } // Discovery V5 if srv.DiscoveryV5 { var ntab *discv5.Network var err error if sconn != nil { ntab, err = discv5.ListenUDP(srv.PrivateKey, sconn, "", srv.NetRestrict) } else { ntab, err = discv5.ListenUDP(srv.PrivateKey, conn, "", srv.NetRestrict) } if err != nil { return err } if err := ntab.SetFallbackNodes(srv.BootstrapNodesV5); err != nil { return err } srv.DiscV5 = ntab } return nil } func (srv *Server) setupListening() error { // Launch the TCP listener. listener, err := net.Listen("tcp", srv.ListenAddr) if err != nil { return err } laddr := listener.Addr().(*net.TCPAddr) srv.ListenAddr = laddr.String() srv.listener = listener srv.localnode.Set(enr.TCP(laddr.Port)) srv.loopWG.Add(1) go srv.listenLoop() // Map the TCP listening port if NAT is configured. if !laddr.IP.IsLoopback() && srv.NAT != nil { srv.loopWG.Add(1) go func() { nat.Map(srv.NAT, srv.quit, "tcp", laddr.Port, laddr.Port, "ethereum p2p") srv.loopWG.Done() }() } return nil } type dialer interface { newTasks(running int, peers map[enode.ID]*Peer, now time.Time) []task taskDone(task, time.Time) addStatic(*enode.Node) removeStatic(*enode.Node) addDirect(*enode.Node) removeDirect(*enode.Node) addGroup(*dialGroup) removeGroup(*dialGroup) } func (srv *Server) run(dialstate dialer) { srv.log.Info("Started P2P networking", "self", srv.localnode.Node()) defer srv.loopWG.Done() defer srv.nodedb.Close() var ( peers = make(map[enode.ID]*Peer) inboundCount = 0 trusted = make(map[enode.ID]bool, len(srv.TrustedNodes)) peerflags = make(map[enode.ID]connFlag) groupRefCount = make(map[enode.ID]int32) groups = make(map[string]*dialGroup) taskdone = make(chan task, maxActiveDialTasks) runningTasks []task queuedTasks []task // tasks that can't run yet ) // Put trusted nodes into a map to speed up checks. // Trusted peers are loaded on startup or added via AddTrustedPeer RPC. for _, n := range srv.TrustedNodes { trusted[n.ID()] = true } // removes t from runningTasks delTask := func(t task) { for i := range runningTasks { if runningTasks[i] == t { runningTasks = append(runningTasks[:i], runningTasks[i+1:]...) break } } } // starts until max number of active tasks is satisfied startTasks := func(ts []task) (rest []task) { i := 0 for ; len(runningTasks) < maxActiveDialTasks && i < len(ts); i++ { t := ts[i] srv.log.Trace("New dial task", "task", t) go func() { t.Do(srv); taskdone <- t }() runningTasks = append(runningTasks, t) } return ts[i:] } scheduleTasks := func() { // Start from queue first. queuedTasks = append(queuedTasks[:0], startTasks(queuedTasks)...) // Query dialer for new tasks and start as many as possible now. if len(runningTasks) < maxActiveDialTasks { nt := dialstate.newTasks(len(runningTasks)+len(queuedTasks), peers, time.Now()) queuedTasks = append(queuedTasks, startTasks(nt)...) } } // remember and maintain the connection flags locally setConnFlags := func(id enode.ID, f connFlag, val bool) { if p, ok := peers[id]; ok { p.rw.set(f, val) } if val { peerflags[id] |= f } else { peerflags[id] &= ^f } if peerflags[id] == 0 { delete(peerflags, id) } } // Put trusted nodes into a map to speed up checks. // Trusted peers are loaded on startup or added via AddTrustedPeer RPC. for _, n := range srv.TrustedNodes { setConnFlags(n.ID(), trustedConn, true) } canDisconnect := func(p *Peer) bool { f, ok := peerflags[p.ID()] if ok && f != 0 { return false } return !p.rw.is(dynDialedConn | inboundConn) } removeGroup := func(g *dialGroup) { if gg, ok := groups[g.name]; ok { for id := range gg.nodes { groupRefCount[id]-- if groupRefCount[id] == 0 { setConnFlags(id, groupDialedConn, false) delete(groupRefCount, id) } } } } addGroup := func(g *dialGroup) { if _, ok := groups[g.name]; ok { removeGroup(groups[g.name]) } for id := range g.nodes { groupRefCount[id]++ if groupRefCount[id] > 0 { setConnFlags(id, groupDialedConn, true) } } groups[g.name] = g } running: for { scheduleTasks() select { case <-srv.quit: // The server was stopped. Run the cleanup logic. break running case n := <-srv.addstatic: // This channel is used by AddPeer to add to the // ephemeral static peer list. Add it to the dialer, // it will keep the node connected. srv.log.Trace("Adding static node", "node", n) setConnFlags(n.ID(), staticDialedConn, true) dialstate.addStatic(n) case n := <-srv.removestatic: // This channel is used by RemovePeer to send a // disconnect request to a peer and begin the // stop keeping the node connected. srv.log.Trace("Removing static node", "node", n) setConnFlags(n.ID(), staticDialedConn, false) dialstate.removeStatic(n) if p, ok := peers[n.ID()]; ok && canDisconnect(p) { p.Disconnect(DiscRequested) } case n := <-srv.adddirect: // This channel is used by AddDirectPeer to add to the // ephemeral direct peer list. Add it to the dialer, // it will keep the node connected. srv.log.Trace("Adding direct node", "node", n) setConnFlags(n.ID(), directDialedConn, true) if p, ok := peers[n.ID()]; ok { p.rw.set(directDialedConn, true) } dialstate.addDirect(n) case n := <-srv.removedirect: // This channel is used by RemoveDirectPeer to send a // disconnect request to a peer and begin the // stop keeping the node connected. srv.log.Trace("Removing direct node", "node", n) setConnFlags(n.ID(), directDialedConn, false) if p, ok := peers[n.ID()]; ok { p.rw.set(directDialedConn, false) if !p.rw.is(trustedConn | groupDialedConn) { p.Disconnect(DiscRequested) } } dialstate.removeDirect(n) case g := <-srv.addgroup: srv.log.Trace("Adding group", "group", g) addGroup(g) dialstate.addGroup(g) case g := <-srv.removegroup: srv.log.Trace("Removing group", "group", g) removeGroup(g) dialstate.removeGroup(g) case n := <-srv.addtrusted: // This channel is used by AddTrustedPeer to add an enode // to the trusted node set. srv.log.Trace("Adding trusted node", "node", n) setConnFlags(n.ID(), trustedConn, true) case n := <-srv.removetrusted: // This channel is used by RemoveTrustedPeer to remove an enode // from the trusted node set. srv.log.Trace("Removing trusted node", "node", n) setConnFlags(n.ID(), trustedConn, false) case op := <-srv.peerOp: // This channel is used by Peers and PeerCount. op(peers) srv.peerOpDone <- struct{}{} case t := <-taskdone: // A task got done. Tell dialstate about it so it // can update its state and remove it from the active // tasks list. srv.log.Trace("Dial task done", "task", t) dialstate.taskDone(t, time.Now()) delTask(t) case c := <-srv.posthandshake: // A connection has passed the encryption handshake so // the remote identity is known (but hasn't been verified yet). if f, ok := peerflags[c.node.ID()]; ok { c.flags |= f } // TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them. select { case c.cont <- srv.encHandshakeChecks(peers, inboundCount, c): case <-srv.quit: break running } case c := <-srv.addpeer: // At this point the connection is past the protocol handshake. // Its capabilities are known and the remote identity is verified. err := srv.protoHandshakeChecks(peers, inboundCount, c) if err == nil { // The handshakes are done and it passed all checks. p := newPeer(c, srv.Protocols) // If message events are enabled, pass the peerFeed // to the peer if srv.EnableMsgEvents { p.events = &srv.peerFeed } name := truncateName(c.name) srv.log.Debug("Adding p2p peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1) go srv.runPeer(p) peers[c.node.ID()] = p if p.Inbound() { inboundCount++ } } // The dialer logic relies on the assumption that // dial tasks complete after the peer has been added or // discarded. Unblock the task last. select { case c.cont <- err: case <-srv.quit: break running } case pd := <-srv.delpeer: // A peer disconnected. d := common.PrettyDuration(mclock.Now() - pd.created) pd.log.Debug("Removing p2p peer", "duration", d, "peers", len(peers)-1, "req", pd.requested, "err", pd.err) delete(peers, pd.ID()) if pd.Inbound() { inboundCount-- } } } srv.log.Trace("P2P networking is spinning down") // Terminate discovery. If there is a running lookup it will terminate soon. if srv.ntab != nil { srv.ntab.Close() } if srv.DiscV5 != nil { srv.DiscV5.Close() } // Disconnect all peers. for _, p := range peers { p.Disconnect(DiscQuitting) } // Wait for peers to shut down. Pending connections and tasks are // not handled here and will terminate soon-ish because srv.quit // is closed. for len(peers) > 0 { p := <-srv.delpeer p.log.Trace("<-delpeer (spindown)", "remainingTasks", len(runningTasks)) delete(peers, p.ID()) } } func (srv *Server) protoHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int, c *conn) error { // Drop connections with no matching protocols. if len(srv.Protocols) > 0 && countMatchingProtocols(srv.Protocols, c.caps) == 0 { return DiscUselessPeer } // Repeat the encryption handshake checks because the // peer set might have changed between the handshakes. return srv.encHandshakeChecks(peers, inboundCount, c) } func (srv *Server) encHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int, c *conn) error { switch { case !c.is(trustedConn|staticDialedConn|directDialedConn|groupDialedConn) && len(peers) >= srv.MaxPeers: return DiscTooManyPeers case !c.is(trustedConn|directDialedConn|groupDialedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns(): return DiscTooManyPeers case peers[c.node.ID()] != nil: return DiscAlreadyConnected case c.node.ID() == srv.localnode.ID(): return DiscSelf default: return nil } } func (srv *Server) maxInboundConns() int { return srv.MaxPeers - srv.maxDialedConns() } func (srv *Server) maxDialedConns() int { if srv.NoDiscovery || srv.NoDial { return 0 } r := srv.DialRatio if r == 0 { r = defaultDialRatio } return srv.MaxPeers / r } // listenLoop runs in its own goroutine and accepts // inbound connections. func (srv *Server) listenLoop() { defer srv.loopWG.Done() srv.log.Debug("TCP listener up", "addr", srv.listener.Addr()) tokens := defaultMaxPendingPeers if srv.MaxPendingPeers > 0 { tokens = srv.MaxPendingPeers } slots := make(chan struct{}, tokens) for i := 0; i < tokens; i++ { slots <- struct{}{} } for { // Wait for a handshake slot before accepting. <-slots var ( fd net.Conn err error ) for { fd, err = srv.listener.Accept() if netutil.IsTemporaryError(err) { srv.log.Debug("Temporary read error", "err", err) continue } else if err != nil { srv.log.Debug("Read error", "err", err) return } break } // Reject connections that do not match NetRestrict. if srv.NetRestrict != nil { if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok && !srv.NetRestrict.Contains(tcp.IP) { srv.log.Debug("Rejected conn (not whitelisted in NetRestrict)", "addr", fd.RemoteAddr()) fd.Close() slots <- struct{}{} continue } } var ip net.IP if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok { ip = tcp.IP } fd = newMeteredConn(fd, true, ip) srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr()) go func() { srv.SetupConn(fd, inboundConn, nil) slots <- struct{}{} }() } } // SetupConn runs the handshakes and attempts to add the connection // as a peer. It returns when the connection has been added as a peer // or the handshakes have failed. func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node) error { c := &conn{fd: fd, transport: srv.newTransport(fd), flags: flags, cont: make(chan error)} err := srv.setupConn(c, flags, dialDest) if err != nil { c.close(err) srv.log.Trace("Setting up connection failed", "addr", fd.RemoteAddr(), "err", err) } return err } func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) error { // Prevent leftover pending conns from entering the handshake. srv.lock.Lock() running := srv.running srv.lock.Unlock() if !running { return errServerStopped } // If dialing, figure out the remote public key. var dialPubkey *ecdsa.PublicKey if dialDest != nil { dialPubkey = new(ecdsa.PublicKey) if err := dialDest.Load((*enode.Secp256k1)(dialPubkey)); err != nil { return errors.New("dial destination doesn't have a secp256k1 public key") } } // Run the encryption handshake. remotePubkey, err := c.doEncHandshake(srv.PrivateKey, dialPubkey) if err != nil { srv.log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err) return err } if dialDest != nil { // For dialed connections, check that the remote public key matches. if dialPubkey.X.Cmp(remotePubkey.X) != 0 || dialPubkey.Y.Cmp(remotePubkey.Y) != 0 { return DiscUnexpectedIdentity } c.node = dialDest } else { c.node = nodeFromConn(remotePubkey, c.fd) } if conn, ok := c.fd.(*meteredConn); ok { conn.handshakeDone(c.node.ID()) } clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags) err = srv.checkpoint(c, srv.posthandshake) if err != nil { clog.Trace("Rejected peer before protocol handshake", "err", err) return err } // Run the protocol handshake phs, err := c.doProtoHandshake(srv.ourHandshake) if err != nil { clog.Trace("Failed proto handshake", "err", err) return err } if id := c.node.ID(); !bytes.Equal(crypto.Keccak256(phs.ID), id[:]) { clog.Trace("Wrong devp2p handshake identity", "phsid", hex.EncodeToString(phs.ID)) return DiscUnexpectedIdentity } c.caps, c.name = phs.Caps, phs.Name err = srv.checkpoint(c, srv.addpeer) if err != nil { clog.Trace("Rejected peer", "err", err) return err } // If the checks completed successfully, runPeer has now been // launched by run. clog.Trace("connection set up", "inbound", dialDest == nil) return nil } func nodeFromConn(pubkey *ecdsa.PublicKey, conn net.Conn) *enode.Node { var ip net.IP var port int if tcp, ok := conn.RemoteAddr().(*net.TCPAddr); ok { ip = tcp.IP port = tcp.Port } return enode.NewV4(pubkey, ip, port, port) } func truncateName(s string) string { if len(s) > 20 { return s[:20] + "..." } return s } // checkpoint sends the conn to run, which performs the // post-handshake checks for the stage (posthandshake, addpeer). func (srv *Server) checkpoint(c *conn, stage chan<- *conn) error { select { case stage <- c: case <-srv.quit: return errServerStopped } select { case err := <-c.cont: return err case <-srv.quit: return errServerStopped } } // runPeer runs in its own goroutine for each peer. // it waits until the Peer logic returns and removes // the peer. func (srv *Server) runPeer(p *Peer) { if srv.newPeerHook != nil { srv.newPeerHook(p) } // broadcast peer add srv.peerFeed.Send(&PeerEvent{ Type: PeerEventTypeAdd, Peer: p.ID(), }) // run the protocol remoteRequested, err := p.run() // broadcast peer drop srv.peerFeed.Send(&PeerEvent{ Type: PeerEventTypeDrop, Peer: p.ID(), Error: err.Error(), }) // Note: run waits for existing peers to be sent on srv.delpeer // before returning, so this send should not select on srv.quit. srv.delpeer <- peerDrop{p, err, remoteRequested} } // NodeInfo represents a short summary of the information known about the host. type NodeInfo struct { ID string `json:"id"` // Unique node identifier (also the encryption key) Name string `json:"name"` // Name of the node, including client type, version, OS, custom data Enode string `json:"enode"` // Enode URL for adding this peer from remote peers ENR string `json:"enr"` // Ethereum Node Record IP string `json:"ip"` // IP address of the node Ports struct { Discovery int `json:"discovery"` // UDP listening port for discovery protocol Listener int `json:"listener"` // TCP listening port for RLPx } `json:"ports"` ListenAddr string `json:"listenAddr"` Protocols map[string]interface{} `json:"protocols"` } // NodeInfo gathers and returns a collection of metadata known about the host. func (srv *Server) NodeInfo() *NodeInfo { // Gather and assemble the generic node infos node := srv.Self() info := &NodeInfo{ Name: srv.Name, Enode: node.String(), ID: node.ID().String(), IP: node.IP().String(), ListenAddr: srv.ListenAddr, Protocols: make(map[string]interface{}), } info.Ports.Discovery = node.UDP() info.Ports.Listener = node.TCP() if enc, err := rlp.EncodeToBytes(node.Record()); err == nil { info.ENR = "0x" + hex.EncodeToString(enc) } // Gather all the running protocol infos (only once per protocol type) for _, proto := range srv.Protocols { if _, ok := info.Protocols[proto.Name]; !ok { nodeInfo := interface{}("unknown") if query := proto.NodeInfo; query != nil { nodeInfo = proto.NodeInfo() } info.Protocols[proto.Name] = nodeInfo } } return info } // PeersInfo returns an array of metadata objects describing connected peers. func (srv *Server) PeersInfo() []*PeerInfo { // Gather all the generic and sub-protocol specific infos infos := make([]*PeerInfo, 0, srv.PeerCount()) for _, peer := range srv.Peers() { if peer != nil { infos = append(infos, peer.Info()) } } // Sort the result array alphabetically by node identifier for i := 0; i < len(infos); i++ { for j := i + 1; j < len(infos); j++ { if infos[i].ID > infos[j].ID { infos[i], infos[j] = infos[j], infos[i] } } } return infos }