Require FixedLengthReadable for structs that always drain reader

fuzz/src/chanmon_consistency.rs

@@ -65,7 +65,7 @@ use lightning::types::payment::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning::util::config::UserConfig;
 use lightning::util::hash_tables::*;
 use lightning::util::logger::Logger;
-use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
+use lightning::util::ser::{FixedLengthReader, LengthReadable, ReadableArgs, Writeable, Writer};
 use lightning::util::test_channel_signer::{EnforcementState, TestChannelSigner};
 
 use lightning_invoice::RawBolt11Invoice;
@@ -1101,7 +1101,9 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 											// update_fail_htlc as we do when we reject a payment.
 											let mut msg_ser = update_add.encode();
 											msg_ser[1000] ^= 0xff;
-											let new_msg = UpdateAddHTLC::read(&mut Cursor::new(&msg_ser)).unwrap();
+											let mut cursor = Cursor::new(&msg_ser);
+											let mut reader = FixedLengthReader::new(&mut cursor, msg_ser.len() as u64);
+											let new_msg = UpdateAddHTLC::read_from_fixed_length_buffer(&mut reader).unwrap();
 											dest.handle_update_add_htlc(nodes[$node].get_our_node_id(), &new_msg);
 										}
 									}

fuzz/src/msg_targets/utils.rs

@@ -49,15 +49,17 @@ macro_rules! test_msg {
 #[macro_export]
 macro_rules! test_msg_simple {
 	($MsgType: path, $data: ident) => {{
-		use lightning::util::ser::{Readable, Writeable};
-		let mut r = ::lightning::io::Cursor::new($data);
-		if let Ok(msg) = <$MsgType as Readable>::read(&mut r) {
+		use lightning::util::ser::{LengthReadable, Writeable};
+		let mut s = ::lightning::io::Cursor::new($data);
+		let mut r = ::lightning::util::ser::FixedLengthReader::new(&mut s, $data.len() as u64);
+		if let Ok(msg) = <$MsgType as LengthReadable>::read_from_fixed_length_buffer(&mut r) {
 			let mut w = VecWriter(Vec::new());
 			msg.write(&mut w).unwrap();
 			assert_eq!(msg.serialized_length(), w.0.len());
 
-			let msg =
-				<$MsgType as Readable>::read(&mut ::lightning::io::Cursor::new(&w.0)).unwrap();
+			let mut s = ::lightning::io::Cursor::new(&w.0);
+			let mut r = ::lightning::util::ser::FixedLengthReader::new(&mut s, w.0.len() as u64);
+			let msg = <$MsgType as LengthReadable>::read_from_fixed_length_buffer(&mut r).unwrap();
 			let mut w_two = VecWriter(Vec::new());
 			msg.write(&mut w_two).unwrap();
 			assert_eq!(&w.0[..], &w_two.0[..]);

lightning/src/ln/channelmanager.rs

@@ -83,7 +83,7 @@ use crate::util::config::{ChannelConfig, ChannelConfigUpdate, ChannelConfigOverr
 use crate::util::wakers::{Future, Notifier};
 use crate::util::scid_utils::fake_scid;
 use crate::util::string::UntrustedString;
-use crate::util::ser::{BigSize, FixedLengthReader, Readable, ReadableArgs, MaybeReadable, Writeable, Writer, VecWriter};
+use crate::util::ser::{BigSize, FixedLengthReader, LengthReadable, Readable, ReadableArgs, MaybeReadable, Writeable, Writer, VecWriter};
 use crate::util::logger::{Level, Logger, WithContext};
 use crate::util::errors::APIError;
 #[cfg(async_payments)] use {
@@ -12882,14 +12882,14 @@ impl Readable for HTLCFailureMsg {
 			2 => {
 				let length: BigSize = Readable::read(reader)?;
 				let mut s = FixedLengthReader::new(reader, length.0);
-				let res = Readable::read(&mut s)?;
+				let res = LengthReadable::read_from_fixed_length_buffer(&mut s)?;
 				s.eat_remaining()?; // Return ShortRead if there's actually not enough bytes
 				Ok(HTLCFailureMsg::Relay(res))
 			},
 			3 => {
 				let length: BigSize = Readable::read(reader)?;
 				let mut s = FixedLengthReader::new(reader, length.0);
-				let res = Readable::read(&mut s)?;
+				let res = LengthReadable::read_from_fixed_length_buffer(&mut s)?;
 				s.eat_remaining()?; // Return ShortRead if there's actually not enough bytes
 				Ok(HTLCFailureMsg::Malformed(res))
 			},
@@ -15055,8 +15055,8 @@ mod tests {
 		let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
 
 		create_announced_chan_between_nodes(&nodes, 0, 1);
-	
-		// Since we do not send peer storage, we manually simulate receiving a dummy 
+
+		// Since we do not send peer storage, we manually simulate receiving a dummy
 		// `PeerStorage` from the channel partner.
 		nodes[0].node.handle_peer_storage(nodes[1].node.get_our_node_id(), msgs::PeerStorage{data: vec![0; 100]});
 

lightning/src/ln/msgs.rs

@@ -729,7 +729,7 @@ pub struct UpdateFulfillHTLC {
 /// A [`peer_storage`] message that can be sent to or received from a peer.
 ///
 /// This message is used to distribute backup data to peers.
-/// If data is lost or corrupted, users can retrieve it through [`PeerStorageRetrieval`]  
+/// If data is lost or corrupted, users can retrieve it through [`PeerStorageRetrieval`]
 /// to recover critical information, such as channel states, for fund recovery.
 ///
 /// [`peer_storage`] is used to send our own encrypted backup data to a peer.
@@ -744,7 +744,7 @@ pub struct PeerStorage {
 /// A [`peer_storage_retrieval`] message that can be sent to or received from a peer.
 ///
 /// This message is sent to peers for whom we store backup data.
-/// If we receive this message, it indicates that the peer had stored our backup data.  
+/// If we receive this message, it indicates that the peer had stored our backup data.
 /// This data can be used for fund recovery in case of data loss.
 ///
 /// [`peer_storage_retrieval`] is used to send the most recent backup of the peer.
@@ -2006,13 +2006,13 @@ impl Writeable for TrampolineOnionPacket {
 }
 
 impl LengthReadable for TrampolineOnionPacket {
-	fn read_from_fixed_length_buffer<R: LengthRead>(r: &mut R) -> Result<Self, DecodeError> {
+	fn read_from_fixed_length_buffer<'a, R: Read>(r: &mut FixedLengthReader<'a, R>) -> Result<Self, DecodeError> {
 		let version = Readable::read(r)?;
 		let public_key = Readable::read(r)?;
 
 		let hop_data_len = r.total_bytes().saturating_sub(66); // 1 (version) + 33 (pubkey) + 32 (HMAC) = 66
 		let mut rd = FixedLengthReader::new(r, hop_data_len);
-		let hop_data = WithoutLength::<Vec<u8>>::read(&mut rd)?.0;
+		let hop_data = WithoutLength::<Vec<u8>>::read_from_fixed_length_buffer(&mut rd)?.0;
 
 		let hmac = Readable::read(r)?;
 
@@ -2094,8 +2094,8 @@ impl Writeable for AcceptChannel {
 	}
 }
 
-impl Readable for AcceptChannel {
-	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
+impl LengthReadable for AcceptChannel {
+	fn read_from_fixed_length_buffer<'a, R: Read>(r: &mut FixedLengthReader<'a, R>) -> Result<Self, DecodeError> {
 		let temporary_channel_id: ChannelId = Readable::read(r)?;
 		let dust_limit_satoshis: u64 = Readable::read(r)?;
 		let max_htlc_value_in_flight_msat: u64 = Readable::read(r)?;
@@ -2179,8 +2179,8 @@ impl Writeable for AcceptChannelV2 {
 	}
 }
 
-impl Readable for AcceptChannelV2 {
-	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
+impl LengthReadable for AcceptChannelV2 {
+	fn read_from_fixed_length_buffer<'a, R: Read>(r: &mut FixedLengthReader<'a, R>) -> Result<Self, DecodeError> {
 		let temporary_channel_id: ChannelId = Readable::read(r)?;
 		let funding_satoshis: u64 = Readable::read(r)?;
 		let dust_limit_satoshis: u64 = Readable::read(r)?;
@@ -2442,8 +2442,8 @@ impl Writeable for Init {
 	}
 }
 
-impl Readable for Init {
-	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
+impl LengthReadable for Init {
+	fn read_from_fixed_length_buffer<'a, R: Read>(r: &mut FixedLengthReader<'a, R>) -> Result<Self, DecodeError> {
 		let global_features: InitFeatures = Readable::read(r)?;
 		let features: InitFeatures = Readable::read(r)?;
 		let mut remote_network_address: Option<SocketAddress> = None;
@@ -2488,8 +2488,8 @@ impl Writeable for OpenChannel {
 	}
 }
 
-impl Readable for OpenChannel {
-	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
+impl LengthReadable for OpenChannel {
+	fn read_from_fixed_length_buffer<'a, R: Read>(r: &mut FixedLengthReader<'a, R>) -> Result<Self, DecodeError> {
 		let chain_hash: ChainHash = Readable::read(r)?;
 		let temporary_channel_id: ChannelId = Readable::read(r)?;
 		let funding_satoshis: u64 = Readable::read(r)?;
@@ -2572,8 +2572,8 @@ impl Writeable for OpenChannelV2 {
 	}
 }
 
-impl Readable for OpenChannelV2 {
-	fn read<R: Read>(r: &mut R) -> Result<Self, DecodeError> {
+impl LengthReadable for OpenChannelV2 {
+	fn read_from_fixed_length_buffer<'a, R: Read>(r: &mut FixedLengthReader<'a, R>) -> Result<Self, DecodeError> {
 		let chain_hash: ChainHash = Readable::read(r)?;
 		let temporary_channel_id: ChannelId = Readable::read(r)?;
 		let funding_feerate_sat_per_1000_weight: u32 = Readable::read(r)?;
@@ -4379,7 +4379,9 @@ mod tests {
 		let encoded_value = closing_signed.encode();
 		let target_value = <Vec<u8>>::from_hex("020202020202020202020202020202020202020202020202020202020202020200083a840000034dd977cb9b53d93a6ff64bb5f1e158b4094b66e798fb12911168a3ccdf80a83096340a6a95da0ae8d9f776528eecdbb747eb6b545495a4319ed5378e35b21e073a").unwrap();
 		assert_eq!(encoded_value, target_value);
-		assert_eq!(msgs::ClosingSigned::read(&mut Cursor::new(&target_value)).unwrap(), closing_signed);
+		let mut msg_stream = Cursor::new(&target_value);
+		let mut msg_reader = FixedLengthReader::new(&mut msg_stream, target_value.len() as u64);
+		assert_eq!(msgs::ClosingSigned::read_from_fixed_length_buffer(&mut msg_reader).unwrap(), closing_signed);
 
 		let closing_signed_with_range = msgs::ClosingSigned {
 			channel_id: ChannelId::from_bytes([2; 32]),
@@ -4393,7 +4395,9 @@ mod tests {
 		let encoded_value_with_range = closing_signed_with_range.encode();
 		let target_value_with_range = <Vec<u8>>::from_hex("020202020202020202020202020202020202020202020202020202020202020200083a840000034dd977cb9b53d93a6ff64bb5f1e158b4094b66e798fb12911168a3ccdf80a83096340a6a95da0ae8d9f776528eecdbb747eb6b545495a4319ed5378e35b21e073a011000000000deadbeef1badcafe01234567").unwrap();
 		assert_eq!(encoded_value_with_range, target_value_with_range);
-		assert_eq!(msgs::ClosingSigned::read(&mut Cursor::new(&target_value_with_range)).unwrap(),
+		let mut msg_stream = Cursor::new(&target_value_with_range);
+		let mut msg_reader = FixedLengthReader::new(&mut msg_stream, target_value_with_range.len() as u64);
+		assert_eq!(msgs::ClosingSigned::read_from_fixed_length_buffer(&mut msg_reader).unwrap(),
 			closing_signed_with_range);
 	}
 
@@ -4557,7 +4561,9 @@ mod tests {
 		let encoded_value = init_msg.encode();
 		let target_value = <Vec<u8>>::from_hex("0000000001206fe28c0ab6f1b372c1a6a246ae63f74f931e8365e15a089c68d61900000000000307017f00000103e8").unwrap();
 		assert_eq!(encoded_value, target_value);
-		assert_eq!(msgs::Init::read(&mut Cursor::new(&target_value)).unwrap(), init_msg);
+		let mut msg_stream = Cursor::new(&target_value);
+		let mut msg_reader = FixedLengthReader::new(&mut msg_stream, target_value.len() as u64);
+		assert_eq!(msgs::Init::read_from_fixed_length_buffer(&mut msg_reader).unwrap(), init_msg);
 	}
 
 	#[test]
@@ -4794,7 +4800,8 @@ mod tests {
 		{ // verify that a codec round trip works
 			let mut reader = Cursor::new(&encoded_trampoline_packet);
 			let mut trampoline_packet_reader = FixedLengthReader::new(&mut reader, encoded_trampoline_packet.len() as u64);
-			let decoded_trampoline_packet: TrampolineOnionPacket = <TrampolineOnionPacket as LengthReadable>::read_from_fixed_length_buffer(&mut trampoline_packet_reader).unwrap();
+			let decoded_trampoline_packet: TrampolineOnionPacket =
+				<TrampolineOnionPacket as LengthReadable>::read_from_fixed_length_buffer(&mut trampoline_packet_reader).unwrap();
 			assert_eq!(decoded_trampoline_packet.encode(), encoded_trampoline_packet);
 		}
 
@@ -4920,7 +4927,9 @@ mod tests {
 		let target_value = <Vec<u8>>::from_hex("06226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f000186a0000005dc").unwrap();
 		assert_eq!(encoded_value, target_value);
 
-		query_channel_range = Readable::read(&mut Cursor::new(&target_value[..])).unwrap();
+		let mut msg_stream = Cursor::new(&target_value[..]);
+		let mut msg_reader = FixedLengthReader::new(&mut msg_stream, target_value.len() as u64);
+		query_channel_range = LengthReadable::read_from_fixed_length_buffer(&mut msg_reader).unwrap();
 		assert_eq!(query_channel_range.first_blocknum, 100000);
 		assert_eq!(query_channel_range.number_of_blocks, 1500);
 	}
@@ -5004,7 +5013,9 @@ mod tests {
 		let target_value = <Vec<u8>>::from_hex("06226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f01").unwrap();
 		assert_eq!(encoded_value, target_value);
 
-		reply_short_channel_ids_end = Readable::read(&mut Cursor::new(&target_value[..])).unwrap();
+		let mut msg_stream = Cursor::new(&target_value[..]);
+		let mut msg_reader = FixedLengthReader::new(&mut msg_stream, target_value.len() as u64);
+		reply_short_channel_ids_end = LengthReadable::read_from_fixed_length_buffer(&mut msg_reader).unwrap();
 		assert_eq!(reply_short_channel_ids_end.chain_hash, expected_chain_hash);
 		assert_eq!(reply_short_channel_ids_end.full_information, true);
 	}
@@ -5021,7 +5032,10 @@ mod tests {
 		let target_value = <Vec<u8>>::from_hex("06226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f5ec57980ffffffff").unwrap();
 		assert_eq!(encoded_value, target_value);
 
-		gossip_timestamp_filter = Readable::read(&mut Cursor::new(&target_value[..])).unwrap();
+		let mut stream = Cursor::new(&target_value[..]);
+		gossip_timestamp_filter = LengthReadable::read_from_fixed_length_buffer(
+			&mut FixedLengthReader::new(&mut stream, target_value.len() as u64)
+		).unwrap();
 		assert_eq!(gossip_timestamp_filter.chain_hash, expected_chain_hash);
 		assert_eq!(gossip_timestamp_filter.first_timestamp, 1590000000);
 		assert_eq!(gossip_timestamp_filter.timestamp_range, 0xffff_ffff);

lightning/src/ln/peer_handler.rs

@@ -1544,7 +1544,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
 									try_potential_handleerror!(peer,
 										peer.channel_encryptor.decrypt_message(&mut peer.pending_read_buffer[..]));
 
-									let mut reader = io::Cursor::new(&peer.pending_read_buffer[..peer.pending_read_buffer.len() - 16]);
+									let mut cursor = io::Cursor::new(&peer.pending_read_buffer[..peer.pending_read_buffer.len() - 16]);
+									let mut reader = crate::util::ser::FixedLengthReader::new(&mut cursor, (peer.pending_read_buffer.len() - 16) as u64);
 									let message_result = wire::read(&mut reader, &*self.message_handler.custom_message_handler);
 
 									// Reset read buffer