server.go 24 KB

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