13211764208

Committed 08 Feb 2025 03:08AM UTC coverage: 49.288% (-9.5%) from 58.815%

Build # 13211764208

Build Type

Pull #9489

github

Committed by

calvinrzachman

Commit Message

itest: verify switchrpc server enforces send then track

We prevent the rpc server from allowing onion dispatches for
attempt IDs which have already been tracked by rpc clients.

This helps protect the client from leaking a duplicate onion
attempt. NOTE: This is not the only method for solving this
issue! The issue could be addressed via careful client side
programming which accounts for the uncertainty and async
nature of dispatching onions to a remote process via RPC.
This would require some lnd ChannelRouter changes for how
we intend to use these RPCs though.

Pull Request Pull Request #9489: multi: add BuildOnion, SendOnion, and TrackOnion RPCs

Run Details

474 of 990 new or added lines in 11 files covered. (47.88%)

27321 existing lines in 435 files now uncovered.

101192 of 205306 relevant lines covered (49.29%)

1.54 hits per line

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

35.61

/peer/ping_manager.go

package peer

import (
        "errors"
        "fmt"
        "sync"
        "sync/atomic"
        "time"

        "github.com/lightningnetwork/lnd/lnwire"
)

// PingManagerConfig is a structure containing various parameters that govern
// how the PingManager behaves.
type PingManagerConfig struct {
        // NewPingPayload is a closure that returns the payload to be packaged
        // in the Ping message.
        NewPingPayload func() []byte

        // NewPongSize is a closure that returns a random value between
        // [0, lnwire.MaxPongBytes]. This random value helps to more effectively
        // pair Pong messages with Ping.
        NewPongSize func() uint16

        // IntervalDuration is the Duration between attempted pings.
        IntervalDuration time.Duration

        // TimeoutDuration is the Duration we wait before declaring a ping
        // attempt failed.
        TimeoutDuration time.Duration

        // SendPing is a closure that is responsible for sending the Ping
        // message out to our peer
        SendPing func(ping *lnwire.Ping)

        // OnPongFailure is a closure that is responsible for executing the
        // logic when a Pong message is either late or does not match our
        // expectations for that Pong
        OnPongFailure func(error)
}

// PingManager is a structure that is designed to manage the internal state
// of the ping pong lifecycle with the remote peer. We assume there is only one
// ping outstanding at once.
//
// NOTE: This structure MUST be initialized with NewPingManager.
type PingManager struct {
        cfg *PingManagerConfig

        // pingTime is a rough estimate of the RTT (round-trip-time) between us
        // and the connected peer.
        // To be used atomically.
        // TODO(roasbeef): also use a WMA or EMA?
        pingTime atomic.Pointer[time.Duration]

        // pingLastSend is the time when we sent our last ping message.
        // To be used atomically.
        pingLastSend *time.Time

        // outstandingPongSize is the current size of the requested pong
        // payload.  This value can only validly range from [0,65531]. Any
        // value < 0 is interpreted as if there is no outstanding ping message.
        outstandingPongSize int32

        // pingTicker is a pointer to a Ticker that fires on every ping
        // interval.
        pingTicker *time.Ticker

        // pingTimeout is a Timer that will fire when we want to time out a
        // ping
        pingTimeout *time.Timer

        // pongChan is the channel on which the pingManager will write Pong
        // messages it is evaluating
        pongChan chan *lnwire.Pong

        started sync.Once
        stopped sync.Once

        quit chan struct{}
        wg   sync.WaitGroup
}

// NewPingManager constructs a pingManager in a valid state. It must be started
// before it does anything useful, though.
func NewPingManager(cfg *PingManagerConfig) *PingManager {
        m := PingManager{
                cfg:                 cfg,
                outstandingPongSize: -1,
                pongChan:            make(chan *lnwire.Pong, 1),
                quit:                make(chan struct{}),
        }

        return &m
}

// Start launches the primary goroutine that is owned by the pingManager.
func (m *PingManager) Start() error {
        var err error
        m.started.Do(func() {
                m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)
                m.pingTimeout = time.NewTimer(0)

                m.wg.Add(1)
                go m.pingHandler()
        })

        return err
}

// pingHandler is the main goroutine responsible for enforcing the ping/pong
// protocol.
func (m *PingManager) pingHandler() {
        defer m.wg.Done()
        defer m.pingTimeout.Stop()

        // Ensure that the pingTimeout channel is empty.
        if !m.pingTimeout.Stop() {
                <-m.pingTimeout.C
        }

        for {
                select {
                case <-m.pingTicker.C:
                        // If this occurs it means that the new ping cycle has
                        // begun while there is still an outstanding ping
                        // awaiting a pong response.  This should never occur,
                        // but if it does, it implies a timeout.
                        if m.outstandingPongSize >= 0 {
                                e := errors.New("impossible: new ping" +
                                        "in unclean state",
                                )
                                m.cfg.OnPongFailure(e)

                                return
                        }

                        pongSize := m.cfg.NewPongSize()
                        ping := &lnwire.Ping{
                                NumPongBytes: pongSize,
                                PaddingBytes: m.cfg.NewPingPayload(),
                        }

                        // Set up our bookkeeping for the new Ping.
                        if err := m.setPingState(pongSize); err != nil {
                                m.cfg.OnPongFailure(err)

                                return
                        }

                        m.cfg.SendPing(ping)

                case <-m.pingTimeout.C:
                        m.resetPingState()

                        e := errors.New("timeout while waiting for " +
                                "pong response",
                        )

                        m.cfg.OnPongFailure(e)

                        return

                case pong := <-m.pongChan:
                        pongSize := int32(len(pong.PongBytes))

                        // Save off values we are about to override when we
                        // call resetPingState.
                        expected := m.outstandingPongSize
                        lastPing := m.pingLastSend

                        m.resetPingState()

                        // If the pong we receive doesn't match the ping we
                        // sent out, then we fail out.
                        if pongSize != expected {
                                e := errors.New("pong response does " +
                                        "not match expected size",
                                )

                                m.cfg.OnPongFailure(e)

                                return
                        }

                        // Compute RTT of ping and save that for future
                        // querying.
                        if lastPing != nil {
                                rtt := time.Since(*lastPing)
                                m.pingTime.Store(&rtt)
                        }

                case <-m.quit:
                        return
                }
        }
}

// Stop interrupts the goroutines that the PingManager owns.
func (m *PingManager) Stop() {
        if m.pingTicker == nil {
                return
        }

        m.stopped.Do(func() {
                close(m.quit)
                m.wg.Wait()

                m.pingTicker.Stop()
                m.pingTimeout.Stop()
        })
}

// setPingState is a private method to keep track of all of the fields we need
// to set when we send out a Ping.
func (m *PingManager) setPingState(pongSize uint16) error {
        t := time.Now()
        m.pingLastSend = &t
        m.outstandingPongSize = int32(pongSize)
        if m.pingTimeout.Reset(m.cfg.TimeoutDuration) {
                return fmt.Errorf(
                        "impossible: ping timeout reset when already active",
                )
        }

        return nil
}

// resetPingState is a private method that resets all of the bookkeeping that
// is tracking a currently outstanding Ping.
func (m *PingManager) resetPingState() {
        m.pingLastSend = nil
        m.outstandingPongSize = -1
        if !m.pingTimeout.Stop() {
                select {
                case <-m.pingTimeout.C:
                default:
                }
        }
}

// GetPingTimeMicroSeconds reports back the RTT calculated by the pingManager.
func (m *PingManager) GetPingTimeMicroSeconds() int64 {
        rtt := m.pingTime.Load()

        if rtt == nil {
                return -1
        }

        return rtt.Microseconds()
}

// ReceivedPong is called to evaluate a Pong message against the expectations
// we have for it. It will cause the PingManager to invoke the supplied
// OnPongFailure function if the Pong argument supplied violates expectations.
func (m *PingManager) ReceivedPong(msg *lnwire.Pong) {
        select {
        case m.pongChan <- msg:
        case <-m.quit:
        }
}

1	package peer
2
3	import (
4	"errors"
5	"fmt"
6	"sync"
7	"sync/atomic"
8	"time"
9
10	"github.com/lightningnetwork/lnd/lnwire"
11	)
12
13	// PingManagerConfig is a structure containing various parameters that govern
14	// how the PingManager behaves.
15	type PingManagerConfig struct {
16	// NewPingPayload is a closure that returns the payload to be packaged
17	// in the Ping message.
18	NewPingPayload func() []byte
19
20	// NewPongSize is a closure that returns a random value between
21	// [0, lnwire.MaxPongBytes]. This random value helps to more effectively
22	// pair Pong messages with Ping.
23	NewPongSize func() uint16
24
25	// IntervalDuration is the Duration between attempted pings.
26	IntervalDuration time.Duration
27
28	// TimeoutDuration is the Duration we wait before declaring a ping
29	// attempt failed.
30	TimeoutDuration time.Duration
31
32	// SendPing is a closure that is responsible for sending the Ping
33	// message out to our peer
34	SendPing func(ping *lnwire.Ping)
35
36	// OnPongFailure is a closure that is responsible for executing the
37	// logic when a Pong message is either late or does not match our
38	// expectations for that Pong
39	OnPongFailure func(error)
40	}
41
42	// PingManager is a structure that is designed to manage the internal state
43	// of the ping pong lifecycle with the remote peer. We assume there is only one
44	// ping outstanding at once.
45	//
46	// NOTE: This structure MUST be initialized with NewPingManager.
47	type PingManager struct {
48	cfg *PingManagerConfig
49
50	// pingTime is a rough estimate of the RTT (round-trip-time) between us
51	// and the connected peer.
52	// To be used atomically.
53	// TODO(roasbeef): also use a WMA or EMA?
54	pingTime atomic.Pointer[time.Duration]
55
56	// pingLastSend is the time when we sent our last ping message.
57	// To be used atomically.
58	pingLastSend *time.Time
59
60	// outstandingPongSize is the current size of the requested pong
61	// payload. This value can only validly range from [0,65531]. Any
62	// value < 0 is interpreted as if there is no outstanding ping message.
63	outstandingPongSize int32
64
65	// pingTicker is a pointer to a Ticker that fires on every ping
66	// interval.
67	pingTicker *time.Ticker
68
69	// pingTimeout is a Timer that will fire when we want to time out a
70	// ping
71	pingTimeout *time.Timer
72
73	// pongChan is the channel on which the pingManager will write Pong
74	// messages it is evaluating
75	pongChan chan *lnwire.Pong
76
77	started sync.Once
78	stopped sync.Once
79
80	quit chan struct{}
81	wg sync.WaitGroup
82	}
83
84	// NewPingManager constructs a pingManager in a valid state. It must be started
85	// before it does anything useful, though.
86	func NewPingManager(cfg PingManagerConfig) PingManager {	3✔
87	m := PingManager{	3✔
88	cfg: cfg,	3✔
89	outstandingPongSize: -1,	3✔
90	pongChan: make(chan *lnwire.Pong, 1),	3✔
91	quit: make(chan struct{}),	3✔
92	}	3✔
93		3✔
94	return &m	3✔
95	}	3✔
96
97	// Start launches the primary goroutine that is owned by the pingManager.
98	func (m *PingManager) Start() error {	3✔
99	var err error	3✔
100	m.started.Do(func() {	6✔
101	m.pingTicker = time.NewTicker(m.cfg.IntervalDuration)	3✔
102	m.pingTimeout = time.NewTimer(0)	3✔
103		3✔
104	m.wg.Add(1)	3✔
105	go m.pingHandler()	3✔
106	})	3✔
107
108	return err	3✔
109	}
110
111	// pingHandler is the main goroutine responsible for enforcing the ping/pong
112	// protocol.
113	func (m *PingManager) pingHandler() {	3✔
114	defer m.wg.Done()	3✔
115	defer m.pingTimeout.Stop()	3✔
116		3✔
117	// Ensure that the pingTimeout channel is empty.	3✔
118	if !m.pingTimeout.Stop() {	6✔
119	<-m.pingTimeout.C	3✔
120	}	3✔
121
122	for {	6✔
123	select {	3✔
UNCOV 124	case <-m.pingTicker.C:	×
UNCOV 125	// If this occurs it means that the new ping cycle has	×
UNCOV 126	// begun while there is still an outstanding ping	×
UNCOV 127	// awaiting a pong response. This should never occur,	×
UNCOV 128	// but if it does, it implies a timeout.	×
UNCOV 129	if m.outstandingPongSize >= 0 {	×
130	e := errors.New("impossible: new ping" +	×
131	"in unclean state",	×
132	)	×
133	m.cfg.OnPongFailure(e)	×
134		×
135	return	×
136	}	×
137
UNCOV 138	pongSize := m.cfg.NewPongSize()	×
UNCOV 139	ping := &lnwire.Ping{	×
UNCOV 140	NumPongBytes: pongSize,	×
UNCOV 141	PaddingBytes: m.cfg.NewPingPayload(),	×
UNCOV 142	}	×
UNCOV 143		×
UNCOV 144	// Set up our bookkeeping for the new Ping.	×
UNCOV 145	if err := m.setPingState(pongSize); err != nil {	×
146	m.cfg.OnPongFailure(err)	×
147		×
148	return	×
149	}	×
150
UNCOV 151	m.cfg.SendPing(ping)	×
152
UNCOV 153	case <-m.pingTimeout.C:	×
UNCOV 154	m.resetPingState()	×
UNCOV 155		×
UNCOV 156	e := errors.New("timeout while waiting for " +	×
UNCOV 157	"pong response",	×
UNCOV 158	)	×
UNCOV 159		×
UNCOV 160	m.cfg.OnPongFailure(e)	×
UNCOV 161		×
UNCOV 162	return	×
163
UNCOV 164	case pong := <-m.pongChan:	×
UNCOV 165	pongSize := int32(len(pong.PongBytes))	×
UNCOV 166		×
UNCOV 167	// Save off values we are about to override when we	×
UNCOV 168	// call resetPingState.	×
UNCOV 169	expected := m.outstandingPongSize	×
UNCOV 170	lastPing := m.pingLastSend	×
UNCOV 171		×
UNCOV 172	m.resetPingState()	×
UNCOV 173		×
UNCOV 174	// If the pong we receive doesn't match the ping we	×
UNCOV 175	// sent out, then we fail out.	×
UNCOV 176	if pongSize != expected {	×
UNCOV 177	e := errors.New("pong response does " +	×
UNCOV 178	"not match expected size",	×
UNCOV 179	)	×
UNCOV 180		×
UNCOV 181	m.cfg.OnPongFailure(e)	×
UNCOV 182		×
UNCOV 183	return	×
UNCOV 184	}	×
185
186	// Compute RTT of ping and save that for future
187	// querying.
UNCOV 188	if lastPing != nil {	×
UNCOV 189	rtt := time.Since(*lastPing)	×
UNCOV 190	m.pingTime.Store(&rtt)	×
UNCOV 191	}	×
192
193	case <-m.quit:	3✔
194	return	3✔
195	}
196	}
197	}
198
199	// Stop interrupts the goroutines that the PingManager owns.
200	func (m *PingManager) Stop() {	3✔
201	if m.pingTicker == nil {	3✔
202	return	×
203	}	×
204
205	m.stopped.Do(func() {	6✔
206	close(m.quit)	3✔
207	m.wg.Wait()	3✔
208		3✔
209	m.pingTicker.Stop()	3✔
210	m.pingTimeout.Stop()	3✔
211	})	3✔
212	}
213
214	// setPingState is a private method to keep track of all of the fields we need
215	// to set when we send out a Ping.
UNCOV 216	func (m *PingManager) setPingState(pongSize uint16) error {	×
UNCOV 217	t := time.Now()	×
UNCOV 218	m.pingLastSend = &t	×
UNCOV 219	m.outstandingPongSize = int32(pongSize)	×
UNCOV 220	if m.pingTimeout.Reset(m.cfg.TimeoutDuration) {	×
221	return fmt.Errorf(	×
222	"impossible: ping timeout reset when already active",	×
223	)	×
224	}	×
225
UNCOV 226	return nil	×
227	}
228
229	// resetPingState is a private method that resets all of the bookkeeping that
230	// is tracking a currently outstanding Ping.
UNCOV 231	func (m *PingManager) resetPingState() {	×
UNCOV 232	m.pingLastSend = nil	×
UNCOV 233	m.outstandingPongSize = -1	×
UNCOV 234	if !m.pingTimeout.Stop() {	×
UNCOV 235	select {	×
236	case <-m.pingTimeout.C:	×
UNCOV 237	default:	×
238	}
239	}
240	}
241
242	// GetPingTimeMicroSeconds reports back the RTT calculated by the pingManager.
243	func (m *PingManager) GetPingTimeMicroSeconds() int64 {	3✔
244	rtt := m.pingTime.Load()	3✔
245		3✔
246	if rtt == nil {	6✔
247	return -1	3✔
248	}	3✔
249
250	return rtt.Microseconds()	×
251	}
252
253	// ReceivedPong is called to evaluate a Pong message against the expectations
254	// we have for it. It will cause the PingManager to invoke the supplied
255	// OnPongFailure function if the Pong argument supplied violates expectations.
UNCOV 256	func (m PingManager) ReceivedPong(msg lnwire.Pong) {	×
UNCOV 257	select {	×
UNCOV 258	case m.pongChan <- msg:	×
259	case <-m.quit:	×
260	}
261	}

lightningnetwork / lnd / 13211764208

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