12312390362

Committed 13 Dec 2024 08:44AM UTC coverage: 57.458% (+8.5%) from 48.92%

Build # 12312390362

Build Type

Pull #9343

github

Committed by

ellemouton

Commit Message

fn: rework the ContextGuard and add tests

In this commit, the ContextGuard struct is re-worked such that the
context that its new main WithCtx method provides is cancelled in sync
with a parent context being cancelled or with it's quit channel being
cancelled. Tests are added to assert the behaviour. In order for the
close of the quit channel to be consistent with the cancelling of the
derived context, the quit channel _must_ be contained internal to the
ContextGuard so that callers are only able to close the channel via the
exposed Quit method which will then take care to first cancel any
derived context that depend on the quit channel before returning.

Pull Request Pull Request #9343: fn: expand the ContextGuard and add tests

Run Details

101853 of 177264 relevant lines covered (57.46%)

24972.93 hits per line

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

72.11

/chanacceptor/rpcacceptor.go

package chanacceptor

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

        "github.com/btcsuite/btcd/btcutil"
        "github.com/btcsuite/btcd/chaincfg"
        "github.com/lightningnetwork/lnd/input"
        "github.com/lightningnetwork/lnd/lnrpc"
        "github.com/lightningnetwork/lnd/lnwallet/chancloser"
        "github.com/lightningnetwork/lnd/lnwire"
)

var (
        errShuttingDown = errors.New("server shutting down")

        // errCustomLength is returned when our custom error's length exceeds
        // our maximum.
        errCustomLength = fmt.Errorf("custom error message exceeds length "+
                "limit: %v", maxErrorLength)

        // errInvalidUpfrontShutdown is returned when we cannot parse the
        // upfront shutdown address returned.
        errInvalidUpfrontShutdown = fmt.Errorf("could not parse upfront " +
                "shutdown address")

        // errInsufficientReserve is returned when the reserve proposed by for
        // a channel is less than the dust limit originally supplied.
        errInsufficientReserve = fmt.Errorf("reserve lower than proposed dust " +
                "limit")

        // errAcceptWithError is returned when we get a response which accepts
        // a channel but ambiguously also sets a custom error message.
        errAcceptWithError = errors.New("channel acceptor response accepts " +
                "channel, but also includes custom error")

        // errMaxHtlcTooHigh is returned if our htlc count exceeds the number
        // hard-set by BOLT 2.
        errMaxHtlcTooHigh = fmt.Errorf("htlc limit exceeds spec limit of: %v",
                input.MaxHTLCNumber/2)

        // maxErrorLength is the maximum error length we allow the error we
        // send to our peer to be.
        maxErrorLength = 500
)

// chanAcceptInfo contains a request for a channel acceptor decision, and a
// channel that the response should be sent on.
type chanAcceptInfo struct {
        request  *ChannelAcceptRequest
        response chan *ChannelAcceptResponse
}

// RPCAcceptor represents the RPC-controlled variant of the ChannelAcceptor.
// One RPCAcceptor allows one RPC client.
type RPCAcceptor struct {
        // receive is a function from which we receive channel acceptance
        // decisions. Note that this function is expected to block.
        receive func() (*lnrpc.ChannelAcceptResponse, error)

        // send is a function which sends requests for channel acceptance
        // decisions into our rpc stream.
        send func(request *lnrpc.ChannelAcceptRequest) error

        // requests is a channel that we send requests for a acceptor response
        // into.
        requests chan *chanAcceptInfo

        // timeout is the amount of time we allow the channel acceptance
        // decision to take. This time includes the time to send a query to the
        // acceptor, and the time it takes to receive a response.
        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
}

// Accept is a predicate on the ChannelAcceptRequest which is sent to the RPC
// client who will respond with the ultimate decision. This function passes the
// request into the acceptor's requests channel, and returns the response it
// receives, failing the request if the timeout elapses.
//
// NOTE: Part of the ChannelAcceptor interface.
func (r *RPCAcceptor) Accept(req *ChannelAcceptRequest) *ChannelAcceptResponse {
        respChan := make(chan *ChannelAcceptResponse, 1)

        newRequest := &chanAcceptInfo{
                request:  req,
                response: respChan,
        }

        // timeout is the time after which ChannelAcceptRequests expire.
        timeout := time.After(r.timeout)

        // Create a rejection response which we can use for the cases where we
        // reject the channel.
        rejectChannel := NewChannelAcceptResponse(
                false, errChannelRejected, nil, 0, 0, 0, 0, 0, 0, false,
        )

        // Send the request to the newRequests channel.
        select {
        case r.requests <- newRequest:

        case <-timeout:
                log.Errorf("RPCAcceptor returned false - reached timeout of %v",
                        r.timeout)
                return rejectChannel

        case <-r.done:
                return rejectChannel

        case <-r.quit:
                return rejectChannel
        }

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

        case <-timeout:
                log.Errorf("RPCAcceptor returned false - reached timeout of %v",
                        r.timeout)
                return rejectChannel

        case <-r.done:
                return rejectChannel

        case <-r.quit:
                return rejectChannel
        }
}

// NewRPCAcceptor creates and returns an instance of the RPCAcceptor.
func NewRPCAcceptor(receive func() (*lnrpc.ChannelAcceptResponse, error),
        send func(*lnrpc.ChannelAcceptRequest) error, timeout time.Duration,
        params *chaincfg.Params, quit chan struct{}) *RPCAcceptor {

        return &RPCAcceptor{
                receive:  receive,
                send:     send,
                requests: make(chan *chanAcceptInfo),
                timeout:  timeout,
                params:   params,
                done:     make(chan struct{}),
                quit:     quit,
        }
}

// Run is the main loop for the RPC Acceptor. This function will block until
// it receives the signal that lnd is shutting down, or the rpc stream is
// cancelled by the client.
func (r *RPCAcceptor) Run() error {
        // Wait for our goroutines to exit before we return.
        defer r.wg.Wait()

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

        // 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 channel acceptor.
        // We expect the receive function to block, so it must be run in a
        // goroutine (otherwise we could not send more than one channel accept
        // request to the client).
        r.wg.Add(1)
        go func() {
                r.receiveResponses(errChan, responses)
                r.wg.Done()
        }()

        return r.sendAcceptRequests(errChan, responses)
}

// receiveResponses receives responses for our channel accept requests and
// dispatches them into the responses channel provided, sending any errors that
// occur into the error channel provided.
func (r *RPCAcceptor) receiveResponses(errChan chan error,
        responses chan *lnrpc.ChannelAcceptResponse) {

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

                var pendingID [32]byte
                copy(pendingID[:], resp.PendingChanId)

                openChanResp := &lnrpc.ChannelAcceptResponse{
                        Accept:          resp.Accept,
                        PendingChanId:   pendingID[:],
                        Error:           resp.Error,
                        UpfrontShutdown: resp.UpfrontShutdown,
                        CsvDelay:        resp.CsvDelay,
                        ReserveSat:      resp.ReserveSat,
                        InFlightMaxMsat: resp.InFlightMaxMsat,
                        MaxHtlcCount:    resp.MaxHtlcCount,
                        MinHtlcIn:       resp.MinHtlcIn,
                        MinAcceptDepth:  resp.MinAcceptDepth,
                        ZeroConf:        resp.ZeroConf,
                }

                // We have received a decision for one of our channel
                // acceptor requests.
                select {
                case responses <- openChanResp:

                case <-r.done:
                        return

                case <-r.quit:
                        return
                }
        }
}

// sendAcceptRequests handles channel acceptor requests sent to us by our
// Accept() function, dispatching them to our acceptor stream and coordinating
// return of responses to their callers.
func (r *RPCAcceptor) sendAcceptRequests(errChan chan error,
        responses chan *lnrpc.ChannelAcceptResponse) error {

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

        // Create a map of pending channel IDs to our original open channel
        // request and a response channel. We keep the original channel open
        // message so that we can validate our response against it.
        acceptRequests := make(map[[32]byte]*chanAcceptInfo)

        for {
                //nolint:ll
                select {
                // Consume requests passed to us from our Accept() function and
                // send them into our stream.
                case newRequest := <-r.requests:

                        req := newRequest.request
                        pendingChanID := req.OpenChanMsg.PendingChannelID

                        // Map the channel commitment type to its RPC
                        // counterpart. Also determine whether the zero-conf or
                        // scid-alias channel types are set.
                        var (
                                commitmentType lnrpc.CommitmentType
                                wantsZeroConf  bool
                                wantsScidAlias bool
                        )

                        if req.OpenChanMsg.ChannelType != nil {
                                channelFeatures := lnwire.RawFeatureVector(
                                        *req.OpenChanMsg.ChannelType,
                                )
                                switch {
                                case channelFeatures.OnlyContains(
                                        lnwire.ZeroConfRequired,
                                        lnwire.ScidAliasRequired,
                                        lnwire.ScriptEnforcedLeaseRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE

                                case channelFeatures.OnlyContains(
                                        lnwire.ZeroConfRequired,
                                        lnwire.ScriptEnforcedLeaseRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE

                                case channelFeatures.OnlyContains(
                                        lnwire.ScidAliasRequired,
                                        lnwire.ScriptEnforcedLeaseRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE

                                case channelFeatures.OnlyContains(
                                        lnwire.ScriptEnforcedLeaseRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE

                                case channelFeatures.OnlyContains(
                                        lnwire.ZeroConfRequired,
                                        lnwire.ScidAliasRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_ANCHORS

                                case channelFeatures.OnlyContains(
                                        lnwire.ZeroConfRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_ANCHORS

                                case channelFeatures.OnlyContains(
                                        lnwire.ScidAliasRequired,
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_ANCHORS

                                case channelFeatures.OnlyContains(
                                        lnwire.AnchorsZeroFeeHtlcTxRequired,
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_ANCHORS

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootChannelsRequiredStaging,
                                        lnwire.ZeroConfRequired,
                                        lnwire.ScidAliasRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootChannelsRequiredStaging,
                                        lnwire.ZeroConfRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootChannelsRequiredStaging,
                                        lnwire.ScidAliasRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootChannelsRequiredStaging,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootOverlayChansRequired,
                                        lnwire.ZeroConfRequired,
                                        lnwire.ScidAliasRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootOverlayChansRequired,
                                        lnwire.ZeroConfRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootOverlayChansRequired,
                                        lnwire.ScidAliasRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY

                                case channelFeatures.OnlyContains(
                                        lnwire.SimpleTaprootOverlayChansRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY

                                case channelFeatures.OnlyContains(
                                        lnwire.StaticRemoteKeyRequired,
                                ):
                                        commitmentType = lnrpc.CommitmentType_STATIC_REMOTE_KEY

                                case channelFeatures.OnlyContains():
                                        commitmentType = lnrpc.CommitmentType_LEGACY

                                default:
                                        log.Warnf("Unhandled commitment type "+
                                                "in channel acceptor request: %v",
                                                req.OpenChanMsg.ChannelType)
                                }

                                if channelFeatures.IsSet(
                                        lnwire.ZeroConfRequired,
                                ) {

                                        wantsZeroConf = true
                                }

                                if channelFeatures.IsSet(
                                        lnwire.ScidAliasRequired,
                                ) {

                                        wantsScidAlias = true
                                }
                        }

                        acceptRequests[pendingChanID] = newRequest

                        // A ChannelAcceptRequest has been received, send it to the client.
                        chanAcceptReq := &lnrpc.ChannelAcceptRequest{
                                NodePubkey:       req.Node.SerializeCompressed(),
                                ChainHash:        req.OpenChanMsg.ChainHash[:],
                                PendingChanId:    req.OpenChanMsg.PendingChannelID[:],
                                FundingAmt:       uint64(req.OpenChanMsg.FundingAmount),
                                PushAmt:          uint64(req.OpenChanMsg.PushAmount),
                                DustLimit:        uint64(req.OpenChanMsg.DustLimit),
                                MaxValueInFlight: uint64(req.OpenChanMsg.MaxValueInFlight),
                                ChannelReserve:   uint64(req.OpenChanMsg.ChannelReserve),
                                MinHtlc:          uint64(req.OpenChanMsg.HtlcMinimum),
                                FeePerKw:         uint64(req.OpenChanMsg.FeePerKiloWeight),
                                CsvDelay:         uint32(req.OpenChanMsg.CsvDelay),
                                MaxAcceptedHtlcs: uint32(req.OpenChanMsg.MaxAcceptedHTLCs),
                                ChannelFlags:     uint32(req.OpenChanMsg.ChannelFlags),
                                CommitmentType:   commitmentType,
                                WantsZeroConf:    wantsZeroConf,
                                WantsScidAlias:   wantsScidAlias,
                        }

                        if err := r.send(chanAcceptReq); err != nil {
                                return err
                        }

                // Process newly received responses from our channel acceptor,
                // looking the original request up in our map of requests and
                // dispatching the response.
                case resp := <-responses:
                        // Look up the appropriate channel to send on given the
                        // pending ID. If a channel is found, send the response
                        // over it.
                        var pendingID [32]byte
                        copy(pendingID[:], resp.PendingChanId)
                        requestInfo, ok := acceptRequests[pendingID]
                        if !ok {
                                continue
                        }

                        // Validate the response we have received. If it is not
                        // valid, we log our error and proceed to deliver the
                        // rejection.
                        accept, acceptErr, shutdown, err := r.validateAcceptorResponse(
                                requestInfo.request.OpenChanMsg.DustLimit, resp,
                        )
                        if err != nil {
                                log.Errorf("Invalid acceptor response: %v", err)
                        }

                        requestInfo.response <- NewChannelAcceptResponse(
                                accept, acceptErr, shutdown,
                                uint16(resp.CsvDelay),
                                uint16(resp.MaxHtlcCount),
                                uint16(resp.MinAcceptDepth),
                                btcutil.Amount(resp.ReserveSat),
                                lnwire.MilliSatoshi(resp.InFlightMaxMsat),
                                lnwire.MilliSatoshi(resp.MinHtlcIn),
                                resp.ZeroConf,
                        )

                        // Delete the channel from the acceptRequests map.
                        delete(acceptRequests, pendingID)

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

                // Exit if we are shutting down.
                case <-r.quit:
                        return errShuttingDown
                }
        }
}

// validateAcceptorResponse validates the response we get from the channel
// acceptor, returning a boolean indicating whether to accept the channel, an
// error to send to the peer, and any validation errors that occurred.
func (r *RPCAcceptor) validateAcceptorResponse(dustLimit btcutil.Amount,
        req *lnrpc.ChannelAcceptResponse) (bool, error, lnwire.DeliveryAddress,
        error) {

        channelStr := hex.EncodeToString(req.PendingChanId)

        // Check that the max htlc count is within the BOLT 2 hard-limit of 483.
        // The initiating side should fail values above this anyway, but we
        // catch the invalid user input here.
        if req.MaxHtlcCount > input.MaxHTLCNumber/2 {
                log.Errorf("Max htlc count: %v for channel: %v is greater "+
                        "than limit of: %v", req.MaxHtlcCount, channelStr,
                        input.MaxHTLCNumber/2)

                return false, errChannelRejected, nil, errMaxHtlcTooHigh
        }

        // Ensure that the reserve that has been proposed, if it is set, is at
        // least the dust limit that was proposed by the remote peer. This is
        // required by BOLT 2.
        reserveSat := btcutil.Amount(req.ReserveSat)
        if reserveSat != 0 && reserveSat < dustLimit {
                log.Errorf("Remote reserve: %v sat for channel: %v must be "+
                        "at least equal to proposed dust limit: %v",
                        req.ReserveSat, channelStr, dustLimit)

                return false, errChannelRejected, nil, errInsufficientReserve
        }

        // Attempt to parse the upfront shutdown address provided.
        upfront, err := chancloser.ParseUpfrontShutdownAddress(
                req.UpfrontShutdown, r.params,
        )
        if err != nil {
                log.Errorf("Could not parse upfront shutdown for "+
                        "%v: %v", channelStr, err)

                return false, errChannelRejected, nil, errInvalidUpfrontShutdown
        }

        // Check that the custom error provided is valid.
        if len(req.Error) > maxErrorLength {
                return false, errChannelRejected, nil, errCustomLength
        }

        var haveCustomError = len(req.Error) != 0

        switch {
        // If accept is true, but we also have an error specified, we fail
        // because this result is ambiguous.
        case req.Accept && haveCustomError:
                return false, errChannelRejected, nil, errAcceptWithError

        // If we accept without an error message, we can just return a nil
        // error.
        case req.Accept:
                return true, nil, upfront, nil

        // If we reject the channel, and have a custom error, then we use it.
        case haveCustomError:
                return false, fmt.Errorf(req.Error), nil, nil

        // Otherwise, we have rejected the channel with no custom error, so we
        // just use a generic error to fail the channel.
        default:
                return false, errChannelRejected, nil, nil
        }
}

// A compile-time constraint to ensure RPCAcceptor implements the ChannelAcceptor
// interface.
var _ ChannelAcceptor = (*RPCAcceptor)(nil)

1	package chanacceptor
2
3	import (
4	"encoding/hex"
5	"errors"
6	"fmt"
7	"sync"
8	"time"
9
10	"github.com/btcsuite/btcd/btcutil"
11	"github.com/btcsuite/btcd/chaincfg"
12	"github.com/lightningnetwork/lnd/input"
13	"github.com/lightningnetwork/lnd/lnrpc"
14	"github.com/lightningnetwork/lnd/lnwallet/chancloser"
15	"github.com/lightningnetwork/lnd/lnwire"
16	)
17
18	var (
19	errShuttingDown = errors.New("server shutting down")
20
21	// errCustomLength is returned when our custom error's length exceeds
22	// our maximum.
23	errCustomLength = fmt.Errorf("custom error message exceeds length "+
24	"limit: %v", maxErrorLength)
25
26	// errInvalidUpfrontShutdown is returned when we cannot parse the
27	// upfront shutdown address returned.
28	errInvalidUpfrontShutdown = fmt.Errorf("could not parse upfront " +
29	"shutdown address")
30
31	// errInsufficientReserve is returned when the reserve proposed by for
32	// a channel is less than the dust limit originally supplied.
33	errInsufficientReserve = fmt.Errorf("reserve lower than proposed dust " +
34	"limit")
35
36	// errAcceptWithError is returned when we get a response which accepts
37	// a channel but ambiguously also sets a custom error message.
38	errAcceptWithError = errors.New("channel acceptor response accepts " +
39	"channel, but also includes custom error")
40
41	// errMaxHtlcTooHigh is returned if our htlc count exceeds the number
42	// hard-set by BOLT 2.
43	errMaxHtlcTooHigh = fmt.Errorf("htlc limit exceeds spec limit of: %v",
44	input.MaxHTLCNumber/2)
45
46	// maxErrorLength is the maximum error length we allow the error we
47	// send to our peer to be.
48	maxErrorLength = 500
49	)
50
51	// chanAcceptInfo contains a request for a channel acceptor decision, and a
52	// channel that the response should be sent on.
53	type chanAcceptInfo struct {
54	request *ChannelAcceptRequest
55	response chan *ChannelAcceptResponse
56	}
57
58	// RPCAcceptor represents the RPC-controlled variant of the ChannelAcceptor.
59	// One RPCAcceptor allows one RPC client.
60	type RPCAcceptor struct {
61	// receive is a function from which we receive channel acceptance
62	// decisions. Note that this function is expected to block.
63	receive func() (*lnrpc.ChannelAcceptResponse, error)
64
65	// send is a function which sends requests for channel acceptance
66	// decisions into our rpc stream.
67	send func(request *lnrpc.ChannelAcceptRequest) error
68
69	// requests is a channel that we send requests for a acceptor response
70	// into.
71	requests chan *chanAcceptInfo
72
73	// timeout is the amount of time we allow the channel acceptance
74	// decision to take. This time includes the time to send a query to the
75	// acceptor, and the time it takes to receive a response.
76	timeout time.Duration
77
78	// params are our current chain params.
79	params *chaincfg.Params
80
81	// done is closed when the rpc client terminates.
82	done chan struct{}
83
84	// quit is closed when lnd is shutting down.
85	quit chan struct{}
86
87	wg sync.WaitGroup
88	}
89
90	// Accept is a predicate on the ChannelAcceptRequest which is sent to the RPC
91	// client who will respond with the ultimate decision. This function passes the
92	// request into the acceptor's requests channel, and returns the response it
93	// receives, failing the request if the timeout elapses.
94	//
95	// NOTE: Part of the ChannelAcceptor interface.
96	func (r RPCAcceptor) Accept(req ChannelAcceptRequest) *ChannelAcceptResponse {	5✔
97	respChan := make(chan *ChannelAcceptResponse, 1)	5✔
98		5✔
99	newRequest := &chanAcceptInfo{	5✔
100	request: req,	5✔
101	response: respChan,	5✔
102	}	5✔
103		5✔
104	// timeout is the time after which ChannelAcceptRequests expire.	5✔
105	timeout := time.After(r.timeout)	5✔
106		5✔
107	// Create a rejection response which we can use for the cases where we	5✔
108	// reject the channel.	5✔
109	rejectChannel := NewChannelAcceptResponse(	5✔
110	false, errChannelRejected, nil, 0, 0, 0, 0, 0, 0, false,	5✔
111	)	5✔
112		5✔
113	// Send the request to the newRequests channel.	5✔
114	select {	5✔
115	case r.requests <- newRequest:	5✔
116
117	case <-timeout:	×
118	log.Errorf("RPCAcceptor returned false - reached timeout of %v",	×
119	r.timeout)	×
120	return rejectChannel	×
121
122	case <-r.done:	×
123	return rejectChannel	×
124
125	case <-r.quit:	×
126	return rejectChannel	×
127	}
128
129	// Receive the response and return it. If no response has been received
130	// in AcceptorTimeout, then return false.
131	select {	5✔
132	case resp := <-respChan:	5✔
133	return resp	5✔
134
135	case <-timeout:	×
136	log.Errorf("RPCAcceptor returned false - reached timeout of %v",	×
137	r.timeout)	×
138	return rejectChannel	×
139
140	case <-r.done:	×
141	return rejectChannel	×
142
143	case <-r.quit:	×
144	return rejectChannel	×
145	}
146	}
147
148	// NewRPCAcceptor creates and returns an instance of the RPCAcceptor.
149	func NewRPCAcceptor(receive func() (*lnrpc.ChannelAcceptResponse, error),
150	send func(*lnrpc.ChannelAcceptRequest) error, timeout time.Duration,
151	params chaincfg.Params, quit chan struct{}) RPCAcceptor {	11✔
152		11✔
153	return &RPCAcceptor{	11✔
154	receive: receive,	11✔
155	send: send,	11✔
156	requests: make(chan *chanAcceptInfo),	11✔
157	timeout: timeout,	11✔
158	params: params,	11✔
159	done: make(chan struct{}),	11✔
160	quit: quit,	11✔
161	}	11✔
162	}	11✔
163
164	// Run is the main loop for the RPC Acceptor. This function will block until
165	// it receives the signal that lnd is shutting down, or the rpc stream is
166	// cancelled by the client.
167	func (r *RPCAcceptor) Run() error {	3✔
168	// Wait for our goroutines to exit before we return.	3✔
169	defer r.wg.Wait()	3✔
170		3✔
171	// Create a channel that responses from acceptors are sent into.	3✔
172	responses := make(chan *lnrpc.ChannelAcceptResponse)	3✔
173		3✔
174	// errChan is used by the receive loop to signal any errors that occur	3✔
175	// during reading from the stream. This is primarily used to shutdown	3✔
176	// the send loop in the case of an RPC client disconnecting.	3✔
177	errChan := make(chan error, 1)	3✔
178		3✔
179	// Start a goroutine to receive responses from the channel acceptor.	3✔
180	// We expect the receive function to block, so it must be run in a	3✔
181	// goroutine (otherwise we could not send more than one channel accept	3✔
182	// request to the client).	3✔
183	r.wg.Add(1)	3✔
184	go func() {	6✔
185	r.receiveResponses(errChan, responses)	3✔
186	r.wg.Done()	3✔
187	}()	3✔
188
189	return r.sendAcceptRequests(errChan, responses)	3✔
190	}
191
192	// receiveResponses receives responses for our channel accept requests and
193	// dispatches them into the responses channel provided, sending any errors that
194	// occur into the error channel provided.
195	func (r *RPCAcceptor) receiveResponses(errChan chan error,
196	responses chan *lnrpc.ChannelAcceptResponse) {	3✔
197		3✔
198	for {	11✔
199	resp, err := r.receive()	8✔
200	if err != nil {	11✔
201	errChan <- err	3✔
202	return	3✔
203	}	3✔
204
205	var pendingID [32]byte	5✔
206	copy(pendingID[:], resp.PendingChanId)	5✔
207		5✔
208	openChanResp := &lnrpc.ChannelAcceptResponse{	5✔
209	Accept: resp.Accept,	5✔
210	PendingChanId: pendingID[:],	5✔
211	Error: resp.Error,	5✔
212	UpfrontShutdown: resp.UpfrontShutdown,	5✔
213	CsvDelay: resp.CsvDelay,	5✔
214	ReserveSat: resp.ReserveSat,	5✔
215	InFlightMaxMsat: resp.InFlightMaxMsat,	5✔
216	MaxHtlcCount: resp.MaxHtlcCount,	5✔
217	MinHtlcIn: resp.MinHtlcIn,	5✔
218	MinAcceptDepth: resp.MinAcceptDepth,	5✔
219	ZeroConf: resp.ZeroConf,	5✔
220	}	5✔
221		5✔
222	// We have received a decision for one of our channel	5✔
223	// acceptor requests.	5✔
224	select {	5✔
225	case responses <- openChanResp:	5✔
226
227	case <-r.done:	×
228	return	×
229
230	case <-r.quit:	×
231	return	×
232	}
233	}
234	}
235
236	// sendAcceptRequests handles channel acceptor requests sent to us by our
237	// Accept() function, dispatching them to our acceptor stream and coordinating
238	// return of responses to their callers.
239	func (r *RPCAcceptor) sendAcceptRequests(errChan chan error,
240	responses chan *lnrpc.ChannelAcceptResponse) error {	3✔
241		3✔
242	// Close the done channel to indicate that the acceptor is no longer	3✔
243	// listening and any in-progress requests should be terminated.	3✔
244	defer close(r.done)	3✔
245		3✔
246	// Create a map of pending channel IDs to our original open channel	3✔
247	// request and a response channel. We keep the original channel open	3✔
248	// message so that we can validate our response against it.	3✔
249	acceptRequests := make(map[[32]byte]*chanAcceptInfo)	3✔
250		3✔
251	for {	16✔
252	//nolint:ll	13✔
253	select {	13✔
254	// Consume requests passed to us from our Accept() function and
255	// send them into our stream.
256	case newRequest := <-r.requests:	5✔
257		5✔
258	req := newRequest.request	5✔
259	pendingChanID := req.OpenChanMsg.PendingChannelID	5✔
260		5✔
261	// Map the channel commitment type to its RPC	5✔
262	// counterpart. Also determine whether the zero-conf or	5✔
263	// scid-alias channel types are set.	5✔
264	var (	5✔
265	commitmentType lnrpc.CommitmentType	5✔
266	wantsZeroConf bool	5✔
267	wantsScidAlias bool	5✔
268	)	5✔
269		5✔
270	if req.OpenChanMsg.ChannelType != nil {	5✔
271	channelFeatures := lnwire.RawFeatureVector(	×
272	*req.OpenChanMsg.ChannelType,	×
273	)	×
274	switch {	×
275	case channelFeatures.OnlyContains(
276	lnwire.ZeroConfRequired,
277	lnwire.ScidAliasRequired,
278	lnwire.ScriptEnforcedLeaseRequired,
279	lnwire.AnchorsZeroFeeHtlcTxRequired,
280	lnwire.StaticRemoteKeyRequired,
281	):	×
282	commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE	×
283
284	case channelFeatures.OnlyContains(
285	lnwire.ZeroConfRequired,
286	lnwire.ScriptEnforcedLeaseRequired,
287	lnwire.AnchorsZeroFeeHtlcTxRequired,
288	lnwire.StaticRemoteKeyRequired,
289	):	×
290	commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE	×
291
292	case channelFeatures.OnlyContains(
293	lnwire.ScidAliasRequired,
294	lnwire.ScriptEnforcedLeaseRequired,
295	lnwire.AnchorsZeroFeeHtlcTxRequired,
296	lnwire.StaticRemoteKeyRequired,
297	):	×
298	commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE	×
299
300	case channelFeatures.OnlyContains(
301	lnwire.ScriptEnforcedLeaseRequired,
302	lnwire.AnchorsZeroFeeHtlcTxRequired,
303	lnwire.StaticRemoteKeyRequired,
304	):	×
305	commitmentType = lnrpc.CommitmentType_SCRIPT_ENFORCED_LEASE	×
306
307	case channelFeatures.OnlyContains(
308	lnwire.ZeroConfRequired,
309	lnwire.ScidAliasRequired,
310	lnwire.AnchorsZeroFeeHtlcTxRequired,
311	lnwire.StaticRemoteKeyRequired,
312	):	×
313	commitmentType = lnrpc.CommitmentType_ANCHORS	×
314
315	case channelFeatures.OnlyContains(
316	lnwire.ZeroConfRequired,
317	lnwire.AnchorsZeroFeeHtlcTxRequired,
318	lnwire.StaticRemoteKeyRequired,
319	):	×
320	commitmentType = lnrpc.CommitmentType_ANCHORS	×
321
322	case channelFeatures.OnlyContains(
323	lnwire.ScidAliasRequired,
324	lnwire.AnchorsZeroFeeHtlcTxRequired,
325	lnwire.StaticRemoteKeyRequired,
326	):	×
327	commitmentType = lnrpc.CommitmentType_ANCHORS	×
328
329	case channelFeatures.OnlyContains(
330	lnwire.AnchorsZeroFeeHtlcTxRequired,
331	lnwire.StaticRemoteKeyRequired,
332	):	×
333	commitmentType = lnrpc.CommitmentType_ANCHORS	×
334
335	case channelFeatures.OnlyContains(
336	lnwire.SimpleTaprootChannelsRequiredStaging,
337	lnwire.ZeroConfRequired,
338	lnwire.ScidAliasRequired,
339	):	×
340	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT	×
341
342	case channelFeatures.OnlyContains(
343	lnwire.SimpleTaprootChannelsRequiredStaging,
344	lnwire.ZeroConfRequired,
345	):	×
346	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT	×
347
348	case channelFeatures.OnlyContains(
349	lnwire.SimpleTaprootChannelsRequiredStaging,
350	lnwire.ScidAliasRequired,
351	):	×
352	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT	×
353
354	case channelFeatures.OnlyContains(
355	lnwire.SimpleTaprootChannelsRequiredStaging,
356	):	×
357	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT	×
358
359	case channelFeatures.OnlyContains(
360	lnwire.SimpleTaprootOverlayChansRequired,
361	lnwire.ZeroConfRequired,
362	lnwire.ScidAliasRequired,
363	):	×
364	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY	×
365
366	case channelFeatures.OnlyContains(
367	lnwire.SimpleTaprootOverlayChansRequired,
368	lnwire.ZeroConfRequired,
369	):	×
370	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY	×
371
372	case channelFeatures.OnlyContains(
373	lnwire.SimpleTaprootOverlayChansRequired,
374	lnwire.ScidAliasRequired,
375	):	×
376	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY	×
377
378	case channelFeatures.OnlyContains(
379	lnwire.SimpleTaprootOverlayChansRequired,
380	):	×
381	commitmentType = lnrpc.CommitmentType_SIMPLE_TAPROOT_OVERLAY	×
382
383	case channelFeatures.OnlyContains(
384	lnwire.StaticRemoteKeyRequired,
385	):	×
386	commitmentType = lnrpc.CommitmentType_STATIC_REMOTE_KEY	×
387
388	case channelFeatures.OnlyContains():	×
389	commitmentType = lnrpc.CommitmentType_LEGACY	×
390
391	default:	×
392	log.Warnf("Unhandled commitment type "+	×
393	"in channel acceptor request: %v",	×
394	req.OpenChanMsg.ChannelType)	×
395	}
396
397	if channelFeatures.IsSet(	×
398	lnwire.ZeroConfRequired,	×
399	) {	×
400		×
401	wantsZeroConf = true	×
402	}	×
403
404	if channelFeatures.IsSet(	×
405	lnwire.ScidAliasRequired,	×
406	) {	×
407		×
408	wantsScidAlias = true	×
409	}	×
410	}
411
412	acceptRequests[pendingChanID] = newRequest	5✔
413		5✔
414	// A ChannelAcceptRequest has been received, send it to the client.	5✔
415	chanAcceptReq := &lnrpc.ChannelAcceptRequest{	5✔
416	NodePubkey: req.Node.SerializeCompressed(),	5✔
417	ChainHash: req.OpenChanMsg.ChainHash[:],	5✔
418	PendingChanId: req.OpenChanMsg.PendingChannelID[:],	5✔
419	FundingAmt: uint64(req.OpenChanMsg.FundingAmount),	5✔
420	PushAmt: uint64(req.OpenChanMsg.PushAmount),	5✔
421	DustLimit: uint64(req.OpenChanMsg.DustLimit),	5✔
422	MaxValueInFlight: uint64(req.OpenChanMsg.MaxValueInFlight),	5✔
423	ChannelReserve: uint64(req.OpenChanMsg.ChannelReserve),	5✔
424	MinHtlc: uint64(req.OpenChanMsg.HtlcMinimum),	5✔
425	FeePerKw: uint64(req.OpenChanMsg.FeePerKiloWeight),	5✔
426	CsvDelay: uint32(req.OpenChanMsg.CsvDelay),	5✔
427	MaxAcceptedHtlcs: uint32(req.OpenChanMsg.MaxAcceptedHTLCs),	5✔
428	ChannelFlags: uint32(req.OpenChanMsg.ChannelFlags),	5✔
429	CommitmentType: commitmentType,	5✔
430	WantsZeroConf: wantsZeroConf,	5✔
431	WantsScidAlias: wantsScidAlias,	5✔
432	}	5✔
433		5✔
434	if err := r.send(chanAcceptReq); err != nil {	5✔
435	return err	×
436	}	×
437
438	// Process newly received responses from our channel acceptor,
439	// looking the original request up in our map of requests and
440	// dispatching the response.
441	case resp := <-responses:	5✔
442	// Look up the appropriate channel to send on given the	5✔
443	// pending ID. If a channel is found, send the response	5✔
444	// over it.	5✔
445	var pendingID [32]byte	5✔
446	copy(pendingID[:], resp.PendingChanId)	5✔
447	requestInfo, ok := acceptRequests[pendingID]	5✔
448	if !ok {	5✔
449	continue	×
450	}
451
452	// Validate the response we have received. If it is not
453	// valid, we log our error and proceed to deliver the
454	// rejection.
455	accept, acceptErr, shutdown, err := r.validateAcceptorResponse(	5✔
456	requestInfo.request.OpenChanMsg.DustLimit, resp,	5✔
457	)	5✔
458	if err != nil {	7✔
459	log.Errorf("Invalid acceptor response: %v", err)	2✔
460	}	2✔
461
462	requestInfo.response <- NewChannelAcceptResponse(	5✔
463	accept, acceptErr, shutdown,	5✔
464	uint16(resp.CsvDelay),	5✔
465	uint16(resp.MaxHtlcCount),	5✔
466	uint16(resp.MinAcceptDepth),	5✔
467	btcutil.Amount(resp.ReserveSat),	5✔
468	lnwire.MilliSatoshi(resp.InFlightMaxMsat),	5✔
469	lnwire.MilliSatoshi(resp.MinHtlcIn),	5✔
470	resp.ZeroConf,	5✔
471	)	5✔
472		5✔
473	// Delete the channel from the acceptRequests map.	5✔
474	delete(acceptRequests, pendingID)	5✔
475
476	// If we failed to receive from our acceptor, we exit.
477	case err := <-errChan:	×
478	log.Errorf("Received an error: %v, shutting down", err)	×
479	return err	×
480
481	// Exit if we are shutting down.
482	case <-r.quit:	3✔
483	return errShuttingDown	3✔
484	}
485	}
486	}
487
488	// validateAcceptorResponse validates the response we get from the channel
489	// acceptor, returning a boolean indicating whether to accept the channel, an
490	// error to send to the peer, and any validation errors that occurred.
491	func (r *RPCAcceptor) validateAcceptorResponse(dustLimit btcutil.Amount,
492	req *lnrpc.ChannelAcceptResponse) (bool, error, lnwire.DeliveryAddress,
493	error) {	13✔
494		13✔
495	channelStr := hex.EncodeToString(req.PendingChanId)	13✔
496		13✔
497	// Check that the max htlc count is within the BOLT 2 hard-limit of 483.	13✔
498	// The initiating side should fail values above this anyway, but we	13✔
499	// catch the invalid user input here.	13✔
500	if req.MaxHtlcCount > input.MaxHTLCNumber/2 {	14✔
501	log.Errorf("Max htlc count: %v for channel: %v is greater "+	1✔
502	"than limit of: %v", req.MaxHtlcCount, channelStr,	1✔
503	input.MaxHTLCNumber/2)	1✔
504		1✔
505	return false, errChannelRejected, nil, errMaxHtlcTooHigh	1✔
506	}	1✔
507
508	// Ensure that the reserve that has been proposed, if it is set, is at
509	// least the dust limit that was proposed by the remote peer. This is
510	// required by BOLT 2.
511	reserveSat := btcutil.Amount(req.ReserveSat)	12✔
512	if reserveSat != 0 && reserveSat < dustLimit {	14✔
513	log.Errorf("Remote reserve: %v sat for channel: %v must be "+	2✔
514	"at least equal to proposed dust limit: %v",	2✔
515	req.ReserveSat, channelStr, dustLimit)	2✔
516		2✔
517	return false, errChannelRejected, nil, errInsufficientReserve	2✔
518	}	2✔
519
520	// Attempt to parse the upfront shutdown address provided.
521	upfront, err := chancloser.ParseUpfrontShutdownAddress(	10✔
522	req.UpfrontShutdown, r.params,	10✔
523	)	10✔
524	if err != nil {	11✔
525	log.Errorf("Could not parse upfront shutdown for "+	1✔
526	"%v: %v", channelStr, err)	1✔
527		1✔
528	return false, errChannelRejected, nil, errInvalidUpfrontShutdown	1✔
529	}	1✔
530
531	// Check that the custom error provided is valid.
532	if len(req.Error) > maxErrorLength {	10✔
533	return false, errChannelRejected, nil, errCustomLength	1✔
534	}	1✔
535
536	var haveCustomError = len(req.Error) != 0	8✔
537		8✔
538	switch {	8✔
539	// If accept is true, but we also have an error specified, we fail
540	// because this result is ambiguous.
541	case req.Accept && haveCustomError:	2✔
542	return false, errChannelRejected, nil, errAcceptWithError	2✔
543
544	// If we accept without an error message, we can just return a nil
545	// error.
546	case req.Accept:	2✔
547	return true, nil, upfront, nil	2✔
548
549	// If we reject the channel, and have a custom error, then we use it.
550	case haveCustomError:	2✔
551	return false, fmt.Errorf(req.Error), nil, nil	2✔
552
553	// Otherwise, we have rejected the channel with no custom error, so we
554	// just use a generic error to fail the channel.
555	default:	2✔
556	return false, errChannelRejected, nil, nil	2✔
557	}
558	}
559
560	// A compile-time constraint to ensure RPCAcceptor implements the ChannelAcceptor
561	// interface.
562	var _ ChannelAcceptor = (*RPCAcceptor)(nil)

lightningnetwork / lnd / 12312390362

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