server.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825
  1. // Copyright 2014 The go-ethereum Authors
  2. // This file is part of the go-ethereum library.
  3. //
  4. // The go-ethereum library is free software: you can redistribute it and/or modify
  5. // it under the terms of the GNU Lesser General Public License as published by
  6. // the Free Software Foundation, either version 3 of the License, or
  7. // (at your option) any later version.
  8. //
  9. // The go-ethereum library is distributed in the hope that it will be useful,
  10. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. // GNU Lesser General Public License for more details.
  13. //
  14. // You should have received a copy of the GNU Lesser General Public License
  15. // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
  16. // Package p2p implements the Ethereum p2p network protocols.
  17. package p2p
  18. import (
  19. "crypto/ecdsa"
  20. "errors"
  21. "fmt"
  22. "net"
  23. "sync"
  24. "time"
  25. "github.com/ethereum/go-ethereum/common"
  26. "github.com/ethereum/go-ethereum/common/mclock"
  27. "github.com/ethereum/go-ethereum/log"
  28. "github.com/ethereum/go-ethereum/p2p/discover"
  29. "github.com/ethereum/go-ethereum/p2p/discv5"
  30. "github.com/ethereum/go-ethereum/p2p/nat"
  31. "github.com/ethereum/go-ethereum/p2p/netutil"
  32. )
  33. const (
  34. defaultDialTimeout = 15 * time.Second
  35. refreshPeersInterval = 30 * time.Second
  36. staticPeerCheckInterval = 15 * time.Second
  37. // Maximum number of concurrently handshaking inbound connections.
  38. maxAcceptConns = 50
  39. // Maximum number of concurrently dialing outbound connections.
  40. maxActiveDialTasks = 16
  41. // Maximum time allowed for reading a complete message.
  42. // This is effectively the amount of time a connection can be idle.
  43. frameReadTimeout = 30 * time.Second
  44. // Maximum amount of time allowed for writing a complete message.
  45. frameWriteTimeout = 20 * time.Second
  46. )
  47. var errServerStopped = errors.New("server stopped")
  48. // Config holds Server options.
  49. type Config struct {
  50. // This field must be set to a valid secp256k1 private key.
  51. PrivateKey *ecdsa.PrivateKey
  52. // MaxPeers is the maximum number of peers that can be
  53. // connected. It must be greater than zero.
  54. MaxPeers int
  55. // MaxPendingPeers is the maximum number of peers that can be pending in the
  56. // handshake phase, counted separately for inbound and outbound connections.
  57. // Zero defaults to preset values.
  58. MaxPendingPeers int
  59. // Discovery specifies whether the peer discovery mechanism should be started
  60. // or not. Disabling is usually useful for protocol debugging (manual topology).
  61. Discovery bool
  62. // DiscoveryV5 specifies whether the the new topic-discovery based V5 discovery
  63. // protocol should be started or not.
  64. DiscoveryV5 bool
  65. // Listener address for the V5 discovery protocol UDP traffic.
  66. DiscoveryV5Addr string
  67. // Name sets the node name of this server.
  68. // Use common.MakeName to create a name that follows existing conventions.
  69. Name string
  70. // BootstrapNodes are used to establish connectivity
  71. // with the rest of the network.
  72. BootstrapNodes []*discover.Node
  73. // BootstrapNodesV5 are used to establish connectivity
  74. // with the rest of the network using the V5 discovery
  75. // protocol.
  76. BootstrapNodesV5 []*discv5.Node
  77. // Static nodes are used as pre-configured connections which are always
  78. // maintained and re-connected on disconnects.
  79. StaticNodes []*discover.Node
  80. // Trusted nodes are used as pre-configured connections which are always
  81. // allowed to connect, even above the peer limit.
  82. TrustedNodes []*discover.Node
  83. // Connectivity can be restricted to certain IP networks.
  84. // If this option is set to a non-nil value, only hosts which match one of the
  85. // IP networks contained in the list are considered.
  86. NetRestrict *netutil.Netlist
  87. // NodeDatabase is the path to the database containing the previously seen
  88. // live nodes in the network.
  89. NodeDatabase string
  90. // Protocols should contain the protocols supported
  91. // by the server. Matching protocols are launched for
  92. // each peer.
  93. Protocols []Protocol
  94. // If ListenAddr is set to a non-nil address, the server
  95. // will listen for incoming connections.
  96. //
  97. // If the port is zero, the operating system will pick a port. The
  98. // ListenAddr field will be updated with the actual address when
  99. // the server is started.
  100. ListenAddr string
  101. // If set to a non-nil value, the given NAT port mapper
  102. // is used to make the listening port available to the
  103. // Internet.
  104. NAT nat.Interface
  105. // If Dialer is set to a non-nil value, the given Dialer
  106. // is used to dial outbound peer connections.
  107. Dialer *net.Dialer
  108. // If NoDial is true, the server will not dial any peers.
  109. NoDial bool
  110. }
  111. // Server manages all peer connections.
  112. type Server struct {
  113. // Config fields may not be modified while the server is running.
  114. Config
  115. // Hooks for testing. These are useful because we can inhibit
  116. // the whole protocol stack.
  117. newTransport func(net.Conn) transport
  118. newPeerHook func(*Peer)
  119. lock sync.Mutex // protects running
  120. running bool
  121. ntab discoverTable
  122. listener net.Listener
  123. ourHandshake *protoHandshake
  124. lastLookup time.Time
  125. DiscV5 *discv5.Network
  126. // These are for Peers, PeerCount (and nothing else).
  127. peerOp chan peerOpFunc
  128. peerOpDone chan struct{}
  129. quit chan struct{}
  130. addstatic chan *discover.Node
  131. removestatic chan *discover.Node
  132. posthandshake chan *conn
  133. addpeer chan *conn
  134. delpeer chan peerDrop
  135. loopWG sync.WaitGroup // loop, listenLoop
  136. }
  137. type peerOpFunc func(map[discover.NodeID]*Peer)
  138. type peerDrop struct {
  139. *Peer
  140. err error
  141. requested bool // true if signaled by the peer
  142. }
  143. type connFlag int
  144. const (
  145. dynDialedConn connFlag = 1 << iota
  146. staticDialedConn
  147. inboundConn
  148. trustedConn
  149. )
  150. // conn wraps a network connection with information gathered
  151. // during the two handshakes.
  152. type conn struct {
  153. fd net.Conn
  154. transport
  155. flags connFlag
  156. cont chan error // The run loop uses cont to signal errors to setupConn.
  157. id discover.NodeID // valid after the encryption handshake
  158. caps []Cap // valid after the protocol handshake
  159. name string // valid after the protocol handshake
  160. }
  161. type transport interface {
  162. // The two handshakes.
  163. doEncHandshake(prv *ecdsa.PrivateKey, dialDest *discover.Node) (discover.NodeID, error)
  164. doProtoHandshake(our *protoHandshake) (*protoHandshake, error)
  165. // The MsgReadWriter can only be used after the encryption
  166. // handshake has completed. The code uses conn.id to track this
  167. // by setting it to a non-nil value after the encryption handshake.
  168. MsgReadWriter
  169. // transports must provide Close because we use MsgPipe in some of
  170. // the tests. Closing the actual network connection doesn't do
  171. // anything in those tests because NsgPipe doesn't use it.
  172. close(err error)
  173. }
  174. func (c *conn) String() string {
  175. s := c.flags.String()
  176. if (c.id != discover.NodeID{}) {
  177. s += " " + c.id.String()
  178. }
  179. s += " " + c.fd.RemoteAddr().String()
  180. return s
  181. }
  182. func (f connFlag) String() string {
  183. s := ""
  184. if f&trustedConn != 0 {
  185. s += "-trusted"
  186. }
  187. if f&dynDialedConn != 0 {
  188. s += "-dyndial"
  189. }
  190. if f&staticDialedConn != 0 {
  191. s += "-staticdial"
  192. }
  193. if f&inboundConn != 0 {
  194. s += "-inbound"
  195. }
  196. if s != "" {
  197. s = s[1:]
  198. }
  199. return s
  200. }
  201. func (c *conn) is(f connFlag) bool {
  202. return c.flags&f != 0
  203. }
  204. // Peers returns all connected peers.
  205. func (srv *Server) Peers() []*Peer {
  206. var ps []*Peer
  207. select {
  208. // Note: We'd love to put this function into a variable but
  209. // that seems to cause a weird compiler error in some
  210. // environments.
  211. case srv.peerOp <- func(peers map[discover.NodeID]*Peer) {
  212. for _, p := range peers {
  213. ps = append(ps, p)
  214. }
  215. }:
  216. <-srv.peerOpDone
  217. case <-srv.quit:
  218. }
  219. return ps
  220. }
  221. // PeerCount returns the number of connected peers.
  222. func (srv *Server) PeerCount() int {
  223. var count int
  224. select {
  225. case srv.peerOp <- func(ps map[discover.NodeID]*Peer) { count = len(ps) }:
  226. <-srv.peerOpDone
  227. case <-srv.quit:
  228. }
  229. return count
  230. }
  231. // AddPeer connects to the given node and maintains the connection until the
  232. // server is shut down. If the connection fails for any reason, the server will
  233. // attempt to reconnect the peer.
  234. func (srv *Server) AddPeer(node *discover.Node) {
  235. select {
  236. case srv.addstatic <- node:
  237. case <-srv.quit:
  238. }
  239. }
  240. // RemovePeer disconnects from the given node
  241. func (srv *Server) RemovePeer(node *discover.Node) {
  242. select {
  243. case srv.removestatic <- node:
  244. case <-srv.quit:
  245. }
  246. }
  247. // Self returns the local node's endpoint information.
  248. func (srv *Server) Self() *discover.Node {
  249. srv.lock.Lock()
  250. defer srv.lock.Unlock()
  251. if !srv.running {
  252. return &discover.Node{IP: net.ParseIP("0.0.0.0")}
  253. }
  254. return srv.makeSelf(srv.listener, srv.ntab)
  255. }
  256. func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *discover.Node {
  257. // If the server's not running, return an empty node.
  258. // If the node is running but discovery is off, manually assemble the node infos.
  259. if ntab == nil {
  260. // Inbound connections disabled, use zero address.
  261. if listener == nil {
  262. return &discover.Node{IP: net.ParseIP("0.0.0.0"), ID: discover.PubkeyID(&srv.PrivateKey.PublicKey)}
  263. }
  264. // Otherwise inject the listener address too
  265. addr := listener.Addr().(*net.TCPAddr)
  266. return &discover.Node{
  267. ID: discover.PubkeyID(&srv.PrivateKey.PublicKey),
  268. IP: addr.IP,
  269. TCP: uint16(addr.Port),
  270. }
  271. }
  272. // Otherwise return the discovery node.
  273. return ntab.Self()
  274. }
  275. // Stop terminates the server and all active peer connections.
  276. // It blocks until all active connections have been closed.
  277. func (srv *Server) Stop() {
  278. srv.lock.Lock()
  279. defer srv.lock.Unlock()
  280. if !srv.running {
  281. return
  282. }
  283. srv.running = false
  284. if srv.listener != nil {
  285. // this unblocks listener Accept
  286. srv.listener.Close()
  287. }
  288. close(srv.quit)
  289. srv.loopWG.Wait()
  290. }
  291. // Start starts running the server.
  292. // Servers can not be re-used after stopping.
  293. func (srv *Server) Start() (err error) {
  294. srv.lock.Lock()
  295. defer srv.lock.Unlock()
  296. if srv.running {
  297. return errors.New("server already running")
  298. }
  299. srv.running = true
  300. log.Info("Starting P2P networking")
  301. // static fields
  302. if srv.PrivateKey == nil {
  303. return fmt.Errorf("Server.PrivateKey must be set to a non-nil key")
  304. }
  305. if srv.newTransport == nil {
  306. srv.newTransport = newRLPX
  307. }
  308. if srv.Dialer == nil {
  309. srv.Dialer = &net.Dialer{Timeout: defaultDialTimeout}
  310. }
  311. srv.quit = make(chan struct{})
  312. srv.addpeer = make(chan *conn)
  313. srv.delpeer = make(chan peerDrop)
  314. srv.posthandshake = make(chan *conn)
  315. srv.addstatic = make(chan *discover.Node)
  316. srv.removestatic = make(chan *discover.Node)
  317. srv.peerOp = make(chan peerOpFunc)
  318. srv.peerOpDone = make(chan struct{})
  319. // node table
  320. if srv.Discovery {
  321. ntab, err := discover.ListenUDP(srv.PrivateKey, srv.ListenAddr, srv.NAT, srv.NodeDatabase, srv.NetRestrict)
  322. if err != nil {
  323. return err
  324. }
  325. if err := ntab.SetFallbackNodes(srv.BootstrapNodes); err != nil {
  326. return err
  327. }
  328. srv.ntab = ntab
  329. }
  330. if srv.DiscoveryV5 {
  331. ntab, err := discv5.ListenUDP(srv.PrivateKey, srv.DiscoveryV5Addr, srv.NAT, "", srv.NetRestrict) //srv.NodeDatabase)
  332. if err != nil {
  333. return err
  334. }
  335. if err := ntab.SetFallbackNodes(srv.BootstrapNodesV5); err != nil {
  336. return err
  337. }
  338. srv.DiscV5 = ntab
  339. }
  340. dynPeers := (srv.MaxPeers + 1) / 2
  341. if !srv.Discovery {
  342. dynPeers = 0
  343. }
  344. dialer := newDialState(srv.StaticNodes, srv.ntab, dynPeers, srv.NetRestrict)
  345. // handshake
  346. srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: discover.PubkeyID(&srv.PrivateKey.PublicKey)}
  347. for _, p := range srv.Protocols {
  348. srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap())
  349. }
  350. // listen/dial
  351. if srv.ListenAddr != "" {
  352. if err := srv.startListening(); err != nil {
  353. return err
  354. }
  355. }
  356. if srv.NoDial && srv.ListenAddr == "" {
  357. log.Warn("P2P server will be useless, neither dialing nor listening")
  358. }
  359. srv.loopWG.Add(1)
  360. go srv.run(dialer)
  361. srv.running = true
  362. return nil
  363. }
  364. func (srv *Server) startListening() error {
  365. // Launch the TCP listener.
  366. listener, err := net.Listen("tcp", srv.ListenAddr)
  367. if err != nil {
  368. return err
  369. }
  370. laddr := listener.Addr().(*net.TCPAddr)
  371. srv.ListenAddr = laddr.String()
  372. srv.listener = listener
  373. srv.loopWG.Add(1)
  374. go srv.listenLoop()
  375. // Map the TCP listening port if NAT is configured.
  376. if !laddr.IP.IsLoopback() && srv.NAT != nil {
  377. srv.loopWG.Add(1)
  378. go func() {
  379. nat.Map(srv.NAT, srv.quit, "tcp", laddr.Port, laddr.Port, "ethereum p2p")
  380. srv.loopWG.Done()
  381. }()
  382. }
  383. return nil
  384. }
  385. type dialer interface {
  386. newTasks(running int, peers map[discover.NodeID]*Peer, now time.Time) []task
  387. taskDone(task, time.Time)
  388. addStatic(*discover.Node)
  389. removeStatic(*discover.Node)
  390. }
  391. func (srv *Server) run(dialstate dialer) {
  392. defer srv.loopWG.Done()
  393. var (
  394. peers = make(map[discover.NodeID]*Peer)
  395. trusted = make(map[discover.NodeID]bool, len(srv.TrustedNodes))
  396. taskdone = make(chan task, maxActiveDialTasks)
  397. runningTasks []task
  398. queuedTasks []task // tasks that can't run yet
  399. )
  400. // Put trusted nodes into a map to speed up checks.
  401. // Trusted peers are loaded on startup and cannot be
  402. // modified while the server is running.
  403. for _, n := range srv.TrustedNodes {
  404. trusted[n.ID] = true
  405. }
  406. // removes t from runningTasks
  407. delTask := func(t task) {
  408. for i := range runningTasks {
  409. if runningTasks[i] == t {
  410. runningTasks = append(runningTasks[:i], runningTasks[i+1:]...)
  411. break
  412. }
  413. }
  414. }
  415. // starts until max number of active tasks is satisfied
  416. startTasks := func(ts []task) (rest []task) {
  417. i := 0
  418. for ; len(runningTasks) < maxActiveDialTasks && i < len(ts); i++ {
  419. t := ts[i]
  420. log.Trace("New dial task", "task", t)
  421. go func() { t.Do(srv); taskdone <- t }()
  422. runningTasks = append(runningTasks, t)
  423. }
  424. return ts[i:]
  425. }
  426. scheduleTasks := func() {
  427. // Start from queue first.
  428. queuedTasks = append(queuedTasks[:0], startTasks(queuedTasks)...)
  429. // Query dialer for new tasks and start as many as possible now.
  430. if len(runningTasks) < maxActiveDialTasks {
  431. nt := dialstate.newTasks(len(runningTasks)+len(queuedTasks), peers, time.Now())
  432. queuedTasks = append(queuedTasks, startTasks(nt)...)
  433. }
  434. }
  435. running:
  436. for {
  437. scheduleTasks()
  438. select {
  439. case <-srv.quit:
  440. // The server was stopped. Run the cleanup logic.
  441. break running
  442. case n := <-srv.addstatic:
  443. // This channel is used by AddPeer to add to the
  444. // ephemeral static peer list. Add it to the dialer,
  445. // it will keep the node connected.
  446. log.Debug("Adding static node", "node", n)
  447. dialstate.addStatic(n)
  448. case n := <-srv.removestatic:
  449. // This channel is used by RemovePeer to send a
  450. // disconnect request to a peer and begin the
  451. // stop keeping the node connected
  452. log.Debug("Removing static node", "node", n)
  453. dialstate.removeStatic(n)
  454. if p, ok := peers[n.ID]; ok {
  455. p.Disconnect(DiscRequested)
  456. }
  457. case op := <-srv.peerOp:
  458. // This channel is used by Peers and PeerCount.
  459. op(peers)
  460. srv.peerOpDone <- struct{}{}
  461. case t := <-taskdone:
  462. // A task got done. Tell dialstate about it so it
  463. // can update its state and remove it from the active
  464. // tasks list.
  465. log.Trace("Dial task done", "task", t)
  466. dialstate.taskDone(t, time.Now())
  467. delTask(t)
  468. case c := <-srv.posthandshake:
  469. // A connection has passed the encryption handshake so
  470. // the remote identity is known (but hasn't been verified yet).
  471. if trusted[c.id] {
  472. // Ensure that the trusted flag is set before checking against MaxPeers.
  473. c.flags |= trustedConn
  474. }
  475. // TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
  476. c.cont <- srv.encHandshakeChecks(peers, c)
  477. case c := <-srv.addpeer:
  478. // At this point the connection is past the protocol handshake.
  479. // Its capabilities are known and the remote identity is verified.
  480. err := srv.protoHandshakeChecks(peers, c)
  481. if err == nil {
  482. // The handshakes are done and it passed all checks.
  483. p := newPeer(c, srv.Protocols)
  484. name := truncateName(c.name)
  485. log.Debug("Adding p2p peer", "id", c.id, "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1)
  486. peers[c.id] = p
  487. go srv.runPeer(p)
  488. }
  489. // The dialer logic relies on the assumption that
  490. // dial tasks complete after the peer has been added or
  491. // discarded. Unblock the task last.
  492. c.cont <- err
  493. case pd := <-srv.delpeer:
  494. // A peer disconnected.
  495. d := common.PrettyDuration(mclock.Now() - pd.created)
  496. pd.log.Debug("Removing p2p peer", "duration", d, "peers", len(peers)-1, "req", pd.requested, "err", pd.err)
  497. delete(peers, pd.ID())
  498. }
  499. }
  500. log.Trace("P2P networking is spinning down")
  501. // Terminate discovery. If there is a running lookup it will terminate soon.
  502. if srv.ntab != nil {
  503. srv.ntab.Close()
  504. }
  505. if srv.DiscV5 != nil {
  506. srv.DiscV5.Close()
  507. }
  508. // Disconnect all peers.
  509. for _, p := range peers {
  510. p.Disconnect(DiscQuitting)
  511. }
  512. // Wait for peers to shut down. Pending connections and tasks are
  513. // not handled here and will terminate soon-ish because srv.quit
  514. // is closed.
  515. for len(peers) > 0 {
  516. p := <-srv.delpeer
  517. p.log.Trace("<-delpeer (spindown)", "remainingTasks", len(runningTasks))
  518. delete(peers, p.ID())
  519. }
  520. }
  521. func (srv *Server) protoHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
  522. // Drop connections with no matching protocols.
  523. if len(srv.Protocols) > 0 && countMatchingProtocols(srv.Protocols, c.caps) == 0 {
  524. return DiscUselessPeer
  525. }
  526. // Repeat the encryption handshake checks because the
  527. // peer set might have changed between the handshakes.
  528. return srv.encHandshakeChecks(peers, c)
  529. }
  530. func (srv *Server) encHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
  531. switch {
  532. case !c.is(trustedConn|staticDialedConn) && len(peers) >= srv.MaxPeers:
  533. return DiscTooManyPeers
  534. case peers[c.id] != nil:
  535. return DiscAlreadyConnected
  536. case c.id == srv.Self().ID:
  537. return DiscSelf
  538. default:
  539. return nil
  540. }
  541. }
  542. type tempError interface {
  543. Temporary() bool
  544. }
  545. // listenLoop runs in its own goroutine and accepts
  546. // inbound connections.
  547. func (srv *Server) listenLoop() {
  548. defer srv.loopWG.Done()
  549. log.Info("RLPx listener up", "self", srv.makeSelf(srv.listener, srv.ntab))
  550. // This channel acts as a semaphore limiting
  551. // active inbound connections that are lingering pre-handshake.
  552. // If all slots are taken, no further connections are accepted.
  553. tokens := maxAcceptConns
  554. if srv.MaxPendingPeers > 0 {
  555. tokens = srv.MaxPendingPeers
  556. }
  557. slots := make(chan struct{}, tokens)
  558. for i := 0; i < tokens; i++ {
  559. slots <- struct{}{}
  560. }
  561. for {
  562. // Wait for a handshake slot before accepting.
  563. <-slots
  564. var (
  565. fd net.Conn
  566. err error
  567. )
  568. for {
  569. fd, err = srv.listener.Accept()
  570. if tempErr, ok := err.(tempError); ok && tempErr.Temporary() {
  571. log.Debug("Temporary read error", "err", err)
  572. continue
  573. } else if err != nil {
  574. log.Debug("Read error", "err", err)
  575. return
  576. }
  577. break
  578. }
  579. // Reject connections that do not match NetRestrict.
  580. if srv.NetRestrict != nil {
  581. if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok && !srv.NetRestrict.Contains(tcp.IP) {
  582. log.Debug("Rejected conn (not whitelisted in NetRestrict)", "addr", fd.RemoteAddr())
  583. fd.Close()
  584. slots <- struct{}{}
  585. continue
  586. }
  587. }
  588. fd = newMeteredConn(fd, true)
  589. log.Trace("Accepted connection", "addr", fd.RemoteAddr())
  590. // Spawn the handler. It will give the slot back when the connection
  591. // has been established.
  592. go func() {
  593. srv.setupConn(fd, inboundConn, nil)
  594. slots <- struct{}{}
  595. }()
  596. }
  597. }
  598. // setupConn runs the handshakes and attempts to add the connection
  599. // as a peer. It returns when the connection has been added as a peer
  600. // or the handshakes have failed.
  601. func (srv *Server) setupConn(fd net.Conn, flags connFlag, dialDest *discover.Node) {
  602. // Prevent leftover pending conns from entering the handshake.
  603. srv.lock.Lock()
  604. running := srv.running
  605. srv.lock.Unlock()
  606. c := &conn{fd: fd, transport: srv.newTransport(fd), flags: flags, cont: make(chan error)}
  607. if !running {
  608. c.close(errServerStopped)
  609. return
  610. }
  611. // Run the encryption handshake.
  612. var err error
  613. if c.id, err = c.doEncHandshake(srv.PrivateKey, dialDest); err != nil {
  614. log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
  615. c.close(err)
  616. return
  617. }
  618. clog := log.New("id", c.id, "addr", c.fd.RemoteAddr(), "conn", c.flags)
  619. // For dialed connections, check that the remote public key matches.
  620. if dialDest != nil && c.id != dialDest.ID {
  621. c.close(DiscUnexpectedIdentity)
  622. clog.Trace("Dialed identity mismatch", "want", c, dialDest.ID)
  623. return
  624. }
  625. if err := srv.checkpoint(c, srv.posthandshake); err != nil {
  626. clog.Trace("Rejected peer before protocol handshake", "err", err)
  627. c.close(err)
  628. return
  629. }
  630. // Run the protocol handshake
  631. phs, err := c.doProtoHandshake(srv.ourHandshake)
  632. if err != nil {
  633. clog.Trace("Failed proto handshake", "err", err)
  634. c.close(err)
  635. return
  636. }
  637. if phs.ID != c.id {
  638. clog.Trace("Wrong devp2p handshake identity", "err", phs.ID)
  639. c.close(DiscUnexpectedIdentity)
  640. return
  641. }
  642. c.caps, c.name = phs.Caps, phs.Name
  643. if err := srv.checkpoint(c, srv.addpeer); err != nil {
  644. clog.Trace("Rejected peer", "err", err)
  645. c.close(err)
  646. return
  647. }
  648. // If the checks completed successfully, runPeer has now been
  649. // launched by run.
  650. }
  651. func truncateName(s string) string {
  652. if len(s) > 20 {
  653. return s[:20] + "..."
  654. }
  655. return s
  656. }
  657. // checkpoint sends the conn to run, which performs the
  658. // post-handshake checks for the stage (posthandshake, addpeer).
  659. func (srv *Server) checkpoint(c *conn, stage chan<- *conn) error {
  660. select {
  661. case stage <- c:
  662. case <-srv.quit:
  663. return errServerStopped
  664. }
  665. select {
  666. case err := <-c.cont:
  667. return err
  668. case <-srv.quit:
  669. return errServerStopped
  670. }
  671. }
  672. // runPeer runs in its own goroutine for each peer.
  673. // it waits until the Peer logic returns and removes
  674. // the peer.
  675. func (srv *Server) runPeer(p *Peer) {
  676. if srv.newPeerHook != nil {
  677. srv.newPeerHook(p)
  678. }
  679. remoteRequested, err := p.run()
  680. // Note: run waits for existing peers to be sent on srv.delpeer
  681. // before returning, so this send should not select on srv.quit.
  682. srv.delpeer <- peerDrop{p, err, remoteRequested}
  683. }
  684. // NodeInfo represents a short summary of the information known about the host.
  685. type NodeInfo struct {
  686. ID string `json:"id"` // Unique node identifier (also the encryption key)
  687. Name string `json:"name"` // Name of the node, including client type, version, OS, custom data
  688. Enode string `json:"enode"` // Enode URL for adding this peer from remote peers
  689. IP string `json:"ip"` // IP address of the node
  690. Ports struct {
  691. Discovery int `json:"discovery"` // UDP listening port for discovery protocol
  692. Listener int `json:"listener"` // TCP listening port for RLPx
  693. } `json:"ports"`
  694. ListenAddr string `json:"listenAddr"`
  695. Protocols map[string]interface{} `json:"protocols"`
  696. }
  697. // NodeInfo gathers and returns a collection of metadata known about the host.
  698. func (srv *Server) NodeInfo() *NodeInfo {
  699. node := srv.Self()
  700. // Gather and assemble the generic node infos
  701. info := &NodeInfo{
  702. Name: srv.Name,
  703. Enode: node.String(),
  704. ID: node.ID.String(),
  705. IP: node.IP.String(),
  706. ListenAddr: srv.ListenAddr,
  707. Protocols: make(map[string]interface{}),
  708. }
  709. info.Ports.Discovery = int(node.UDP)
  710. info.Ports.Listener = int(node.TCP)
  711. // Gather all the running protocol infos (only once per protocol type)
  712. for _, proto := range srv.Protocols {
  713. if _, ok := info.Protocols[proto.Name]; !ok {
  714. nodeInfo := interface{}("unknown")
  715. if query := proto.NodeInfo; query != nil {
  716. nodeInfo = proto.NodeInfo()
  717. }
  718. info.Protocols[proto.Name] = nodeInfo
  719. }
  720. }
  721. return info
  722. }
  723. // PeersInfo returns an array of metadata objects describing connected peers.
  724. func (srv *Server) PeersInfo() []*PeerInfo {
  725. // Gather all the generic and sub-protocol specific infos
  726. infos := make([]*PeerInfo, 0, srv.PeerCount())
  727. for _, peer := range srv.Peers() {
  728. if peer != nil {
  729. infos = append(infos, peer.Info())
  730. }
  731. }
  732. // Sort the result array alphabetically by node identifier
  733. for i := 0; i < len(infos); i++ {
  734. for j := i + 1; j < len(infos); j++ {
  735. if infos[i].ID > infos[j].ID {
  736. infos[i], infos[j] = infos[j], infos[i]
  737. }
  738. }
  739. }
  740. return infos
  741. }