fxcp_core/nfs/

rpc.rs

// SPDX-License-Identifier: GPL-2.0-or-later
// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
//
// fxcp-core/src/nfs/rpc.rs  --  NFSv4.2 compound RPC builder and reply parser

//! Constructs NFSv4.2 compound RPC payloads and parses server replies.
//!
//! Implements the minimal ONC RPC + NFSv4.2 wire format needed for
//! atomic file creation via compound operations.

use super::xdr::{XdrEncoder, XdrDecoder};
use super::NfsError;
use crate::constants;
use tracing::debug;

/// NFSv4.2 SEQUENCE operation code (RFC 8881 S18.46).
pub const OP_SEQUENCE: u32 = 53;
/// NFSv4 PUTFH operation code: set the current filehandle (RFC 8881 S18.19).
pub const OP_PUTFH: u32 = 22;
/// NFSv4 OPEN operation code: open or create a file (RFC 8881 S18.16).
pub const OP_OPEN: u32 = 18;
/// NFSv4 WRITE operation code: write data to a file (RFC 8881 S18.32).
pub const OP_WRITE: u32 = 38;
/// NFSv4 READ operation code: read data from a file (RFC 7530 S16.23).
pub const OP_READ: u32 = 25;
/// NFSv4 SETATTR operation code: set file attributes (RFC 8881 S18.30).
pub const OP_SETATTR: u32 = 34;
/// NFSv4 CLOSE operation code: release an open stateid (RFC 8881 S18.2).
pub const OP_CLOSE: u32 = 4;
/// NFSv4 GETATTR operation code: retrieve file attributes (RFC 8881 S18.7).
pub const OP_GETATTR: u32 = 9;
/// NFSv4.1 EXCHANGE_ID operation code: establish client identity (RFC 8881 S18.35).
pub const OP_EXCHANGE_ID: u32 = 42;
/// NFSv4.1 CREATE_SESSION operation code: create a session (RFC 8881 S18.36).
pub const OP_CREATE_SESSION: u32 = 43;
/// NFSv4.1 DESTROY_SESSION operation code: tear down a session (RFC 8881 S18.37).
pub const OP_DESTROY_SESSION: u32 = 44;
/// NFSv4 PUTROOTFH operation code: set the current filehandle to the root (RFC 8881 S18.20).
pub const OP_PUTROOTFH: u32 = 24;
/// NFSv4 LOOKUP operation code: look up a name in a directory (RFC 8881 S18.13).
pub const OP_LOOKUP: u32 = 15;
/// NFSv4 GETFH operation code: retrieve the current filehandle (RFC 8881 S18.8).
pub const OP_GETFH: u32 = 10;
/// NFSv4.1 RECLAIM_COMPLETE operation code: end grace period (RFC 8881 S18.51).
pub const OP_RECLAIM_COMPLETE: u32 = 58;

/// NFS4 status: operation succeeded.
pub const NFS4_OK: u32 = 0;
/// NFS4 error: stale filehandle (target was deleted or renamed).
pub const NFS4ERR_STALE: u32 = 70;
/// NFS4 error: file or directory not found.
pub const NFS4ERR_NOENT: u32 = 2;
/// NFS4 error: file already exists.
pub const NFS4ERR_EXIST: u32 = 17;
/// NFS4 error: permission denied.
pub const NFS4ERR_ACCESS: u32 = 13;
/// NFS4 error: invalid argument (e.g. bad attribute value, nseconds out of range).
pub const NFS4ERR_INVAL: u32 = 22;
/// NFS4 error: not a directory.
pub const NFS4ERR_NOTDIR: u32 = 20;
/// NFS4 error: is a directory (write to directory attempted).
pub const NFS4ERR_ISDIR: u32 = 21;
/// NFS4 error: file too large.
pub const NFS4ERR_FBIG: u32 = 27;
/// NFS4 error: no space on device.
pub const NFS4ERR_NOSPC: u32 = 28;
/// NFS4 error: read-only filesystem.
pub const NFS4ERR_ROFS: u32 = 30;
/// NFS4 error: filename too long.
pub const NFS4ERR_NAMETOOLONG: u32 = 63;
/// NFS4 error: directory not empty.
pub const NFS4ERR_NOTEMPTY: u32 = 66;
/// NFS4 error: server is busy, client should retry.
pub const NFS4ERR_DELAY: u32 = 10008;
/// NFS4 error: bad stateid (expired or wrong open).
pub const NFS4ERR_BAD_STATEID: u32 = 10025;
/// NFS4 error: session ID is invalid or expired.
pub const NFS4ERR_BADSESSION: u32 = 10052;
/// NFS4 error: bad sequence ID in a stateful operation.
pub const NFS4ERR_BADSEQ: u32 = 10026;
/// NFS4 error: sequence operations arrived out of order.
pub const NFS4ERR_SEQ_MISORDERED: u32 = 10063;

/// ONC RPC protocol version (always 2).
pub const RPC_VERSION: u32 = 2;
/// ONC RPC program number for NFS.
pub const NFS_PROGRAM: u32 = 100003;
/// NFS protocol major version (4).
pub const NFS_V4: u32 = 4;
/// NFS procedure number for COMPOUND (batched operations).
pub const NFSPROC4_COMPOUND: u32 = 1;
/// NFS procedure number for NULL (liveness probe).
pub const NFSPROC4_NULL: u32 = 0;
/// ONC RPC AUTH_SYS authentication flavor (Unix uid/gid credentials).
pub const AUTH_SYS: u32 = 1;
/// ONC RPC AUTH_NONE authentication flavor (no credentials).
pub const AUTH_NONE: u32 = 0;

/// OPEN share access: read only.
pub const OPEN4_SHARE_ACCESS_READ: u32 = 0x00000001;
/// OPEN share access: write only.
pub const OPEN4_SHARE_ACCESS_WRITE: u32 = 0x00000002;
/// OPEN share access: read and write.
pub const OPEN4_SHARE_ACCESS_BOTH: u32 = 0x00000003;
/// OPEN share deny: no deny (allow concurrent access).
pub const OPEN4_SHARE_DENY_NONE: u32 = 0x00000000;
/// OPEN claim type: open by filename in the current directory.
pub const CLAIM_NULL: u32 = 0;
/// OPEN type: create the file if it doesn't exist.
pub const OPEN4_CREATE: u32 = 1;
/// OPEN type: fail if the file doesn't exist.
pub const OPEN4_NOCREATE: u32 = 0;
/// Create mode: unchecked (overwrite existing attributes).
pub const CREATEMODE4_UNCHECKED: u32 = 0;

/// Write stability: server may cache, no durability guarantee.
pub const UNSTABLE4: u32 = 0;
/// Write stability: data flushed to stable storage, metadata may be cached.
pub const DATA_SYNC4: u32 = 1;
/// Write stability: data and metadata flushed to stable storage.
pub const FILE_SYNC4: u32 = 2;

/// FATTR4 bitmap position for file size (bit 4, word 0).
pub const FATTR4_SIZE: u32 = 4;

/// FATTR4 bitmap position for file mode/permissions (bit 33, word 1).
pub const FATTR4_MODE: u32 = 33;
/// FATTR4 bitmap position for file owner (bit 36, word 1).
pub const FATTR4_OWNER: u32 = 36;
/// FATTR4 bitmap position for file owner group (bit 37, word 1).
pub const FATTR4_OWNER_GROUP: u32 = 37;
/// FATTR4 bitmap position for modification time (bit 53, word 1).
pub const FATTR4_TIME_MODIFY: u32 = 53;
/// FATTR4 bitmap position for settable modification time (bit 52, word 1).
pub const FATTR4_TIME_MODIFY_SET: u32 = 52;

/// RPC reply status: message accepted by the server.
pub const MSG_ACCEPTED: u32 = 0;
/// RPC accept status: procedure executed successfully.
pub const SUCCESS: u32 = 0;

// -----------------------------------------------------------------------
// Transport abstraction types (Wave 1)
// -----------------------------------------------------------------------

/// An RPC message separated from transport framing.
///
/// Contains the XDR-encoded RPC body (call or reply) without the 4-byte
/// TCP record mark.  Wave 2 will use this struct to dispatch between
/// TCP record-mark framing and RDMA `rdma_msg` framing (RFC 8166).
#[derive(Debug, Clone)]
pub struct RpcMessage {
    /// Transaction identifier  --  echoed in the server reply.
    pub xid: u32,
    /// XDR-encoded RPC body (no record mark, no RDMA header).
    pub body: Vec<u8>,
}

/// Prepend a TCP record mark to an XDR body.
///
/// Sets the last-fragment bit (MSB) and encodes the payload length in
/// the lower 31 bits.  Standalone callers (e.g. [`super::client::probe_server_alive`])
/// use this when they cannot go through [`super::client::NfsCompoundClient::rpc_call`].
pub fn frame_rpc_message(body: &[u8]) -> Vec<u8> {
    let rm = 0x80000000u32 | (body.len() as u32);
    let mut msg = Vec::with_capacity(4 + body.len());
    msg.extend_from_slice(&rm.to_be_bytes());
    msg.extend_from_slice(body);
    msg
}

// -----------------------------------------------------------------------
// NFS4 compound operation types
// -----------------------------------------------------------------------

/// Represents an NFSv4 state identifier (returned by OPEN, used by WRITE/SETATTR/CLOSE).
#[derive(Debug, Clone, Copy)]
#[derive(Default)]
pub struct StateId {
    pub seqid: u32,
    pub other: [u8; 12],
}


impl StateId {
    /// The "current stateid" special value  --  tells the server to use the stateid
    /// from the most recent stateful operation (OPEN) in this compound.
    /// RFC 5661 S16.2.3.1.2: seqid=1, other=all-zeros.
    /// (NOT seqid=0xFFFFFFFF/other=all-0xFF, which is the anonymous/READ bypass stateid.)
    pub fn current() -> Self {
        Self { seqid: 1, other: [0u8; 12] }
    }
}

/// Write stability level.
#[derive(Debug, Clone, Copy)]
pub enum WriteStable {
    Unstable,
    DataSync,
    FileSync,
}

/// A single NFSv4.2 compound operation.
#[derive(Debug, Clone)]
pub enum Nfs4Op {
    Sequence {
        session_id: [u8; 16],
        sequence_id: u32,
        slot_id: u32,
        highest_slot_id: u32,
        cache_this: bool,
    },
    PutFh {
        handle: Vec<u8>,
    },
    PutRootFh,
    Lookup {
        name: String,
    },
    Open {
        seqid: u32,
        share_access: u32,
        share_deny: u32,
        clientid: u64,
        owner: Vec<u8>,
        filename: String,
        mode: u32,
        create_owner: Option<String>,
        create_group: Option<String>,
        create_mtime: Option<(i64, i64)>,
    },
    Write {
        stateid: StateId,
        offset: u64,
        stable: WriteStable,
        data: Vec<u8>,
    },
    Read {
        stateid: StateId,
        offset: u64,
        count: u32,
    },
    SetAttr {
        stateid: StateId,
        mode: Option<u32>,
        owner: Option<String>,
        owner_group: Option<String>,
        mtime: Option<(i64, i64)>,
    },
    Close {
        seqid: u32,
        stateid: StateId,
    },
    GetAttr {
        attr_request: [u32; 2],
    },
    GetFh,
    /// Tell server we have no state to reclaim (ends grace period for this session).
    ReclaimComplete,
}

/// Parsed result of a single operation in a compound reply.
#[derive(Debug)]
pub struct OpResult {
    pub op: u32,
    pub status: u32,
    /// For OPEN: the returned stateid.
    pub stateid: Option<StateId>,
    /// For GETFH: the returned filehandle.
    pub filehandle: Option<Vec<u8>>,
    /// For GETATTR: file size (if SIZE bit was in bitmap).
    pub size: Option<u64>,
    /// For GETATTR: modification time (seconds, nanoseconds).
    pub mtime: Option<(i64, i64)>,
    /// For READ: the returned file data.
    pub data: Option<Vec<u8>>,
}

/// Parsed NFSv4.2 compound reply.
#[derive(Debug)]
pub struct CompoundReply {
    pub xid: u32,
    pub status: u32,
    pub tag: Vec<u8>,
    pub op_results: Vec<OpResult>,
}

// -----------------------------------------------------------------------
// Encoding
// -----------------------------------------------------------------------

/// Encode AUTH_SYS credentials into XDR.
fn encode_auth_sys(enc: &mut XdrEncoder, uid: u32, gid: u32, machine: &str) {
    enc.encode_u32(AUTH_SYS);  // flavor
    // Auth body (length-prefixed)
    let mut body = XdrEncoder::new(64);
    body.encode_u32(0);         // stamp
    body.encode_string(machine); // machinename
    body.encode_u32(uid);        // uid
    body.encode_u32(gid);        // gid
    body.encode_u32(1);          // gids count
    body.encode_u32(gid);        // gids[0]
    let body_bytes = body.into_bytes();
    enc.encode_opaque(&body_bytes);
}

/// Encode a single NFSv4 operation into XDR.
fn encode_op(enc: &mut XdrEncoder, op: &Nfs4Op) {
    match op {
        Nfs4Op::Sequence { session_id, sequence_id, slot_id, highest_slot_id, cache_this } => {
            enc.encode_u32(OP_SEQUENCE);
            enc.encode_opaque_fixed(session_id);  // 16 bytes, no length prefix
            enc.encode_u32(*sequence_id);
            enc.encode_u32(*slot_id);
            enc.encode_u32(*highest_slot_id);
            enc.encode_bool(*cache_this);
        }
        Nfs4Op::PutFh { handle } => {
            enc.encode_u32(OP_PUTFH);
            enc.encode_opaque(handle);
        }
        Nfs4Op::PutRootFh => {
            enc.encode_u32(OP_PUTROOTFH);
        }
        Nfs4Op::Lookup { name } => {
            enc.encode_u32(OP_LOOKUP);
            enc.encode_string(name);
        }
        Nfs4Op::Open { seqid, share_access, share_deny, clientid, owner, filename, mode,
                       create_owner, create_group, create_mtime } => {
            enc.encode_u32(OP_OPEN);
            enc.encode_u32(*seqid);
            enc.encode_u32(*share_access);
            enc.encode_u32(*share_deny);
            enc.encode_u64(*clientid);
            enc.encode_opaque(owner);
            enc.encode_u32(OPEN4_CREATE);
            enc.encode_u32(CREATEMODE4_UNCHECKED);
            // createattrs fattr4: bitmap word 1 bits in ascending order
            let mut bitmap_w1: u32 = 1 << (FATTR4_MODE - 32);
            if create_owner.is_some() { bitmap_w1 |= 1 << (FATTR4_OWNER - 32); }
            if create_group.is_some() { bitmap_w1 |= 1 << (FATTR4_OWNER_GROUP - 32); }
            if create_mtime.is_some() { bitmap_w1 |= 1 << (FATTR4_TIME_MODIFY_SET - 32); }
            enc.encode_u32(2);
            enc.encode_u32(0);
            enc.encode_u32(bitmap_w1);
            let mut attr = XdrEncoder::new(128);
            attr.encode_u32(*mode & 0o7777);
            if let Some(o) = create_owner {
                attr.encode_string(o);
            }
            if let Some(g) = create_group {
                attr.encode_string(g);
            }
            if let Some((secs, nsecs)) = create_mtime {
                attr.encode_u32(1); // SET_TO_CLIENT_TIME4
                attr.encode_i64(*secs);
                attr.encode_u32((*nsecs as u32).min(999_999_999));
            }
            let attr_bytes = attr.into_bytes();
            enc.encode_opaque(&attr_bytes);
            enc.encode_u32(CLAIM_NULL);
            enc.encode_string(filename);
        }
        Nfs4Op::Write { stateid, offset, stable, data } => {
            enc.encode_u32(OP_WRITE);
            // stateid4
            enc.encode_u32(stateid.seqid);
            enc.encode_opaque_fixed(&stateid.other);
            enc.encode_u64(*offset);
            enc.encode_u32(match stable {
                WriteStable::Unstable => UNSTABLE4,
                WriteStable::DataSync => DATA_SYNC4,
                WriteStable::FileSync => FILE_SYNC4,
            });
            enc.encode_opaque(data);
        }
        Nfs4Op::Read { stateid, offset, count } => {
            enc.encode_u32(OP_READ);
            enc.encode_u32(stateid.seqid);
            enc.encode_opaque_fixed(&stateid.other);
            enc.encode_u64(*offset);
            enc.encode_u32(*count);
        }
        Nfs4Op::SetAttr { stateid, mode, owner, owner_group, mtime } => {
            enc.encode_u32(OP_SETATTR);
            enc.encode_u32(stateid.seqid);
            enc.encode_opaque_fixed(&stateid.other);
            let mut bitmap_w1: u32 = 0;
            if mode.is_some() { bitmap_w1 |= 1 << (FATTR4_MODE - 32); }
            if owner.is_some() { bitmap_w1 |= 1 << (FATTR4_OWNER - 32); }
            if owner_group.is_some() { bitmap_w1 |= 1 << (FATTR4_OWNER_GROUP - 32); }
            if mtime.is_some() { bitmap_w1 |= 1 << (FATTR4_TIME_MODIFY_SET - 32); }
            enc.encode_u32(2);   // 2 bitmap words
            enc.encode_u32(0);   // word 0
            enc.encode_u32(bitmap_w1);
            let mut attr = XdrEncoder::new(128);
            if let Some(m) = mode {
                attr.encode_u32(*m);
            }
            if let Some(o) = owner {
                attr.encode_string(o);
            }
            if let Some(g) = owner_group {
                attr.encode_string(g);
            }
            if let Some((secs, nsecs)) = mtime {
                let clamped_nsecs = (*nsecs as u32).min(999_999_999);
                if *nsecs < 0 || *nsecs > 999_999_999 {
                    debug!(
                        "SETATTR: clamping out-of-range nseconds {} -> {}",
                        nsecs, clamped_nsecs
                    );
                }
                attr.encode_u32(1); // SET_TO_CLIENT_TIME4
                attr.encode_i64(*secs);
                attr.encode_u32(clamped_nsecs);
            }
            let attr_bytes = attr.into_bytes();
            debug!(
                "SETATTR: bitmap_w1={:#010x} attr_vals={} bytes owner={:?} group={:?} mtime={:?}",
                bitmap_w1, attr_bytes.len(), owner, owner_group, mtime
            );
            enc.encode_opaque(&attr_bytes);
        }
        Nfs4Op::Close { seqid, stateid } => {
            enc.encode_u32(OP_CLOSE);
            enc.encode_u32(*seqid);
            enc.encode_u32(stateid.seqid);
            enc.encode_opaque_fixed(&stateid.other);
        }
        Nfs4Op::GetAttr { attr_request } => {
            enc.encode_u32(OP_GETATTR);
            enc.encode_u32(2);  // bitmap length
            enc.encode_u32(attr_request[0]);
            enc.encode_u32(attr_request[1]);
        }
        Nfs4Op::GetFh => {
            enc.encode_u32(OP_GETFH);
        }
        Nfs4Op::ReclaimComplete => {
            enc.encode_u32(OP_RECLAIM_COMPLETE);
            enc.encode_bool(false); // rca_one_fs = false (complete for all filesystems)
        }
    }
}

/// Build an NFSv4.2 COMPOUND RPC call body (XDR, no record mark).
///
/// Returns the XDR-encoded RPC body ready for [`frame_rpc_message`] or
/// [`super::client::NfsCompoundClient::rpc_call`].
pub fn build_compound(
    xid: u32,
    tag: &str,
    uid: u32,
    gid: u32,
    machine: &str,
    ops: &[Nfs4Op],
) -> Vec<u8> {
    let mut body = XdrEncoder::new(constants::NFS_XDR_ENCODER_BODY_CAPACITY);

    // ONC RPC header
    body.encode_u32(xid);            // XID
    body.encode_u32(0);              // CALL (not REPLY)
    body.encode_u32(RPC_VERSION);    // RPC version 2
    body.encode_u32(NFS_PROGRAM);    // program: NFS
    body.encode_u32(NFS_V4);         // version: 4
    body.encode_u32(NFSPROC4_COMPOUND); // procedure: COMPOUND

    // AUTH_SYS credentials
    encode_auth_sys(&mut body, uid, gid, machine);

    // Verifier (AUTH_NONE)
    body.encode_u32(AUTH_NONE);
    body.encode_u32(0); // verifier body length

    // COMPOUND args
    body.encode_string(tag);         // tag
    body.encode_u32(2);              // minor version (4.2)
    body.encode_u32(ops.len() as u32); // argarray count

    for op in ops {
        encode_op(&mut body, op);
    }

    body.into_bytes()
}

/// Parse a compound reply, extracting status codes and the OPEN stateid.
///
/// This is a minimal parser  --  it extracts enough to determine success/failure
/// and to get the stateid from OPEN (needed for WRITE, SETATTR, CLOSE).
pub fn parse_compound_reply(data: &[u8]) -> Result<CompoundReply, NfsError> {
    // Skip record mark (4 bytes)
    if data.len() < 4 {
        return Err(NfsError::XdrDecode("reply too short".into()));
    }
    let mut dec = XdrDecoder::new(&data[4..]);

    // RPC reply header
    let xid = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let msg_type = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    if msg_type != 1 { // REPLY
        return Err(NfsError::RpcError(format!("expected REPLY (1), got {}", msg_type)));
    }

    let reply_stat = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    if reply_stat != MSG_ACCEPTED {
        return Err(NfsError::RpcError(format!("RPC rejected: {}", reply_stat)));
    }

    // Verifier
    let _verf_flavor = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let _verf_body = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;

    // Accept status
    let accept_stat = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    if accept_stat != SUCCESS {
        return Err(NfsError::RpcError(format!("RPC accept error: {}", accept_stat)));
    }

    // COMPOUND reply
    let compound_status = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let tag = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?.to_vec();
    let num_results = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;

    let mut op_results = Vec::with_capacity(num_results as usize);
    for _ in 0..num_results {
        let op = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
        let status = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;

        let mut filehandle = None;
        let stateid = None;
        let mut size = None;
        let mut mtime = None;
        let mut data_field: Option<Vec<u8>> = None;

        // Parse enough of each op result to skip to the next
        if status == NFS4_OK {
            match op {
                OP_SEQUENCE => {
                    dec.skip_raw(16 + 4 + 4 + 4 + 4 + 4).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                }
                OP_PUTFH | OP_PUTROOTFH | OP_LOOKUP | OP_RECLAIM_COMPLETE => {
                    // No result data
                }
                OP_GETFH => {
                    let fh = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    filehandle = Some(fh.to_vec());
                }
                OP_OPEN => {
                    // OPEN4resok: stateid + change_info4 + rflags + bitmap4 + delegation
                    let sid_seqid = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    let sid_other = dec.decode_opaque_fixed(12).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    let mut other = [0u8; 12];
                    other.copy_from_slice(sid_other);

                    // change_info4: atomic(bool=4) + before(u64=8) + after(u64=8) = 20 bytes
                    dec.skip_raw(20).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    // rflags: 4 bytes
                    dec.skip_raw(4).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    // bitmap4 attrset: count(u32) + count x u32
                    let bm_len = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    dec.skip_raw(bm_len as usize * 4).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    // open_delegation4: discriminant + type-dependent body
                    let deleg_type = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    match deleg_type {
                        0 => { /* OPEN_DELEGATE_NONE: void */ }
                        3 => {
                            // OPEN_DELEGATE_NONE_EXT: ond_why(u32) + optional bool
                            let why = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                            if why == 1 || why == 2 {
                                dec.skip_raw(4).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                            }
                        }
                        _ => {
                            // READ/WRITE delegation  --  variable-length nfsace4, can't skip safely
                            op_results.push(OpResult { op, status, stateid: Some(StateId { seqid: sid_seqid, other }), filehandle: None, size: None, mtime: None, data: None });
                            break;
                        }
                    }

                    op_results.push(OpResult { op, status, stateid: Some(StateId { seqid: sid_seqid, other }), filehandle: None, size: None, mtime: None, data: None });
                    continue;
                }
                OP_WRITE => {
                    dec.skip_raw(4 + 4 + 8).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                }
                OP_SETATTR => {
                    let bm_len = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    dec.skip_raw(bm_len as usize * 4).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                }
                OP_CLOSE => {
                    dec.skip_raw(4 + 12).map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                }
                OP_READ => {
                    let _eof = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    let raw = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    data_field = Some(raw.to_vec());
                }
                OP_GETATTR => {
                    // Parse bitmap + attribute values for SIZE and TIME_MODIFY
                    let bm_len = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    let mut bitmap = vec![0u32; bm_len as usize];
                    for b in bitmap.iter_mut() {
                        *b = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    }
                    let attr_data = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
                    // Parse requested attributes from the opaque value in bitmap order
                    let mut attr_dec = XdrDecoder::new(attr_data);
                    if bm_len > 0 && (bitmap[0] & (1 << FATTR4_SIZE)) != 0 {
                        size = attr_dec.decode_u64().ok();
                    }
                    if bm_len > 1 && (bitmap[1] & (1 << (FATTR4_TIME_MODIFY - 32))) != 0 {
                        // nfstime4: seconds (i64) + nseconds (u32)
                        if let (Ok(secs), Ok(nsecs)) = (attr_dec.decode_u64(), attr_dec.decode_u32()) {
                            mtime = Some((secs as i64, nsecs as i64));
                        }
                    }
                }
                _ => {
                    break; // Unknown op  --  can't skip safely
                }
            }
        }

        op_results.push(OpResult { op, status, stateid, filehandle, size, mtime, data: data_field });
    }

    Ok(CompoundReply {
        xid,
        status: compound_status,
        tag,
        op_results,
    })
}

/// Get the human-readable name for an NFS4 error code.
pub fn nfs4_error_name(code: u32) -> &'static str {
    match code {
        NFS4_OK => "NFS4_OK",
        1 => "NFS4ERR_PERM",
        NFS4ERR_NOENT => "NFS4ERR_NOENT",
        5 => "NFS4ERR_IO",
        6 => "NFS4ERR_NXIO",
        NFS4ERR_ACCESS => "NFS4ERR_ACCESS",
        NFS4ERR_EXIST => "NFS4ERR_EXIST",
        18 => "NFS4ERR_XDEV",
        NFS4ERR_NOTDIR => "NFS4ERR_NOTDIR",
        NFS4ERR_ISDIR => "NFS4ERR_ISDIR",
        NFS4ERR_INVAL => "NFS4ERR_INVAL",
        NFS4ERR_FBIG => "NFS4ERR_FBIG",
        NFS4ERR_NOSPC => "NFS4ERR_NOSPC",
        NFS4ERR_ROFS => "NFS4ERR_ROFS",
        NFS4ERR_NAMETOOLONG => "NFS4ERR_NAMETOOLONG",
        NFS4ERR_NOTEMPTY => "NFS4ERR_NOTEMPTY",
        NFS4ERR_STALE => "NFS4ERR_STALE",
        10003 => "NFS4ERR_BADHANDLE",
        10006 => "NFS4ERR_SERVERFAULT",
        10008 => "NFS4ERR_DELAY",
        10010 => "NFS4ERR_SAME",
        10011 => "NFS4ERR_DENIED",
        10013 => "NFS4ERR_GRACE",
        10015 => "NFS4ERR_SHARE_DENIED",
        10016 => "NFS4ERR_WRONGSEC",
        10019 => "NFS4ERR_REQ_TOO_BIG",
        NFS4ERR_BAD_STATEID => "NFS4ERR_BAD_STATEID",
        NFS4ERR_BADSEQ => "NFS4ERR_BADSEQ",
        10028 => "NFS4ERR_REP_TOO_BIG",
        NFS4ERR_BADSESSION => "NFS4ERR_BADSESSION",
        10053 => "NFS4ERR_BADSLOT",
        10054 => "NFS4ERR_COMPLETE_ALREADY",
        NFS4ERR_SEQ_MISORDERED => "NFS4ERR_SEQ_MISORDERED",
        10044 => "NFS4ERR_OP_ILLEGAL",
        10068 => "NFS4ERR_RETRY_UNCACHED_REP",
        _ => "NFS4ERR_UNKNOWN",
    }
}

/// ONC RPC RPCSEC_GSS authentication flavor (RFC 2203).
pub const AUTH_RPCSEC_GSS: u32 = 6;
/// ONC RPC AUTH_TLS authentication flavor (RFC 9289 S4.1).
///
/// Used in the credential field of a NULL RPC call to probe whether the
/// server supports RPC-over-TLS.  The server replies with a verifier body
/// of `b"STARTTLS"` when TLS is available.
pub const AUTH_TLS: u32 = 7;
/// RPCSEC_GSS protocol version.
pub const RPCSEC_GSS_VERSION: u32 = 1;
/// RPCSEC_GSS procedure: data exchange (after context established).
pub const RPCSEC_GSS_DATA: u32 = 0;
/// RPCSEC_GSS procedure: initial context establishment (RFC 2203 S5.2.2).
#[cfg(feature = "krb5")]
pub const RPCSEC_GSS_INIT: u32 = 1;
/// RPCSEC_GSS procedure: continuation of multi-round context establishment.
#[cfg(feature = "krb5")]
pub const RPCSEC_GSS_CONTINUE_INIT: u32 = 2;

/// Build an RPCSEC_GSS-authenticated NFSv4.2 COMPOUND call.
///
/// Replaces AUTH_SYS credentials with RPCSEC_GSS credentials and computes
/// the verifier MIC over the RPC header per RFC 2203 S5.3.3.1.
/// For krb5p (privacy), the compound args body is wrapped via `gss.wrap_body()`.
#[cfg(feature = "krb5")]
pub fn build_compound_gss(
    xid: u32,
    tag: &str,
    ops: &[Nfs4Op],
    gss: &mut super::gss::GssContext,
) -> Result<Vec<u8>, super::NfsError> {
    use super::gss::RpcGssService;

    let service = gss.rpc_gss_service();

    // RFC 2203 S5.3.3.3: seq_num starts at 1, increments by 1 per DATA call
    gss.sequence_num += 1;

    // Phase 1: Encode compound args (tag + minor_version + ops).
    let mut args_enc = XdrEncoder::new(constants::NFS_XDR_ENCODER_BODY_CAPACITY);
    args_enc.encode_string(tag);
    args_enc.encode_u32(2); // minor version 4.2
    args_enc.encode_u32(ops.len() as u32);
    for op in ops {
        encode_op(&mut args_enc, op);
    }
    let compound_args = args_enc.into_bytes();

    debug!(
        "build_compound_gss: tag={} ops={} compound_args={} bytes, last 32: {:02x?}",
        tag, ops.len(), compound_args.len(),
        &compound_args[compound_args.len().saturating_sub(32)..]
    );

    // Phase 2: Encode RPC header + RPCSEC_GSS credential.
    let mut header = XdrEncoder::new(constants::NFS_XDR_ENCODER_TINY_CAPACITY);
    header.encode_u32(xid);
    header.encode_u32(0); // CALL
    header.encode_u32(RPC_VERSION);
    header.encode_u32(NFS_PROGRAM);
    header.encode_u32(NFS_V4);
    header.encode_u32(NFSPROC4_COMPOUND);

    // RPCSEC_GSS credential (RFC 2203 S5.2.2)
    let mut cred_body = XdrEncoder::new(64);
    cred_body.encode_u32(RPCSEC_GSS_VERSION);
    cred_body.encode_u32(RPCSEC_GSS_DATA);
    cred_body.encode_u32(gss.sequence_num);
    cred_body.encode_u32(service as u32);
    cred_body.encode_opaque(&gss.server_handle);
    let cred_bytes = cred_body.into_bytes();

    header.encode_u32(AUTH_RPCSEC_GSS);
    header.encode_opaque(&cred_bytes);

    let header_bytes = header.into_bytes();

    // Phase 3: Verifier = MIC over RPC header (xid through credential).
    let mic = gss.get_mic(&header_bytes)?;

    // Phase 4: Assemble full message.
    let mut body = XdrEncoder::new(header_bytes.len() + mic.len() + compound_args.len() + 64);
    body.encode_raw(&header_bytes);

    // Verifier: flavor=RPCSEC_GSS, body=MIC
    body.encode_u32(AUTH_RPCSEC_GSS);
    body.encode_opaque(&mic);

    // Phase 5: Procedure args  --  service-dependent encoding.
    match service {
        RpcGssService::None => {
            // krb5: plaintext args
            body.encode_raw(&compound_args);
        }
        RpcGssService::Integrity => {
            // krb5i: { seq_num || args } + MIC over that (RFC 2203 S5.3.2.2)
            let mut integ_data = XdrEncoder::new(compound_args.len() + 8);
            integ_data.encode_u32(gss.sequence_num);
            integ_data.encode_raw(&compound_args);
            let integ_bytes = integ_data.into_bytes();

            let integ_mic = gss.get_mic(&integ_bytes)?;
            body.encode_opaque(&integ_bytes);
            body.encode_opaque(&integ_mic);
        }
        RpcGssService::Privacy => {
            // krb5p: wrap(seq_num || args) as single unit (RFC 2203 S5.3.2.3)
            let mut priv_data = XdrEncoder::new(compound_args.len() + 8);
            priv_data.encode_u32(gss.sequence_num);
            priv_data.encode_raw(&compound_args);
            let priv_bytes = priv_data.into_bytes();

            let wrapped = gss.wrap_body(&priv_bytes)?;
            debug!(
                "krb5p wrap: priv_bytes={} wrapped={} overhead={} token_hdr={:02x?}",
                priv_bytes.len(), wrapped.len(),
                wrapped.len().saturating_sub(priv_bytes.len()),
                &wrapped[..wrapped.len().min(16)]
            );
            body.encode_opaque(&wrapped);
        }
    }

    Ok(body.into_bytes())
}

/// Build an RPCSEC_GSS INIT RPC to exchange the client's GSS token for a
/// server-assigned context handle (RFC 2203 S5.2.2).
///
/// Uses NFS NULL procedure with RPCSEC_GSS credential (proc=INIT, seq=0).
/// `handle` is empty for the first INIT; set to the server's handle for
/// CONTINUE_INIT rounds.
#[cfg(feature = "krb5")]
pub fn build_rpcsec_gss_init(
    xid: u32,
    handle: &[u8],
    gss_token: &[u8],
) -> Vec<u8> {
    let mut body = XdrEncoder::new(256 + gss_token.len());

    // RPC header
    body.encode_u32(xid);
    body.encode_u32(0); // CALL
    body.encode_u32(RPC_VERSION);
    body.encode_u32(NFS_PROGRAM);
    body.encode_u32(NFS_V4);
    body.encode_u32(NFSPROC4_NULL); // NULL procedure

    // RPCSEC_GSS credential
    let proc = if handle.is_empty() { RPCSEC_GSS_INIT } else { RPCSEC_GSS_CONTINUE_INIT };
    let mut cred_body = XdrEncoder::new(64);
    cred_body.encode_u32(RPCSEC_GSS_VERSION); // version = 1
    cred_body.encode_u32(proc);                // RPCSEC_GSS_INIT or CONTINUE_INIT
    cred_body.encode_u32(0);                   // seq_num = 0 for INIT
    cred_body.encode_u32(1);                   // service = rpc_gss_svc_none
    cred_body.encode_opaque(handle);           // context handle (empty for first INIT)
    let cred_bytes = cred_body.into_bytes();

    body.encode_u32(AUTH_RPCSEC_GSS); // credential flavor
    body.encode_opaque(&cred_bytes);  // credential body

    // Verifier: AUTH_NONE
    body.encode_u32(AUTH_NONE);
    body.encode_u32(0);

    // Call body: gss_init_arg = opaque gss_token<>
    body.encode_opaque(gss_token);

    body.into_bytes()
}

/// Parse an RPCSEC_GSS INIT reply (RFC 2203 S5.2.2).
///
/// Returns `(server_handle, gss_major, gss_minor, seq_window, continuation_token)`.
#[cfg(feature = "krb5")]
pub fn parse_rpcsec_gss_init_reply(
    data: &[u8],
) -> Result<(Vec<u8>, u32, u32, u32, Vec<u8>), NfsError> {
    if data.len() < 4 {
        return Err(NfsError::XdrDecode("RPCSEC_GSS INIT reply too short".into()));
    }
    let mut dec = XdrDecoder::new(&data[4..]); // skip record mark

    // RPC reply header
    let _xid = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let msg_type = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    if msg_type != 1 {
        return Err(NfsError::RpcError(format!("expected REPLY (1), got {}", msg_type)));
    }

    let reply_stat = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    if reply_stat != MSG_ACCEPTED {
        return Err(NfsError::RpcError(format!("RPCSEC_GSS INIT rejected: {}", reply_stat)));
    }

    // Verifier
    let _verf_flavor = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let _verf_body = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;

    // Accept status
    let accept_stat = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    if accept_stat != SUCCESS {
        return Err(NfsError::RpcError(format!("RPCSEC_GSS INIT accept error: {}", accept_stat)));
    }

    // gss_init_res body
    let server_handle = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?.to_vec();
    let gss_major = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let gss_minor = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let seq_window = dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
    let cont_token = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?.to_vec();

    Ok((server_handle, gss_major, gss_minor, seq_window, cont_token))
}

/// Unwrap a GSS-protected NFS compound reply for `parse_compound_reply()`.
///
/// For krb5p: decrypts the wrapped body. For krb5i: verifies the MIC.
/// For krb5 (auth-only): passes through unchanged.
/// Returns data suitable for `parse_compound_reply()`.
#[cfg(feature = "krb5")]
pub fn unwrap_gss_reply(
    data: &[u8],
    gss: &mut super::gss::GssContext,
) -> Result<Vec<u8>, NfsError> {
    use super::gss::RpcGssService;

    let service = gss.rpc_gss_service();
    if matches!(service, RpcGssService::None) {
        return Ok(data.to_vec());
    }

    if data.len() < 4 {
        return Err(NfsError::XdrDecode("GSS reply too short".into()));
    }
    let mut dec = XdrDecoder::new(&data[4..]); // skip record mark

    // Parse RPC reply header (always plaintext)
    dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?; // xid
    dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?; // msg_type
    dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?; // reply_stat
    dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?; // verf_flavor
    dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?; // verf_body
    dec.decode_u32().map_err(|e| NfsError::XdrDecode(e.to_string()))?; // accept_stat

    let header_end = 4 + dec.position(); // offset after RPC header in original data

    let compound_body = match service {
        RpcGssService::None => unreachable!(),
        RpcGssService::Privacy => {
            let wrapped = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
            let plaintext = gss.unwrap_body(wrapped)?;
            if plaintext.len() < 4 {
                return Err(NfsError::XdrDecode("unwrapped krb5p reply too short for seq_num".into()));
            }
            plaintext[4..].to_vec() // skip seq_num prefix
        }
        RpcGssService::Integrity => {
            let integ_data = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
            let checksum = dec.decode_opaque().map_err(|e| NfsError::XdrDecode(e.to_string()))?;
            gss.verify_mic(integ_data, checksum)?;
            if integ_data.len() < 4 {
                return Err(NfsError::XdrDecode("krb5i integrity data too short for seq_num".into()));
            }
            integ_data[4..].to_vec() // skip seq_num prefix
        }
    };

    debug!(
        "unwrap_gss_reply: header_end={} compound_body={} bytes, hex: {:02x?}",
        header_end, compound_body.len(), &compound_body[..compound_body.len().min(200)]
    );

    // Reconstruct: original RPC header + plaintext compound body
    let total_len = header_end + compound_body.len();
    let mut result = Vec::with_capacity(total_len);
    result.extend_from_slice(&data[..header_end]);
    result.extend_from_slice(&compound_body);
    // Fix record mark to reflect new payload size
    let rm = 0x80000000u32 | ((total_len - 4) as u32);
    result[..4].copy_from_slice(&rm.to_be_bytes());

    Ok(result)
}

/// Build a minimal NFS NULL RPC call (zero-overhead server liveness check).
///
/// The NULL procedure requires no session, no filehandle, no authentication.
/// Response time is <1ms when the server is reachable. Standard NFS liveness probe.
pub fn build_null_call(xid: u32) -> Vec<u8> {
    let mut body = XdrEncoder::new(64);
    body.encode_u32(xid);
    body.encode_u32(0);              // CALL
    body.encode_u32(RPC_VERSION);
    body.encode_u32(NFS_PROGRAM);
    body.encode_u32(NFS_V4);
    body.encode_u32(NFSPROC4_NULL);  // procedure 0 = NULL
    // AUTH_NONE credentials + verifier
    body.encode_u32(AUTH_NONE);
    body.encode_u32(0);
    body.encode_u32(AUTH_NONE);
    body.encode_u32(0);
    body.into_bytes()
}

/// Build an AUTH_TLS probe (RFC 9289 S4.1) to detect RPC-over-TLS support.
///
/// Identical to [`build_null_call`] except the credential flavor is
/// `AUTH_TLS` (7) instead of `AUTH_NONE` (0).  The verifier remains
/// `AUTH_NONE`.  If the server supports TLS it replies with a verifier
/// body of `b"STARTTLS"`.
pub fn build_auth_tls_probe(xid: u32) -> Vec<u8> {
    let mut body = XdrEncoder::new(64);
    body.encode_u32(xid);
    body.encode_u32(0); // CALL
    body.encode_u32(RPC_VERSION);
    body.encode_u32(NFS_PROGRAM);
    body.encode_u32(NFS_V4);
    body.encode_u32(NFSPROC4_NULL);
    // AUTH_TLS credential (flavor=7, body_len=0)
    body.encode_u32(AUTH_TLS);
    body.encode_u32(0);
    // AUTH_NONE verifier
    body.encode_u32(AUTH_NONE);
    body.encode_u32(0);
    body.into_bytes()
}

/// Parse an ONC RPC reply to an AUTH_TLS probe (RFC 9289 S4.1).
///
/// Returns `true` when the server indicates TLS support: the reply must be
/// `MSG_ACCEPTED` with a verifier whose body is exactly `b"STARTTLS"`.
pub fn parse_auth_tls_reply(reply: &[u8]) -> bool {
    // Minimum: 4 RM + 4 XID + 4 REPLY + 4 accept_status
    //        + 4 verf_flavor + 4 verf_len + 8 verf_body = 32
    if reply.len() < 32 {
        return false;
    }
    let mut dec = XdrDecoder::new(&reply[4..]); // skip record mark
    let Ok(_xid) = dec.decode_u32() else { return false };
    let Ok(msg_type) = dec.decode_u32() else { return false };
    if msg_type != 1 {
        return false;
    }
    let Ok(accept_status) = dec.decode_u32() else { return false };
    if accept_status != MSG_ACCEPTED {
        return false;
    }
    let Ok(verf_flavor) = dec.decode_u32() else { return false };
    let Ok(verf_len) = dec.decode_u32() else { return false };
    if verf_flavor != AUTH_NONE || verf_len != 8 {
        return false;
    }
    if dec.remaining() < 8 {
        return false;
    }
    let Ok(verf_body) = dec.decode_opaque_fixed(8) else { return false };
    verf_body == b"STARTTLS"
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    use super::*;

    #[test]
    fn auth_tls_probe_xdr_encoding() {
        let probe = build_auth_tls_probe(0x42);
        // Body only (no record mark): 10 x u32 = 40 bytes
        assert_eq!(probe.len(), 40);
        assert_eq!(u32::from_be_bytes(probe[0..4].try_into().unwrap()), 0x42); // XID
        assert_eq!(u32::from_be_bytes(probe[4..8].try_into().unwrap()), 0); // CALL
        assert_eq!(u32::from_be_bytes(probe[8..12].try_into().unwrap()), 2); // RPC ver
        assert_eq!(u32::from_be_bytes(probe[12..16].try_into().unwrap()), 100003); // NFS prog
        assert_eq!(u32::from_be_bytes(probe[16..20].try_into().unwrap()), 4); // NFSv4
        assert_eq!(u32::from_be_bytes(probe[20..24].try_into().unwrap()), 0); // NULL proc
        assert_eq!(u32::from_be_bytes(probe[24..28].try_into().unwrap()), 7); // AUTH_TLS
        assert_eq!(u32::from_be_bytes(probe[28..32].try_into().unwrap()), 0); // cred len
        assert_eq!(u32::from_be_bytes(probe[32..36].try_into().unwrap()), 0); // verf AUTH_NONE
        assert_eq!(u32::from_be_bytes(probe[36..40].try_into().unwrap()), 0); // verf len
    }

    #[test]
    fn auth_tls_probe_differs_from_null_only_in_credential_flavor() {
        let null_call = build_null_call(0xAA);
        let tls_probe = build_auth_tls_probe(0xAA);
        assert_eq!(null_call.len(), tls_probe.len());
        // Everything before credential flavor (offset 24) is identical
        assert_eq!(&null_call[..24], &tls_probe[..24]);
        // Credential flavor: AUTH_NONE(0) vs AUTH_TLS(7)
        assert_eq!(u32::from_be_bytes(null_call[24..28].try_into().unwrap()), 0);
        assert_eq!(u32::from_be_bytes(tls_probe[24..28].try_into().unwrap()), 7);
        // Rest (cred body len + verifier) identical
        assert_eq!(&null_call[28..], &tls_probe[28..]);
    }

    #[test]
    fn frame_rpc_message_prepends_record_mark() {
        let body = build_null_call(0x42);
        let framed = frame_rpc_message(&body);
        assert_eq!(framed.len(), 4 + body.len());
        let rm = u32::from_be_bytes(framed[0..4].try_into().unwrap());
        assert_eq!(rm, 0x80000000 | (body.len() as u32));
        assert_eq!(&framed[4..], &body);
    }

    #[test]
    fn parse_auth_tls_reply_detects_starttls() {
        let mut reply = Vec::new();
        reply.extend_from_slice(&(0x80000000u32 | 32).to_be_bytes()); // RM
        reply.extend_from_slice(&0xAAu32.to_be_bytes()); // XID
        reply.extend_from_slice(&1u32.to_be_bytes()); // REPLY
        reply.extend_from_slice(&0u32.to_be_bytes()); // MSG_ACCEPTED
        reply.extend_from_slice(&0u32.to_be_bytes()); // verf flavor AUTH_NONE
        reply.extend_from_slice(&8u32.to_be_bytes()); // verf body len
        reply.extend_from_slice(b"STARTTLS"); // verf body
        reply.extend_from_slice(&0u32.to_be_bytes()); // accept_stat SUCCESS
        assert!(parse_auth_tls_reply(&reply));
    }

    #[test]
    fn parse_auth_tls_reply_rejects_no_tls_support() {
        // Normal AUTH_NONE NULL reply  --  empty verifier
        let mut reply = Vec::new();
        reply.extend_from_slice(&(0x80000000u32 | 24).to_be_bytes());
        reply.extend_from_slice(&0xBBu32.to_be_bytes());
        reply.extend_from_slice(&1u32.to_be_bytes());
        reply.extend_from_slice(&0u32.to_be_bytes());
        reply.extend_from_slice(&0u32.to_be_bytes()); // verf flavor
        reply.extend_from_slice(&0u32.to_be_bytes()); // verf body len = 0
        reply.extend_from_slice(&0u32.to_be_bytes()); // accept_stat
        assert!(!parse_auth_tls_reply(&reply));
    }

    #[test]
    fn parse_auth_tls_reply_rejects_truncated_input() {
        assert!(!parse_auth_tls_reply(&[]));
        assert!(!parse_auth_tls_reply(&[0u8; 16]));
        assert!(!parse_auth_tls_reply(&[0u8; 31]));
    }

    #[test]
    fn test_read_op_encode() {
        let stateid = StateId { seqid: 0, other: [0u8; 12] };
        let op = Nfs4Op::Read { stateid, offset: 0, count: 8192 };
        let mut enc = XdrEncoder::new(64);
        encode_op(&mut enc, &op);
        let bytes = enc.into_bytes();
        // OP_READ (4) + seqid (4) + other (12) + offset (8) + count (4) = 32 bytes
        assert_eq!(bytes.len(), 32, "READ op XDR should be 32 bytes");
        // First 4 bytes = OP_READ = 25 in big-endian
        assert_eq!(&bytes[0..4], &25u32.to_be_bytes(), "First 4 bytes should be OP_READ (25)");
        // Offset field (bytes 20..28) = 0
        assert_eq!(&bytes[20..28], &0u64.to_be_bytes(), "Offset should be 0");
        // Count field (last 4 bytes) = 8192
        assert_eq!(&bytes[28..32], &8192u32.to_be_bytes(), "Last 4 bytes should be count (8192)");
    }

    #[test]
    fn test_read_op_result_data_field() {
        let result = OpResult {
            op: OP_READ,
            status: 0,
            stateid: None,
            filehandle: None,
            size: None,
            mtime: None,
            data: Some(vec![1u8, 2, 3, 4]),
        };
        assert_eq!(result.data, Some(vec![1u8, 2, 3, 4]));
        assert_eq!(result.op, OP_READ);
        assert_eq!(result.status, 0);
    }
}