12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199 |
- // 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 <http://www.gnu.org/licenses/>.
- // Package p2p implements the Ethereum p2p network protocols.
- package p2p
- import (
- "bytes"
- "crypto/ecdsa"
- "encoding/hex"
- "errors"
- "fmt"
- "net"
- "sort"
- "sync"
- "sync/atomic"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/common/mclock"
- "github.com/ethereum/go-ethereum/crypto"
- "github.com/ethereum/go-ethereum/event"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/p2p/discover"
- "github.com/ethereum/go-ethereum/p2p/enode"
- "github.com/ethereum/go-ethereum/p2p/enr"
- "github.com/ethereum/go-ethereum/p2p/nat"
- "github.com/ethereum/go-ethereum/p2p/netutil"
- "github.com/ethereum/go-ethereum/permission/core"
- )
- const (
- defaultDialTimeout = 15 * time.Second
- // This is the fairness knob for the discovery mixer. When looking for peers, we'll
- // wait this long for a single source of candidates before moving on and trying other
- // sources.
- discmixTimeout = 5 * time.Second
- // Connectivity defaults.
- defaultMaxPendingPeers = 50
- defaultDialRatio = 3
- // This time limits inbound connection attempts per source IP.
- inboundThrottleTime = 30 * time.Second
- // 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 []*enode.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
- EnableNodePermission bool `toml:",omitempty"`
- DataDir string `toml:",omitempty"`
- // Logger is a custom logger to use with the p2p.Server.
- Logger log.Logger `toml:",omitempty"`
- clock mclock.Clock
- }
- // 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, *ecdsa.PublicKey) transport
- newPeerHook func(*Peer)
- listenFunc func(network, addr string) (net.Listener, error)
- lock sync.Mutex // protects running
- running bool
- listener net.Listener
- ourHandshake *protoHandshake
- loopWG sync.WaitGroup // loop, listenLoop
- peerFeed event.Feed
- log log.Logger
- nodedb *enode.DB
- localnode *enode.LocalNode
- ntab *discover.UDPv4
- DiscV5 *discover.UDPv5
- discmix *enode.FairMix
- dialsched *dialScheduler
- // Channels into the run loop.
- quit chan struct{}
- addtrusted chan *enode.Node
- removetrusted chan *enode.Node
- peerOp chan peerOpFunc
- peerOpDone chan struct{}
- delpeer chan peerDrop
- checkpointPostHandshake chan *conn
- checkpointAddPeer chan *conn
- // State of run loop and listenLoop.
- inboundHistory expHeap
- // raft peers info
- checkPeerInRaft func(*enode.Node) bool
- // permissions - check if node is permissioned
- isNodePermissionedFunc func(node *enode.Node, nodename string, currentNode string, datadir string, direction string) bool
- }
- 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
- 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) (*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&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
- }
- }
- }
- // LocalNode returns the local node record.
- func (srv *Server) LocalNode() *enode.LocalNode {
- return srv.localnode
- }
- // Peers returns all connected peers.
- func (srv *Server) Peers() []*Peer {
- var ps []*Peer
- srv.doPeerOp(func(peers map[enode.ID]*Peer) {
- for _, p := range peers {
- ps = append(ps, p)
- }
- })
- return ps
- }
- // PeerCount returns the number of connected peers.
- func (srv *Server) PeerCount() int {
- var count int
- srv.doPeerOp(func(ps map[enode.ID]*Peer) {
- count = len(ps)
- })
- return count
- }
- // AddPeer adds the given node to the static node set. When there is room in the peer set,
- // the server will connect to the node. If the connection fails for any reason, the server
- // will attempt to reconnect the peer.
- func (srv *Server) AddPeer(node *enode.Node) {
- srv.dialsched.addStatic(node)
- }
- // RemovePeer removes a node from the static node set. It also disconnects from the given
- // node if it is currently connected as a peer.
- //
- // This method blocks until all protocols have exited and the peer is removed. Do not use
- // RemovePeer in protocol implementations, call Disconnect on the Peer instead.
- func (srv *Server) RemovePeer(node *enode.Node) {
- var (
- ch chan *PeerEvent
- sub event.Subscription
- )
- // Disconnect the peer on the main loop.
- srv.doPeerOp(func(peers map[enode.ID]*Peer) {
- srv.dialsched.removeStatic(node)
- if peer := peers[node.ID()]; peer != nil {
- ch = make(chan *PeerEvent, 1)
- sub = srv.peerFeed.Subscribe(ch)
- peer.Disconnect(DiscRequested)
- }
- })
- // Wait for the peer connection to end.
- if ch != nil {
- defer sub.Unsubscribe()
- for ev := range ch {
- if ev.Peer == node.ID() && ev.Type == PeerEventTypeDrop {
- return
- }
- }
- }
- }
- // 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()
- }
- // 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 discover.UDPConn
- 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 discover.UDPConn
- 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.Root()
- }
- if srv.clock == nil {
- srv.clock = mclock.System{}
- }
- 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.listenFunc == nil {
- srv.listenFunc = net.Listen
- }
- srv.quit = make(chan struct{})
- srv.delpeer = make(chan peerDrop)
- srv.checkpointPostHandshake = make(chan *conn)
- srv.checkpointAddPeer = make(chan *conn)
- 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
- }
- srv.setupDialScheduler()
- srv.loopWG.Add(1)
- go srv.run()
- 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})
- // 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 {
- srv.discmix = enode.NewFairMix(discmixTimeout)
- // Add protocol-specific discovery sources.
- added := make(map[string]bool)
- for _, proto := range srv.Protocols {
- if proto.DialCandidates != nil && !added[proto.Name] {
- srv.discmix.AddSource(proto.DialCandidates)
- added[proto.Name] = true
- }
- }
- // Don't listen on UDP endpoint if DHT is disabled.
- 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() {
- srv.loopWG.Add(1)
- go func() {
- nat.Map(srv.NAT, srv.quit, "udp", realaddr.Port, realaddr.Port, "ethereum discovery")
- srv.loopWG.Done()
- }()
- }
- }
- 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,
- Log: srv.log,
- }
- ntab, err := discover.ListenV4(conn, srv.localnode, cfg)
- if err != nil {
- return err
- }
- srv.ntab = ntab
- srv.discmix.AddSource(ntab.RandomNodes())
- }
- // Discovery V5
- if srv.DiscoveryV5 {
- cfg := discover.Config{
- PrivateKey: srv.PrivateKey,
- NetRestrict: srv.NetRestrict,
- Bootnodes: srv.BootstrapNodesV5,
- Log: srv.log,
- }
- var err error
- if sconn != nil {
- srv.DiscV5, err = discover.ListenV5(sconn, srv.localnode, cfg)
- } else {
- srv.DiscV5, err = discover.ListenV5(conn, srv.localnode, cfg)
- }
- if err != nil {
- return err
- }
- }
- return nil
- }
- func (srv *Server) setupDialScheduler() {
- config := dialConfig{
- self: srv.localnode.ID(),
- maxDialPeers: srv.maxDialedConns(),
- maxActiveDials: srv.MaxPendingPeers,
- log: srv.Logger,
- netRestrict: srv.NetRestrict,
- dialer: srv.Dialer,
- clock: srv.clock,
- }
- if srv.ntab != nil {
- config.resolver = srv.ntab
- }
- if config.dialer == nil {
- config.dialer = tcpDialer{&net.Dialer{Timeout: defaultDialTimeout}}
- }
- srv.dialsched = newDialScheduler(config, srv.discmix, srv.SetupConn)
- for _, n := range srv.StaticNodes {
- srv.dialsched.addStatic(n)
- }
- }
- func (srv *Server) maxInboundConns() int {
- return srv.MaxPeers - srv.maxDialedConns()
- }
- func (srv *Server) maxDialedConns() (limit int) {
- if srv.NoDial || srv.MaxPeers == 0 {
- return 0
- }
- if srv.DialRatio == 0 {
- limit = srv.MaxPeers / defaultDialRatio
- } else {
- limit = srv.MaxPeers / srv.DialRatio
- }
- if limit == 0 {
- limit = 1
- }
- return limit
- }
- func (srv *Server) setupListening() error {
- // Launch the listener.
- listener, err := srv.listenFunc("tcp", srv.ListenAddr)
- if err != nil {
- return err
- }
- srv.listener = listener
- srv.ListenAddr = listener.Addr().String()
- // Update the local node record and map the TCP listening port if NAT is configured.
- if tcp, ok := listener.Addr().(*net.TCPAddr); ok {
- srv.localnode.Set(enr.TCP(tcp.Port))
- if !tcp.IP.IsLoopback() && srv.NAT != nil {
- srv.loopWG.Add(1)
- go func() {
- nat.Map(srv.NAT, srv.quit, "tcp", tcp.Port, tcp.Port, "ethereum p2p")
- srv.loopWG.Done()
- }()
- }
- }
- srv.loopWG.Add(1)
- go srv.listenLoop()
- return nil
- }
- // doPeerOp runs fn on the main loop.
- func (srv *Server) doPeerOp(fn peerOpFunc) {
- select {
- case srv.peerOp <- fn:
- <-srv.peerOpDone
- case <-srv.quit:
- }
- }
- // run is the main loop of the server.
- func (srv *Server) run() {
- srv.log.Info("Started P2P networking", "self", srv.localnode.Node().URLv4())
- defer srv.loopWG.Done()
- defer srv.nodedb.Close()
- defer srv.discmix.Close()
- defer srv.dialsched.stop()
- var (
- peers = make(map[enode.ID]*Peer)
- inboundCount = 0
- trusted = make(map[enode.ID]bool, len(srv.TrustedNodes))
- )
- // 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
- }
- running:
- for {
- select {
- case <-srv.quit:
- // The server was stopped. Run the cleanup logic.
- break running
- case n := <-srv.addtrusted:
- // This channel is used by AddTrustedPeer to add a node
- // to the trusted node set.
- srv.log.Trace("Adding trusted node", "node", n)
- trusted[n.ID()] = true
- if p, ok := peers[n.ID()]; ok {
- p.rw.set(trustedConn, true)
- }
- case n := <-srv.removetrusted:
- // This channel is used by RemoveTrustedPeer to remove a node
- // from the trusted node set.
- srv.log.Trace("Removing trusted node", "node", n)
- delete(trusted, n.ID())
- if p, ok := peers[n.ID()]; ok {
- p.rw.set(trustedConn, false)
- }
- case op := <-srv.peerOp:
- // This channel is used by Peers and PeerCount.
- op(peers)
- srv.peerOpDone <- struct{}{}
- case c := <-srv.checkpointPostHandshake:
- // A connection has passed the encryption handshake so
- // the remote identity is known (but hasn't been verified yet).
- if trusted[c.node.ID()] {
- // Ensure that the trusted flag is set before checking against MaxPeers.
- c.flags |= trustedConn
- }
- // TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
- c.cont <- srv.postHandshakeChecks(peers, inboundCount, c)
- case c := <-srv.checkpointAddPeer:
- // At this point the connection is past the protocol handshake.
- // Its capabilities are known and the remote identity is verified.
- err := srv.addPeerChecks(peers, inboundCount, c)
- if err == nil {
- // The handshakes are done and it passed all checks.
- p := srv.launchPeer(c)
- peers[c.node.ID()] = p
- srv.log.Debug("Adding p2p peer", "peercount", len(peers), "id", p.ID(), "conn", c.flags, "addr", p.RemoteAddr(), "name", p.Name())
- srv.dialsched.peerAdded(c)
- if p.Inbound() {
- inboundCount++
- }
- }
- c.cont <- err
- case pd := <-srv.delpeer:
- // A peer disconnected.
- d := common.PrettyDuration(mclock.Now() - pd.created)
- delete(peers, pd.ID())
- srv.log.Debug("Removing p2p peer", "peercount", len(peers), "id", pd.ID(), "duration", d, "req", pd.requested, "err", pd.err)
- srv.dialsched.peerRemoved(pd.rw)
- 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)")
- delete(peers, p.ID())
- }
- }
- func (srv *Server) postHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int, c *conn) error {
- switch {
- case !c.is(trustedConn) && len(peers) >= srv.MaxPeers:
- return DiscTooManyPeers
- case !c.is(trustedConn) && 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) addPeerChecks(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 post-handshake checks because the
- // peer set might have changed since those checks were performed.
- return srv.postHandshakeChecks(peers, inboundCount, c)
- }
- // listenLoop runs in its own goroutine and accepts
- // inbound connections.
- func (srv *Server) listenLoop() {
- srv.log.Debug("TCP listener up", "addr", srv.listener.Addr())
- // The slots channel limits accepts of new connections.
- tokens := defaultMaxPendingPeers
- if srv.MaxPendingPeers > 0 {
- tokens = srv.MaxPendingPeers
- }
- slots := make(chan struct{}, tokens)
- for i := 0; i < tokens; i++ {
- slots <- struct{}{}
- }
- // Wait for slots to be returned on exit. This ensures all connection goroutines
- // are down before listenLoop returns.
- defer srv.loopWG.Done()
- defer func() {
- for i := 0; i < cap(slots); i++ {
- <-slots
- }
- }()
- for {
- // Wait for a free slot before accepting.
- <-slots
- var (
- fd net.Conn
- err error
- lastLog time.Time
- )
- for {
- fd, err = srv.listener.Accept()
- if netutil.IsTemporaryError(err) {
- if time.Since(lastLog) > 1*time.Second {
- srv.log.Debug("Temporary read error", "err", err)
- lastLog = time.Now()
- }
- time.Sleep(time.Millisecond * 200)
- continue
- } else if err != nil {
- srv.log.Debug("Read error", "err", err)
- slots <- struct{}{}
- return
- }
- break
- }
- remoteIP := netutil.AddrIP(fd.RemoteAddr())
- if err := srv.checkInboundConn(remoteIP); err != nil {
- srv.log.Debug("Rejected inbound connection", "addr", fd.RemoteAddr(), "err", err)
- fd.Close()
- slots <- struct{}{}
- continue
- }
- if remoteIP != nil {
- var addr *net.TCPAddr
- if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok {
- addr = tcp
- }
- fd = newMeteredConn(fd, true, addr)
- srv.log.Trace("Accepted connection", "addr", fd.RemoteAddr())
- }
- go func() {
- srv.SetupConn(fd, inboundConn, nil)
- slots <- struct{}{}
- }()
- }
- }
- func (srv *Server) checkInboundConn(remoteIP net.IP) error {
- if remoteIP == nil {
- return nil
- }
- // Reject connections that do not match NetRestrict.
- if srv.NetRestrict != nil && !srv.NetRestrict.Contains(remoteIP) {
- return fmt.Errorf("not whitelisted in NetRestrict")
- }
- // Reject Internet peers that try too often.
- now := srv.clock.Now()
- srv.inboundHistory.expire(now, nil)
- if !netutil.IsLAN(remoteIP) && srv.inboundHistory.contains(remoteIP.String()) {
- return fmt.Errorf("too many attempts")
- }
- srv.inboundHistory.add(remoteIP.String(), now.Add(inboundThrottleTime))
- return nil
- }
- // 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, flags: flags, cont: make(chan error)}
- if dialDest == nil {
- c.transport = srv.newTransport(fd, nil)
- } else {
- c.transport = srv.newTransport(fd, dialDest.Pubkey())
- }
- err := srv.setupConn(c, flags, dialDest)
- if err != nil {
- c.close(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 {
- err = errors.New("dial destination doesn't have a secp256k1 public key")
- srv.log.Trace("Setting up connection failed", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
- return err
- }
- }
- // Run the RLPx handshake.
- remotePubkey, err := c.doEncHandshake(srv.PrivateKey)
- if err != nil {
- srv.log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
- return err
- }
- if dialDest != nil {
- c.node = dialDest
- } else {
- c.node = nodeFromConn(remotePubkey, c.fd)
- }
- clog := srv.log.New("id", c.node.ID(), "addr", c.fd.RemoteAddr(), "conn", c.flags)
- // If raft is running, check if the dialing node is in the raft cluster
- // Node doesn't belong to raft cluster is not allowed to join the p2p network
- if srv.checkPeerInRaft != nil && !srv.checkPeerInRaft(c.node) {
- node := c.node.ID().String()
- log.Trace("incoming connection peer is not in the raft cluster", "enode.id", node)
- return newPeerError(errNotInRaftCluster, "id=%s…%s", node[:4], node[len(node)-4:])
- }
- //START - QUORUM Permissioning
- currentNode := srv.NodeInfo().ID
- cnodeName := srv.NodeInfo().Name
- clog.Trace("Quorum permissioning",
- "EnableNodePermission", srv.EnableNodePermission,
- "DataDir", srv.DataDir,
- "Current Node ID", currentNode,
- "Node Name", cnodeName,
- "Dialed Dest", dialDest,
- "Connection ID", c.node.ID(),
- "Connection String", c.node.ID().String())
- if srv.EnableNodePermission {
- clog.Trace("Node Permissioning is Enabled.")
- nodeId := c.node.ID().String()
- node := c.node
- direction := "INCOMING"
- if dialDest != nil {
- node = dialDest
- nodeId = dialDest.ID().String()
- direction = "OUTGOING"
- log.Trace("Node Permissioning", "Connection Direction", direction)
- }
- if srv.isNodePermissionedFunc == nil {
- if !core.IsNodePermissioned(nodeId, currentNode, srv.DataDir, direction) {
- return newPeerError(errPermissionDenied, "id=%s…%s %s id=%s…%s", currentNode[:4], currentNode[len(currentNode)-4:], direction, nodeId[:4], nodeId[len(nodeId)-4:])
- }
- } else if !srv.isNodePermissionedFunc(node, nodeId, currentNode, srv.DataDir, direction) {
- return newPeerError(errPermissionDenied, "id=%s…%s %s id=%s…%s", currentNode[:4], currentNode[len(currentNode)-4:], direction, nodeId[:4], nodeId[len(nodeId)-4:])
- }
- } else {
- clog.Trace("Node Permissioning is Disabled.")
- }
- //END - QUORUM Permissioning
- err = srv.checkpoint(c, srv.checkpointPostHandshake)
- if err != nil {
- clog.Trace("Rejected peer", "err", err)
- return err
- }
- // Run the capability negotiation handshake.
- phs, err := c.doProtoHandshake(srv.ourHandshake)
- if err != nil {
- clog.Trace("Failed p2p 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
- }
- // To continue support of IBFT1.0 with besu
- if len(phs.Caps) == 1 && phs.Caps[0].Name == "istanbul" && phs.Caps[0].Version == 99 {
- phs.Caps = []Cap{{
- Name: "eth",
- Version: 64,
- }}
- }
- c.caps, c.name = phs.Caps, phs.Name
- err = srv.checkpoint(c, srv.checkpointAddPeer)
- if err != nil {
- clog.Trace("Rejected peer", "err", err)
- return err
- }
- 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)
- }
- // 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
- }
- return <-c.cont
- }
- func (srv *Server) launchPeer(c *conn) *Peer {
- p := newPeer(srv.log, c, srv.Protocols)
- if srv.EnableMsgEvents {
- // If message events are enabled, pass the peerFeed
- // to the peer.
- p.events = &srv.peerFeed
- }
- go srv.runPeer(p)
- return p
- }
- // runPeer runs in its own goroutine for each peer.
- func (srv *Server) runPeer(p *Peer) {
- if srv.newPeerHook != nil {
- srv.newPeerHook(p)
- }
- srv.peerFeed.Send(&PeerEvent{
- Type: PeerEventTypeAdd,
- Peer: p.ID(),
- RemoteAddress: p.RemoteAddr().String(),
- LocalAddress: p.LocalAddr().String(),
- })
- // Run the per-peer main loop.
- remoteRequested, err := p.run()
- // Announce disconnect on the main loop to update the peer set.
- // The main loop 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}
- // Broadcast peer drop to external subscribers. This needs to be
- // after the send to delpeer so subscribers have a consistent view of
- // the peer set (i.e. Server.Peers() doesn't include the peer when the
- // event is received.
- srv.peerFeed.Send(&PeerEvent{
- Type: PeerEventTypeDrop,
- Peer: p.ID(),
- Error: err.Error(),
- RemoteAddress: p.RemoteAddr().String(),
- LocalAddress: p.LocalAddr().String(),
- })
- }
- // 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.URLv4(),
- 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()
- info.ENR = node.String()
- // 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
- }
- func (srv *Server) SetCheckPeerInRaft(f func(*enode.Node) bool) {
- srv.checkPeerInRaft = f
- }
- func (srv *Server) SetIsNodePermissioned(f func(*enode.Node, string, string, string, string) bool) {
- if srv.isNodePermissionedFunc == nil {
- srv.isNodePermissionedFunc = f
- }
- }
- func (srv *Server) SetNewTransportFunc(f func(net.Conn, *ecdsa.PublicKey) transport) {
- srv.newTransport = f
- }
|