sandfs 0.9.0

use std::io::{Cursor, Read, Write};

use anyhow::anyhow;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader};
use tracing::{debug, error, trace, warn};

use crate::nfsserve::context::{RPCContext, Verdict};
use crate::nfsserve::rpc::*;
use crate::nfsserve::xdr::*;
use crate::nfsserve::{mount, mount_handlers, nfs, nfs_handlers, portmap, portmap_handlers};

// Information from RFC 5531
// https://datatracker.ietf.org/doc/html/rfc5531

const NFS_ACL_PROGRAM: u32 = 100227;
const NFS_ID_MAP_PROGRAM: u32 = 100270;
const NFS_METADATA_PROGRAM: u32 = 200024;

/// The largest message read: a write of 1 MiB and its header, with room to spare.
const MAX_MESSAGE: usize = 2 << 20;

async fn handle_rpc(
    input: &mut impl Read,
    output: &mut impl Write,
    mut context: RPCContext,
) -> Result<bool, anyhow::Error> {
    let mut recv = rpc_msg::default();
    recv.deserialize(input)?;
    let xid = recv.xid;
    if let rpc_body::CALL(call) = recv.body {
        if let auth_flavor::AUTH_UNIX = call.cred.flavor {
            let mut auth = auth_unix::default();
            auth.deserialize(&mut Cursor::new(&call.cred.body))?;
            context.auth = auth;
        }
        if call.rpcvers != 2 {
            warn!("Invalid RPC version {} != 2", call.rpcvers);
            rpc_vers_mismatch(xid).serialize(output)?;
            return Ok(true);
        }

        if context.transaction_tracker.is_retransmission(xid, &context.client_addr) {
            // This is a retransmission
            // Drop the message and return
            debug!("Retransmission detected, xid: {}, client_addr: {}, call: {:?}", xid, context.client_addr, call);
            return Ok(false);
        }

        let res = {
            if call.prog == nfs::PROGRAM {
                nfs_handlers::handle_nfs(xid, call, input, output, &context).await
            } else if call.prog == portmap::PROGRAM {
                portmap_handlers::handle_portmap(xid, call, input, output, &context)
            } else if call.prog == mount::PROGRAM {
                mount_handlers::handle_mount(xid, call, input, output, &context).await
            } else if call.prog == NFS_ACL_PROGRAM
                || call.prog == NFS_ID_MAP_PROGRAM
                || call.prog == NFS_METADATA_PROGRAM
            {
                trace!("ignoring NFS_ACL packet");
                prog_unavail_reply_message(xid).serialize(output)?;
                Ok(())
            } else {
                warn!("Unknown RPC Program number {} != {}", call.prog, nfs::PROGRAM);
                prog_unavail_reply_message(xid).serialize(output)?;
                Ok(())
            }
        }
        .map(|_| true);
        context.transaction_tracker.mark_processed(xid, &context.client_addr);
        res
    } else {
        error!("Unexpectedly received a Reply instead of a Call");
        Err(anyhow!("Bad RPC Call format"))
    }
}

/// RFC 1057 Section 10
/// When RPC messages are passed on top of a byte stream transport
/// protocol (like TCP), it is necessary to delimit one message from
/// another in order to detect and possibly recover from protocol errors.
/// This is called record marking (RM).  Sun uses this RM/TCP/IP
/// transport for passing RPC messages on TCP streams.  One RPC message
/// fits into one RM record.
///
/// A record is composed of one or more record fragments.  A record
/// fragment is a four-byte header followed by 0 to (2**31) - 1 bytes of
/// fragment data.  The bytes encode an unsigned binary number; as with
/// XDR integers, the byte order is from highest to lowest.  The number
/// encodes two values -- a boolean which indicates whether the fragment
/// is the last fragment of the record (bit value 1 implies the fragment
/// is the last fragment) and a 31-bit unsigned binary value which is the
/// length in bytes of the fragment's data.  The boolean value is the
/// highest-order bit of the header; the length is the 31 low-order bits.
/// (Note that this record specification is NOT in XDR standard form!)
async fn read_fragment(socket: &mut (impl AsyncRead + Unpin), append_to: &mut Vec<u8>) -> Result<bool, anyhow::Error> {
    let mut header_buf = [0_u8; 4];
    socket.read_exact(&mut header_buf).await?;
    let fragment_header = u32::from_be_bytes(header_buf);
    let is_last = (fragment_header & (1 << 31)) > 0;
    let length = (fragment_header & ((1 << 31) - 1)) as usize;
    trace!("Reading fragment length:{}, last:{}", length, is_last);
    let start_offset = append_to.len();
    if start_offset + length > MAX_MESSAGE {
        return Err(anyhow!("RPC message too large"));
    }
    append_to.resize(append_to.len() + length, 0);
    socket.read_exact(&mut append_to[start_offset..]).await?;
    trace!("Finishing Reading fragment length:{}, last:{}", length, is_last);
    Ok(is_last)
}

pub async fn write_fragment(socket: &mut (impl AsyncWrite + Unpin), buf: &[u8]) -> Result<(), anyhow::Error> {
    // TODO: split into many fragments
    assert!(buf.len() < (1 << 31));
    // set the last flag
    let fragment_header = buf.len() as u32 + (1 << 31);
    let header_buf = u32::to_be_bytes(fragment_header);
    trace!("Writing fragment length:{}", buf.len());
    // One write, so the header doesn't go out in a packet of its own.
    socket.write_all(&[&header_buf[..], buf].concat()).await?;
    Ok(())
}

/// Answer the calls that come over `socket`, one after another, until it closes or a
/// call can't be read. Each call is answered before the next is read: the file system
/// answers each without waiting on anything, so answering them in tasks of their own
/// would only add a hand-off per call.
pub async fn serve_connection(socket: tokio::net::TcpStream, context: RPCContext) -> Result<(), anyhow::Error> {
    let (read, mut write) = socket.into_split();
    // Calls are small but for writes, which a read this size passes straight through.
    let mut read = BufReader::with_capacity(64 << 10, read);
    let mut call = Vec::new();
    loop {
        call.clear();
        while !read_fragment(&mut read, &mut call).await? {}
        let correct = match context.filter.as_ref().map(|filter| filter.check(&call)) {
            Some(Verdict::Reply(reply)) => {
                write_fragment(&mut write, &reply).await?;
                continue;
            }
            Some(Verdict::Correct(correct)) => Some(correct),
            Some(Verdict::Serve) | None => None,
        };
        let mut reply = Vec::new();
        match handle_rpc(&mut Cursor::new(&call), &mut reply, context.clone()).await {
            Err(e) => {
                error!("RPC Error: {:?}", e);
                return Err(e);
            }
            Ok(true) => {
                if let Some(correct) = correct {
                    correct(&mut reply);
                }
                write_fragment(&mut write, &reply).await?;
            }
            // A retransmission of a call already answered.
            Ok(false) => {}
        }
    }
}