server.go 24 KB

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