17132206455

Committed 21 Aug 2025 03:56PM UTC coverage: 54.685% (-2.6%) from 57.321%

Build # 17132206455

Build Type

Pull #10167

github

Committed by

web-flow

Commit Message

Merge 5dd2ed093 into 0c2f045f5

Pull Request Pull Request #10167: multi: bump Go to 1.24.6

Run Details

4 of 31 new or added lines in 10 files covered. (12.9%)

23854 existing lines in 284 files now uncovered.

108937 of 199210 relevant lines covered (54.68%)

22026.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

5.38

/rpcperms/middleware_handler.go

package rpcperms

import (
        "context"
        "encoding/hex"
        "errors"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/btcsuite/btcd/chaincfg"
        "github.com/lightningnetwork/lnd/lnrpc"
        "github.com/lightningnetwork/lnd/macaroons"
        "google.golang.org/grpc/metadata"
        "google.golang.org/protobuf/proto"
        "google.golang.org/protobuf/reflect/protoreflect"
        "google.golang.org/protobuf/reflect/protoregistry"
        "gopkg.in/macaroon.v2"
)

var (
        // ErrShuttingDown is the error that's returned when the server is
        // shutting down and a request cannot be served anymore.
        ErrShuttingDown = errors.New("server shutting down")

        // ErrTimeoutReached is the error that's returned if any of the
        // middleware's tasks is not completed in the given time.
        ErrTimeoutReached = errors.New("intercept timeout reached")

        // errClientQuit is the error that's returned if the client closes the
        // middleware communication stream before a request was fully handled.
        errClientQuit = errors.New("interceptor RPC client quit")
)

// MiddlewareHandler is a type that communicates with a middleware over the
// established bi-directional RPC stream. It sends messages to the middleware
// whenever the custom business logic implemented there should give feedback to
// a request or response that's happening on the main gRPC server.
type MiddlewareHandler struct {
        // lastMsgID is the ID of the last intercept message that was forwarded
        // to the middleware.
        //
        // NOTE: Must be used atomically!
        lastMsgID uint64

        middlewareName string

        readOnly bool

        customCaveatName string

        receive func() (*lnrpc.RPCMiddlewareResponse, error)

        send func(request *lnrpc.RPCMiddlewareRequest) error

        interceptRequests chan *interceptRequest

        timeout time.Duration

        // params are our current chain params.
        params *chaincfg.Params

        // done is closed when the rpc client terminates.
        done chan struct{}

        // quit is closed when lnd is shutting down.
        quit chan struct{}

        wg sync.WaitGroup
}

// NewMiddlewareHandler creates a new handler for the middleware with the given
// name and custom caveat name.
func NewMiddlewareHandler(name, customCaveatName string, readOnly bool,
        receive func() (*lnrpc.RPCMiddlewareResponse, error),
        send func(request *lnrpc.RPCMiddlewareRequest) error,
        timeout time.Duration, params *chaincfg.Params,
        quit chan struct{}) *MiddlewareHandler {

        // We explicitly want to log this as a warning since intercepting any
        // gRPC messages can also be used for malicious purposes and the user
        // should be made aware of the risks.
        log.Warnf("A new gRPC middleware with the name '%s' was registered "+
                " with custom_macaroon_caveat='%s', read_only=%v. Make sure "+
                "you trust the middleware author since that code will be able "+
                "to intercept and possibly modify any gRPC messages sent/"+
                "received to/from a client that has a macaroon with that "+
                "custom caveat.", name, customCaveatName, readOnly)

        return &MiddlewareHandler{
                middlewareName:    name,
                customCaveatName:  customCaveatName,
                readOnly:          readOnly,
                receive:           receive,
                send:              send,
                interceptRequests: make(chan *interceptRequest),
                timeout:           timeout,
                params:            params,
                done:              make(chan struct{}),
                quit:              quit,
        }
}

// intercept handles the full interception lifecycle of a single middleware
// event (stream authentication, request interception or response interception).
// The lifecycle consists of sending a message to the middleware, receiving a
// feedback on it and sending the feedback to the appropriate channel. All steps
// are guarded by the configured timeout to make sure a middleware cannot slow
// down requests too much.
func (h *MiddlewareHandler) intercept(requestID uint64,
        req *InterceptionRequest) (*interceptResponse, error) {

        respChan := make(chan *interceptResponse, 1)

        newRequest := &interceptRequest{
                requestID: requestID,
                request:   req,
                response:  respChan,
        }

        // timeout is the time after which intercept requests expire.
        timeout := time.After(h.timeout)

        // Send the request to the interceptRequests channel for the main
        // goroutine to be picked up.
        select {
        case h.interceptRequests <- newRequest:

        case <-timeout:
                log.Errorf("MiddlewareHandler returned error - reached "+
                        "timeout of %v for request interception", h.timeout)

                return nil, ErrTimeoutReached

        case <-h.done:
                return nil, errClientQuit

        case <-h.quit:
                return nil, ErrShuttingDown
        }

        // Receive the response and return it. If no response has been received
        // in AcceptorTimeout, then return false.
        select {
        case resp := <-respChan:
                return resp, nil

        case <-timeout:
                log.Errorf("MiddlewareHandler returned error - reached "+
                        "timeout of %v for response interception", h.timeout)
                return nil, ErrTimeoutReached

        case <-h.done:
                return nil, errClientQuit

        case <-h.quit:
                return nil, ErrShuttingDown
        }
}

// Run is the main loop for the middleware handler. This function will block
// until it receives the signal that lnd is shutting down, or the rpc stream is
// cancelled by the client.
func (h *MiddlewareHandler) Run() error {
        // Wait for our goroutines to exit before we return.
        defer h.wg.Wait()
        defer log.Debugf("Exiting middleware run loop for %s", h.middlewareName)

        // Create a channel that responses from middlewares are sent into.
        responses := make(chan *lnrpc.RPCMiddlewareResponse)

        // errChan is used by the receive loop to signal any errors that occur
        // during reading from the stream. This is primarily used to shutdown
        // the send loop in the case of an RPC client disconnecting.
        errChan := make(chan error, 1)

        // Start a goroutine to receive responses from the interceptor. We
        // expect the receive function to block, so it must be run in a
        // goroutine (otherwise we could not send more than one intercept
        // request to the client).
        h.wg.Add(1)
        go func() {
                defer h.wg.Done()

                h.receiveResponses(errChan, responses)
        }()

        return h.sendInterceptRequests(errChan, responses)
}

// receiveResponses receives responses for our intercept requests and dispatches
// them into the responses channel provided, sending any errors that occur into
// the error channel provided.
func (h *MiddlewareHandler) receiveResponses(errChan chan error,
        responses chan *lnrpc.RPCMiddlewareResponse) {

        for {
                resp, err := h.receive()
                if err != nil {
                        errChan <- err
                        return
                }

                select {
                case responses <- resp:

                case <-h.done:
                        return

                case <-h.quit:
                        return
                }
        }
}

// sendInterceptRequests handles intercept requests sent to us by our Accept()
// function, dispatching them to our acceptor stream and coordinating return of
// responses to their callers.
func (h *MiddlewareHandler) sendInterceptRequests(errChan chan error,
        responses chan *lnrpc.RPCMiddlewareResponse) error {

        // Close the done channel to indicate that the interceptor is no longer
        // listening and any in-progress requests should be terminated.
        defer close(h.done)

        interceptRequests := make(map[uint64]*interceptRequest)

        for {
                select {
                // Consume requests passed to us from our Accept() function and
                // send them into our stream.
                case newRequest := <-h.interceptRequests:
                        msgID := atomic.AddUint64(&h.lastMsgID, 1)

                        req := newRequest.request
                        interceptRequests[msgID] = newRequest

                        interceptReq, err := req.ToRPC(
                                newRequest.requestID, msgID,
                        )
                        if err != nil {
                                return err
                        }

                        if err := h.send(interceptReq); err != nil {
                                return err
                        }

                // Process newly received responses from our interceptor,
                // looking the original request up in our map of requests and
                // dispatching the response.
                case resp := <-responses:
                        requestInfo, ok := interceptRequests[resp.RefMsgId]
                        if !ok {
                                continue
                        }

                        response := &interceptResponse{}
                        switch msg := resp.GetMiddlewareMessage().(type) {
                        case *lnrpc.RPCMiddlewareResponse_Feedback:
                                t := msg.Feedback
                                if t.Error != "" {
                                        response.err = fmt.Errorf("%s", t.Error)
                                        break
                                }

                                // If there's nothing to replace, we're done,
                                // this request was just accepted.
                                if !t.ReplaceResponse {
                                        break
                                }

                                // We are replacing the response, the question
                                // now just is: was it an error or a proper
                                // proto message?
                                response.replace = true
                                if requestInfo.request.IsError {
                                        response.replacement = errors.New(
                                                string(t.ReplacementSerialized),
                                        )

                                        break
                                }

                                // Not an error but a proper proto message that
                                // needs to be replaced. For that we need to
                                // parse it from the raw bytes into the full RPC
                                // message.
                                protoMsg, err := parseProto(
                                        requestInfo.request.ProtoTypeName,
                                        t.ReplacementSerialized,
                                )

                                if err != nil {
                                        response.err = err

                                        break
                                }

                                response.replacement = protoMsg

                        default:
                                return fmt.Errorf("unknown middleware "+
                                        "message: %v", msg)
                        }

                        select {
                        case requestInfo.response <- response:
                        case <-h.quit:
                        }

                        delete(interceptRequests, resp.RefMsgId)

                // If we failed to receive from our middleware, we exit.
                case err := <-errChan:
                        log.Errorf("Received an error: %v, shutting down", err)
                        return err

                // Exit if we are shutting down.
                case <-h.quit:
                        return ErrShuttingDown
                }
        }
}

// InterceptType defines the different types of intercept messages a middleware
// can receive.
type InterceptType uint8

const (
        // TypeStreamAuth is the type of intercept message that is sent when a
        // client or streaming RPC is initialized. A message with this type will
        // be sent out during stream initialization so a middleware can
        // accept/deny the whole stream instead of only single messages on the
        // stream.
        TypeStreamAuth InterceptType = 1

        // TypeRequest is the type of intercept message that is sent when an RPC
        // request message is sent to lnd. For client-streaming RPCs a new
        // message of this type is sent for each individual RPC request sent to
        // the stream. Middleware has the option to modify a request message
        // before it is delivered to lnd.
        TypeRequest InterceptType = 2

        // TypeResponse is the type of intercept message that is sent when an
        // RPC response message is sent from lnd to a client. For
        // server-streaming RPCs a new message of this type is sent for each
        // individual RPC response sent to the stream. Middleware has the option
        // to modify a response message before it is sent out to the client.
        TypeResponse InterceptType = 3
)

// InterceptionRequest is a struct holding all information that is sent to a
// middleware whenever there is something to intercept (auth, request,
// response).
type InterceptionRequest struct {
        // Type is the type of the interception message.
        Type InterceptType

        // StreamRPC is set to true if the invoked RPC method is client or
        // server streaming.
        StreamRPC bool

        // Macaroon holds the macaroon that the client sent to lnd.
        Macaroon *macaroon.Macaroon

        // RawMacaroon holds the raw binary serialized macaroon that the client
        // sent to lnd.
        RawMacaroon []byte

        // CustomCaveatName is the name of the custom caveat that the middleware
        // was intercepting for.
        CustomCaveatName string

        // CustomCaveatCondition is the condition of the custom caveat that the
        // middleware was intercepting for. This can be empty for custom caveats
        // that only have a name (marker caveats).
        CustomCaveatCondition string

        // FullURI is the full RPC method URI that was invoked.
        FullURI string

        // ProtoSerialized is the full request or response object in the
        // protobuf binary serialization format.
        ProtoSerialized []byte

        // ProtoTypeName is the fully qualified name of the protobuf type of the
        // request or response message that is serialized in the field above.
        ProtoTypeName string

        // IsError indicates that the message contained within this request is
        // an error. Will only ever be true for response messages.
        IsError bool

        // CtxMetadataPairs contains the metadata pairs that were sent along
        // with the RPC request via the context.
        CtxMetadataPairs metadata.MD
}

// NewMessageInterceptionRequest creates a new interception request for either
// a request or response message.
func NewMessageInterceptionRequest(ctx context.Context,
        authType InterceptType, isStream bool, fullMethod string,
        m interface{}) (*InterceptionRequest, error) {

        mac, rawMacaroon, err := macaroonFromContext(ctx)
        if err != nil {
                return nil, err
        }

        md, _ := metadata.FromIncomingContext(ctx)

        req := &InterceptionRequest{
                Type:             authType,
                StreamRPC:        isStream,
                Macaroon:         mac,
                RawMacaroon:      rawMacaroon,
                FullURI:          fullMethod,
                CtxMetadataPairs: md,
        }

        // The message is either a proto message or an error, we don't support
        // any other types being intercepted.
        switch t := m.(type) {
        case proto.Message:
                req.ProtoSerialized, err = proto.Marshal(t)
                if err != nil {
                        return nil, fmt.Errorf("cannot marshal proto msg: %w",
                                err)
                }
                req.ProtoTypeName = string(proto.MessageName(t))

        case error:
                req.ProtoSerialized = []byte(t.Error())
                req.ProtoTypeName = "error"
                req.IsError = true

        default:
                return nil, fmt.Errorf("unsupported type for interception "+
                        "request: %v", m)
        }

        return req, nil
}

// NewStreamAuthInterceptionRequest creates a new interception request for a
// stream authentication message.
func NewStreamAuthInterceptionRequest(ctx context.Context,
        fullMethod string) (*InterceptionRequest, error) {

        mac, rawMacaroon, err := macaroonFromContext(ctx)
        if err != nil {
                return nil, err
        }

        return &InterceptionRequest{
                Type:        TypeStreamAuth,
                StreamRPC:   true,
                Macaroon:    mac,
                RawMacaroon: rawMacaroon,
                FullURI:     fullMethod,
        }, nil
}

// macaroonFromContext tries to extract the macaroon from the incoming context.
// If there is no macaroon, a nil error is returned since some RPCs might not
// require a macaroon. But in case there is something in the macaroon header
// field that cannot be parsed, a non-nil error is returned.
func macaroonFromContext(ctx context.Context) (*macaroon.Macaroon, []byte,
        error) {

        macHex, err := macaroons.RawMacaroonFromContext(ctx)
        if err != nil {
                // If there is no macaroon, we continue anyway as it might be an
                // RPC that doesn't require a macaroon.
                return nil, nil, nil
        }

        macBytes, err := hex.DecodeString(macHex)
        if err != nil {
                return nil, nil, err
        }

        mac := &macaroon.Macaroon{}
        if err := mac.UnmarshalBinary(macBytes); err != nil {
                return nil, nil, err
        }

        return mac, macBytes, nil
}

// ToRPC converts the interception request to its RPC counterpart.
func (r *InterceptionRequest) ToRPC(requestID,
        msgID uint64) (*lnrpc.RPCMiddlewareRequest, error) {

        mdPairs := make(
                map[string]*lnrpc.MetadataValues, len(r.CtxMetadataPairs),
        )
        for key, values := range r.CtxMetadataPairs {
                mdPairs[key] = &lnrpc.MetadataValues{
                        Values: values,
                }
        }

        rpcRequest := &lnrpc.RPCMiddlewareRequest{
                RequestId:             requestID,
                MsgId:                 msgID,
                RawMacaroon:           r.RawMacaroon,
                CustomCaveatCondition: r.CustomCaveatCondition,
                MetadataPairs:         mdPairs,
        }

        switch r.Type {
        case TypeStreamAuth:
                rpcRequest.InterceptType = &lnrpc.RPCMiddlewareRequest_StreamAuth{
                        StreamAuth: &lnrpc.StreamAuth{
                                MethodFullUri: r.FullURI,
                        },
                }

        case TypeRequest:
                rpcRequest.InterceptType = &lnrpc.RPCMiddlewareRequest_Request{
                        Request: &lnrpc.RPCMessage{
                                MethodFullUri: r.FullURI,
                                StreamRpc:     r.StreamRPC,
                                TypeName:      r.ProtoTypeName,
                                Serialized:    r.ProtoSerialized,
                        },
                }

        case TypeResponse:
                rpcRequest.InterceptType = &lnrpc.RPCMiddlewareRequest_Response{
                        Response: &lnrpc.RPCMessage{
                                MethodFullUri: r.FullURI,
                                StreamRpc:     r.StreamRPC,
                                TypeName:      r.ProtoTypeName,
                                Serialized:    r.ProtoSerialized,
                                IsError:       r.IsError,
                        },
                }

        default:
                return nil, fmt.Errorf("unknown intercept type %v", r.Type)
        }

        return rpcRequest, nil
}

// interceptRequest is a struct that keeps track of an interception request sent
// out to a middleware and the response that is eventually sent back by the
// middleware.
type interceptRequest struct {
        requestID uint64
        request   *InterceptionRequest
        response  chan *interceptResponse
}

// interceptResponse is the response a middleware sends back for each
// intercepted message.
type interceptResponse struct {
        err         error
        replace     bool
        replacement interface{}
}

// parseProto parses a proto serialized message of the given type into its
// native version.
func parseProto(typeName string, serialized []byte) (proto.Message, error) {
        messageType, err := protoregistry.GlobalTypes.FindMessageByName(
                protoreflect.FullName(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
}

// replaceProtoMsg replaces the given target message with the content of the
// replacement message.
func replaceProtoMsg(target interface{}, replacement interface{}) error {
        targetMsg, ok := target.(proto.Message)
        if !ok {
                return fmt.Errorf("target is not a proto message: %v", target)
        }

        replacementMsg, ok := replacement.(proto.Message)
        if !ok {
                return fmt.Errorf("replacement is not a proto message: %v",
                        replacement)
        }

        if targetMsg.ProtoReflect().Type() !=
                replacementMsg.ProtoReflect().Type() {

                return fmt.Errorf("replacement message is of wrong type")
        }

        replacementBytes, err := proto.Marshal(replacementMsg)
        if err != nil {
                return fmt.Errorf("error marshaling replacement: %w", err)
        }
        err = proto.Unmarshal(replacementBytes, targetMsg)
        if err != nil {
                return fmt.Errorf("error unmarshaling replacement: %w", err)
        }

        return nil
}

1	package rpcperms
2
3	import (
4	"context"
5	"encoding/hex"
6	"errors"
7	"fmt"
8	"sync"
9	"sync/atomic"
10	"time"
11
12	"github.com/btcsuite/btcd/chaincfg"
13	"github.com/lightningnetwork/lnd/lnrpc"
14	"github.com/lightningnetwork/lnd/macaroons"
15	"google.golang.org/grpc/metadata"
16	"google.golang.org/protobuf/proto"
17	"google.golang.org/protobuf/reflect/protoreflect"
18	"google.golang.org/protobuf/reflect/protoregistry"
19	"gopkg.in/macaroon.v2"
20	)
21
22	var (
23	// ErrShuttingDown is the error that's returned when the server is
24	// shutting down and a request cannot be served anymore.
25	ErrShuttingDown = errors.New("server shutting down")
26
27	// ErrTimeoutReached is the error that's returned if any of the
28	// middleware's tasks is not completed in the given time.
29	ErrTimeoutReached = errors.New("intercept timeout reached")
30
31	// errClientQuit is the error that's returned if the client closes the
32	// middleware communication stream before a request was fully handled.
33	errClientQuit = errors.New("interceptor RPC client quit")
34	)
35
36	// MiddlewareHandler is a type that communicates with a middleware over the
37	// established bi-directional RPC stream. It sends messages to the middleware
38	// whenever the custom business logic implemented there should give feedback to
39	// a request or response that's happening on the main gRPC server.
40	type MiddlewareHandler struct {
41	// lastMsgID is the ID of the last intercept message that was forwarded
42	// to the middleware.
43	//
44	// NOTE: Must be used atomically!
45	lastMsgID uint64
46
47	middlewareName string
48
49	readOnly bool
50
51	customCaveatName string
52
53	receive func() (*lnrpc.RPCMiddlewareResponse, error)
54
55	send func(request *lnrpc.RPCMiddlewareRequest) error
56
57	interceptRequests chan *interceptRequest
58
59	timeout time.Duration
60
61	// params are our current chain params.
62	params *chaincfg.Params
63
64	// done is closed when the rpc client terminates.
65	done chan struct{}
66
67	// quit is closed when lnd is shutting down.
68	quit chan struct{}
69
70	wg sync.WaitGroup
71	}
72
73	// NewMiddlewareHandler creates a new handler for the middleware with the given
74	// name and custom caveat name.
75	func NewMiddlewareHandler(name, customCaveatName string, readOnly bool,
76	receive func() (*lnrpc.RPCMiddlewareResponse, error),
77	send func(request *lnrpc.RPCMiddlewareRequest) error,
78	timeout time.Duration, params *chaincfg.Params,
UNCOV 79	quit chan struct{}) *MiddlewareHandler {	×
UNCOV 80		×
UNCOV 81	// We explicitly want to log this as a warning since intercepting any	×
UNCOV 82	// gRPC messages can also be used for malicious purposes and the user	×
UNCOV 83	// should be made aware of the risks.	×
UNCOV 84	log.Warnf("A new gRPC middleware with the name '%s' was registered "+	×
UNCOV 85	" with custom_macaroon_caveat='%s', read_only=%v. Make sure "+	×
UNCOV 86	"you trust the middleware author since that code will be able "+	×
UNCOV 87	"to intercept and possibly modify any gRPC messages sent/"+	×
UNCOV 88	"received to/from a client that has a macaroon with that "+	×
UNCOV 89	"custom caveat.", name, customCaveatName, readOnly)	×
UNCOV 90		×
UNCOV 91	return &MiddlewareHandler{	×
UNCOV 92	middlewareName: name,	×
UNCOV 93	customCaveatName: customCaveatName,	×
UNCOV 94	readOnly: readOnly,	×
UNCOV 95	receive: receive,	×
UNCOV 96	send: send,	×
UNCOV 97	interceptRequests: make(chan *interceptRequest),	×
UNCOV 98	timeout: timeout,	×
UNCOV 99	params: params,	×
UNCOV 100	done: make(chan struct{}),	×
UNCOV 101	quit: quit,	×
UNCOV 102	}	×
UNCOV 103	}	×
104
105	// intercept handles the full interception lifecycle of a single middleware
106	// event (stream authentication, request interception or response interception).
107	// The lifecycle consists of sending a message to the middleware, receiving a
108	// feedback on it and sending the feedback to the appropriate channel. All steps
109	// are guarded by the configured timeout to make sure a middleware cannot slow
110	// down requests too much.
111	func (h *MiddlewareHandler) intercept(requestID uint64,
UNCOV 112	req InterceptionRequest) (interceptResponse, error) {	×
UNCOV 113		×
UNCOV 114	respChan := make(chan *interceptResponse, 1)	×
UNCOV 115		×
UNCOV 116	newRequest := &interceptRequest{	×
UNCOV 117	requestID: requestID,	×
UNCOV 118	request: req,	×
UNCOV 119	response: respChan,	×
UNCOV 120	}	×
UNCOV 121		×
UNCOV 122	// timeout is the time after which intercept requests expire.	×
UNCOV 123	timeout := time.After(h.timeout)	×
UNCOV 124		×
UNCOV 125	// Send the request to the interceptRequests channel for the main	×
UNCOV 126	// goroutine to be picked up.	×
UNCOV 127	select {	×
UNCOV 128	case h.interceptRequests <- newRequest:	×
129
130	case <-timeout:	×
131	log.Errorf("MiddlewareHandler returned error - reached "+	×
132	"timeout of %v for request interception", h.timeout)	×
133		×
134	return nil, ErrTimeoutReached	×
135
136	case <-h.done:	×
137	return nil, errClientQuit	×
138
139	case <-h.quit:	×
140	return nil, ErrShuttingDown	×
141	}
142
143	// Receive the response and return it. If no response has been received
144	// in AcceptorTimeout, then return false.
UNCOV 145	select {	×
UNCOV 146	case resp := <-respChan:	×
UNCOV 147	return resp, nil	×
148
149	case <-timeout:	×
150	log.Errorf("MiddlewareHandler returned error - reached "+	×
151	"timeout of %v for response interception", h.timeout)	×
152	return nil, ErrTimeoutReached	×
153
154	case <-h.done:	×
155	return nil, errClientQuit	×
156
157	case <-h.quit:	×
158	return nil, ErrShuttingDown	×
159	}
160	}
161
162	// Run is the main loop for the middleware handler. This function will block
163	// until it receives the signal that lnd is shutting down, or the rpc stream is
164	// cancelled by the client.
UNCOV 165	func (h *MiddlewareHandler) Run() error {	×
UNCOV 166	// Wait for our goroutines to exit before we return.	×
UNCOV 167	defer h.wg.Wait()	×
UNCOV 168	defer log.Debugf("Exiting middleware run loop for %s", h.middlewareName)	×
UNCOV 169		×
UNCOV 170	// Create a channel that responses from middlewares are sent into.	×
UNCOV 171	responses := make(chan *lnrpc.RPCMiddlewareResponse)	×
UNCOV 172		×
UNCOV 173	// errChan is used by the receive loop to signal any errors that occur	×
UNCOV 174	// during reading from the stream. This is primarily used to shutdown	×
UNCOV 175	// the send loop in the case of an RPC client disconnecting.	×
UNCOV 176	errChan := make(chan error, 1)	×
UNCOV 177		×
UNCOV 178	// Start a goroutine to receive responses from the interceptor. We	×
UNCOV 179	// expect the receive function to block, so it must be run in a	×
UNCOV 180	// goroutine (otherwise we could not send more than one intercept	×
UNCOV 181	// request to the client).	×
UNCOV 182	h.wg.Add(1)	×
UNCOV 183	go func() {	×
UNCOV 184	defer h.wg.Done()	×
UNCOV 185		×
UNCOV 186	h.receiveResponses(errChan, responses)	×
UNCOV 187	}()	×
188
UNCOV 189	return h.sendInterceptRequests(errChan, responses)	×
190	}
191
192	// receiveResponses receives responses for our intercept requests and dispatches
193	// them into the responses channel provided, sending any errors that occur into
194	// the error channel provided.
195	func (h *MiddlewareHandler) receiveResponses(errChan chan error,
UNCOV 196	responses chan *lnrpc.RPCMiddlewareResponse) {	×
UNCOV 197		×
UNCOV 198	for {	×
UNCOV 199	resp, err := h.receive()	×
UNCOV 200	if err != nil {	×
UNCOV 201	errChan <- err	×
UNCOV 202	return	×
UNCOV 203	}	×
204
UNCOV 205	select {	×
UNCOV 206	case responses <- resp:	×
207
208	case <-h.done:	×
209	return	×
210
211	case <-h.quit:	×
212	return	×
213	}
214	}
215	}
216
217	// sendInterceptRequests handles intercept requests sent to us by our Accept()
218	// function, dispatching them to our acceptor stream and coordinating return of
219	// responses to their callers.
220	func (h *MiddlewareHandler) sendInterceptRequests(errChan chan error,
UNCOV 221	responses chan *lnrpc.RPCMiddlewareResponse) error {	×
UNCOV 222		×
UNCOV 223	// Close the done channel to indicate that the interceptor is no longer	×
UNCOV 224	// listening and any in-progress requests should be terminated.	×
UNCOV 225	defer close(h.done)	×
UNCOV 226		×
UNCOV 227	interceptRequests := make(map[uint64]*interceptRequest)	×
UNCOV 228		×
UNCOV 229	for {	×
UNCOV 230	select {	×
231	// Consume requests passed to us from our Accept() function and
232	// send them into our stream.
UNCOV 233	case newRequest := <-h.interceptRequests:	×
UNCOV 234	msgID := atomic.AddUint64(&h.lastMsgID, 1)	×
UNCOV 235		×
UNCOV 236	req := newRequest.request	×
UNCOV 237	interceptRequests[msgID] = newRequest	×
UNCOV 238		×
UNCOV 239	interceptReq, err := req.ToRPC(	×
UNCOV 240	newRequest.requestID, msgID,	×
UNCOV 241	)	×
UNCOV 242	if err != nil {	×
243	return err	×
244	}	×
245
UNCOV 246	if err := h.send(interceptReq); err != nil {	×
247	return err	×
248	}	×
249
250	// Process newly received responses from our interceptor,
251	// looking the original request up in our map of requests and
252	// dispatching the response.
UNCOV 253	case resp := <-responses:	×
UNCOV 254	requestInfo, ok := interceptRequests[resp.RefMsgId]	×
UNCOV 255	if !ok {	×
256	continue	×
257	}
258
UNCOV 259	response := &interceptResponse{}	×
UNCOV 260	switch msg := resp.GetMiddlewareMessage().(type) {	×
UNCOV 261	case *lnrpc.RPCMiddlewareResponse_Feedback:	×
UNCOV 262	t := msg.Feedback	×
UNCOV 263	if t.Error != "" {	×
264	response.err = fmt.Errorf("%s", t.Error)	×
265	break	×
266	}
267
268	// If there's nothing to replace, we're done,
269	// this request was just accepted.
UNCOV 270	if !t.ReplaceResponse {	×
UNCOV 271	break	×
272	}
273
274	// We are replacing the response, the question
275	// now just is: was it an error or a proper
276	// proto message?
UNCOV 277	response.replace = true	×
UNCOV 278	if requestInfo.request.IsError {	×
UNCOV 279	response.replacement = errors.New(	×
UNCOV 280	string(t.ReplacementSerialized),	×
UNCOV 281	)	×
UNCOV 282		×
UNCOV 283	break	×
284	}
285
286	// Not an error but a proper proto message that
287	// needs to be replaced. For that we need to
288	// parse it from the raw bytes into the full RPC
289	// message.
UNCOV 290	protoMsg, err := parseProto(	×
UNCOV 291	requestInfo.request.ProtoTypeName,	×
UNCOV 292	t.ReplacementSerialized,	×
UNCOV 293	)	×
UNCOV 294		×
UNCOV 295	if err != nil {	×
296	response.err = err	×
297		×
298	break	×
299	}
300
UNCOV 301	response.replacement = protoMsg	×
302
303	default:	×
304	return fmt.Errorf("unknown middleware "+	×
305	"message: %v", msg)	×
306	}
307
UNCOV 308	select {	×
UNCOV 309	case requestInfo.response <- response:	×
310	case <-h.quit:	×
311	}
312
UNCOV 313	delete(interceptRequests, resp.RefMsgId)	×
314
315	// If we failed to receive from our middleware, we exit.
UNCOV 316	case err := <-errChan:	×
UNCOV 317	log.Errorf("Received an error: %v, shutting down", err)	×
UNCOV 318	return err	×
319
320	// Exit if we are shutting down.
321	case <-h.quit:	×
322	return ErrShuttingDown	×
323	}
324	}
325	}
326
327	// InterceptType defines the different types of intercept messages a middleware
328	// can receive.
329	type InterceptType uint8
330
331	const (
332	// TypeStreamAuth is the type of intercept message that is sent when a
333	// client or streaming RPC is initialized. A message with this type will
334	// be sent out during stream initialization so a middleware can
335	// accept/deny the whole stream instead of only single messages on the
336	// stream.
337	TypeStreamAuth InterceptType = 1
338
339	// TypeRequest is the type of intercept message that is sent when an RPC
340	// request message is sent to lnd. For client-streaming RPCs a new
341	// message of this type is sent for each individual RPC request sent to
342	// the stream. Middleware has the option to modify a request message
343	// before it is delivered to lnd.
344	TypeRequest InterceptType = 2
345
346	// TypeResponse is the type of intercept message that is sent when an
347	// RPC response message is sent from lnd to a client. For
348	// server-streaming RPCs a new message of this type is sent for each
349	// individual RPC response sent to the stream. Middleware has the option
350	// to modify a response message before it is sent out to the client.
351	TypeResponse InterceptType = 3
352	)
353
354	// InterceptionRequest is a struct holding all information that is sent to a
355	// middleware whenever there is something to intercept (auth, request,
356	// response).
357	type InterceptionRequest struct {
358	// Type is the type of the interception message.
359	Type InterceptType
360
361	// StreamRPC is set to true if the invoked RPC method is client or
362	// server streaming.
363	StreamRPC bool
364
365	// Macaroon holds the macaroon that the client sent to lnd.
366	Macaroon *macaroon.Macaroon
367
368	// RawMacaroon holds the raw binary serialized macaroon that the client
369	// sent to lnd.
370	RawMacaroon []byte
371
372	// CustomCaveatName is the name of the custom caveat that the middleware
373	// was intercepting for.
374	CustomCaveatName string
375
376	// CustomCaveatCondition is the condition of the custom caveat that the
377	// middleware was intercepting for. This can be empty for custom caveats
378	// that only have a name (marker caveats).
379	CustomCaveatCondition string
380
381	// FullURI is the full RPC method URI that was invoked.
382	FullURI string
383
384	// ProtoSerialized is the full request or response object in the
385	// protobuf binary serialization format.
386	ProtoSerialized []byte
387
388	// ProtoTypeName is the fully qualified name of the protobuf type of the
389	// request or response message that is serialized in the field above.
390	ProtoTypeName string
391
392	// IsError indicates that the message contained within this request is
393	// an error. Will only ever be true for response messages.
394	IsError bool
395
396	// CtxMetadataPairs contains the metadata pairs that were sent along
397	// with the RPC request via the context.
398	CtxMetadataPairs metadata.MD
399	}
400
401	// NewMessageInterceptionRequest creates a new interception request for either
402	// a request or response message.
403	func NewMessageInterceptionRequest(ctx context.Context,
404	authType InterceptType, isStream bool, fullMethod string,
UNCOV 405	m interface{}) (*InterceptionRequest, error) {	×
UNCOV 406		×
UNCOV 407	mac, rawMacaroon, err := macaroonFromContext(ctx)	×
UNCOV 408	if err != nil {	×
409	return nil, err	×
410	}	×
411
UNCOV 412	md, _ := metadata.FromIncomingContext(ctx)	×
UNCOV 413		×
UNCOV 414	req := &InterceptionRequest{	×
UNCOV 415	Type: authType,	×
UNCOV 416	StreamRPC: isStream,	×
UNCOV 417	Macaroon: mac,	×
UNCOV 418	RawMacaroon: rawMacaroon,	×
UNCOV 419	FullURI: fullMethod,	×
UNCOV 420	CtxMetadataPairs: md,	×
UNCOV 421	}	×
UNCOV 422		×
UNCOV 423	// The message is either a proto message or an error, we don't support	×
UNCOV 424	// any other types being intercepted.	×
UNCOV 425	switch t := m.(type) {	×
UNCOV 426	case proto.Message:	×
UNCOV 427	req.ProtoSerialized, err = proto.Marshal(t)	×
UNCOV 428	if err != nil {	×
429	return nil, fmt.Errorf("cannot marshal proto msg: %w",	×
430	err)	×
431	}	×
UNCOV 432	req.ProtoTypeName = string(proto.MessageName(t))	×
433
UNCOV 434	case error:	×
UNCOV 435	req.ProtoSerialized = []byte(t.Error())	×
UNCOV 436	req.ProtoTypeName = "error"	×
UNCOV 437	req.IsError = true	×
438
439	default:	×
440	return nil, fmt.Errorf("unsupported type for interception "+	×
441	"request: %v", m)	×
442	}
443
UNCOV 444	return req, nil	×
445	}
446
447	// NewStreamAuthInterceptionRequest creates a new interception request for a
448	// stream authentication message.
449	func NewStreamAuthInterceptionRequest(ctx context.Context,
UNCOV 450	fullMethod string) (*InterceptionRequest, error) {	×
UNCOV 451		×
UNCOV 452	mac, rawMacaroon, err := macaroonFromContext(ctx)	×
UNCOV 453	if err != nil {	×
454	return nil, err	×
455	}	×
456
UNCOV 457	return &InterceptionRequest{	×
UNCOV 458	Type: TypeStreamAuth,	×
UNCOV 459	StreamRPC: true,	×
UNCOV 460	Macaroon: mac,	×
UNCOV 461	RawMacaroon: rawMacaroon,	×
UNCOV 462	FullURI: fullMethod,	×
UNCOV 463	}, nil	×
464	}
465
466	// macaroonFromContext tries to extract the macaroon from the incoming context.
467	// If there is no macaroon, a nil error is returned since some RPCs might not
468	// require a macaroon. But in case there is something in the macaroon header
469	// field that cannot be parsed, a non-nil error is returned.
470	func macaroonFromContext(ctx context.Context) (*macaroon.Macaroon, []byte,
UNCOV 471	error) {	×
UNCOV 472		×
UNCOV 473	macHex, err := macaroons.RawMacaroonFromContext(ctx)	×
UNCOV 474	if err != nil {	×
475	// If there is no macaroon, we continue anyway as it might be an	×
476	// RPC that doesn't require a macaroon.	×
477	return nil, nil, nil	×
478	}	×
479
UNCOV 480	macBytes, err := hex.DecodeString(macHex)	×
UNCOV 481	if err != nil {	×
482	return nil, nil, err	×
483	}	×
484
UNCOV 485	mac := &macaroon.Macaroon{}	×
UNCOV 486	if err := mac.UnmarshalBinary(macBytes); err != nil {	×
487	return nil, nil, err	×
488	}	×
489
UNCOV 490	return mac, macBytes, nil	×
491	}
492
493	// ToRPC converts the interception request to its RPC counterpart.
494	func (r *InterceptionRequest) ToRPC(requestID,
UNCOV 495	msgID uint64) (*lnrpc.RPCMiddlewareRequest, error) {	×
UNCOV 496		×
UNCOV 497	mdPairs := make(	×
UNCOV 498	map[string]*lnrpc.MetadataValues, len(r.CtxMetadataPairs),	×
UNCOV 499	)	×
UNCOV 500	for key, values := range r.CtxMetadataPairs {	×
UNCOV 501	mdPairs[key] = &lnrpc.MetadataValues{	×
UNCOV 502	Values: values,	×
UNCOV 503	}	×
UNCOV 504	}	×
505
UNCOV 506	rpcRequest := &lnrpc.RPCMiddlewareRequest{	×
UNCOV 507	RequestId: requestID,	×
UNCOV 508	MsgId: msgID,	×
UNCOV 509	RawMacaroon: r.RawMacaroon,	×
UNCOV 510	CustomCaveatCondition: r.CustomCaveatCondition,	×
UNCOV 511	MetadataPairs: mdPairs,	×
UNCOV 512	}	×
UNCOV 513		×
UNCOV 514	switch r.Type {	×
UNCOV 515	case TypeStreamAuth:	×
UNCOV 516	rpcRequest.InterceptType = &lnrpc.RPCMiddlewareRequest_StreamAuth{	×
UNCOV 517	StreamAuth: &lnrpc.StreamAuth{	×
UNCOV 518	MethodFullUri: r.FullURI,	×
UNCOV 519	},	×
UNCOV 520	}	×
521
UNCOV 522	case TypeRequest:	×
UNCOV 523	rpcRequest.InterceptType = &lnrpc.RPCMiddlewareRequest_Request{	×
UNCOV 524	Request: &lnrpc.RPCMessage{	×
UNCOV 525	MethodFullUri: r.FullURI,	×
UNCOV 526	StreamRpc: r.StreamRPC,	×
UNCOV 527	TypeName: r.ProtoTypeName,	×
UNCOV 528	Serialized: r.ProtoSerialized,	×
UNCOV 529	},	×
UNCOV 530	}	×
531
UNCOV 532	case TypeResponse:	×
UNCOV 533	rpcRequest.InterceptType = &lnrpc.RPCMiddlewareRequest_Response{	×
UNCOV 534	Response: &lnrpc.RPCMessage{	×
UNCOV 535	MethodFullUri: r.FullURI,	×
UNCOV 536	StreamRpc: r.StreamRPC,	×
UNCOV 537	TypeName: r.ProtoTypeName,	×
UNCOV 538	Serialized: r.ProtoSerialized,	×
UNCOV 539	IsError: r.IsError,	×
UNCOV 540	},	×
UNCOV 541	}	×
542
543	default:	×
544	return nil, fmt.Errorf("unknown intercept type %v", r.Type)	×
545	}
546
UNCOV 547	return rpcRequest, nil	×
548	}
549
550	// interceptRequest is a struct that keeps track of an interception request sent
551	// out to a middleware and the response that is eventually sent back by the
552	// middleware.
553	type interceptRequest struct {
554	requestID uint64
555	request *InterceptionRequest
556	response chan *interceptResponse
557	}
558
559	// interceptResponse is the response a middleware sends back for each
560	// intercepted message.
561	type interceptResponse struct {
562	err error
563	replace bool
564	replacement interface{}
565	}
566
567	// parseProto parses a proto serialized message of the given type into its
568	// native version.
UNCOV 569	func parseProto(typeName string, serialized []byte) (proto.Message, error) {	×
UNCOV 570	messageType, err := protoregistry.GlobalTypes.FindMessageByName(	×
UNCOV 571	protoreflect.FullName(typeName),	×
UNCOV 572	)	×
UNCOV 573	if err != nil {	×
574	return nil, err	×
575	}	×
UNCOV 576	msg := messageType.New()	×
UNCOV 577	err = proto.Unmarshal(serialized, msg.Interface())	×
UNCOV 578	if err != nil {	×
579	return nil, err	×
580	}	×
581
UNCOV 582	return msg.Interface(), nil	×
583	}
584
585	// replaceProtoMsg replaces the given target message with the content of the
586	// replacement message.
587	func replaceProtoMsg(target interface{}, replacement interface{}) error {	5✔
588	targetMsg, ok := target.(proto.Message)	5✔
589	if !ok {	6✔
590	return fmt.Errorf("target is not a proto message: %v", target)	1✔
591	}	1✔
592
593	replacementMsg, ok := replacement.(proto.Message)	4✔
594	if !ok {	4✔
595	return fmt.Errorf("replacement is not a proto message: %v",	×
596	replacement)	×
597	}	×
598
599	if targetMsg.ProtoReflect().Type() !=	4✔
600	replacementMsg.ProtoReflect().Type() {	5✔
601		1✔
602	return fmt.Errorf("replacement message is of wrong type")	1✔
603	}	1✔
604
605	replacementBytes, err := proto.Marshal(replacementMsg)	3✔
606	if err != nil {	3✔
607	return fmt.Errorf("error marshaling replacement: %w", err)	×
608	}	×
609	err = proto.Unmarshal(replacementBytes, targetMsg)	3✔
610	if err != nil {	3✔
611	return fmt.Errorf("error unmarshaling replacement: %w", err)	×
612	}	×
613
614	return nil	3✔
615	}

lightningnetwork / lnd / 17132206455

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous