removes raw indexing into packet data (#25554)

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.

(cherry picked from commit 5dbf7d8)

# Conflicts:
#	ledger/src/shred.rs
#	ledger/src/sigverify_shreds.rs
#	perf/src/sigverify.rs

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

@@ -583,7 +583,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

@@ -812,7 +812,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());
@@ -896,7 +896,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);
@@ -1348,7 +1348,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

@@ -377,12 +377,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,