lightningnetwork / lnd / 12313002221

package peer

import (

        "bytes"

        "container/list"

        "errors"

        "fmt"

        "math/rand"

        "net"

        "strings"

        "sync"

        "sync/atomic"

        "time"

        "github.com/btcsuite/btcd/btcec/v2"

        "github.com/btcsuite/btcd/chaincfg/chainhash"

        "github.com/btcsuite/btcd/connmgr"

        "github.com/btcsuite/btcd/txscript"

        "github.com/btcsuite/btcd/wire"

        "github.com/btcsuite/btclog/v2"

        "github.com/davecgh/go-spew/spew"

        "github.com/lightningnetwork/lnd/buffer"

        "github.com/lightningnetwork/lnd/chainntnfs"

        "github.com/lightningnetwork/lnd/channeldb"

        "github.com/lightningnetwork/lnd/channelnotifier"

        "github.com/lightningnetwork/lnd/contractcourt"

        "github.com/lightningnetwork/lnd/discovery"

        "github.com/lightningnetwork/lnd/feature"

        "github.com/lightningnetwork/lnd/fn/v2"

        "github.com/lightningnetwork/lnd/funding"

        graphdb "github.com/lightningnetwork/lnd/graph/db"

        "github.com/lightningnetwork/lnd/graph/db/models"

        "github.com/lightningnetwork/lnd/htlcswitch"

        "github.com/lightningnetwork/lnd/htlcswitch/hodl"

        "github.com/lightningnetwork/lnd/htlcswitch/hop"

        "github.com/lightningnetwork/lnd/input"

        "github.com/lightningnetwork/lnd/invoices"

        "github.com/lightningnetwork/lnd/keychain"

        "github.com/lightningnetwork/lnd/lnpeer"

        "github.com/lightningnetwork/lnd/lntypes"

        "github.com/lightningnetwork/lnd/lnutils"

        "github.com/lightningnetwork/lnd/lnwallet"

        "github.com/lightningnetwork/lnd/lnwallet/chainfee"

        "github.com/lightningnetwork/lnd/lnwallet/chancloser"

        "github.com/lightningnetwork/lnd/lnwire"

        "github.com/lightningnetwork/lnd/msgmux"

        "github.com/lightningnetwork/lnd/netann"

        "github.com/lightningnetwork/lnd/pool"

        "github.com/lightningnetwork/lnd/queue"

        "github.com/lightningnetwork/lnd/subscribe"

        "github.com/lightningnetwork/lnd/ticker"

        "github.com/lightningnetwork/lnd/tlv"

        "github.com/lightningnetwork/lnd/watchtower/wtclient"

)

const (

        // pingInterval is the interval at which ping messages are sent.

        pingInterval = 1 * time.Minute

        // pingTimeout is the amount of time we will wait for a pong response

        // before considering the peer to be unresponsive.

        //

        // This MUST be a smaller value than the pingInterval.

        pingTimeout = 30 * time.Second

        // idleTimeout is the duration of inactivity before we time out a peer.

        idleTimeout = 5 * time.Minute

        // writeMessageTimeout is the timeout used when writing a message to the

        // peer.

        writeMessageTimeout = 5 * time.Second

        // readMessageTimeout is the timeout used when reading a message from a

        // peer.

        readMessageTimeout = 5 * time.Second

        // handshakeTimeout is the timeout used when waiting for the peer's init

        // message.

        handshakeTimeout = 15 * time.Second

        // ErrorBufferSize is the number of historic peer errors that we store.

        ErrorBufferSize = 10

        // pongSizeCeiling is the upper bound on a uniformly distributed random

        // variable that we use for requesting pong responses. We don't use the

        // MaxPongBytes (upper bound accepted by the protocol) because it is

        // needlessly wasteful of precious Tor bandwidth for little to no gain.

        pongSizeCeiling = 4096

        // torTimeoutMultiplier is the scaling factor we use on network timeouts

        // for Tor peers.

        torTimeoutMultiplier = 3

)

var (

        // ErrChannelNotFound is an error returned when a channel is queried and

        // either the Brontide doesn't know of it, or the channel in question

        // is pending.

        ErrChannelNotFound = fmt.Errorf("channel not found")

)

// outgoingMsg packages an lnwire.Message to be sent out on the wire, along with

// a buffered channel which will be sent upon once the write is complete. This

// buffered channel acts as a semaphore to be used for synchronization purposes.

type outgoingMsg struct {

        priority bool

        msg      lnwire.Message

        errChan  chan error // MUST be buffered.

}

// newChannelMsg packages a channeldb.OpenChannel with a channel that allows

// the receiver of the request to report when the channel creation process has

// completed.

type newChannelMsg struct {

        // channel is used when the pending channel becomes active.

        channel *lnpeer.NewChannel

        // channelID is used when there's a new pending channel.

        channelID lnwire.ChannelID

        err chan error

}

type customMsg struct {

        peer [33]byte

        msg  lnwire.Custom

}

// closeMsg is a wrapper struct around any wire messages that deal with the

// cooperative channel closure negotiation process. This struct includes the

// raw channel ID targeted along with the original message.

type closeMsg struct {

        cid lnwire.ChannelID

        msg lnwire.Message

}

// PendingUpdate describes the pending state of a closing channel.

type PendingUpdate struct {

        Txid        []byte

        OutputIndex uint32

}

// ChannelCloseUpdate contains the outcome of the close channel operation.

type ChannelCloseUpdate struct {

        ClosingTxid []byte

        Success     bool

        // LocalCloseOutput is an optional, additional output on the closing

        // transaction that the local party should be paid to. This will only be

        // populated if the local balance isn't dust.

        LocalCloseOutput fn.Option[chancloser.CloseOutput]

        // RemoteCloseOutput is an optional, additional output on the closing

        // transaction that the remote party should be paid to. This will only

        // be populated if the remote balance isn't dust.

        RemoteCloseOutput fn.Option[chancloser.CloseOutput]

        // AuxOutputs is an optional set of additional outputs that might be

        // included in the closing transaction. These are used for custom

        // channel types.

        AuxOutputs fn.Option[chancloser.AuxCloseOutputs]

}

// TimestampedError is a timestamped error that is used to store the most recent

// errors we have experienced with our peers.

type TimestampedError struct {

        Error     error

        Timestamp time.Time

}

// Config defines configuration fields that are necessary for a peer object

// to function.

type Config struct {

        // Conn is the underlying network connection for this peer.

        Conn MessageConn

        // ConnReq stores information related to the persistent connection request

        // for this peer.

        ConnReq *connmgr.ConnReq

        // PubKeyBytes is the serialized, compressed public key of this peer.

        PubKeyBytes [33]byte

        // Addr is the network address of the peer.

        Addr *lnwire.NetAddress

        // Inbound indicates whether or not the peer is an inbound peer.

        Inbound bool

        // Features is the set of features that we advertise to the remote party.

        Features *lnwire.FeatureVector

        // LegacyFeatures is the set of features that we advertise to the remote

        // peer for backwards compatibility. Nodes that have not implemented

        // flat features will still be able to read our feature bits from the

        // legacy global field, but we will also advertise everything in the

        // default features field.

        LegacyFeatures *lnwire.FeatureVector

        // OutgoingCltvRejectDelta defines the number of blocks before expiry of

        // an htlc where we don't offer it anymore.

        OutgoingCltvRejectDelta uint32

        // ChanActiveTimeout specifies the duration the peer will wait to request

        // a channel reenable, beginning from the time the peer was started.

        ChanActiveTimeout time.Duration

        // ErrorBuffer stores a set of errors related to a peer. It contains error

        // messages that our peer has recently sent us over the wire and records of

        // unknown messages that were sent to us so that we can have a full track

        // record of the communication errors we have had with our peer. If we

        // choose to disconnect from a peer, it also stores the reason we had for

        // disconnecting.

        ErrorBuffer *queue.CircularBuffer

        // WritePool is the task pool that manages reuse of write buffers. Write

        // tasks are submitted to the pool in order to conserve the total number of

        // write buffers allocated at any one time, and decouple write buffer

        // allocation from the peer life cycle.

        WritePool *pool.Write

        // ReadPool is the task pool that manages reuse of read buffers.

        ReadPool *pool.Read

        // Switch is a pointer to the htlcswitch. It is used to setup, get, and

        // tear-down ChannelLinks.

        Switch messageSwitch

        // InterceptSwitch is a pointer to the InterceptableSwitch, a wrapper around

        // the regular Switch. We only export it here to pass ForwardPackets to the

        // ChannelLinkConfig.

        InterceptSwitch *htlcswitch.InterceptableSwitch

        // ChannelDB is used to fetch opened channels, and closed channels.

        ChannelDB *channeldb.ChannelStateDB

        // ChannelGraph is a pointer to the channel graph which is used to

        // query information about the set of known active channels.

        ChannelGraph *graphdb.ChannelGraph

        // ChainArb is used to subscribe to channel events, update contract signals,

        // and force close channels.

        ChainArb *contractcourt.ChainArbitrator

        // AuthGossiper is needed so that the Brontide impl can register with the

        // gossiper and process remote channel announcements.

        AuthGossiper *discovery.AuthenticatedGossiper

        // ChanStatusMgr is used to set or un-set the disabled bit in channel

        // updates.

        ChanStatusMgr *netann.ChanStatusManager

        // ChainIO is used to retrieve the best block.

        ChainIO lnwallet.BlockChainIO

        // FeeEstimator is used to compute our target ideal fee-per-kw when

        // initializing the coop close process.

        FeeEstimator chainfee.Estimator

        // Signer is used when creating *lnwallet.LightningChannel instances.

        Signer input.Signer

        // SigPool is used when creating *lnwallet.LightningChannel instances.

        SigPool *lnwallet.SigPool

        // Wallet is used to publish transactions and generates delivery

        // scripts during the coop close process.

        Wallet *lnwallet.LightningWallet

        // ChainNotifier is used to receive confirmations of a coop close

        // transaction.

        ChainNotifier chainntnfs.ChainNotifier

        // BestBlockView is used to efficiently query for up-to-date

        // blockchain state information

        BestBlockView chainntnfs.BestBlockView

        // RoutingPolicy is used to set the forwarding policy for links created by

        // the Brontide.

        RoutingPolicy models.ForwardingPolicy

        // Sphinx is used when setting up ChannelLinks so they can decode sphinx

        // onion blobs.

        Sphinx *hop.OnionProcessor

        // WitnessBeacon is used when setting up ChannelLinks so they can add any

        // preimages that they learn.

        WitnessBeacon contractcourt.WitnessBeacon

        // Invoices is passed to the ChannelLink on creation and handles all

        // invoice-related logic.

        Invoices *invoices.InvoiceRegistry

        // ChannelNotifier is used by the link to notify other sub-systems about

        // channel-related events and by the Brontide to subscribe to

        // ActiveLinkEvents.

        ChannelNotifier *channelnotifier.ChannelNotifier

        // HtlcNotifier is used when creating a ChannelLink.

        HtlcNotifier *htlcswitch.HtlcNotifier

        // TowerClient is used to backup revoked states.

        TowerClient wtclient.ClientManager

        // DisconnectPeer is used to disconnect this peer if the cooperative close

        // process fails.

        DisconnectPeer func(*btcec.PublicKey) error

        // GenNodeAnnouncement is used to send our node announcement to the remote

        // on startup.

        GenNodeAnnouncement func(...netann.NodeAnnModifier) (

                lnwire.NodeAnnouncement, error)

        // PrunePersistentPeerConnection is used to remove all internal state

        // related to this peer in the server.

        PrunePersistentPeerConnection func([33]byte)

        // FetchLastChanUpdate fetches our latest channel update for a target

        // channel.

        FetchLastChanUpdate func(lnwire.ShortChannelID) (*lnwire.ChannelUpdate1,

                error)

        // FundingManager is an implementation of the funding.Controller interface.

        FundingManager funding.Controller

        // Hodl is used when creating ChannelLinks to specify HodlFlags as

        // breakpoints in dev builds.

        Hodl *hodl.Config

        // UnsafeReplay is used when creating ChannelLinks to specify whether or

        // not to replay adds on its commitment tx.

        UnsafeReplay bool

        // MaxOutgoingCltvExpiry is used when creating ChannelLinks and is the max

        // number of blocks that funds could be locked up for when forwarding

        // payments.

        MaxOutgoingCltvExpiry uint32

        // MaxChannelFeeAllocation is used when creating ChannelLinks and is the

        // maximum percentage of total funds that can be allocated to a channel's

        // commitment fee. This only applies for the initiator of the channel.

        MaxChannelFeeAllocation float64

        // MaxAnchorsCommitFeeRate is the maximum fee rate we'll use as an

        // initiator for anchor channel commitments.

        MaxAnchorsCommitFeeRate chainfee.SatPerKWeight

        // CoopCloseTargetConfs is the confirmation target that will be used

        // to estimate the fee rate to use during a cooperative channel

        // closure initiated by the remote peer.

        CoopCloseTargetConfs uint32

        // ServerPubKey is the serialized, compressed public key of our lnd node.

        // It is used to determine which policy (channel edge) to pass to the

        // ChannelLink.

        ServerPubKey [33]byte

        // ChannelCommitInterval is the maximum time that is allowed to pass between

        // receiving a channel state update and signing the next commitment.

        // Setting this to a longer duration allows for more efficient channel

        // operations at the cost of latency.

        ChannelCommitInterval time.Duration

        // PendingCommitInterval is the maximum time that is allowed to pass

        // while waiting for the remote party to revoke a locally initiated

        // commitment state. Setting this to a longer duration if a slow

        // response is expected from the remote party or large number of

        // payments are attempted at the same time.

        PendingCommitInterval time.Duration

        // ChannelCommitBatchSize is the maximum number of channel state updates

        // that is accumulated before signing a new commitment.

        ChannelCommitBatchSize uint32

        // HandleCustomMessage is called whenever a custom message is received

        // from the peer.

        HandleCustomMessage func(peer [33]byte, msg *lnwire.Custom) error

        // GetAliases is passed to created links so the Switch and link can be

        // aware of the channel's aliases.

        GetAliases func(base lnwire.ShortChannelID) []lnwire.ShortChannelID

        // RequestAlias allows the Brontide struct to request an alias to send

        // to the peer.

        RequestAlias func() (lnwire.ShortChannelID, error)

        // AddLocalAlias persists an alias to an underlying alias store.

        AddLocalAlias func(alias, base lnwire.ShortChannelID,

                gossip, liveUpdate bool) error

        // AuxLeafStore is an optional store that can be used to store auxiliary

        // leaves for certain custom channel types.

        AuxLeafStore fn.Option[lnwallet.AuxLeafStore]

        // AuxSigner is an optional signer that can be used to sign auxiliary

        // leaves for certain custom channel types.

        AuxSigner fn.Option[lnwallet.AuxSigner]

        // AuxResolver is an optional interface that can be used to modify the

        // way contracts are resolved.

        AuxResolver fn.Option[lnwallet.AuxContractResolver]

        // AuxTrafficShaper is an optional auxiliary traffic shaper that can be

        // used to manage the bandwidth of peer links.

        AuxTrafficShaper fn.Option[htlcswitch.AuxTrafficShaper]

        // PongBuf is a slice we'll reuse instead of allocating memory on the

        // heap. Since only reads will occur and no writes, there is no need

        // for any synchronization primitives. As a result, it's safe to share

        // this across multiple Peer struct instances.

        PongBuf []byte

        // Adds the option to disable forwarding payments in blinded routes

        // by failing back any blinding-related payloads as if they were

        // invalid.

        DisallowRouteBlinding bool

        // DisallowQuiescence is a flag that indicates whether the Brontide

        // should have the quiescence feature disabled.

        DisallowQuiescence bool

        // MaxFeeExposure limits the number of outstanding fees in a channel.

        // This value will be passed to created links.

        MaxFeeExposure lnwire.MilliSatoshi

        // MsgRouter is an optional instance of the main message router that

        // the peer will use. If None, then a new default version will be used

        // in place.

        MsgRouter fn.Option[msgmux.Router]

        // AuxChanCloser is an optional instance of an abstraction that can be

        // used to modify the way the co-op close transaction is constructed.

        AuxChanCloser fn.Option[chancloser.AuxChanCloser]

        // ShouldFwdExpEndorsement is a closure that indicates whether

        // experimental endorsement signals should be set.

        ShouldFwdExpEndorsement func() bool

        // Quit is the server's quit channel. If this is closed, we halt operation.

        Quit chan struct{}

}

// Brontide is an active peer on the Lightning Network. This struct is responsible

// for managing any channel state related to this peer. To do so, it has

// several helper goroutines to handle events such as HTLC timeouts, new

// funding workflow, and detecting an uncooperative closure of any active

// channels.

// TODO(roasbeef): proper reconnection logic.

type Brontide struct {

        // MUST be used atomically.

        started    int32

        disconnect int32

        // MUST be used atomically.

        bytesReceived uint64

        bytesSent     uint64

        // isTorConnection is a flag that indicates whether or not we believe

        // the remote peer is a tor connection. It is not always possible to

        // know this with certainty but we have heuristics we use that should

        // catch most cases.

        //

        // NOTE: We judge the tor-ness of a connection by if the remote peer has

        // ".onion" in the address OR if it's connected over localhost.

        // This will miss cases where our peer is connected to our clearnet

        // address over the tor network (via exit nodes). It will also misjudge

        // actual localhost connections as tor. We need to include this because

        // inbound connections to our tor address will appear to come from the

        // local socks5 proxy. This heuristic is only used to expand the timeout

        // window for peers so it is OK to misjudge this. If you use this field

        // for any other purpose you should seriously consider whether or not

        // this heuristic is good enough for your use case.

        isTorConnection bool

        pingManager *PingManager

        // lastPingPayload stores an unsafe pointer wrapped as an atomic

        // variable which points to the last payload the remote party sent us

        // as their ping.

        //

        // MUST be used atomically.

        lastPingPayload atomic.Value

        cfg Config

        // activeSignal when closed signals that the peer is now active and

        // ready to process messages.

        activeSignal chan struct{}

        // startTime is the time this peer connection was successfully established.

        // It will be zero for peers that did not successfully call Start().

        startTime time.Time

        // sendQueue is the channel which is used to queue outgoing messages to be

        // written onto the wire. Note that this channel is unbuffered.

        sendQueue chan outgoingMsg

        // outgoingQueue is a buffered channel which allows second/third party

        // objects to queue messages to be sent out on the wire.

        outgoingQueue chan outgoingMsg

        // activeChannels is a map which stores the state machines of all

        // active channels. Channels are indexed into the map by the txid of

        // the funding transaction which opened the channel.

        //

        // NOTE: On startup, pending channels are stored as nil in this map.

        // Confirmed channels have channel data populated in the map. This means

        // that accesses to this map should nil-check the LightningChannel to

        // see if this is a pending channel or not. The tradeoff here is either

        // having two maps everywhere (one for pending, one for confirmed chans)

        // or having an extra nil-check per access.

        activeChannels *lnutils.SyncMap[

                lnwire.ChannelID, *lnwallet.LightningChannel]

        // addedChannels tracks any new channels opened during this peer's

        // lifecycle. We use this to filter out these new channels when the time

        // comes to request a reenable for active channels, since they will have

        // waited a shorter duration.

        addedChannels *lnutils.SyncMap[lnwire.ChannelID, struct{}]

        // newActiveChannel is used by the fundingManager to send fully opened

        // channels to the source peer which handled the funding workflow.

        newActiveChannel chan *newChannelMsg

        // newPendingChannel is used by the fundingManager to send pending open

        // channels to the source peer which handled the funding workflow.

        newPendingChannel chan *newChannelMsg

        // removePendingChannel is used by the fundingManager to cancel pending

        // open channels to the source peer when the funding flow is failed.

        removePendingChannel chan *newChannelMsg

        // activeMsgStreams is a map from channel id to the channel streams that

        // proxy messages to individual, active links.

        activeMsgStreams map[lnwire.ChannelID]*msgStream

        // activeChanCloses is a map that keeps track of all the active

        // cooperative channel closures. Any channel closing messages are directed

        // to one of these active state machines. Once the channel has been closed,

        // the state machine will be deleted from the map.

        activeChanCloses map[lnwire.ChannelID]*chancloser.ChanCloser

        // localCloseChanReqs is a channel in which any local requests to close

        // a particular channel are sent over.

        localCloseChanReqs chan *htlcswitch.ChanClose

        // linkFailures receives all reported channel failures from the switch,

        // and instructs the channelManager to clean remaining channel state.

        linkFailures chan linkFailureReport

        // chanCloseMsgs is a channel that any message related to channel

        // closures are sent over. This includes lnwire.Shutdown message as

        // well as lnwire.ClosingSigned messages.

        chanCloseMsgs chan *closeMsg

        // remoteFeatures is the feature vector received from the peer during

        // the connection handshake.

        remoteFeatures *lnwire.FeatureVector

        // resentChanSyncMsg is a set that keeps track of which channels we

        // have re-sent channel reestablishment messages for. This is done to

        // avoid getting into loop where both peers will respond to the other

        // peer's chansync message with its own over and over again.

        resentChanSyncMsg map[lnwire.ChannelID]struct{}

        // channelEventClient is the channel event subscription client that's

        // used to assist retry enabling the channels. This client is only

        // created when the reenableTimeout is no greater than 1 minute. Once

        // created, it is canceled once the reenabling has been finished.

        //

        // NOTE: we choose to create the client conditionally to avoid

        // potentially holding lots of un-consumed events.

        channelEventClient *subscribe.Client

        // msgRouter is an instance of the msgmux.Router which is used to send

        // off new wire messages for handing.

        msgRouter fn.Option[msgmux.Router]

        // globalMsgRouter is a flag that indicates whether we have a global

        // msg router. If so, then we don't worry about stopping the msg router

        // when a peer disconnects.

        globalMsgRouter bool

        startReady chan struct{}

        quit       chan struct{}

        wg         sync.WaitGroup

        // log is a peer-specific logging instance.

        log btclog.Logger

}

// A compile-time check to ensure that Brontide satisfies the lnpeer.Peer interface.

var _ lnpeer.Peer = (*Brontide)(nil)

// NewBrontide creates a new Brontide from a peer.Config struct.

        }

        // TODO(roasbeef): make dynamic in order to create fake cover traffic.

        //

        // NOTE(proofofkeags): this was changed to be dynamic to allow better

        // pong identification, however, more thought is needed to make this

        // actually usable as a traffic decoy.

        })

}

// Start starts all helper goroutines the peer needs for normal operations.  In

// the case this peer has already been started, then this function is a noop.

        // Once we've finished starting up the peer, we'll signal to other

        // goroutines that the they can move forward to tear down the peer, or

        // carry out other relevant changes.

        // Quickly check if we have any existing legacy channels with this

        // peer.

                }

        }

        // Exchange local and global features, the init message should be very

        // first between two nodes.

        // Before we launch any of the helper goroutines off the peer struct,

        // we'll first ensure proper adherence to the p2p protocol. The init

        // message MUST be sent before any other message.

        }()

        // In order to avoid blocking indefinitely, we'll give the other peer

        // an upper timeout to respond before we bail out early.

        }

        // Once the init message arrives, we can parse it so we can figure out

        // the negotiation of features for this session.

        // Next, load all the active channels we have with this peer,

        // registering them with the switch and launching the necessary

        // goroutines required to operate them.

        // Register the message router now as we may need to register some

        // endpoints while loading the channels below.

                }

        }

}

// initGossipSync initializes either a gossip syncer or an initial routing

// dump, depending on the negotiated synchronization method.

                // Register the peer's gossip syncer with the gossiper.

                // This blocks synchronously to ensure the gossip syncer is

                // registered with the gossiper before attempting to read

                // messages from the remote peer.

                //

                // TODO(wilmer): Only sync updates from non-channel peers. This

                // requires an improved version of the current network

                // bootstrapper to ensure we can find and connect to non-channel

                // peers.

        }

}

// taprootShutdownAllowed returns true if both parties have negotiated the

// shutdown-any-segwit feature.

// QuitSignal is a method that should return a channel which will be sent upon

// or closed once the backing peer exits. This allows callers using the

// interface to cancel any processing in the event the backing implementation

// exits.

//

// NOTE: Part of the lnpeer.Peer interface.

// addrWithInternalKey takes a delivery script, then attempts to supplement it

// with information related to the internal key for the addr, but only if it's

// a taproot addr.

func (p *Brontide) addrWithInternalKey(

                )(internalKeyDesc),

        }, nil

}

// loadActiveChannels creates indexes within the peer for tracking all active

// channels returned by the database. It returns a slice of channel reestablish

// messages that should be sent to the peer immediately, in case we have borked

// channels that haven't been closed yet.

func (p *Brontide) loadActiveChannels(chans []*channeldb.OpenChannel) (