server.go 22 KB

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