faraday/faraday.go

// Package faraday contains the main function for faraday.
package faraday

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	proxy "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
	"github.com/jessevdk/go-flags"
	"github.com/lightninglabs/faraday/chain"
	"github.com/lightninglabs/faraday/chanevents"
	"github.com/lightninglabs/faraday/frdrpc"
	"github.com/lightninglabs/faraday/frdrpcserver"
	"github.com/lightninglabs/faraday/frdrpcserver/perms"
	"github.com/lightninglabs/lndclient"
	"github.com/lightningnetwork/lnd/build"
	"github.com/lightningnetwork/lnd/clock"
	"github.com/lightningnetwork/lnd/kvdb"
	"github.com/lightningnetwork/lnd/lncfg"
	"github.com/lightningnetwork/lnd/lnrpc/verrpc"
	"github.com/lightningnetwork/lnd/macaroons"
	"github.com/lightningnetwork/lnd/signal"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/protobuf/encoding/protojson"
	"gopkg.in/macaroon-bakery.v2/bakery"
)

var (
	// customMarshalerOption is the configuration we use for the JSON
	// marshaler of the REST proxy. The default JSON marshaler only sets
	// OrigName to true, which instructs it to use the same field names as
	// specified in the proto file and not switch to camel case. What we
	// also want is that the marshaler prints all values, even if they are
	// falsey.
	customMarshalerOption = proxy.WithMarshalerOption(
		proxy.MIMEWildcard, &proxy.JSONPb{
			MarshalOptions: protojson.MarshalOptions{
				UseProtoNames:   true,
				EmitUnpopulated: true,
			},
		},
	)

	// maxMsgRecvSize is the largest message our REST proxy will receive. We
	// set this to 600MiB atm.
	maxMsgRecvSize = grpc.MaxCallRecvMsgSize(600 * 1024 * 1024)

	// errServerAlreadyStarted is the error that is returned if the server
	// is requested to start while it's already been started.
	errServerAlreadyStarted = fmt.Errorf("server can only be started once")

	// errServerStopped is the error that is returned if the server is
	// requested to start after it has been stopped. The Faraday struct is
	// not reusable after Stop.
	errServerStopped = fmt.Errorf("server has been stopped and cannot " +
		"be restarted")
)

// MinLndVersion is the minimum lnd version required. Note that apis that are
// only available in more recent versions are available at compile time, so this
// version should be bumped if additional functionality is included that depends
// on newer apis.
var MinLndVersion = &verrpc.Version{
	AppMajor: 0,
	AppMinor: 15,
	AppPatch: 4,
}

// Faraday is a struct that houses the faraday daemon and its dependencies.
type Faraday struct {
	*frdrpcserver.RPCServer

	// cfg is the faraday config.
	cfg *Config

	// started is used to ensure we only start/stop the faraday once.
	started atomic.Bool

	// stopped is set once Stop completes or Start fails. It prevents
	// reuse of the struct, since internal fields are not reset.
	stopped atomic.Bool

	// monitor is the channel events monitor.
	monitor *chanevents.Monitor

	// stores contains all the stores used by faraday.
	stores *stores

	// ctxCancel is a function that can be used to cancel the main context.
	ctxCancel context.CancelFunc

	lnd *lndclient.GrpcLndServices

	// lndOwned indicates whether Faraday created the lnd connection
	// itself (standalone mode via Start). When true, Stop will close
	// the connection. When false (subserver mode via StartAsSubserver),
	// the parent process manages the lnd lifecycle.
	lndOwned bool

	// bitcoinClient is set if the client opted to connect to a bitcoin
	// backend, if not, it will be nil.
	bitcoinClient chain.BitcoinClient

	macaroonService *lndclient.MacaroonService
	macaroonDB      kvdb.Backend

	// grpcServer is the main gRPC server that this service will register
	// itself with and accept client requests from.
	grpcServer *grpc.Server

	// rpcListener is the listener to use when starting the gRPC server.
	rpcListener net.Listener

	// restServer is the REST proxy server.
	restServer *http.Server
	restCancel func()

	wg sync.WaitGroup
}

// New creates a new Faraday instance with the given configuration.
func New(cfg *Config) *Faraday {
	return &Faraday{cfg: cfg}
}

// Start starts faraday and its dependencies with an RPC server included.
func (f *Faraday) Start() error {
	if f.stopped.Load() {
		return errServerStopped
	}

	if !f.started.CompareAndSwap(false, true) {
		return errServerAlreadyStarted
	}

	log.Infof("Starting Faraday version %s", Version())

	// Connect to the full suite of lightning services offered by lnd's
	// subservers.
	var err error
	f.lnd, err = lndclient.NewLndServices(&lndclient.LndServicesConfig{
		LndAddress:         f.cfg.Lnd.RPCServer,
		Network:            lndclient.Network(f.cfg.Network),
		CustomMacaroonPath: f.cfg.Lnd.MacaroonPath,
		TLSPath:            f.cfg.Lnd.TLSCertPath,
		CheckVersion:       MinLndVersion,
		RPCTimeout:         f.cfg.Lnd.RequestTimeout,
	})
	if err != nil {
		f.stopped.Store(true)
		f.started.Store(false)

		return fmt.Errorf("cannot connect to lightning services: %v",
			err)
	}
	f.lndOwned = true

	// Initialize faraday with its dependencies. If anything from here
	// on fails, we need to clean up the lnd connection.
	err = f.initialize(true)
	if err != nil {
		f.lnd.Close()
		f.stopped.Store(true)
		f.started.Store(false)

		return fmt.Errorf("error initializing faraday: %v", err)
	}

	fwdAnalyzer := chanevents.NewForwardingAnalyzer(
		f.stores.ChanEventsStore, f.lnd.LndServices,
	)
	cfg := &frdrpcserver.Config{
		Lnd:                f.lnd.LndServices,
		ChanEvents:         f.stores.ChanEventsStore,
		ForwardingAnalyzer: fwdAnalyzer,
		BitcoinClient:      f.bitcoinClient,
	}

	// Create the RPC server.
	f.RPCServer = frdrpcserver.NewRPCServer(cfg)

	err = f.startRPCServer()
	if err != nil {
		if f.macaroonService != nil {
			if e := f.macaroonService.Stop(); e != nil {
				log.Errorf("Error stopping macaroon "+
					"service: %v", e)
			}
			if e := f.macaroonDB.Close(); e != nil {
				log.Errorf("Error closing macaroon "+
					"DB: %v", e)
			}
		}
		f.lnd.Close()
		f.stopped.Store(true)
		f.started.Store(false)

		return fmt.Errorf("error starting RPC server: %v", err)
	}

	return nil
}

// startRPCServer starts the gRPC and REST RPC servers.
func (f *Faraday) startRPCServer() error {
	// Prepare the RPC server.
	serverTLSCfg, restClientCreds, err := getTLSConfig(f.cfg)
	if err != nil {
		return fmt.Errorf("error loading TLS config: %v", err)
	}

	// Depending on how far we got in initializing the server, we might need
	// to clean up certain services that were already started. Keep track of
	// them with this map of service name to shutdown function.
	shutdownFuncs := make(map[string]func() error)
	defer func() {
		for serviceName, shutdownFn := range shutdownFuncs {
			if err := shutdownFn(); err != nil {
				log.Errorf("Error shutting down %s service: %v",
					serviceName, err)
			}
		}
	}()

	// First we add the security interceptor to our gRPC server options that
	// checks the macaroons for validity.
	if f.macaroonService == nil {
		return fmt.Errorf("macaroon service must be initialized " +
			"before starting the RPC server")
	}

	unaryInterceptor, streamInterceptor, err :=
		f.macaroonService.Interceptors()

	if err != nil {
		return fmt.Errorf("error with macaroon interceptor: %v", err)
	}

	// Add our TLS configuration and then create our server instance. It's
	// important that we let gRPC create the TLS listener and we don't just
	// use tls.NewListener(). Otherwise we run into the ALPN error with non-
	// golang clients.
	tlsCredentials := credentials.NewTLS(serverTLSCfg)
	f.grpcServer = grpc.NewServer(
		grpc.UnaryInterceptor(unaryInterceptor),
		grpc.StreamInterceptor(streamInterceptor),
		grpc.Creds(tlsCredentials),
	)

	// Start the gRPC RPCServer listening for HTTP/2 connections.
	log.Info("Starting gRPC listener")
	f.rpcListener, err = net.Listen("tcp", f.cfg.RPCListen)
	if err != nil {
		return fmt.Errorf("gRPC server unable to listen on %v",
			f.cfg.RPCListen)
	}
	shutdownFuncs["gRPC listener"] = f.rpcListener.Close
	log.Infof("gRPC server listening on %s", f.rpcListener.Addr())

	frdrpc.RegisterFaradayServerServer(f.grpcServer, f)

	// We'll also create and start an accompanying proxy to serve clients
	// through REST. An empty address indicates REST is disabled.
	if f.cfg.RESTListen != "" {
		log.Infof("Starting REST proxy listener ")
		restListener, err := net.Listen("tcp", f.cfg.RESTListen)
		if err != nil {
			return fmt.Errorf("REST server unable to listen on "+
				"%v: %v", f.cfg.RESTListen, err)
		}
		restListener = tls.NewListener(
			restListener, serverTLSCfg,
		)
		shutdownFuncs["REST listener"] = restListener.Close
		log.Infof("REST server listening on %s", restListener.Addr())

		// We'll dial into the local gRPC server so we need to set some
		// gRPC dial options and CORS settings.
		var restCtx context.Context
		restCtx, f.restCancel = context.WithCancel(context.Background())
		mux := proxy.NewServeMux(customMarshalerOption)
		var restHandler http.Handler = mux
		if f.cfg.CORSOrigin != "" {
			restHandler = allowCORS(restHandler, f.cfg.CORSOrigin)
		}
		proxyOpts := []grpc.DialOption{
			grpc.WithTransportCredentials(*restClientCreds),
			grpc.WithDefaultCallOptions(maxMsgRecvSize),
		}

		// With TLS enabled by default, we cannot call 0.0.0.0
		// internally from the REST proxy as that IP address isn't in
		// the cert. We need to rewrite it to the loopback address.
		restProxyDest := f.cfg.RPCListen
		switch {
		case strings.Contains(restProxyDest, "0.0.0.0"):
			restProxyDest = strings.Replace(
				restProxyDest, "0.0.0.0", "127.0.0.1", 1,
			)

		case strings.Contains(restProxyDest, "[::]"):
			restProxyDest = strings.Replace(
				restProxyDest, "[::]", "[::1]", 1,
			)
		}
		err = frdrpc.RegisterFaradayServerHandlerFromEndpoint(
			restCtx, mux, restProxyDest, proxyOpts,
		)
		if err != nil {
			return err
		}
		f.restServer = &http.Server{
			Handler:           restHandler,
			ReadHeaderTimeout: 3 * time.Second,
		}

		f.wg.Add(1)
		go func() {
			defer f.wg.Done()
			err := f.restServer.Serve(restListener)
			// ErrServerClosed is always returned when the proxy is
			// shut down, so don't log it.
			if err != nil && err != http.ErrServerClosed {
				log.Error(err)
			}
		}()
	} else {
		log.Infof("REST proxy disabled")
	}

	f.wg.Add(1)
	go func() {
		defer f.wg.Done()
		if err := f.grpcServer.Serve(f.rpcListener); err != nil {
			log.Errorf("could not serve grpc server: %v", err)
		}
	}()

	// If we got here successfully, there's no need to shutdown anything
	// anymore.
	shutdownFuncs = nil

	return nil
}

// stopRPCServer stops the gRPC and REST RPC servers.
func (f *Faraday) stopRPCServer() {
	if f.restServer != nil {
		f.restCancel()
		err := f.restServer.Close()
		if err != nil {
			log.Errorf("unable to close REST listener: %v", err)
		}
	}

	if f.grpcServer != nil {
		f.grpcServer.Stop()
	}
}

// StartAsSubserver is an alternative to Start where the RPC server does not
// create its own gRPC server but registers to an existing one. The same goes
// for REST (if enabled), instead of creating an own mux and HTTP server, we
// register to an existing one.
func (f *Faraday) StartAsSubserver(lndGrpc *lndclient.GrpcLndServices,
	withMacaroonService bool) error {

	log.Infof("Starting Faraday subserver version %s", Version())

	// There should be no reason to start the daemon twice. Therefore,
	// return an error if that's tried. This is mostly to guard against
	// Start and StartAsSubserver both being called.
	if f.stopped.Load() {
		return errServerStopped
	}

	if !f.started.CompareAndSwap(false, true) {
		return errServerAlreadyStarted
	}

	// When starting as a subserver, we get passed in an already established
	// connection to lnd that might be shared among other subservers.
	f.lnd = lndGrpc

	// With lnd already pre-connected, initialize everything else, such as
	// the RPC server instance. If this fails, then nothing has been
	// started yet, and we can just return the error.
	err := f.initialize(withMacaroonService)
	if err != nil {
		f.stopped.Store(true)
		f.started.Store(false)

		return fmt.Errorf("error initializing faraday: %v", err)
	}

	fwdAnalyzer := chanevents.NewForwardingAnalyzer(
		f.stores.ChanEventsStore, lndGrpc.LndServices,
	)
	cfg := &frdrpcserver.Config{
		Lnd:                lndGrpc.LndServices,
		ChanEvents:         f.stores.ChanEventsStore,
		ForwardingAnalyzer: fwdAnalyzer,
		BitcoinClient:      f.bitcoinClient,
	}

	// Create the RPC server, but don't start it.
	f.RPCServer = frdrpcserver.NewRPCServer(cfg)

	return nil
}

// ValidateMacaroon extracts the macaroon from the context's gRPC metadata,
// checks its signature, makes sure all specified permissions for the called
// method are contained within and finally ensures all caveat conditions are
// met. A non-nil error is returned if any of the checks fail. This method is
// needed to enable faraday running as an external subserver in the same process
// as lnd but still validate its own macaroons.
func (f *Faraday) ValidateMacaroon(ctx context.Context,
	requiredPermissions []bakery.Op, fullMethod string) error {

	if f.macaroonService == nil {
		return fmt.Errorf("macaroon service not yet initialised")
	}

	// Delegate the call to faraday's own macaroon validator service.
	return f.macaroonService.ValidateMacaroon(
		ctx, requiredPermissions, fullMethod,
	)
}

// Stop shuts down Faraday: the RPC servers, macaroon service, and, if Faraday
// owns the lnd connection (standalone mode via Start), the lnd connection as
// well. In subserver mode (started via StartAsSubserver) the lnd connection is
// left open for the parent process to manage.
//
// Calling Stop on an already stopped or never-started instance is a no-op
// and returns nil.
func (f *Faraday) Stop() error {
	if !f.started.CompareAndSwap(true, false) {
		return nil
	}

	// Mark as permanently stopped so the struct cannot be reused.
	f.stopped.Store(true)

	log.Infof("Stopping Faraday")

	f.stopRPCServer()

	// Wait for the gRPC and REST serve goroutines to exit before
	// tearing down the macaroon service, so that in-flight RPCs
	// can complete cleanly.
	f.wg.Wait()

	if f.ctxCancel != nil {
		f.ctxCancel()
	}

	if f.monitor != nil {
		if err := f.monitor.Stop(); err != nil {
			log.Errorf("Error stopping channel event monitor: %v",
				err)
		}
	}

	if f.stores != nil {
		if err := f.stores.Close(); err != nil {
			log.Errorf("Error closing stores: %v", err)
		}
	}

	var stopErr error
	if f.macaroonService != nil {
		err := f.macaroonService.Stop()
		if err != nil {
			log.Errorf("Error stopping macaroon service: %v", err)
			stopErr = errors.Join(stopErr, err)
		}

		if err := f.macaroonDB.Close(); err != nil {
			log.Errorf("Error closing macaroon DB: %v", err)
			stopErr = errors.Join(stopErr, err)
		}
	}

	// Only close the lnd connection if we created it ourselves
	// (standalone mode). In subserver mode, the parent process
	// manages the shared lnd connection.
	if f.lndOwned && f.lnd != nil {
		f.lnd.Close()
	}

	return stopErr
}

// initialize sets up faraday with its dependencies.
func (f *Faraday) initialize(withMacaroonService bool) error {
	var err error

	if withMacaroonService {
		// Set up the macaroon service.
		var rks bakery.RootKeyStore
		rks, f.macaroonDB, err = lndclient.NewBoltMacaroonStore(
			f.cfg.FaradayDir, lncfg.MacaroonDBName,
			macDatabaseOpenTimeout,
		)
		if err != nil {
			return err
		}

		f.macaroonService, err = lndclient.NewMacaroonService(
			&lndclient.MacaroonServiceConfig{
				RootKeyStore:     rks,
				MacaroonLocation: faradayMacaroonLocation,
				MacaroonPath:     f.cfg.MacaroonPath,
				Checkers: []macaroons.Checker{
					macaroons.IPLockChecker,
				},
				RequiredPerms: perms.RequiredPermissions,
				DBPassword:    macDbDefaultPw,
				LndClient:     &f.lnd.LndServices,
				EphemeralKey:  lndclient.SharedKeyNUMS,
				KeyLocator:    lndclient.SharedKeyLocator,
			},
		)
		if err != nil {
			if e := f.macaroonDB.Close(); e != nil {
				log.Errorf("Error closing macaroon DB: %v", e)
			}

			return fmt.Errorf("error creating macaroon "+
				"service: %v", err)
		}

		// Start the macaroon service and let it create its default
		// macaroon in case it doesn't exist yet.
		if err := f.macaroonService.Start(); err != nil {
			if e := f.macaroonDB.Close(); e != nil {
				log.Errorf("Error closing macaroon DB: %v", e)
			}

			return fmt.Errorf("error starting macaroon "+
				"service: %v", err)
		}
	}

	// If the client chose to connect to a bitcoin client, get one now.
	if f.cfg.ChainConn {
		f.bitcoinClient, err = chain.NewBitcoinClient(f.cfg.Bitcoin)
		if err != nil {
			if f.macaroonService != nil {
				if e := f.macaroonService.Stop(); e != nil {
					log.Errorf("Error stopping macaroon "+
						"service: %v", e)
				}
				if e := f.macaroonDB.Close(); e != nil {
					log.Errorf("Error closing macaroon "+
						"DB: %v", e)
				}
			}

			return err
		}
	}

	// Create any relevant stores.
	f.stores, err = NewStores(*f.cfg, clock.NewDefaultClock())
	if err != nil {
		return fmt.Errorf("could not create stores: %v", err)
	}

	// Create the channel event monitor. ChanEvents may be nil on
	// initialization paths that don't go through DefaultConfig (e.g. when
	// faraday runs as a subserver), so fall back to a zero-value config
	// instead of dereferencing a nil pointer.
	var chanEventsCfg chanevents.Config
	if f.cfg.ChanEvents != nil {
		chanEventsCfg = *f.cfg.ChanEvents
	}

	f.monitor = chanevents.NewMonitor(
		f.lnd.Client, f.stores.ChanEventsStore, chanEventsCfg,
	)

	ctx, cancel := context.WithCancel(context.Background())
	f.ctxCancel = cancel

	if err := f.monitor.Start(ctx); err != nil {
		cancel()

		return fmt.Errorf("could not start channel event "+
			"monitor: %v", err)
	}

	return nil
}

// allowCORS wraps the given http.Handler with a function that adds the
// Access-Control-Allow-Origin header to the response.
func allowCORS(handler http.Handler, origin string) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Access-Control-Allow-Origin", origin)
		handler.ServeHTTP(w, r)
	})
}

// Main is the real entry point for faraday. It is required to ensure that
// defers are properly executed when os.Exit() is called.
func Main() error {
	// Start with a default config.
	config := DefaultConfig()

	// Parse command line options to obtain user specified values.
	if _, err := flags.Parse(&config); err != nil {
		return err
	}

	// Show the version and exit if the version flag was specified.
	appName := filepath.Base(os.Args[0])
	appName = strings.TrimSuffix(appName, filepath.Ext(appName))
	if config.ShowVersion {
		fmt.Println(appName, "version", Version())
		os.Exit(0)
	}

	// Hook interceptor for os signals.
	shutdownInterceptor, err := signal.Intercept()
	if err != nil {
		return err
	}

	// Setup logging before parsing the config.
	logWriter := build.NewRotatingLogWriter()
	subLogMgr := build.NewSubLoggerManager(
		build.NewDefaultLogHandlers(config.Logging, logWriter)...,
	)
	SetupLoggers(subLogMgr, shutdownInterceptor)
	err = build.ParseAndSetDebugLevels(config.DebugLevel, subLogMgr)
	if err != nil {
		return err
	}

	if err := ValidateConfig(&config); err != nil {
		return fmt.Errorf("error validating config: %v", err)
	}

	server := New(&config)
	err = server.Start()
	if err != nil {
		return fmt.Errorf("error starting faraday: %w", err)
	}

	// Run until the user terminates.
	<-shutdownInterceptor.ShutdownChannel()
	log.Infof("Received shutdown signal.")

	if err := server.Stop(); err != nil {
		return fmt.Errorf("error stopping faraday: %w", err)
	}

	return nil
}