[watchtower/server] Server-Side Wire Protocol

watchtower/server/interface.go

@@ -0,0 +1,66 @@
+package server
+
+import (
+	"io"
+	"net"
+	"time"
+
+	"github.com/btcsuite/btcd/btcec"
+	"github.com/lightningnetwork/lnd/watchtower/wtdb"
+)
+
+// Interface represents a simple, listen-only service that accepts watchtower
+// clients, and provides responses to their requests.
+type Interface interface {
+	// InboundPeerConnected accepts a new watchtower client, and handles any
+	// requests sent by the peer.
+	InboundPeerConnected(Peer)
+
+	// Start sets up the watchtower server.
+	Start() error
+
+	// Stop cleans up the watchtower's current connections and resources.
+	Stop() error
+}
+
+// Peer is the primary interface used to abstract watchtower clients.
+type Peer interface {
+	io.WriteCloser
+
+	// ReadNextMessage pulls the next framed message from the client.
+	ReadNextMessage() ([]byte, error)
+
+	// SetWriteDeadline specifies the time by which the client must have
+	// read a message sent by the server. In practice, the connection is
+	// buffered, so the client must read enough from the connection to
+	// support the server adding another reply.
+	SetWriteDeadline(time.Time) error
+
+	// SetReadDeadline specifies the time by which the client must send
+	// another message.
+	SetReadDeadline(time.Time) error
+
+	// RemotePub returns the client's public key.
+	RemotePub() *btcec.PublicKey
+
+	// RemoteAddr returns the client's network address.
+	RemoteAddr() net.Addr
+}
+
+// DB provides the server access to session creation and retrieval, as well as
+// persisting state updates sent by clients.
+type DB interface {
+	// InsertSessionInfo saves a newly agreed-upon session from a client.
+	// This method should fail if a session with the same session id already
+	// exists.
+	InsertSessionInfo(*wtdb.SessionInfo) error
+
+	// GetSessionInfo retrieves the SessionInfo associated with the session
+	// id, if it exists.
+	GetSessionInfo(*wtdb.SessionID) (*wtdb.SessionInfo, error)
+
+	// InsertStateUpdate persists a state update sent by a client, and
+	// validates the update against the current SessionInfo stored under the
+	// update's session id..
+	InsertStateUpdate(*wtdb.SessionStateUpdate) (uint16, error)
+}

watchtower/server/log.go

@@ -0,0 +1,29 @@
+package server
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger("WTSV", nil))
+}
+
+// DisableLog disables all library log output.  Logging output is disabled
+// by default until UseLogger is called.
+func DisableLog() {
+	UseLogger(btclog.Disabled)
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

watchtower/server/mock.go

@@ -0,0 +1,98 @@
+// +build dev
+
+package server
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/btcsuite/btcd/btcec"
+)
+
+type MockPeer struct {
+	remotePub  *btcec.PublicKey
+	remoteAddr net.Addr
+
+	IncomingMsgs chan []byte
+	OutgoingMsgs chan []byte
+
+	writeDeadline <-chan time.Time
+	readDeadline  <-chan time.Time
+
+	Quit chan struct{}
+}
+
+func NewMockPeer(pk *btcec.PublicKey, addr net.Addr, bufferSize int) *MockPeer {
+	return &MockPeer{
+		remotePub:    pk,
+		remoteAddr:   addr,
+		IncomingMsgs: make(chan []byte, bufferSize),
+		OutgoingMsgs: make(chan []byte, bufferSize),
+		Quit:         make(chan struct{}),
+	}
+}
+
+func (p *MockPeer) Write(b []byte) (n int, err error) {
+	select {
+	case p.OutgoingMsgs <- b:
+		return len(b), nil
+	case <-p.writeDeadline:
+		return 0, fmt.Errorf("write timeout expired")
+	case <-p.Quit:
+		return 0, fmt.Errorf("connection closed")
+	}
+}
+
+func (p *MockPeer) Close() error {
+	select {
+	case <-p.Quit:
+		return fmt.Errorf("connection already closed")
+	default:
+		close(p.Quit)
+		return nil
+	}
+}
+
+func (p *MockPeer) ReadNextMessage() ([]byte, error) {
+	select {
+	case b := <-p.IncomingMsgs:
+		return b, nil
+	case <-p.readDeadline:
+		return nil, fmt.Errorf("read timeout expired")
+	case <-p.Quit:
+		return nil, fmt.Errorf("connection closed")
+	}
+}
+
+func (p *MockPeer) SetWriteDeadline(t time.Time) error {
+	if t.IsZero() {
+		p.writeDeadline = nil
+		return nil
+	}
+
+	duration := time.Until(t)
+	p.writeDeadline = time.After(duration)
+
+	return nil
+}
+
+func (p *MockPeer) SetReadDeadline(t time.Time) error {
+	if t.IsZero() {
+		p.readDeadline = nil
+		return nil
+	}
+
+	duration := time.Until(t)
+	p.readDeadline = time.After(duration)
+
+	return nil
+}
+
+func (p *MockPeer) RemotePub() *btcec.PublicKey {
+	return p.remotePub
+}
+
+func (p *MockPeer) RemoteAddr() net.Addr {
+	return p.remoteAddr
+}