suricata/rust/src/nfs/nfs.rs

/* Copyright (C) 2017-2025 Open Information Security Foundation
 *
 * You can copy, redistribute or modify this Program under the terms of
 * the GNU General Public License version 2 as published by the Free
 * Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

// written by Victor Julien

use lru::LruCache;
use std;
use std::cmp;
use std::ffi::CString;
use std::num::NonZeroUsize;

use nom7::{Err, Needed};
use suricata_sys::sys::{
    AppLayerParserState, AppProto, SCAppLayerParserConfParserEnabled,
    SCAppLayerProtoDetectConfProtoDetectionEnabled, SCAppLayerProtoDetectPPParseConfPorts,
    SCAppLayerProtoDetectPPRegister, SCFileFlowFlagsToFlags,
};

use crate::applayer;
use crate::applayer::*;
use crate::conf::*;
use crate::core::*;
use crate::direction::Direction;
use crate::direction::DIR_BOTH;
use crate::filecontainer::FileContainerWrapper;
use crate::filetracker::*;
use crate::flow::{flow_get_last_time, Flow};
use crate::frames::*;

use crate::nfs::nfs2_records::*;
use crate::nfs::nfs3_records::*;
use crate::nfs::nfs4_records::*;
use crate::nfs::nfs_records::*;
use crate::nfs::rpc_records::*;
use crate::nfs::types::*;

pub static mut SURICATA_NFS_FILE_CONFIG: Option<&'static SuricataFileContext> = None;

pub const NFS_MIN_FRAME_LEN: u16 = 32;

static mut NFS_MAX_TX: usize = 1024;

pub const RPC_TCP_PRE_CREDS: usize = 28;
pub const RPC_UDP_PRE_CREDS: usize = 24;

static mut ALPROTO_NFS: AppProto = ALPROTO_UNKNOWN;
/*
 * Record parsing.
 *
 * Incomplete records come in due to TCP splicing. For all record types
 * except READ and WRITE, processing only begins when the full record
 * is available. For READ/WRITE partial records are processed as well to
 * avoid queuing too much data.
 *
 * Getting file names.
 *
 * NFS makes heavy use of 'file handles' for operations. In many cases it
 * uses a file name just once and after that just the handle. For example,
 * if a client did a file listing (e.g. READDIRPLUS) and would READ the
 * file afterwards, the name will only appear in the READDIRPLUS answer.
 * To be able to log the names we store a mapping between file handles
 * and file names in NFSState::namemap.
 *
 * Mapping NFS to Suricata's transaction model.
 *
 * The easiest way to do transactions would be to map each command/reply with
 * the same XID to a transaction. This would allow for per XID logging, detect
 * etc. However this model doesn't fit well with file tracking. The file
 * tracking in Suricata is really expecting to be one or more files to live
 * inside a single transaction. Would XID pairs be a transaction however,
 * there would be many transactions forming a single file. This will be very
 * inefficient.
 *
 * The model implemented here is as follows: each file transfer is a single
 * transaction. All XID pairs unrelated to those file transfers create
 * transactions per pair.
 *
 * A complicating factor is that the procedure matching is per tx, and a
 * file transfer may have multiple procedures involved. Currently now only
 * a COMMIT after WRITEs. A vector of additional procedures is kept to
 * match on this.
 *
 * File tracking
 *
 * Files are tracked per 'FileTransferTracker' and are stored in the
 * NFSTransaction where they can be looked up per handle as part of the
 * Transaction lookup.
 */

pub static mut NFS_CFG_MAX_REQ: usize = 512;
pub static mut NFS_CFG_MAX_NAMES: usize = 512;

/// Max queued (out-of-order) read file data (bytes). 0 = no limit.
pub static mut NFS_CFG_MAX_READ_QUEUE_SIZE: u32 = 67108864;
/// Max queued (out-of-order) read chunks. 0 = no limit.
pub static mut NFS_CFG_MAX_READ_QUEUE_CNT: u32 = 64;
/// Max queued (out-of-order) write file data (bytes). 0 = no limit.
pub static mut NFS_CFG_MAX_WRITE_QUEUE_SIZE: u32 = 67108864;
/// Max queued (out-of-order) write chunks. 0 = no limit.
pub static mut NFS_CFG_MAX_WRITE_QUEUE_CNT: u32 = 64;
/// Max claimed size of a single WRITE request (bytes). Larger requests
/// are rejected with an event and their payload remainder skipped.
pub static mut NFS_CFG_MAX_WRITE_SIZE: u32 = 16777216;
/// Max data size of a single READ response (bytes). Larger responses
/// are rejected with an event.
pub static mut NFS_CFG_MAX_READ_SIZE: u32 = 16777216;

#[derive(AppLayerFrameType)]
pub enum NFSFrameType {
    RPCPdu,
    RPCHdr,
    RPCData,
    RPCCreds, // for rpc calls | rpc.creds [creds_flavor + creds_len + creds]

    NFSPdu,
    NFSStatus,

    NFS4Pdu,
    NFS4Hdr,
    NFS4Ops,
    NFS4Status,
}

#[derive(FromPrimitive, Debug, AppLayerEvent, Clone, Copy)]
pub enum NFSEvent {
    MalformedData = 0,
    NonExistingVersion = 1,
    UnsupportedVersion = 2,
    TooManyTransactions = 3,
    /// Queued (out-of-order) read file data exceeds `max-read-queue-size`
    ReadQueueSizeExceeded = 4,
    /// Number of queued (out-of-order) read chunks exceeds `max-read-queue-cnt`
    ReadQueueCntExceeded = 5,
    /// Queued (out-of-order) write file data exceeds `max-write-queue-size`
    WriteQueueSizeExceeded = 6,
    /// Number of queued (out-of-order) write chunks exceeds `max-write-queue-cnt`
    WriteQueueCntExceeded = 7,
    /// WRITE request claims more data than the configured `max-write-size`
    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)]
pub enum NFSTransactionTypeData {
    RENAME(Vec<u8>),
    FILE(NFSTransactionFile),
}

#[derive(Default, Debug)]
pub struct NFSTransactionFile {
    /// file transactions are unidirectional in the sense that they track
    /// a single file on one direction
    pub direction: Direction, // Direction::ToClient or Direction::ToServer

    /// additional procedures part of a single file transfer. Currently
    /// only COMMIT on WRITEs.
    pub file_additional_procs: Vec<u32>,

    pub chunk_count: u32,

    /// last xid of this file transfer. Last READ or COMMIT normally.
    pub file_last_xid: u32,

    /// after a gap, this will be set to a time in the future. If the file
    /// receives no updates before that, it will be considered complete.
    pub post_gap_ts: u64,

    /// file tracker for a single file. Boxed so that we don't use
    /// as much space if we're not a file tx.
    pub file_tracker: FileTransferTracker,
}

impl NFSTransactionFile {
    pub fn new() -> Self {
        return Self {
            file_tracker: FileTransferTracker::new(),
            ..Default::default()
        };
    }
    pub fn update_file_flags(&mut self, flow_file_flags: u16) {
        let dir_flag = if self.direction == Direction::ToServer {
            STREAM_TOSERVER
        } else {
            STREAM_TOCLIENT
        };
        self.file_tracker.file_flags = unsafe { SCFileFlowFlagsToFlags(flow_file_flags, dir_flag) };
    }
}

unsafe extern "C" fn nfs_gettxfiles(
    tx: *mut std::ffi::c_void, direction: u8,
) -> AppLayerGetFileState {
    let tx = cast_pointer!(tx, NFSTransaction);
    if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data {
        let tx_dir: u8 = tdf.direction.into();
        if direction & tx_dir != 0 {
            if let Some(sfcm) = SURICATA_NFS_FILE_CONFIG {
                return AppLayerGetFileState {
                    fc: &mut tdf.file_tracker.file,
                    cfg: sfcm.files_sbcfg,
                };
            }
        }
    }
    AppLayerGetFileState::err()
}

#[derive(Debug)]
pub struct NFSTransaction {
    pub id: u64,
    /// internal id
    pub xid: u32,
    /// nfs req/reply pair id
    pub procedure: u32,
    /// file name of the object we're dealing with. In case of RENAME
    /// this is the 'from' or original name.
    pub file_name: Vec<u8>,

    pub auth_type: u32,
    pub request_machine_name: Vec<u8>,
    pub request_uid: u32,
    pub request_gid: u32,

    pub rpc_response_status: u32,
    pub nfs_response_status: u32,

    pub is_first: bool,
    pub is_last: bool,

    /// for state tracking. false means this side is in progress, true
    /// that it's complete.
    pub request_done: bool,
    pub response_done: bool,

    pub nfs_version: u16,

    /// is a special file tx that we look up by file_handle instead of XID
    pub is_file_tx: bool,
    pub is_file_closed: bool,
    pub file_handle: Vec<u8>,

    /// Procedure type specific data
    /// TODO see if this can be an `Option<Box<NFSTransactionTypeData>>`. Initial
    /// attempt failed.
    pub type_data: Option<NFSTransactionTypeData>,

    pub tx_data: AppLayerTxData,
}

impl Default for NFSTransaction {
    fn default() -> Self {
        Self::new()
    }
}

impl NFSTransaction {
    pub fn new() -> Self {
        return Self {
            id: 0,
            xid: 0,
            procedure: 0,
            file_name: Vec::new(),
            request_machine_name: Vec::new(),
            request_uid: 0,
            request_gid: 0,
            rpc_response_status: 0,
            nfs_response_status: 0,
            auth_type: 0,
            is_first: false,
            is_last: false,
            request_done: false,
            response_done: false,
            nfs_version: 0,
            is_file_tx: false,
            is_file_closed: false,
            file_handle: Vec::new(),
            type_data: None,
            tx_data: AppLayerTxData::new(),
        };
    }

    pub fn free(&mut self) {
        debug_validate_bug_on!(self.tx_data.0.files_opened > 1);
        debug_validate_bug_on!(self.tx_data.0.files_logged > 1);
    }
}

impl Transaction for NFSTransaction {
    fn id(&self) -> u64 {
        self.id
    }
}

impl Drop for NFSTransaction {
    fn drop(&mut self) {
        if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = self.type_data {
            if let Some(sfcm) = unsafe { SURICATA_NFS_FILE_CONFIG } {
                tdf.file_tracker.file.free(sfcm);
            }
        }
        self.free();
    }
}

#[derive(Debug)]
pub struct NFSRequestXidMap {
    pub progver: u32,
    pub procedure: u32,
    pub chunk_offset: u64,
    pub file_name: Vec<u8>,

    /// READ replies can use this to get to the handle the request used
    pub file_handle: Vec<u8>,

    pub gssapi_proc: u32,
    pub gssapi_service: u32,
}

impl NFSRequestXidMap {
    pub fn new(progver: u32, procedure: u32, chunk_offset: u64) -> NFSRequestXidMap {
        NFSRequestXidMap {
            progver,
            procedure,
            chunk_offset,
            file_name: Vec::new(),
            file_handle: Vec::new(),
            gssapi_proc: 0,
            gssapi_service: 0,
        }
    }
}

/// little wrapper around the FileTransferTracker::new_chunk method
pub fn filetracker_newchunk(
    ft: &mut FileTransferTracker, name: &[u8], data: &[u8], chunk_offset: u64, chunk_size: u32,
    fill_bytes: u8, is_last: bool, xid: &u32,
) {
    if let Some(sfcm) = unsafe { SURICATA_NFS_FILE_CONFIG } {
        ft.new_chunk(
            sfcm,
            name,
            data,
            chunk_offset,
            chunk_size,
            fill_bytes,
            is_last,
            xid,
        );
    }
}

fn filetracker_trunc(ft: &mut FileTransferTracker) {
    if let Some(sfcm) = unsafe { SURICATA_NFS_FILE_CONFIG } {
        ft.trunc(sfcm);
    }
}

pub fn filetracker_close(ft: &mut FileTransferTracker) {
    if let Some(sfcm) = unsafe { SURICATA_NFS_FILE_CONFIG } {
        ft.close(sfcm);
    }
}

fn filetracker_update(ft: &mut FileTransferTracker, data: &[u8], gap_size: u32) -> u32 {
    if let Some(sfcm) = unsafe { SURICATA_NFS_FILE_CONFIG } {
        ft.update(sfcm, data, gap_size)
    } else {
        0
    }
}

/// Event to raise if enqueuing `len` bytes at `offset` (read direction) would
/// exceed the limits. OOO data at a new offset grows the queue; in-order appends drain it.
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.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.new_chunk_entry(offset) && ft.get_inflight_cnt() + 1 > max_cnt as usize {
        return Some(NFSEvent::ReadQueueCntExceeded);
    }
    None
}

/// 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.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.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
}

/// OOO-queue backstop (bytes): configured queue size, or a 1 GiB hard
/// cap (16x the 64 MiB default) when limits are disabled.
pub(crate) fn ooo_queue_backstop(direction: Direction) -> u64 {
    let queue_size = if direction == Direction::ToClient {
        unsafe { NFS_CFG_MAX_READ_QUEUE_SIZE }
    } else {
        unsafe { NFS_CFG_MAX_WRITE_QUEUE_SIZE }
    };
    if queue_size != 0 {
        u64::from(queue_size)
    } else {
        1 << 30 // 1 GiB hard backstop when the limits are disabled
    }
}

/// OOO-queue backstop (chunk count): configured count, or a 1024 hard cap
/// (16x the default) when limits are off; chunks carry per-entry overhead.
pub(crate) fn ooo_queue_backstop_cnt(direction: Direction) -> u64 {
    let queue_cnt = if direction == Direction::ToClient {
        unsafe { NFS_CFG_MAX_READ_QUEUE_CNT }
    } else {
        unsafe { NFS_CFG_MAX_WRITE_QUEUE_CNT }
    };
    if queue_cnt != 0 {
        u64::from(queue_cnt)
    } else {
        1024 // hard chunk-count backstop when the limits are disabled
    }
}

#[derive(Debug)]
pub struct NFSState {
    state_data: AppLayerStateData,

    /// map xid to procedure so replies can lookup the procedure
    pub requestmap: LruCache<u32, NFSRequestXidMap>,

    /// map file handle (1) to name (2)
    pub namemap: LruCache<Vec<u8>, Vec<u8>>,

    /// transactions list
    pub transactions: Vec<NFSTransaction>,

    /// partial record tracking
    pub ts_chunk_xid: u32,
    pub tc_chunk_xid: u32,
    /// size of the current chunk that we still need to receive
    pub ts_chunk_left: u32,
    pub tc_chunk_left: u32,
    /// unbuffered tail of the in-progress chunk's RPC record. Tracked
    /// separately from the count-based chunk promise (file_len <= count):
    /// a rejected continuation must never consume/skip past the record
    /// boundary into the next RPC.
    pub ts_chunk_tail: u32,
    pub tc_chunk_tail: u32,
    /// file handle of in progress toserver WRITE file chunk
    ts_chunk_fh: Vec<u8>,
    /// file handle of in progress toclient READ file chunk. Kept here as
    /// the requestmap is bounded and may evict the READ's xid before the
    /// chunk is complete.
    tc_chunk_fh: Vec<u8>,

    /// number of stream bytes to skip (queued file data over the queue limit)
    skip_ts: u32,
    skip_tc: u32,

    ts_ssn_gap: bool,
    tc_ssn_gap: bool,

    ts_gap: bool, // last TS update was gap
    tc_gap: bool, // last TC update was gap

    is_udp: bool,

    /// true as long as we have file txs that are in a post-gap
    /// state. It means we'll do extra house keeping for those.
    check_post_gap_file_txs: bool,
    post_gap_files_checked: bool,

    pub nfs_version: u16,

    /// tx counter for assigning incrementing id's to tx's
    tx_id: u64,

    /// Timestamp in seconds of last update. This is packet time,
    /// potentially coming from pcaps.
    ts: u64,
}

impl Default for NFSState {
    fn default() -> Self {
        Self::new()
    }
}

impl State<NFSTransaction> for NFSState {
    fn get_transaction_count(&self) -> usize {
        self.transactions.len()
    }

    fn get_transaction_by_index(&self, index: usize) -> Option<&NFSTransaction> {
        self.transactions.get(index)
    }
}

impl NFSState {
    /// Allocation function for a new TLS parser instance
    pub fn new() -> NFSState {
        NFSState {
            state_data: AppLayerStateData::default(),
            requestmap: LruCache::new(NonZeroUsize::new(unsafe { NFS_CFG_MAX_REQ }).unwrap()),
            namemap: LruCache::new(NonZeroUsize::new(unsafe { NFS_CFG_MAX_NAMES }).unwrap()),
            transactions: Vec::new(),
            ts_chunk_xid: 0,
            tc_chunk_xid: 0,
            skip_ts: 0,
            skip_tc: 0,
            ts_chunk_left: 0,
            tc_chunk_left: 0,
            ts_chunk_tail: 0,
            tc_chunk_tail: 0,
            ts_chunk_fh: Vec::new(),
            tc_chunk_fh: Vec::new(),
            ts_ssn_gap: false,
            tc_ssn_gap: false,
            ts_gap: false,
            tc_gap: false,
            is_udp: false,
            check_post_gap_file_txs: false,
            post_gap_files_checked: false,
            nfs_version: 0,
            tx_id: 0,
            ts: 0,
        }
    }

    fn update_ts(&mut self, ts: u64) {
        if ts != self.ts {
            self.ts = ts;
            self.post_gap_files_checked = false;
        }
    }

    pub fn new_tx(&mut self) -> NFSTransaction {
        let mut tx = NFSTransaction::new();
        self.tx_id += 1;
        tx.id = self.tx_id;
        if self.transactions.len() > unsafe { NFS_MAX_TX } {
            // set at least one another transaction to the drop state
            for tx_old in &mut self.transactions {
                if !tx_old.request_done || !tx_old.response_done {
                    tx_old.tx_data.0.updated_tc = true;
                    tx_old.tx_data.0.updated_ts = true;
                    tx_old.request_done = true;
                    tx_old.response_done = true;
                    tx_old.is_file_closed = true;
                    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;
                }
            }
        }
        return tx;
    }

    pub fn free_tx(&mut self, tx_id: u64) {
        //SCLogNotice!("Freeing TX with ID {}", tx_id);
        let len = self.transactions.len();
        let mut found = false;
        let mut index = 0;
        for i in 0..len {
            let tx = &self.transactions[i];
            if tx.id == tx_id + 1 {
                found = true;
                index = i;
                break;
            }
        }
        if found {
            SCLogDebug!("freeing TX with ID {} at index {}", tx_id, index);
            self.transactions.remove(index);
        }
    }

    pub fn get_tx_by_id(&mut self, tx_id: u64) -> Option<&NFSTransaction> {
        return self.transactions.iter().find(|&tx| tx.id == tx_id + 1);
    }

    pub fn get_tx_by_xid(&mut self, tx_xid: u32) -> Option<&mut NFSTransaction> {
        return self
            .transactions
            .iter_mut()
            .find(|tx| !tx.is_file_tx && tx.xid == tx_xid);
    }

    /// Set an event. The event is set on the most recent transaction.
    pub fn set_event(&mut self, event: NFSEvent) {
        let len = self.transactions.len();
        if len == 0 {
            return;
        }

        let tx = &mut self.transactions[len - 1];
        tx.tx_data.set_event(event as u8);
    }

    /// Set how many stream bytes to skip (queued file data over the
    /// queue limit, not buffered).
    pub fn set_skip(&mut self, direction: Direction, skip: u32) {
        if direction == Direction::ToServer {
            self.skip_ts = self.skip_ts.saturating_add(skip);
        } else {
            self.skip_tc = self.skip_tc.saturating_add(skip);
        }
    }

    /// Consume up to `input_size` bytes of pending skip. Returns how
    /// much we consumed.
    fn handle_skip(&mut self, direction: Direction, input_size: u32) -> u32 {
        let skip_left = if direction == Direction::ToServer {
            self.skip_ts
        } else {
            self.skip_tc
        };
        if skip_left == 0 {
            return 0;
        }
        SCLogDebug!("skip_left {} input_size {}", skip_left, input_size);

        let consumed = if skip_left >= input_size {
            input_size
        } else {
            skip_left
        };

        let remaining = skip_left.saturating_sub(input_size);
        if direction == Direction::ToServer {
            self.skip_ts = remaining;
        } else {
            self.skip_tc = remaining;
        }
        return consumed;
    }

    // TODO maybe not enough users to justify a func
    pub fn mark_response_tx_done(
        &mut self, flow: *mut Flow, xid: u32, rpc_status: u32, nfs_status: u32, resp_handle: &[u8],
    ) {
        if let Some(mytx) = self.get_tx_by_xid(xid) {
            mytx.tx_data.0.updated_tc = true;
            mytx.tx_data.0.updated_ts = true;
            mytx.response_done = true;
            mytx.rpc_response_status = rpc_status;
            mytx.nfs_response_status = nfs_status;
            if mytx.file_handle.is_empty() && !resp_handle.is_empty() {
                mytx.file_handle = resp_handle.to_vec();
            }

            SCLogDebug!(
                "process_reply_record: tx ID {} XID {:04X} REQUEST {} RESPONSE {}",
                mytx.id,
                mytx.xid,
                mytx.request_done,
                mytx.response_done
            );
            sc_app_layer_parser_trigger_raw_stream_inspection(flow, Direction::ToClient as i32);
        } else {
            //SCLogNotice!("process_reply_record: not TX found for XID {}", r.hdr.xid);
        }
    }

    fn add_rpc_udp_ts_pdu(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64,
    ) -> Option<Frame> {
        let rpc_udp_ts_pdu = Frame::new(
            flow,
            stream_slice,
            input,
            rpc_len,
            NFSFrameType::RPCPdu as u8,
            None,
        );
        SCLogDebug!("rpc_udp_pdu ts frame {:?}", rpc_udp_ts_pdu);
        rpc_udp_ts_pdu
    }

    fn add_rpc_udp_ts_creds(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], creds_len: i64,
    ) {
        let _rpc_udp_ts_creds = Frame::new(
            flow,
            stream_slice,
            input,
            creds_len,
            NFSFrameType::RPCCreds as u8,
            None,
        );
        SCLogDebug!("rpc_creds ts frame {:?}", _rpc_udp_ts_creds);
    }

    fn add_rpc_tcp_ts_pdu(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64,
    ) -> Option<Frame> {
        let rpc_tcp_ts_pdu = Frame::new(
            flow,
            stream_slice,
            input,
            rpc_len,
            NFSFrameType::RPCPdu as u8,
            None,
        );
        SCLogDebug!("rpc_tcp_pdu ts frame {:?}", rpc_tcp_ts_pdu);
        rpc_tcp_ts_pdu
    }

    fn add_rpc_tcp_ts_creds(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], creds_len: i64,
    ) {
        let _rpc_tcp_ts_creds = Frame::new(
            flow,
            stream_slice,
            input,
            creds_len,
            NFSFrameType::RPCCreds as u8,
            None,
        );
        SCLogDebug!("rpc_tcp_ts_creds {:?}", _rpc_tcp_ts_creds);
    }

    fn add_nfs_ts_frame(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], nfs_len: i64,
    ) {
        let _nfs_req_pdu = Frame::new(
            flow,
            stream_slice,
            input,
            nfs_len,
            NFSFrameType::NFSPdu as u8,
            None,
        );
        SCLogDebug!("nfs_ts_pdu Frame {:?}", _nfs_req_pdu);
    }

    fn add_nfs4_ts_frames(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], nfs4_len: i64,
    ) {
        let _nfs4_ts_pdu = Frame::new(
            flow,
            stream_slice,
            input,
            nfs4_len,
            NFSFrameType::NFS4Pdu as u8,
            None,
        );
        SCLogDebug!("nfs4_ts_pdu Frame: {:?}", _nfs4_ts_pdu);
        if nfs4_len > 8 {
            let _nfs4_ts_hdr = Frame::new(
                flow,
                stream_slice,
                input,
                8,
                NFSFrameType::NFS4Hdr as u8,
                None,
            );
            SCLogDebug!("nfs4_ts_hdr Frame {:?}", _nfs4_ts_hdr);
            let _nfs4_ts_ops = Frame::new(
                flow,
                stream_slice,
                &input[8..],
                nfs4_len - 8,
                NFSFrameType::NFS4Ops as u8,
                None,
            );
            SCLogDebug!("nfs4_ts_ops Frame {:?}", _nfs4_ts_ops);
        }
    }

    fn add_rpc_udp_tc_pdu(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64,
    ) -> Option<Frame> {
        let rpc_udp_tc_pdu = Frame::new(
            flow,
            stream_slice,
            input,
            rpc_len,
            NFSFrameType::RPCPdu as u8,
            None,
        );
        SCLogDebug!("rpc_tc_pdu frame {:?}", rpc_udp_tc_pdu);
        rpc_udp_tc_pdu
    }

    fn add_rpc_udp_tc_frames(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], rpc_len: i64,
    ) {
        if rpc_len > 8 {
            let _rpc_udp_tc_hdr = Frame::new(
                flow,
                stream_slice,
                input,
                8,
                NFSFrameType::RPCHdr as u8,
                None,
            );
            let _rpc_udp_tc_data = Frame::new(
                flow,
                stream_slice,
                &input[8..],
                rpc_len - 8,
                NFSFrameType::RPCData as u8,
                None,
            );
            SCLogDebug!("rpc_udp_tc_hdr frame {:?}", _rpc_udp_tc_hdr);
            SCLogDebug!("rpc_udp_tc_data frame {:?}", _rpc_udp_tc_data);
        }
    }

    fn add_rpc_tcp_tc_pdu(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], rpc_tcp_len: i64,
    ) -> Option<Frame> {
        let rpc_tcp_tc_pdu = Frame::new(
            flow,
            stream_slice,
            input,
            rpc_tcp_len,
            NFSFrameType::RPCPdu as u8,
            None,
        );
        SCLogDebug!("rpc_tcp_pdu tc frame {:?}", rpc_tcp_tc_pdu);
        rpc_tcp_tc_pdu
    }

    fn add_rpc_tcp_tc_frames(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], rpc_tcp_len: i64,
    ) {
        if rpc_tcp_len > 12 {
            let _rpc_tcp_tc_hdr = Frame::new(
                flow,
                stream_slice,
                input,
                12,
                NFSFrameType::RPCHdr as u8,
                None,
            );
            let _rpc_tcp_tc_data = Frame::new(
                flow,
                stream_slice,
                &input[12..],
                rpc_tcp_len - 12,
                NFSFrameType::RPCData as u8,
                None,
            );
            SCLogDebug!("rpc_tcp_tc_hdr frame {:?}", _rpc_tcp_tc_hdr);
            SCLogDebug!("rpc_tcp_tc_data frame {:?}", _rpc_tcp_tc_data);
        }
    }

    fn add_nfs_tc_frames(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], nfs_len: i64,
    ) {
        if nfs_len > 0 {
            let _nfs_tc_pdu = Frame::new(
                flow,
                stream_slice,
                input,
                nfs_len,
                NFSFrameType::NFSPdu as u8,
                None,
            );
            SCLogDebug!("nfs_tc_pdu frame {:?}", _nfs_tc_pdu);
            let _nfs_res_status = Frame::new(
                flow,
                stream_slice,
                input,
                4,
                NFSFrameType::NFSStatus as u8,
                None,
            );
            SCLogDebug!("nfs_tc_status frame {:?}", _nfs_res_status);
        }
    }

    fn add_nfs4_tc_frames(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, input: &[u8], nfs4_len: i64,
    ) {
        if nfs4_len > 0 {
            let _nfs4_tc_pdu = Frame::new(
                flow,
                stream_slice,
                input,
                nfs4_len,
                NFSFrameType::NFS4Pdu as u8,
                None,
            );
            SCLogDebug!("nfs4_tc_pdu frame {:?}", _nfs4_tc_pdu);
            let _nfs4_tc_status = Frame::new(
                flow,
                stream_slice,
                input,
                4,
                NFSFrameType::NFS4Status as u8,
                None,
            );
            SCLogDebug!("nfs4_tc_status frame {:?}", _nfs4_tc_status);
        }
        if nfs4_len > 8 {
            let _nfs4_tc_hdr = Frame::new(
                flow,
                stream_slice,
                input,
                8,
                NFSFrameType::NFS4Hdr as u8,
                None,
            );
            SCLogDebug!("nfs4_tc_hdr frame {:?}", _nfs4_tc_hdr);
            let _nfs4_tc_ops = Frame::new(
                flow,
                stream_slice,
                &input[8..],
                nfs4_len - 8,
                NFSFrameType::NFS4Ops as u8,
                None,
            );
            SCLogDebug!("nfs4_tc_ops frame {:?}", _nfs4_tc_ops);
        }
    }

    fn post_gap_housekeeping_for_files(&mut self) {
        let mut post_gap_txs = false;
        for tx in &mut self.transactions {
            if let Some(NFSTransactionTypeData::FILE(ref mut f)) = tx.type_data {
                if f.post_gap_ts > 0 {
                    if self.ts > f.post_gap_ts {
                        tx.request_done = true;
                        tx.response_done = true;
                        filetracker_trunc(&mut f.file_tracker);
                    } else {
                        post_gap_txs = true;
                    }
                }
            }
        }
        self.check_post_gap_file_txs = post_gap_txs;
    }

    /* after a gap we will consider all transactions complete for our
     * direction. File transfer transactions are an exception. Those
     * can handle gaps. For the file transactions we set the current
     * (flow) time and prune them in 60 seconds if no update for them
     * was received. */
    fn post_gap_housekeeping(&mut self, dir: Direction) {
        if self.ts_ssn_gap && dir == Direction::ToServer {
            for tx in &mut self.transactions {
                if tx.id >= self.tx_id {
                    SCLogDebug!("post_gap_housekeeping: done");
                    break;
                }
                if let Some(NFSTransactionTypeData::FILE(ref mut f)) = tx.type_data {
                    // leaving FILE txs open as they can deal with gaps. We
                    // remove them after 60 seconds of no activity though.
                    if f.post_gap_ts == 0 {
                        f.post_gap_ts = self.ts + 60;
                        self.check_post_gap_file_txs = true;
                    }
                } else {
                    SCLogDebug!("post_gap_housekeeping: tx {} marked as done TS", tx.id);
                    tx.request_done = true;
                }
            }
        } else if self.tc_ssn_gap && dir == Direction::ToClient {
            for tx in &mut self.transactions {
                if tx.id >= self.tx_id {
                    SCLogDebug!("post_gap_housekeeping: done");
                    break;
                }
                if let Some(NFSTransactionTypeData::FILE(ref mut f)) = tx.type_data {
                    // leaving FILE txs open as they can deal with gaps. We
                    // remove them after 60 seconds of no activity though.
                    if f.post_gap_ts == 0 {
                        f.post_gap_ts = self.ts + 60;
                        self.check_post_gap_file_txs = true;
                    }
                } else {
                    SCLogDebug!("post_gap_housekeeping: tx {} marked as done TC", tx.id);
                    tx.request_done = true;
                    tx.response_done = true;
                }
            }
        }
    }

    pub fn process_request_record_lookup(&mut self, r: &RpcPacket, xidmap: &mut NFSRequestXidMap) {
        match parse_nfs3_request_lookup(r.prog_data) {
            Ok((_, lookup)) => {
                SCLogDebug!("LOOKUP {:?}", lookup);
                xidmap.file_name = lookup.name_vec;
            }
            _ => {
                self.set_event(NFSEvent::MalformedData);
            }
        };
    }

    pub fn xidmap_handle2name(&mut self, xidmap: &mut NFSRequestXidMap) {
        if let Some(n) = self.namemap.get(&xidmap.file_handle) {
            SCLogDebug!("xidmap_handle2name: name {:?}", n);
            xidmap.file_name = n.to_vec();
        } else {
            SCLogDebug!(
                "xidmap_handle2name: object {:?} not found",
                xidmap.file_handle
            );
        }
    }

    /// complete request record
    fn process_request_record(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, r: &RpcPacket,
    ) {
        SCLogDebug!(
            "REQUEST {} procedure {} ({}) blob size {}",
            r.hdr.xid,
            r.procedure,
            self.requestmap.len(),
            r.prog_data.len()
        );

        match r.progver {
            4 => {
                self.add_nfs4_ts_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                self.process_request_record_v4(r)
            }
            3 => {
                self.add_nfs_ts_frame(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                self.process_request_record_v3(flow, r)
            }
            2 => {
                self.add_nfs_ts_frame(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                self.process_request_record_v2(r)
            }
            _ => {}
        }
    }

    pub fn new_file_tx(
        &mut self, file_handle: &[u8], file_name: &[u8], direction: Direction,
    ) -> &mut NFSTransaction {
        let mut tx = self.new_tx();
        tx.file_name = file_name.to_vec();
        tx.file_handle = file_handle.to_vec();
        tx.is_file_tx = true;

        tx.type_data = Some(NFSTransactionTypeData::FILE(NFSTransactionFile::new()));
        if let Some(NFSTransactionTypeData::FILE(ref mut d)) = tx.type_data {
            d.direction = direction;
            d.file_tracker.tx_id = tx.id - 1;
            tx.tx_data.update_file_flags(self.state_data.file_flags);
            d.update_file_flags(tx.tx_data.0.file_flags);
        }
        tx.tx_data.init_files_opened();
        tx.tx_data.0.file_tx = if direction == Direction::ToServer {
            STREAM_TOSERVER
        } else {
            STREAM_TOCLIENT
        }; // TODO direction to flag func?
        SCLogDebug!(
            "new_file_tx: TX FILE created: ID {} NAME {}",
            tx.id,
            String::from_utf8_lossy(file_name)
        );
        self.transactions.push(tx);
        let tx_ref = self.transactions.last_mut();
        return tx_ref.unwrap();
    }

    pub fn get_file_tx_by_handle(
        &mut self, file_handle: &[u8], direction: Direction,
    ) -> Option<&mut NFSTransaction> {
        let fh = file_handle.to_vec();
        for tx in &mut self.transactions {
            if let Some(NFSTransactionTypeData::FILE(ref mut d)) = tx.type_data {
                if tx.is_file_tx
                    && !tx.is_file_closed
                    && direction == d.direction
                    && tx.file_handle == fh
                {
                    tx.tx_data.update_file_flags(self.state_data.file_flags);
                    d.update_file_flags(tx.tx_data.0.file_flags);
                    SCLogDebug!("Found NFS file TX with ID {} XID {:04X}", tx.id, tx.xid);
                    tx.tx_data.0.updated_tc = true;
                    tx.tx_data.0.updated_ts = true;
                    return Some(tx);
                }
            }
        }
        SCLogDebug!("Failed to find NFS TX with handle {:?}", file_handle);
        return None;
    }

    pub fn process_write_record<'b>(
        &mut self, flow: *mut Flow, r: &RpcPacket<'b>, w: &Nfs3RequestWrite<'b>,
    ) -> u32 {
        let mut fill_bytes = 0;
        // the parser derives the XDR padding from file_len (the data that is
        // actually on the wire), not from the claimed count
        let pad = w.file_len % 4;
        if pad != 0 {
            fill_bytes = 4 - pad;
        }

        // for now assume that stable FILE_SYNC flags means a single chunk
        let is_last = w.stable == 2;
        // a partial record's file tx must stay open (found by handle) until
        // filetracker_update has received the rest of its data
        let is_partial = w.file_data.len() < w.file_len as usize;
        let file_handle = w.handle.value.to_vec();
        let file_name = if let Some(name) = self.namemap.get(w.handle.value) {
            SCLogDebug!("WRITE name {:?}", name);
            name.to_vec()
        } else {
            SCLogDebug!("WRITE object {:?} not found", w.handle.value);
            Vec::new()
        };

        if w.count == 0 {
            return 0;
        }

        // Oversized WRITE: log + skip to the record boundary to stay in sync.
        // A zero limit disables the check.
        let max_write_size = unsafe { NFS_CFG_MAX_WRITE_SIZE };
        if max_write_size != 0 && w.count > max_write_size {
            let tx = self.new_file_tx(&file_handle, &file_name, Direction::ToServer);
            tx.procedure = NFSPROC3_WRITE;
            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 file transaction so a later same-handle
            // operation cannot reuse this rejected one.
            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;
            // 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);
            if !self.is_udp {
                self.set_skip(Direction::ToServer, pending);
                self.ts_chunk_xid = 0;
                self.ts_chunk_left = 0;
                self.ts_chunk_tail = 0;
                self.ts_chunk_fh.clear();
            }
            SCLogDebug!(
                "WRITE too large (claim {} > {}): skipping {} stream bytes",
                w.count,
                unsafe { NFS_CFG_MAX_WRITE_SIZE },
                pending
            );
            return 0;
        }

        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.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(
                            &mut tdf.file_tracker,
                            &file_name,
                            w.file_data,
                            w.offset,
                            w.file_len,
                            fill_bytes as u8,
                            is_last,
                            &r.hdr.xid,
                        );
                        tdf.chunk_count += 1;
                        if is_last {
                            tdf.file_last_xid = r.hdr.xid;
                            tx.is_last = true;
                            tx.response_done = true;
                            if !is_partial {
                                tx.is_file_closed = true;
                            }
                            sc_app_layer_parser_trigger_raw_stream_inspection(
                                flow,
                                Direction::ToClient as i32,
                            );
                        }
                    }
                    true
                } else {
                    false
                }
            }
            None => false,
        };
        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.offset, w.file_data.len() as u64)
                {
                    // 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,
                        &file_name,
                        w.file_data,
                        w.offset,
                        w.file_len,
                        fill_bytes as u8,
                        is_last,
                        &r.hdr.xid,
                    );
                }
                if is_last && !queue_exceeded {
                    tdf.file_last_xid = r.hdr.xid;
                    tx.is_last = true;
                    /* if this is a partial record we will close the tx
                     * when we've received the final data */
                    if !is_partial {
                        tx.request_done = true;
                        tx.is_file_closed = true;
                        sc_app_layer_parser_trigger_raw_stream_inspection(
                            flow,
                            Direction::ToServer as i32,
                        );
                    }
                }
            }
            tx.procedure = NFSPROC3_WRITE;
            tx.xid = r.hdr.xid;
            tx.is_first = true;
            tx.nfs_version = r.progver as u16;
        }
        if !self.is_udp {
            if queue_exceeded {
                // 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;
                self.ts_chunk_tail = 0;
                self.ts_chunk_fh.clear();
                SCLogDebug!("WRITE queue exceeded: skipping {} stream bytes", pending);
            } else {
                self.ts_chunk_xid = r.hdr.xid;
                // the chunk promise follows what the parser/tracker actually
                // consume (file_len-based, incl. partial data): a count-based
                // promise outlives the record when count > file_len and would
                // eat the following record's bytes as chunk continuation
                debug_validate_bug_on!(w.file_data.len() as u32 > w.file_len);
                self.ts_chunk_left = w.file_len - w.file_data.len() as u32;
                // the record's unbuffered tail: file_len <= count, so this
                // (not the count-based promise) bounds what a rejected
                // continuation may consume or skip
                self.ts_chunk_tail = r.prog_data_size.saturating_sub(r.prog_data.len() as u32);
                self.ts_chunk_fh = file_handle;
                SCLogDebug!(
                    "REQUEST chunk_xid {:04X} chunk_left {} chunk_tail {}",
                    self.ts_chunk_xid,
                    self.ts_chunk_left,
                    self.ts_chunk_tail
                );
            }
        }
        0
    }

    fn process_partial_write_request_record<'b>(
        &mut self, flow: *mut Flow, r: &RpcPacket<'b>, w: &Nfs3RequestWrite<'b>,
    ) -> u32 {
        SCLogDebug!(
            "REQUEST {} procedure {} blob size {}",
            r.hdr.xid,
            r.procedure,
            r.prog_data.len()
        );

        let mut xidmap = NFSRequestXidMap::new(r.progver, r.procedure, 0);
        xidmap.file_handle = w.handle.value.to_vec();
        self.requestmap.put(r.hdr.xid, xidmap);

        return self.process_write_record(flow, r, w);
    }
    /// Reject a partial v4 WRITE compound whose claimed length exceeds the
    /// limit: log the tx and skip the record so the data blob is never buffered.
    fn process_partial_v4_write_request_record(&mut self, r: &RpcPacket<'_>) {
        SCLogDebug!(
            "REQUEST {} partial v4 WRITE rejected: claim exceeds {}",
            r.hdr.xid,
            unsafe { NFS_CFG_MAX_WRITE_SIZE }
        );

        let xidmap = NFSRequestXidMap::new(r.progver, r.procedure, 0);
        self.requestmap.put(r.hdr.xid, xidmap);

        let tx = self.new_file_tx(&b""[..], &b""[..], Direction::ToServer);
        tx.procedure = NFSPROC4_WRITE;
        tx.xid = r.hdr.xid;
        tx.is_first = true;
        tx.nfs_version = 4;
        tx.request_done = true;
        tx.response_done = true;
        // Close the file transaction so a later same-handle
        // operation cannot reuse this rejected one.
        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;
        // 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;
        self.ts_chunk_tail = 0;
        SCLogDebug!(
            "v4 WRITE too large (record claim {} > {}): skipping {} stream bytes",
            r.prog_data_size,
            unsafe { NFS_CFG_MAX_WRITE_SIZE },
            pending
        );
    }

    /// Mark a partially buffered v4 COMPOUND request record malformed and skip it
    /// so the (attacker-controlled) record length is never buffered: the scanner's
    /// structural errors are definitive and the full parser rejects the same record.
    fn process_partial_v4_malformed_request_record(&mut self, r: &RpcPacket<'_>) {
        SCLogDebug!(
            "REQUEST {} partial v4 COMPOUND rejected: structural scan error",
            r.hdr.xid
        );
        // A definitive structural error from the buffered bytes: log a malformed
        // tx and skip the record (bounded fallback, no buffering toward the claim).
        let mut tx = self.new_tx();
        tx.xid = r.hdr.xid;
        tx.procedure = r.procedure;
        tx.nfs_version = r.progver as u16;
        tx.request_done = true;
        tx.response_done = true;
        tx.is_file_closed = true;
        tx.tx_data.set_event(NFSEvent::MalformedData as u8);
        tx.tx_data.0.updated_ts = true;
        tx.tx_data.0.updated_tc = true;
        self.transactions.push(tx);
        // Mirror the full record path: register the xid so a matching reply
        // is processed like any other v4 request.
        let xidmap = NFSRequestXidMap::new(r.progver, r.procedure, 0);
        self.requestmap.put(r.hdr.xid, xidmap);
        // Skip to the RPC record boundary without buffering the claimed rest
        // of the 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;
        self.ts_chunk_tail = 0;
        SCLogDebug!(
            "v4 COMPOUND malformed (record claim {}): skipping {} stream bytes",
            r.prog_data_size,
            pending
        );
    }

    /// Mark a partially buffered v4 COMPOUND reply record malformed and skip it
    /// (bounded fallback; cf. process_partial_v4_malformed_request_record).
    fn process_partial_v4_malformed_response_record(&mut self, r: &RpcReplyPacket<'_>) {
        SCLogDebug!(
            "REPLY {} partial v4 COMPOUND rejected: structural scan error",
            r.hdr.xid
        );
        // Same bounded fallback on the response side: malformed tx + skip
        // the record, no buffering toward the claimed length.
        let mut tx = self.new_tx();
        tx.xid = r.hdr.xid;
        tx.nfs_version = self.nfs_version;
        tx.request_done = true;
        tx.response_done = true;
        tx.is_file_closed = true;
        tx.tx_data.set_event(NFSEvent::MalformedData as u8);
        tx.tx_data.0.updated_ts = true;
        tx.tx_data.0.updated_tc = true;
        self.transactions.push(tx);
        // Skip to the RPC record boundary without buffering the claimed rest
        // of the record.
        // NB: no requestmap entry is left behind: a reply to this (already
        // rejected) request is dropped by process_reply_record with no
        // event -- the request-side MalformedData tx is the only signal.
        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;
        self.tc_chunk_tail = 0;
        self.tc_chunk_fh.clear();
        SCLogDebug!(
            "v4 COMPOUND reply malformed (record claim {}): skipping {} stream bytes",
            r.prog_data_size,
            pending
        );
    }

    fn process_reply_record(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice, r: &RpcReplyPacket,
    ) -> u32 {
        let mut xidmap;
        match self.requestmap.pop(&r.hdr.xid) {
            Some(p) => {
                xidmap = p;
            }
            _ => {
                SCLogDebug!(
                    "REPLY: xid {:04X} NOT FOUND. GAPS? TS:{} TC:{}",
                    r.hdr.xid,
                    self.ts_ssn_gap,
                    self.tc_ssn_gap
                );

                // TODO we might be able to try to infer from the size + data
                // that this is a READ reply and pass the data to the file API anyway?
                return 0;
            }
        }
        SCLogDebug!(
            "process_reply_record: removed xid {:04X} from requestmap",
            r.hdr.xid
        );

        if self.nfs_version == 0 {
            self.nfs_version = xidmap.progver as u16;
        }

        match xidmap.progver {
            2 => {
                SCLogDebug!("NFSv2 reply record");
                self.add_nfs_tc_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                self.process_reply_record_v2(flow, r, &xidmap);
                return 0;
            }
            3 => {
                SCLogDebug!("NFSv3 reply record");
                self.add_nfs_tc_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                self.process_reply_record_v3(flow, r, &mut xidmap);
                return 0;
            }
            4 => {
                SCLogDebug!("NFSv4 reply record");
                self.add_nfs4_tc_frames(flow, stream_slice, r.prog_data, r.prog_data_size as i64);
                self.process_reply_record_v4(flow, r, &mut xidmap);
                return 0;
            }
            _ => {
                SCLogDebug!("Invalid NFS version");
                self.set_event(NFSEvent::NonExistingVersion);
                return 0;
            }
        }
    }

    // update in progress chunks for file transfers
    // return how much data we consumed
    fn filetracker_update(
        &mut self, flow: *mut Flow, direction: Direction, data: &[u8], gap_size: u32,
    ) -> u32 {
        let mut chunk_left = if direction == Direction::ToServer {
            self.ts_chunk_left
        } else {
            self.tc_chunk_left
        };
        // the record tail is tracked alongside the chunk promise (see the
        // struct field): a rejected continuation may only consume/skip it
        let mut chunk_tail = if direction == Direction::ToServer {
            self.ts_chunk_tail
        } else {
            self.tc_chunk_tail
        };

        let ssn_gap = self.ts_ssn_gap | self.tc_ssn_gap;
        // Backstop: cap the OOO queue on every open file tx (an interleaved
        // handle can hide a first file's queue), running even with chunk_left == 0.
        if ssn_gap {
            // Last-resort cap for OOO data in gapped streams: configured
            // size/count (1 GiB / 1024 when limits are off).
            let backstop = ooo_queue_backstop(direction);
            let cnt_backstop = ooo_queue_backstop_cnt(direction);
            for tx in &mut self.transactions {
                if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data {
                    if tx.is_file_tx && !tx.is_file_closed && tdf.direction == direction {
                        let queued_data = tdf.file_tracker.get_queued_size();
                        let queued_cnt = tdf.file_tracker.get_inflight_cnt() as u64;
                        if queued_data > backstop || queued_cnt > cnt_backstop {
                            SCLogDebug!(
                                "QUEUED size {} / count {} > backstop {} / {} while we've seen GAPs. Truncating file.",
                                queued_data,
                                queued_cnt,
                                backstop,
                                cnt_backstop
                            );
                            filetracker_trunc(&mut tdf.file_tracker);
                            // Attribute the truncation to the file tx
                            // actually truncated (not the newest tx).
                            tx.tx_data.set_event(NFSEvent::TruncatedFileData as u8);
                            tx.tx_data.0.updated_ts = true;
                            tx.tx_data.0.updated_tc = true;
                        }
                    }
                }
            }
        }

        if chunk_left == 0 {
            if chunk_tail == 0 {
                return 0;
            }
            // The chunk's file data is complete, but the rest of its record
            // (its XDR padding) was not part of that input. Consume it now,
            // never past the record tail: for a still open file these bytes
            // are the tracker's fill bytes, never file data.
            let pad = cmp::min(chunk_tail, data.len() as u32);
            let file_handle = if direction == Direction::ToServer {
                self.ts_chunk_fh.to_vec()
            } else {
                self.tc_chunk_fh.to_vec()
            };
            if let Some(tx) = self.get_file_tx_by_handle(&file_handle, direction) {
                if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data {
                    // only a tracker with an open file owes fill bytes
                    if !tdf.file_tracker.is_done() {
                        filetracker_update(&mut tdf.file_tracker, &data[..pad as usize], gap_size);
                    }
                }
            }
            chunk_tail -= pad;
            if direction == Direction::ToServer {
                self.ts_chunk_tail = chunk_tail;
                if chunk_tail == 0 {
                    self.ts_chunk_xid = 0;
                    self.ts_chunk_fh.clear();
                }
            } else {
                self.tc_chunk_tail = chunk_tail;
                if chunk_tail == 0 {
                    self.tc_chunk_xid = 0;
                    self.tc_chunk_fh.clear();
                }
            }
            return pad;
        }
        // bytes of this call's input that belong to the in-progress chunk
        let chunk_total = chunk_left;
        // record tail at call start: bounds the stream bytes a rejected
        // continuation may consume/skip
        let chunk_tail0 = chunk_tail;
        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 {
            self.tc_chunk_xid
        };
        SCLogDebug!(
            "filetracker_update: chunk left {}, input {} chunk_xid {:04X}",
            chunk_left,
            data.len(),
            xid
        );

        let file_handle;
        // we have the data that we expect
        if chunk_left <= data.len() as u32 {
            chunk_left = 0;
            // chunk_tail is reduced by what is consumed below: if the record's
            // XDR padding is not (all) in this input, the chunk stays active
            // until it is (see the chunk_left == 0 case above)

            if direction == Direction::ToServer {
                file_handle = self.ts_chunk_fh.to_vec();
            } else {
                file_handle = self.tc_chunk_fh.to_vec();

                // chunk done, remove requestmap entry (unless evicted)
                self.requestmap.pop(&xid);
            }
        } else {
            chunk_left -= data.len() as u32;

            if direction == Direction::ToServer {
                file_handle = self.ts_chunk_fh.to_vec();
            } else {
                file_handle = self.tc_chunk_fh.to_vec();
            }
        }

        // 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. A tx whose tracker has no open file cannot
        // take the chunk: its tracker never got a chunk, as it replaced the
        // chunk's tx after that was closed (e.g. by the max-tx limit), or its
        // file was already closed. Handle it like a missing tx.
        let tx = self
            .get_file_tx_by_handle(&file_handle, direction)
            .filter(|tx| match tx.type_data {
                Some(NFSTransactionTypeData::FILE(ref tdf)) => !tdf.file_tracker.is_done(),
                _ => false,
            });
        match tx {
            Some(tx) => {
                if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data {
                    // Chunk-path queue limit (not a newchunk caller): enforce the operator
                    // max-{read,write}-queue-size/cnt; gap bytes count in `add`; 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;
                            // The record's XDR padding, before the tracker state
                            // is cleared below: it follows the claimed data on
                            // the wire, so the claim-relative tail does not
                            // cover it.
                            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);
                            // Stream accounting follows the record's unbuffered
                            // tail (file_len-based) plus its XDR padding, not
                            // the count-based promise: count > file_len would
                            // run the consume/skip past the record into the
                            // next RPC.
                            let total = chunk_tail0.saturating_add(fill);
                            if total > data.len() as u32 {
                                // all of this segment is record tail; the rest is
                                // skipped on later calls
                                consumed = data.len() as u32;
                                limit_skip = total - data.len() as u32;
                            } else {
                                // consume exactly the tail + padding: the stream
                                // lands on the next RPC boundary
                                consumed = total;
                            }
                            chunk_tail = 0;
                        }
                    }
                    if limit_event.is_none() {
                        // 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. A truncated tracker is
                         * never done, so also close it when the data of its last
                         * chunk is complete. */
                        if tdf.file_tracker.is_done() || (tx.is_last && chunk_left == 0) {
                            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,
                                    );
                                }
                            } 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;
                    }
                } else {
                    consumed = 0;
                }
            }
            None => {
                // The file tx is gone (e.g. closed by the max-tx limit), but
                // the record boundary is still known: consume the chunk up to
                // the record tail so its data is not parsed as an RPC record.
                SCLogDebug!("no file tx for chunk XID {:04X}: consuming it", xid);
                consumed = cmp::min(chunk_tail0, data.len() as u32);
            }
        };

        // the record tail shrinks by what the tracker consumed from the stream
        if limit_event.is_none() && consumed > 0 {
            chunk_tail = chunk_tail.saturating_sub(consumed);
        }
        // once the file data is done, the rest of the record that is in this
        // input is consumed too, so it is not parsed as an RPC record
        if limit_event.is_none() && chunk_left == 0 && chunk_tail > 0 {
            let rest = cmp::min(chunk_tail, (data.len() as u32).saturating_sub(consumed));
            consumed += rest;
            chunk_tail -= rest;
        }

        // persist after the limit bookkeeping: the rejected-chunk reset and
        // the normal-path tail decrement both land in the stream state
        if direction == Direction::ToServer {
            self.ts_chunk_left = chunk_left;
            self.ts_chunk_tail = chunk_tail;
        } else {
            self.tc_chunk_left = chunk_left;
            self.tc_chunk_tail = chunk_tail;
        }
        // all of the chunk's record is consumed: forget the chunk
        if chunk_left == 0 && chunk_tail == 0 {
            if direction == Direction::ToServer {
                self.ts_chunk_xid = 0;
                self.ts_chunk_fh.clear();
            } else {
                self.tc_chunk_xid = 0;
                self.tc_chunk_fh.clear();
            }
        }

        // 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.tc_chunk_fh.clear();
                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);
            }
        }
        return consumed;
    }

    /// xidmapr is an Option as it's already removed from the map if we
    /// have a complete record. Otherwise we do a lookup ourselves.
    pub fn process_read_record<'b>(
        &mut self, flow: *mut Flow, r: &RpcReplyPacket<'b>, reply: &NfsReplyRead<'b>,
        xidmapr: Option<&NFSRequestXidMap>,
    ) -> u32 {
        let file_name;
        let file_handle;
        let chunk_offset;
        let nfs_version;

        let mut fill_bytes = 0;
        // the XDR padding derives from data_len (the data actually present in
        // the reply), not from the requested count
        let pad = reply.data_len % 4;
        if pad != 0 {
            fill_bytes = 4 - pad;
        }

        if reply.count == 0 {
            return 0;
        }

        match xidmapr {
            Some(xidmap) => {
                file_name = xidmap.file_name.to_vec();
                file_handle = xidmap.file_handle.to_vec();
                chunk_offset = xidmap.chunk_offset;
                nfs_version = xidmap.progver;
            }
            None => {
                if let Some(xidmap) = self.requestmap.get(&r.hdr.xid) {
                    file_name = xidmap.file_name.to_vec();
                    file_handle = xidmap.file_handle.to_vec();
                    chunk_offset = xidmap.chunk_offset;
                    nfs_version = xidmap.progver;
                } else {
                    // Unknown xid (unsolicited reply): the data cannot be
                    // associated. Skip to the record boundary instead of
                    // leaving the stream buffered toward the claim.
                    let pending = r.prog_data_size.saturating_sub(r.prog_data.len() as u32);
                    self.set_skip(Direction::ToClient, pending);
                    return 0;
                }
            }
        }
        SCLogDebug!("chunk_offset {}", chunk_offset);

        // Oversized READ reply: reject with an event, log the file as a tx when
        // one doesn't exist, skip the payload tail to stay in sync.
        let max_read_size = unsafe { NFS_CFG_MAX_READ_SIZE };
        if max_read_size != 0 && reply.count > max_read_size {
            let event = NFSEvent::ReadResponseTooLarge;
            match self.get_file_tx_by_handle(&file_handle, Direction::ToClient) {
                Some(tx) => {
                    // Rejected response is the final word for this read:
                    // complete the tx so it is actually logged.
                    tx.tx_data.set_event(event as u8);
                    tx.is_last = true;
                    tx.request_done = true;
                    tx.response_done = true;
                    // Close it so a later same-handle op cannot reuse it.
                    tx.is_file_closed = true;
                    tx.tx_data.0.updated_ts = true;
                    tx.tx_data.0.updated_tc = true;
                    // Read can no longer complete: truncate the file so it is
                    // logged now and the tx accounting balances.
                    if let Some(NFSTransactionTypeData::FILE(ref mut tdf)) = tx.type_data {
                        filetracker_trunc(&mut tdf.file_tracker);
                    }
                }
                None => {
                    let tx = self.new_file_tx(&file_handle, &file_name, Direction::ToClient);
                    tx.procedure = if nfs_version < 4 {
                        NFSPROC3_READ
                    } else {
                        NFSPROC4_READ
                    };
                    tx.xid = r.hdr.xid;
                    tx.is_first = true;
                    tx.is_last = true;
                    tx.request_done = true;
                    tx.response_done = true;
                    tx.is_file_closed = true;
                    tx.nfs_version = if nfs_version < 4 { 3 } else { 4 };
                    tx.tx_data.set_event(event as u8);
                    tx.tx_data.0.updated_ts = true;
                    tx.tx_data.0.updated_tc = true;
                    // Rejected tx never opened a file: zero files_opened, else it
                    // stays live until teardown and grows unbounded on repeats.
                    tx.tx_data.0.files_opened = 0;
                }
            }
            // 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);
            if !self.is_udp {
                self.set_skip(Direction::ToClient, pending);
                self.tc_chunk_xid = 0;
                self.tc_chunk_left = 0;
                self.tc_chunk_tail = 0;
                self.tc_chunk_fh.clear();
            }
            SCLogDebug!(
                "READ too large ({} > {}): skipping {} stream bytes",
                reply.count,
                unsafe { NFS_CFG_MAX_READ_SIZE },
                pending
            );
            return 0;
        }

        let mut is_last = reply.eof;
        SCLogDebug!(
            "XID {} is_last {} fill_bytes {} reply.count {} reply.data_len {} reply.data.len() {}",
            r.hdr.xid,
            is_last,
            fill_bytes,
            reply.count,
            reply.data_len,
            reply.data.len()
        );

        if nfs_version == 2 {
            let size = match parse_nfs2_attribs(reply.attr_blob) {
                Ok((_, ref attr)) => attr.asize,
                _ => 0,
            };
            SCLogDebug!(
                "NFSv2 READ reply record: File size {}. Offset {} data len {}: total {}",
                size,
                chunk_offset,
                reply.data_len,
                chunk_offset + reply.data_len as u64
            );

            if size as u64 == chunk_offset + reply.data_len as u64 {
                is_last = true;
            }
        }

        let is_partial = reply.data.len() < reply.data_len as usize;
        SCLogDebug!("partial data? {}", is_partial);

        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,
                        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(
                            &mut tdf.file_tracker,
                            &file_name,
                            reply.data,
                            chunk_offset,
                            reply.data_len,
                            fill_bytes as u8,
                            is_last,
                            &r.hdr.xid,
                        );
                        tdf.chunk_count += 1;
                        if is_last {
                            tdf.file_last_xid = r.hdr.xid;
                            tx.rpc_response_status = r.reply_state;
                            tx.nfs_response_status = reply.status;
                            tx.is_last = true;
                            tx.request_done = true;
                            sc_app_layer_parser_trigger_raw_stream_inspection(
                                flow,
                                Direction::ToServer as i32,
                            );

                            /* if this is a partial record we will close the tx
                             * when we've received the final data */
                            if !is_partial {
                                tx.response_done = true;
                                SCLogDebug!("TX {} is DONE", tx.id);
                                sc_app_layer_parser_trigger_raw_stream_inspection(
                                    flow,
                                    Direction::ToClient as i32,
                                );
                            }
                        }
                    }
                    true
                } else {
                    false
                }
            }
            None => false,
        };
        if !found {
            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, chunk_offset, reply.data.len() as u64)
                {
                    // 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,
                        &file_name,
                        reply.data,
                        chunk_offset,
                        reply.data_len,
                        fill_bytes as u8,
                        is_last,
                        &r.hdr.xid,
                    );
                }
                if is_last && !queue_exceeded {
                    tdf.file_last_xid = r.hdr.xid;
                    tx.rpc_response_status = r.reply_state;
                    tx.nfs_response_status = reply.status;
                    tx.is_last = true;
                    tx.request_done = true;
                    sc_app_layer_parser_trigger_raw_stream_inspection(
                        flow,
                        Direction::ToServer as i32,
                    );

                    /* if this is a partial record we will close the tx
                     * when we've received the final data */
                    if !is_partial {
                        tx.response_done = true;
                        SCLogDebug!("TX {} is DONE", tx.id);
                        sc_app_layer_parser_trigger_raw_stream_inspection(
                            flow,
                            Direction::ToClient as i32,
                        );
                    }
                }
            }
            tx.procedure = if nfs_version < 4 {
                NFSPROC3_READ
            } else {
                NFSPROC4_READ
            };
            tx.xid = r.hdr.xid;
            tx.is_first = true;
        }
        if !self.is_udp {
            if queue_exceeded {
                // 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;
                self.tc_chunk_tail = 0;
                self.tc_chunk_fh.clear();
                SCLogDebug!("READ queue exceeded: skipping {} stream bytes", pending);
            } else {
                self.tc_chunk_xid = r.hdr.xid;
                // the chunk promise follows the data actually present in the
                // reply (data_len-based): a count-based promise outlives the
                // record on a short read and would eat the following record's
                // bytes as chunk continuation
                debug_validate_bug_on!(reply.data.len() as u32 > reply.data_len);
                self.tc_chunk_left = reply.data_len - reply.data.len() as u32;
                // the record's unbuffered tail: data_len <= count, so this
                // (not the count-based promise) bounds what a rejected
                // continuation may consume or skip
                self.tc_chunk_tail = r.prog_data_size.saturating_sub(r.prog_data.len() as u32);
                self.tc_chunk_fh = file_handle;
                // a partial reply that holds all of its data only has its padding
                // left: done with the request, as at the end of a chunk in
                // filetracker_update
                if xidmapr.is_none() && self.tc_chunk_left == 0 {
                    self.requestmap.pop(&r.hdr.xid);
                }
            }
        }

        SCLogDebug!(
            "REPLY {} to procedure {} blob size {} / {}: chunk_left {} chunk_xid {:04X}",
            r.hdr.xid,
            NFSPROC3_READ,
            r.prog_data.len(),
            reply.count,
            self.tc_chunk_left,
            self.tc_chunk_xid
        );
        0
    }

    fn process_partial_read_reply_record<'b>(
        &mut self, flow: *mut Flow, r: &RpcReplyPacket<'b>, reply: &NfsReplyRead<'b>,
    ) -> u32 {
        SCLogDebug!(
            "REPLY {} to procedure READ blob size {} / {}",
            r.hdr.xid,
            r.prog_data.len(),
            reply.count
        );

        return self.process_read_record(flow, r, reply, None);
    }

    fn peek_reply_record(&mut self, r: &RpcPacketHeader) -> u32 {
        if let Some(xidmap) = self.requestmap.get(&r.xid) {
            return xidmap.procedure;
        } else {
            SCLogDebug!("REPLY: xid {} NOT FOUND", r.xid);
            return 0;
        }
    }

    /// A partial record failed to parse, but its record marker gave its size:
    /// skip the rest of the record, like the complete record parsing does,
    /// instead of disabling the parser for the flow.
    fn skip_partial_record(
        &mut self, direction: Direction, cur_i: &[u8], rec_size: usize,
    ) -> AppLayerResult {
        let pending = (rec_size - cur_i.len()) as u32;
        SCLogDebug!(
            "malformed partial record: skipping {} stream bytes",
            pending
        );
        self.set_skip(direction, pending);
        return AppLayerResult::ok();
    }

    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();
        }
        if consumed < gap_size {
            let new_gap_size = gap_size - consumed;
            let gap = vec![0; new_gap_size as usize];
            self.filetracker_update(
                std::ptr::null_mut(),
                Direction::ToServer,
                &gap,
                new_gap_size,
            );
        }
        SCLogDebug!("parse_tcp_data_ts_gap ({}) done", gap_size);
        return AppLayerResult::ok();
    }

    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();
        }
        if consumed < gap_size {
            let new_gap_size = gap_size - consumed;
            let gap = vec![0; new_gap_size as usize];
            self.filetracker_update(
                std::ptr::null_mut(),
                Direction::ToClient,
                &gap,
                new_gap_size,
            );
        }
        SCLogDebug!("parse_tcp_data_tc_gap ({}) done", gap_size);
        return AppLayerResult::ok();
    }

    /// Handle partial records
    fn parse_tcp_partial_data_ts<'b>(
        &mut self, flow: *mut Flow, base_input: &'b [u8], cur_i: &'b [u8],
        phdr: &RpcRequestPacketPartial, rec_size: usize,
    ) -> AppLayerResult {
        // special case: avoid buffering file write blobs
        // as these can be large.
        if rec_size >= 512 && cur_i.len() >= 44 {
            // large record, likely file xfer
            SCLogDebug!("large record {}, likely file xfer", rec_size);

            // quick peek, are we in WRITE mode?
            if phdr.procedure == NFSPROC3_WRITE {
                SCLogDebug!(
                    "CONFIRMED WRITE: large record {}, file chunk xfer",
                    rec_size
                );

                // lets try to parse the RPC record. Might fail with Incomplete.
                match parse_rpc(cur_i, false) {
                    Ok((_rem, ref hdr)) => {
                        // we got here because rec_size > input, so we should never have
                        // remaining data
                        debug_validate_bug_on!(!_rem.is_empty());

                        match parse_nfs3_request_write(hdr.prog_data, false) {
                            Ok((_, ref w)) => {
                                // deal with the partial nfs write data
                                self.process_partial_write_request_record(flow, hdr, w);
                                return AppLayerResult::ok();
                            }
                            Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                                self.set_event(NFSEvent::MalformedData);
                                SCLogDebug!("Parsing failed: {:?}", _e);
                                return self.skip_partial_record(
                                    Direction::ToServer,
                                    cur_i,
                                    rec_size,
                                );
                            }
                            Err(Err::Incomplete(_)) => {
                                // this is normal, fall through to incomplete handling
                            }
                        }
                    }
                    Err(Err::Incomplete(_)) => {
                        // size check was done for a minimal RPC record size,
                        // so Incomplete is normal.
                        SCLogDebug!("TS data incomplete");
                    }
                    Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                        self.set_event(NFSEvent::MalformedData);
                        SCLogDebug!("Parsing failed: {:?}", _e);
                        return self.skip_partial_record(Direction::ToServer, cur_i, rec_size);
                    }
                }
            }
            // NFSv4 WRITE data is inside a compound op: scan the compound to the claimed
            // write length so oversized records are rejected before the 31-bit claim is buffered.
            if phdr.progver == 4 && phdr.procedure == NFSPROC4_COMPOUND {
                let max_write = unsafe { NFS_CFG_MAX_WRITE_SIZE };
                // a zero limit disables the check
                if max_write != 0 {
                    match parse_rpc(cur_i, false) {
                        Ok((_rem, ref hdr)) => {
                            // RPCSEC_GSS integrity compounds arrive wrapped (length, seqnum, data):
                            // unwrap like the full record path, or the envelope length is read as the
                            // compound tag and the scan stays incomplete (buffering toward the claim).
                            let (gss_proc, gss_service) = match &hdr.creds {
                                RpcRequestCreds::GssApi(ref g) => (g.procedure, g.service),
                                _ => (0, 0),
                            };
                            let scanres = match gss_compound_scan_data(
                                gss_proc,
                                gss_service,
                                hdr.prog_data,
                            ) {
                                GssCompoundScan::Compound(data) => {
                                    scan_nfs4_request_compound_write_len(data, max_write)
                                }
                                GssCompoundScan::EnvelopeIncomplete => Nfs4CompoundScan::Incomplete,
                                GssCompoundScan::EnvelopeMalformed => Nfs4CompoundScan::Malformed,
                            };
                            match scanres {
                                Nfs4CompoundScan::Oversized(_write_len) => {
                                    self.process_partial_v4_write_request_record(hdr);
                                    return AppLayerResult::ok();
                                }
                                // Definitive structural error: skip the record (bounded
                                // fallback) instead of buffering toward the claimed length.
                                Nfs4CompoundScan::Malformed => {
                                    self.process_partial_v4_malformed_request_record(hdr);
                                    return AppLayerResult::ok();
                                }
                                // Incomplete (needs more data) and Clean (no oversized op)
                                // both fall through to the normal incomplete handling.
                                Nfs4CompoundScan::Incomplete | Nfs4CompoundScan::Clean => {}
                            }
                        }
                        Err(Err::Incomplete(_)) => {
                            // normal: the compound scan needs more data
                            SCLogDebug!("TS v4 data incomplete");
                        }
                        Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                            self.set_event(NFSEvent::MalformedData);
                            SCLogDebug!("Parsing failed: {:?}", _e);
                            return self.skip_partial_record(Direction::ToServer, cur_i, rec_size);
                        }
                    }
                }
            }
        }
        // make sure we pass a value higher than current input
        // but lower than the record size
        let n1 = cmp::max(cur_i.len(), 1024);
        let n2 = cmp::min(n1, rec_size);
        return AppLayerResult::incomplete((base_input.len() - cur_i.len()) as u32, n2 as u32);
    }

    /// Parsing function, handling TCP chunks fragmentation
    pub fn parse_tcp_data_ts(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice,
    ) -> AppLayerResult {
        let mut cur_i = stream_slice.as_slice();
        let consumed = self.handle_skip(Direction::ToServer, cur_i.len() as u32);
        if consumed > 0 {
            if consumed as usize > cur_i.len() {
                return AppLayerResult::err();
            }
            cur_i = &cur_i[consumed as usize..];
        }
        // take care of in progress file chunk transfers
        // and skip buffer beyond it
        let consumed = self.filetracker_update(flow, Direction::ToServer, cur_i, 0);
        if consumed > 0 {
            if consumed > cur_i.len() as u32 {
                return AppLayerResult::err();
            }
            cur_i = &cur_i[consumed as usize..];
        }
        if cur_i.is_empty() {
            return AppLayerResult::ok();
        }
        if self.ts_gap {
            SCLogDebug!("TS trying to catch up after GAP (input {})", cur_i.len());

            let mut _cnt = 0;
            while !cur_i.is_empty() {
                _cnt += 1;
                match nfs_probe(cur_i, Direction::ToServer) {
                    1 => {
                        SCLogDebug!("expected data found");
                        self.ts_gap = false;
                        break;
                    }
                    0 => {
                        SCLogDebug!("incomplete, queue and retry with the next block (input {}). Looped {} times.",
                                cur_i.len(), _cnt);
                        return AppLayerResult::incomplete(
                            stream_slice.len() - cur_i.len() as u32,
                            (cur_i.len() + 1) as u32,
                        );
                    }
                    -1 => {
                        cur_i = &cur_i[1..];
                        if cur_i.is_empty() {
                            SCLogDebug!(
                                "all post-GAP data in this chunk was bad. Looped {} times.",
                                _cnt
                            );
                        }
                    }
                    _ => {
                        return AppLayerResult::err();
                    }
                }
            }
            SCLogDebug!("TS GAP handling done (input {})", cur_i.len());
        }

        while !cur_i.is_empty() {
            // min record size
            self.add_rpc_tcp_ts_pdu(flow, stream_slice, cur_i, cur_i.len() as i64);
            match parse_rpc_request_partial(cur_i) {
                Ok((_, ref rpc_phdr)) => {
                    let rec_size = (rpc_phdr.hdr.frag_len + 4) as usize;

                    // Handle partial records
                    if rec_size > cur_i.len() {
                        return self.parse_tcp_partial_data_ts(
                            flow,
                            stream_slice.as_slice(),
                            cur_i,
                            rpc_phdr,
                            rec_size,
                        );
                    }

                    // we have the full records size worth of data,
                    // let's parse it. Errors lead to event, but are
                    // not fatal as we already have enough info to
                    // go to the next record.
                    match parse_rpc(cur_i, true) {
                        Ok((_, ref rpc_record)) => {
                            self.add_rpc_tcp_ts_creds(
                                flow,
                                stream_slice,
                                &cur_i[RPC_TCP_PRE_CREDS..],
                                (rpc_record.creds_len + 8) as i64,
                            );
                            self.process_request_record(flow, stream_slice, rpc_record);
                        }
                        Err(Err::Incomplete(_)) => {
                            self.set_event(NFSEvent::MalformedData);
                        }
                        Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                            self.set_event(NFSEvent::MalformedData);
                            SCLogDebug!("Parsing failed: {:?}", _e);
                        }
                    }
                    cur_i = &cur_i[rec_size..];
                }
                Err(Err::Incomplete(needed)) => {
                    if let Needed::Size(n) = needed {
                        SCLogDebug!("Not enough data for partial RPC header {:?}", needed);
                        // 28 is the partial RPC header size parse_rpc_request_partial
                        // looks for.
                        let n = usize::from(n);
                        let need = if n > 28 { n } else { 28 };
                        return AppLayerResult::incomplete(
                            stream_slice.len() - cur_i.len() as u32,
                            need as u32,
                        );
                    }
                    return AppLayerResult::err();
                }
                /* This error is fatal. If we failed to parse the RPC hdr we don't
                 * have a length and we don't know where the next record starts. */
                Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                    self.set_event(NFSEvent::MalformedData);
                    SCLogDebug!("Parsing failed: {:?}", _e);
                    return AppLayerResult::err();
                }
            }
        }

        self.post_gap_housekeeping(Direction::ToServer);
        if self.check_post_gap_file_txs && !self.post_gap_files_checked {
            self.post_gap_housekeeping_for_files();
            self.post_gap_files_checked = true;
        }

        AppLayerResult::ok()
    }

    /// Handle partial records
    fn parse_tcp_partial_data_tc<'b>(
        &mut self, flow: *mut Flow, base_input: &'b [u8], cur_i: &'b [u8], phdr: &RpcPacketHeader,
        rec_size: usize,
    ) -> AppLayerResult {
        // A partial reply that cannot carry file data must never be buffered
        // toward its claimed length (31-bit, attacker-controlled): a
        // legitimate reply always follows its request on this flow, so an
        // unknown xid or a non-file procedure (only READ replies carry file
        // data) means the record is unsolicited or a lost request's reply.
        // Skip to the claimed record boundary instead.
        match parse_rpc_reply(cur_i, false) {
            Ok((_, ref hdr)) => {
                let carries_file_data = self
                    .requestmap
                    .get(&hdr.hdr.xid)
                    .map(|xm| xm.procedure == NFSPROC3_READ || xm.procedure == NFSPROC4_COMPOUND)
                    .unwrap_or(false);
                if !carries_file_data {
                    let pending = hdr
                        .prog_data_size
                        .saturating_sub(hdr.prog_data.len() as u32);
                    self.set_skip(Direction::ToClient, pending);
                    SCLogDebug!(
                        "partial reply xid {:08X} carries no file data: skipping {} stream bytes",
                        hdr.hdr.xid,
                        pending
                    );
                    return AppLayerResult::ok();
                }
            }
            Err(Err::Incomplete(_)) => {
                // not enough header bytes yet: keep buffering (bounded by the
                // 28-byte header) and fall through to the incomplete handling
            }
            Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                self.set_event(NFSEvent::MalformedData);
                SCLogDebug!("Parsing failed: {:?}", _e);
                return self.skip_partial_record(Direction::ToClient, cur_i, rec_size);
            }
        }
        // special case: avoid buffering file read blobs
        // as these can be large.
        if rec_size >= 512 && cur_i.len() >= 128 {
            //36 {
            // large record, likely file xfer
            SCLogDebug!("large record {}, likely file xfer", rec_size);

            // quick peek, are in READ mode?
            if self.peek_reply_record(phdr) == NFSPROC3_READ {
                SCLogDebug!(
                    "CONFIRMED large READ record {}, likely file chunk xfer",
                    rec_size
                );

                // we should have enough data to parse the RPC record
                match parse_rpc_reply(cur_i, false) {
                    Ok((_rem, ref hdr)) => {
                        // we got here because rec_size > input, so we should never have
                        // remaining data
                        debug_validate_bug_on!(!_rem.is_empty());

                        match parse_nfs3_reply_read(hdr.prog_data, false) {
                            Ok((_, ref r)) => {
                                // deal with the partial nfs read data
                                self.process_partial_read_reply_record(flow, hdr, r);
                                return AppLayerResult::ok();
                            }
                            Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                                self.set_event(NFSEvent::MalformedData);
                                SCLogDebug!("Parsing failed: {:?}", _e);
                                // done with the request, as for a complete record
                                self.requestmap.pop(&hdr.hdr.xid);
                                return self.skip_partial_record(
                                    Direction::ToClient,
                                    cur_i,
                                    rec_size,
                                );
                            }
                            Err(Err::Incomplete(_)) => {
                                // this is normal, fall through to incomplete handling
                            }
                        }
                    }
                    Err(Err::Incomplete(_)) => {
                        // size check was done for a minimal RPC record size,
                        // so Incomplete is normal.
                        SCLogDebug!("TC data incomplete");
                    }
                    Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                        self.set_event(NFSEvent::MalformedData);
                        SCLogDebug!("Parsing failed: {:?}", _e);
                        return self.skip_partial_record(Direction::ToClient, cur_i, rec_size);
                    }
                }
            }
            // NFSv4 READ data is inside a COMPOUND reply: scan the compound to the
            // claimed read length so oversized replies are rejected before the 31-bit claim is buffered.
            if self.peek_reply_record(phdr) == NFSPROC4_COMPOUND {
                let max_read = unsafe { NFS_CFG_MAX_READ_SIZE };
                // a zero limit disables the check
                if max_read != 0 {
                    // the request recorded its RPCSEC_GSS credential combination in the
                    // xidmap; integrity-wrapped compounds (proc 0, service 2) are unwrapped first.
                    let gss_integrity = self
                        .requestmap
                        .get(&phdr.xid)
                        .map(|xm| xm.gssapi_proc == 0 && xm.gssapi_service == 2)
                        .unwrap_or(false);
                    match parse_rpc_reply(cur_i, false) {
                        Ok((_rem, ref hdr)) => {
                            let scanres = if gss_integrity {
                                match gss_compound_scan_data(0, 2, hdr.prog_data) {
                                    GssCompoundScan::Compound(data) => {
                                        scan_nfs4_response_compound_read_len(data, max_read)
                                    }
                                    GssCompoundScan::EnvelopeIncomplete => {
                                        Nfs4CompoundScan::Incomplete
                                    }
                                    GssCompoundScan::EnvelopeMalformed => {
                                        Nfs4CompoundScan::Malformed
                                    }
                                }
                            } else {
                                scan_nfs4_response_compound_read_len(hdr.prog_data, max_read)
                            };
                            match scanres {
                                Nfs4CompoundScan::Oversized(read_len) => {
                                    // run the shared reject path (transaction
                                    // handling + skip) with the claimed length
                                    let reply = NfsReplyRead {
                                        status: 0,
                                        attr_follows: 0,
                                        attr_blob: &[],
                                        count: read_len,
                                        eof: false,
                                        data_len: 0,
                                        data: &[],
                                    };
                                    self.process_read_record(flow, hdr, &reply, None);
                                    return AppLayerResult::ok();
                                }
                                // Definitive structural error: skip the record (bounded
                                // fallback) instead of buffering toward the claimed length.
                                Nfs4CompoundScan::Malformed => {
                                    self.process_partial_v4_malformed_response_record(hdr);
                                    return AppLayerResult::ok();
                                }
                                // Incomplete (needs more data) and Clean (no oversized op)
                                // both fall through to the normal incomplete handling.
                                Nfs4CompoundScan::Incomplete | Nfs4CompoundScan::Clean => {}
                            }
                        }
                        Err(Err::Incomplete(_)) => {
                            // normal: the compound scan needs more data
                            SCLogDebug!("TC v4 data incomplete");
                        }
                        Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                            self.set_event(NFSEvent::MalformedData);
                            SCLogDebug!("Parsing failed: {:?}", _e);
                            return self.skip_partial_record(Direction::ToClient, cur_i, rec_size);
                        }
                    }
                }
            }
        }
        // make sure we pass a value higher than current input
        // but lower than the record size
        let n1 = cmp::max(cur_i.len(), 1024);
        let n2 = cmp::min(n1, rec_size);
        return AppLayerResult::incomplete((base_input.len() - cur_i.len()) as u32, n2 as u32);
    }

    /// Parsing function, handling TCP chunks fragmentation
    pub fn parse_tcp_data_tc(
        &mut self, flow: *mut Flow, stream_slice: &StreamSlice,
    ) -> AppLayerResult {
        let mut cur_i = stream_slice.as_slice();
        let consumed = self.handle_skip(Direction::ToClient, cur_i.len() as u32);
        if consumed > 0 {
            if consumed as usize > cur_i.len() {
                return AppLayerResult::err();
            }
            cur_i = &cur_i[consumed as usize..];
        }
        // take care of in progress file chunk transfers
        // and skip buffer beyond it
        let consumed = self.filetracker_update(flow, Direction::ToClient, cur_i, 0);
        if consumed > 0 {
            if consumed > cur_i.len() as u32 {
                return AppLayerResult::err();
            }
            cur_i = &cur_i[consumed as usize..];
        }
        if cur_i.is_empty() {
            return AppLayerResult::ok();
        }
        if self.tc_gap {
            SCLogDebug!("TC trying to catch up after GAP (input {})", cur_i.len());

            let mut _cnt = 0;
            while !cur_i.is_empty() {
                _cnt += 1;
                match nfs_probe(cur_i, Direction::ToClient) {
                    1 => {
                        SCLogDebug!("expected data found");
                        self.tc_gap = false;
                        break;
                    }
                    0 => {
                        SCLogDebug!("incomplete, queue and retry with the next block (input {}). Looped {} times.",
                                cur_i.len(), _cnt);
                        return AppLayerResult::incomplete(
                            stream_slice.len() - cur_i.len() as u32,
                            (cur_i.len() + 1) as u32,
                        );
                    }
                    -1 => {
                        cur_i = &cur_i[1..];
                        if cur_i.is_empty() {
                            SCLogDebug!(
                                "all post-GAP data in this chunk was bad. Looped {} times.",
                                _cnt
                            );
                        }
                    }
                    _ => {
                        return AppLayerResult::err();
                    }
                }
            }
            SCLogDebug!("TC GAP handling done (input {})", cur_i.len());
        }

        while !cur_i.is_empty() {
            self.add_rpc_tcp_tc_pdu(flow, stream_slice, cur_i, cur_i.len() as i64);
            match parse_rpc_packet_header(cur_i) {
                Ok((_, ref rpc_phdr)) => {
                    let rec_size = (rpc_phdr.frag_len + 4) as usize;
                    // see if we have all data available
                    if rec_size > cur_i.len() {
                        return self.parse_tcp_partial_data_tc(
                            flow,
                            stream_slice.as_slice(),
                            cur_i,
                            rpc_phdr,
                            rec_size,
                        );
                    }

                    // we have the full data of the record, lets parse
                    match parse_rpc_reply(cur_i, true) {
                        Ok((_, ref rpc_record)) => {
                            self.add_rpc_tcp_tc_frames(
                                flow,
                                stream_slice,
                                cur_i,
                                cur_i.len() as i64,
                            );
                            self.process_reply_record(flow, stream_slice, rpc_record);
                        }
                        Err(Err::Incomplete(_)) => {
                            // we shouldn't get incomplete as we have the full data
                            // so if we got incomplete anyway it's the data that is
                            // bad.
                            self.set_event(NFSEvent::MalformedData);
                        }
                        Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                            self.set_event(NFSEvent::MalformedData);
                            SCLogDebug!("Parsing failed: {:?}", _e);
                        }
                    }
                    cur_i = &cur_i[rec_size..]; // progress input past parsed record
                }
                Err(Err::Incomplete(needed)) => {
                    if let Needed::Size(n) = needed {
                        SCLogDebug!("Not enough data for partial RPC header {:?}", needed);
                        // 12 is the partial RPC header size parse_rpc_packet_header
                        // looks for.
                        let n = usize::from(n);
                        let need = if n > 12 { n } else { 12 };
                        return AppLayerResult::incomplete(
                            stream_slice.len() - cur_i.len() as u32,
                            need as u32,
                        );
                    }
                    return AppLayerResult::err();
                }
                /* This error is fatal. If we failed to parse the RPC hdr we don't
                 * have a length and we don't know where the next record starts. */
                Err(Err::Error(_e)) | Err(Err::Failure(_e)) => {
                    self.set_event(NFSEvent::MalformedData);
                    SCLogDebug!("Parsing failed: {:?}", _e);
                    return AppLayerResult::err();
                }
            }
        }
        self.post_gap_housekeeping(Direction::ToClient);
        if self.check_post_gap_file_txs && !self.post_gap_files_checked {
            self.post_gap_housekeeping_for_files();
            self.post_gap_files_checked = true;
        }
        AppLayerResult::ok()
    }
    /// Parsing function
    pub fn parse_udp_ts(&mut self, flow: *mut Flow, stream_slice: &StreamSlice) -> AppLayerResult {
        let input = stream_slice.as_slice();
        SCLogDebug!("parse_udp_ts ({})", input.len());
        self.add_rpc_udp_ts_pdu(flow, stream_slice, input, input.len() as i64);
        if !input.is_empty() {
            match parse_rpc_udp_request(input) {
                Ok((_, ref rpc_record)) => {
                    self.is_udp = true;
                    self.add_rpc_udp_ts_creds(
                        flow,
                        stream_slice,
                        &input[RPC_UDP_PRE_CREDS..],
                        (rpc_record.creds_len + 8) as i64,
                    );
                    match rpc_record.progver {
                        3 => {
                            self.process_request_record(flow, stream_slice, rpc_record);
                        }
                        2 => {
                            self.add_nfs_ts_frame(
                                flow,
                                stream_slice,
                                rpc_record.prog_data,
                                rpc_record.prog_data_size as i64,
                            );
                            self.process_request_record_v2(rpc_record);
                        }
                        _ => {}
                    }
                }
                Err(_e) => {
                    // UDP: a datagram cannot be completed by a later one; an
                    // unparseable one is corrupt -> log it and drop just this
                    // datagram. Records are self-delimited (no stream to keep
                    // in sync), so abandoning the whole flow would let a single
                    // crafted datagram suppress inspection of everything after.
                    SCLogDebug!("UDP ts parse failed: {:?}", _e);
                    if self.is_udp {
                        let mut tx = self.new_tx();
                        tx.nfs_version = self.nfs_version;
                        tx.request_done = true;
                        tx.response_done = true;
                        tx.is_file_closed = true;
                        tx.tx_data.set_event(NFSEvent::MalformedData as u8);
                        tx.tx_data.0.updated_ts = true;
                        tx.tx_data.0.updated_tc = true;
                        self.transactions.push(tx);
                        return AppLayerResult::ok();
                    }
                }
            }
        }
        AppLayerResult::ok()
    }

    /// Parsing function
    pub fn parse_udp_tc(&mut self, flow: *mut Flow, stream_slice: &StreamSlice) -> AppLayerResult {
        let input = stream_slice.as_slice();
        SCLogDebug!("parse_udp_tc ({})", input.len());
        self.add_rpc_udp_tc_pdu(flow, stream_slice, input, input.len() as i64);
        if !input.is_empty() {
            match parse_rpc_udp_reply(input) {
                Ok((_, ref rpc_record)) => {
                    self.is_udp = true;
                    self.add_rpc_udp_tc_frames(flow, stream_slice, input, input.len() as i64);
                    self.process_reply_record(flow, stream_slice, rpc_record);
                }
                Err(_e) => {
                    // See the parse_udp_ts arm above: a corrupt datagram on a
                    // locked-in UDP flow is logged and dropped (datagram only,
                    // the flow keeps being inspected).
                    SCLogDebug!("UDP tc parse failed: {:?}", _e);
                    if self.is_udp {
                        let mut tx = self.new_tx();
                        tx.nfs_version = self.nfs_version;
                        tx.request_done = true;
                        tx.response_done = true;
                        tx.is_file_closed = true;
                        tx.tx_data.set_event(NFSEvent::MalformedData as u8);
                        tx.tx_data.0.updated_ts = true;
                        tx.tx_data.0.updated_tc = true;
                        self.transactions.push(tx);
                        return AppLayerResult::ok();
                    }
                }
            }
        }
        AppLayerResult::ok()
    }
}

/// Returns *mut NFSState
extern "C" fn nfs_state_new(
    _orig_state: *mut std::os::raw::c_void, _orig_proto: AppProto,
) -> *mut std::os::raw::c_void {
    let state = NFSState::new();
    let boxed = Box::new(state);
    SCLogDebug!("allocating state");
    return Box::into_raw(boxed) as *mut _;
}

/// Params:
/// - state: *mut NFSState as void pointer
extern "C" fn nfs_state_free(state: *mut std::os::raw::c_void) {
    // Just unbox...
    SCLogDebug!("freeing state");
    std::mem::drop(unsafe { Box::from_raw(state as *mut NFSState) });
}

/// C binding parse a NFS TCP request. Returns 1 on success, -1 on failure.
unsafe extern "C" fn nfs_parse_request(
    flow: *mut Flow, state: *mut std::os::raw::c_void, _pstate: *mut AppLayerParserState,
    stream_slice: StreamSlice, _data: *mut std::os::raw::c_void,
) -> AppLayerResult {
    let state = cast_pointer!(state, NFSState);
    let flow = cast_pointer!(flow, Flow);

    if stream_slice.is_gap() {
        return nfs_parse_request_tcp_gap(state, stream_slice.gap_size());
    }
    SCLogDebug!("parsing {} bytes of request data", stream_slice.len());

    state.update_ts(flow_get_last_time(flow).as_secs());
    state.parse_tcp_data_ts(flow, &stream_slice)
}

extern "C" fn nfs_parse_request_tcp_gap(state: &mut NFSState, input_len: u32) -> AppLayerResult {
    state.parse_tcp_data_ts_gap(input_len)
}

unsafe extern "C" fn nfs_parse_response(
    flow: *mut Flow, state: *mut std::os::raw::c_void, _pstate: *mut AppLayerParserState,
    stream_slice: StreamSlice, _data: *mut std::os::raw::c_void,
) -> AppLayerResult {
    let state = cast_pointer!(state, NFSState);
    let flow = cast_pointer!(flow, Flow);

    if stream_slice.is_gap() {
        return nfs_parse_response_tcp_gap(state, stream_slice.gap_size());
    }
    SCLogDebug!("parsing {} bytes of response data", stream_slice.len());

    state.update_ts(flow_get_last_time(flow).as_secs());
    state.parse_tcp_data_tc(flow, &stream_slice)
}

extern "C" fn nfs_parse_response_tcp_gap(state: &mut NFSState, input_len: u32) -> AppLayerResult {
    state.parse_tcp_data_tc_gap(input_len)
}

/// C binding to parse an NFS/UDP request. Returns 1 on success, -1 on failure.
unsafe extern "C" fn nfs_parse_request_udp(
    f: *mut Flow, state: *mut std::os::raw::c_void, _pstate: *mut AppLayerParserState,
    stream_slice: StreamSlice, _data: *mut std::os::raw::c_void,
) -> AppLayerResult {
    let state = cast_pointer!(state, NFSState);

    SCLogDebug!("parsing {} bytes of request data", stream_slice.len());
    state.parse_udp_ts(f, &stream_slice)
}

unsafe extern "C" fn nfs_parse_response_udp(
    f: *mut Flow, state: *mut std::os::raw::c_void, _pstate: *mut AppLayerParserState,
    stream_slice: StreamSlice, _data: *mut std::os::raw::c_void,
) -> AppLayerResult {
    let state = cast_pointer!(state, NFSState);
    SCLogDebug!("parsing {} bytes of response data", stream_slice.len());
    state.parse_udp_tc(f, &stream_slice)
}

pub unsafe extern "C" fn nfs_state_get_tx_count(state: *mut std::os::raw::c_void) -> u64 {
    let state = cast_pointer!(state, NFSState);
    SCLogDebug!("nfs_state_get_tx_count: returning {}", state.tx_id);
    return state.tx_id;
}

unsafe extern "C" fn nfs_state_get_tx(
    state: *mut std::os::raw::c_void, tx_id: u64,
) -> *mut std::os::raw::c_void {
    let state = cast_pointer!(state, NFSState);
    match state.get_tx_by_id(tx_id) {
        Some(tx) => {
            return tx as *const _ as *mut _;
        }
        None => {
            return std::ptr::null_mut();
        }
    }
}

unsafe extern "C" fn nfs_state_tx_free(state: *mut std::os::raw::c_void, tx_id: u64) {
    let state = cast_pointer!(state, NFSState);
    state.free_tx(tx_id);
}

unsafe extern "C" fn nfs_tx_get_alstate_progress(
    tx: *mut std::os::raw::c_void, direction: u8,
) -> std::os::raw::c_int {
    let tx = cast_pointer!(tx, NFSTransaction);
    if direction == u8::from(Direction::ToServer) && tx.request_done {
        SCLogDebug!("TOSERVER progress 1");
        return 1;
    } else if direction == u8::from(Direction::ToClient) && tx.response_done {
        SCLogDebug!("TOCLIENT progress 1");
        return 1;
    } else {
        SCLogDebug!("{} progress 0", direction);
        return 0;
    }
}

unsafe extern "C" fn nfs_get_tx_data(
    tx: *mut std::os::raw::c_void,
) -> *mut suricata_sys::sys::AppLayerTxData {
    let tx = cast_pointer!(tx, NFSTransaction);
    return &mut tx.tx_data.0;
}

export_state_data_get!(nfs_get_state_data, NFSState);

#[no_mangle]
pub unsafe extern "C" fn SCNfsInit(context: &'static mut SuricataFileContext) {
    SURICATA_NFS_FILE_CONFIG = Some(context);
}

fn nfs_probe_dir(i: &[u8], rdir: *mut u8) -> i8 {
    match parse_rpc_packet_header(i) {
        Ok((_, ref hdr)) => {
            let dir = if hdr.msgtype == 0 {
                Direction::ToServer
            } else {
                Direction::ToClient
            };
            unsafe { *rdir = dir as u8 };
            return 1;
        }
        Err(Err::Incomplete(_)) => {
            return 0;
        }
        Err(_) => {
            return -1;
        }
    }
}

pub fn nfs_probe(i: &[u8], direction: Direction) -> i32 {
    if direction == Direction::ToClient {
        match parse_rpc_reply(i, false) {
            Ok((_, ref rpc)) => {
                if rpc.hdr.frag_len >= 24
                    && rpc.hdr.frag_len <= 35000
                    && rpc.hdr.msgtype == 1
                    && rpc.reply_state == 0
                    && rpc.accept_state == 0
                {
                    SCLogDebug!(
                        "TC PROBE LEN {} XID {} TYPE {}",
                        rpc.hdr.frag_len,
                        rpc.hdr.xid,
                        rpc.hdr.msgtype
                    );
                    return 1;
                } else {
                    return -1;
                }
            }
            Err(Err::Incomplete(_)) => match parse_rpc_packet_header(i) {
                Ok((_, ref rpc_hdr)) => {
                    if rpc_hdr.frag_len >= 24
                        && rpc_hdr.frag_len <= 35000
                        && rpc_hdr.xid != 0
                        && rpc_hdr.msgtype == 1
                    {
                        SCLogDebug!(
                            "TC PROBE LEN {} XID {} TYPE {}",
                            rpc_hdr.frag_len,
                            rpc_hdr.xid,
                            rpc_hdr.msgtype
                        );
                        return 1;
                    } else {
                        return -1;
                    }
                }
                Err(Err::Incomplete(_)) => {
                    return 0;
                }
                Err(_) => {
                    return -1;
                }
            },
            Err(_) => {
                return -1;
            }
        }
    } else {
        match parse_rpc(i, false) {
            Ok((_, ref rpc)) => {
                if rpc.hdr.frag_len >= 40
                    && rpc.hdr.msgtype == 0
                    && rpc.rpcver == 2
                    && (rpc.progver == 3 || rpc.progver == 4)
                    && rpc.program == 100003
                    && rpc.procedure <= NFSPROC3_COMMIT
                {
                    return rpc_auth_type_known(rpc.creds_flavor) as i32;
                } else {
                    return -1;
                }
            }
            Err(Err::Incomplete(_)) => {
                return 0;
            }
            Err(_) => {
                return -1;
            }
        }
    }
}

pub fn nfs_probe_udp(i: &[u8], direction: Direction) -> i32 {
    if direction == Direction::ToClient {
        match parse_rpc_udp_reply(i) {
            Ok((_, ref rpc)) => {
                if i.len() >= 32
                    && rpc.hdr.msgtype == 1
                    && rpc.reply_state == 0
                    && rpc.accept_state == 0
                {
                    SCLogDebug!(
                        "TC PROBE LEN {} XID {} TYPE {}",
                        rpc.hdr.frag_len,
                        rpc.hdr.xid,
                        rpc.hdr.msgtype
                    );
                    return 1;
                } else {
                    return -1;
                }
            }
            Err(_) => {
                return -1;
            }
        }
    } else {
        match parse_rpc_udp_request(i) {
            Ok((_, ref rpc)) => {
                if i.len() >= 48
                    && rpc.hdr.msgtype == 0
                    && rpc.progver == 3
                    && rpc.program == 100003
                {
                    return 1;
                } else if i.len() >= 48
                    && rpc.hdr.msgtype == 0
                    && rpc.progver == 2
                    && rpc.program == 100003
                {
                    SCLogDebug!("NFSv2!");
                    return 1;
                } else {
                    return -1;
                }
            }
            Err(_) => {
                return -1;
            }
        }
    }
}

/// MIDSTREAM
unsafe extern "C" fn nfs_probe_ms(
    _flow: *const Flow, direction: u8, input: *const u8, len: u32, rdir: *mut u8,
) -> AppProto {
    if input.is_null() {
        return ALPROTO_UNKNOWN;
    }
    let slice: &[u8] = build_slice!(input, len as usize);
    SCLogDebug!("nfs_probe_ms: probing direction {:02x}", direction);
    let mut adirection: u8 = 0;
    match nfs_probe_dir(slice, &mut adirection) {
        1 => {
            if adirection == u8::from(Direction::ToServer) {
                SCLogDebug!("nfs_probe_dir said Direction::ToServer");
            } else {
                SCLogDebug!("nfs_probe_dir said Direction::ToClient");
            }
            match nfs_probe(slice, adirection.into()) {
                1 => {
                    SCLogDebug!(
                        "nfs_probe success: dir {:02x} adir {:02x}",
                        direction,
                        adirection
                    );
                    if (direction & DIR_BOTH) != adirection {
                        *rdir = adirection;
                    }
                    ALPROTO_NFS
                }
                0 => ALPROTO_UNKNOWN,
                _ => ALPROTO_FAILED,
            }
        }
        0 => ALPROTO_UNKNOWN,
        _ => ALPROTO_FAILED,
    }
}

unsafe extern "C" fn ffi_nfs_probe(
    _f: *const Flow, direction: u8, input: *const u8, len: u32, _rdir: *mut u8,
) -> AppProto {
    if input.is_null() {
        return ALPROTO_UNKNOWN;
    }
    let slice: &[u8] = build_slice!(input, len as usize);
    SCLogDebug!("ffi_nfs_probe: running probe");
    match nfs_probe(slice, direction.into()) {
        1 => ALPROTO_NFS,
        -1 => ALPROTO_FAILED,
        _ => ALPROTO_UNKNOWN,
    }
}

/// TOSERVER probe function
unsafe extern "C" fn nfs_probe_udp_ts(
    _f: *const Flow, _direction: u8, input: *const u8, len: u32, _rdir: *mut u8,
) -> AppProto {
    if input.is_null() {
        return ALPROTO_UNKNOWN;
    }
    let slice: &[u8] = build_slice!(input, len as usize);
    match nfs_probe_udp(slice, Direction::ToServer) {
        1 => ALPROTO_NFS,
        -1 => ALPROTO_FAILED,
        _ => ALPROTO_UNKNOWN,
    }
}

/// TOCLIENT probe function
unsafe extern "C" fn nfs_probe_udp_tc(
    _f: *const Flow, _direction: u8, input: *const u8, len: u32, _rdir: *mut u8,
) -> AppProto {
    if input.is_null() {
        return ALPROTO_UNKNOWN;
    }
    let slice: &[u8] = build_slice!(input, len as usize);
    match nfs_probe_udp(slice, Direction::ToClient) {
        1 => ALPROTO_NFS,
        -1 => ALPROTO_FAILED,
        _ => ALPROTO_UNKNOWN,
    }
}

// Parser name as a C style string.
const PARSER_NAME: &[u8] = b"nfs\0";

/// Load the NFS queue and record-size limits from the config. Called from
/// both TCP and UDP registration so single-transport configs load them.
unsafe fn load_nfs_limit_config() {
    let retval = conf_get("app-layer.protocols.nfs.max-read-queue-size");
    if let Some(val) = retval {
        match get_memval(val) {
            Ok(v) => match u32::try_from(v) {
                Ok(v32) => {
                    NFS_CFG_MAX_READ_QUEUE_SIZE = v32;
                }
                Err(_) => {
                    SCLogError!("Invalid max-read-queue-size value (too large)");
                }
            },
            Err(_) => {
                SCLogError!("Invalid max-read-queue-size value");
            }
        }
    }
    let retval = conf_get("app-layer.protocols.nfs.max-read-queue-cnt");
    if let Some(val) = retval {
        if let Ok(v) = val.parse::<u32>() {
            NFS_CFG_MAX_READ_QUEUE_CNT = v;
        } else {
            SCLogError!("Invalid max-read-queue-cnt value");
        }
    }
    let retval = conf_get("app-layer.protocols.nfs.max-write-queue-size");
    if let Some(val) = retval {
        match get_memval(val) {
            Ok(v) => match u32::try_from(v) {
                Ok(v32) => {
                    NFS_CFG_MAX_WRITE_QUEUE_SIZE = v32;
                }
                Err(_) => {
                    SCLogError!("Invalid max-write-queue-size value (too large)");
                }
            },
            Err(_) => {
                SCLogError!("Invalid max-write-queue-size value");
            }
        }
    }
    let retval = conf_get("app-layer.protocols.nfs.max-write-queue-cnt");
    if let Some(val) = retval {
        if let Ok(v) = val.parse::<u32>() {
            NFS_CFG_MAX_WRITE_QUEUE_CNT = v;
        } else {
            SCLogError!("Invalid max-write-queue-cnt value");
        }
    }
    let retval = conf_get("app-layer.protocols.nfs.max-write-size");
    if let Some(val) = retval {
        match get_memval(val) {
            Ok(v) => match u32::try_from(v) {
                Ok(v32) => {
                    // a value of 0 disables the check (documented)
                    NFS_CFG_MAX_WRITE_SIZE = v32;
                }
                Err(_) => {
                    SCLogError!("Invalid max-write-size value (too large)");
                }
            },
            Err(_) => {
                SCLogError!("Invalid max-write-size value");
            }
        }
    }
    let retval = conf_get("app-layer.protocols.nfs.max-read-size");
    if let Some(val) = retval {
        match get_memval(val) {
            Ok(v) => match u32::try_from(v) {
                Ok(v32) => {
                    // a value of 0 disables the check (documented)
                    NFS_CFG_MAX_READ_SIZE = v32;
                }
                Err(_) => {
                    SCLogError!("Invalid max-read-size value (too large)");
                }
            },
            Err(_) => {
                SCLogError!("Invalid max-read-size value");
            }
        }
    }
}

#[no_mangle]
pub unsafe extern "C" fn SCRegisterNfsParser() {
    let default_port = CString::new("[2049]").unwrap();
    let parser = RustParser {
        name: PARSER_NAME.as_ptr() as *const std::os::raw::c_char,
        default_port: std::ptr::null(),
        ipproto: IPPROTO_TCP,
        probe_ts: None,
        probe_tc: None,
        min_depth: 0,
        max_depth: 16,
        state_new: nfs_state_new,
        state_free: nfs_state_free,
        tx_free: nfs_state_tx_free,
        parse_ts: nfs_parse_request,
        parse_tc: nfs_parse_response,
        get_tx_count: nfs_state_get_tx_count,
        get_tx: nfs_state_get_tx,
        tx_comp_st_ts: 1,
        tx_comp_st_tc: 1,
        tx_get_progress: nfs_tx_get_alstate_progress,
        get_eventinfo: Some(NFSEvent::get_event_info),
        get_eventinfo_byid: Some(NFSEvent::get_event_info_by_id),
        localstorage_new: None,
        localstorage_free: None,
        get_tx_files: Some(nfs_gettxfiles),
        get_tx_iterator: Some(applayer::state_get_tx_iterator::<NFSState, NFSTransaction>),
        get_tx_data: nfs_get_tx_data,
        get_state_data: nfs_get_state_data,
        apply_tx_config: None,
        flags: APP_LAYER_PARSER_OPT_ACCEPT_GAPS,
        get_frame_id_by_name: Some(NFSFrameType::ffi_id_from_name),
        get_frame_name_by_id: Some(NFSFrameType::ffi_name_from_id),
        get_state_id_by_name: None,
        get_state_name_by_id: None,
    };

    let ip_proto_str = CString::new("tcp").unwrap();

    if SCAppLayerProtoDetectConfProtoDetectionEnabled(ip_proto_str.as_ptr(), parser.name) != 0 {
        let alproto = applayer_register_protocol_detection(&parser, 1);
        ALPROTO_NFS = alproto;

        let midstream = conf_get_bool("stream.midstream");
        if midstream {
            if SCAppLayerProtoDetectPPParseConfPorts(
                ip_proto_str.as_ptr(),
                IPPROTO_TCP,
                parser.name,
                ALPROTO_NFS,
                0,
                NFS_MIN_FRAME_LEN,
                Some(nfs_probe_ms),
                Some(nfs_probe_ms),
            ) == 0
            {
                SCLogDebug!("No NFSTCP app-layer configuration, enabling NFSTCP detection TCP detection on port {:?}.",
                            default_port);
                /* register 'midstream' probing parsers if midstream is enabled. */
                SCAppLayerProtoDetectPPRegister(
                    IPPROTO_TCP,
                    default_port.as_ptr(),
                    ALPROTO_NFS,
                    0,
                    NFS_MIN_FRAME_LEN,
                    Direction::ToServer.into(),
                    Some(nfs_probe_ms),
                    Some(nfs_probe_ms),
                );
            }
        } else {
            SCAppLayerProtoDetectPPRegister(
                IPPROTO_TCP,
                default_port.as_ptr(),
                ALPROTO_NFS,
                0,
                NFS_MIN_FRAME_LEN,
                Direction::ToServer.into(),
                Some(ffi_nfs_probe),
                Some(ffi_nfs_probe),
            );
        }
        if SCAppLayerParserConfParserEnabled(ip_proto_str.as_ptr(), parser.name) != 0 {
            let _ = AppLayerRegisterParser(&parser, alproto);
        }
        SCLogDebug!("Rust nfs parser registered.");
        let retval = conf_get("app-layer.protocols.nfs.max-requests");
        if let Some(val) = retval {
            if let Ok(v) = val.parse::<usize>() {
                if v > 0 {
                    NFS_CFG_MAX_REQ = v;
                } else {
                    SCLogError!("Invalid max-requests value, must be >0");
                }
            } else {
                SCLogError!("Invalid max-requests value");
            }
        }
        let retval = conf_get("app-layer.protocols.nfs.max-names");
        if let Some(val) = retval {
            if let Ok(v) = val.parse::<usize>() {
                if v > 0 {
                    NFS_CFG_MAX_NAMES = v;
                } else {
                    SCLogError!("Invalid max-names value, must be >0");
                }
            } else {
                SCLogError!("Invalid max-names value");
            }
        }
        load_nfs_limit_config();
    } else {
        SCLogDebug!("Protocol detector and parser disabled for nfs.");
    }
}

#[no_mangle]
pub unsafe extern "C" fn SCRegisterNfsUdpParser() {
    let default_port = CString::new("[2049]").unwrap();
    let parser = RustParser {
        name: PARSER_NAME.as_ptr() as *const std::os::raw::c_char,
        default_port: std::ptr::null(),
        ipproto: IPPROTO_UDP,
        probe_ts: None,
        probe_tc: None,
        min_depth: 0,
        max_depth: 16,
        state_new: nfs_state_new,
        state_free: nfs_state_free,
        tx_free: nfs_state_tx_free,
        parse_ts: nfs_parse_request_udp,
        parse_tc: nfs_parse_response_udp,
        get_tx_count: nfs_state_get_tx_count,
        get_tx: nfs_state_get_tx,
        tx_comp_st_ts: 1,
        tx_comp_st_tc: 1,
        tx_get_progress: nfs_tx_get_alstate_progress,
        get_eventinfo: Some(NFSEvent::get_event_info),
        get_eventinfo_byid: Some(NFSEvent::get_event_info_by_id),
        localstorage_new: None,
        localstorage_free: None,
        get_tx_files: Some(nfs_gettxfiles),
        get_tx_iterator: Some(applayer::state_get_tx_iterator::<NFSState, NFSTransaction>),
        get_tx_data: nfs_get_tx_data,
        get_state_data: nfs_get_state_data,
        apply_tx_config: None,
        flags: 0,
        get_frame_id_by_name: Some(NFSFrameType::ffi_id_from_name),
        get_frame_name_by_id: Some(NFSFrameType::ffi_name_from_id),
        get_state_id_by_name: None,
        get_state_name_by_id: None,
    };

    let ip_proto_str = CString::new("udp").unwrap();

    if SCAppLayerProtoDetectConfProtoDetectionEnabled(ip_proto_str.as_ptr(), parser.name) != 0 {
        let alproto = applayer_register_protocol_detection(&parser, 1);
        ALPROTO_NFS = alproto;

        if SCAppLayerProtoDetectPPParseConfPorts(
            ip_proto_str.as_ptr(),
            IPPROTO_UDP,
            parser.name,
            ALPROTO_NFS,
            0,
            NFS_MIN_FRAME_LEN,
            Some(nfs_probe_udp_ts),
            Some(nfs_probe_udp_tc),
        ) == 0
        {
            SCLogDebug!("No NFSUDP app-layer configuration, enabling NFSUDP detection UDP detection on port {:?}.",
                        default_port);
            SCAppLayerProtoDetectPPRegister(
                IPPROTO_UDP,
                default_port.as_ptr(),
                ALPROTO_NFS,
                0,
                NFS_MIN_FRAME_LEN,
                Direction::ToServer.into(),
                Some(nfs_probe_udp_ts),
                Some(nfs_probe_udp_tc),
            );
        }
        if SCAppLayerParserConfParserEnabled(ip_proto_str.as_ptr(), parser.name) != 0 {
            let _ = AppLayerRegisterParser(&parser, alproto);
        }
        if let Some(val) = conf_get("app-layer.protocols.nfs.max-tx") {
            if let Ok(v) = val.parse::<usize>() {
                NFS_MAX_TX = v;
            } else {
                SCLogError!("Invalid value for nfs.max-tx");
            }
        }
        load_nfs_limit_config();
        SCLogDebug!("Rust nfs parser registered.");
    } else {
        SCLogDebug!("Protocol detector and parser disabled for nfs.");
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    // The queue-limit tests rewrite the shared NFS_CFG_* statics: serialize.
    static NFS_CFG_TEST_LOCK: Mutex<()> = Mutex::new(());

    #[test]
    fn test_ooo_queue_backstop_thresholds() {
        let _lock = NFS_CFG_TEST_LOCK.lock().unwrap();
        // defaults: 64 MiB queue caps and 64 chunk counts
        assert_eq!(ooo_queue_backstop(Direction::ToClient), 67108864);
        assert_eq!(ooo_queue_backstop(Direction::ToServer), 67108864);
        assert_eq!(ooo_queue_backstop_cnt(Direction::ToClient), 64);
        assert_eq!(ooo_queue_backstop_cnt(Direction::ToServer), 64);

        // enabled (non-zero) limits: the backstops follow them
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_SIZE = 4096;
        }
        unsafe {
            NFS_CFG_MAX_WRITE_QUEUE_SIZE = 8192;
        }
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_CNT = 16;
        }
        unsafe {
            NFS_CFG_MAX_WRITE_QUEUE_CNT = 32;
        }
        assert_eq!(ooo_queue_backstop(Direction::ToClient), 4096);
        assert_eq!(ooo_queue_backstop(Direction::ToServer), 8192);
        assert_eq!(ooo_queue_backstop_cnt(Direction::ToClient), 16);
        assert_eq!(ooo_queue_backstop_cnt(Direction::ToServer), 32);

        // disabled (0) limits: the 1 GiB / 1024 hard caps apply
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_SIZE = 0;
        }
        unsafe {
            NFS_CFG_MAX_WRITE_QUEUE_SIZE = 0;
        }
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_CNT = 0;
        }
        unsafe {
            NFS_CFG_MAX_WRITE_QUEUE_CNT = 0;
        }
        assert_eq!(ooo_queue_backstop(Direction::ToClient), 1 << 30);
        assert_eq!(ooo_queue_backstop(Direction::ToServer), 1 << 30);
        assert_eq!(ooo_queue_backstop_cnt(Direction::ToClient), 1024);
        assert_eq!(ooo_queue_backstop_cnt(Direction::ToServer), 1024);

        // restore defaults
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_SIZE = 67108864;
        }
        unsafe {
            NFS_CFG_MAX_WRITE_QUEUE_SIZE = 67108864;
        }
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_CNT = 64;
        }
        unsafe {
            NFS_CFG_MAX_WRITE_QUEUE_CNT = 64;
        }
    }

    #[test]
    fn test_nfs_queue_limit_projection() {
        let _lock = NFS_CFG_TEST_LOCK.lock().unwrap();
        // the FFI file ops are no-ops under cfg(test), so a default file
        // 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,
        };
        let xid: u32 = 0x1234;

        unsafe {
            NFS_CFG_MAX_READ_QUEUE_SIZE = 150;
            NFS_CFG_MAX_READ_QUEUE_CNT = 0;
            NFS_CFG_MAX_WRITE_QUEUE_SIZE = 0;
            NFS_CFG_MAX_WRITE_QUEUE_CNT = 0;
        }

        let buf140 = vec![0u8; 140];

        let mut ft = FileTransferTracker::new();
        // one 140-byte OOO chunk already queued (limit is 150)
        ft.new_chunk(&TEST_FC, b"f", &buf140, 128, 140, 0, false, &xid);
        assert_eq!(ft.get_inflight_size(), 140);

        // An in-order append (offset == tracked) can only drain the queue: never
        // reject it, even though the naive projection (in_flight + len > limit) would.
        assert!(read_queue_limit_event(&ft, 0, 140).is_none());

        // a genuinely new OOO append would grow the queue past the limit:
        // rejected
        assert!(matches!(
            read_queue_limit_event(&ft, 268, 140),
            Some(NFSEvent::ReadQueueSizeExceeded)
        ));

        // the chunk path (continuation of the active chunk) only sees the
        // byte limit: 140 + 11 > 150
        assert!(read_chunk_queue_limit_event(&ft, 11).is_some());
        // exactly at the limit (140 + 10 == 150) is still allowed
        assert!(read_chunk_queue_limit_event(&ft, 10).is_none());

        // count dimension: two queued entries, limit 2
        unsafe {
            NFS_CFG_MAX_READ_QUEUE_SIZE = 0;
            NFS_CFG_MAX_READ_QUEUE_CNT = 2;
        }
        let mut ft2 = FileTransferTracker::new();
        ft2.new_chunk(&TEST_FC, b"f", &buf140, 128, 140, 0, false, &xid);
        ft2.new_chunk(&TEST_FC, b"f", &buf140, 256, 140, 0, false, &xid);
        assert_eq!(ft2.get_inflight_cnt(), 2);

        // a continuation of an already queued chunk adds no map entry:
        // not rejected even though it is OOO
        assert!(read_queue_limit_event(&ft2, 128, 140).is_none());
        // a new chunk would be the third entry: rejected on count
        assert!(matches!(
            read_queue_limit_event(&ft2, 544, 140),
            Some(NFSEvent::ReadQueueCntExceeded)
        ));

        unsafe {
            NFS_CFG_MAX_READ_QUEUE_SIZE = 67108864;
            NFS_CFG_MAX_WRITE_QUEUE_SIZE = 67108864;
            NFS_CFG_MAX_READ_QUEUE_CNT = 64;
            NFS_CFG_MAX_WRITE_QUEUE_CNT = 64;
        }
    }
}