diff --git a/rust/src/filetracker.rs b/rust/src/filetracker.rs index 820683662f..7c8fba9c08 100644 --- a/rust/src/filetracker.rs +++ b/rust/src/filetracker.rs @@ -65,6 +65,13 @@ pub struct FileTransferTracker { chunk_is_last: bool, chunk_is_ooo: bool, file_is_truncated: bool, + // Consumed but not stored: a discarded retransmission, so a split record's + // continuation is skipped without touching the file. + chunk_consume_only: bool, + // Consume-and-discard before the in-progress chunk stores: the part of a + // backward chunk's covered prefix its first fragment did not buffer (see + // the new_chunk re-anchor); the continuation carries it first. + chunk_pending_discard: u32, chunks: HashMap, cur_ooo_chunk_offset: u64, @@ -107,16 +114,23 @@ impl FileTransferTracker { self.file_open = false; self.file_closed = true; self.tracked = 0; + // no chunk state survives a close + self.chunk_consume_only = false; } pub fn trunc(&mut self, config: &'static SuricataFileContext) { if self.file_is_truncated || !self.file_open { + // no file state to drop: clear the discard markers as well + self.chunk_consume_only = false; + self.chunk_pending_discard = 0; return; } let myflags = self.file_flags | 1; // TODO util-file.c::FILE_TRUNCATED self.file.file_close(config, &self.track_id, myflags); SCLogDebug!("truncated file"); self.file_is_truncated = true; + self.chunk_consume_only = false; + self.chunk_pending_discard = 0; self.chunks.clear(); self.in_flight = 0; self.cur_ooo = 0; @@ -126,9 +140,27 @@ impl FileTransferTracker { &mut self, config: &'static SuricataFileContext, name: &[u8], data: &[u8], chunk_offset: u64, chunk_size: u32, fill_bytes: u8, is_last: bool, xid: &u32, ) -> u32 { + // Shadowed so a partial-overlap trim below can re-anchor the chunk + // at the tracked offset instead of discarding its new tail. + let mut data = data; + let mut chunk_offset = chunk_offset; + let mut chunk_size = chunk_size; + + // A new chunk arms fresh discard state; the re-anchor below sets the + // real value when a fragment leaves uncovered covered prefix behind. + self.chunk_pending_discard = 0; + if self.chunk_left != 0 || self.fill_bytes != 0 { - SCLogDebug!("current chunk incomplete: truncating"); - self.trunc(config); + if self.chunk_consume_only { + // an unconsumed discarded chunk holds no file state: + // reset it without truncating the file + self.chunk_consume_only = false; + self.chunk_left = 0; + self.fill_bytes = 0; + } else { + SCLogDebug!("current chunk incomplete: truncating"); + self.trunc(config); + } } SCLogDebug!( @@ -140,6 +172,58 @@ impl FileTransferTracker { // for now assume that is_last means its really the last chunk // so no out of order chunks coming after. This means that if // the last chunk is out or order, we've missed chunks before. + // + // Chunk before `tracked` but extending past it: drop the covered prefix, + // re-anchor at `tracked`, keep the LOGICAL tail (chunk_size - overlap), + // not the buffered tail (else a fragmented chunk finishes early). + // Trigger on the logical extent, not the buffered one: a partial record + // can carry all its buffered bytes inside the tracked region while the + // chunk still extends past it (else the retransmission arm below would + // consume its new tail instead of storing it). + if chunk_offset < self.tracked { + let overlap = (self.tracked - chunk_offset) as usize; + if chunk_size as usize > overlap { + let drop = overlap.min(data.len()); + SCLogDebug!( + "NEW CHUNK partial overlap {}/{}: dropping {} covered bytes, logical tail {} ({} buffered this fragment)", + chunk_offset, + self.tracked, + drop, + (chunk_size as usize) - overlap, + data.len() - drop + ); + data = &data[drop..]; + chunk_offset = self.tracked; + chunk_size = chunk_size.saturating_sub(overlap as u32); + // The continuation still carries the covered prefix this + // fragment did not buffer: discard it as it arrives, only + // the new extent is stored. + self.chunk_pending_discard = (overlap - drop) as u32; + } + } + if chunk_offset < self.tracked { + // Retransmission of a tracked region: discard (never queue), and arm a + // consume-only chunk so a split record's continuation is consumed cleanly. + SCLogDebug!( + "NEW CHUNK retransmits tracked offset {}/{}; dropping", + chunk_offset, + self.tracked + ); + if is_last { + // Terminal backward write: the file can no longer complete (the queue + // only drains at `tracked`); truncate now so it is logged and released. + self.trunc(config); + } + self.chunk_left = chunk_size; + self.fill_bytes = fill_bytes; + self.chunk_is_ooo = false; + self.chunk_is_last = false; + self.chunk_consume_only = true; + if !self.file_open { + return 0; + } + return self.update(config, data, 0); + } if chunk_offset != self.tracked { SCLogDebug!( "NEW CHUNK IS OOO: expected {}, got {}", @@ -158,7 +242,15 @@ impl FileTransferTracker { self.fill_bytes = fill_bytes; self.chunk_is_last = is_last; - if self.file_is_truncated || self.file_closed { + if self.file_is_truncated { + // File can't store data, but a split record's payload + XDR padding must + // still be consumed (as the truncated update path does) to stay in sync. + if !data.is_empty() { + return self.update(config, data, 0); + } + return 0; + } + if self.file_closed { return 0; } if !self.file_open { @@ -182,12 +274,36 @@ impl FileTransferTracker { pub fn update( &mut self, config: &'static SuricataFileContext, data: &[u8], gap_size: u32, ) -> u32 { - if self.file_is_truncated { - let consumed = std::cmp::min(data.len() as u32, self.chunk_left); - self.chunk_left = self.chunk_left.saturating_sub(data.len() as u32); - return consumed; - } + // A backward chunk's re-anchor may leave a covered prefix its + // continuation carries before any storable byte: consume-and-discard + // it first (see chunk_pending_discard). + let mut data = data; let mut consumed = 0_usize; + if self.chunk_pending_discard > 0 { + let d = std::cmp::min(data.len(), self.chunk_pending_discard as usize); + data = &data[d..]; + self.chunk_pending_discard -= d as u32; + consumed += d; + if data.is_empty() { + return consumed as u32; + } + } + if self.file_is_truncated || self.chunk_consume_only { + // Consumed but not stored (truncated file or discarded retransmission): consume the + // chunk data and the record's XDR padding together so the next record starts at a frame. + let mut c = std::cmp::min(data.len() as u32, self.chunk_left); + self.chunk_left = self.chunk_left.saturating_sub(c); + if self.chunk_left == 0 && self.fill_bytes > 0 { + let extra = (data.len() as u32).saturating_sub(c); + let f = std::cmp::min(extra, self.fill_bytes as u32); + c += f; + self.fill_bytes = self.fill_bytes.saturating_sub(f as u8); + } + if self.chunk_left == 0 && self.fill_bytes == 0 { + self.chunk_consume_only = false; + } + return consumed as u32 + c; + } let is_gap = gap_size > 0; if is_gap || gap_size > 0 { SCLogDebug!( @@ -205,7 +321,7 @@ impl FileTransferTracker { self.close(config); self.chunk_is_last = false; } - return 0; + return consumed as u32; } else if self.chunk_left == 0 { SCLogDebug!("FILL BYTES {} from prev run", self.fill_bytes); if data.len() >= self.fill_bytes as usize { @@ -271,57 +387,59 @@ impl FileTransferTracker { self.fill_bytes -= extra as u8; SCLogDebug!("CHUNK(post) fill bytes now still {}", self.fill_bytes); } - self.chunk_left = 0; - } else { - self.chunk_left = 0; - - if !self.chunk_is_ooo { - loop { - let _offset = self.tracked; - match self.chunks.remove(&self.tracked) { - Some(c) => { - self.in_flight -= c.chunk.len() as u64; - - let res = self.file.file_append( - config, - &self.track_id, - &c.chunk, - c.contains_gap, - ); - match res { - 0 => {} - -2 => { - self.file_is_truncated = true; - } - _ => { - SCLogDebug!("got error so truncating file"); - self.file_is_truncated = true; - } - } - - self.tracked += c.chunk.len() as u64; - self.cur_ooo -= c.chunk.len() as u64; + } + self.chunk_left = 0; - SCLogDebug!("STORED OOO CHUNK at offset {}, tracked now {}, stored len {}", _offset, self.tracked, c.chunk.len()); - } - _ => { - SCLogDebug!( - "NO STORED CHUNK found at _offset {}", - self.tracked - ); - break; + // Chunk completed in this call, padded or not: settle its state either + // way (skipping leaves the OOO marker/offset stale for the next record). + if !self.chunk_is_ooo { + loop { + let _offset = self.tracked; + match self.chunks.remove(&self.tracked) { + Some(c) => { + self.in_flight -= c.chunk.len() as u64; + + let res = self.file.file_append( + config, + &self.track_id, + &c.chunk, + c.contains_gap, + ); + match res { + 0 => {} + -2 => { + self.file_is_truncated = true; + } + _ => { + SCLogDebug!("got error so truncating file"); + self.file_is_truncated = true; + } } - }; - } - } else { - SCLogDebug!( - "UPDATE: complete ooo chunk. Offset {}", - self.cur_ooo_chunk_offset - ); - self.chunk_is_ooo = false; - self.cur_ooo_chunk_offset = 0; + self.tracked += c.chunk.len() as u64; + self.cur_ooo -= c.chunk.len() as u64; + + SCLogDebug!( + "STORED OOO CHUNK at offset {}, tracked now {}, stored len {}", + _offset, + self.tracked, + c.chunk.len() + ); + } + _ => { + SCLogDebug!("NO STORED CHUNK found at _offset {}", self.tracked); + break; + } + }; } + } else { + SCLogDebug!( + "UPDATE: complete ooo chunk. Offset {}", + self.cur_ooo_chunk_offset + ); + + self.chunk_is_ooo = false; + self.cur_ooo_chunk_offset = 0; } if self.chunk_is_last { SCLogDebug!("last chunk, closing"); @@ -372,4 +490,144 @@ impl FileTransferTracker { pub fn get_inflight_cnt(&self) -> usize { self.chunks.len() } + + /// The XDR padding bytes of the record whose chunk data is in flight. + pub fn get_fill_bytes(&self) -> u8 { + self.fill_bytes + } + + /// True if an append at `offset` is out of order and grows the OOO queue. + pub fn is_ooo_offset(&self, offset: u64) -> bool { + offset > self.tracked + } + + /// True if an append at `offset` creates a new OOO chunk-map entry + /// (not a continuation of an already enqueued chunk). + pub fn new_chunk_entry(&self, offset: u64) -> bool { + self.is_ooo_offset(offset) && !self.chunks.contains_key(&offset) + } + + /// True if the chunk currently being filled is out of order. + pub fn current_chunk_is_ooo(&self) -> bool { + self.chunk_is_ooo + } +} + +#[cfg(test)] +mod tests { + use crate::core::{StreamingBufferConfig, SuricataFileContext}; + use crate::filetracker::FileTransferTracker; + + // the FFI file ops are no-ops under cfg(test), so a default context is + // safe to use with the tracker + static TEST_SBCFG: StreamingBufferConfig = StreamingBufferConfig { + buf_size: 2048, + max_regions: 8, + region_gap: 0, + Calloc: None, + Realloc: None, + Free: None, + }; + static TEST_FC: SuricataFileContext = SuricataFileContext { + files_sbcfg: &TEST_SBCFG, + }; + + #[test] + fn test_partial_overlap_chunk_keeps_tail_past_tracked() { + // tracked=4096; a chunk at offset 3072 of length 2048 spans 3072..5120. + // The covered 1024-byte prefix is discarded; the new tail (4096..5120) + // is stored, not lost (a permanent inspection/extraction gap). + let xid: u32 = 0x42; + let mut ft = FileTransferTracker::new(); + let head = vec![0u8; 4096]; + ft.new_chunk(&TEST_FC, b"f", &head, 0, 4096, 0, false, &xid); + assert_eq!(ft.tracked, 4096); + + let overlap = vec![0u8; 2048]; + ft.new_chunk(&TEST_FC, b"f", &overlap, 3072, 2048, 0, false, &xid); + // the 1024-byte tail past the tracked region must be stored + assert_eq!(ft.tracked, 5120); + } + + #[test] + fn test_partial_overlap_fragmented_chunk_keeps_logical_tail() { + // tracked=4096. A chunk at offset 3072 of LOGICAL size 8192, first fragment + // only 4096 buffered bytes (3072..7168): drop the covered 1024-byte prefix, + // keep the LOGICAL tail (4096..11264), not just the buffered tail. + let xid: u32 = 0x44; + let mut ft = FileTransferTracker::new(); + let head = vec![0u8; 4096]; + ft.new_chunk(&TEST_FC, b"f", &head, 0, 4096, 0, false, &xid); + assert_eq!(ft.tracked, 4096); + + // first fragment: 4096 buffered bytes of the 8192-byte logical chunk + let frag1 = vec![0u8; 4096]; + ft.new_chunk(&TEST_FC, b"f", &frag1, 3072, 8192, 0, false, &xid); + // covered 1024-byte prefix dropped, buffered tail (3072 bytes) stored + // (tracked -> 7168), chunk still expects its remaining 4096 logical bytes. + assert_eq!(ft.tracked, 7168); + + // second fragment: the remaining 4096 logical bytes (7168..11264). The + // logical-tail size makes it consume exactly 4096 and complete, where the + // buggy buffered-tail size would leave the chunk already complete. + let frag2 = vec![0u8; 4096]; + let consumed = ft.update(&TEST_FC, &frag2, 0); + assert_eq!(consumed, 4096); + assert_eq!(ft.tracked, 11264); + } + + #[test] + fn test_fully_covered_chunk_is_discarded() { + // a chunk entirely within the tracked region is a pure + // retransmission: discarded wholesale, tracked unchanged. + let xid: u32 = 0x43; + let mut ft = FileTransferTracker::new(); + let head = vec![0u8; 4096]; + ft.new_chunk(&TEST_FC, b"f", &head, 0, 4096, 0, false, &xid); + assert_eq!(ft.tracked, 4096); + + let retrans = vec![0u8; 512]; + ft.new_chunk(&TEST_FC, b"f", &retrans, 100, 512, 0, false, &xid); + assert_eq!(ft.tracked, 4096); + } + + #[test] + fn test_partial_overlap_fragment_inside_tracked_keeps_logical_tail() { + // tracked=4096. A chunk at offset 3072 of LOGICAL size 4096 spans + // 3072..7168, but the first fragment only buffers 1000 bytes + // (3072..4072): all buffered bytes are inside the tracked region while + // the logical chunk extends 952 bytes past it. The re-anchor must + // trigger on the logical extent (not the buffered one) so the + // arriving continuation is stored, not consumed away; and the 24 + // covered bytes the continuation still carries (4072..4096) are + // discarded, not stored. + let xid: u32 = 0x45; + let mut ft = FileTransferTracker::new(); + let head = vec![0u8; 4096]; + ft.new_chunk(&TEST_FC, b"f", &head, 0, 4096, 0, false, &xid); + assert_eq!(ft.tracked, 4096); + + // first fragment: 1000 buffered bytes, all covered (overlap = 1024) + let frag1 = vec![0xAAu8; 1000]; + ft.new_chunk(&TEST_FC, b"f", &frag1, 3072, 4096, 0, false, &xid); + // nothing new stored yet: covered prefix dropped, chunk re-anchored at + // 4096 with a 3072-byte logical tail still to come + assert_eq!(ft.tracked, 4096); + + // continuation: 24 covered bytes (4072..4096) followed by the logical + // tail 4096..7168. The full 3096-byte continuation must be consumed + // (the pre-fix re-anchor shrank the chunk by the whole overlap, so the + // chunk finished after 3072 bytes and the trailing 24 spilled into the + // next record's framing) + let mut frag2 = vec![0xAAu8; 24]; + frag2.extend(vec![0xBBu8; 3072]); + let consumed = ft.update(&TEST_FC, &frag2, 0); + assert_eq!(consumed, 3096); + assert_eq!(ft.tracked, 7168); + + // another record's chunk right after: framing is intact + let frag3 = vec![0xCCu8; 100]; + ft.new_chunk(&TEST_FC, b"f", &frag3, 7168, 100, 0, false, &xid); + assert_eq!(ft.tracked, 7268); + } } diff --git a/rust/src/nfs/nfs.rs b/rust/src/nfs/nfs.rs index 047334aa13..c8f62fad47 100644 --- a/rust/src/nfs/nfs.rs +++ b/rust/src/nfs/nfs.rs @@ -135,7 +135,7 @@ pub enum NFSFrameType { NFS4Status, } -#[derive(FromPrimitive, Debug, AppLayerEvent)] +#[derive(FromPrimitive, Debug, AppLayerEvent, Clone, Copy)] pub enum NFSEvent { MalformedData = 0, NonExistingVersion = 1, @@ -153,6 +153,9 @@ pub enum NFSEvent { WriteRequestTooLarge = 8, /// READ response returns more data than the configured `max-read-size` ReadResponseTooLarge = 9, + /// OOO file data exceeded the queue backstop (gapped stream); queued + /// data truncated to bound memory + TruncatedFileData = 10, } #[derive(Debug)] @@ -384,34 +387,65 @@ fn filetracker_update(ft: &mut FileTransferTracker, data: &[u8], gap_size: u32) } } -/// Returns the event to raise if enqueuing `len` bytes of queued file data -/// in the read direction would exceed the configured limits. -pub(crate) fn read_queue_limit_event(ft: &FileTransferTracker, len: u64) -> Option { +/// Event to raise if enqueuing `len` bytes at `offset` (read direction) +/// would exceed the configured limits. OOO data at a new offset grows the +/// queue; in-order appends drain it and are never rejected. +pub(crate) fn read_queue_limit_event( + ft: &FileTransferTracker, offset: u64, len: u64, +) -> Option { let max_size = unsafe { NFS_CFG_MAX_READ_QUEUE_SIZE }; - if max_size != 0 && ft.get_inflight_size() + len > u64::from(max_size) { + if max_size != 0 + && ft.is_ooo_offset(offset) + && ft.get_inflight_size() + len > u64::from(max_size) + { return Some(NFSEvent::ReadQueueSizeExceeded); } let max_cnt = unsafe { NFS_CFG_MAX_READ_QUEUE_CNT }; - if max_cnt != 0 && ft.get_inflight_cnt() >= max_cnt as usize { + if max_cnt != 0 && ft.new_chunk_entry(offset) && ft.get_inflight_cnt() + 1 > max_cnt as usize { return Some(NFSEvent::ReadQueueCntExceeded); } None } -/// Returns the event to raise if enqueuing `len` bytes of queued file data -/// in the write direction would exceed the configured limits. -pub(crate) fn write_queue_limit_event(ft: &FileTransferTracker, len: u64) -> Option { +/// Write-direction counterpart of read_queue_limit_event. +pub(crate) fn write_queue_limit_event( + ft: &FileTransferTracker, offset: u64, len: u64, +) -> Option { let max_size = unsafe { NFS_CFG_MAX_WRITE_QUEUE_SIZE }; - if max_size != 0 && ft.get_inflight_size() + len > u64::from(max_size) { + if max_size != 0 + && ft.is_ooo_offset(offset) + && ft.get_inflight_size() + len > u64::from(max_size) + { return Some(NFSEvent::WriteQueueSizeExceeded); } let max_cnt = unsafe { NFS_CFG_MAX_WRITE_QUEUE_CNT }; - if max_cnt != 0 && ft.get_inflight_cnt() >= max_cnt as usize { + if max_cnt != 0 && ft.new_chunk_entry(offset) && ft.get_inflight_cnt() + 1 > max_cnt as usize { return Some(NFSEvent::WriteQueueCntExceeded); } None } +/// Size-only queue check for the chunk path (continuation of the +/// active chunk, which is already a map entry). +pub(crate) fn read_chunk_queue_limit_event(ft: &FileTransferTracker, add: u64) -> Option { + let max_size = unsafe { NFS_CFG_MAX_READ_QUEUE_SIZE }; + if max_size != 0 && ft.get_inflight_size() + add > u64::from(max_size) { + return Some(NFSEvent::ReadQueueSizeExceeded); + } + None +} + +/// Write-direction counterpart of read_chunk_queue_limit_event. +pub(crate) fn write_chunk_queue_limit_event( + ft: &FileTransferTracker, add: u64, +) -> Option { + let max_size = unsafe { NFS_CFG_MAX_WRITE_QUEUE_SIZE }; + if max_size != 0 && ft.get_inflight_size() + add > u64::from(max_size) { + return Some(NFSEvent::WriteQueueSizeExceeded); + } + None +} + #[derive(Debug)] pub struct NFSState { state_data: AppLayerStateData, @@ -528,6 +562,11 @@ impl NFSState { tx_old .tx_data .set_event(NFSEvent::TooManyTransactions as u8); + // Close the tracker now: without trunc the evicted file and its + // OOO queue stay live until flow teardown (memory retained). + if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx_old.type_data { + filetracker_trunc(&mut tdf.file_tracker); + } break; } } @@ -1133,16 +1172,20 @@ impl NFSState { return 0; } - let mut queue_event: Option = None; let mut queue_exceeded = false; let found = match self.get_file_tx_by_handle(&file_handle, Direction::ToServer) { Some(tx) => { if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { - if let Some(event) = - write_queue_limit_event(&tdf.file_tracker, w.file_data.len() as u64) - { - queue_event = Some(event); + if let Some(event) = write_queue_limit_event( + &tdf.file_tracker, + w.offset, + w.file_data.len() as u64, + ) { + // Attach to the affected file tx, not the newest tx. + tx.tx_data.set_event(event as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; queue_exceeded = true; } else { filetracker_newchunk( @@ -1178,10 +1221,19 @@ impl NFSState { let tx = self.new_file_tx(&file_handle, &file_name, Direction::ToServer); if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { if let Some(event) = - write_queue_limit_event(&tdf.file_tracker, w.file_data.len() as u64) + write_queue_limit_event(&tdf.file_tracker, w.offset, w.file_data.len() as u64) { - queue_event = Some(event); + // tx just created: set the event on it directly. + tx.tx_data.set_event(event as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; queue_exceeded = true; + // Chunk rejected before the file was opened: complete + close the + // tx and zero files_opened, else it dangles until teardown. + tx.request_done = true; + tx.response_done = true; + tx.is_file_closed = true; + tx.tx_data.0.files_opened = 0; } else { filetracker_newchunk( &mut tdf.file_tracker, @@ -1210,14 +1262,10 @@ impl NFSState { tx.is_first = true; tx.nfs_version = r.progver as u16; } - if let Some(event) = queue_event { - self.set_event(event); - } if !self.is_udp { if queue_exceeded { - // don't buffer the remainder of this record; skip it from - // the stream instead of adding it to the file tracker - let pending = w.count.saturating_sub(w.file_data.len() as u32); + // Don't buffer the rest of the record: skip to its RPC boundary. + let pending = r.prog_data_size.saturating_sub(r.prog_data.len() as u32); self.set_skip(Direction::ToServer, pending); self.ts_chunk_xid = 0; self.ts_chunk_left = 0; @@ -1445,6 +1493,12 @@ impl NFSState { if chunk_left == 0 { return 0; } + // bytes of this call's input that belong to the in-progress chunk + let chunk_total = chunk_left; + let to_client = direction == Direction::ToClient; + // bytes for the in-progress chunk, gap bytes included: gaps entering + // file state count toward queue growth. + let add = std::cmp::min(data.len() as u64, chunk_total as u64); let xid = if direction == Direction::ToServer { self.ts_chunk_xid } else { @@ -1506,66 +1560,156 @@ impl NFSState { } let ssn_gap = self.ts_ssn_gap | self.tc_ssn_gap; + // Whether the OOO-queue backstop truncated this update; raised as an + // event after the tx borrow ends. + let mut truncated = false; + // Chunk-path queue-limit bookkeeping, acted on after the tx borrow + // ends: the event to raise and how many chunk bytes to skip/reset. + let mut limit_event: Option = None; + let mut limit_skip: u32 = 0; + let mut consumed: u32 = 0; // get the tx and update it - let consumed = match self.get_file_tx_by_handle(&file_handle, direction) { + match self.get_file_tx_by_handle(&file_handle, direction) { Some(tx) => { if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { - if ssn_gap { - let queued_data = tdf.file_tracker.get_queued_size(); - if queued_data > 2000000 { - // TODO should probably be configurable - SCLogDebug!( - "QUEUED size {} while we've seen GAPs. Truncating file.", - queued_data - ); + // Chunk-path queue limit: this site is not a newchunk caller, so + // enforce the operator max-{read,write}-queue-size/cnt directly. + // Gap bytes count in `add` (they enter file state and grow the + // queue when the active chunk is OOO); overflow -> event + skip. + if add > 0 && limit_event.is_none() && tdf.file_tracker.current_chunk_is_ooo() { + // Only an active OOO chunk continuation grows the queue; in-order + // data drains it, so it is never rejected here. + let ev = if to_client { + read_chunk_queue_limit_event(&tdf.file_tracker, add) + } else { + write_chunk_queue_limit_event(&tdf.file_tracker, add) + }; + if let Some(e) = ev { + // Attach to the affected file tx (not the newest tx). + tx.tx_data.set_event(e as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; + // XDR padding of the rejected record's chunk, before the + // tracker state is cleared. + let fill = tdf.file_tracker.get_fill_bytes() as u32; + // Drop the rejected chunk: clear queue + in-progress chunk. filetracker_trunc(&mut tdf.file_tracker); + limit_event = Some(e); + if chunk_left == 0 { + // Record completes in this segment: consume chunk data + + // trailing padding together (if the segment ended before + // the padding, the rest is skipped next call). + let total = chunk_total.saturating_add(fill); + consumed = std::cmp::min(data.len() as u32, total); + limit_skip = total.saturating_sub(consumed); + } else { + // All of this segment is chunk data; the promised + // remainder and padding are skipped on later calls. + consumed = data.len() as u32; + limit_skip = chunk_left.saturating_add(fill); + } } } + if limit_event.is_none() { + if ssn_gap { + // Last-resort backstop for out-of-order file data in a + // gapped stream. The configurable queue limits + // (max-{read,write}-queue-size) bound the normal + // enqueuing path; this caps OOO growth when those + // limits are disabled (0) so an attacker cannot drive + // unbounded memory. It is rebased to the configured + // queue size (or a 1 GiB hard cap when disabled) so it + // no longer silently truncates a legitimate transfer + // that is within the operator's queue cap (the old + // hardcoded 2 MiB value undercut the 64 MiB default + // silently). + let queue_size = if direction == Direction::ToClient { + unsafe { NFS_CFG_MAX_READ_QUEUE_SIZE } + } else { + unsafe { NFS_CFG_MAX_WRITE_QUEUE_SIZE } + }; + let backstop = if queue_size != 0 { + u64::from(queue_size) + } else { + 1 << 30 // 1 GiB hard backstop when limits disabled + }; + let queued_data = tdf.file_tracker.get_queued_size(); + if queued_data > backstop { + SCLogDebug!( + "QUEUED size {} > backstop {} while we've seen GAPs. Truncating file.", + queued_data, + backstop + ); + filetracker_trunc(&mut tdf.file_tracker); + truncated = true; + } + } - // reset timestamp if we get called after a gap - if tdf.post_gap_ts > 0 { - tdf.post_gap_ts = 0; - } + // reset timestamp if we get called after a gap + if tdf.post_gap_ts > 0 { + tdf.post_gap_ts = 0; + } - tdf.chunk_count += 1; - let cs = filetracker_update(&mut tdf.file_tracker, data, gap_size); - /* see if we need to close the tx */ - if tdf.file_tracker.is_done() { - if direction == Direction::ToClient { - tx.response_done = true; - tx.is_file_closed = true; - SCLogDebug!( - "TX {} response is done now that the file track is ready", - tx.id - ); - if !flow.is_null() { - sc_app_layer_parser_trigger_raw_stream_inspection( - flow, - Direction::ToClient as i32, + tdf.chunk_count += 1; + let cs = filetracker_update(&mut tdf.file_tracker, data, gap_size); + /* see if we need to close the tx */ + if tdf.file_tracker.is_done() { + if direction == Direction::ToClient { + tx.response_done = true; + tx.is_file_closed = true; + SCLogDebug!( + "TX {} response is done now that the file track is ready", + tx.id ); - } - } else { - tx.request_done = true; - tx.is_file_closed = true; - SCLogDebug!( - "TX {} request is done now that the file track is ready", - tx.id - ); - if !flow.is_null() { - sc_app_layer_parser_trigger_raw_stream_inspection( - flow, - Direction::ToServer as i32, + if !flow.is_null() { + sc_app_layer_parser_trigger_raw_stream_inspection( + flow, + Direction::ToClient as i32, + ); + } + } else { + tx.request_done = true; + tx.is_file_closed = true; + SCLogDebug!( + "TX {} request is done now that the file track is ready", + tx.id ); + if !flow.is_null() { + sc_app_layer_parser_trigger_raw_stream_inspection( + flow, + Direction::ToServer as i32, + ); + } } } + consumed = cs; } - cs } else { - 0 + consumed = 0; } } - None => 0, + None => consumed = 0, }; + + // Chunk-limit overflow: reset chunk state and skip the promised + // remainder (event attached to the file tx under the borrow). + if let Some(_ev) = limit_event { + if to_client { + self.tc_chunk_left = 0; + self.tc_chunk_xid = 0; + self.requestmap.pop(&xid); + } else { + self.ts_chunk_left = 0; + self.ts_chunk_xid = 0; + self.ts_chunk_fh.clear(); + } + if limit_skip > 0 { + self.set_skip(direction, limit_skip); + } + } + if truncated { + self.set_event(NFSEvent::TruncatedFileData); + } return consumed; } @@ -1710,17 +1854,21 @@ impl NFSState { let is_partial = reply.data.len() < reply.count as usize; SCLogDebug!("partial data? {}", is_partial); - let mut queue_event: Option = None; let mut queue_exceeded = false; let found = match self.get_file_tx_by_handle(&file_handle, Direction::ToClient) { Some(tx) => { SCLogDebug!("updated TX {:?}", tx); if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { - if let Some(event) = - read_queue_limit_event(&tdf.file_tracker, reply.data.len() as u64) - { - queue_event = Some(event); + if let Some(event) = read_queue_limit_event( + &tdf.file_tracker, + chunk_offset, + reply.data.len() as u64, + ) { + // Attach to the affected file tx, not the newest tx. + tx.tx_data.set_event(event as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; queue_exceeded = true; } else { filetracker_newchunk( @@ -1768,10 +1916,19 @@ impl NFSState { let tx = self.new_file_tx(&file_handle, &file_name, Direction::ToClient); if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { if let Some(event) = - read_queue_limit_event(&tdf.file_tracker, reply.data.len() as u64) + read_queue_limit_event(&tdf.file_tracker, chunk_offset, reply.data.len() as u64) { - queue_event = Some(event); + // tx just created: set the event on it directly. + tx.tx_data.set_event(event as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; queue_exceeded = true; + // Chunk rejected before the file was opened: complete + close the + // tx and zero files_opened, else it dangles until teardown. + tx.request_done = true; + tx.response_done = true; + tx.is_file_closed = true; + tx.tx_data.0.files_opened = 0; } else { filetracker_newchunk( &mut tdf.file_tracker, @@ -1815,15 +1972,10 @@ impl NFSState { tx.xid = r.hdr.xid; tx.is_first = true; } - if let Some(event) = queue_event { - self.set_event(event); - } - if !self.is_udp { if queue_exceeded { - // don't buffer the remainder of this record; skip it from - // the stream instead of adding it to the file tracker - let pending = reply.count.saturating_sub(reply.data.len() as u32); + // Don't buffer the rest of the record: skip to its RPC boundary. + let pending = r.prog_data_size.saturating_sub(r.prog_data.len() as u32); self.set_skip(Direction::ToClient, pending); self.tc_chunk_xid = 0; self.tc_chunk_left = 0; @@ -1872,6 +2024,11 @@ impl NFSState { pub fn parse_tcp_data_ts_gap(&mut self, gap_size: u32) -> AppLayerResult { SCLogDebug!("parse_tcp_data_ts_gap ({})", gap_size); let consumed = self.handle_skip(Direction::ToServer, gap_size); + // Tag the session gapped even when the skip ate the gap: the OOO + // backstop is gated on this flag and runs on every subsequent + // filetracker_update (SMB parity). + self.ts_ssn_gap = true; + self.ts_gap = true; if consumed == gap_size { /* gap was part of skipped file data: nothing else to do */ return AppLayerResult::ok(); @@ -1879,19 +2036,13 @@ impl NFSState { if consumed < gap_size { let new_gap_size = gap_size - consumed; let gap = vec![0; new_gap_size as usize]; - let consumed2 = self.filetracker_update( + self.filetracker_update( std::ptr::null_mut(), Direction::ToServer, &gap, new_gap_size, ); - if consumed2 == new_gap_size { - /* fully consumed by the file tracker, no need to tag as gap */ - return AppLayerResult::ok(); - } } - self.ts_ssn_gap = true; - self.ts_gap = true; SCLogDebug!("parse_tcp_data_ts_gap ({}) done", gap_size); return AppLayerResult::ok(); } @@ -1899,6 +2050,10 @@ impl NFSState { pub fn parse_tcp_data_tc_gap(&mut self, gap_size: u32) -> AppLayerResult { SCLogDebug!("parse_tcp_data_tc_gap ({})", gap_size); let consumed = self.handle_skip(Direction::ToClient, gap_size); + // Tag the session gapped even when the skip ate the gap (see the TS + // gap handler above). + self.tc_ssn_gap = true; + self.tc_gap = true; if consumed == gap_size { /* gap was part of skipped file data: nothing else to do */ return AppLayerResult::ok(); @@ -1906,19 +2061,13 @@ impl NFSState { if consumed < gap_size { let new_gap_size = gap_size - consumed; let gap = vec![0; new_gap_size as usize]; - let consumed2 = self.filetracker_update( + self.filetracker_update( std::ptr::null_mut(), Direction::ToClient, &gap, new_gap_size, ); - if consumed2 == new_gap_size { - /* fully consumed by the file tracker, no need to tag as gap */ - return AppLayerResult::ok(); - } } - self.tc_ssn_gap = true; - self.tc_gap = true; SCLogDebug!("parse_tcp_data_tc_gap ({}) done", gap_size); return AppLayerResult::ok(); } diff --git a/rust/src/nfs/nfs4.rs b/rust/src/nfs/nfs4.rs index e30a4712df..a53ea5cedb 100644 --- a/rust/src/nfs/nfs4.rs +++ b/rust/src/nfs/nfs4.rs @@ -65,9 +65,8 @@ impl NFSState { Vec::new() }; - // Reject an oversized WRITE: log it as a transaction and skip to - // the record boundary to keep the flow in sync. - // A zero limit disables the check. + // Reject an oversized WRITE: log it as a tx and skip to the record + // boundary to keep the flow in sync. let max_write_size = unsafe { NFS_CFG_MAX_WRITE_SIZE }; if max_write_size != 0 && w.write_len > max_write_size { let tx = self.new_file_tx(&file_handle, &file_name, Direction::ToServer); @@ -75,11 +74,21 @@ impl NFSState { tx.xid = r.hdr.xid; tx.is_first = true; tx.nfs_version = r.progver as u16; + tx.request_done = true; + tx.response_done = true; + // Close the tx so a later same-handle op cannot reuse it. + tx.is_file_closed = true; tx.tx_data.set_event(NFSEvent::WriteRequestTooLarge as u8); + // Complete the rejected transaction immediately: without + // this the tx and its event only surface at flow teardown. + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; // Rejected tx never opened a file: zero files_opened, else the tx // (and file) stays live until teardown and grows unbounded on repeats. tx.tx_data.0.files_opened = 0; - let pending = w.write_len.saturating_sub(w.data.len() as u32); + // Skip to the RPC record boundary: claim vs data can disagree, and + // claimed-count math lands mid-padding or in a later record. + let pending = r.prog_data_size.saturating_sub(r.prog_data.len() as u32); self.set_skip(Direction::ToServer, pending); self.ts_chunk_xid = 0; self.ts_chunk_left = 0; @@ -92,16 +101,18 @@ impl NFSState { return; } - let mut queue_event: Option = None; let mut queue_exceeded = false; let found = match self.get_file_tx_by_handle(&file_handle, Direction::ToServer) { Some(tx) => { if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { if let Some(event) = - write_queue_limit_event(&tdf.file_tracker, w.data.len() as u64) + write_queue_limit_event(&tdf.file_tracker, w.offset, w.data.len() as u64) { - queue_event = Some(event); + // Attach to the affected file tx, not the newest tx. + tx.tx_data.set_event(event as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; queue_exceeded = true; } else { filetracker_newchunk( @@ -129,10 +140,20 @@ impl NFSState { if !found { let tx = self.new_file_tx(&file_handle, &file_name, Direction::ToServer); if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data { - if let Some(event) = write_queue_limit_event(&tdf.file_tracker, w.data.len() as u64) + if let Some(event) = + write_queue_limit_event(&tdf.file_tracker, w.offset, w.data.len() as u64) { - queue_event = Some(event); + // tx just created: set the event on it directly. + tx.tx_data.set_event(event as u8); + tx.tx_data.0.updated_ts = true; + tx.tx_data.0.updated_tc = true; queue_exceeded = true; + // Chunk rejected before the file was opened: complete + close the + // tx and zero files_opened, else it dangles until teardown. + tx.request_done = true; + tx.response_done = true; + tx.is_file_closed = true; + tx.tx_data.0.files_opened = 0; } else { filetracker_newchunk( &mut tdf.file_tracker, @@ -157,13 +178,9 @@ impl NFSState { tx.is_first = true; tx.nfs_version = r.progver as u16; } - if let Some(event) = queue_event { - self.set_event(event); - } if queue_exceeded { - // don't buffer the remainder of this record; skip it from the - // stream instead of adding it to the file tracker - let pending = w.write_len.saturating_sub(w.data.len() as u32); + // Don't buffer the rest of the record: skip to its RPC boundary. + let pending = r.prog_data_size.saturating_sub(r.prog_data.len() as u32); self.set_skip(Direction::ToServer, pending); self.ts_chunk_xid = 0; self.ts_chunk_left = 0;