9915780197

Committed 13 Jul 2024 12:30AM UTC coverage: 49.268% (-9.1%) from 58.413%

Build # 9915780197

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #8653 from ProofOfKeags/fn-prim

DynComms [0/n]: `fn` package additions

Run Details

92837 of 188433 relevant lines covered (49.27%)

1.55 hits per line

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

65.32

/htlcswitch/interceptable_switch.go

package htlcswitch

import (
        "crypto/sha256"
        "fmt"
        "sync"

        "github.com/go-errors/errors"
        "github.com/lightningnetwork/lnd/chainntnfs"
        "github.com/lightningnetwork/lnd/channeldb/models"
        "github.com/lightningnetwork/lnd/htlcswitch/hop"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/lnwire"
)

var (
        // ErrFwdNotExists is an error returned when the caller tries to resolve
        // a forward that doesn't exist anymore.
        ErrFwdNotExists = errors.New("forward does not exist")

        // ErrUnsupportedFailureCode when processing of an unsupported failure
        // code is attempted.
        ErrUnsupportedFailureCode = errors.New("unsupported failure code")

        errBlockStreamStopped = errors.New("block epoch stream stopped")
)

// InterceptableSwitch is an implementation of ForwardingSwitch interface.
// This implementation is used like a proxy that wraps the switch and
// intercepts forward requests. A reference to the Switch is held in order
// to communicate back the interception result where the options are:
// Resume - forwards the original request to the switch as is.
// Settle - routes UpdateFulfillHTLC to the originating link.
// Fail - routes UpdateFailHTLC to the originating link.
type InterceptableSwitch struct {
        // htlcSwitch is the underline switch
        htlcSwitch *Switch

        // intercepted is where we stream all intercepted packets coming from
        // the switch.
        intercepted chan *interceptedPackets

        // resolutionChan is where we stream all responses coming from the
        // interceptor client.
        resolutionChan chan *fwdResolution

        onchainIntercepted chan InterceptedForward

        // interceptorRegistration is a channel that we use to synchronize
        // client connect and disconnect.
        interceptorRegistration chan ForwardInterceptor

        // requireInterceptor indicates whether processing should block if no
        // interceptor is connected.
        requireInterceptor bool

        // interceptor is the handler for intercepted packets.
        interceptor ForwardInterceptor

        // heldHtlcSet keeps track of outstanding intercepted forwards.
        heldHtlcSet *heldHtlcSet

        // cltvRejectDelta defines the number of blocks before the expiry of the
        // htlc where we no longer intercept it and instead cancel it back.
        cltvRejectDelta uint32

        // cltvInterceptDelta defines the number of blocks before the expiry of
        // the htlc where we don't intercept anymore. This value must be greater
        // than CltvRejectDelta, because we don't want to offer htlcs to the
        // interceptor client for which there is no time left to resolve them
        // anymore.
        cltvInterceptDelta uint32

        // notifier is an instance of a chain notifier that we'll use to signal
        // the switch when a new block has arrived.
        notifier chainntnfs.ChainNotifier

        // blockEpochStream is an active block epoch event stream backed by an
        // active ChainNotifier instance. This will be used to retrieve the
        // latest height of the chain.
        blockEpochStream *chainntnfs.BlockEpochEvent

        // currentHeight is the currently best known height.
        currentHeight int32

        wg   sync.WaitGroup
        quit chan struct{}
}

type interceptedPackets struct {
        packets  []*htlcPacket
        linkQuit chan struct{}
        isReplay bool
}

// FwdAction defines the various resolution types.
type FwdAction int

const (
        // FwdActionResume forwards the intercepted packet to the switch.
        FwdActionResume FwdAction = iota

        // FwdActionSettle settles the intercepted packet with a preimage.
        FwdActionSettle

        // FwdActionFail fails the intercepted packet back to the sender.
        FwdActionFail
)

// FwdResolution defines the action to be taken on an intercepted packet.
type FwdResolution struct {
        // Key is the incoming circuit key of the htlc.
        Key models.CircuitKey

        // Action is the action to take on the intercepted htlc.
        Action FwdAction

        // Preimage is the preimage that is to be used for settling if Action is
        // FwdActionSettle.
        Preimage lntypes.Preimage

        // FailureMessage is the encrypted failure message that is to be passed
        // back to the sender if action is FwdActionFail.
        FailureMessage []byte

        // FailureCode is the failure code that is to be passed back to the
        // sender if action is FwdActionFail.
        FailureCode lnwire.FailCode
}

type fwdResolution struct {
        resolution *FwdResolution
        errChan    chan error
}

// InterceptableSwitchConfig contains the configuration of InterceptableSwitch.
type InterceptableSwitchConfig struct {
        // Switch is a reference to the actual switch implementation that
        // packets get sent to on resume.
        Switch *Switch

        // Notifier is an instance of a chain notifier that we'll use to signal
        // the switch when a new block has arrived.
        Notifier chainntnfs.ChainNotifier

        // CltvRejectDelta defines the number of blocks before the expiry of the
        // htlc where we auto-fail an intercepted htlc to prevent channel
        // force-closure.
        CltvRejectDelta uint32

        // CltvInterceptDelta defines the number of blocks before the expiry of
        // the htlc where we don't intercept anymore. This value must be greater
        // than CltvRejectDelta, because we don't want to offer htlcs to the
        // interceptor client for which there is no time left to resolve them
        // anymore.
        CltvInterceptDelta uint32

        // RequireInterceptor indicates whether processing should block if no
        // interceptor is connected.
        RequireInterceptor bool
}

// NewInterceptableSwitch returns an instance of InterceptableSwitch.
func NewInterceptableSwitch(cfg *InterceptableSwitchConfig) (
        *InterceptableSwitch, error) {

        if cfg.CltvInterceptDelta <= cfg.CltvRejectDelta {
                return nil, fmt.Errorf("cltv intercept delta %v not greater "+
                        "than cltv reject delta %v",
                        cfg.CltvInterceptDelta, cfg.CltvRejectDelta)
        }

        return &InterceptableSwitch{
                htlcSwitch:              cfg.Switch,
                intercepted:             make(chan *interceptedPackets),
                onchainIntercepted:      make(chan InterceptedForward),
                interceptorRegistration: make(chan ForwardInterceptor),
                heldHtlcSet:             newHeldHtlcSet(),
                resolutionChan:          make(chan *fwdResolution),
                requireInterceptor:      cfg.RequireInterceptor,
                cltvRejectDelta:         cfg.CltvRejectDelta,
                cltvInterceptDelta:      cfg.CltvInterceptDelta,
                notifier:                cfg.Notifier,

                quit: make(chan struct{}),
        }, nil
}

// SetInterceptor sets the ForwardInterceptor to be used. A nil argument
// unregisters the current interceptor.
func (s *InterceptableSwitch) SetInterceptor(
        interceptor ForwardInterceptor) {

        // Synchronize setting the handler with the main loop to prevent race
        // conditions.
        select {
        case s.interceptorRegistration <- interceptor:

        case <-s.quit:
        }
}

func (s *InterceptableSwitch) Start() error {
        blockEpochStream, err := s.notifier.RegisterBlockEpochNtfn(nil)
        if err != nil {
                return err
        }
        s.blockEpochStream = blockEpochStream

        s.wg.Add(1)
        go func() {
                defer s.wg.Done()

                err := s.run()
                if err != nil {
                        log.Errorf("InterceptableSwitch stopped: %v", err)
                }
        }()

        return nil
}

func (s *InterceptableSwitch) Stop() error {
        close(s.quit)
        s.wg.Wait()

        s.blockEpochStream.Cancel()

        return nil
}

func (s *InterceptableSwitch) run() error {
        // The block epoch stream will immediately stream the current height.
        // Read it out here.
        select {
        case currentBlock, ok := <-s.blockEpochStream.Epochs:
                if !ok {
                        return errBlockStreamStopped
                }
                s.currentHeight = currentBlock.Height

        case <-s.quit:
                return nil
        }

        log.Debugf("InterceptableSwitch running: height=%v, "+
                "requireInterceptor=%v", s.currentHeight, s.requireInterceptor)

        for {
                select {
                // An interceptor registration or de-registration came in.
                case interceptor := <-s.interceptorRegistration:
                        s.setInterceptor(interceptor)

                case packets := <-s.intercepted:
                        var notIntercepted []*htlcPacket
                        for _, p := range packets.packets {
                                intercepted, err := s.interceptForward(
                                        p, packets.isReplay,
                                )
                                if err != nil {
                                        return err
                                }

                                if !intercepted {
                                        notIntercepted = append(
                                                notIntercepted, p,
                                        )
                                }
                        }
                        err := s.htlcSwitch.ForwardPackets(
                                packets.linkQuit, notIntercepted...,
                        )
                        if err != nil {
                                log.Errorf("Cannot forward packets: %v", err)
                        }

                case fwd := <-s.onchainIntercepted:
                        // For on-chain interceptions, we don't know if it has
                        // already been offered before. This information is in
                        // the forwarding package which isn't easily accessible
                        // from contractcourt. It is likely though that it was
                        // already intercepted in the off-chain flow. And even
                        // if not, it is safe to signal replay so that we won't
                        // unexpectedly skip over this htlc.
                        if _, err := s.forward(fwd, true); err != nil {
                                return err
                        }

                case res := <-s.resolutionChan:
                        res.errChan <- s.resolve(res.resolution)

                case currentBlock, ok := <-s.blockEpochStream.Epochs:
                        if !ok {
                                return errBlockStreamStopped
                        }

                        s.currentHeight = currentBlock.Height

                        // A new block is appended. Fail any held htlcs that
                        // expire at this height to prevent channel force-close.
                        s.failExpiredHtlcs()

                case <-s.quit:
                        return nil
                }
        }
}

func (s *InterceptableSwitch) failExpiredHtlcs() {
        s.heldHtlcSet.popAutoFails(
                uint32(s.currentHeight),
                func(fwd InterceptedForward) {
                        err := fwd.FailWithCode(
                                lnwire.CodeTemporaryChannelFailure,
                        )
                        if err != nil {
                                log.Errorf("Cannot fail packet: %v", err)
                        }
                },
        )
}

func (s *InterceptableSwitch) sendForward(fwd InterceptedForward) {
        err := s.interceptor(fwd.Packet())
        if err != nil {
                // Only log the error. If we couldn't send the packet, we assume
                // that the interceptor will reconnect so that we can retry.
                log.Debugf("Interceptor cannot handle forward: %v", err)
        }
}

func (s *InterceptableSwitch) setInterceptor(interceptor ForwardInterceptor) {
        s.interceptor = interceptor

        // Replay all currently held htlcs. When an interceptor is not required,
        // there may be none because they've been cleared after the previous
        // disconnect.
        if interceptor != nil {
                log.Debugf("Interceptor connected")

                s.heldHtlcSet.forEach(s.sendForward)

                return
        }

        // The interceptor disconnects. If an interceptor is required, keep the
        // held htlcs.
        if s.requireInterceptor {
                log.Infof("Interceptor disconnected, retaining held packets")

                return
        }

        // Interceptor is not required. Release held forwards.
        log.Infof("Interceptor disconnected, resolving held packets")

        s.heldHtlcSet.popAll(func(fwd InterceptedForward) {
                err := fwd.Resume()
                if err != nil {
                        log.Errorf("Failed to resume hold forward %v", err)
                }
        })
}

func (s *InterceptableSwitch) resolve(res *FwdResolution) error {
        intercepted, err := s.heldHtlcSet.pop(res.Key)
        if err != nil {
                return err
        }

        switch res.Action {
        case FwdActionResume:
                return intercepted.Resume()

        case FwdActionSettle:
                return intercepted.Settle(res.Preimage)

        case FwdActionFail:
                if len(res.FailureMessage) > 0 {
                        return intercepted.Fail(res.FailureMessage)
                }

                return intercepted.FailWithCode(res.FailureCode)

        default:
                return fmt.Errorf("unrecognized action %v", res.Action)
        }
}

// Resolve resolves an intercepted packet.
func (s *InterceptableSwitch) Resolve(res *FwdResolution) error {
        internalRes := &fwdResolution{
                resolution: res,
                errChan:    make(chan error, 1),
        }

        select {
        case s.resolutionChan <- internalRes:

        case <-s.quit:
                return errors.New("switch shutting down")
        }

        select {
        case err := <-internalRes.errChan:
                return err

        case <-s.quit:
                return errors.New("switch shutting down")
        }
}

// ForwardPackets attempts to forward the batch of htlcs to a connected
// interceptor. If the interceptor signals the resume action, the htlcs are
// forwarded to the switch. The link's quit signal should be provided to allow
// cancellation of forwarding during link shutdown.
func (s *InterceptableSwitch) ForwardPackets(linkQuit chan struct{}, isReplay bool,
        packets ...*htlcPacket) error {

        // Synchronize with the main event loop. This should be light in the
        // case where there is no interceptor.
        select {
        case s.intercepted <- &interceptedPackets{
                packets:  packets,
                linkQuit: linkQuit,
                isReplay: isReplay,
        }:

        case <-linkQuit:
                log.Debugf("Forward cancelled because link quit")

        case <-s.quit:
                return errors.New("interceptable switch quit")
        }

        return nil
}

// ForwardPacket forwards a single htlc to the external interceptor.
func (s *InterceptableSwitch) ForwardPacket(
        fwd InterceptedForward) error {

        select {
        case s.onchainIntercepted <- fwd:

        case <-s.quit:
                return errors.New("interceptable switch quit")
        }

        return nil
}

// interceptForward forwards the packet to the external interceptor after
// checking the interception criteria.
func (s *InterceptableSwitch) interceptForward(packet *htlcPacket,
        isReplay bool) (bool, error) {

        switch htlc := packet.htlc.(type) {
        case *lnwire.UpdateAddHTLC:
                // We are not interested in intercepting initiated payments.
                if packet.incomingChanID == hop.Source {
                        return false, nil
                }

                intercepted := &interceptedForward{
                        htlc:       htlc,
                        packet:     packet,
                        htlcSwitch: s.htlcSwitch,
                        autoFailHeight: int32(packet.incomingTimeout -
                                s.cltvRejectDelta),
                }

                // Handle forwards that are too close to expiry.
                handled, err := s.handleExpired(intercepted)
                if err != nil {
                        log.Errorf("Error handling intercepted htlc "+
                                "that expires too soon: circuit=%v, "+
                                "incoming_timeout=%v, err=%v",
                                packet.inKey(), packet.incomingTimeout, err)

                        // Return false so that the packet is offered as normal
                        // to the switch. This isn't ideal because interception
                        // may be configured as always-on and is skipped now.
                        // Returning true isn't great either, because the htlc
                        // will remain stuck and potentially force-close the
                        // channel. But in the end, we should never get here, so
                        // the actual return value doesn't matter that much.
                        return false, nil
                }
                if handled {
                        return true, nil
                }

                return s.forward(intercepted, isReplay)

        default:
                return false, nil
        }
}

// forward records the intercepted htlc and forwards it to the interceptor.
func (s *InterceptableSwitch) forward(
        fwd InterceptedForward, isReplay bool) (bool, error) {

        inKey := fwd.Packet().IncomingCircuit

        // Ignore already held htlcs.
        if s.heldHtlcSet.exists(inKey) {
                return true, nil
        }

        // If there is no interceptor currently registered, configuration and packet
        // replay status determine how the packet is handled.
        if s.interceptor == nil {
                // Process normally if an interceptor is not required.
                if !s.requireInterceptor {
                        return false, nil
                }

                // We are in interceptor-required mode. If this is a new packet, it is
                // still safe to fail back. The interceptor has never seen this packet
                // yet. This limits the backlog of htlcs when the interceptor is down.
                if !isReplay {
                        err := fwd.FailWithCode(
                                lnwire.CodeTemporaryChannelFailure,
                        )
                        if err != nil {
                                log.Errorf("Cannot fail packet: %v", err)
                        }

                        return true, nil
                }

                // This packet is a replay. It is not safe to fail back, because the
                // interceptor may still signal otherwise upon reconnect. Keep the
                // packet in the queue until then.
                if err := s.heldHtlcSet.push(inKey, fwd); err != nil {
                        return false, err
                }

                return true, nil
        }

        // There is an interceptor registered. We can forward the packet right now.
        // Hold it in the queue too to track what is outstanding.
        if err := s.heldHtlcSet.push(inKey, fwd); err != nil {
                return false, err
        }

        s.sendForward(fwd)

        return true, nil
}

// handleExpired checks that the htlc isn't too close to the channel
// force-close broadcast height. If it is, it is cancelled back.
func (s *InterceptableSwitch) handleExpired(fwd *interceptedForward) (
        bool, error) {

        height := uint32(s.currentHeight)
        if fwd.packet.incomingTimeout >= height+s.cltvInterceptDelta {
                return false, nil
        }

        log.Debugf("Interception rejected because htlc "+
                "expires too soon: circuit=%v, "+
                "height=%v, incoming_timeout=%v",
                fwd.packet.inKey(), height,
                fwd.packet.incomingTimeout)

        err := fwd.FailWithCode(
                lnwire.CodeExpiryTooSoon,
        )
        if err != nil {
                return false, err
        }

        return true, nil
}

// interceptedForward implements the InterceptedForward interface.
// It is passed from the switch to external interceptors that are interested
// in holding forwards and resolve them manually.
type interceptedForward struct {
        htlc           *lnwire.UpdateAddHTLC
        packet         *htlcPacket
        htlcSwitch     *Switch
        autoFailHeight int32
}

// Packet returns the intercepted htlc packet.
func (f *interceptedForward) Packet() InterceptedPacket {
        return InterceptedPacket{
                IncomingCircuit: models.CircuitKey{
                        ChanID: f.packet.incomingChanID,
                        HtlcID: f.packet.incomingHTLCID,
                },
                OutgoingChanID: f.packet.outgoingChanID,
                Hash:           f.htlc.PaymentHash,
                OutgoingExpiry: f.htlc.Expiry,
                OutgoingAmount: f.htlc.Amount,
                IncomingAmount: f.packet.incomingAmount,
                IncomingExpiry: f.packet.incomingTimeout,
                CustomRecords:  f.packet.customRecords,
                OnionBlob:      f.htlc.OnionBlob,
                AutoFailHeight: f.autoFailHeight,
        }
}

// Resume resumes the default behavior as if the packet was not intercepted.
func (f *interceptedForward) Resume() error {
        // Forward to the switch. A link quit channel isn't needed, because we
        // are on a different thread now.
        return f.htlcSwitch.ForwardPackets(nil, f.packet)
}

// Fail notifies the intention to Fail an existing hold forward with an
// encrypted failure reason.
func (f *interceptedForward) Fail(reason []byte) error {
        obfuscatedReason := f.packet.obfuscator.IntermediateEncrypt(reason)

        return f.resolve(&lnwire.UpdateFailHTLC{
                Reason: obfuscatedReason,
        })
}

// FailWithCode notifies the intention to fail an existing hold forward with the
// specified failure code.
func (f *interceptedForward) FailWithCode(code lnwire.FailCode) error {
        shaOnionBlob := func() [32]byte {
                return sha256.Sum256(f.htlc.OnionBlob[:])
        }

        // Create a local failure.
        var failureMsg lnwire.FailureMessage

        switch code {
        case lnwire.CodeInvalidOnionVersion:
                failureMsg = &lnwire.FailInvalidOnionVersion{
                        OnionSHA256: shaOnionBlob(),
                }

        case lnwire.CodeInvalidOnionHmac:
                failureMsg = &lnwire.FailInvalidOnionHmac{
                        OnionSHA256: shaOnionBlob(),
                }

        case lnwire.CodeInvalidOnionKey:
                failureMsg = &lnwire.FailInvalidOnionKey{
                        OnionSHA256: shaOnionBlob(),
                }

        case lnwire.CodeTemporaryChannelFailure:
                update := f.htlcSwitch.failAliasUpdate(
                        f.packet.incomingChanID, true,
                )
                if update == nil {
                        // Fallback to the original, non-alias behavior.
                        var err error
                        update, err = f.htlcSwitch.cfg.FetchLastChannelUpdate(
                                f.packet.incomingChanID,
                        )
                        if err != nil {
                                return err
                        }
                }

                failureMsg = lnwire.NewTemporaryChannelFailure(update)

        case lnwire.CodeExpiryTooSoon:
                update, err := f.htlcSwitch.cfg.FetchLastChannelUpdate(
                        f.packet.incomingChanID,
                )
                if err != nil {
                        return err
                }

                failureMsg = lnwire.NewExpiryTooSoon(*update)

        default:
                return ErrUnsupportedFailureCode
        }

        // Encrypt the failure for the first hop. This node will be the origin
        // of the failure.
        reason, err := f.packet.obfuscator.EncryptFirstHop(failureMsg)
        if err != nil {
                return fmt.Errorf("failed to encrypt failure reason %w", err)
        }

        return f.resolve(&lnwire.UpdateFailHTLC{
                Reason: reason,
        })
}

// Settle forwards a settled packet to the switch.
func (f *interceptedForward) Settle(preimage lntypes.Preimage) error {
        if !preimage.Matches(f.htlc.PaymentHash) {
                return errors.New("preimage does not match hash")
        }
        return f.resolve(&lnwire.UpdateFulfillHTLC{
                PaymentPreimage: preimage,
        })
}

// resolve is used for both Settle and Fail and forwards the message to the
// switch.
func (f *interceptedForward) resolve(message lnwire.Message) error {
        pkt := &htlcPacket{
                incomingChanID: f.packet.incomingChanID,
                incomingHTLCID: f.packet.incomingHTLCID,
                outgoingChanID: f.packet.outgoingChanID,
                outgoingHTLCID: f.packet.outgoingHTLCID,
                isResolution:   true,
                circuit:        f.packet.circuit,
                htlc:           message,
                obfuscator:     f.packet.obfuscator,
                sourceRef:      f.packet.sourceRef,
        }
        return f.htlcSwitch.mailOrchestrator.Deliver(pkt.incomingChanID, pkt)
}

lightningnetwork / lnd / 9915780197

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