server.go

package server

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

	"github.com/btcsuite/btcd/btcec"
	"github.com/btcsuite/btcd/connmgr"
	"github.com/btcsuite/btcutil"
	"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")
)

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

	// NodePrivKey is private key to be used in accepting new brontide
	// connections.
	NodePrivKey *btcec.PrivateKey

	// 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)
}

// 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  int32 // atomic
	shutdown int32 // atomic

	cfg *Config

	connMgr *connmgr.ConnManager

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

	globalFeatures *lnwire.RawFeatureVector
	localFeatures  *lnwire.RawFeatureVector

	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) {
	localFeatures := lnwire.NewRawFeatureVector(
		wtwire.WtSessionsOptional,
	)

	s := &Server{
		cfg:            cfg,
		clients:        make(map[wtdb.SessionID]Peer),
		globalFeatures: lnwire.NewRawFeatureVector(),
		localFeatures:  localFeatures,
		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 {
	// Already running?
	if !atomic.CompareAndSwapInt32(&s.started, 0, 1) {
		return nil
	}

	s.connMgr.Start()

	return nil
}

// Stop shutdowns down the server's listeners and any active requests.
func (s *Server) Stop() error {
	// Bail if we're already shutting down.
	if !atomic.CompareAndSwapInt32(&s.shutdown, 0, 1) {
		return nil
	}

	s.connMgr.Stop()

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

	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) {
	s.wg.Add(1)
	go s.handleClient(peer)
}

// 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)

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

	localInit := wtwire.NewInitMessage(
		s.localFeatures, s.globalFeatures,
	)

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

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

	// stateUpdateOnlyMode will become true if the client's first message is
	// a StateUpdate. If instead, it is a CreateSession, this method will exit
	// immediately after replying. We track this to ensure that the client
	// can't send a CreateSession after having already sent a StateUpdate.
	var stateUpdateOnlyMode bool
	for {
		select {
		case <-s.quit:
			return
		default:
		}

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

		// Process the request according to the message's type.
		switch msg := nextMsg.(type) {

		// A CreateSession indicates a request to establish a new session
		// with our watchtower.
		case *wtwire.CreateSession:
			// Ensure CreateSession can only be sent as the first
			// message.
			if stateUpdateOnlyMode {
				log.Errorf("client %x sent CreateSession after "+
					"StateUpdate", id)
				return
			}

			log.Infof("Received CreateSession from %s, "+
				"version=%d nupdates=%d rewardrate=%d "+
				"sweepfeerate=%d", id, msg.BlobVersion,
				msg.MaxUpdates, msg.RewardRate,
				msg.SweepFeeRate)

			// 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)
			}

			// Exit after replying to CreateSession.
			return

		// A StateUpdate indicates an existing client attempting to
		// back-up a revoked commitment state.
		case *wtwire.StateUpdate:
			log.Infof("Received SessionUpdate from %s, seqnum=%d "+
				"lastapplied=%d complete=%v hint=%x", id,
				msg.SeqNum, msg.LastApplied, msg.IsComplete,
				msg.Hint[:])

			// Try to accept the state update from the client.
			err := s.handleStateUpdate(peer, &id, msg)
			if err != nil {
				log.Errorf("unable to handle StateUpdate "+
					"from %s: %v", id, err)
				return
			}

			// If the client signals that this is last StateUpdate
			// message, we can disconnect the client.
			if msg.IsComplete == 1 {
				return
			}

			// The client has signaled that more StateUpdates are
			// yet to come. Enter state-update-only mode to disallow
			// future sends of CreateSession messages.
			stateUpdateOnlyMode = true

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

func (s *Server) handleInit(localInit, remoteInit *wtwire.Init) error {
	remoteLocalFeatures := lnwire.NewFeatureVector(
		remoteInit.LocalFeatures, wtwire.LocalFeatures,
	)
	remoteGlobalFeatures := lnwire.NewFeatureVector(
		remoteInit.GlobalFeatures, wtwire.GlobalFeatures,
	)

	unknownLocalFeatures := remoteLocalFeatures.UnknownRequiredFeatures()
	if len(unknownLocalFeatures) > 0 {
		err := fmt.Errorf("Peer set unknown local feature bits: %v",
			unknownLocalFeatures)
		return err
	}

	unknownGlobalFeatures := remoteGlobalFeatures.UnknownRequiredFeatures()
	if len(unknownGlobalFeatures) > 0 {
		err := fmt.Errorf("Peer set unknown global feature bits: %v",
			unknownGlobalFeatures)
		return err
	}

	return nil
}

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)
	nextMsg, err := wtwire.ReadMessage(msgReader, 0)
	if err != nil {
		err = fmt.Errorf("unable to parse message: %v", err)
		return nil, err
	}

	return nextMsg, nil
}

// handleCreateSession processes a CreateSession message from the peer, and returns
// a CreateSessionReply in response. This method will only succeed if no existing
// session info is known about the session id. If an existing session is found,
// the reward address is returned in case the client lost our reply.
func (s *Server) handleCreateSession(peer Peer, id *wtdb.SessionID,
	init *wtwire.CreateSession) error {

	// TODO(conner): validate accept against policy

	// Query the db for session info belonging to the client's session id.
	existingInfo, err := s.cfg.DB.GetSessionInfo(id)
	switch {

	// We already have a session corresponding to this session id, return an
	// error signaling that it already exists in our database. We return the
	// reward address to the client in case they were not able to process
	// our reply earlier.
	case err == nil:
		log.Debugf("Already have session for %s", id)
		return s.replyCreateSession(
			peer, id, wtwire.CreateSessionCodeAlreadyExists,
			[]byte(existingInfo.RewardAddress),
		)

	// Some other database error occurred, return a temporary failure.
	case err != wtdb.ErrSessionNotFound:
		log.Errorf("unable to load session info for %s", id)
		return s.replyCreateSession(
			peer, id, wtwire.CodeTemporaryFailure, nil,
		)
	}

	// Now that we've established that this session does not exist in the
	// database, retrieve the sweep address that will be given to the
	// client. This address is to be included by the client when signing
	// sweep transactions destined for this tower, if its negotiated output
	// is not dust.
	rewardAddress, err := s.cfg.NewAddress()
	if err != nil {
		log.Errorf("unable to generate reward addr for %s", id)
		return s.replyCreateSession(
			peer, id, wtwire.CodeTemporaryFailure, nil,
		)
	}

	rewardAddrBytes := rewardAddress.ScriptAddress()

	// TODO(conner): create invoice for upfront payment

	// Assemble the session info using the agreed upon parameters, reward
	// address, and session id.
	info := wtdb.SessionInfo{
		ID:            *id,
		Version:       init.BlobVersion,
		MaxUpdates:    init.MaxUpdates,
		RewardRate:    init.RewardRate,
		SweepFeeRate:  init.SweepFeeRate,
		RewardAddress: rewardAddrBytes,
	}

	// Insert the session info into the watchtower's database. If
	// successful, the session will now be ready for use.
	err = s.cfg.DB.InsertSessionInfo(&info)
	if err != nil {
		log.Errorf("unable to create session for %s", id)
		return s.replyCreateSession(
			peer, id, wtwire.CodeTemporaryFailure, nil,
		)
	}

	log.Infof("Accepted session for %s", id)

	return s.replyCreateSession(
		peer, id, wtwire.CodeOK, rewardAddrBytes,
	)
}

// handleStateUpdate processes a StateUpdate message request from a client. An
// attempt will be made to insert the update into the db, where it is validated
// against the client's session. The possible errors are then mapped back to
// StateUpdateCodes specified by the watchtower wire protocol, and sent back
// using a StateUpdateReply message.
func (s *Server) handleStateUpdate(peer Peer, id *wtdb.SessionID,
	update *wtwire.StateUpdate) error {

	var (
		lastApplied uint16
		failCode    wtwire.ErrorCode
		err         error
	)

	sessionUpdate := wtdb.SessionStateUpdate{
		ID:            *id,
		Hint:          update.Hint,
		SeqNum:        update.SeqNum,
		LastApplied:   update.LastApplied,
		EncryptedBlob: update.EncryptedBlob,
	}

	lastApplied, err = s.cfg.DB.InsertStateUpdate(&sessionUpdate)
	switch {
	case err == nil:
		log.Infof("State update %d accepted for %s",
			update.SeqNum, id)

		failCode = wtwire.CodeOK

	// Return a permanent failure if a client tries to send an update for
	// which we have no session.
	case err == wtdb.ErrSessionNotFound:
		failCode = wtwire.CodePermanentFailure

	case err == wtdb.ErrSeqNumAlreadyApplied:
		failCode = wtwire.CodePermanentFailure

		// TODO(conner): remove session state for protocol
		// violation. Could also double as clean up method for
		// session-related state.

	case err == wtdb.ErrLastAppliedReversion:
		failCode = wtwire.StateUpdateCodeClientBehind

	case err == wtdb.ErrSessionConsumed:
		failCode = wtwire.StateUpdateCodeMaxUpdatesExceeded

	case err == wtdb.ErrUpdateOutOfOrder:
		failCode = wtwire.StateUpdateCodeSeqNumOutOfOrder

	default:
		failCode = wtwire.CodeTemporaryFailure
	}

	return s.replyStateUpdate(
		peer, id, failCode, lastApplied,
	)
}

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

// 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=%v",
		f.ID, f.Code,
	)
}

// replyCreateSession sends a response to a CreateSession from a client. If the
// status code in the reply is OK, the error from the write will be bubbled up.
// Otherwise, this method returns a connection error to ensure we don't continue
// communication with the client.
func (s *Server) replyCreateSession(peer Peer, id *wtdb.SessionID,
	code wtwire.ErrorCode, data []byte) error {

	msg := &wtwire.CreateSessionReply{
		Code: code,
		Data: data,
	}

	err := s.sendMessage(peer, msg)
	if err != nil {
		log.Errorf("unable to send CreateSessionReply to %s", id)
	}

	// Return the write error if the request succeeded.
	if code == wtwire.CodeOK {
		return err
	}

	// Otherwise the request failed, return a connection failure to
	// disconnect the client.
	return &connFailure{
		ID:   *id,
		Code: uint16(code),
	}
}

// replyStateUpdate sends a response to a StateUpdate from a client. If the
// status code in the reply is OK, the error from the write will be bubbled up.
// Otherwise, this method returns a connection error to ensure we don't continue
// communication with the client.
func (s *Server) replyStateUpdate(peer Peer, id *wtdb.SessionID,
	code wtwire.StateUpdateCode, lastApplied uint16) error {

	msg := &wtwire.StateUpdateReply{
		Code:        code,
		LastApplied: lastApplied,
	}

	err := s.sendMessage(peer, msg)
	if err != nil {
		log.Errorf("unable to send StateUpdateReply to %s", id)
	}

	// Return the write error if the request succeeded.
	if code == wtwire.CodeOK {
		return err
	}

	// Otherwise the request failed, return a connection failure to
	// disconnect the client.
	return &connFailure{
		ID:   *id,
		Code: uint16(code),
	}
}

// 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
	}

	_, 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) {
	log.Infof("Releasing incoming peer %s", id)

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

	if ok {
		peer.Close()
	}
}

// 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")
}