12058234999

Committed 27 Nov 2024 09:06PM UTC coverage: 57.847% (-1.1%) from 58.921%

Build # 12058234999

Build Type

Pull #9148

github

Committed by

ProofOfKeags

Commit Message

lnwire: convert DynPropose and DynCommit to use typed tlv records

Pull Request Pull Request #9148: DynComms [2/n]: lnwire: add authenticated wire messages for Dyn*

Run Details

142 of 177 new or added lines in 4 files covered. (80.23%)

19365 existing lines in 251 files now uncovered.

100876 of 174383 relevant lines covered (57.85%)

25338.28 hits per line

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

5.59

/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/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
}

// 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
        }

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

        // 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) {

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

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

lightningnetwork / lnd / 12058234999

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