removes raw indexing into packet data

Packets are at the boundary of the system where, vast majority of the
time, they are received from an untrusted source. Raw indexing into the
data buffer can open attack vectors if the offsets are invalid.
Validating offsets beforehand is verbose and error prone.

The commit updates Packet::data() api to take a SliceIndex and always to
return an Option. The call-sites are so forced to explicitly handle the
case where the offsets are invalid.

Author: behzadnouri

bench-streamer/src/main.rs

@@ -37,7 +37,8 @@ fn producer(addr: &SocketAddr, exit: Arc<AtomicBool>) -> JoinHandle<()> {
         for p in packet_batch.iter() {
             let a = p.meta.socket_addr();
             assert!(p.meta.size <= PACKET_DATA_SIZE);
-            send.send_to(p.data(), &a).unwrap();
+            let data = p.data(..).unwrap_or_default();
+            send.send_to(data, &a).unwrap();
             num += 1;
         }
         assert_eq!(num, 10);

core/src/banking_stage.rs

@@ -508,7 +508,7 @@ impl BankingStage {
             .iter()
             .filter_map(|p| {
                 if !p.meta.forwarded() && data_budget.take(p.meta.size) {
-                    Some(p.data().to_vec())
+                    Some(p.data(..)?.to_vec())
                 } else {
                     None
                 }

core/src/packet_hasher.rs

@@ -26,7 +26,7 @@ impl Default for PacketHasher {
 
 impl PacketHasher {
     pub(crate) fn hash_packet(&self, packet: &Packet) -> u64 {
-        self.hash_data(packet.data())
+        self.hash_data(packet.data(..).unwrap_or_default())
     }
 
     pub(crate) fn hash_shred(&self, shred: &Shred) -> u64 {

core/src/serve_repair.rs

@@ -814,7 +814,7 @@ mod tests {
                 .into_iter()
                 .filter_map(|p| {
                     assert_eq!(repair_response::nonce(p).unwrap(), nonce);
-                    Shred::new_from_serialized_shred(p.data().to_vec()).ok()
+                    Shred::new_from_serialized_shred(p.data(..).unwrap().to_vec()).ok()
                 })
                 .collect();
             assert!(!rv.is_empty());
@@ -898,7 +898,7 @@ mod tests {
                 .into_iter()
                 .filter_map(|p| {
                     assert_eq!(repair_response::nonce(p).unwrap(), nonce);
-                    Shred::new_from_serialized_shred(p.data().to_vec()).ok()
+                    Shred::new_from_serialized_shred(p.data(..).unwrap().to_vec()).ok()
                 })
                 .collect();
             assert_eq!(rv[0].index(), 1);
@@ -1347,7 +1347,7 @@ mod tests {
 
     fn verify_responses<'a>(request: &ShredRepairType, packets: impl Iterator<Item = &'a Packet>) {
         for packet in packets {
-            let shred_payload = packet.data().to_vec();
+            let shred_payload = packet.data(..).unwrap().to_vec();
             let shred = Shred::new_from_serialized_shred(shred_payload).unwrap();
             request.verify_response(&shred);
         }

core/src/unprocessed_packet_batches.rs

@@ -355,12 +355,14 @@ pub fn deserialize_packets<'a>(
 
 /// Read the transaction message from packet data
 pub fn packet_message(packet: &Packet) -> Result<&[u8], DeserializedPacketError> {
-    let (sig_len, sig_size) =
-        decode_shortu16_len(packet.data()).map_err(DeserializedPacketError::ShortVecError)?;
+    let (sig_len, sig_size) = packet
+        .data(..)
+        .and_then(|bytes| decode_shortu16_len(bytes).ok())
+        .ok_or(DeserializedPacketError::ShortVecError(()))?;
     sig_len
         .checked_mul(size_of::<Signature>())
         .and_then(|v| v.checked_add(sig_size))
-        .and_then(|msg_start| packet.data().get(msg_start..))
+        .and_then(|msg_start| packet.data(msg_start..))
         .ok_or(DeserializedPacketError::SignatureOverflowed(sig_size))
 }
 

core/src/window_service.rs

@@ -363,7 +363,7 @@ where
             inc_new_counter_debug!("streamer-recv_window-invalid_or_unnecessary_packet", 1);
             return None;
         }
-        let serialized_shred = packet.data().to_vec();
+        let serialized_shred = packet.data(..)?.to_vec();
         let shred = Shred::new_from_serialized_shred(serialized_shred).ok()?;
         if !shred_filter(&shred, working_bank.clone(), last_root) {
             return None;

gossip/tests/gossip.rs

@@ -260,7 +260,7 @@ pub fn cluster_info_retransmit() {
     let retransmit_peers: Vec<_> = peers.iter().collect();
     retransmit_to(
         &retransmit_peers,
-        p.data(),
+        p.data(..).unwrap(),
         &tn1,
         false,
         &SocketAddrSpace::Unspecified,

ledger/src/shred.rs

@@ -575,12 +575,15 @@ pub fn get_shred_slot_index_type(
         }
     };
 
-    let shred_type = match ShredType::try_from(p.data()[OFFSET_OF_SHRED_TYPE]) {
-        Err(_) => {
+    let shred_type = match p
+        .data(OFFSET_OF_SHRED_TYPE)
+        .and_then(|shred_type| ShredType::try_from(*shred_type).ok())
+    {
+        None => {
             stats.bad_shred_type += 1;
             return None;
         }
-        Ok(shred_type) => shred_type,
+        Some(shred_type) => shred_type,
     };
     Some((slot, index, shred_type))
 }
@@ -733,7 +736,7 @@ mod tests {
         let shred = Shred::new_from_data(10, 0, 1000, &[1, 2, 3], ShredFlags::empty(), 0, 1, 0);
         let mut packet = Packet::default();
         shred.copy_to_packet(&mut packet);
-        let shred_res = Shred::new_from_serialized_shred(packet.data().to_vec());
+        let shred_res = Shred::new_from_serialized_shred(packet.data(..).unwrap().to_vec());
         assert_matches!(
             shred.parent(),
             Err(Error::InvalidParentOffset {
@@ -898,7 +901,7 @@ mod tests {
         assert_eq!(shred, Shred::new_from_serialized_shred(payload).unwrap());
         assert_eq!(
             shred.reference_tick(),
-            Shred::reference_tick_from_data(packet.data()).unwrap()
+            Shred::reference_tick_from_data(packet.data(..).unwrap()).unwrap()
         );
         assert_eq!(Shred::get_slot_from_packet(&packet), Some(shred.slot()));
         assert_eq!(
@@ -939,7 +942,7 @@ mod tests {
         assert_eq!(shred, Shred::new_from_serialized_shred(payload).unwrap());
         assert_eq!(
             shred.reference_tick(),
-            Shred::reference_tick_from_data(packet.data()).unwrap()
+            Shred::reference_tick_from_data(packet.data(..).unwrap()).unwrap()
         );
         assert_eq!(Shred::get_slot_from_packet(&packet), Some(shred.slot()));
         assert_eq!(

ledger/src/sigverify_shreds.rs

@@ -62,9 +62,9 @@ pub fn verify_shred_cpu(packet: &Packet, slot_leaders: &HashMap<u64, [u8; 32]>)
     };
     trace!("slot {}", slot);
     let pubkey = slot_leaders.get(&slot)?;
-    let signature = Signature::new(packet.data().get(sig_start..sig_end)?);
+    let signature = Signature::new(packet.data(sig_start..sig_end)?);
     trace!("signature {}", signature);
-    if !signature.verify(pubkey, packet.data().get(msg_start..msg_end)?) {
+    if !signature.verify(pubkey, packet.data(msg_start..msg_end)?) {
         return Some(0);
     }
     Some(1)
@@ -304,7 +304,8 @@ fn sign_shred_cpu(keypair: &Keypair, packet: &mut Packet) {
     let sig_start = 0;
     let sig_end = sig_start + size_of::<Signature>();
     let msg_start = sig_end;
-    let signature = keypair.sign_message(&packet.data()[msg_start..]);
+    let message = packet.data(msg_start..).unwrap_or_default();
+    let signature = keypair.sign_message(message);
     trace!("signature {:?}", signature);
     packet.buffer_mut()[..sig_end].copy_from_slice(signature.as_ref());
 }