server.go

package wtserver

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

	"github.com/btcsuite/btcd/chaincfg/chainhash"
	"github.com/btcsuite/btcd/connmgr"
	"github.com/btcsuite/btcutil"
	"github.com/lightningnetwork/lnd/keychain"
	"github.com/lightningnetwork/lnd/lnwire"
	"github.com/lightningnetwork/lnd/watchtower/wtdb"
	"github.com/lightningnetwork/lnd/watchtower/wtwire"
)

var (
	// ErrPeerAlreadyConnected signals that a peer with the same session id
	// is already active within the server.
	ErrPeerAlreadyConnected = errors.New("peer already connected")

	// ErrServerExiting signals that a request could not be processed
	// because the server has been requested to shut down.
	ErrServerExiting = errors.New("server shutting down")
)

// Config abstracts the primary components and dependencies of the server.
type Config struct {
	// DB provides persistent access to the server's sessions and for
	// storing state updates.
	DB DB

	// NodeKeyECDH is the the ECDH capable wrapper of the key to be used in
	// accepting new brontide connections.
	NodeKeyECDH keychain.SingleKeyECDH

	// Listeners specifies which address to which clients may connect.
	Listeners []net.Listener

	// ReadTimeout specifies how long a client may go without sending a
	// message.
	ReadTimeout time.Duration

	// WriteTimeout specifies how long a client may go without reading a
	// message from the other end, if the connection has stopped buffering
	// the server's replies.
	WriteTimeout time.Duration

	// NewAddress is used to generate reward addresses, where a cut of
	// successfully sent funds can be received.
	NewAddress func() (btcutil.Address, error)

	// ChainHash identifies the network that the server is watching.
	ChainHash chainhash.Hash

	// NoAckCreateSession causes the server to not reply to create session
	// requests, this should only be used for testing.
	NoAckCreateSession bool

	// NoAckUpdates causes the server to not acknowledge state updates, this
	// should only be used for testing.
	NoAckUpdates bool

	// DisableReward causes the server to reject any session creation
	// attempts that request rewards.
	DisableReward bool
}

// Server houses the state required to handle watchtower peers. It's primary job
// is to accept incoming connections, and dispatch processing of the client
// message streams.
type Server struct {
	started sync.Once
	stopped sync.Once

	cfg *Config

	connMgr *connmgr.ConnManager

	clientMtx sync.RWMutex
	clients   map[wtdb.SessionID]Peer

	newPeers chan Peer

	localInit *wtwire.Init

	wg   sync.WaitGroup
	quit chan struct{}
}

// New creates a new server to handle watchtower clients. The server will accept
// clients connecting to the listener addresses, and allows them to open
// sessions and send state updates.
func New(cfg *Config) (*Server, error) {
	localInit := wtwire.NewInitMessage(
		lnwire.NewRawFeatureVector(
			wtwire.AltruistSessionsOptional,
			wtwire.AnchorCommitOptional,
		),
		cfg.ChainHash,
	)

	s := &Server{
		cfg:       cfg,
		clients:   make(map[wtdb.SessionID]Peer),
		newPeers:  make(chan Peer),
		localInit: localInit,
		quit:      make(chan struct{}),
	}

	connMgr, err := connmgr.New(&connmgr.Config{
		Listeners: cfg.Listeners,
		OnAccept:  s.inboundPeerConnected,
		Dial:      noDial,
	})
	if err != nil {
		return nil, err
	}

	s.connMgr = connMgr

	return s, nil
}

// Start begins listening on the server's listeners.
func (s *Server) Start() error {
	s.started.Do(func() {
		log.Infof("Starting watchtower server")

		s.wg.Add(1)
		go s.peerHandler()

		s.connMgr.Start()

		log.Infof("Watchtower server started successfully")
	})
	return nil
}

// Stop shutdowns down the server's listeners and any active requests.
func (s *Server) Stop() error {
	s.stopped.Do(func() {
		log.Infof("Stopping watchtower server")

		s.connMgr.Stop()

		close(s.quit)
		s.wg.Wait()

		log.Infof("Watchtower server stopped successfully")
	})
	return nil
}

// inboundPeerConnected is the callback given to the connection manager, and is
// called each time a new connection is made to the watchtower. This method
// proxies the new peers by filtering out those that do not satisfy the
// server.Peer interface, and closes their connection. Successful connections
// will be passed on to the public InboundPeerConnected method.
func (s *Server) inboundPeerConnected(c net.Conn) {
	peer, ok := c.(Peer)
	if !ok {
		log.Warnf("incoming connection %T does not satisfy "+
			"server.Peer interface", c)
		c.Close()
		return
	}

	s.InboundPeerConnected(peer)
}

// InboundPeerConnected accepts a server.Peer, and handles the request submitted
// by the client. This method serves also as a public endpoint for locally
// registering new clients with the server.
func (s *Server) InboundPeerConnected(peer Peer) {
	select {
	case s.newPeers <- peer:
	case <-s.quit:
	}
}

// peerHandler processes newly accepted peers and spawns a client handler for
// each. The peerHandler is used to ensure that waitgrouped client handlers are
// spawned from a waitgrouped goroutine.
func (s *Server) peerHandler() {
	defer s.wg.Done()
	defer s.removeAllPeers()

	for {
		select {
		case peer := <-s.newPeers:
			s.wg.Add(1)
			go s.handleClient(peer)

		case <-s.quit:
			return
		}
	}
}

// handleClient processes a series watchtower messages sent by a client. The
// client may either send:
//  * a single CreateSession message.
//  * a series of StateUpdate messages.
//
// This method uses the server's peer map to ensure at most one peer using the
// same session id can enter the main event loop. The connection will be
// dropped by the watchtower if no messages are sent or received by the
// configured Read/WriteTimeouts.
//
// NOTE: This method MUST be run as a goroutine.
func (s *Server) handleClient(peer Peer) {
	defer s.wg.Done()

	// Use the connection's remote pubkey as the client's session id.
	id := wtdb.NewSessionIDFromPubKey(peer.RemotePub())

	// Register this peer in the server's client map, and defer the
	// connection's cleanup. If the peer already exists, we will close the
	// connection and exit immediately.
	err := s.addPeer(&id, peer)
	if err != nil {
		peer.Close()
		return
	}
	defer s.removePeer(&id, peer.RemoteAddr())

	msg, err := s.readMessage(peer)
	if err != nil {
		log.Errorf("Unable to read message from client %s@%s: %v",
			id, peer.RemoteAddr(), err)
		return
	}

	remoteInit, ok := msg.(*wtwire.Init)
	if !ok {
		log.Errorf("Client %s@%s did not send Init msg as first "+
			"message", id, peer.RemoteAddr())
		return
	}

	err = s.sendMessage(peer, s.localInit)
	if err != nil {
		log.Errorf("Unable to send Init msg to %s: %v", id, err)
		return
	}

	err = s.localInit.CheckRemoteInit(remoteInit, wtwire.FeatureNames)
	if err != nil {
		log.Errorf("Cannot support client %s: %v", id, err)
		return
	}

	nextMsg, err := s.readMessage(peer)
	if err != nil {
		log.Errorf("Unable to read watchtower msg from %s: %v",
			id, err)
		return
	}

	switch msg := nextMsg.(type) {
	case *wtwire.CreateSession:
		// Attempt to open a new session for this client.
		err = s.handleCreateSession(peer, &id, msg)
		if err != nil {
			log.Errorf("Unable to handle CreateSession "+
				"from %s: %v", id, err)
		}

	case *wtwire.DeleteSession:
		err = s.handleDeleteSession(peer, &id)
		if err != nil {
			log.Errorf("Unable to handle DeleteSession "+
				"from %s: %v", id, err)
		}

	case *wtwire.StateUpdate:
		err = s.handleStateUpdates(peer, &id, msg)
		if err != nil {
			log.Errorf("Unable to handle StateUpdate "+
				"from %s: %v", id, err)
		}

	default:
		log.Errorf("Received unsupported message type: %T "+
			"from %s", nextMsg, id)
	}
}

// connFailure is a default error used when a request failed with a non-zero
// error code.
type connFailure struct {
	ID   wtdb.SessionID
	Code wtwire.ErrorCode
}

// Error displays the SessionID and Code that caused the connection failure.
func (f *connFailure) Error() string {
	return fmt.Sprintf("connection with %s failed with code=%s",
		f.ID, f.Code,
	)
}

// readMessage receives and parses the next message from the given Peer. An
// error is returned if a message is not received before the server's read
// timeout, the read off the wire failed, or the message could not be
// deserialized.
func (s *Server) readMessage(peer Peer) (wtwire.Message, error) {
	// Set a read timeout to ensure we drop the client if not sent in a
	// timely manner.
	err := peer.SetReadDeadline(time.Now().Add(s.cfg.ReadTimeout))
	if err != nil {
		err = fmt.Errorf("unable to set read deadline: %v", err)
		return nil, err
	}

	// Pull the next message off the wire, and parse it according to the
	// watchtower wire specification.
	rawMsg, err := peer.ReadNextMessage()
	if err != nil {
		err = fmt.Errorf("unable to read message: %v", err)
		return nil, err
	}

	msgReader := bytes.NewReader(rawMsg)
	msg, err := wtwire.ReadMessage(msgReader, 0)
	if err != nil {
		err = fmt.Errorf("unable to parse message: %v", err)
		return nil, err
	}

	logMessage(peer, msg, true)

	return msg, nil
}

// sendMessage sends a watchtower wire message to the target peer.
func (s *Server) sendMessage(peer Peer, msg wtwire.Message) error {
	// TODO(conner): use buffer pool?

	var b bytes.Buffer
	_, err := wtwire.WriteMessage(&b, msg, 0)
	if err != nil {
		err = fmt.Errorf("unable to encode msg: %v", err)
		return err
	}

	err = peer.SetWriteDeadline(time.Now().Add(s.cfg.WriteTimeout))
	if err != nil {
		err = fmt.Errorf("unable to set write deadline: %v", err)
		return err
	}

	logMessage(peer, msg, false)

	_, err = peer.Write(b.Bytes())
	return err
}

// addPeer stores a client in the server's client map. An error is returned if a
// client with the same session id already exists.
func (s *Server) addPeer(id *wtdb.SessionID, peer Peer) error {
	s.clientMtx.Lock()
	defer s.clientMtx.Unlock()

	if existingPeer, ok := s.clients[*id]; ok {
		log.Infof("Already connected to peer %s@%s, disconnecting %s",
			id, existingPeer.RemoteAddr(), peer.RemoteAddr())
		return ErrPeerAlreadyConnected
	}
	s.clients[*id] = peer

	log.Infof("Accepted incoming peer %s@%s",
		id, peer.RemoteAddr())

	return nil
}

// removePeer deletes a client from the server's client map. If a peer is found,
// this method will close the peer's connection.
func (s *Server) removePeer(id *wtdb.SessionID, addr net.Addr) {
	log.Infof("Releasing incoming peer %s@%s", id, addr)

	s.clientMtx.Lock()
	peer, ok := s.clients[*id]
	delete(s.clients, *id)
	s.clientMtx.Unlock()

	if ok {
		peer.Close()
	}
}

// removeAllPeers iterates through the server's current set of peers and closes
// all open connections.
func (s *Server) removeAllPeers() {
	s.clientMtx.Lock()
	defer s.clientMtx.Unlock()

	for id, peer := range s.clients {
		log.Infof("Releasing incoming peer %s@%s", id,
			peer.RemoteAddr())

		delete(s.clients, id)
		peer.Close()
	}
}

// logMessage writes information about a message exchanged with a remote peer,
// using directional prepositions to signal whether the message was sent or
// received.
func logMessage(peer Peer, msg wtwire.Message, read bool) {
	var action = "Received"
	var preposition = "from"
	if !read {
		action = "Sending"
		preposition = "to"
	}

	summary := wtwire.MessageSummary(msg)
	if len(summary) > 0 {
		summary = "(" + summary + ")"
	}

	log.Debugf("%s %s%v %s %x@%s", action, msg.MsgType(), summary,
		preposition, peer.RemotePub().SerializeCompressed(),
		peer.RemoteAddr())
}

// noDial is a dummy dial method passed to the server's connmgr.
func noDial(_ net.Addr) (net.Conn, error) {
	return nil, fmt.Errorf("watchtower cannot make outgoing conns")
}