diff --git a/watchtower/server/interface.go b/watchtower/wtserver/interface.go similarity index 99% rename from watchtower/server/interface.go rename to watchtower/wtserver/interface.go index 98a992136..69d4ec9cf 100644 --- a/watchtower/server/interface.go +++ b/watchtower/wtserver/interface.go @@ -1,4 +1,4 @@ -package server +package wtserver import ( "io" diff --git a/watchtower/server/log.go b/watchtower/wtserver/log.go similarity index 92% rename from watchtower/server/log.go rename to watchtower/wtserver/log.go index 5d5740e8f..7939bd97f 100644 --- a/watchtower/server/log.go +++ b/watchtower/wtserver/log.go @@ -1,4 +1,4 @@ -package server +package wtserver import ( "github.com/btcsuite/btclog" @@ -12,7 +12,7 @@ var log btclog.Logger // The default amount of logging is none. func init() { - UseLogger(build.NewSubLogger("WTSV", nil)) + UseLogger(build.NewSubLogger("WTWR", nil)) } // DisableLog disables all library log output. Logging output is disabled diff --git a/watchtower/server/mock.go b/watchtower/wtserver/mock.go similarity index 99% rename from watchtower/server/mock.go rename to watchtower/wtserver/mock.go index 54e0bf963..758e4ae2e 100644 --- a/watchtower/server/mock.go +++ b/watchtower/wtserver/mock.go @@ -1,6 +1,6 @@ // +build dev -package server +package wtserver import ( "fmt" diff --git a/watchtower/server/server.go b/watchtower/wtserver/server.go similarity index 96% rename from watchtower/server/server.go rename to watchtower/wtserver/server.go index 274f06d57..24483d3ac 100644 --- a/watchtower/server/server.go +++ b/watchtower/wtserver/server.go @@ -1,4 +1,4 @@ -package server +package wtserver import ( "bytes" @@ -108,8 +108,12 @@ func (s *Server) Start() error { return nil } + log.Infof("Starting watchtower server") + s.connMgr.Start() + log.Infof("Watchtower server started successfully") + return nil } @@ -120,11 +124,15 @@ func (s *Server) Stop() error { return nil } + log.Infof("Stopping watchtower server") + s.connMgr.Stop() close(s.quit) s.wg.Wait() + log.Infof("Watchtower server stopped successfully") + return nil } @@ -287,6 +295,8 @@ func (s *Server) handleClient(peer Peer) { } } +// handleInit accepts the local and remote Init messages, and verifies that the +// client is not requesting any required features that are unknown to the tower. func (s *Server) handleInit(localInit, remoteInit *wtwire.Init) error { remoteLocalFeatures := lnwire.NewFeatureVector( remoteInit.LocalFeatures, wtwire.LocalFeatures, @@ -312,33 +322,6 @@ func (s *Server) handleInit(localInit, remoteInit *wtwire.Init) error { 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, @@ -550,6 +533,37 @@ func (s *Server) replyStateUpdate(peer Peer, id *wtdb.SessionID, } } +// 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) + nextMsg, err := wtwire.ReadMessage(msgReader, 0) + if err != nil { + err = fmt.Errorf("unable to parse message: %v", err) + return nil, err + } + + return nextMsg, 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? diff --git a/watchtower/server/server_test.go b/watchtower/wtserver/server_test.go similarity index 94% rename from watchtower/server/server_test.go rename to watchtower/wtserver/server_test.go index f36e26982..3ca056af2 100644 --- a/watchtower/server/server_test.go +++ b/watchtower/wtserver/server_test.go @@ -1,6 +1,6 @@ // +build dev -package server_test +package wtserver_test import ( "bytes" @@ -12,8 +12,8 @@ import ( "github.com/btcsuite/btcd/chaincfg" "github.com/btcsuite/btcutil" "github.com/lightningnetwork/lnd/lnwire" - "github.com/lightningnetwork/lnd/watchtower/server" "github.com/lightningnetwork/lnd/watchtower/wtdb" + "github.com/lightningnetwork/lnd/watchtower/wtserver" "github.com/lightningnetwork/lnd/watchtower/wtwire" ) @@ -36,8 +36,8 @@ func randPubKey(t *testing.T) *btcec.PublicKey { // initServer creates and starts a new server using the server.DB and timeout. // If the provided database is nil, a mock db will be used. -func initServer(t *testing.T, db server.DB, - timeout time.Duration) server.Interface { +func initServer(t *testing.T, db wtserver.DB, + timeout time.Duration) wtserver.Interface { t.Helper() @@ -45,7 +45,7 @@ func initServer(t *testing.T, db server.DB, db = wtdb.NewMockDB() } - s, err := server.New(&server.Config{ + s, err := wtserver.New(&wtserver.Config{ DB: db, ReadTimeout: timeout, WriteTimeout: timeout, @@ -79,8 +79,8 @@ func TestServerOnlyAcceptOnePeer(t *testing.T) { // Create two peers using the same session id. peerPub := randPubKey(t) - peer1 := server.NewMockPeer(peerPub, nil, 0) - peer2 := server.NewMockPeer(peerPub, nil, 0) + peer1 := wtserver.NewMockPeer(peerPub, nil, 0) + peer2 := wtserver.NewMockPeer(peerPub, nil, 0) // Serialize a Init message to be sent by both peers. init := wtwire.NewInitMessage( @@ -106,8 +106,8 @@ func TestServerOnlyAcceptOnePeer(t *testing.T) { // Try to send a message on either peer, and record the opposite peer as // the one we assume to be rejected. var ( - rejectedPeer *server.MockPeer - acceptedPeer *server.MockPeer + rejectedPeer *wtserver.MockPeer + acceptedPeer *wtserver.MockPeer ) select { case peer1.IncomingMsgs <- msg: @@ -192,7 +192,7 @@ func testServerCreateSession(t *testing.T, i int, test createSessionTestCase) { // Create a new client and connect to server. peerPub := randPubKey(t) - peer := server.NewMockPeer(peerPub, nil, 0) + peer := wtserver.NewMockPeer(peerPub, nil, 0) connect(t, i, s, peer, test.initMsg, timeoutDuration) // Send the CreateSession message, and wait for a reply. @@ -220,7 +220,7 @@ func testServerCreateSession(t *testing.T, i int, test createSessionTestCase) { // Simulate a peer with the same session id connection to the server // again. - peer = server.NewMockPeer(peerPub, nil, 0) + peer = wtserver.NewMockPeer(peerPub, nil, 0) connect(t, i, s, peer, test.initMsg, timeoutDuration) // Send the _same_ CreateSession message as the first attempt. @@ -492,7 +492,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) { // Create a new client and connect to the server. peerPub := randPubKey(t) - peer := server.NewMockPeer(peerPub, nil, 0) + peer := wtserver.NewMockPeer(peerPub, nil, 0) connect(t, i, s, peer, test.initMsg, timeoutDuration) // Register a session for this client to use in the subsequent tests. @@ -512,7 +512,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) { // Now that the original connection has been closed, connect a new // client with the same session id. - peer = server.NewMockPeer(peerPub, nil, 0) + peer = wtserver.NewMockPeer(peerPub, nil, 0) connect(t, i, s, peer, test.initMsg, timeoutDuration) // Send the intended StateUpdate messages in series. @@ -523,7 +523,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) { if update == nil { assertConnClosed(t, peer, 2*timeoutDuration) - peer = server.NewMockPeer(peerPub, nil, 0) + peer = wtserver.NewMockPeer(peerPub, nil, 0) connect(t, i, s, peer, test.initMsg, timeoutDuration) continue @@ -547,7 +547,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) { assertConnClosed(t, peer, 2*timeoutDuration) } -func connect(t *testing.T, i int, s server.Interface, peer *server.MockPeer, +func connect(t *testing.T, i int, s wtserver.Interface, peer *wtserver.MockPeer, initMsg *wtwire.Init, timeout time.Duration) { s.InboundPeerConnected(peer) @@ -555,9 +555,9 @@ func connect(t *testing.T, i int, s server.Interface, peer *server.MockPeer, recvReply(t, i, "Init", peer, timeout) } -// sendMsg sends a wtwire.Message message via a server.MockPeer. +// sendMsg sends a wtwire.Message message via a wtserver.MockPeer. func sendMsg(t *testing.T, i int, msg wtwire.Message, - peer *server.MockPeer, timeout time.Duration) { + peer *wtserver.MockPeer, timeout time.Duration) { t.Helper() @@ -579,7 +579,7 @@ func sendMsg(t *testing.T, i int, msg wtwire.Message, // expected reply type. The supported replies are CreateSessionReply and // StateUpdateReply. func recvReply(t *testing.T, i int, name string, - peer *server.MockPeer, timeout time.Duration) wtwire.Message { + peer *wtserver.MockPeer, timeout time.Duration) wtwire.Message { t.Helper() @@ -623,7 +623,7 @@ func recvReply(t *testing.T, i int, name string, // assertConnClosed checks that the peer's connection is closed before the // timeout expires. -func assertConnClosed(t *testing.T, peer *server.MockPeer, duration time.Duration) { +func assertConnClosed(t *testing.T, peer *wtserver.MockPeer, duration time.Duration) { t.Helper() select {