server.go 23 KB

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