server.go 23.9 KB
Newer Older
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 6 7 8
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
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 13 14
// GNU Lesser General Public License for more details.
//
// 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

17
// Package p2p implements the Ethereum p2p network protocols.
zelig's avatar
zelig committed
18 19 20
package p2p

import (
21
	"crypto/ecdsa"
22
	"errors"
zelig's avatar
zelig committed
23 24 25 26 27
	"fmt"
	"net"
	"sync"
	"time"

28 29
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/common/mclock"
30
	"github.com/ethereum/go-ethereum/log"
31
	"github.com/ethereum/go-ethereum/p2p/discover"
32
	"github.com/ethereum/go-ethereum/p2p/discv5"
33
	"github.com/ethereum/go-ethereum/p2p/nat"
34
	"github.com/ethereum/go-ethereum/p2p/netutil"
zelig's avatar
zelig committed
35 36 37
)

const (
38
	defaultDialTimeout      = 15 * time.Second
39 40
	refreshPeersInterval    = 30 * time.Second
	staticPeerCheckInterval = 15 * time.Second
41

42
	// Maximum number of concurrently handshaking inbound connections.
43
	maxAcceptConns = 50
44

45
	// Maximum number of concurrently dialing outbound connections.
46
	maxActiveDialTasks = 16
47

48 49 50 51 52
	// Maximum time allowed for reading a complete message.
	// This is effectively the amount of time a connection can be idle.
	frameReadTimeout = 30 * time.Second

	// Maximum amount of time allowed for writing a complete message.
53
	frameWriteTimeout = 20 * time.Second
zelig's avatar
zelig committed
54 55
)

56 57
var errServerStopped = errors.New("server stopped")

58 59
// Config holds Server options.
type Config struct {
60
	// This field must be set to a valid secp256k1 private key.
61
	PrivateKey *ecdsa.PrivateKey `toml:"-"`
62 63 64 65 66

	// MaxPeers is the maximum number of peers that can be
	// connected. It must be greater than zero.
	MaxPeers int

67 68 69
	// MaxPendingPeers is the maximum number of peers that can be pending in the
	// handshake phase, counted separately for inbound and outbound connections.
	// Zero defaults to preset values.
70
	MaxPendingPeers int `toml:",omitempty"`
71

72 73 74
	// NoDiscovery can be used to disable the peer discovery mechanism.
	// Disabling is useful for protocol debugging (manual topology).
	NoDiscovery bool
75

76 77
	// DiscoveryV5 specifies whether the the new topic-discovery based V5 discovery
	// protocol should be started or not.
78
	DiscoveryV5 bool `toml:",omitempty"`
79

80
	// Listener address for the V5 discovery protocol UDP traffic.
81
	DiscoveryV5Addr string `toml:",omitempty"`
82

83
	// Name sets the node name of this server.
obscuren's avatar
obscuren committed
84
	// Use common.MakeName to create a name that follows existing conventions.
85
	Name string `toml:"-"`
86

87
	// BootstrapNodes are used to establish connectivity
88
	// with the rest of the network.
89
	BootstrapNodes []*discover.Node
90

91 92 93
	// BootstrapNodesV5 are used to establish connectivity
	// with the rest of the network using the V5 discovery
	// protocol.
94
	BootstrapNodesV5 []*discv5.Node `toml:",omitempty"`
95

96 97 98
	// Static nodes are used as pre-configured connections which are always
	// maintained and re-connected on disconnects.
	StaticNodes []*discover.Node
99

100 101 102 103
	// Trusted nodes are used as pre-configured connections which are always
	// allowed to connect, even above the peer limit.
	TrustedNodes []*discover.Node

104 105 106
	// Connectivity can be restricted to certain IP networks.
	// If this option is set to a non-nil value, only hosts which match one of the
	// IP networks contained in the list are considered.
107
	NetRestrict *netutil.Netlist `toml:",omitempty"`
108

109 110
	// NodeDatabase is the path to the database containing the previously seen
	// live nodes in the network.
111
	NodeDatabase string `toml:",omitempty"`
112

113 114 115
	// Protocols should contain the protocols supported
	// by the server. Matching protocols are launched for
	// each peer.
116
	Protocols []Protocol `toml:"-"`
117 118 119 120 121 122 123 124 125 126 127 128

	// If ListenAddr is set to a non-nil address, the server
	// will listen for incoming connections.
	//
	// If the port is zero, the operating system will pick a port. The
	// ListenAddr field will be updated with the actual address when
	// the server is started.
	ListenAddr string

	// If set to a non-nil value, the given NAT port mapper
	// is used to make the listening port available to the
	// Internet.
129
	NAT nat.Interface `toml:",omitempty"`
130 131 132

	// If Dialer is set to a non-nil value, the given Dialer
	// is used to dial outbound peer connections.
133
	Dialer *net.Dialer `toml:"-"`
134 135

	// If NoDial is true, the server will not dial any peers.
136
	NoDial bool `toml:",omitempty"`
137 138 139 140 141 142
}

// Server manages all peer connections.
type Server struct {
	// Config fields may not be modified while the server is running.
	Config
143

144
	// Hooks for testing. These are useful because we can inhibit
145
	// the whole protocol stack.
146 147 148 149 150
	newTransport func(net.Conn) transport
	newPeerHook  func(*Peer)

	lock    sync.Mutex // protects running
	running bool
zelig's avatar
zelig committed
151

152 153
	ntab         discoverTable
	listener     net.Listener
154
	ourHandshake *protoHandshake
155
	lastLookup   time.Time
156
	DiscV5       *discv5.Network
157

158 159 160 161 162 163
	// These are for Peers, PeerCount (and nothing else).
	peerOp     chan peerOpFunc
	peerOpDone chan struct{}

	quit          chan struct{}
	addstatic     chan *discover.Node
164
	removestatic  chan *discover.Node
165 166
	posthandshake chan *conn
	addpeer       chan *conn
167
	delpeer       chan peerDrop
168 169 170 171 172
	loopWG        sync.WaitGroup // loop, listenLoop
}

type peerOpFunc func(map[discover.NodeID]*Peer)

173 174 175 176 177 178
type peerDrop struct {
	*Peer
	err       error
	requested bool // true if signaled by the peer
}

179
type connFlag int
180

181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198
const (
	dynDialedConn connFlag = 1 << iota
	staticDialedConn
	inboundConn
	trustedConn
)

// conn wraps a network connection with information gathered
// during the two handshakes.
type conn struct {
	fd net.Conn
	transport
	flags connFlag
	cont  chan error      // The run loop uses cont to signal errors to setupConn.
	id    discover.NodeID // valid after the encryption handshake
	caps  []Cap           // valid after the protocol handshake
	name  string          // valid after the protocol handshake
}
199

200 201 202 203 204 205 206 207 208 209 210 211
type transport interface {
	// The two handshakes.
	doEncHandshake(prv *ecdsa.PrivateKey, dialDest *discover.Node) (discover.NodeID, error)
	doProtoHandshake(our *protoHandshake) (*protoHandshake, error)
	// The MsgReadWriter can only be used after the encryption
	// handshake has completed. The code uses conn.id to track this
	// by setting it to a non-nil value after the encryption handshake.
	MsgReadWriter
	// transports must provide Close because we use MsgPipe in some of
	// the tests. Closing the actual network connection doesn't do
	// anything in those tests because NsgPipe doesn't use it.
	close(err error)
zelig's avatar
zelig committed
212 213
}

214
func (c *conn) String() string {
215
	s := c.flags.String()
216
	if (c.id != discover.NodeID{}) {
217
		s += " " + c.id.String()
218 219 220 221 222 223 224 225
	}
	s += " " + c.fd.RemoteAddr().String()
	return s
}

func (f connFlag) String() string {
	s := ""
	if f&trustedConn != 0 {
226
		s += "-trusted"
227 228
	}
	if f&dynDialedConn != 0 {
229
		s += "-dyndial"
230 231
	}
	if f&staticDialedConn != 0 {
232
		s += "-staticdial"
233 234
	}
	if f&inboundConn != 0 {
235
		s += "-inbound"
236 237 238 239 240 241 242 243 244 245
	}
	if s != "" {
		s = s[1:]
	}
	return s
}

func (c *conn) is(f connFlag) bool {
	return c.flags&f != 0
}
zelig's avatar
zelig committed
246

247
// Peers returns all connected peers.
248 249 250 251 252 253 254 255 256
func (srv *Server) Peers() []*Peer {
	var ps []*Peer
	select {
	// Note: We'd love to put this function into a variable but
	// that seems to cause a weird compiler error in some
	// environments.
	case srv.peerOp <- func(peers map[discover.NodeID]*Peer) {
		for _, p := range peers {
			ps = append(ps, p)
zelig's avatar
zelig committed
257
		}
258 259 260
	}:
		<-srv.peerOpDone
	case <-srv.quit:
zelig's avatar
zelig committed
261
	}
262
	return ps
zelig's avatar
zelig committed
263 264
}

265 266
// PeerCount returns the number of connected peers.
func (srv *Server) PeerCount() int {
267 268 269 270 271 272 273
	var count int
	select {
	case srv.peerOp <- func(ps map[discover.NodeID]*Peer) { count = len(ps) }:
		<-srv.peerOpDone
	case <-srv.quit:
	}
	return count
zelig's avatar
zelig committed
274 275
}

276 277 278 279
// AddPeer connects to the given node and maintains the connection until the
// server is shut down. If the connection fails for any reason, the server will
// attempt to reconnect the peer.
func (srv *Server) AddPeer(node *discover.Node) {
280 281 282 283 284 285
	select {
	case srv.addstatic <- node:
	case <-srv.quit:
	}
}

286 287 288 289 290 291 292 293
// RemovePeer disconnects from the given node
func (srv *Server) RemovePeer(node *discover.Node) {
	select {
	case srv.removestatic <- node:
	case <-srv.quit:
	}
}

294 295
// Self returns the local node's endpoint information.
func (srv *Server) Self() *discover.Node {
296 297
	srv.lock.Lock()
	defer srv.lock.Unlock()
298

299 300 301
	if !srv.running {
		return &discover.Node{IP: net.ParseIP("0.0.0.0")}
	}
302 303 304 305 306 307 308 309 310
	return srv.makeSelf(srv.listener, srv.ntab)
}

func (srv *Server) makeSelf(listener net.Listener, ntab discoverTable) *discover.Node {
	// If the server's not running, return an empty node.
	// If the node is running but discovery is off, manually assemble the node infos.
	if ntab == nil {
		// Inbound connections disabled, use zero address.
		if listener == nil {
311 312 313
			return &discover.Node{IP: net.ParseIP("0.0.0.0"), ID: discover.PubkeyID(&srv.PrivateKey.PublicKey)}
		}
		// Otherwise inject the listener address too
314
		addr := listener.Addr().(*net.TCPAddr)
315 316 317 318 319 320
		return &discover.Node{
			ID:  discover.PubkeyID(&srv.PrivateKey.PublicKey),
			IP:  addr.IP,
			TCP: uint16(addr.Port),
		}
	}
321 322
	// Otherwise return the discovery node.
	return ntab.Self()
323
}
324

325 326 327 328 329 330 331 332 333 334 335 336 337 338 339
// Stop terminates the server and all active peer connections.
// It blocks until all active connections have been closed.
func (srv *Server) Stop() {
	srv.lock.Lock()
	defer srv.lock.Unlock()
	if !srv.running {
		return
	}
	srv.running = false
	if srv.listener != nil {
		// this unblocks listener Accept
		srv.listener.Close()
	}
	close(srv.quit)
	srv.loopWG.Wait()
zelig's avatar
zelig committed
340 341
}

342
// Start starts running the server.
343
// Servers can not be re-used after stopping.
344 345 346 347 348 349
func (srv *Server) Start() (err error) {
	srv.lock.Lock()
	defer srv.lock.Unlock()
	if srv.running {
		return errors.New("server already running")
	}
350
	srv.running = true
351
	log.Info("Starting P2P networking")
352

353
	// static fields
354 355
	if srv.PrivateKey == nil {
		return fmt.Errorf("Server.PrivateKey must be set to a non-nil key")
zelig's avatar
zelig committed
356
	}
357 358
	if srv.newTransport == nil {
		srv.newTransport = newRLPX
359
	}
360 361
	if srv.Dialer == nil {
		srv.Dialer = &net.Dialer{Timeout: defaultDialTimeout}
362
	}
363 364
	srv.quit = make(chan struct{})
	srv.addpeer = make(chan *conn)
365
	srv.delpeer = make(chan peerDrop)
366 367
	srv.posthandshake = make(chan *conn)
	srv.addstatic = make(chan *discover.Node)
368
	srv.removestatic = make(chan *discover.Node)
369 370
	srv.peerOp = make(chan peerOpFunc)
	srv.peerOpDone = make(chan struct{})
371

372
	// node table
373
	if !srv.NoDiscovery {
374
		ntab, err := discover.ListenUDP(srv.PrivateKey, srv.ListenAddr, srv.NAT, srv.NodeDatabase, srv.NetRestrict)
375 376 377
		if err != nil {
			return err
		}
378 379 380
		if err := ntab.SetFallbackNodes(srv.BootstrapNodes); err != nil {
			return err
		}
381
		srv.ntab = ntab
382
	}
383

384
	if srv.DiscoveryV5 {
385
		ntab, err := discv5.ListenUDP(srv.PrivateKey, srv.DiscoveryV5Addr, srv.NAT, "", srv.NetRestrict) //srv.NodeDatabase)
386 387 388
		if err != nil {
			return err
		}
389
		if err := ntab.SetFallbackNodes(srv.BootstrapNodesV5); err != nil {
390 391 392 393 394
			return err
		}
		srv.DiscV5 = ntab
	}

395
	dynPeers := (srv.MaxPeers + 1) / 2
396
	if srv.NoDiscovery {
397 398
		dynPeers = 0
	}
399
	dialer := newDialState(srv.StaticNodes, srv.BootstrapNodes, srv.ntab, dynPeers, srv.NetRestrict)
400

401
	// handshake
402
	srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: discover.PubkeyID(&srv.PrivateKey.PublicKey)}
403 404 405
	for _, p := range srv.Protocols {
		srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap())
	}
406 407 408 409 410 411
	// listen/dial
	if srv.ListenAddr != "" {
		if err := srv.startListening(); err != nil {
			return err
		}
	}
412
	if srv.NoDial && srv.ListenAddr == "" {
413
		log.Warn("P2P server will be useless, neither dialing nor listening")
414 415
	}

416 417
	srv.loopWG.Add(1)
	go srv.run(dialer)
418 419
	srv.running = true
	return nil
zelig's avatar
zelig committed
420 421
}

422
func (srv *Server) startListening() error {
423
	// Launch the TCP listener.
424 425 426 427
	listener, err := net.Listen("tcp", srv.ListenAddr)
	if err != nil {
		return err
	}
428 429
	laddr := listener.Addr().(*net.TCPAddr)
	srv.ListenAddr = laddr.String()
430
	srv.listener = listener
431
	srv.loopWG.Add(1)
432
	go srv.listenLoop()
433
	// Map the TCP listening port if NAT is configured.
434
	if !laddr.IP.IsLoopback() && srv.NAT != nil {
435
		srv.loopWG.Add(1)
436 437 438 439
		go func() {
			nat.Map(srv.NAT, srv.quit, "tcp", laddr.Port, laddr.Port, "ethereum p2p")
			srv.loopWG.Done()
		}()
440 441 442 443
	}
	return nil
}

444 445 446 447
type dialer interface {
	newTasks(running int, peers map[discover.NodeID]*Peer, now time.Time) []task
	taskDone(task, time.Time)
	addStatic(*discover.Node)
448
	removeStatic(*discover.Node)
449 450 451 452 453
}

func (srv *Server) run(dialstate dialer) {
	defer srv.loopWG.Done()
	var (
454 455
		peers        = make(map[discover.NodeID]*Peer)
		trusted      = make(map[discover.NodeID]bool, len(srv.TrustedNodes))
456
		taskdone     = make(chan task, maxActiveDialTasks)
457 458
		runningTasks []task
		queuedTasks  []task // tasks that can't run yet
459 460 461 462 463 464 465 466
	)
	// Put trusted nodes into a map to speed up checks.
	// Trusted peers are loaded on startup and cannot be
	// modified while the server is running.
	for _, n := range srv.TrustedNodes {
		trusted[n.ID] = true
	}

467
	// removes t from runningTasks
468
	delTask := func(t task) {
469 470 471
		for i := range runningTasks {
			if runningTasks[i] == t {
				runningTasks = append(runningTasks[:i], runningTasks[i+1:]...)
472 473 474
				break
			}
		}
zelig's avatar
zelig committed
475
	}
476 477 478 479 480
	// starts until max number of active tasks is satisfied
	startTasks := func(ts []task) (rest []task) {
		i := 0
		for ; len(runningTasks) < maxActiveDialTasks && i < len(ts); i++ {
			t := ts[i]
481
			log.Trace("New dial task", "task", t)
482 483
			go func() { t.Do(srv); taskdone <- t }()
			runningTasks = append(runningTasks, t)
484
		}
485 486 487 488 489 490 491 492 493
		return ts[i:]
	}
	scheduleTasks := func() {
		// Start from queue first.
		queuedTasks = append(queuedTasks[:0], startTasks(queuedTasks)...)
		// Query dialer for new tasks and start as many as possible now.
		if len(runningTasks) < maxActiveDialTasks {
			nt := dialstate.newTasks(len(runningTasks)+len(queuedTasks), peers, time.Now())
			queuedTasks = append(queuedTasks, startTasks(nt)...)
494 495 496 497 498
		}
	}

running:
	for {
499
		scheduleTasks()
500

501 502 503 504 505 506 507 508
		select {
		case <-srv.quit:
			// The server was stopped. Run the cleanup logic.
			break running
		case n := <-srv.addstatic:
			// This channel is used by AddPeer to add to the
			// ephemeral static peer list. Add it to the dialer,
			// it will keep the node connected.
509
			log.Debug("Adding static node", "node", n)
510
			dialstate.addStatic(n)
511 512 513 514
		case n := <-srv.removestatic:
			// This channel is used by RemovePeer to send a
			// disconnect request to a peer and begin the
			// stop keeping the node connected
515
			log.Debug("Removing static node", "node", n)
516 517 518 519
			dialstate.removeStatic(n)
			if p, ok := peers[n.ID]; ok {
				p.Disconnect(DiscRequested)
			}
520 521 522 523 524 525 526 527
		case op := <-srv.peerOp:
			// This channel is used by Peers and PeerCount.
			op(peers)
			srv.peerOpDone <- struct{}{}
		case t := <-taskdone:
			// A task got done. Tell dialstate about it so it
			// can update its state and remove it from the active
			// tasks list.
528
			log.Trace("Dial task done", "task", t)
529
			dialstate.taskDone(t, time.Now())
530 531 532 533 534 535 536 537 538 539 540 541 542 543
			delTask(t)
		case c := <-srv.posthandshake:
			// A connection has passed the encryption handshake so
			// the remote identity is known (but hasn't been verified yet).
			if trusted[c.id] {
				// Ensure that the trusted flag is set before checking against MaxPeers.
				c.flags |= trustedConn
			}
			// TODO: track in-progress inbound node IDs (pre-Peer) to avoid dialing them.
			c.cont <- srv.encHandshakeChecks(peers, c)
		case c := <-srv.addpeer:
			// At this point the connection is past the protocol handshake.
			// Its capabilities are known and the remote identity is verified.
			err := srv.protoHandshakeChecks(peers, c)
544
			if err == nil {
545 546
				// The handshakes are done and it passed all checks.
				p := newPeer(c, srv.Protocols)
547 548
				name := truncateName(c.name)
				log.Debug("Adding p2p peer", "id", c.id, "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1)
549 550 551 552 553 554 555
				peers[c.id] = p
				go srv.runPeer(p)
			}
			// The dialer logic relies on the assumption that
			// dial tasks complete after the peer has been added or
			// discarded. Unblock the task last.
			c.cont <- err
556
		case pd := <-srv.delpeer:
557
			// A peer disconnected.
558 559 560
			d := common.PrettyDuration(mclock.Now() - pd.created)
			pd.log.Debug("Removing p2p peer", "duration", d, "peers", len(peers)-1, "req", pd.requested, "err", pd.err)
			delete(peers, pd.ID())
561 562 563
		}
	}

564 565
	log.Trace("P2P networking is spinning down")

566
	// Terminate discovery. If there is a running lookup it will terminate soon.
567 568 569
	if srv.ntab != nil {
		srv.ntab.Close()
	}
570 571 572
	if srv.DiscV5 != nil {
		srv.DiscV5.Close()
	}
573 574 575 576 577 578 579 580 581
	// Disconnect all peers.
	for _, p := range peers {
		p.Disconnect(DiscQuitting)
	}
	// Wait for peers to shut down. Pending connections and tasks are
	// not handled here and will terminate soon-ish because srv.quit
	// is closed.
	for len(peers) > 0 {
		p := <-srv.delpeer
582
		p.log.Trace("<-delpeer (spindown)", "remainingTasks", len(runningTasks))
583
		delete(peers, p.ID())
584
	}
585
}
586

587 588 589 590
func (srv *Server) protoHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
	// Drop connections with no matching protocols.
	if len(srv.Protocols) > 0 && countMatchingProtocols(srv.Protocols, c.caps) == 0 {
		return DiscUselessPeer
591
	}
592 593 594
	// Repeat the encryption handshake checks because the
	// peer set might have changed between the handshakes.
	return srv.encHandshakeChecks(peers, c)
595
}
zelig's avatar
zelig committed
596

597 598 599 600 601 602
func (srv *Server) encHandshakeChecks(peers map[discover.NodeID]*Peer, c *conn) error {
	switch {
	case !c.is(trustedConn|staticDialedConn) && len(peers) >= srv.MaxPeers:
		return DiscTooManyPeers
	case peers[c.id] != nil:
		return DiscAlreadyConnected
603
	case c.id == srv.Self().ID:
604 605 606
		return DiscSelf
	default:
		return nil
607
	}
608 609
}

610 611 612 613
type tempError interface {
	Temporary() bool
}

614 615
// listenLoop runs in its own goroutine and accepts
// inbound connections.
616
func (srv *Server) listenLoop() {
617
	defer srv.loopWG.Done()
618
	log.Info("RLPx listener up", "self", srv.makeSelf(srv.listener, srv.ntab))
619 620 621 622

	// This channel acts as a semaphore limiting
	// active inbound connections that are lingering pre-handshake.
	// If all slots are taken, no further connections are accepted.
623 624 625 626 627 628
	tokens := maxAcceptConns
	if srv.MaxPendingPeers > 0 {
		tokens = srv.MaxPendingPeers
	}
	slots := make(chan struct{}, tokens)
	for i := 0; i < tokens; i++ {
629 630 631
		slots <- struct{}{}
	}

zelig's avatar
zelig committed
632
	for {
633
		// Wait for a handshake slot before accepting.
634
		<-slots
635 636 637 638 639 640 641 642

		var (
			fd  net.Conn
			err error
		)
		for {
			fd, err = srv.listener.Accept()
			if tempErr, ok := err.(tempError); ok && tempErr.Temporary() {
643
				log.Debug("Temporary read error", "err", err)
644 645
				continue
			} else if err != nil {
646
				log.Debug("Read error", "err", err)
647 648 649
				return
			}
			break
zelig's avatar
zelig committed
650
		}
651 652 653 654

		// Reject connections that do not match NetRestrict.
		if srv.NetRestrict != nil {
			if tcp, ok := fd.RemoteAddr().(*net.TCPAddr); ok && !srv.NetRestrict.Contains(tcp.IP) {
655
				log.Debug("Rejected conn (not whitelisted in NetRestrict)", "addr", fd.RemoteAddr())
656 657 658 659 660 661
				fd.Close()
				slots <- struct{}{}
				continue
			}
		}

662
		fd = newMeteredConn(fd, true)
663
		log.Trace("Accepted connection", "addr", fd.RemoteAddr())
664

665 666
		// Spawn the handler. It will give the slot back when the connection
		// has been established.
667
		go func() {
668
			srv.setupConn(fd, inboundConn, nil)
669 670
			slots <- struct{}{}
		}()
zelig's avatar
zelig committed
671 672 673
	}
}

674 675 676 677 678 679 680 681 682 683 684 685
// setupConn runs the handshakes and attempts to add the connection
// as a peer. It returns when the connection has been added as a peer
// or the handshakes have failed.
func (srv *Server) setupConn(fd net.Conn, flags connFlag, dialDest *discover.Node) {
	// Prevent leftover pending conns from entering the handshake.
	srv.lock.Lock()
	running := srv.running
	srv.lock.Unlock()
	c := &conn{fd: fd, transport: srv.newTransport(fd), flags: flags, cont: make(chan error)}
	if !running {
		c.close(errServerStopped)
		return
686
	}
687 688 689
	// Run the encryption handshake.
	var err error
	if c.id, err = c.doEncHandshake(srv.PrivateKey, dialDest); err != nil {
690
		log.Trace("Failed RLPx handshake", "addr", c.fd.RemoteAddr(), "conn", c.flags, "err", err)
691 692
		c.close(err)
		return
693
	}
694
	clog := log.New("id", c.id, "addr", c.fd.RemoteAddr(), "conn", c.flags)
695 696 697
	// For dialed connections, check that the remote public key matches.
	if dialDest != nil && c.id != dialDest.ID {
		c.close(DiscUnexpectedIdentity)
698
		clog.Trace("Dialed identity mismatch", "want", c, dialDest.ID)
699
		return
zelig's avatar
zelig committed
700
	}
701
	if err := srv.checkpoint(c, srv.posthandshake); err != nil {
702
		clog.Trace("Rejected peer before protocol handshake", "err", err)
703
		c.close(err)
Felix Lange's avatar
Felix Lange committed
704
		return
zelig's avatar
zelig committed
705
	}
706 707
	// Run the protocol handshake
	phs, err := c.doProtoHandshake(srv.ourHandshake)
708
	if err != nil {
709
		clog.Trace("Failed proto handshake", "err", err)
710
		c.close(err)
711 712
		return
	}
713
	if phs.ID != c.id {
714
		clog.Trace("Wrong devp2p handshake identity", "err", phs.ID)
715 716
		c.close(DiscUnexpectedIdentity)
		return
717
	}
718 719
	c.caps, c.name = phs.Caps, phs.Name
	if err := srv.checkpoint(c, srv.addpeer); err != nil {
720
		clog.Trace("Rejected peer", "err", err)
721
		c.close(err)
722
		return
zelig's avatar
zelig committed
723
	}
724 725
	// If the checks completed successfully, runPeer has now been
	// launched by run.
726
}
727

728 729 730 731 732 733 734
func truncateName(s string) string {
	if len(s) > 20 {
		return s[:20] + "..."
	}
	return s
}

735 736 737 738 739 740 741
// checkpoint sends the conn to run, which performs the
// post-handshake checks for the stage (posthandshake, addpeer).
func (srv *Server) checkpoint(c *conn, stage chan<- *conn) error {
	select {
	case stage <- c:
	case <-srv.quit:
		return errServerStopped
742
	}
743 744 745 746 747
	select {
	case err := <-c.cont:
		return err
	case <-srv.quit:
		return errServerStopped
748 749 750
	}
}

751 752 753
// runPeer runs in its own goroutine for each peer.
// it waits until the Peer logic returns and removes
// the peer.
754
func (srv *Server) runPeer(p *Peer) {
755 756 757
	if srv.newPeerHook != nil {
		srv.newPeerHook(p)
	}
758
	remoteRequested, err := p.run()
759 760
	// Note: run waits for existing peers to be sent on srv.delpeer
	// before returning, so this send should not select on srv.quit.
761
	srv.delpeer <- peerDrop{p, err, remoteRequested}
zelig's avatar
zelig committed
762
}
763 764 765 766 767 768 769 770 771 772 773 774 775 776 777

// NodeInfo represents a short summary of the information known about the host.
type NodeInfo struct {
	ID    string `json:"id"`    // Unique node identifier (also the encryption key)
	Name  string `json:"name"`  // Name of the node, including client type, version, OS, custom data
	Enode string `json:"enode"` // Enode URL for adding this peer from remote peers
	IP    string `json:"ip"`    // IP address of the node
	Ports struct {
		Discovery int `json:"discovery"` // UDP listening port for discovery protocol
		Listener  int `json:"listener"`  // TCP listening port for RLPx
	} `json:"ports"`
	ListenAddr string                 `json:"listenAddr"`
	Protocols  map[string]interface{} `json:"protocols"`
}

778
// NodeInfo gathers and returns a collection of metadata known about the host.
779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825
func (srv *Server) NodeInfo() *NodeInfo {
	node := srv.Self()

	// Gather and assemble the generic node infos
	info := &NodeInfo{
		Name:       srv.Name,
		Enode:      node.String(),
		ID:         node.ID.String(),
		IP:         node.IP.String(),
		ListenAddr: srv.ListenAddr,
		Protocols:  make(map[string]interface{}),
	}
	info.Ports.Discovery = int(node.UDP)
	info.Ports.Listener = int(node.TCP)

	// Gather all the running protocol infos (only once per protocol type)
	for _, proto := range srv.Protocols {
		if _, ok := info.Protocols[proto.Name]; !ok {
			nodeInfo := interface{}("unknown")
			if query := proto.NodeInfo; query != nil {
				nodeInfo = proto.NodeInfo()
			}
			info.Protocols[proto.Name] = nodeInfo
		}
	}
	return info
}

// PeersInfo returns an array of metadata objects describing connected peers.
func (srv *Server) PeersInfo() []*PeerInfo {
	// Gather all the generic and sub-protocol specific infos
	infos := make([]*PeerInfo, 0, srv.PeerCount())
	for _, peer := range srv.Peers() {
		if peer != nil {
			infos = append(infos, peer.Info())
		}
	}
	// Sort the result array alphabetically by node identifier
	for i := 0; i < len(infos); i++ {
		for j := i + 1; j < len(infos); j++ {
			if infos[i].ID > infos[j].ID {
				infos[i], infos[j] = infos[j], infos[i]
			}
		}
	}
	return infos
}