config+log+rpcmiddleware: add RPC middleware package

config.go

@@ -18,6 +18,7 @@ import (
 	"github.com/lightninglabs/faraday"
 	"github.com/lightninglabs/faraday/chain"
 	"github.com/lightninglabs/faraday/frdrpcserver"
+	mid "github.com/lightninglabs/lightning-terminal/rpcmiddleware"
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/loopd"
 	"github.com/lightninglabs/pool"
@@ -183,6 +184,8 @@ type Config struct {
 	PoolMode string       `long:"pool-mode" description:"The mode to run pool in, either 'integrated' (default) or 'remote'. 'integrated' means poold is started alongside the UI and everything is stored in pool's main data directory, configure everything by using the --pool.* flags. 'remote' means the UI connects to an existing poold node and acts as a proxy for gRPC calls to it." choice:"integrated" choice:"remote"`
 	Pool     *pool.Config `group:"Integrated pool options (use when pool-mode=integrated)" namespace:"pool"`
 
+	RPCMiddleware *mid.Config `group:"RPC middleware options" namespace:"rpcmiddleware"`
+
 	// faradayRpcConfig is a subset of faraday's full configuration that is
 	// passed into faraday's RPC server.
 	faradayRpcConfig *frdrpcserver.Config
@@ -321,6 +324,7 @@ func defaultConfig() *Config {
 		Loop:                 &loopDefaultConfig,
 		PoolMode:             defaultPoolMode,
 		Pool:                 &poolDefaultConfig,
+		RPCMiddleware:        mid.DefaultConfig(),
 		FirstLNCConnDeadline: defaultFirstLNCConnTimeout,
 	}
 }
@@ -382,6 +386,14 @@ func loadAndValidateConfig(interceptor signal.Interceptor) (*Config, error) {
 		return nil, err
 	}
 
+	// To enable the rpc middleware interceptor, we need LND's RPC
+	// middleware interceptor to be enabled. In remote mode we can't
+	// influence whether that's enabled on lnd. But in integrated mode we
+	// can overwrite the flag here.
+	if !cfg.lndRemote && !cfg.RPCMiddleware.Disabled {
+		cfg.Lnd.RPCMiddleware.Enable = true
+	}
+
 	// Validate the lightning-terminal config options.
 	litDir := lnd.CleanAndExpandPath(preCfg.LitDir)
 	cfg.LetsEncryptDir = lncfg.CleanAndExpandPath(cfg.LetsEncryptDir)

log.go

@@ -4,6 +4,7 @@ import (
 	"github.com/btcsuite/btclog"
 	"github.com/lightninglabs/faraday"
 	"github.com/lightninglabs/lightning-node-connect/mailbox"
+	mid "github.com/lightninglabs/lightning-terminal/rpcmiddleware"
 	"github.com/lightninglabs/lightning-terminal/session"
 	"github.com/lightninglabs/loop/loopd"
 	"github.com/lightninglabs/pool"
@@ -59,6 +60,7 @@ func SetupLoggers(root *build.RotatingLogWriter, intercept signal.Interceptor) {
 	lnd.AddSubLogger(root, Subsystem, intercept, UseLogger)
 	lnd.AddSubLogger(root, session.Subsystem, intercept, session.UseLogger)
 	lnd.AddSubLogger(root, mailbox.Subsystem, intercept, mailbox.UseLogger)
+	lnd.AddSubLogger(root, mid.Subsystem, intercept, mid.UseLogger)
 
 	// Add daemon loggers to lnd's root logger.
 	faraday.SetupLoggers(root, intercept)

rpcmiddleware/config.go

@@ -0,0 +1,22 @@
+package rpcmiddleware
+
+import "time"
+
+const (
+	// DefaultInterceptTimeout is the default maximum time we're allowed to
+	// take to respond to an RPC middleware interception request.
+	DefaultInterceptTimeout = time.Second * 2
+)
+
+// Config is the configuration struct for the RPC middleware.
+type Config struct {
+	Disabled         bool          `long:"disabled" description:"Disable the RPC middleware"`
+	InterceptTimeout time.Duration `long:"intercept-timeout" description:"The maximum time the RPC middleware is allowed to take for intercepting each RPC request"`
+}
+
+// DefaultConfig returns the default RPC middleware configuration.
+func DefaultConfig() *Config {
+	return &Config{
+		InterceptTimeout: DefaultInterceptTimeout,
+	}
+}

rpcmiddleware/interface.go

@@ -0,0 +1,340 @@
+package rpcmiddleware
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
+)
+
+var (
+	// errorType is the reflection type of the error interface.
+	errorType = reflect.TypeOf((*error)(nil)).Elem()
+
+	// protoMessageType is the reflection type of the proto.Message
+	// interface.
+	protoMessageType = reflect.TypeOf((*proto.Message)(nil)).Elem()
+
+	// acceptRequestHandler is a RequestHandler that accepts all requests.
+	acceptRequestHandler requestHandler = func(proto.Message) error {
+		return nil
+	}
+
+	// passThroughResponseHandler is a responseHandler that does not modify
+	// the response and just passes it through.
+	passThroughResponseHandler responseHandler = func(
+		proto.Message) (proto.Message, error) {
+
+		return nil, nil
+	}
+)
+
+// RequestInterceptor is a type that can intercept an RPC request.
+type RequestInterceptor interface {
+	// Name returns the name of the interceptor.
+	Name() string
+
+	// ReadOnly returns true if this interceptor should be registered in
+	// read-only mode. In read-only mode no custom caveat name can be
+	// specified.
+	ReadOnly() bool
+
+	// CustomCaveatName returns the name of the custom caveat that is
+	// expected to be handled by this interceptor. Cannot be specified in
+	// read-only mode.
+	CustomCaveatName() string
+
+	// Intercept processes an RPC middleware interception request and
+	// returns the interception result which either accepts or rejects the
+	// intercepted message.
+	Intercept(context.Context,
+		*lnrpc.RPCMiddlewareRequest) (*lnrpc.RPCMiddlewareResponse,
+		error)
+}
+
+// requestHandler is a function type for a generic gRPC request message handler
+// that can accept (=return nil) or refuse (=return non-nil error with rejection
+// reason) an incoming request.
+type requestHandler func(req proto.Message) error
+
+// responseHandler is a function type for a generic gRPC response message
+// handler that can pass through the response (=return nil, nil), replace the
+// response with a new message of the same type (=return non-nil message, nil
+// error) or abort the call by returning a non-nil error.
+type responseHandler func(req proto.Message) (proto.Message, error)
+
+// RoundTripChecker is a type that represents a basic request/response round
+// trip checker.
+type RoundTripChecker interface {
+	// AcceptsRequest returns true if the checker accepts protobuf request
+	// messages of the given type. This is mainly a safety feature to make
+	// sure a round trip checker is implemented correctly.
+	AcceptsRequest(protoreflect.MessageType) bool
+
+	// HandlesResponse returns true if the checker can handle protobuf
+	// response messages of the given type. This is mainly a safety feature
+	// to make sure a round trip checker is implemented correctly.
+	HandlesResponse(protoreflect.MessageType) bool
+
+	// HandleRequest is called for each incoming gRPC request message of the
+	// type declared to be accepted by AcceptsRequest. The handler can
+	// accept (=return nil) or refuse (=return non-nil error with rejection
+	// reason) an incoming request.
+	HandleRequest(proto.Message) error
+
+	// HandleResponse is called for each outgoing gRPC response message of
+	// the type declared to be handled by HandlesResponse. The handler can
+	// pass through the response (=return nil, nil), replace the response
+	// with a new message of the same type (=return non-nil message, nil
+	// error) or abort the call by returning a non-nil error.
+	HandleResponse(proto.Message) (proto.Message, error)
+}
+
+// DefaultChecker is the default implementation of a round trip checker.
+type DefaultChecker struct {
+	requestType     protoreflect.MessageType
+	responseType    protoreflect.MessageType
+	requestHandler  requestHandler
+	responseHandler responseHandler
+}
+
+// AcceptsRequest returns true if the checker accepts protobuf request messages
+// of the given type. This is mainly a safety feature to make sure a round trip
+// checker is implemented correctly.
+func (r *DefaultChecker) AcceptsRequest(t protoreflect.MessageType) bool {
+	return t == r.requestType
+}
+
+// HandlesResponse returns true if the checker can handle protobuf response
+// messages of the given type. This is mainly a safety feature to make sure a
+// round trip checker is implemented correctly.
+func (r *DefaultChecker) HandlesResponse(t protoreflect.MessageType) bool {
+	return t == r.responseType
+}
+
+// HandleRequest is called for each incoming gRPC request message of the
+// type declared to be accepted by AcceptsRequest. The handler can
+// accept (=return nil) or refuse (=return non-nil error with rejection
+// reason) an incoming request.
+func (r *DefaultChecker) HandleRequest(req proto.Message) error {
+	return r.requestHandler(req)
+}
+
+// HandleResponse is called for each outgoing gRPC response message of
+// the type declared to be handled by HandlesResponse. The handler can
+// pass through the response (=return nil, nil), replace the response
+// with a new message of the same type (=return non-nil message, nil
+// error) or abort the call by returning a non-nil error.
+func (r *DefaultChecker) HandleResponse(resp proto.Message) (proto.Message,
+	error) {
+
+	return r.responseHandler(resp)
+}
+
+// NewPassThrough returns a round trip checker that allows the incoming request
+// and passes through the response unmodified.
+func NewPassThrough(requestSample proto.Message,
+	responseSample proto.Message) *DefaultChecker {
+
+	return &DefaultChecker{
+		requestType:     requestSample.ProtoReflect().Type(),
+		responseType:    responseSample.ProtoReflect().Type(),
+		requestHandler:  acceptRequestHandler,
+		responseHandler: passThroughResponseHandler,
+	}
+}
+
+// NewRequestChecker returns a round trip checker that inspects the incoming
+// request and passes through the response unmodified.
+func NewRequestChecker(requestSample proto.Message,
+	responseSample proto.Message,
+	typedRequestHandler interface{}) *DefaultChecker {
+
+	return &DefaultChecker{
+		requestType:  requestSample.ProtoReflect().Type(),
+		responseType: responseSample.ProtoReflect().Type(),
+		requestHandler: newReflectionRequestHandler(
+			requestSample, typedRequestHandler,
+		),
+		responseHandler: passThroughResponseHandler,
+	}
+}
+
+// NewResponseRewriter returns a round trip checker that allows the incoming
+// request and inspects and potentially modifies the response.
+func NewResponseRewriter(requestSample proto.Message,
+	responseSample proto.Message,
+	typedResponseHandler interface{}) *DefaultChecker {
+
+	return &DefaultChecker{
+		requestType:    requestSample.ProtoReflect().Type(),
+		responseType:   responseSample.ProtoReflect().Type(),
+		requestHandler: acceptRequestHandler,
+		responseHandler: newReflectionResponseHandler(
+			responseSample, typedResponseHandler,
+		),
+	}
+}
+
+// NewFullChecker returns a round trip checker that both inspects the incoming
+// request and response and potentially modifies the response.
+func NewFullChecker(requestSample proto.Message,
+	responseSample proto.Message, typedRequestHandler interface{},
+	typedResponseHandler interface{}) *DefaultChecker {
+
+	return &DefaultChecker{
+		requestType:  requestSample.ProtoReflect().Type(),
+		responseType: responseSample.ProtoReflect().Type(),
+		requestHandler: newReflectionRequestHandler(
+			requestSample, typedRequestHandler,
+		),
+		responseHandler: newReflectionResponseHandler(
+			responseSample, typedResponseHandler,
+		),
+	}
+}
+
+// newReflectionRequestHandler returns a request handler that adapts the generic
+// proto.Message capable request handler into one that is type specific for the
+// given request message sample message. This requires reflection and cannot be
+// implemented with Generics.
+func newReflectionRequestHandler(requestSample proto.Message,
+	typedHandler interface{}) requestHandler {
+
+	requestType := reflect.TypeOf(requestSample)
+	requestProtoType := requestSample.ProtoReflect().Type()
+	handlerValue := reflect.ValueOf(typedHandler)
+
+	err := validateRequestHandler(handlerValue.Type(), requestType)
+	if err != nil {
+		// This is covered by unit tests and shouldn't happen in the
+		// first place, as this would be an implementation error.
+		panic(err)
+	}
+
+	return func(req proto.Message) error {
+		if req.ProtoReflect().Type() != requestProtoType {
+			return fmt.Errorf("request handler called for "+
+				"unsupported type %v (expected %v)",
+				req.ProtoReflect().Type(), requestProtoType)
+		}
+
+		// We made sure this call would succeed when creating the
+		// handler.
+		resp := handlerValue.Call([]reflect.Value{
+			reflect.ValueOf(req),
+		})
+
+		// We also made sure the types returned from the function would
+		// be compatible with what we expect.
+		var err error
+		if resp[0].Interface() != nil {
+			err = resp[0].Interface().(error)
+		}
+
+		return err
+	}
+}
+
+// newReflectionResponseHandler returns a response handler that adapts the
+// generic proto.Message capable response handler into one that is type specific
+// for the given request message sample message. This requires reflection and
+// cannot be implemented with Generics.
+func newReflectionResponseHandler(responseSample proto.Message,
+	typedHandler interface{}) responseHandler {
+
+	responseType := reflect.TypeOf(responseSample)
+	responseProtoType := responseSample.ProtoReflect().Type()
+	handlerValue := reflect.ValueOf(typedHandler)
+
+	err := validateResponseHandler(handlerValue.Type(), responseType)
+	if err != nil {
+		// This is covered by unit tests and shouldn't happen in the
+		// first place, as this would be an implementation error.
+		panic(err)
+	}
+
+	return func(req proto.Message) (proto.Message, error) {
+		if req.ProtoReflect().Type() != responseProtoType {
+			return nil, fmt.Errorf("response handler called for "+
+				"unsupported type %v (expected %v)",
+				req.ProtoReflect().Type(), responseProtoType)
+		}
+
+		// We made sure this call would succeed when creating the
+		// handler.
+		resp := handlerValue.Call([]reflect.Value{
+			reflect.ValueOf(req),
+		})
+
+		// We also made sure the types returned from the function would
+		// be compatible with what we expect.
+		var (
+			replacementMessage proto.Message
+			err                error
+		)
+		if resp[0].Interface() != nil {
+			replacementMessage = resp[0].Interface().(proto.Message)
+		}
+		if resp[1].Interface() != nil {
+			err = resp[1].Interface().(error)
+		}
+
+		return replacementMessage, err
+	}
+}
+
+// validateRequestHandler makes sure that the given request handler function
+// has the correct number and types of parameters and return values.
+func validateRequestHandler(typedHandlerType reflect.Type,
+	requestType reflect.Type) error {
+
+	if typedHandlerType.Kind() != reflect.Func {
+		return fmt.Errorf("request handler must be a function")
+	}
+	if typedHandlerType.NumIn() != 1 || typedHandlerType.NumOut() != 1 {
+		return fmt.Errorf("request handler must have exactly one " +
+			"parameter and one return value")
+	}
+	if !typedHandlerType.In(0).ConvertibleTo(requestType) {
+		return fmt.Errorf("request handler must have one parameter " +
+			"with a sub type of proto.Message")
+	}
+	if typedHandlerType.Out(0) != errorType {
+		return fmt.Errorf("request handler must return exactly one " +
+			"error value")
+	}
+
+	return nil
+}
+
+// validateResponseHandler makes sure that the given response handler function
+// has the correct number and types of parameters and return values.
+func validateResponseHandler(typedHandlerType reflect.Type,
+	responseType reflect.Type) error {
+
+	if typedHandlerType.Kind() != reflect.Func {
+		return fmt.Errorf("response handler must be a function")
+	}
+	if typedHandlerType.NumIn() != 1 || typedHandlerType.NumOut() != 2 {
+		return fmt.Errorf("response handler must have exactly one " +
+			"parameter and two return values")
+	}
+	if !typedHandlerType.In(0).ConvertibleTo(responseType) {
+		return fmt.Errorf("response handler must have one parameter " +
+			"with a sub type of proto.Message")
+	}
+	outType0 := typedHandlerType.Out(0)
+	pmt := protoMessageType
+	if outType0 != pmt ||
+		typedHandlerType.Out(1) != errorType {
+
+		return fmt.Errorf("response handler must return exactly two " +
+			"values, proto.Message and error")
+	}
+
+	return nil
+}

rpcmiddleware/interface_test.go

@@ -0,0 +1,139 @@
+package rpcmiddleware
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+)
+
+var (
+	listPeersReq     = &lnrpc.ListPeersRequest{}
+	listPeersReqType = listPeersReq.ProtoReflect().Type()
+
+	listPeersResp     = &lnrpc.ListPeersResponse{}
+	listPeersRespType = listPeersResp.ProtoReflect().Type()
+)
+
+// TestMessageTypeOf tests that parsing the fully qualified protobuf message
+// type into its reflection type works correctly.
+func TestMessageTypeOf(t *testing.T) {
+	listPeersReqTypeParsed, err := MessageTypeOf("lnrpc.ListPeersRequest")
+	require.NoError(t, err)
+	require.Equal(t, listPeersReqType, listPeersReqTypeParsed)
+
+	listPeersRespTypeParsed, err := MessageTypeOf("lnrpc.ListPeersResponse")
+	require.NoError(t, err)
+	require.Equal(t, listPeersRespType, listPeersRespTypeParsed)
+}
+
+// TestPassThrough tests that the pass through round trip checker behaves as
+// expected.
+func TestPassThrough(t *testing.T) {
+	peersChecker := NewPassThrough(listPeersReq, listPeersResp)
+
+	require.True(t, peersChecker.AcceptsRequest(listPeersReqType))
+	require.True(t, peersChecker.HandlesResponse(listPeersRespType))
+	require.Nil(t, peersChecker.HandleRequest(listPeersReq))
+
+	resp, err := peersChecker.HandleResponse(listPeersResp)
+	require.NoError(t, err)
+	require.Nil(t, resp)
+
+	require.False(t, peersChecker.AcceptsRequest(listPeersRespType))
+	require.False(t, peersChecker.HandlesResponse(listPeersReqType))
+}
+
+// TestResponseRewriter tests that the response rewriter round trip checker
+// behaves as expected.
+func TestResponseRewriter(t *testing.T) {
+	peersChecker := NewResponseRewriter(
+		listPeersReq, listPeersResp,
+		func(peer *lnrpc.ListPeersResponse) (proto.Message, error) {
+			return peer, nil
+		},
+	)
+
+	require.True(t, peersChecker.AcceptsRequest(listPeersReqType))
+	require.True(t, peersChecker.HandlesResponse(listPeersRespType))
+	require.Nil(t, peersChecker.HandleRequest(listPeersReq))
+
+	resp, err := peersChecker.HandleResponse(listPeersResp)
+	require.NoError(t, err)
+	require.Equal(t, listPeersResp, resp)
+
+	require.False(t, peersChecker.AcceptsRequest(listPeersRespType))
+	require.False(t, peersChecker.HandlesResponse(listPeersReqType))
+}
+
+// TestFullChecker tests that the full checker round trip checker behaves as
+// expected.
+func TestFullChecker(t *testing.T) {
+	myErr := fmt.Errorf("some error happened")
+	peersChecker := NewFullChecker(
+		listPeersReq, listPeersResp,
+		func(peer *lnrpc.ListPeersRequest) error {
+			return myErr
+		},
+		func(*lnrpc.ListPeersResponse) (proto.Message, error) {
+			return nil, myErr
+		},
+	)
+
+	require.True(t, peersChecker.AcceptsRequest(listPeersReqType))
+	require.True(t, peersChecker.HandlesResponse(listPeersRespType))
+	require.Equal(t, myErr, peersChecker.HandleRequest(listPeersReq))
+
+	resp, err := peersChecker.HandleResponse(listPeersResp)
+	require.Error(t, err)
+	require.Equal(t, myErr, err)
+	require.Nil(t, resp)
+
+	require.False(t, peersChecker.AcceptsRequest(listPeersRespType))
+	require.False(t, peersChecker.HandlesResponse(listPeersReqType))
+}
+
+// TestImplementationPanics makes sure implementation errors are caught with
+// panics.
+func TestImplementationPanics(t *testing.T) {
+	require.PanicsWithError(
+		t, "request handler must be a function", func() {
+			_ = NewRequestChecker(
+				listPeersReq, listPeersResp, "wrong",
+			)
+		},
+	)
+	require.PanicsWithError(
+		t, "request handler must have exactly one parameter and one "+
+			"return value",
+		func() {
+			_ = NewRequestChecker(
+				listPeersReq, listPeersResp,
+				func() error {
+					return nil
+				},
+			)
+		},
+	)
+	require.PanicsWithError(
+		t, "response handler must be a function", func() {
+			_ = NewResponseRewriter(
+				listPeersReq, listPeersResp, "wrong",
+			)
+		},
+	)
+	require.PanicsWithError(
+		t, "response handler must have exactly one parameter and two "+
+			"return values",
+		func() {
+			_ = NewResponseRewriter(
+				listPeersReq, listPeersResp,
+				func() error {
+					return nil
+				},
+			)
+		},
+	)
+}

rpcmiddleware/log.go

@@ -0,0 +1,25 @@
+package rpcmiddleware
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+const Subsystem = "RMID"
+
+// 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(Subsystem, nil))
+}
+
+// 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
+}

rpcmiddleware/manager.go

@@ -0,0 +1,103 @@
+package rpcmiddleware
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/lightninglabs/lndclient"
+)
+
+// Manager is the main middleware manager service.
+type Manager struct {
+	interceptTimeout time.Duration
+	lndClient        lndclient.LightningClient
+	interceptors     []RequestInterceptor
+
+	errChan  chan error
+	wg       sync.WaitGroup
+	cancel   context.CancelFunc
+	quit     chan struct{}
+	stopOnce sync.Once
+}
+
+// NewManager returns a new middleware manager.
+func NewManager(interceptTimeout time.Duration,
+	lndClient lndclient.LightningClient,
+	interceptors ...RequestInterceptor) *Manager {
+
+	return &Manager{
+		interceptTimeout: interceptTimeout,
+		lndClient:        lndClient,
+		interceptors:     interceptors,
+		errChan:          make(chan error),
+		quit:             make(chan struct{}),
+	}
+}
+
+// Start starts the firewall by registering the interceptors with lnd.
+func (f *Manager) Start() error {
+	ctxc, cancel := context.WithCancel(context.Background())
+	f.cancel = cancel
+
+	for _, i := range f.interceptors {
+		errChan, err := f.lndClient.RegisterRPCMiddleware(
+			ctxc, i.Name(), i.CustomCaveatName(), i.ReadOnly(),
+			f.interceptTimeout, i.Intercept,
+		)
+		if err != nil {
+			cancel()
+			f.wg.Wait()
+
+			return err
+		}
+
+		f.wg.Add(1)
+		go func(i RequestInterceptor, errChan chan error) {
+			defer f.wg.Done()
+
+			for {
+				select {
+				case <-f.quit:
+					log.Debugf("Quitting interceptor %v, "+
+						"shutting down", i.Name())
+					return
+
+				case <-ctxc.Done():
+					log.Debugf("Quitting interceptor %v, "+
+						"context canceled", i.Name())
+
+					return
+
+				case err := <-errChan:
+					log.Errorf("Error in interceptor: %v",
+						err)
+
+					select {
+					case f.errChan <- err:
+					case <-f.quit:
+					case <-ctxc.Done():
+					}
+				}
+			}
+		}(i, errChan)
+	}
+
+	return nil
+}
+
+// Errors returns a channel on which any errors the manager encounters will be
+// delivered.
+func (f *Manager) Errors() chan error {
+	return f.errChan
+}
+
+// Stop shuts down the middleware manager.
+func (f *Manager) Stop() {
+	f.stopOnce.Do(func() {
+		close(f.quit)
+		f.cancel()
+
+		f.wg.Wait()
+	})
+}

rpcmiddleware/proto.go

@@ -0,0 +1,92 @@
+package rpcmiddleware
+
+import (
+	"fmt"
+
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
+)
+
+// MessageTypeOf return the protobuf reflection type of the given fully
+// qualified message type name.
+func MessageTypeOf(typeName string) (protoreflect.MessageType, error) {
+	return protoregistry.GlobalTypes.FindMessageByName(
+		protoreflect.FullName(typeName),
+	)
+}
+
+// ParseProtobuf parses a proto serialized message of the given type into its
+// native version.
+func ParseProtobuf(typeName string, serialized []byte) (proto.Message, error) {
+	messageType, err := MessageTypeOf(typeName)
+	if err != nil {
+		return nil, err
+	}
+	msg := messageType.New()
+	err = proto.Unmarshal(serialized, msg.Interface())
+	if err != nil {
+		return nil, err
+	}
+
+	return msg.Interface(), nil
+}
+
+func RPCOk(req *lnrpc.RPCMiddlewareRequest) (*lnrpc.RPCMiddlewareResponse,
+	error) {
+
+	return RPCErrString(req, "")
+}
+
+func RPCErr(req *lnrpc.RPCMiddlewareRequest,
+	err error) (*lnrpc.RPCMiddlewareResponse, error) {
+
+	if err != nil {
+		return RPCErrString(req, err.Error())
+	}
+
+	return RPCErrString(req, "")
+}
+
+func RPCErrString(req *lnrpc.RPCMiddlewareRequest, format string,
+	args ...interface{}) (*lnrpc.RPCMiddlewareResponse, error) {
+
+	feedback := &lnrpc.InterceptFeedback{}
+	resp := &lnrpc.RPCMiddlewareResponse{
+		RefMsgId: req.MsgId,
+		MiddlewareMessage: &lnrpc.RPCMiddlewareResponse_Feedback{
+			Feedback: feedback,
+		},
+	}
+
+	if format != "" {
+		feedback.Error = fmt.Sprintf(format, args...)
+	}
+
+	return resp, nil
+}
+
+func RPCReplacement(req *lnrpc.RPCMiddlewareRequest,
+	replacementResponse proto.Message) (*lnrpc.RPCMiddlewareResponse,
+	error) {
+
+	rawResponse, err := proto.Marshal(replacementResponse)
+	if err != nil {
+		return RPCErr(
+			req, fmt.Errorf("cannot marshal proto msg: %v", err),
+		)
+	}
+
+	feedback := &lnrpc.InterceptFeedback{
+		ReplaceResponse:       true,
+		ReplacementSerialized: rawResponse,
+	}
+
+	return &lnrpc.RPCMiddlewareResponse{
+		RefMsgId: req.MsgId,
+		MiddlewareMessage: &lnrpc.RPCMiddlewareResponse_Feedback{
+			Feedback: feedback,
+		},
+	}, nil
+}