fix(nfs): reject NFSv4 calls with PROG_MISMATCH so clients fall back to v3 (#9262)

* feat(nfs): add NFSv3-only RPC version filter

The upstream willscott/go-nfs library dispatches RPC calls by (program,
procedure) only — it does not validate the program version. A client
sending NFSv4 (prog 100003 vers 4 proc 1 COMPOUND) lands on the same
handler map as NFSv3 and gets routed to v3 SETATTR, which parses the
COMPOUND args as SETATTR3args and writes a malformed reply. The kernel
then returns EPROTONOSUPPORT and mount.nfs prints "requested NFS version
or transport protocol is not supported" without retrying v3.

This commit adds a listener wrapper that peeks the first RPC frame on
each new TCP connection. If the program is NFS or MOUNT and the version
is not 3, it writes a protocol-correct PROG_MISMATCH reply (supported
range 3..3, per RFC 5531) directly to the socket and closes the
connection. v3 frames are replayed unchanged via a bufio reader so go-nfs
sees the original bytes. Unknown programs pass through so go-nfs's own
PROG_UNAVAIL handling stays in charge.

The filter is not yet wired into the server; the next commit activates
it. Tests cover NFSv4 reject, MOUNTv4 reject, NFSv3 pass-through, and
unknown-program pass-through.

* fix(nfs): wire NFSv3 version filter into the listener chain

Place the version filter after the optional client allowlist so that
unauthorized peers are still rejected first by IP/CIDR before we look at
RPC content. With the filter active, a Linux client doing the default
v4-first probe gets a clean PROG_MISMATCH reply pointing at v3, which
lets mount.nfs (and the in-kernel client) skip v4 and reuse the same v3
mountOptions that already work for rclone serve nfs against this
deployment.

* test(nfs): exercise MOUNT v4 in the v4-rejection test, not v1

TestVersionFilterRejectsMOUNTv4WithProgMismatch was sending
mountProgramID with version 1, so the test never actually covered the
"reject MOUNT v4" path it claims to exercise. The filter does reject any
non-v3 version uniformly, so the test still passed, but a future change
that tightened the version check (for example, only rejecting v4) would
let this test silently lie about coverage. Bump the call to version 4 so
the name matches what is actually exercised.

* refactor(nfs): reuse package RPC constants and io.ReadFull in version filter

The RPC numeric constants (msg_type=CALL/REPLY, MSG_ACCEPTED, PROG_MISMATCH,
AUTH_NONE, the NFS/MOUNT program numbers) are already named in
portmap.go alongside the portmap responder. Reuse them here instead of
defining a parallel set in rpc_version_filter.go: keeping one source of
truth per package means a future correction in one spot can't drift away
from the other. The filter-only constants (peek timeout, peek length,
supportedNFSVer) stay local because they have no portmap analog.

In the test, drop the bespoke readFull loop in favor of io.ReadFull.
The custom version was a near-identical reimplementation that did not
return io.ErrUnexpectedEOF on short reads, so the standard library is
both shorter and more diagnostic-friendly.

* fix(nfs): move RPC peek off the Accept path

The previous wrapper called filterFirstRPCFrame inline inside
versionFilterListener.Accept, which meant a single slow or idle TCP
connect could hold rpcVersionFilterPeekTimeout (10s) of head-of-line
blocking against every other accept: gonfs.Serve calls Accept serially,
so each in-flight peek stalled the next legitimate client until the
deadline expired. An attacker who simply opens a TCP connection without
sending any RPC payload could trivially throttle accept throughput.

Restructure the wrapper so a background goroutine drives the inner
Accept loop and hands each raw conn to its own short-lived goroutine
that runs the peek. Validated conns are sent on a buffered-once channel,
which the wrapper's Accept reads from; rejected conns finish their
PROG_MISMATCH reply and disappear without ever reaching the channel.
This means N concurrent slow clients only block themselves, not the
N+1th fast client that connects after them.

Add Close coordination — sync.WaitGroup for the accept loop and per-conn
peek goroutines, plus a closed channel so Accept unblocks immediately on
shutdown — so the wrapper now satisfies the full net.Listener contract
instead of relying on the embedded listener.

Add a regression test that opens a slow conn (TCP only, never writes)
and a fast conn (sends a v3 frame) and asserts the fast conn reaches
the inner accept handler well below the peek timeout.

* test(nfs): assert io.EOF (not just any error) after PROG_MISMATCH close

The post-rejection check was only failing when conn.Read succeeded; any
error — including a deadline timeout because the server kept the socket
open — let the test pass. That defeats the point of the assertion: a
regression where the filter replies but forgets to close would slip
through silently.

Match against io.EOF explicitly. The TCP semantics are deterministic
here: the server writes PROG_MISMATCH, calls conn.Close(), the client
reads what's left in flight and then sees a clean FIN, which surfaces
as io.EOF on the next zero-byte read.

* fix(nfs): reject short first fragments before parsing RPC header fields

bufio.Reader.Peek(28) is willing to read across record boundaries to
satisfy the requested length, so a final fragment whose body is shorter
than the 24-byte fixed RPC CALL header (xid + msg_type + rpcvers + prog
+ vers + proc) leaves the trailing peek bytes pointing at the next
RPC's framing or whatever bytes happen to follow on the wire. Indexing
hdr[16:24] for prog/vers in that state can spuriously reject (or pass
through) traffic based on data that doesn't belong to the request being
classified.

Drop those frames out of the filter early: if the first fragment can't
possibly hold a full CALL header, pass the connection straight to
go-nfs, which has its own framing-error handling for malformed input.

Add a regression test that crafts a 12-byte first fragment whose
trailing peek bytes are deliberately shaped like an NFSv4 CALL — without
the length check the filter sends a PROG_MISMATCH; with it, the conn
passes through silently. Verified by stashing the production-code change
and running the test in isolation: it fails as expected without the fix.

* fix(nfs): retry transient Accept() errors instead of treating any error as terminal

acceptLoop previously exited on the first error returned by the inner
listener's Accept(). That conflates two very different failure modes:
permanent shutdown (the listener was Close()d, OS-level fatal failure)
and transient resource pressure (EMFILE, EAGAIN, ECONNABORTED on
accept). The transient case should not take the entire NFS server down
— a single fd-table-full event would leave the deployment offline until
restart.

Classify the error: errors.Is(err, net.ErrClosed) is the permanent
signal we already wanted to surface to Accept(); everything else is
transient. Log at V(1) and back off rpcVersionFilterAcceptBackoff
(50ms, mirroring portmap.go's portmapRetryBackoff) before retrying. The
backoff sleep is interruptible via the closed channel so Close() still
shuts the loop down promptly.

Add a regression test that wraps a real listener with one that injects
3 fake transient errors before delegating, and asserts Accept() still
delivers the next real connection. Verified the test fails on the old
"any error is terminal" loop and passes with this change.

* fix(nfs): only synthesize PROG_MISMATCH for ONC RPC v2 traffic

The filter was rejecting any CALL-shaped record with prog=100003 or
100005 and vers!=3, regardless of the rpcvers field. If the caller is
speaking some other protocol that happens to share the port — or just
sending garbled bytes — pretending to be an NFSv3 server replying
PROG_MISMATCH is misleading at best, and at worst fabricates a coherent
RPC reply for traffic we don't actually understand.

Add an rpcvers==2 check between the msg_type and prog/vers parses. Any
non-v2 record now passes through to go-nfs, whose RFC 5531 §9
RPC_MISMATCH handling is the correct place to reject mis-versioned RPC.

Regression test takes a normal v3 NFS CALL frame, overwrites the rpcvers
field with 99, and asserts no PROG_MISMATCH-shaped reply lands on the
client and that the conn is delivered to the inner accept handler.
Verified the test fails on the previous code (filter still rejected on
prog/vers alone) and passes with the guard in place.

* fix(nfs): bound Close() latency by evicting in-flight prefilter conns

Close() does wg.Wait() to drain handleConn goroutines, but each of those
goroutines can be parked inside filterFirstRPCFrame's bufio.Peek for up
to rpcVersionFilterPeekTimeout (10s) waiting for the very first RPC
header. A client that completes the TCP handshake but never sends a
byte therefore stretched shutdown by 10s per such conn — a real
regression for stop/restart paths and for tests that just want to tear
the listener down.

Track raw (pre-peek) conns in versionFilterListener.inFlight as
handleConn enters, untrack on exit, and have Close() forcibly close
every tracked conn before wg.Wait. Closing the underlying conn breaks
its Peek immediately, so handleConn returns within a single scheduler
hop. trackInFlight also short-circuits if shutdown has already started,
so a conn accepted after signalClose can't slip past the eviction.

Black-box regression test opens 4 idle TCP-handshake-only conns, lets
their handleConn goroutines settle into Peek, and asserts Close()
returns under 2s. Verified: same test fails on the previous code with
Close taking ~9.9s; passes here at ~100ms.

weed/server/nfs/rpc_version_filter.go

@@ -0,0 +1,377 @@
+package nfs
+
+import (
+	"bufio"
+	"encoding/binary"
+	"errors"
+	"io"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+)
+
+// The upstream willscott/go-nfs library dispatches RPC calls by (program,
+// procedure) only — it does not validate the RPC program version. That means
+// a Linux client speaking NFSv4 (program 100003 vers 4) lands on the same
+// handler map as NFSv3: proc=1 routes to NFSv3 SETATTR, which parses the
+// NFSv4 COMPOUND args as if they were SETATTR3args and writes a malformed
+// reply. The client cannot decode that reply, the kernel returns
+// EPROTONOSUPPORT, and mount.nfs prints "requested NFS version or transport
+// protocol is not supported" without ever falling back to v3.
+//
+// The default Linux mount.nfs path is to try NFSv4 first, so this affects
+// every plain `mount -t nfs <host>:<export> /mnt` against a `weed nfs`
+// deployment. To make the v4→v3 fallback work, we wrap the listener so the
+// first RPC frame on each new TCP connection is inspected: if the program is
+// NFS or MOUNT and the version is not 3, we synthesize a PROG_MISMATCH reply
+// (with the supported version range 3..3) directly to the socket and close
+// the connection. The client then retries with v3 and proceeds normally.
+//
+// Clients keep the same program/version for the lifetime of a TCP connection
+// in practice, so we only need to check the first frame; subsequent frames
+// flow through to go-nfs unchanged. This avoids vendoring go-nfs while still
+// producing protocol-correct rejections.
+
+// RPC numeric constants used here (rpcMsgCall, rpcMsgReply, rpcMsgAccepted,
+// rpcAcceptProgMismatch, rpcAuthNone, nfsProgram, mountProgram) are defined
+// alongside the portmap responder in portmap.go to keep one source of truth
+// per package.
+const (
+	// rpcVersionFilterPeekTimeout bounds how long we wait for the first frame
+	// header on a new connection before giving up and letting go-nfs handle
+	// the (possibly half-open) socket.
+	rpcVersionFilterPeekTimeout = 10 * time.Second
+
+	// peeked length: 4-byte fragment marker + 24 bytes of fixed RPC header
+	// (xid + msg_type + rpcvers + prog + vers + proc).
+	rpcVersionFilterPeekLen = 28
+
+	// rpcVersionFilterAcceptBackoff is how long the accept loop sleeps
+	// after a transient Accept() error (EMFILE, EAGAIN, ECONNABORTED,
+	// etc.) before retrying. Mirrors portmapRetryBackoff in portmap.go so
+	// both NFS-listening goroutines back off identically under host
+	// resource pressure.
+	rpcVersionFilterAcceptBackoff = 50 * time.Millisecond
+
+	supportedNFSVer = 3
+)
+
+// versionFilterListener moves the per-connection RPC peek off the
+// Listener.Accept() critical path. Peeking inline would let one slow or idle
+// client (or a TCP three-way handshake without any RPC payload) hold
+// rpcVersionFilterPeekTimeout — i.e. up to 10 seconds — of head-of-line
+// blocking against every other connect, since gonfs.Serve only calls Accept
+// serially. Instead, a background goroutine runs the inner Accept() loop and
+// hands each raw conn to its own short-lived goroutine that does the peek;
+// validated conns are sent on acceptCh and the wrapper's Accept() reads from
+// that channel. Rejected conns never reach the channel — PROG_MISMATCH is
+// already on the wire by the time the per-conn goroutine returns.
+type versionFilterListener struct {
+	inner    net.Listener
+	acceptCh chan net.Conn
+
+	// closed is signalled either by Close() or by the accept loop after the
+	// inner listener returns a terminal error. After it fires Accept() will
+	// stop blocking and return acceptErr (or net.ErrClosed if none).
+	closed    chan struct{}
+	closeOnce sync.Once
+
+	mu        sync.Mutex
+	acceptErr error
+	// inFlight tracks raw (pre-peek) conns that are currently in
+	// handleConn so Close() can break their Peek() deadline by closing
+	// them, instead of waiting up to rpcVersionFilterPeekTimeout per
+	// idle client for the timeout to fire on its own.
+	inFlight map[net.Conn]struct{}
+
+	startOnce sync.Once
+	wg        sync.WaitGroup
+}
+
+func newVersionFilterListener(inner net.Listener) net.Listener {
+	return &versionFilterListener{
+		inner:    inner,
+		acceptCh: make(chan net.Conn),
+		closed:   make(chan struct{}),
+	}
+}
+
+// start lazily kicks off the background accept loop the first time someone
+// calls Accept(). This matches the behaviour of the embedded-listener form we
+// replaced — no goroutines spawn just from constructing the wrapper.
+func (l *versionFilterListener) start() {
+	l.startOnce.Do(func() {
+		l.wg.Add(1)
+		go l.acceptLoop()
+	})
+}
+
+func (l *versionFilterListener) Accept() (net.Conn, error) {
+	l.start()
+	select {
+	case c := <-l.acceptCh:
+		return c, nil
+	case <-l.closed:
+		return nil, l.terminalErr()
+	}
+}
+
+func (l *versionFilterListener) Close() error {
+	l.signalClose()
+	err := l.inner.Close()
+	// Eagerly close any raw conns currently blocked in filterFirstRPCFrame's
+	// Peek so handleConn returns promptly. Without this, an idle client
+	// (TCP handshake without any RPC payload) holds Close() up to
+	// rpcVersionFilterPeekTimeout — 10s of stop-the-world per such conn.
+	l.evictInFlight()
+	l.wg.Wait()
+	return err
+}
+
+func (l *versionFilterListener) Addr() net.Addr {
+	return l.inner.Addr()
+}
+
+func (l *versionFilterListener) signalClose() {
+	l.closeOnce.Do(func() {
+		close(l.closed)
+	})
+}
+
+func (l *versionFilterListener) terminalErr() error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.acceptErr != nil {
+		return l.acceptErr
+	}
+	return net.ErrClosed
+}
+
+// trackInFlight records a raw conn that's about to be peeked, so Close()
+// can break its Peek() deadline by closing it. Returns false if shutdown
+// has already started; the caller must close the conn and bail.
+func (l *versionFilterListener) trackInFlight(c net.Conn) bool {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	select {
+	case <-l.closed:
+		return false
+	default:
+	}
+	if l.inFlight == nil {
+		l.inFlight = make(map[net.Conn]struct{})
+	}
+	l.inFlight[c] = struct{}{}
+	return true
+}
+
+func (l *versionFilterListener) untrackInFlight(c net.Conn) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	delete(l.inFlight, c)
+}
+
+// evictInFlight closes every conn currently in handleConn so their
+// in-flight Peek() returns immediately. delete(nil-map, k) is a no-op,
+// so handleConn's deferred untrackInFlight is safe even after we've
+// nilled the map here.
+func (l *versionFilterListener) evictInFlight() {
+	l.mu.Lock()
+	conns := l.inFlight
+	l.inFlight = nil
+	l.mu.Unlock()
+	for c := range conns {
+		_ = c.Close()
+	}
+}
+
+func (l *versionFilterListener) acceptLoop() {
+	defer l.wg.Done()
+	defer l.signalClose()
+	for {
+		conn, err := l.inner.Accept()
+		if err != nil {
+			// Permanent: the inner listener has been closed (Close(),
+			// shutdown, or an unrecoverable error from the OS). Surface
+			// the error to Accept() and stop.
+			if errors.Is(err, net.ErrClosed) {
+				l.mu.Lock()
+				if l.acceptErr == nil {
+					l.acceptErr = err
+				}
+				l.mu.Unlock()
+				return
+			}
+			// Transient (EMFILE, EAGAIN, ECONNABORTED on accept,
+			// timeouts if a deadline is ever set): treating these as
+			// terminal would tear the whole NFS server down on a
+			// resource blip. Back off briefly and retry, mirroring the
+			// pattern in portmap.go's serveTCP.
+			glog.V(1).Infof("nfs version filter: transient accept error: %v", err)
+			select {
+			case <-l.closed:
+				return
+			case <-time.After(rpcVersionFilterAcceptBackoff):
+				continue
+			}
+		}
+		l.wg.Add(1)
+		go l.handleConn(conn)
+	}
+}
+
+// handleConn runs the version peek for a single accepted conn. Because each
+// conn has its own goroutine, a slow client only blocks itself; concurrent
+// peeks proceed in parallel up to whatever the runtime can schedule. If
+// Close() fires before the peek completes we drop the validated conn so we
+// don't leak a socket past shutdown.
+func (l *versionFilterListener) handleConn(conn net.Conn) {
+	defer l.wg.Done()
+	if !l.trackInFlight(conn) {
+		// Shutdown beat us: don't start the Peek that we'd then
+		// have to break, just close the raw conn.
+		_ = conn.Close()
+		return
+	}
+	defer l.untrackInFlight(conn)
+
+	wrapped, accepted := filterFirstRPCFrame(conn)
+	if !accepted {
+		// Already replied with PROG_MISMATCH and closed conn.
+		return
+	}
+	select {
+	case l.acceptCh <- wrapped:
+	case <-l.closed:
+		_ = wrapped.Close()
+	}
+}
+
+// peekedConn returns the bytes that filterFirstRPCFrame already buffered when
+// it peeked the first RPC header, then transparently reads from the
+// underlying connection. Writes go straight to the socket; the bufio reader
+// only buffers the read side.
+type peekedConn struct {
+	net.Conn
+	reader io.Reader
+}
+
+func (c *peekedConn) Read(p []byte) (int, error) {
+	return c.reader.Read(p)
+}
+
+// filterFirstRPCFrame inspects the first RPC frame on conn and decides whether
+// to pass it through to go-nfs. Returns (wrappedConn, true) if the frame is
+// for a supported (program, version) — including programs we don't recognize,
+// since go-nfs handles its own PROG_UNAVAIL response. Returns (nil, false) if
+// we already replied with PROG_MISMATCH and closed conn.
+//
+// On peek failure (early close, deadline) we pass the connection through:
+// returning an error here would silently drop legitimate clients on a flaky
+// link, and go-nfs has its own per-frame error handling.
+func filterFirstRPCFrame(conn net.Conn) (net.Conn, bool) {
+	r := bufio.NewReader(conn)
+
+	deadlineErr := conn.SetReadDeadline(time.Now().Add(rpcVersionFilterPeekTimeout))
+
+	hdr, peekErr := r.Peek(rpcVersionFilterPeekLen)
+
+	// Always clear the deadline before returning to go-nfs; failing to do so
+	// would make every subsequent Read() time out at the same instant.
+	if deadlineErr == nil {
+		_ = conn.SetReadDeadline(time.Time{})
+	}
+
+	if peekErr != nil {
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+
+	fragMark := binary.BigEndian.Uint32(hdr[0:4])
+	if fragMark&(1<<31) == 0 {
+		// Multi-fragment record: portmap-style filtering of the first frame
+		// would need reassembly. Fall through to go-nfs which handles this.
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+
+	// Peek(28) can read across record boundaries — the first fragment may
+	// be shorter than the fixed RPC CALL header (24 bytes after the marker)
+	// with the remaining bytes belonging to the *next* RPC. Indexing into
+	// hdr[16:24] without first checking the fragment length would parse
+	// fields from a different RPC and either spuriously reject or pass it.
+	// Pass through if the first fragment can't possibly hold a full header
+	// and let go-nfs surface the framing error.
+	if fragLen := fragMark &^ uint32(1<<31); fragLen < 24 {
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+
+	xid := binary.BigEndian.Uint32(hdr[4:8])
+	if msgType := binary.BigEndian.Uint32(hdr[8:12]); msgType != rpcMsgCall {
+		// Not a CALL — odd, but pass through.
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+	if rpcVers := binary.BigEndian.Uint32(hdr[12:16]); rpcVers != 2 {
+		// ONC RPC v2 is the only version we and go-nfs speak; if the
+		// rpcvers field is anything else the rest of the header is
+		// untrusted (could be a non-RPC protocol that happens to share
+		// the port, or simply garbled traffic). Don't synthesize a
+		// PROG_MISMATCH that lies about supporting NFS — pass it
+		// through and let go-nfs / RFC 5531 §9 RPC_MISMATCH handling
+		// in the upstream library do the right thing.
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+
+	prog := binary.BigEndian.Uint32(hdr[16:20])
+	vers := binary.BigEndian.Uint32(hdr[20:24])
+
+	switch prog {
+	case nfsProgram, mountProgram:
+	default:
+		// Unknown program: let go-nfs reply PROG_UNAVAIL itself.
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+
+	if vers == supportedNFSVer {
+		return &peekedConn{Conn: conn, reader: r}, true
+	}
+
+	glog.V(1).Infof("nfs: rejecting client %s with PROG_MISMATCH: prog=%d vers=%d (supported=%d)",
+		conn.RemoteAddr(), prog, vers, supportedNFSVer)
+
+	if err := writeProgMismatchTCP(conn, xid, supportedNFSVer, supportedNFSVer); err != nil {
+		glog.V(1).Infof("nfs: write PROG_MISMATCH to %s: %v", conn.RemoteAddr(), err)
+	}
+	_ = conn.Close()
+	return nil, false
+}
+
+// writeProgMismatchTCP encodes a single-frame TCP RPC reply carrying
+// MSG_ACCEPTED + PROG_MISMATCH along with the supported version range, per
+// RFC 5531 section 9. The frame layout is:
+//
+//	uint32 fragment_header (last-fragment | length)
+//	uint32 xid
+//	uint32 msg_type=REPLY(1)
+//	uint32 reply_stat=MSG_ACCEPTED(0)
+//	uint32 verf_flavor=AUTH_NONE(0)
+//	uint32 verf_len=0
+//	uint32 accept_stat=PROG_MISMATCH(2)
+//	uint32 low
+//	uint32 high
+const progMismatchBodyLen = 32
+
+func writeProgMismatchTCP(w io.Writer, xid, low, high uint32) error {
+	out := make([]byte, 4+progMismatchBodyLen)
+	binary.BigEndian.PutUint32(out[0:4], uint32(progMismatchBodyLen)|(1<<31))
+	binary.BigEndian.PutUint32(out[4:8], xid)
+	binary.BigEndian.PutUint32(out[8:12], rpcMsgReply)
+	binary.BigEndian.PutUint32(out[12:16], rpcMsgAccepted)
+	binary.BigEndian.PutUint32(out[16:20], rpcAuthNone)
+	binary.BigEndian.PutUint32(out[20:24], 0) // verf opaque length (always zero for AUTH_NONE)
+	binary.BigEndian.PutUint32(out[24:28], rpcAcceptProgMismatch)
+	binary.BigEndian.PutUint32(out[28:32], low)
+	binary.BigEndian.PutUint32(out[32:36], high)
+	_, err := w.Write(out)
+	return err
+}

weed/server/nfs/rpc_version_filter_test.go

@@ -0,0 +1,560 @@
+package nfs
+
+import (
+	"encoding/binary"
+	"errors"
+	"io"
+	"net"
+	"sync"
+	"testing"
+	"time"
+)
+
+// buildRPCCallFrame constructs a single TCP-framed RPC CALL header without
+// procedure arguments — enough for the version filter to decide whether to
+// reject the connection. The frame layout matches RFC 5531 (Open Network
+// Computing RPC v2): a 4-byte fragment marker (last-fragment bit set on a
+// 40-byte body) followed by xid + msg_type=CALL + rpcvers=2 + prog + vers +
+// proc + two empty AUTH_NONE opaque_auth structs.
+func buildRPCCallFrame(xid, prog, vers, proc uint32) []byte {
+	const bodyLen = 40
+	frame := make([]byte, 4+bodyLen)
+	binary.BigEndian.PutUint32(frame[0:4], uint32(bodyLen)|(1<<31))
+	binary.BigEndian.PutUint32(frame[4:8], xid)
+	binary.BigEndian.PutUint32(frame[8:12], 0) // msg_type CALL
+	binary.BigEndian.PutUint32(frame[12:16], 2)
+	binary.BigEndian.PutUint32(frame[16:20], prog)
+	binary.BigEndian.PutUint32(frame[20:24], vers)
+	binary.BigEndian.PutUint32(frame[24:28], proc)
+	// cred + verf both AUTH_NONE / length 0
+	return frame
+}
+
+// readPROGMismatchReply parses a TCP-framed PROG_MISMATCH reply produced by
+// writeProgMismatchTCP and returns the xid plus the supported (low, high)
+// version range advertised by the server.
+func readPROGMismatchReply(t *testing.T, conn net.Conn) (xid, low, high uint32) {
+	t.Helper()
+	_ = conn.SetReadDeadline(time.Now().Add(2 * time.Second))
+	buf := make([]byte, 4+progMismatchBodyLen)
+	n, err := io.ReadFull(conn, buf)
+	if err != nil {
+		t.Fatalf("read reply: %v (got %d bytes)", err, n)
+	}
+	frag := binary.BigEndian.Uint32(buf[0:4])
+	if frag&(1<<31) == 0 {
+		t.Fatalf("reply frame missing last-fragment bit: %x", frag)
+	}
+	if got := frag &^ (1 << 31); got != progMismatchBodyLen {
+		t.Fatalf("reply body length=%d want %d", got, progMismatchBodyLen)
+	}
+	xid = binary.BigEndian.Uint32(buf[4:8])
+	if mt := binary.BigEndian.Uint32(buf[8:12]); mt != 1 {
+		t.Fatalf("reply msg_type=%d want REPLY(1)", mt)
+	}
+	if rs := binary.BigEndian.Uint32(buf[12:16]); rs != 0 {
+		t.Fatalf("reply reply_stat=%d want MSG_ACCEPTED(0)", rs)
+	}
+	if as := binary.BigEndian.Uint32(buf[24:28]); as != 2 {
+		t.Fatalf("reply accept_stat=%d want PROG_MISMATCH(2)", as)
+	}
+	low = binary.BigEndian.Uint32(buf[28:32])
+	high = binary.BigEndian.Uint32(buf[32:36])
+	return
+}
+
+func TestVersionFilterRejectsNFSv4WithProgMismatch(t *testing.T) {
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+
+	// In a real server, accepted conns are passed to go-nfs. We just need
+	// to drive Accept() so the filter runs; the test never sees a wrapped
+	// conn because the v4 frame is rejected.
+	accepted := make(chan net.Conn, 1)
+	go func() {
+		for {
+			c, aerr := listener.Accept()
+			if aerr != nil {
+				return
+			}
+			accepted <- c
+		}
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer conn.Close()
+
+	// NFSv4 NULL: the first probe Linux mount.nfs sends when trying v4.
+	if _, err := conn.Write(buildRPCCallFrame(0xdeadbeef, nfsProgram, 4, 0)); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	xid, low, high := readPROGMismatchReply(t, conn)
+	if xid != 0xdeadbeef {
+		t.Errorf("xid=%x want %x", xid, 0xdeadbeef)
+	}
+	if low != supportedNFSVer || high != supportedNFSVer {
+		t.Errorf("supported range=(%d,%d) want (%d,%d)", low, high, supportedNFSVer, supportedNFSVer)
+	}
+
+	// Filter must close the connection after replying so the client knows
+	// not to send another RPC on this socket. Insist on io.EOF specifically:
+	// "any error" would let a stuck (but still-open) connection pass this
+	// check via a deadline timeout, which is exactly the regression we want
+	// to catch.
+	_ = conn.SetReadDeadline(time.Now().Add(time.Second))
+	one := make([]byte, 1)
+	n, err := conn.Read(one)
+	switch {
+	case err == nil:
+		t.Errorf("expected EOF after PROG_MISMATCH but read returned %d bytes", n)
+	case !errors.Is(err, io.EOF):
+		t.Errorf("expected io.EOF after PROG_MISMATCH, got %v (likely a regression where the filter replies but does not close)", err)
+	}
+
+	select {
+	case c := <-accepted:
+		c.Close()
+		t.Error("rejected connection should not be returned to caller")
+	case <-time.After(100 * time.Millisecond):
+	}
+}
+
+func TestVersionFilterRejectsMOUNTv4WithProgMismatch(t *testing.T) {
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+	go func() {
+		for {
+			c, aerr := listener.Accept()
+			if aerr != nil {
+				return
+			}
+			c.Close()
+		}
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	if _, err := conn.Write(buildRPCCallFrame(42, mountProgram, 4, 0)); err != nil {
+		t.Fatal(err)
+	}
+
+	xid, low, high := readPROGMismatchReply(t, conn)
+	if xid != 42 {
+		t.Errorf("xid=%d want 42", xid)
+	}
+	if low != supportedNFSVer || high != supportedNFSVer {
+		t.Errorf("supported range=(%d,%d) want (3,3)", low, high)
+	}
+}
+
+func TestVersionFilterPassesThroughNFSv3(t *testing.T) {
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+	got := make(chan []byte, 1)
+	go func() {
+		c, aerr := listener.Accept()
+		if aerr != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, 44)
+		_, rerr := io.ReadFull(c, buf)
+		if rerr != nil {
+			return
+		}
+		got <- buf
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	frame := buildRPCCallFrame(7, nfsProgram, 3, 0)
+	if _, err := conn.Write(frame); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case received := <-got:
+		if string(received) != string(frame) {
+			t.Error("v3 frame was modified or partially consumed by filter")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("v3 frame not delivered to inner accept handler")
+	}
+}
+
+func TestVersionFilterPassesThroughUnknownProgram(t *testing.T) {
+	// The filter should only police NFS / MOUNT versions; other programs
+	// reach go-nfs which already responds PROG_UNAVAIL itself. Otherwise
+	// adding a new program (e.g. NLM) here would require updating the
+	// filter, which would defeat the point of using it as a thin shim.
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+	delivered := make(chan struct{}, 1)
+	go func() {
+		c, aerr := listener.Accept()
+		if aerr != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, 44)
+		if _, rerr := io.ReadFull(c, buf); rerr == nil {
+			delivered <- struct{}{}
+		}
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	// Program 100021 is NLM, which weed nfs doesn't run; let go-nfs handle
+	// the unsupported-program reply.
+	if _, err := conn.Write(buildRPCCallFrame(99, 100021, 4, 0)); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-delivered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("unknown-program frame should pass through filter")
+	}
+}
+
+// transientErrListener wraps a real net.Listener but injects a configurable
+// number of transient Accept() errors before delegating. It exists only to
+// regression-test the version filter's transient-retry behaviour without
+// having to provoke real EMFILE conditions on the host.
+type transientErrListener struct {
+	inner   net.Listener
+	mu      sync.Mutex
+	remaining int
+}
+
+type fakeAcceptError struct{}
+
+func (fakeAcceptError) Error() string { return "fake transient accept error" }
+
+func (l *transientErrListener) Accept() (net.Conn, error) {
+	l.mu.Lock()
+	if l.remaining > 0 {
+		l.remaining--
+		l.mu.Unlock()
+		return nil, fakeAcceptError{}
+	}
+	l.mu.Unlock()
+	return l.inner.Accept()
+}
+
+func (l *transientErrListener) Close() error   { return l.inner.Close() }
+func (l *transientErrListener) Addr() net.Addr { return l.inner.Addr() }
+
+func TestVersionFilterRetriesTransientAcceptErrors(t *testing.T) {
+	// Regression test: previously the accept loop exited on any error
+	// from the inner listener, which meant a single transient EMFILE /
+	// EAGAIN under host resource pressure would tear the entire NFS
+	// server down. Inject a few fake transient errors and assert the
+	// filter still delivers the next real connection.
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	injected := &transientErrListener{inner: innerListener, remaining: 3}
+	listener := newVersionFilterListener(injected)
+
+	delivered := make(chan struct{}, 1)
+	go func() {
+		c, aerr := listener.Accept()
+		if aerr != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, 44)
+		if _, rerr := io.ReadFull(c, buf); rerr == nil {
+			delivered <- struct{}{}
+		}
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	if _, err := conn.Write(buildRPCCallFrame(1, nfsProgram, 3, 0)); err != nil {
+		t.Fatal(err)
+	}
+
+	// 3 transient errors × ~50ms backoff plus normal accept latency. Allow
+	// a generous bound so flakes on slow CI don't surface here, but still
+	// tight enough to catch a regression to "any error is terminal".
+	select {
+	case <-delivered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("filter did not retry transient Accept() errors and recover")
+	}
+}
+
+func TestVersionFilterCloseReturnsPromptlyWithIdlePeekConns(t *testing.T) {
+	// Regression test: Close() used to wait on every handleConn goroutine
+	// via wg.Wait, but those goroutines could be stuck in
+	// filterFirstRPCFrame's Peek() until rpcVersionFilterPeekTimeout (10s)
+	// fired. An idle client that completed a TCP handshake but never sent
+	// a byte would stretch shutdown by up to that timeout per conn.
+	// Close() now eagerly closes any tracked in-flight raw conns, which
+	// forces Peek() to return immediately and lets handleConn finish.
+	//
+	// Black-box test: only observes Close() latency. With the regression
+	// in place Close() would block ~10s; with the fix it returns in well
+	// under a second.
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	listener := newVersionFilterListener(innerListener)
+	// Drive Accept once so the background accept loop is running.
+	go func() { _, _ = listener.Accept() }()
+
+	const idleConns = 4
+	dialed := make([]net.Conn, 0, idleConns)
+	defer func() {
+		for _, c := range dialed {
+			_ = c.Close()
+		}
+	}()
+	for i := 0; i < idleConns; i++ {
+		c, err := net.Dial("tcp", innerListener.Addr().String())
+		if err != nil {
+			t.Fatal(err)
+		}
+		dialed = append(dialed, c)
+	}
+
+	// Give handleConn time to invoke Peek for each idle conn — without
+	// this the test could race ahead and Close() while no goroutine has
+	// actually started peeking yet, masking the regression.
+	time.Sleep(100 * time.Millisecond)
+
+	// Close() must finish in well under rpcVersionFilterPeekTimeout (10s).
+	// 2s is a generous bound that still clearly distinguishes "broke the
+	// peek by closing the conn" from "waited for the peek deadline".
+	start := time.Now()
+	if err := listener.Close(); err != nil {
+		t.Errorf("Close: %v", err)
+	}
+	elapsed := time.Since(start)
+	if elapsed > 2*time.Second {
+		t.Errorf("Close took %v with %d idle pre-peek conns; should be sub-second once they're forcibly closed", elapsed, idleConns)
+	}
+}
+
+func TestVersionFilterPassesThroughNonV2RPC(t *testing.T) {
+	// Anything that isn't ONC RPC v2 isn't ours to classify — even if the
+	// bytes at hdr[16:24] happen to look like nfsProgram + vers=4, we
+	// shouldn't synthesize a PROG_MISMATCH advertising NFSv3 support for
+	// what could be a completely different protocol sharing the port.
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+	delivered := make(chan struct{}, 1)
+	go func() {
+		c, aerr := listener.Accept()
+		if aerr != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, 44)
+		if _, rerr := io.ReadFull(c, buf); rerr == nil {
+			delivered <- struct{}{}
+		}
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	// Build a CALL frame, then overwrite the rpcvers field with 99.
+	// Without the rpcvers guard the filter would still parse prog=NFS,
+	// vers=4 from the same buffer and reject with PROG_MISMATCH.
+	frame := buildRPCCallFrame(0xfeedbeef, nfsProgram, 4, 0)
+	binary.BigEndian.PutUint32(frame[12:16], 99) // bogus rpcvers
+	if _, err := conn.Write(frame); err != nil {
+		t.Fatal(err)
+	}
+
+	// Try to read a PROG_MISMATCH reply with a short deadline — none
+	// should arrive because the filter shouldn't pretend to know what
+	// this protocol is.
+	_ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
+	hdr := make([]byte, 4)
+	if n, err := io.ReadFull(conn, hdr); err == nil && n == 4 {
+		if got := binary.BigEndian.Uint32(hdr); got == uint32(progMismatchBodyLen)|(1<<31) {
+			t.Fatal("filter sent PROG_MISMATCH for a non-v2 RPC frame")
+		}
+	}
+
+	// And the connection should reach the inner accept handler.
+	select {
+	case <-delivered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("non-v2 RPC frame should pass through filter to inner accept")
+	}
+}
+
+func TestVersionFilterIgnoresShortFirstFragment(t *testing.T) {
+	// Peek(28) can read past the first fragment's body when the body is
+	// shorter than the 24-byte fixed RPC CALL header. Without a length
+	// check, the prog/vers fields would be sourced from bytes belonging to
+	// the *next* RPC (or a syntactic accident), and the filter could
+	// spuriously reject the connection. Send a 12-byte first fragment whose
+	// trailing peek-region bytes look like an NFSv4 CALL header, and assert
+	// the filter does NOT emit a PROG_MISMATCH reply.
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+	go func() {
+		for {
+			c, aerr := listener.Accept()
+			if aerr != nil {
+				return
+			}
+			c.Close()
+		}
+	}()
+
+	conn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	const shortBody = 12
+	payload := make([]byte, 4+24)
+	binary.BigEndian.PutUint32(payload[0:4], shortBody|(1<<31)) // last-fragment, body=12
+	// Bytes 4..16 are the actual fragment body (12 bytes — too short for a
+	// CALL header; the filter must not look at them as one).
+	// Bytes 16..28 sit past the fragment in the peek window. If we were to
+	// (incorrectly) read prog/vers from hdr[16:24], we'd see NFS+v4 here.
+	binary.BigEndian.PutUint32(payload[16:20], nfsProgram)
+	binary.BigEndian.PutUint32(payload[20:24], 4)
+
+	if _, err := conn.Write(payload); err != nil {
+		t.Fatal(err)
+	}
+
+	// If the filter erroneously rejected, it would send a 36-byte TCP RPC
+	// reply (4-byte frag marker + 32-byte PROG_MISMATCH body) within ms.
+	// Wait briefly and assert nothing PROG_MISMATCH-shaped came back.
+	_ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
+	hdr := make([]byte, 4)
+	n, err := io.ReadFull(conn, hdr)
+	if err == nil && n == 4 {
+		if got := binary.BigEndian.Uint32(hdr); got == uint32(progMismatchBodyLen)|(1<<31) {
+			t.Fatal("filter sent PROG_MISMATCH on a short fragment whose trailing peek bytes only superficially resembled a v4 call")
+		}
+	}
+	// Anything else (timeout, EOF, or unrelated bytes) is fine — we only
+	// care that the filter did NOT misclassify the short fragment.
+}
+
+func TestVersionFilterDoesNotHeadOfLineBlockOnSlowConn(t *testing.T) {
+	// Regression test: the previous implementation peeked the first RPC
+	// frame inline in Accept(), so an idle TCP-only connect would block
+	// every later Accept() call for up to rpcVersionFilterPeekTimeout.
+	// The peek now runs in a per-conn goroutine; a fast follow-up connect
+	// must reach the inner accept handler well before the slow conn's
+	// peek deadline.
+	innerListener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer innerListener.Close()
+
+	listener := newVersionFilterListener(innerListener)
+
+	delivered := make(chan struct{}, 1)
+	go func() {
+		c, aerr := listener.Accept()
+		if aerr != nil {
+			return
+		}
+		defer c.Close()
+		buf := make([]byte, 44)
+		if _, rerr := io.ReadFull(c, buf); rerr == nil {
+			delivered <- struct{}{}
+		}
+	}()
+
+	// Slow client: connect, never write. Holds a goroutine inside the
+	// filter peeking until the deadline, but must not block the next conn.
+	slowConn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer slowConn.Close()
+
+	// Fast client: send a valid v3 frame straight away; this conn must be
+	// delivered to the inner accept handler without waiting for slowConn.
+	fastConn, err := net.Dial("tcp", innerListener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer fastConn.Close()
+
+	if _, err := fastConn.Write(buildRPCCallFrame(11, nfsProgram, 3, 0)); err != nil {
+		t.Fatal(err)
+	}
+
+	// Bound the wait well below rpcVersionFilterPeekTimeout (10s) so a
+	// regression to inline peeking would clearly time out here.
+	select {
+	case <-delivered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("fast conn should not be head-of-line blocked by slow conn's peek")
+	}
+}

weed/server/nfs/server.go

@@ -143,6 +143,7 @@ func (s *Server) serve(listener net.Listener) error {
 			authorizer: s.clientAuthorizer,
 		}
 	}
+	listener = newVersionFilterListener(listener)
 
 	handler, err := s.newHandler()
 	if err != nil {