OpenPXE/crates/iso-store/src/nfs_share.rs

//! NFSv3 share consumer — in-process userspace client.
//!
//! v0.4.67 brings NFS back, this time *the right way*: a pure-Rust
//! NFSv3 client (`nfs3_client` from the xetdata family of crates)
//! that speaks the NFS protocol entirely in userspace over TCP. No
//! `mount.nfs`, no kernel modules, no `CAP_SYS_ADMIN`. Works in every
//! container the SMB path (v0.4.65) works in, including Unraid and
//! restricted-SCC OpenShift.
//!
//! ## How it differs from the v0.4.64 NFS path
//!
//! v0.4.64 shelled out to `mount(8)` and asked the kernel to attach
//! the remote share to the local filesystem. That required nfs client
//! modules on the **host** kernel. Containers that ran on hosts
//! without those modules — Unraid being the dominant case — failed
//! with `mount.nfs: failed to apply fstab options` no matter what
//! capabilities they were granted.
//!
//! v0.4.67 doesn't touch the kernel. The client opens a TCP socket to
//! the NFS server, performs the MOUNT3 RPC to get a root file handle,
//! and uses NFSv3 READDIR3 / LOOKUP3 / READ3 / GETATTR3 to walk the
//! share and stream files. The kernel sees plain TCP traffic, nothing
//! else.
//!
//! ## How it differs from the v0.4.65 SMB path
//!
//! SMB ships in v0.4.65 via the `smbclient` CLI subprocess. NFS ships
//! here via a Rust crate — no subprocess, no PATH lookup, no parsing
//! of human-formatted output. That also means:
//!
//!   - **Range requests work** for NFS-sourced ISOs. NFSv3 READ3
//!     takes an explicit offset, so the HTTP handler can seek into
//!     the middle of a 5 GB ISO without downloading what comes
//!     before. SMB-sourced ISOs still return 416 for ranges because
//!     `smbclient -c 'get file -'` is a sequential stream.
//!   - **Introspection could work** (we could LOOKUP the ISO9660 PVD
//!     at offset 0x8000 and parse the volume label). For v0.4.67 we
//!     don't yet — same as SMB, NFS-sourced ISOs register with an
//!     `Unknown` family and fall back to generic sanboot. A follow-up
//!     can add bounded read-based detection.
//!
//! ## How it's the same as SMB
//!
//! The public surface is deliberately parallel: `NfsShareManager`
//! mirrors `SmbShareManager`, `NfsShare` mirrors `SmbShare`, the API
//! responses use the same `{error, stderr, hint}` shape so the UI's
//! storage tab renders both protocols through one code path. The
//! state file (`<work_dir>/nfs_shares.json`) sits next to
//! `smb_shares.json`.
//!
//! ## No auth
//!
//! NFSv3 uses AUTH_SYS by default — the client tells the server "I'm
//! uid X, gid Y" and the server decides whether to trust that. Most
//! NAS appliances exposing ISO libraries via NFS gate access by
//! **client IP address** rather than uid, so there's nothing for the
//! UI to ask for. (If a future server needs Kerberos or non-default
//! uid mapping we can add those, but for ISO read access nobody does.)

use crate::introspect::{introspect_reader, provisional_report};
use crate::iso_fs::{self, FileLocation, IsoReadAt};
use crate::remote_cache::RemoteIntrospectCache;
use crate::store::{generate_boot_entries_for, slugify_str, IsoSource, IsoStore};
use bytes::Bytes;
use nfs3_client::tokio::TokioConnector;
use nfs3_client::Nfs3ConnectionBuilder;
use nfs3_types::nfs3::{
    self as nfs3, diropargs3, entry3, filename3, nfs_fh3, GETATTR3args, LOOKUP3args, Nfs3Result,
    READ3args, READDIR3args,
};
use nfs3_types::rpc::{auth_unix, opaque_auth};
use nfs3_types::xdr_codec::Opaque;
use openpxe_core::{Error, Result};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::borrow::Cow;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use time::OffsetDateTime;

/// Default TCP port for the NFSv3 protocol. The classic split (111
/// portmapper + dynamic mountd) is supported by the underlying crate;
/// for direct connection-mode NAS appliances 2049 is what answers.
const DEFAULT_NFS_PORT: u16 = 2049;

/// How long we wait for the initial MOUNT3 + portmap handshake before
/// giving up. Mirrors the SMB pre-flight probe shape so the UI's
/// timeout banner reads consistently across protocols.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);

/// NFSv3 READ3 chunk size. The protocol's hard cap is 1 MiB per
/// reply; 64 KiB is a polite default that nearly every server hands
/// back without fragmentation and keeps in-flight memory bounded
/// during a streaming response.
const READ_CHUNK_BYTES: u32 = 64 * 1024;

/// Bound on the in-flight queue between the read-loop task and the
/// HTTP body stream. 16 * 64 KiB ≈ 1 MiB max buffer per stream —
/// enough to keep the network pipe full without letting a slow
/// client park gigabytes of decoded ISO in RAM.
const STREAM_BUFFER_DEPTH: usize = 16;

/// v0.7.4: per-ISO budget for a background introspection probe. A probe
/// is one connection plus a few dozen KiB-sized reads — sub-second on a
/// LAN — so anything past this is a wedged server, not a slow one.
const INTROSPECT_TIMEOUT: Duration = Duration::from_secs(30);

/// One queued background-introspection unit (v0.7.4).
struct ProbeJob {
    iso_id: String,
    filename: String,
    size: u64,
}

/// One configured NFS share. The id is derived from server+export so
/// re-adding the same coordinates is idempotent.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NfsShare {
    pub id: String,
    pub server: String,
    /// Export path on the server (e.g. "/srv/isos"). Must start with
    /// "/" to match the NFS server's view; we validate on add.
    pub export: String,
    /// TCP port — 2049 unless the operator overrode it.
    #[serde(default = "default_port")]
    pub port: u16,
    /// Most recent error talking to the share, or `None` on success.
    pub last_error: Option<String>,
    /// Operator-friendly translation of `last_error`. None when we
    /// don't have a friendlier rendition.
    pub last_hint: Option<String>,
    #[serde(with = "time::serde::rfc3339::option")]
    pub last_scan: Option<OffsetDateTime>,
    pub iso_count: u32,
    /// True after a successful scan, false on failure. Drives the
    /// UI dot.
    pub reachable: bool,
}

/// Submission from the UI / API.
#[derive(Debug, Clone, Deserialize)]
pub struct NfsAddRequest {
    pub server: String,
    pub export: String,
    #[serde(default)]
    pub port: Option<u16>,
}

fn default_port() -> u16 {
    DEFAULT_NFS_PORT
}

/// Structured error surfaced to the API and rendered in the UI. Same
/// shape as `SmbShareError` so the storage tab uses one rendering
/// path for both protocols.
#[derive(Debug, Clone, Serialize)]
pub struct NfsShareError {
    pub error: String,
    pub stderr: String,
    pub hint: Option<String>,
}

impl NfsShareError {
    fn from_raw(error: impl Into<String>, stderr: impl Into<String>) -> Self {
        let stderr = stderr.into();
        let error = error.into();
        let hint = hint_for(&stderr).or_else(|| hint_for(&error));
        Self {
            error,
            stderr,
            hint,
        }
    }
}

#[derive(Debug, Default)]
struct Inner {
    shares: HashMap<String, NfsShare>,
}

/// Manages NFS shares. Cheap to clone — internal state is
/// `Arc<Mutex<...>>`.
#[derive(Debug, Clone)]
pub struct NfsShareManager {
    state_path: Arc<PathBuf>,
    inner: Arc<Mutex<Inner>>,
    iso_store: IsoStore,
    /// Serializes scan operations on the same manager. Each scan
    /// opens its own NFS connection so concurrency isn't a hard
    /// requirement, but serializing keeps log output predictable.
    op_lock: Arc<tokio::sync::Mutex<()>>,
    /// v0.7.4: persisted introspection results keyed `share/path@size`,
    /// so a restart re-probes only new or replaced ISOs.
    introspect_cache: RemoteIntrospectCache,
}

impl NfsShareManager {
    /// Construct a manager. State persists to
    /// `<work_dir>/nfs_shares.json`.
    #[must_use]
    pub fn new(work_dir: &Path, iso_store: IsoStore) -> Self {
        let state_path = work_dir.join("nfs_shares.json");
        Self {
            state_path: Arc::new(state_path),
            inner: Arc::new(Mutex::new(Inner::default())),
            iso_store,
            op_lock: Arc::new(tokio::sync::Mutex::new(())),
            introspect_cache: RemoteIntrospectCache::open(work_dir, "nfs_introspect_cache.json"),
        }
    }

    /// Load persisted state and re-scan every share. Per-share
    /// failures are logged but never propagated — startup must not
    /// block on a single offline server.
    pub async fn load_and_rescan(&self) -> Result<()> {
        let shares = match tokio::fs::read_to_string(self.state_path.as_path()).await {
            Ok(text) => serde_json::from_str::<Vec<NfsShare>>(&text).unwrap_or_default(),
            Err(_) => Vec::new(),
        };
        for mut s in shares {
            s.last_error = None;
            s.last_hint = None;
            s.reachable = false;
            self.inner.lock().shares.insert(s.id.clone(), s.clone());
            if let Err(e) = self.rescan_inner(&s.id).await {
                tracing::warn!(
                    target: "openpxe::nfs",
                    id = %s.id, server = %s.server, export = %s.export,
                    "rescan on startup failed: {e}"
                );
            }
        }
        Ok(())
    }

    /// Register an NFS share. Validates, probes connectivity by
    /// performing a real MOUNT3 + READDIR3, and registers the
    /// resulting ISOs with the store.
    pub async fn add(&self, req: NfsAddRequest) -> std::result::Result<NfsShare, NfsShareError> {
        let server = normalize_server(&req.server);
        let export = req.export.trim().to_string();
        if server.is_empty() {
            return Err(NfsShareError::from_raw("server is required", ""));
        }
        if !export.starts_with('/') {
            return Err(NfsShareError::from_raw(
                "export path must start with '/' (e.g. /srv/isos)",
                "",
            ));
        }
        if server.contains('\0') || export.contains('\0') {
            return Err(NfsShareError::from_raw("NUL bytes are not allowed", ""));
        }
        let port = req.port.filter(|p| *p != 0).unwrap_or(DEFAULT_NFS_PORT);

        let id = share_id(&server, &export);
        let spec = NfsShare {
            id: id.clone(),
            server,
            export,
            port,
            last_error: None,
            last_hint: None,
            last_scan: None,
            iso_count: 0,
            reachable: false,
        };
        self.inner.lock().shares.insert(id.clone(), spec);
        self.persist_locked();

        if let Err(e) = self.rescan_inner(&id).await {
            let m = self.get(&id);
            return Err(NfsShareError {
                error: m
                    .as_ref()
                    .and_then(|m| m.last_error.clone())
                    .unwrap_or_else(|| e.to_string()),
                stderr: String::new(),
                hint: m.and_then(|m| m.last_hint),
            });
        }
        Ok(self.get(&id).expect("just inserted"))
    }

    /// Remove a share. Drops every ISO sourced from it. Idempotent.
    /// Async for symmetry with [`SmbShareManager::remove`] — the
    /// SMB version is async because it tears down a credentials
    /// file; NFS has nothing to clean up but we keep the signature
    /// uniform so the call sites in app.rs / terminal.rs match.
    #[allow(clippy::unused_async)]
    pub async fn remove(&self, id: &str) -> Result<()> {
        let removed = self.inner.lock().shares.remove(id).is_some();
        if removed {
            self.iso_store.drop_external_source(id);
            self.persist_locked();
        }
        Ok(())
    }

    /// Re-walk a share for new / removed ISOs.
    pub async fn rescan(&self, id: &str) -> Result<u32> {
        self.rescan_inner(id).await
    }

    #[must_use]
    pub fn list(&self) -> Vec<NfsShare> {
        let g = self.inner.lock();
        let mut v: Vec<_> = g.shares.values().cloned().collect();
        v.sort_by(|a, b| a.id.cmp(&b.id));
        v
    }

    #[must_use]
    pub fn get(&self, id: &str) -> Option<NfsShare> {
        self.inner.lock().shares.get(id).cloned()
    }

    /// Open a byte stream reading `filename` out of share `share_id`,
    /// starting at `start_offset` and reading at most `max_len`
    /// bytes. The returned stream yields `Bytes` chunks (≤
    /// `READ_CHUNK_BYTES`) and ends at EOF, after `max_len` bytes, or
    /// on the first transport error.
    ///
    /// Used by the HTTP ISO download handler. Supports HTTP Range
    /// requests because NFSv3 READ3 takes an explicit offset — this
    /// is the protocol-level advantage NFS has over the SMB
    /// userspace path.
    ///
    /// Kept `async` for symmetry with [`SmbShareManager::stream_iso`]
    /// even though the body doesn't await today — a future
    /// refinement (e.g. throttling, connection pooling) will need to
    /// await without changing call sites.
    #[allow(clippy::unused_async)]
    pub async fn stream_iso(
        &self,
        share_id: &str,
        filename: &str,
        start_offset: u64,
        max_len: Option<u64>,
    ) -> Result<NfsStream> {
        let share = self
            .get(share_id)
            .ok_or_else(|| Error::Invalid(format!("no such NFS share '{share_id}'")))?;
        // Defensive: NFSv3 LOOKUP3 takes a single name relative to
        // the export root, not a path. We don't support nested
        // directories in v0.4.67 — ISOs live at the top of the share.
        if filename.contains('/') || filename.contains('\\') || filename.contains("..") {
            return Err(Error::Invalid(format!("invalid filename '{filename}'")));
        }

        let (tx, rx) = tokio::sync::mpsc::channel::<std::io::Result<Bytes>>(STREAM_BUFFER_DEPTH);
        let server = share.server.clone();
        let export = share.export.clone();
        let port = share.port;
        let fname = filename.to_string();

        // Spawn a task that owns the NFS connection. Owning the
        // connection inside the spawn means we don't have to worry
        // about borrowing across awaits or sharing the connection
        // between scan and stream — each stream gets its own.
        let task = tokio::spawn(async move {
            let result = stream_loop(
                &server,
                &export,
                port,
                &fname,
                start_offset,
                max_len,
                tx.clone(),
            )
            .await;
            if let Err(e) = result {
                // Best-effort signal of the error to the consumer.
                // If the receiver has already dropped we just exit.
                let _ = tx.send(Err(std::io::Error::other(e.to_string()))).await;
            }
        });

        Ok(NfsStream { rx, _task: task })
    }

    // ── internals ─────────────────────────────────────────────────────

    async fn rescan_inner(&self, id: &str) -> Result<u32> {
        let _g = self.op_lock.lock().await;
        let share = self
            .get(id)
            .ok_or_else(|| Error::Invalid(format!("no such share '{id}'")))?;
        let now = OffsetDateTime::now_utc();

        // Drop prior entries so a deleted file disappears from the
        // store on the next scan.
        self.iso_store.drop_external_source(id);

        let listing = match list_isos(&share.server, &share.export, share.port).await {
            Ok(l) => l,
            Err(err) => {
                let stderr = err.to_string();
                let hint = hint_for(&stderr);
                self.update_status(id, 0, false, Some(stderr.clone()), hint, now);
                return Err(Error::Invalid(stderr));
            }
        };

        let mut count = 0u32;
        let mut to_probe: Vec<ProbeJob> = Vec::new();
        for entry in listing {
            let iso_id = format!("nfs-{}-{}", share.id, slugify_str(&entry.filename));
            // v0.7.4: real introspection over the share — NFSv3 READ3
            // takes an offset, so the ISO9660 probes work remotely. A
            // cache hit registers the full report immediately; a miss
            // registers a provisional filename-based report (so the scan
            // returns fast) and queues a background probe that upgrades
            // the entry in place.
            let cached = self
                .introspect_cache
                .get(&share.id, &entry.filename, entry.size);
            let report = cached.unwrap_or_else(|| {
                to_probe.push(ProbeJob {
                    iso_id: iso_id.clone(),
                    filename: entry.filename.clone(),
                    size: entry.size,
                });
                provisional_report(&entry.filename)
            });
            let boot_entries = generate_boot_entries_for(&iso_id, &entry.filename, &report);
            let source = IsoSource::Nfs {
                share_id: share.id.clone(),
                relative_path: entry.filename.clone(),
            };
            self.iso_store.register_external(
                iso_id,
                entry.filename,
                entry.size,
                report,
                boot_entries,
                source,
            );
            count += 1;
        }
        self.update_status(id, count, true, None, None, now);
        tracing::info!(
            target: "openpxe::nfs",
            id = %id, server = %share.server, export = %share.export,
            iso_count = count,
            pending_introspection = to_probe.len(),
            "NFS share scanned"
        );
        if !to_probe.is_empty() {
            self.spawn_introspection_pass(&share, to_probe);
        }
        Ok(count)
    }

    /// v0.7.4: probe each queued ISO over its own NFS connection and swap
    /// the full introspection into the store as results land. Runs
    /// detached so neither startup nor the share-add API call waits on a
    /// 40-ISO library; per-ISO failures (or a share removed mid-pass)
    /// leave the provisional entry in place, which still sanboots.
    fn spawn_introspection_pass(&self, share: &NfsShare, work: Vec<ProbeJob>) {
        let store = self.iso_store.clone();
        let cache = self.introspect_cache.clone();
        let share_id = share.id.clone();
        let server = share.server.clone();
        let export = share.export.clone();
        let port = share.port;
        let queued = work.len();
        tokio::spawn(async move {
            let mut upgraded = 0usize;
            for job in work {
                let probe = async {
                    let mut reader = NfsReadAt::open(&server, &export, port, &job.filename).await?;
                    let report =
                        introspect_reader(&mut reader, job.size, &job.filename, false).await;
                    reader.finish().await;
                    Ok::<_, NfsClientError>(report)
                };
                match tokio::time::timeout(INTROSPECT_TIMEOUT, probe).await {
                    Ok(Ok(report)) => {
                        cache.put(&share_id, &job.filename, job.size, report.clone());
                        if store.update_external_introspection(&job.iso_id, report) {
                            upgraded += 1;
                        }
                    }
                    Ok(Err(e)) => tracing::warn!(
                        target: "openpxe::nfs",
                        share = %share_id, iso = %job.filename,
                        "introspection failed: {e}"
                    ),
                    Err(_) => tracing::warn!(
                        target: "openpxe::nfs",
                        share = %share_id, iso = %job.filename,
                        "introspection timed out after {}s", INTROSPECT_TIMEOUT.as_secs()
                    ),
                }
            }
            tracing::info!(
                target: "openpxe::nfs",
                share = %share_id, queued, upgraded,
                "remote introspection pass complete"
            );
        });
    }

    /// v0.7.4: locate `in_iso_path` inside a share-hosted ISO. Returns the
    /// byte range so the HTTP layer can serve kernel/initrd files out of
    /// remote ISOs with a follow-up ranged [`Self::stream_iso`].
    pub async fn locate_in_iso(
        &self,
        share_id: &str,
        filename: &str,
        in_iso_path: &str,
    ) -> Result<Option<FileLocation>> {
        let share = self
            .get(share_id)
            .ok_or_else(|| Error::Invalid(format!("no such NFS share '{share_id}'")))?;
        if filename.contains('/') || filename.contains('\\') || filename.contains("..") {
            return Err(Error::Invalid(format!("invalid filename '{filename}'")));
        }
        let mut reader = NfsReadAt::open(&share.server, &share.export, share.port, filename)
            .await
            .map_err(|e| Error::Invalid(format!("nfs open '{filename}': {e}")))?;
        let loc = iso_fs::lookup(&mut reader, in_iso_path).await;
        reader.finish().await;
        Ok(loc)
    }

    fn update_status(
        &self,
        id: &str,
        iso_count: u32,
        reachable: bool,
        err: Option<String>,
        hint: Option<String>,
        ts: OffsetDateTime,
    ) {
        if let Some(s) = self.inner.lock().shares.get_mut(id) {
            s.iso_count = iso_count;
            s.reachable = reachable;
            s.last_error = err;
            s.last_hint = hint;
            s.last_scan = Some(ts);
        }
        self.persist_locked();
    }

    fn persist_locked(&self) {
        let shares: Vec<NfsShare> = self.inner.lock().shares.values().cloned().collect();
        let path = self.state_path.as_path();
        let tmp = path.with_extension("json.tmp");
        let body = match serde_json::to_vec_pretty(&shares) {
            Ok(b) => b,
            Err(e) => {
                tracing::warn!(target: "openpxe::nfs", "serialize: {e}");
                return;
            }
        };
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        if let Err(e) = std::fs::write(&tmp, body) {
            tracing::warn!(target: "openpxe::nfs", "write tmp: {e}");
            return;
        }
        if let Err(e) = std::fs::rename(&tmp, path) {
            tracing::warn!(target: "openpxe::nfs", "rename: {e}");
        }
    }
}

/// HTTP body stream for an NFS-sourced ISO read.
///
/// Implements `Stream<Item = io::Result<Bytes>>` so axum can convert
/// it into a response body via `Body::from_stream`.
#[derive(Debug)]
pub struct NfsStream {
    rx: tokio::sync::mpsc::Receiver<std::io::Result<Bytes>>,
    /// Kept alive so the read-loop task isn't dropped while the HTTP
    /// client is still consuming bytes. Dropping the stream cancels
    /// the task, which is the right behavior on client disconnect.
    _task: tokio::task::JoinHandle<()>,
}

impl futures::Stream for NfsStream {
    type Item = std::io::Result<Bytes>;
    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        self.rx.poll_recv(cx)
    }
}

#[derive(Debug, Clone)]
struct NfsListEntry {
    filename: String,
    size: u64,
}

/// The connection type [`build_connection`] yields.
type NfsConn = nfs3_client::Nfs3Connection<nfs3_client::tokio::TokioIo<tokio::net::TcpStream>>;

/// v0.7.4: random-access reader over one NFS connection + file handle —
/// the [`IsoReadAt`] impl that lets the ISO9660 walker and introspection
/// probes run against share-hosted images.
struct NfsReadAt {
    conn: NfsConn,
    fh: nfs_fh3,
}

impl NfsReadAt {
    /// Connect, mount, and LOOKUP `filename` at the export root.
    async fn open(
        server: &str,
        export: &str,
        port: u16,
        filename: &str,
    ) -> std::result::Result<Self, NfsClientError> {
        let mut conn = build_connection(server, export, port).await?;
        let root = conn.root_nfs_fh3();
        let lookup = conn
            .lookup(&LOOKUP3args {
                what: diropargs3 {
                    dir: root,
                    name: filename3(Opaque::borrowed(filename.as_bytes())),
                },
            })
            .await
            .map_err(NfsClientError::Rpc)?;
        let fh = match lookup {
            Nfs3Result::Ok(o) => o.object,
            Nfs3Result::Err((status, _)) => {
                return Err(NfsClientError::Nfsstat(status_label(status)));
            }
        };
        Ok(Self { conn, fh })
    }

    /// Best-effort unmount. Consumes the reader — it's done.
    async fn finish(self) {
        let _ = self.conn.unmount().await;
    }
}

impl IsoReadAt for NfsReadAt {
    async fn read_at(&mut self, offset: u64, len: u32) -> std::io::Result<Vec<u8>> {
        let mut out: Vec<u8> = Vec::with_capacity(len as usize);
        let mut off = offset;
        // READ3 may legally return fewer bytes than asked (server cap);
        // loop until the exact-read contract is satisfied or the file
        // genuinely ends short.
        while (out.len() as u32) < len {
            let want = (len - out.len() as u32).min(READ_CHUNK_BYTES);
            let res = self
                .conn
                .read(&READ3args {
                    file: self.fh.clone(),
                    offset: off,
                    count: want,
                })
                .await
                .map_err(|e| std::io::Error::other(e.to_string()))?;
            let ok = match res {
                Nfs3Result::Ok(o) => o,
                Nfs3Result::Err((status, _)) => {
                    return Err(std::io::Error::other(status_label(status)));
                }
            };
            let data = ok.data.as_ref();
            if data.is_empty() {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::UnexpectedEof,
                    "NFS read past end of file",
                ));
            }
            out.extend_from_slice(data);
            off += data.len() as u64;
            if ok.eof && (out.len() as u32) < len {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::UnexpectedEof,
                    "NFS read past end of file",
                ));
            }
        }
        out.truncate(len as usize);
        Ok(out)
    }
}

/// Connect, READDIR the export root, look up each `*.iso` to get its
/// size + file handle. Returns a flat list. Errors are returned with
/// a human-readable message; the caller decides how to surface them.
async fn list_isos(
    server: &str,
    export: &str,
    port: u16,
) -> std::result::Result<Vec<NfsListEntry>, NfsClientError> {
    // build_connection applies its own per-attempt timeout (privileged
    // source port first, then a non-privileged fallback).
    let mut conn = build_connection(server, export, port).await?;

    let root = conn.root_nfs_fh3();
    let mut entries = Vec::new();
    let mut cookie: u64 = 0;
    let mut cookieverf = nfs3::cookieverf3::default();
    loop {
        let res = conn
            .readdir(&READDIR3args {
                dir: root.clone(),
                cookie,
                cookieverf,
                count: 32 * 1024,
            })
            .await
            .map_err(NfsClientError::Rpc)?;
        let ok = match res {
            Nfs3Result::Ok(ok) => ok,
            Nfs3Result::Err((status, _)) => {
                return Err(NfsClientError::Nfsstat(status_label(status)));
            }
        };
        cookieverf = ok.cookieverf;
        let eof = ok.reply.eof;
        let dir_entries: Vec<entry3<'_>> = ok.reply.entries.0;
        let next_cookie = dir_entries.last().map(|e| e.cookie);
        for entry in dir_entries {
            // entry.name is `filename3<'a>(Opaque<'a>)` — XDR opaque
            // bytes. Decode to UTF-8 best-effort.
            let name_bytes = entry.name.0.as_ref();
            let Ok(name) = std::str::from_utf8(name_bytes) else {
                continue;
            };
            if name == "." || name == ".." {
                continue;
            }
            if !name.to_ascii_lowercase().ends_with(".iso") {
                continue;
            }
            // Look up the file to get its size + verify it's a
            // regular file (not a symlink / directory matching the
            // .iso pattern).
            let lookup = conn
                .lookup(&LOOKUP3args {
                    what: diropargs3 {
                        dir: root.clone(),
                        name: filename3(Opaque::borrowed(name_bytes)),
                    },
                })
                .await
                .map_err(NfsClientError::Rpc)?;
            let lookup_ok = match lookup {
                Nfs3Result::Ok(o) => o,
                Nfs3Result::Err(_) => continue,
            };
            let getattr = conn
                .getattr(&GETATTR3args {
                    object: lookup_ok.object,
                })
                .await
                .map_err(NfsClientError::Rpc)?;
            let attrs = match getattr {
                Nfs3Result::Ok(o) => o.obj_attributes,
                Nfs3Result::Err(_) => continue,
            };
            // ftype3::NF3REG == 1 (regular file). Skip everything
            // else — directories, symlinks, devices.
            if attrs.type_ as u32 != nfs3::ftype3::NF3REG as u32 {
                continue;
            }
            entries.push(NfsListEntry {
                filename: name.to_string(),
                size: attrs.size,
            });
        }
        if eof {
            break;
        }
        match next_cookie {
            Some(c) if c != 0 => cookie = c,
            _ => break,
        }
    }

    let _ = conn.unmount().await;
    Ok(entries)
}

async fn stream_loop(
    server: &str,
    export: &str,
    port: u16,
    filename: &str,
    start_offset: u64,
    max_len: Option<u64>,
    tx: tokio::sync::mpsc::Sender<std::io::Result<Bytes>>,
) -> std::result::Result<(), NfsClientError> {
    let mut conn = build_connection(server, export, port).await?;

    let root = conn.root_nfs_fh3();
    // Look up the file to get its handle.
    let name_bytes = filename.as_bytes();
    let lookup = conn
        .lookup(&LOOKUP3args {
            what: diropargs3 {
                dir: root,
                name: filename3(Opaque::borrowed(name_bytes)),
            },
        })
        .await
        .map_err(NfsClientError::Rpc)?;
    let lookup_ok = match lookup {
        Nfs3Result::Ok(o) => o,
        Nfs3Result::Err((status, _)) => {
            return Err(NfsClientError::Nfsstat(status_label(status)));
        }
    };
    let file_handle: nfs_fh3 = lookup_ok.object;

    let mut offset = start_offset;
    let mut remaining = max_len;
    loop {
        if let Some(r) = remaining {
            if r == 0 {
                break;
            }
        }
        // Cap the chunk at READ_CHUNK_BYTES and at the remaining budget.
        let chunk = match remaining {
            Some(r) if r < u64::from(READ_CHUNK_BYTES) => r as u32,
            _ => READ_CHUNK_BYTES,
        };
        let res = conn
            .read(&READ3args {
                file: file_handle.clone(),
                offset,
                count: chunk,
            })
            .await
            .map_err(NfsClientError::Rpc)?;
        let ok = match res {
            Nfs3Result::Ok(o) => o,
            Nfs3Result::Err((status, _)) => {
                return Err(NfsClientError::Nfsstat(status_label(status)));
            }
        };
        let bytes = Bytes::copy_from_slice(ok.data.as_ref());
        let bytes_len = bytes.len() as u64;
        if tx.send(Ok(bytes)).await.is_err() {
            // HTTP client dropped — abort gracefully.
            break;
        }
        offset += bytes_len;
        if let Some(r) = remaining.as_mut() {
            *r = r.saturating_sub(bytes_len);
        }
        if ok.eof {
            break;
        }
        // Defensive: a server that returns 0 bytes without EOF
        // would have us busy-looping. Bail out instead.
        if bytes_len == 0 {
            break;
        }
    }

    let _ = conn.unmount().await;
    Ok(())
}

/// Mount the export and return a live connection.
///
/// ## Privileged source port (the v0.4.68 fix)
///
/// Linux kernel `nfsd` — which is what UniFi UNAS, Synology, TrueNAS,
/// and essentially every appliance NAS runs underneath — exports with
/// the `secure` option **by default**. `secure` means the server only
/// accepts MOUNT3 / NFS3 requests whose TCP **source** port is in the
/// privileged range (< 1024). A client connecting from an ephemeral
/// high port gets `MNT3ERR_ACCES` at mount time — which is exactly the
/// error operators hit in v0.4.67 even with their host IP correctly in
/// the export's allow-list.
///
/// v0.4.67 disabled privileged source ports because the openpxe
/// process runs as uid 10001 and "can't bind sub-1024 ports". That
/// reasoning was wrong: the binary carries `CAP_NET_BIND_SERVICE`
/// (granted via `setcap` in the Dockerfile so it can bind the DHCP /
/// TFTP / HTTP low ports as non-root), and that capability also lets
/// it bind a privileged *source* port for an outbound connection.
///
/// So we now try a privileged source port first — the common case for
/// real NAS appliances — and fall back to a non-privileged port for
/// servers exported `insecure` (or environments where we genuinely
/// can't grab a low port). Each attempt gets its own connect timeout.
async fn build_connection(
    server: &str,
    export: &str,
    port: u16,
) -> std::result::Result<
    nfs3_client::Nfs3Connection<nfs3_client::tokio::TokioIo<tokio::net::TcpStream>>,
    NfsClientError,
> {
    match connect_once(server, export, port, true).await {
        Ok(conn) => Ok(conn),
        // A timeout means the server didn't answer at all — a
        // non-privileged retry would just time out again and double
        // the operator's wait. Surface the timeout immediately.
        Err(primary @ NfsClientError::Timeout(..)) => Err(primary),
        Err(primary) => match connect_once(server, export, port, false).await {
            Ok(conn) => Ok(conn),
            // Surface the privileged-attempt error: for the dominant
            // `secure`-export case it's the one whose hint points at
            // the real fix.
            Err(_) => Err(primary),
        },
    }
}

/// One mount attempt with a specific source-port policy, bounded by
/// [`CONNECT_TIMEOUT`].
async fn connect_once(
    server: &str,
    export: &str,
    port: u16,
    privileged: bool,
) -> std::result::Result<
    nfs3_client::Nfs3Connection<nfs3_client::tokio::TokioIo<tokio::net::TcpStream>>,
    NfsClientError,
> {
    // v0.4.69: present an AUTH_SYS (AUTH_UNIX) credential instead of
    // the crate default (AUTH_NONE). This is the fix for the
    // `NFS3ERR_ACCES` operators hit *after* the v0.4.68 privileged-port
    // fix got them past the mount: nearly every NFS server exports
    // `sec=sys` and rejects AUTH_NONE callers on the actual file ops
    // (READDIR/LOOKUP/READ) even when MOUNT succeeded. We send uid 0 /
    // gid 0 — a server with `no_root_squash` treats us as root (full
    // read), and the far more common `root_squash` maps us to the
    // anonymous user, which can still read any world-readable ISO
    // share (the normal case). We deliberately keep this fixed rather
    // than a UI knob: ISO libraries are read-only shared data, and a
    // uid field is exactly the kind of thing that makes a "dead simple"
    // tool confusing for an L1 tech.
    let cred = opaque_auth::auth_unix(&auth_unix {
        uid: 0,
        gid: 0,
        ..Default::default()
    });
    let fut = Nfs3ConnectionBuilder::new(TokioConnector, server, export)
        .connect_from_privileged_port(privileged)
        .nfs3_port(port)
        .credential(cred)
        .mount();
    match tokio::time::timeout(CONNECT_TIMEOUT, fut).await {
        Ok(Ok(conn)) => Ok(conn),
        Ok(Err(e)) => Err(NfsClientError::Connect(e.to_string())),
        Err(_) => Err(NfsClientError::Timeout(server.to_string(), port)),
    }
}

#[derive(Debug)]
enum NfsClientError {
    Connect(String),
    Timeout(String, u16),
    Rpc(nfs3_client::RpcError),
    Nfsstat(String),
}

impl std::fmt::Display for NfsClientError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Connect(msg) => write!(f, "connect failed: {msg}"),
            Self::Timeout(host, port) => write!(
                f,
                "connect timed out after {}s talking to {host}:{port}",
                CONNECT_TIMEOUT.as_secs()
            ),
            Self::Rpc(e) => write!(f, "RPC failure: {e}"),
            Self::Nfsstat(s) => write!(f, "NFS server returned {s}"),
        }
    }
}

impl std::error::Error for NfsClientError {}

/// Best-effort label for an `nfsstat3` so the UI shows a readable
/// name like `NFS3ERR_ACCES` instead of a magic number.
fn status_label(status: nfs3::nfsstat3) -> String {
    use nfs3::nfsstat3 as S;
    let name = match status {
        S::NFS3_OK => "NFS3_OK",
        S::NFS3ERR_PERM => "NFS3ERR_PERM",
        S::NFS3ERR_NOENT => "NFS3ERR_NOENT",
        S::NFS3ERR_IO => "NFS3ERR_IO",
        S::NFS3ERR_NXIO => "NFS3ERR_NXIO",
        S::NFS3ERR_ACCES => "NFS3ERR_ACCES",
        S::NFS3ERR_EXIST => "NFS3ERR_EXIST",
        S::NFS3ERR_XDEV => "NFS3ERR_XDEV",
        S::NFS3ERR_NODEV => "NFS3ERR_NODEV",
        S::NFS3ERR_NOTDIR => "NFS3ERR_NOTDIR",
        S::NFS3ERR_ISDIR => "NFS3ERR_ISDIR",
        S::NFS3ERR_INVAL => "NFS3ERR_INVAL",
        S::NFS3ERR_FBIG => "NFS3ERR_FBIG",
        S::NFS3ERR_NOSPC => "NFS3ERR_NOSPC",
        S::NFS3ERR_ROFS => "NFS3ERR_ROFS",
        S::NFS3ERR_MLINK => "NFS3ERR_MLINK",
        S::NFS3ERR_NAMETOOLONG => "NFS3ERR_NAMETOOLONG",
        S::NFS3ERR_NOTEMPTY => "NFS3ERR_NOTEMPTY",
        S::NFS3ERR_DQUOT => "NFS3ERR_DQUOT",
        S::NFS3ERR_STALE => "NFS3ERR_STALE",
        S::NFS3ERR_REMOTE => "NFS3ERR_REMOTE",
        S::NFS3ERR_BADHANDLE => "NFS3ERR_BADHANDLE",
        S::NFS3ERR_NOT_SYNC => "NFS3ERR_NOT_SYNC",
        S::NFS3ERR_BAD_COOKIE => "NFS3ERR_BAD_COOKIE",
        S::NFS3ERR_NOTSUPP => "NFS3ERR_NOTSUPP",
        S::NFS3ERR_TOOSMALL => "NFS3ERR_TOOSMALL",
        S::NFS3ERR_SERVERFAULT => "NFS3ERR_SERVERFAULT",
        S::NFS3ERR_BADTYPE => "NFS3ERR_BADTYPE",
        S::NFS3ERR_JUKEBOX => "NFS3ERR_JUKEBOX",
    };
    name.to_string()
}

/// Translate well-known NFS error strings into actionable hints for
/// the UI. Mirrors the SMB hint table in spirit; the patterns are
/// different because NFS errors travel via NFS3ERR_* codes plus
/// transport-level messages from the Rust crate.
fn hint_for(text: &str) -> Option<String> {
    let s = text.to_ascii_lowercase();
    if s.contains("mnt3err_acces") || s.contains("mount") && s.contains("acces") {
        // Mount-protocol access denial. Two common causes, in order of
        // likelihood for an appliance NAS: (1) the export requires a
        // privileged source port (`secure`, the Linux default) — we
        // already retry with one, so reaching here means even that was
        // refused; (2) the client IP isn't in the allow-list.
        Some(
            "the NFS server denied the mount (MNT3ERR_ACCES). Two things to \
             check on the server: (1) this OpenPXE host's IP is in the \
             export's allowed-clients list, and (2) if your export uses the \
             default `secure` option, OpenPXE already connects from a \
             privileged port — but if the server still refuses, add \
             `insecure` to the export. On UniFi UNAS, confirm the host IP is \
             listed under the share's NFS permissions and the export path is \
             /var/nfs/shared/<share> (not just /<share>)."
                .into(),
        )
    } else if s.contains("nfs3err_acces") || s.contains("permission denied") {
        // The mount succeeded but a file op was denied. OpenPXE
        // already presents an AUTH_SYS uid-0 credential, so this is a
        // server-side permission/squash issue, not an IP or auth-flavor
        // one. Point the operator at the share's filesystem permissions.
        Some(
            "the mount succeeded but the server denied reading the share \
             (NFS3ERR_ACCES). OpenPXE connects as AUTH_SYS uid 0, so this is \
             a server-side permission issue: make sure the export's \
             directory is readable (most ISO shares are world-readable / \
             0755), and that the export isn't restricted to a specific \
             non-root user via all_squash/anonuid. On UniFi UNAS, confirm \
             the share's NFS permission for this host is Read-Write (or \
             Read-Only) and that the share itself grants read access."
                .into(),
        )
    } else if s.contains("nfs3err_noent")
        || s.contains("nfs3err_notdir")
        || s.contains("mnt3err_noent")
    {
        Some(
            "the export path doesn't exist on the server, or it isn't a \
             directory. Double-check the path (e.g. /srv/isos vs /isos — \
             UniFi UNAS Pro exposes shares under /var/nfs/shared/<name>)."
                .into(),
        )
    } else if s.contains("nfs3err_stale") || s.contains("nfs3err_badhandle") {
        Some(
            "the server's view of the share changed under us. Re-scan; \
             if that doesn't help, remove and re-add the share."
                .into(),
        )
    } else if s.contains("connect timed out")
        || s.contains("timed out")
        || s.contains("connection timed out")
    {
        Some(
            "no answer from the server within the connect timeout. \
             Verify the IP, the port (default 2049), and any firewall \
             between OpenPXE and the NAS."
                .into(),
        )
    } else if s.contains("connection refused") {
        Some(
            "the NFS service isn't accepting connections on this port. \
             Make sure nfsd is running on the server and (for v3) the \
             portmapper on port 111 is reachable."
                .into(),
        )
    } else if s.contains("no route to host") || s.contains("network is unreachable") {
        Some(
            "the server isn't reachable on this network. Check the IP \
             and routes."
                .into(),
        )
    } else if s.contains("mount") && s.contains("denied") {
        Some(
            "MOUNT3 was denied. The classic cause is that the export's \
             `rw=<host>` / `ro=<host>` list doesn't include this client. \
             Some servers also require `insecure` in /etc/exports for \
             non-privileged source ports — which is what this client \
             uses (we run as uid 10001, no privileged-port capability)."
                .into(),
        )
    } else {
        None
    }
}

fn share_id(server: &str, export: &str) -> String {
    slugify_str(&format!("{server}{export}"))
}

/// Normalize a server input: trim, strip schemes, drop trailing
/// slashes. Matches the SMB normalizer so paste-from-anywhere works.
fn normalize_server(raw: &str) -> String {
    let s = raw.trim();
    let s = s
        .strip_prefix("nfs://")
        .or_else(|| s.strip_prefix("http://"))
        .or_else(|| s.strip_prefix("https://"))
        .unwrap_or(s);
    s.trim_end_matches('/').to_string()
}

// Silence the unused-import warning on `Cow` — we use it implicitly
// via `Opaque::borrowed` constructions in `list_isos`.
#[allow(dead_code)]
fn _unused() -> Cow<'static, str> {
    Cow::Borrowed("")
}

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

    #[test]
    fn share_id_is_stable_and_safe() {
        let a = share_id("10.0.0.5", "/srv/isos");
        let b = share_id("10.0.0.5", "/srv/isos");
        assert_eq!(a, b);
        assert!(!a.contains('/'));
        assert!(!a.contains('.'));
    }

    #[test]
    fn normalize_server_strips_schemes() {
        assert_eq!(normalize_server("  10.0.0.5  "), "10.0.0.5");
        assert_eq!(normalize_server("nfs://nas.lan/"), "nas.lan");
        assert_eq!(normalize_server("http://192.168.1.51"), "192.168.1.51");
        assert_eq!(normalize_server("nas.lan"), "nas.lan");
    }

    #[test]
    fn hint_for_acces_points_to_exports_table() {
        let h = hint_for("NFS server returned NFS3ERR_ACCES").unwrap();
        assert!(
            h.to_lowercase().contains("export"),
            "expected exports guidance, got: {h}"
        );
    }

    #[test]
    fn hint_for_mount_acces_calls_out_privileged_port_and_allowlist() {
        // The dominant v0.4.67 field failure: mount denied even with the
        // host IP allow-listed, because the export is `secure` and the
        // client used a high source port. The hint should mention both
        // the allow-list and the secure/insecure angle.
        let h = hint_for("connect failed: MNT3ERR_ACCES").unwrap();
        let lc = h.to_lowercase();
        assert!(
            lc.contains("insecure") || lc.contains("privileged"),
            "got: {h}"
        );
        assert!(
            lc.contains("allow") || lc.contains("permission"),
            "got: {h}"
        );
    }

    #[test]
    fn hint_for_noent_points_to_export_path() {
        let h = hint_for("NFS server returned NFS3ERR_NOENT").unwrap();
        assert!(
            h.to_lowercase().contains("path") || h.to_lowercase().contains("directory"),
            "expected export-path guidance, got: {h}"
        );
    }

    #[test]
    fn hint_for_mount_denied_calls_out_insecure_option() {
        let h = hint_for("mount denied").unwrap();
        assert!(h.to_lowercase().contains("insecure"));
    }

    #[test]
    fn hint_for_unknown_is_none() {
        assert!(hint_for("some entirely unrelated string").is_none());
    }

    #[test]
    fn status_label_covers_common_codes() {
        assert_eq!(status_label(nfs3::nfsstat3::NFS3_OK), "NFS3_OK");
        assert_eq!(status_label(nfs3::nfsstat3::NFS3ERR_ACCES), "NFS3ERR_ACCES");
        assert_eq!(status_label(nfs3::nfsstat3::NFS3ERR_NOENT), "NFS3ERR_NOENT");
    }
}