aiden
/
suricata
mirror of https://github.com/OISF/suricata
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

nfs: enforce the OOO queue limits on the chunk path

Browse Source 
The chunk path (filetracker_update) enqueued per-segment data with no
queue limit. Check the limits on every enqueuing path, projected onto
actual queue growth: only OOO data at a new offset grows the queue,
and only a new OOO chunk adds a map entry. Overflow raises the queue
event on the affected tx and skips to the record boundary.

Replace the 2 MiB gap backstop with a queue-limit one (the chunk-count
cap and all-open-file scan come next), and tag the session gapped on
every gap (even one fully covered by a skip), since the backstop is
gated on that flag. Rejected and evicted tx now release their file
resources so the file logger can clean them up.

Ticket: #8791.
pull/16213/head
Victor Julien 2 months ago committed by Jason Ish
parent 29a5370e71
commit 3a37959005
3 changed files with 590 additions and 166 deletions
Show Stats Download Patch File Download Diff File

370
rust/src/filetracker.rs
Unescape Escape View File

@ -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<u64, FileChunk>,
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);
}
}

339
rust/src/nfs/nfs.rs
Unescape Escape View File

@ -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<NFSEvent> {
/// 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<NFSEvent> {
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<NFSEvent> {
/// Write-direction counterpart of read_queue_limit_event.
pub(crate) fn write_queue_limit_event(
ft: &FileTransferTracker, offset: u64, len: u64,
) -> Option<NFSEvent> {
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<NFSEvent> {
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<NFSEvent> {
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<NFSEvent> = 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<NFSEvent> = 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<NFSEvent> = 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();
}

47
rust/src/nfs/nfs4.rs
Unescape Escape View File

@ -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<NFSEvent> = 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;

Write Preview
Loading…
Cancel
Save