Skip to content

Commit 0189722

Browse files
f0rkiMichael RodlerDaniele Ahmed
authored
Fix for a fuzzer-discovered integer underflow of the flow control window size (#692)
Removed the SubAssign, etc. syntactic sugar functions and switched to return Result on over/underflow Whenever possible, switched to returning a library GoAway protocol error. Otherwise we check for over/underflow only with `debug_assert!`, assuming that those code paths do not over/underflow. Signed-off-by: Michael Rodler <[email protected]> Signed-off-by: Daniele Ahmed <[email protected]> Co-authored-by: Michael Rodler <[email protected]> Co-authored-by: Daniele Ahmed <[email protected]>
1 parent 478f7b9 commit 0189722

File tree

9 files changed

+271
-65
lines changed

9 files changed

+271
-65
lines changed

‎src/proto/connection.rs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,9 @@ where
145145

146146
/// connection flow control
147147
pub(crate) fn set_target_window_size(&mut self, size: WindowSize) {
148-
self.inner.streams.set_target_connection_window_size(size);
148+
let _res = self.inner.streams.set_target_connection_window_size(size);
149+
// TODO: proper error handling
150+
debug_assert!(_res.is_ok());
149151
}
150152

151153
/// Send a new SETTINGS frame with an updated initial window size.

‎src/proto/mod.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ pub type PingPayload = [u8; 8];
3030
pub type WindowSize = u32;
3131

3232
// Constants
33-
pub const MAX_WINDOW_SIZE: WindowSize = (1 << 31) - 1;
33+
pub const MAX_WINDOW_SIZE: WindowSize = (1 << 31) - 1; // i32::MAX as u32
3434
pub const DEFAULT_REMOTE_RESET_STREAM_MAX: usize = 20;
3535
pub const DEFAULT_RESET_STREAM_MAX: usize = 10;
3636
pub const DEFAULT_RESET_STREAM_SECS: u64 = 30;

‎src/proto/streams/flow_control.rs

Lines changed: 40 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -75,12 +75,12 @@ impl FlowControl {
7575
self.window_size > self.available
7676
}
7777

78-
pub fn claim_capacity(&mut self, capacity: WindowSize) {
79-
self.available -= capacity;
78+
pub fn claim_capacity(&mut self, capacity: WindowSize) -> Result<(), Reason> {
79+
self.available.decrease_by(capacity)
8080
}
8181

82-
pub fn assign_capacity(&mut self, capacity: WindowSize) {
83-
self.available += capacity;
82+
pub fn assign_capacity(&mut self, capacity: WindowSize) -> Result<(), Reason> {
83+
self.available.increase_by(capacity)
8484
}
8585

8686
/// If a WINDOW_UPDATE frame should be sent, returns a positive number
@@ -136,36 +136,38 @@ impl FlowControl {
136136
///
137137
/// This is called after receiving a SETTINGS frame with a lower
138138
/// INITIAL_WINDOW_SIZE value.
139-
pub fn dec_send_window(&mut self, sz: WindowSize) {
139+
pub fn dec_send_window(&mut self, sz: WindowSize) -> Result<(), Reason> {
140140
tracing::trace!(
141141
"dec_window; sz={}; window={}, available={}",
142142
sz,
143143
self.window_size,
144144
self.available
145145
);
146-
// This should not be able to overflow `window_size` from the bottom.
147-
self.window_size -= sz;
146+
// ~~This should not be able to overflow `window_size` from the bottom.~~ wrong. it can.
147+
self.window_size.decrease_by(sz)?;
148+
Ok(())
148149
}
149150

150151
/// Decrement the recv-side window size.
151152
///
152153
/// This is called after receiving a SETTINGS ACK frame with a lower
153154
/// INITIAL_WINDOW_SIZE value.
154-
pub fn dec_recv_window(&mut self, sz: WindowSize) {
155+
pub fn dec_recv_window(&mut self, sz: WindowSize) -> Result<(), Reason> {
155156
tracing::trace!(
156157
"dec_recv_window; sz={}; window={}, available={}",
157158
sz,
158159
self.window_size,
159160
self.available
160161
);
161162
// This should not be able to overflow `window_size` from the bottom.
162-
self.window_size -= sz;
163-
self.available -= sz;
163+
self.window_size.decrease_by(sz)?;
164+
self.available.decrease_by(sz)?;
165+
Ok(())
164166
}
165167

166168
/// Decrements the window reflecting data has actually been sent. The caller
167169
/// must ensure that the window has capacity.
168-
pub fn send_data(&mut self, sz: WindowSize) {
170+
pub fn send_data(&mut self, sz: WindowSize) -> Result<(), Reason> {
169171
tracing::trace!(
170172
"send_data; sz={}; window={}; available={}",
171173
sz,
@@ -176,12 +178,13 @@ impl FlowControl {
176178
// If send size is zero it's meaningless to update flow control window
177179
if sz > 0 {
178180
// Ensure that the argument is correct
179-
assert!(self.window_size >= sz as usize);
181+
assert!(self.window_size.0 >= sz as i32);
180182

181183
// Update values
182-
self.window_size -= sz;
183-
self.available -= sz;
184+
self.window_size.decrease_by(sz)?;
185+
self.available.decrease_by(sz)?;
184186
}
187+
Ok(())
185188
}
186189
}
187190

@@ -208,6 +211,29 @@ impl Window {
208211
assert!(self.0 >= 0, "negative Window");
209212
self.0 as WindowSize
210213
}
214+
215+
pub fn decrease_by(&mut self, other: WindowSize) -> Result<(), Reason> {
216+
if let Some(v) = self.0.checked_sub(other as i32) {
217+
self.0 = v;
218+
Ok(())
219+
} else {
220+
Err(Reason::FLOW_CONTROL_ERROR)
221+
}
222+
}
223+
224+
pub fn increase_by(&mut self, other: WindowSize) -> Result<(), Reason> {
225+
let other = self.add(other)?;
226+
self.0 = other.0;
227+
Ok(())
228+
}
229+
230+
pub fn add(&self, other: WindowSize) -> Result<Self, Reason> {
231+
if let Some(v) = self.0.checked_add(other as i32) {
232+
Ok(Self(v))
233+
} else {
234+
Err(Reason::FLOW_CONTROL_ERROR)
235+
}
236+
}
211237
}
212238

213239
impl PartialEq<usize> for Window {
@@ -230,25 +256,6 @@ impl PartialOrd<usize> for Window {
230256
}
231257
}
232258

233-
impl ::std::ops::SubAssign<WindowSize> for Window {
234-
fn sub_assign(&mut self, other: WindowSize) {
235-
self.0 -= other as i32;
236-
}
237-
}
238-
239-
impl ::std::ops::Add<WindowSize> for Window {
240-
type Output = Self;
241-
fn add(self, other: WindowSize) -> Self::Output {
242-
Window(self.0 + other as i32)
243-
}
244-
}
245-
246-
impl ::std::ops::AddAssign<WindowSize> for Window {
247-
fn add_assign(&mut self, other: WindowSize) {
248-
self.0 += other as i32;
249-
}
250-
}
251-
252259
impl fmt::Display for Window {
253260
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
254261
fmt::Display::fmt(&self.0, f)

‎src/proto/streams/prioritize.rs

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,9 @@ impl Prioritize {
8787
flow.inc_window(config.remote_init_window_sz)
8888
.expect("invalid initial window size");
8989

90-
flow.assign_capacity(config.remote_init_window_sz);
90+
// TODO: proper error handling
91+
let _res = flow.assign_capacity(config.remote_init_window_sz);
92+
debug_assert!(_res.is_ok());
9193

9294
tracing::trace!("Prioritize::new; flow={:?}", flow);
9395

@@ -253,7 +255,9 @@ impl Prioritize {
253255
if available as usize > capacity {
254256
let diff = available - capacity as WindowSize;
255257

256-
stream.send_flow.claim_capacity(diff);
258+
// TODO: proper error handling
259+
let _res = stream.send_flow.claim_capacity(diff);
260+
debug_assert!(_res.is_ok());
257261

258262
self.assign_connection_capacity(diff, stream, counts);
259263
}
@@ -324,7 +328,9 @@ impl Prioritize {
324328
pub fn reclaim_all_capacity(&mut self, stream: &mut store::Ptr, counts: &mut Counts) {
325329
let available = stream.send_flow.available().as_size();
326330
if available > 0 {
327-
stream.send_flow.claim_capacity(available);
331+
// TODO: proper error handling
332+
let _res = stream.send_flow.claim_capacity(available);
333+
debug_assert!(_res.is_ok());
328334
// Re-assign all capacity to the connection
329335
self.assign_connection_capacity(available, stream, counts);
330336
}
@@ -337,7 +343,9 @@ impl Prioritize {
337343
if stream.requested_send_capacity as usize > stream.buffered_send_data {
338344
let reserved = stream.requested_send_capacity - stream.buffered_send_data as WindowSize;
339345

340-
stream.send_flow.claim_capacity(reserved);
346+
// TODO: proper error handling
347+
let _res = stream.send_flow.claim_capacity(reserved);
348+
debug_assert!(_res.is_ok());
341349
self.assign_connection_capacity(reserved, stream, counts);
342350
}
343351
}
@@ -363,7 +371,9 @@ impl Prioritize {
363371
let span = tracing::trace_span!("assign_connection_capacity", inc);
364372
let _e = span.enter();
365373

366-
self.flow.assign_capacity(inc);
374+
// TODO: proper error handling
375+
let _res = self.flow.assign_capacity(inc);
376+
debug_assert!(_res.is_ok());
367377

368378
// Assign newly acquired capacity to streams pending capacity.
369379
while self.flow.available() > 0 {
@@ -443,7 +453,9 @@ impl Prioritize {
443453
stream.assign_capacity(assign, self.max_buffer_size);
444454

445455
// Claim the capacity from the connection
446-
self.flow.claim_capacity(assign);
456+
// TODO: proper error handling
457+
let _res = self.flow.claim_capacity(assign);
458+
debug_assert!(_res.is_ok());
447459
}
448460

449461
tracing::trace!(
@@ -763,12 +775,16 @@ impl Prioritize {
763775
// Assign the capacity back to the connection that
764776
// was just consumed from the stream in the previous
765777
// line.
766-
self.flow.assign_capacity(len);
778+
// TODO: proper error handling
779+
let _res = self.flow.assign_capacity(len);
780+
debug_assert!(_res.is_ok());
767781
});
768782

769783
let (eos, len) = tracing::trace_span!("updating connection flow")
770784
.in_scope(|| {
771-
self.flow.send_data(len);
785+
// TODO: proper error handling
786+
let _res = self.flow.send_data(len);
787+
debug_assert!(_res.is_ok());
772788

773789
// Wrap the frame's data payload to ensure that the
774790
// correct amount of data gets written.

‎src/proto/streams/recv.rs

Lines changed: 36 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -90,7 +90,7 @@ impl Recv {
9090
// settings
9191
flow.inc_window(DEFAULT_INITIAL_WINDOW_SIZE)
9292
.expect("invalid initial remote window size");
93-
flow.assign_capacity(DEFAULT_INITIAL_WINDOW_SIZE);
93+
flow.assign_capacity(DEFAULT_INITIAL_WINDOW_SIZE).unwrap();
9494

9595
Recv {
9696
init_window_sz: config.local_init_window_sz,
@@ -363,7 +363,9 @@ impl Recv {
363363
self.in_flight_data -= capacity;
364364

365365
// Assign capacity to connection
366-
self.flow.assign_capacity(capacity);
366+
// TODO: proper error handling
367+
let _res = self.flow.assign_capacity(capacity);
368+
debug_assert!(_res.is_ok());
367369

368370
if self.flow.unclaimed_capacity().is_some() {
369371
if let Some(task) = task.take() {
@@ -391,7 +393,9 @@ impl Recv {
391393
stream.in_flight_recv_data -= capacity;
392394

393395
// Assign capacity to stream
394-
stream.recv_flow.assign_capacity(capacity);
396+
// TODO: proper error handling
397+
let _res = stream.recv_flow.assign_capacity(capacity);
398+
debug_assert!(_res.is_ok());
395399

396400
if stream.recv_flow.unclaimed_capacity().is_some() {
397401
// Queue the stream for sending the WINDOW_UPDATE frame.
@@ -437,7 +441,11 @@ impl Recv {
437441
///
438442
/// The `task` is an optional parked task for the `Connection` that might
439443
/// be blocked on needing more window capacity.
440-
pub fn set_target_connection_window(&mut self, target: WindowSize, task: &mut Option<Waker>) {
444+
pub fn set_target_connection_window(
445+
&mut self,
446+
target: WindowSize,
447+
task: &mut Option<Waker>,
448+
) -> Result<(), Reason> {
441449
tracing::trace!(
442450
"set_target_connection_window; target={}; available={}, reserved={}",
443451
target,
@@ -450,11 +458,15 @@ impl Recv {
450458
//
451459
// Update the flow controller with the difference between the new
452460
// target and the current target.
453-
let current = (self.flow.available() + self.in_flight_data).checked_size();
461+
let current = self
462+
.flow
463+
.available()
464+
.add(self.in_flight_data)?
465+
.checked_size();
454466
if target > current {
455-
self.flow.assign_capacity(target - current);
467+
self.flow.assign_capacity(target - current)?;
456468
} else {
457-
self.flow.claim_capacity(current - target);
469+
self.flow.claim_capacity(current - target)?;
458470
}
459471

460472
// If changing the target capacity means we gained a bunch of capacity,
@@ -465,6 +477,7 @@ impl Recv {
465477
task.wake();
466478
}
467479
}
480+
Ok(())
468481
}
469482

470483
pub(crate) fn apply_local_settings(
@@ -504,9 +517,13 @@ impl Recv {
504517
let dec = old_sz - target;
505518
tracing::trace!("decrementing all windows; dec={}", dec);
506519

507-
store.for_each(|mut stream| {
508-
stream.recv_flow.dec_recv_window(dec);
509-
})
520+
store.try_for_each(|mut stream| {
521+
stream
522+
.recv_flow
523+
.dec_recv_window(dec)
524+
.map_err(proto::Error::library_go_away)?;
525+
Ok::<_, proto::Error>(())
526+
})?;
510527
}
511528
Ordering::Greater => {
512529
// We must increase the (local) window on every open stream.
@@ -519,7 +536,10 @@ impl Recv {
519536
.recv_flow
520537
.inc_window(inc)
521538
.map_err(proto::Error::library_go_away)?;
522-
stream.recv_flow.assign_capacity(inc);
539+
stream
540+
.recv_flow
541+
.assign_capacity(inc)
542+
.map_err(proto::Error::library_go_away)?;
523543
Ok::<_, proto::Error>(())
524544
})?;
525545
}
@@ -626,7 +646,10 @@ impl Recv {
626646
}
627647

628648
// Update stream level flow control
629-
stream.recv_flow.send_data(sz);
649+
stream
650+
.recv_flow
651+
.send_data(sz)
652+
.map_err(proto::Error::library_go_away)?;
630653

631654
// Track the data as in-flight
632655
stream.in_flight_recv_data += sz;
@@ -667,7 +690,7 @@ impl Recv {
667690
}
668691

669692
// Update connection level flow control
670-
self.flow.send_data(sz);
693+
self.flow.send_data(sz).map_err(Error::library_go_away)?;
671694

672695
// Track the data as in-flight
673696
self.in_flight_data += sz;

0 commit comments

Comments
 (0)