quorum/p2p/server.go

package p2p

import (
	"bytes"
	"errors"
	"fmt"
	"net"
	"sync"
	"time"

	"github.com/ethereum/go-ethereum/logger"
)

const (
	outboundAddressPoolSize   = 500
	defaultDialTimeout        = 10 * time.Second
	portMappingUpdateInterval = 15 * time.Minute
	portMappingTimeout        = 20 * time.Minute
)

var srvlog = logger.NewLogger("P2P Server")

// Server manages all peer connections.
//
// The fields of Server are used as configuration parameters.
// You should set them before starting the Server. Fields may not be
// modified while the server is running.
type Server struct {
	// This field must be set to a valid client identity.
	Identity ClientIdentity

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

	// Protocols should contain the protocols supported
	// by the server. Matching protocols are launched for
	// each peer.
	Protocols []Protocol

	// If Blacklist is set to a non-nil value, the given Blacklist
	// is used to verify peer connections.
	Blacklist Blacklist

	// 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.
	NAT NAT

	// If Dialer is set to a non-nil value, the given Dialer
	// is used to dial outbound peer connections.
	Dialer *net.Dialer

	// If NoDial is true, the server will not dial any peers.
	NoDial bool

	// Hook for testing. This is useful because we can inhibit
	// the whole protocol stack.
	newPeerFunc peerFunc

	lock      sync.RWMutex
	running   bool
	listener  net.Listener
	laddr     *net.TCPAddr // real listen addr
	peers     []*Peer
	peerSlots chan int
	peerCount int

	quit           chan struct{}
	wg             sync.WaitGroup
	peerConnect    chan *peerAddr
	peerDisconnect chan *Peer
}

// NAT is implemented by NAT traversal methods.
type NAT interface {
	GetExternalAddress() (net.IP, error)
	AddPortMapping(protocol string, extport, intport int, name string, lifetime time.Duration) error
	DeletePortMapping(protocol string, extport, intport int) error

	// Should return name of the method.
	String() string
}

type peerFunc func(srv *Server, c net.Conn, dialAddr *peerAddr) *Peer

// Peers returns all connected peers.
func (srv *Server) Peers() (peers []*Peer) {
	srv.lock.RLock()
	defer srv.lock.RUnlock()
	for _, peer := range srv.peers {
		if peer != nil {
			peers = append(peers, peer)
		}
	}
	return
}

// PeerCount returns the number of connected peers.
func (srv *Server) PeerCount() int {
	srv.lock.RLock()
	defer srv.lock.RUnlock()
	return srv.peerCount
}

// SuggestPeer injects an address into the outbound address pool.
func (srv *Server) SuggestPeer(ip net.IP, port int, nodeID []byte) {
	addr := &peerAddr{ip, uint64(port), nodeID}
	select {
	case srv.peerConnect <- addr:
	default: // don't block
		srvlog.Warnf("peer suggestion %v ignored", addr)
	}
}

// Broadcast sends an RLP-encoded message to all connected peers.
// This method is deprecated and will be removed later.
func (srv *Server) Broadcast(protocol string, code uint64, data ...interface{}) {
	var payload []byte
	if data != nil {
		payload = encodePayload(data...)
	}
	srv.lock.RLock()
	defer srv.lock.RUnlock()
	for _, peer := range srv.peers {
		if peer != nil {
			var msg = Msg{Code: code}
			if data != nil {
				msg.Payload = bytes.NewReader(payload)
				msg.Size = uint32(len(payload))
			}
			peer.writeProtoMsg(protocol, msg)
		}
	}
}

// Start starts running the server.
// Servers can be re-used and started again after stopping.
func (srv *Server) Start() (err error) {
	srv.lock.Lock()
	defer srv.lock.Unlock()
	if srv.running {
		return errors.New("server already running")
	}
	srvlog.Infoln("Starting Server")

	// initialize fields
	if srv.Identity == nil {
		return fmt.Errorf("Server.Identity must be set to a non-nil identity")
	}
	if srv.MaxPeers <= 0 {
		return fmt.Errorf("Server.MaxPeers must be > 0")
	}
	srv.quit = make(chan struct{})
	srv.peers = make([]*Peer, srv.MaxPeers)
	srv.peerSlots = make(chan int, srv.MaxPeers)
	srv.peerConnect = make(chan *peerAddr, outboundAddressPoolSize)
	srv.peerDisconnect = make(chan *Peer)
	if srv.newPeerFunc == nil {
		srv.newPeerFunc = newServerPeer
	}
	if srv.Blacklist == nil {
		srv.Blacklist = NewBlacklist()
	}
	if srv.Dialer == nil {
		srv.Dialer = &net.Dialer{Timeout: defaultDialTimeout}
	}

	if srv.ListenAddr != "" {
		if err := srv.startListening(); err != nil {
			return err
		}
	}
	if !srv.NoDial {
		srv.wg.Add(1)
		go srv.dialLoop()
	}
	if srv.NoDial && srv.ListenAddr == "" {
		srvlog.Warnln("I will be kind-of useless, neither dialing nor listening.")
	}

	// make all slots available
	for i := range srv.peers {
		srv.peerSlots <- i
	}
	// note: discLoop is not part of WaitGroup
	go srv.discLoop()
	srv.running = true
	return nil
}

func (srv *Server) startListening() error {
	listener, err := net.Listen("tcp", srv.ListenAddr)
	if err != nil {
		return err
	}
	srv.ListenAddr = listener.Addr().String()
	srv.laddr = listener.Addr().(*net.TCPAddr)
	srv.listener = listener
	srv.wg.Add(1)
	go srv.listenLoop()
	if !srv.laddr.IP.IsLoopback() && srv.NAT != nil {
		srv.wg.Add(1)
		go srv.natLoop(srv.laddr.Port)
	}
	return nil
}

// 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()
	if !srv.running {
		srv.lock.Unlock()
		return
	}
	srv.running = false
	srv.lock.Unlock()

	srvlog.Infoln("Stopping server")
	if srv.listener != nil {
		// this unblocks listener Accept
		srv.listener.Close()
	}
	close(srv.quit)
	for _, peer := range srv.Peers() {
		peer.Disconnect(DiscQuitting)
	}
	srv.wg.Wait()

	// wait till they actually disconnect
	// this is checked by claiming all peerSlots.
	// slots become available as the peers disconnect.
	for i := 0; i < cap(srv.peerSlots); i++ {
		<-srv.peerSlots
	}
	// terminate discLoop
	close(srv.peerDisconnect)
}

func (srv *Server) discLoop() {
	for peer := range srv.peerDisconnect {
		srv.removePeer(peer)
	}
}

// main loop for adding connections via listening
func (srv *Server) listenLoop() {
	defer srv.wg.Done()

	srvlog.Infoln("Listening on", srv.listener.Addr())
	for {
		select {
		case slot := <-srv.peerSlots:
			srvlog.Debugf("grabbed slot %v for listening", slot)
			conn, err := srv.listener.Accept()
			if err != nil {
				srv.peerSlots <- slot
				return
			}
			srvlog.Debugf("Accepted conn %v (slot %d)\n", conn.RemoteAddr(), slot)
			srv.addPeer(conn, nil, slot)
		case <-srv.quit:
			return
		}
	}
}

func (srv *Server) natLoop(port int) {
	defer srv.wg.Done()
	for {
		srv.updatePortMapping(port)
		select {
		case <-time.After(portMappingUpdateInterval):
			// one more round
		case <-srv.quit:
			srv.removePortMapping(port)
			return
		}
	}
}

func (srv *Server) updatePortMapping(port int) {
	srvlog.Infoln("Attempting to map port", port, "with", srv.NAT)
	err := srv.NAT.AddPortMapping("tcp", port, port, "ethereum p2p", portMappingTimeout)
	if err != nil {
		srvlog.Errorln("Port mapping error:", err)
		return
	}
	extip, err := srv.NAT.GetExternalAddress()
	if err != nil {
		srvlog.Errorln("Error getting external IP:", err)
		return
	}
	srv.lock.Lock()
	extaddr := *(srv.listener.Addr().(*net.TCPAddr))
	extaddr.IP = extip
	srvlog.Infoln("Mapped port, external addr is", &extaddr)
	srv.laddr = &extaddr
	srv.lock.Unlock()
}

func (srv *Server) removePortMapping(port int) {
	srvlog.Infoln("Removing port mapping for", port, "with", srv.NAT)
	srv.NAT.DeletePortMapping("tcp", port, port)
}

func (srv *Server) dialLoop() {
	defer srv.wg.Done()
	var (
		suggest chan *peerAddr
		slot    *int
		slots   = srv.peerSlots
	)
	for {
		select {
		case i := <-slots:
			// we need a peer in slot i, slot reserved
			slot = &i
			// now we can watch for candidate peers in the next loop
			suggest = srv.peerConnect
			// do not consume more until candidate peer is found
			slots = nil

		case desc := <-suggest:
			// candidate peer found, will dial out asyncronously
			// if connection fails slot will be released
			srvlog.DebugDetailf("dial %v (%v)", desc, *slot)
			go srv.dialPeer(desc, *slot)
			// we can watch if more peers needed in the next loop
			slots = srv.peerSlots
			// until then we dont care about candidate peers
			suggest = nil

		case <-srv.quit:
			// give back the currently reserved slot
			if slot != nil {
				srv.peerSlots <- *slot
			}
			return
		}
	}
}

// connect to peer via dial out
func (srv *Server) dialPeer(desc *peerAddr, slot int) {
	srvlog.Debugf("Dialing %v (slot %d)\n", desc, slot)
	conn, err := srv.Dialer.Dial(desc.Network(), desc.String())
	if err != nil {
		srvlog.DebugDetailf("dial error: %v", err)
		srv.peerSlots <- slot
		return
	}
	go srv.addPeer(conn, desc, slot)
}

// creates the new peer object and inserts it into its slot
func (srv *Server) addPeer(conn net.Conn, desc *peerAddr, slot int) *Peer {
	srv.lock.Lock()
	defer srv.lock.Unlock()
	if !srv.running {
		conn.Close()
		srv.peerSlots <- slot // release slot
		return nil
	}
	peer := srv.newPeerFunc(srv, conn, desc)
	peer.slot = slot
	srv.peers[slot] = peer
	srv.peerCount++
	go func() { peer.loop(); srv.peerDisconnect <- peer }()
	return peer
}

// removes peer: sending disconnect msg, stop peer, remove rom list/table, release slot
func (srv *Server) removePeer(peer *Peer) {
	srv.lock.Lock()
	defer srv.lock.Unlock()
	srvlog.Debugf("Removing %v (slot %v)\n", peer, peer.slot)
	if srv.peers[peer.slot] != peer {
		srvlog.Warnln("Invalid peer to remove:", peer)
		return
	}
	// remove from list and index
	srv.peerCount--
	srv.peers[peer.slot] = nil
	// release slot to signal need for a new peer, last!
	srv.peerSlots <- peer.slot
}

func (srv *Server) verifyPeer(addr *peerAddr) error {
	if srv.Blacklist.Exists(addr.Pubkey) {
		return errors.New("blacklisted")
	}
	if bytes.Equal(srv.Identity.Pubkey()[1:], addr.Pubkey) {
		return newPeerError(errPubkeyForbidden, "not allowed to connect to srv")
	}
	srv.lock.RLock()
	defer srv.lock.RUnlock()
	for _, peer := range srv.peers {
		if peer != nil {
			id := peer.Identity()
			if id != nil && bytes.Equal(id.Pubkey(), addr.Pubkey) {
				return errors.New("already connected")
			}
		}
	}
	return nil
}

// TODO replace with "Set"
type Blacklist interface {
	Get([]byte) (bool, error)
	Put([]byte) error
	Delete([]byte) error
	Exists(pubkey []byte) (ok bool)
}

type BlacklistMap struct {
	blacklist map[string]bool
	lock      sync.RWMutex
}

func NewBlacklist() *BlacklistMap {
	return &BlacklistMap{
		blacklist: make(map[string]bool),
	}
}

func (self *BlacklistMap) Get(pubkey []byte) (bool, error) {
	self.lock.RLock()
	defer self.lock.RUnlock()
	v, ok := self.blacklist[string(pubkey)]
	var err error
	if !ok {
		err = fmt.Errorf("not found")
	}
	return v, err
}

func (self *BlacklistMap) Exists(pubkey []byte) (ok bool) {
	self.lock.RLock()
	defer self.lock.RUnlock()
	_, ok = self.blacklist[string(pubkey)]
	return
}

func (self *BlacklistMap) Put(pubkey []byte) error {
	self.lock.RLock()
	defer self.lock.RUnlock()
	self.blacklist[string(pubkey)] = true
	return nil
}

func (self *BlacklistMap) Delete(pubkey []byte) error {
	self.lock.RLock()
	defer self.lock.RUnlock()
	delete(self.blacklist, string(pubkey))
	return nil
}