// Copyright (c) 2026 Petr Balvín <opensource@petrbalvin.org> (https://petrbalvin.org)
// SPDX-License-Identifier: MIT

// Package nfsclient is the client half of the nfs project: it speaks the
// same wire vocabulary as the server and serves as the second oracle
// against it. It speaks NULL and COMPOUND, from the session establishment
// through the stateful operations.
package nfsclient

import (
	crand "crypto/rand"
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"os"
	"sync"
	"time"

	"sourcedock.dev/petrbalvin/nfs/internal/xdr"

	"sourcedock.dev/petrbalvin/nfs/internal/krb5"
	"sourcedock.dev/petrbalvin/nfs/internal/nfs4"
	"sourcedock.dev/petrbalvin/nfs/internal/rpc"
)

// maxRecord bounds one ONC RPC record of a reply.
const maxRecord = 4 << 20

// defaultTimeout bounds one call and the dial when the caller set no
// Timeout of its own, so a wedged server fails the caller instead of
// hanging the process.
const defaultTimeout = 30 * time.Second

// A wireResult is one routed reply, or the failure that ended the
// connection before it arrived.
type wireResult struct {
	rec []byte
	err error
}

// A Client is one NFS connection. Its methods are safe for concurrent use
// and are multiplexed over the single connection the protocol runs on.
// The same connection carries the server's CB_COMPOUND calls: the reader
// demultiplexes the two directions by message type and xid.
type Client struct {
	conn    net.Conn
	writeMu sync.Mutex
	mu      sync.Mutex
	next    uint32
	// Timeout bounds one call and the dial; zero means the default.
	Timeout   time.Duration
	sessionID *nfs4.SessionID
	cbProgram uint32

	// The AUTH_SYS identity every call asserts, root by default. SetIdentity
	// changes it between calls; a compound already in flight keeps the
	// identity it was written with.
	uid    uint32
	gid    uint32
	groups []uint32

	// The session slots this client drives. Slot zero alone is the
	// default and serialises the compounds; SetConcurrency admits more
	// slots, each with its own sequence, so several compounds run in
	// flight at once. The channel holds the free slot numbers.
	slotSeq [8]uint32
	slots   chan uint32

	pendingMu sync.Mutex
	pending   map[uint32]chan wireResult
	// deadErr holds the failure that ended the reader; calls that arrive
	// after it fail at once instead of waiting out their deadline on a
	// connection nobody reads any more.
	deadErr error

	recMu         sync.Mutex
	recalled      []nfs4.Stateid
	notified      []nfs4.CBNotify
	notifiedLocks []nfs4.CBNotifyLock

	gssMu sync.Mutex
	gss   *gssSession

	// tlsProbe hands a TLS upgrade to the reader goroutine, the only
	// goroutine that may touch the raw wire: the probe exchange, the
	// token check and the handshake all run there.
	tlsProbe chan *tlsProbeReq
}

// A tlsProbeReq is one upgrade request handed to the reader.
type tlsProbeReq struct {
	cfg    *tls.Config
	result chan error
}

// timeout answers the call budget in force.
func (c *Client) timeout() time.Duration {
	if c.Timeout > 0 {
		return c.Timeout
	}
	return defaultTimeout
}

// SetIdentity sets the AUTH_SYS identity the client asserts on every
// subsequent call. The server trusts this claim the way AUTH_SYS does; the
// method exists to test the server as different owners, not to escape it.
func (c *Client) SetIdentity(uid, gid uint32, groups []uint32) {
	c.mu.Lock()
	c.uid, c.gid, c.groups = uid, gid, groups
	c.mu.Unlock()
}

// Dial connects to an NFS server and starts the reader that
// demultiplexes replies from the server's callback calls.
func Dial(addr string) (*Client, error) {
	c := &Client{
		pending:   make(map[uint32]chan wireResult),
		cbProgram: nfs4.CBDefaultProgram,
		next:      1, // zero is the "allocate" sentinel of rawCall
		tlsProbe:  make(chan *tlsProbeReq, 1),
	}
	conn, err := net.DialTimeout("tcp", addr, c.timeout())
	if err != nil {
		return nil, fmt.Errorf("nfsclient: dial %s: %w", addr, err)
	}
	c.conn = conn
	go c.reader()
	return c, nil
}

// SessionID reports the session the client established.
func (c *Client) SessionID() (nfs4.SessionID, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.sessionID == nil {
		return nfs4.SessionID{}, false
	}
	return *c.sessionID, true
}

// Notified lists the directory change notifications the server pushed.
func (c *Client) Notified() []nfs4.CBNotify {
	c.recMu.Lock()
	defer c.recMu.Unlock()
	return append([]nfs4.CBNotify(nil), c.notified...)
}

// NotifiedLocks lists the lock availability notifications the server
// pushed.
func (c *Client) NotifiedLocks() []nfs4.CBNotifyLock {
	c.recMu.Lock()
	defer c.recMu.Unlock()
	return append([]nfs4.CBNotifyLock(nil), c.notifiedLocks...)
}

// Recalled lists the delegation stateids the server recalled over the
// back channel since the client started.
func (c *Client) Recalled() []nfs4.Stateid {
	c.recMu.Lock()
	defer c.recMu.Unlock()
	return append([]nfs4.Stateid(nil), c.recalled...)
}

// reader reads records forever and routes them: a reply goes to the call
// that waits for it, a call is answered as a callback. A TLS upgrade
// request from EnableTLS runs here, because only this goroutine may
// touch the raw wire: it sends the probe, waits out the reply, checks
// the token and runs the handshake, routing anything else that arrives
// in between.
func (c *Client) reader() {
	for {
		select {
		case probe := <-c.tlsProbe:
			probe.result <- c.runTLSProbe(probe.cfg)
			continue
		default:
		}
		rec, err := rpc.ReadRecord(c.conn, maxRecord)
		if err != nil {
			// A deadline set by EnableTLS is the wake-up signal for a
			// queued probe, not a wire failure; at a record boundary it
			// leaves the stream in place.
			if errors.Is(err, os.ErrDeadlineExceeded) {
				continue
			}
			c.failPending(err)
			return
		}
		xid, mtype, err := rpc.PeekHeader(rec)
		if err != nil {
			continue
		}
		if mtype == rpc.MsgReply {
			c.route(xid, rec)
			continue
		}
		c.handleCBCall(xid, rec)
	}
}

// route delivers one reply record to its waiter, if it still waits.
func (c *Client) route(xid uint32, rec []byte) {
	c.pendingMu.Lock()
	ch, ok := c.pending[xid]
	if ok {
		delete(c.pending, xid)
	}
	c.pendingMu.Unlock()
	if ok {
		ch <- wireResult{rec: rec}
	}
}

// runTLSProbe performs the whole RPC-with-TLS upgrade of RFC 9289 on the
// reader goroutine: the AUTH_TLS probe, the STARTTLS token check and
// the handshake itself.
func (c *Client) runTLSProbe(cfg *tls.Config) error {
	c.mu.Lock()
	conn := c.conn
	c.mu.Unlock()
	// The wake-up deadline ends with the probe: reads run open ended
	// again from here on.
	_ = conn.SetReadDeadline(time.Time{})
	xid := c.nextXID()
	header, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: nfs4.ProcNull,
		Cred:      rpc.Auth{Flavor: rpc.FlavorTLS},
		Verifier:  rpc.Auth{Flavor: rpc.FlavorNone},
	})
	if err != nil {
		return err
	}
	if err := c.send(header); err != nil {
		return fmt.Errorf("nfsclient: write: %w", err)
	}
	var reply rpc.Reply
	for {
		rec, rerr := rpc.ReadRecord(c.conn, maxRecord)
		if rerr != nil {
			c.failPending(rerr)
			return rerr
		}
		rxid, mtype, perr := rpc.PeekHeader(rec)
		if perr != nil {
			continue
		}
		if mtype == rpc.MsgReply && rxid == xid {
			reply, err = rpc.DecodeReply(rec)
			break
		}
		if mtype == rpc.MsgReply {
			c.route(rxid, rec)
			continue
		}
		c.handleCBCall(rxid, rec)
	}
	if err != nil {
		return err
	}
	if reply.Rejected || reply.Status != rpc.AcceptSuccess ||
		reply.Verifier.Flavor != rpc.FlavorNone ||
		string(reply.Verifier.Body) != rpc.StarttlsToken {
		return fmt.Errorf("nfsclient: the server refused the tls upgrade")
	}
	return c.runTLS(cfg)
}

// runTLS performs the TLS handshake on the reader goroutine, which owns
// the wire: the handshake and the record reader would otherwise fight
// over the bytes of the raw connection.
func (c *Client) runTLS(cfg *tls.Config) error {
	c.mu.Lock()
	wire := c.conn
	c.mu.Unlock()
	tlsConn := tls.Client(wire, cfg)
	if err := tlsConn.SetDeadline(time.Now().Add(c.timeout())); err != nil {
		return err
	}
	if err := tlsConn.Handshake(); err != nil {
		return fmt.Errorf("nfsclient: tls handshake: %w", err)
	}
	if err := tlsConn.SetDeadline(time.Time{}); err != nil {
		return err
	}
	c.mu.Lock()
	c.conn = tlsConn
	c.mu.Unlock()
	return nil
}

// failPending wakes every caller waiting on a reply and records the
// failure for the calls that arrive after the reader is gone; they all
// report the connection failure that ended it.
func (c *Client) failPending(err error) {
	wrapped := fmt.Errorf("nfsclient: connection lost: %w", err)
	c.pendingMu.Lock()
	c.deadErr = wrapped
	for xid, ch := range c.pending {
		ch <- wireResult{err: wrapped}
		delete(c.pending, xid)
	}
	c.pendingMu.Unlock()
}

// dead reports the failure that ended the reader, if it already ended.
func (c *Client) dead() error {
	c.pendingMu.Lock()
	defer c.pendingMu.Unlock()
	return c.deadErr
}

// handleCBCall answers one server callback: it decodes the CB_COMPOUND,
// walks the operations and sends the result back over the same
// connection. CB_RECALL records the recalled delegation stateid.
func (c *Client) handleCBCall(xid uint32, rec []byte) {
	call, args, err := rpc.DecodeCall(rec)
	if err != nil {
		return
	}
	var reply []byte
	switch {
	case call.Program != c.cbProgram:
		reply, err = rpc.AppendAcceptedReply(nil, xid, rpc.AuthNull,
			rpc.AcceptProgUnavail, rpc.Mismatch{})
	case call.Procedure == 0: // CB NULL
		reply, err = rpc.AppendAcceptedReply(nil, xid, rpc.AuthNull,
			rpc.AcceptSuccess, rpc.Mismatch{})
	default:
		var body []byte
		body, err = c.cbCompound(xid, args)
		if err == nil {
			reply, err = rpc.AppendAcceptedReply(nil, xid, rpc.AuthNull,
				rpc.AcceptSuccess, rpc.Mismatch{})
			reply = append(reply, body...)
		}
	}
	if err != nil {
		return
	}
	c.mu.Lock()
	conn := c.conn
	c.mu.Unlock()
	c.writeMu.Lock()
	defer c.writeMu.Unlock()
	_ = rpc.WriteRecord(conn, reply)
}

// cbCompound decodes the CB_COMPOUND args, walks the operations and
// encodes the result. CB_RECALL is recorded for the caller; the answer
// bodies the protocol defines are carried.
func (c *Client) cbCompound(xid uint32, args []byte) ([]byte, error) {
	header, d, err := nfs4.DecodeCBCompoundArgs(args)
	if err != nil {
		return nil, err
	}
	if header.Minor != nfs4.MinorVersion {
		return nfs4.AppendCBCompoundRes(nil, nfs4.ErrMinorVersMismatch, header.Tag, nil), nil
	}
	var results [][]byte
	top := uint32(nfs4.ErrOK)
	for range header.OpCount {
		op, err := d.Uint32()
		if err != nil {
			return nil, err
		}
		var payload []byte
		switch op {
		case nfs4.OpCBSequence:
			a, err := nfs4.DecodeCBSequenceArgs(d)
			if err != nil {
				return nil, err
			}
			payload = nfs4.AppendCBSequenceRes(nil, a.SessionID, a.Sequence, a.Slot,
				a.HighestSlot)
		case nfs4.OpCBRecall:
			a, err := nfs4.DecodeCBRecallArgs(d)
			if err != nil {
				return nil, err
			}
			c.recMu.Lock()
			c.recalled = append(c.recalled, a.Stateid)
			c.recMu.Unlock()
		case nfs4.OpCBNotify:
			a, err := nfs4.DecodeCBNotifyArgs(d)
			if err != nil {
				return nil, err
			}
			c.recMu.Lock()
			c.notified = append(c.notified, a)
			c.recMu.Unlock()
		case nfs4.OpCBNotifyLock:
			a, err := nfs4.DecodeCBNotifyLockArgs(d)
			if err != nil {
				return nil, err
			}
			c.recMu.Lock()
			c.notifiedLocks = append(c.notifiedLocks, a)
			c.recMu.Unlock()
		default:
			payload = nil
			top = nfs4.ErrNotSupp
		}
		result := nfs4.AppendOpHeader(nil, op, top)
		if top == nfs4.ErrOK || (op == nfs4.OpCBRecall && top == nfs4.ErrDenied) {
			result = append(result, payload...)
		}
		results = append(results, result)
		if top != nfs4.ErrOK {
			break
		}
	}
	return nfs4.AppendCBCompoundRes(nil, top, header.Tag, results), nil
}

// send writes one request record, serialised against callback replies.
func (c *Client) send(record []byte) error {
	c.mu.Lock()
	conn := c.conn
	c.mu.Unlock()
	c.writeMu.Lock()
	defer c.writeMu.Unlock()
	return rpc.WriteRecord(conn, record)
}

// Close ends the connection.
func (c *Client) Close() error {
	c.mu.Lock()
	conn := c.conn
	c.mu.Unlock()
	return conn.Close()
}

// call sends one ONC RPC call. It returns the result body and the accept
// status of the reply; the body is meaningful only when the status is
// success, and an error means the wire or the reply failed outright. The
// reader goroutine routes the reply by xid, so callback calls from the
// server interleave freely. A reply the server rejected instead of
// accepting is an error, not a zero status.
func (c *Client) call(proc uint32, args []byte) ([]byte, uint32, error) {
	if dead := c.dead(); dead != nil {
		return nil, 0, dead
	}
	xid := c.nextXID()

	ch := make(chan wireResult, 1)
	c.pendingMu.Lock()
	c.pending[xid] = ch
	c.pendingMu.Unlock()

	cred, err := (rpc.AuthSys{Machine: "nfsclient", UID: c.uid, GID: c.gid, GIDs: c.groups}).Body()
	if err != nil {
		return nil, 0, err
	}
	header, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: proc,
		Cred:      rpc.Auth{Flavor: rpc.FlavorSys, Body: cred},
	})
	if err != nil {
		return nil, 0, err
	}
	if err := c.send(append(header, args...)); err != nil {
		return nil, 0, fmt.Errorf("nfsclient: write: %w", err)
	}

	wr, ok := c.await(ch, xid)
	if !ok {
		return nil, 0, fmt.Errorf("nfsclient: reply for xid %d timed out", xid)
	}
	if wr.err != nil {
		return nil, 0, wr.err
	}
	reply, err := rpc.DecodeReply(wr.rec)
	if err != nil {
		return nil, 0, err
	}
	if reply.XID != xid {
		return nil, 0, fmt.Errorf("nfsclient: reply xid %d, want %d", reply.XID, xid)
	}
	if reply.Rejected {
		return nil, 0, fmt.Errorf("nfsclient: call rejected, auth status %d", reply.AuthStat)
	}
	return reply.Body, reply.Status, nil
}

// await waits for one routed reply under the call timeout, and drops
// the pending registration when nothing came. A connection whose reader
// already died fails at once: nothing will ever route the reply. The
// boolean reports delivery.
func (c *Client) await(ch chan wireResult, xid uint32) (wireResult, bool) {
	select {
	case wr := <-ch:
		return wr, true
	case <-time.After(c.timeout()):
		c.pendingMu.Lock()
		delete(c.pending, xid)
		dead := c.deadErr
		c.pendingMu.Unlock()
		if dead != nil {
			return wireResult{err: dead}, true
		}
		return wireResult{}, false
	}
}

// Null pings the NULL procedure.
func (c *Client) Null() error {
	_, status, err := c.call(nfs4.ProcNull, nil)
	if err != nil {
		return err
	}
	if status != rpc.AcceptSuccess {
		return fmt.Errorf("nfsclient: rpc status %d", status)
	}
	return nil
}

// Procedure calls an arbitrary procedure number and reports the accept
// status the server answered with, for diagnostics against servers that
// carry more procedures than this client speaks.
func (c *Client) Procedure(proc uint32) (uint32, error) {
	_, status, err := c.call(proc, nil)
	return status, err
}

// A ResOp is one operation result of a COMPOUND.
type ResOp = nfs4.ResOp

// Compound sends one COMPOUND and returns the decoded result array with
// each operation's body. Inside a session the client prepends its
// SEQUENCE and strips its result again, so callers keep seeing the
// operations they passed.
func (c *Client) Compound(tag string, ops [][]byte) (nfs4.CompoundRes, [][]byte, error) {
	c.mu.Lock()
	inSession := c.sessionID != nil
	var slot uint32
	var seqArgs []byte
	if inSession {
		// One slot per compound in flight: taking a free slot caps the
		// concurrency, returning it lets the next compound start. The
		// sequence is per slot, the way the server tracks it.
		slot = <-c.slots
		seq := c.slotSeq[slot]
		c.slotSeq[slot] = seq + 1
		seqArgs = nfs4.AppendSequenceArgs(nil, *c.sessionID, seq, slot, 7, true)
	}
	c.mu.Unlock()
	if inSession {
		defer func() { c.mu.Lock(); c.slots <- slot; c.mu.Unlock() }()
		ops = append([][]byte{seqArgs}, ops...)
	}
	body := nfs4.AppendCompoundArgs(nil, tag, nfs4.MinorVersion, ops)
	var res []byte
	var status uint32
	var err error
	c.gssMu.Lock()
	g := c.gss
	c.gssMu.Unlock()
	if g != nil {
		res, status, err = c.callGSS(g, body)
	} else {
		res, status, err = c.call(nfs4.ProcCompound, body)
	}
	if err != nil {
		return nfs4.CompoundRes{}, nil, err
	}
	if status != rpc.AcceptSuccess {
		return nfs4.CompoundRes{}, nil, fmt.Errorf("nfsclient: rpc status %d", status)
	}
	decoded, bodies, err := nfs4.DecodeCompoundResBodies(res)
	if err != nil {
		return nfs4.CompoundRes{}, nil, err
	}
	if inSession {
		// The SEQUENCE result this client prepended comes back first; a
		// reply without it, or with nothing at all, is a server that
		// answers a different conversation.
		if len(decoded.Ops) == 0 || len(bodies) == 0 {
			return nfs4.CompoundRes{}, nil, fmt.Errorf("nfsclient: empty compound reply")
		}
		decoded.Ops = decoded.Ops[1:]
		bodies = bodies[1:]
	}
	return decoded, bodies, nil
}

// Establish runs EXCHANGE_ID and CREATE_SESSION against the server, after
// which every Compound travels inside the session: the client prepends its
// SEQUENCE and strips its result again, so callers keep seeing their own
// operations.
func (c *Client) Establish(owner string) error {
	var verifier [8]byte
	if _, err := crand.Read(verifier[:]); err != nil {
		return err
	}
	ex := nfs4.AppendExchangeIDArgs(nil, verifier, []byte(owner), 0)
	body, status, err := c.call(nfs4.ProcCompound,
		nfs4.AppendCompoundArgs(nil, "exchange", nfs4.MinorVersion, [][]byte{ex}))
	if err != nil {
		return err
	}
	if status != rpc.AcceptSuccess {
		return fmt.Errorf("nfsclient: exchange id rpc status %d", status)
	}
	res, _, err := nfs4.DecodeCompoundResBodies(body)
	if err != nil || res.Status != nfs4.ErrOK {
		return fmt.Errorf("nfsclient: exchange id status %d", res.Status)
	}
	if len(res.Ops) == 0 {
		return fmt.Errorf("nfsclient: empty exchange id reply")
	}
	clientid, err := xdr.NewDecoder(res.Ops[0].Body).Uint64()
	if err != nil {
		return err
	}

	cs := nfs4.AppendCreateSessionArgs(nil, clientid, 0, nfs4.CreateSessionFlagConnBackChan,
		nfs4.DefaultForeChannel, nfs4.DefaultBackChannel, 0x40000000)
	body, status, err = c.call(nfs4.ProcCompound,
		nfs4.AppendCompoundArgs(nil, "create", nfs4.MinorVersion, [][]byte{cs}))
	if err != nil {
		return err
	}
	if status != rpc.AcceptSuccess {
		return fmt.Errorf("nfsclient: create session rpc status %d", status)
	}
	res, _, err = nfs4.DecodeCompoundResBodies(body)
	if err != nil || res.Status != nfs4.ErrOK {
		return fmt.Errorf("nfsclient: create session status %d", res.Status)
	}
	if len(res.Ops) == 0 {
		return fmt.Errorf("nfsclient: empty create session reply")
	}
	var id nfs4.SessionID
	if len(res.Ops[0].Body) < len(id) {
		return fmt.Errorf("nfsclient: short session id")
	}
	copy(id[:], res.Ops[0].Body)

	c.mu.Lock()
	c.sessionID = &id
	for i := range c.slotSeq {
		c.slotSeq[i] = 1
	}
	c.slots = make(chan uint32, 8)
	c.slots <- 0
	c.mu.Unlock()
	return nil
}

// SetConcurrency sets how many session slots the client drives at once,
// one to eight: several compounds then run in flight over the one
// connection, each on its own slot with its own sequence. The default of
// one serialises the compounds exactly as before. Call it after
// Establish and before the parallel use begins; lowering takes back idle
// slots only.
func (c *Client) SetConcurrency(n int) error {
	if n < 1 || n > 8 {
		return fmt.Errorf("nfsclient: concurrency %d outside 1 to 8", n)
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.slots == nil {
		return fmt.Errorf("nfsclient: SetConcurrency before Establish")
	}
	for len(c.slots) < n {
		c.slots <- uint32(len(c.slots))
	}
	for len(c.slots) > n {
		<-c.slots
	}
	return nil
}

// A gssSession is the client half of an established RPCSEC_GSS
// context: the session key context, the handle the server assigned, the
// negotiated service level and the credential sequence counter.
type gssSession struct {
	ctx     *krb5.Context
	handle  []byte
	service uint32
	seq     uint32
	v3      bool
}

// EnableGSS establishes an RPCSEC_GSS context through RPCSEC_GSS_INIT
// and switches every further COMPOUND of this client to the service
// level, none, integrity or privacy. The service key is the long term
// nfs service key the server knows.
func (c *Client) EnableGSS(etype uint32, serverKey []byte, realm, service, client string, svc uint32) error {
	ctx, token, err := krb5.ClientInit(etype, serverKey, realm, service, client)
	if err != nil {
		return err
	}
	// The control procedures ride the NULL procedure with the context
	// token in the call data, RFC 2203 section 5.1.3.
	cred := rpc.AppendGSSCred(nil, rpc.GSSProcInit, 0, 0, nil)
	reply, _, status, err := c.rawCall(0, rpc.ProcedureNull, cred, nil, nil, token)
	if err != nil {
		return err
	}
	if status != rpc.AcceptSuccess {
		return fmt.Errorf("nfsclient: gss init status %d", status)
	}
	handle, major, _, _, aprep, err := rpc.DecodeGSSInitRes(reply)
	if err != nil {
		return err
	}
	if major != 0 || len(handle) == 0 {
		return fmt.Errorf("nfsclient: gss init major %d", major)
	}
	// The acceptor answer completes the context: the AP-REP must verify
	// under the session key before anything is signed with it.
	if err := ctx.ClientAcceptRep(aprep); err != nil {
		return fmt.Errorf("nfsclient: gss ap-rep: %w", err)
	}
	c.gssMu.Lock()
	c.gss = &gssSession{ctx: ctx, handle: handle, service: svc}
	c.gssMu.Unlock()
	return nil
}

// DisableGSS destroys the RPCSEC_GSS context and returns the client to
// AUTH_SYS. The destroy request carries a verifier MIC under the
// context it retires, as RFC 2203 section 5.2.3 requires.
func (c *Client) DisableGSS() error {
	c.gssMu.Lock()
	defer c.gssMu.Unlock()
	g := c.gss
	c.gss = nil
	if g == nil {
		return nil
	}
	g.seq++
	cred := rpc.AppendGSSCred(nil, rpc.GSSProcDestroy, g.seq, rpc.SvcIntegrity, g.handle)
	xid := c.nextXID()
	prefix, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: rpc.ProcedureNull,
		Cred:      rpc.Auth{Flavor: rpc.FlavorGSS, Body: cred},
	})
	if err != nil {
		return err
	}
	verf, err := g.ctx.GetMIC(prefix)
	if err != nil {
		return err
	}
	if _, _, status, err := c.rawCall(xid, rpc.ProcedureNull, cred, verf, nil, nil); err != nil {
		return err
	} else if status != rpc.AcceptSuccess {
		return fmt.Errorf("nfsclient: gss destroy status %d", status)
	}
	return nil
}

// rawCall sends one call with an explicit RPCSEC_GSS credential. The
// wrap and mic functions apply the service protection to the arguments;
// nil means the none service. The answer body and the accept status come
// back, the verifier and the protection of the results are the caller's
// business.
func (c *Client) rawCall(xid uint32, proc uint32, cred []byte, verf []byte, wrap func([]byte) ([]byte, error), body []byte) (replyBody []byte, replyVerf []byte, status uint32, err error) {
	if dead := c.dead(); dead != nil {
		return nil, nil, 0, dead
	}
	if xid == 0 {
		xid = c.nextXID()
	}

	ch := make(chan wireResult, 1)
	c.pendingMu.Lock()
	c.pending[xid] = ch
	c.pendingMu.Unlock()

	var wireArgs []byte
	if wrap != nil {
		protected, werr := wrap(body)
		if werr != nil {
			return nil, nil, 0, werr
		}
		wireArgs = protected
	} else {
		wireArgs = body
	}
	flavor := uint32(rpc.FlavorNone)
	if len(cred) > 0 {
		flavor = rpc.FlavorGSS
	}
	header, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: proc,
		Cred:      rpc.Auth{Flavor: flavor, Body: cred},
		Verifier:  rpc.Auth{Flavor: flavor, Body: verf},
	})
	if err != nil {
		return nil, nil, 0, err
	}
	if err := c.send(append(header, wireArgs...)); err != nil {
		return nil, nil, 0, fmt.Errorf("nfsclient: write: %w", err)
	}
	wr, ok := c.await(ch, xid)
	if !ok {
		return nil, nil, 0, fmt.Errorf("nfsclient: reply for xid %d timed out", xid)
	}
	if wr.err != nil {
		return nil, nil, 0, wr.err
	}
	reply, err := rpc.DecodeReply(wr.rec)
	if err != nil {
		return nil, nil, 0, err
	}
	if reply.XID != xid {
		return nil, nil, 0, fmt.Errorf("nfsclient: reply xid %d, want %d", reply.XID, xid)
	}
	if reply.Rejected {
		return nil, nil, 0, fmt.Errorf("nfsclient: call rejected, auth status %d", reply.AuthStat)
	}
	return reply.Body, reply.Verifier.Body, reply.Status, nil
}

// callGSS runs one COMPOUND over the established RPCSEC_GSS context at
// the negotiated service level: the verifier MIC covers the call header,
// the arguments and results are checksummed at integrity and sealed at
// privacy. The whole sequence counter arithmetic runs under gssMu, so
// concurrent calls draw distinct sequence numbers.
func (c *Client) callGSS(g *gssSession, args []byte) ([]byte, uint32, error) {
	c.gssMu.Lock()
	defer c.gssMu.Unlock()
	g.seq++
	var cred []byte
	if g.v3 {
		cred = rpc.AppendGSSv3Cred(nil, rpc.GSSProcData, g.seq, g.service, g.handle)
	} else {
		cred = rpc.AppendGSSCred(nil, rpc.GSSProcData, g.seq, g.service, g.handle)
	}

	// The verifier MIC comes first in the sequence order, then the
	// protection of the arguments.
	var wrap func([]byte) ([]byte, error)
	var body []byte
	xid := c.nextXID()
	prefix, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: nfs4.ProcCompound,
		Cred:      rpc.Auth{Flavor: rpc.FlavorGSS, Body: cred},
	})
	if err != nil {
		return nil, 0, err
	}
	verf, err := g.ctx.GetMIC(prefix)
	if err != nil {
		return nil, 0, err
	}
	switch g.service {
	case rpc.SvcPrivacy:
		wrap = g.ctx.Wrap
		body = args
	case rpc.SvcIntegrity:
		mic, merr := g.ctx.GetMIC(args)
		if merr != nil {
			return nil, 0, merr
		}
		body = append(append([]byte{}, args...), mic...)
	default:
		body = args
	}
	replyBody, replyVerf, status, err := c.rawCall(xid, nfs4.ProcCompound, cred, verf, wrap, body)
	if err != nil {
		return nil, 0, err
	}
	if status != rpc.AcceptSuccess {
		return nil, status, nil
	}
	// The reply verifier is a MIC over the reply header with an empty
	// verifier field; the integrity and privacy levels verify it before
	// the results are touched.
	if g.service != rpc.SvcNone {
		replyPrefix, perr := rpc.AppendAcceptedReply(nil, xid, rpc.Auth{}, rpc.AcceptSuccess, rpc.Mismatch{})
		if perr != nil {
			return nil, 0, perr
		}
		if verr := g.ctx.VerifyMIC(replyPrefix, replyVerf); verr != nil {
			return nil, 0, verr
		}
	}
	// At the integrity level the results carry their own MIC, at
	// privacy they are sealed with the header appended.
	switch g.service {
	case rpc.SvcPrivacy:
		opened, uerr := g.ctx.Unwrap(replyBody)
		if uerr != nil {
			return nil, 0, uerr
		}
		return opened, status, nil
	case rpc.SvcIntegrity:
		if len(replyBody) < 28 {
			return nil, 0, fmt.Errorf("nfsclient: gss results too short %d", len(replyBody))
		}
		results := replyBody[:len(replyBody)-28]
		mic := replyBody[len(replyBody)-28:]
		if mic[0] != 0x04 || mic[1] != 0x04 {

			return nil, 0, krb5.ErrToken
		}
		if err := g.ctx.VerifyMIC(results, mic); err != nil {
			return nil, 0, err
		}
		return results, status, nil
	default:
		return replyBody, status, nil
	}
}

// nextXID reserves the next request identifier, so a caller can compute
// a verifier over the whole call header before the call is sent.
func (c *Client) nextXID() uint32 {
	c.mu.Lock()
	defer c.mu.Unlock()
	xid := c.next
	c.next++
	return xid
}

// CreateGSSChild runs RPCSEC_GSS_CREATE over the established context:
// the assertions are bound to a new version three child handle that
// replaces the parent for every further COMPOUND, RFC 7861 section
// 2.7.1. The control message rides on NULLPROC at the integrity level.
func (c *Client) CreateGSSChild(assertions []rpc.Assertion) ([]byte, error) {
	c.gssMu.Lock()
	defer c.gssMu.Unlock()
	g := c.gss
	if g == nil {
		return nil, fmt.Errorf("nfsclient: no gss context")
	}
	g.seq++
	cred := rpc.AppendGSSv3Cred(nil, rpc.GSSProcCreate, g.seq, rpc.SvcIntegrity, g.handle)
	xid := c.nextXID()
	prefix, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: nfs4.ProcNull,
		Cred:      rpc.Auth{Flavor: rpc.FlavorGSS, Body: cred},
	})
	if err != nil {
		return nil, err
	}
	verf, err := g.ctx.GetMIC(prefix)
	if err != nil {
		return nil, err
	}
	callData := rpc.AppendCreateArgs(nil, nil, nil, assertions)
	mic, err := g.ctx.GetMIC(callData)
	if err != nil {
		return nil, err
	}
	replyBody, replyVerf, status, err := c.rawCall(xid, nfs4.ProcNull, cred, verf,
		nil, append(append([]byte{}, callData...), mic...))
	if err != nil {
		return nil, err
	}
	if status != rpc.AcceptSuccess {
		return nil, fmt.Errorf("nfsclient: gss create status %d", status)
	}
	replyPrefix, err := rpc.AppendAcceptedReply(nil, xid, rpc.Auth{}, rpc.AcceptSuccess, rpc.Mismatch{})
	if err != nil {
		return nil, err
	}
	if err := g.ctx.VerifyMIC(replyPrefix, replyVerf); err != nil {
		return nil, err
	}
	if len(replyBody) < 28 {
		return nil, fmt.Errorf("nfsclient: gss create results too short")
	}
	results := replyBody[:len(replyBody)-28]
	if err := g.ctx.VerifyMIC(results, replyBody[len(replyBody)-28:]); err != nil {
		return nil, err
	}
	child, _, _, _, err := rpc.DecodeCreateRes(results)
	if err != nil {
		return nil, err
	}
	g.handle = child
	g.v3 = true
	return child, nil
}

// EnableTLS probes the server for RPC-with-TLS support with the
// AUTH_TLS NULL procedure of RFC 9289 and, when the server answers with
// the STARTTLS token, upgrades this connection to TLS. Every further
// call rides the encrypted channel. The whole exchange runs on the
// reader goroutine, the only one that may touch the raw wire.
func (c *Client) EnableTLS(cfg *tls.Config) error {
	probe := &tlsProbeReq{cfg: cfg, result: make(chan error, 1)}
	select {
	case c.tlsProbe <- probe:
	default:
		return fmt.Errorf("nfsclient: another tls upgrade is in flight")
	}
	// Wake the reader out of its blocking read: the deadline error is
	// its signal to poll the probe queue. EnableTLS is a connection
	// level switch and must not race other calls.
	c.mu.Lock()
	conn := c.conn
	c.mu.Unlock()
	_ = conn.SetReadDeadline(time.Now())
	select {
	case err := <-probe.result:
		return err
	case <-time.After(c.timeout() + c.timeout()):
		// The probe and the handshake each wait out one budget, so the
		// caller allows two.
		return fmt.Errorf("nfsclient: tls upgrade timed out")
	}
}

// ListGSSAssertions queries the server for the assertion types it
// grants, RFC 7861 section 2.7.2. The query rides the established
// context at the integrity level.
func (c *Client) ListGSSAssertions() ([]uint32, error) {
	c.gssMu.Lock()
	defer c.gssMu.Unlock()
	g := c.gss
	if g == nil {
		return nil, fmt.Errorf("nfsclient: no gss context")
	}
	g.seq++
	cred := rpc.AppendGSSv3Cred(nil, rpc.GSSProcList, g.seq, rpc.SvcIntegrity, g.handle)
	xid := c.nextXID()
	prefix, err := rpc.AppendCall(nil, rpc.Call{
		XID:       xid,
		Program:   nfs4.Program,
		Version:   nfs4.Version,
		Procedure: nfs4.ProcNull,
		Cred:      rpc.Auth{Flavor: rpc.FlavorGSS, Body: cred},
	})
	if err != nil {
		return nil, err
	}
	verf, err := g.ctx.GetMIC(prefix)
	if err != nil {
		return nil, err
	}
	mic, err := g.ctx.GetMIC(nil)
	if err != nil {
		return nil, err
	}
	replyBody, replyVerf, status, err := c.rawCall(xid, nfs4.ProcNull, cred, verf,
		nil, append([]byte{}, mic...))
	if err != nil {
		return nil, err
	}
	if status != rpc.AcceptSuccess {
		return nil, fmt.Errorf("nfsclient: gss list status %d", status)
	}
	replyPrefix, err := rpc.AppendAcceptedReply(nil, xid, rpc.Auth{}, rpc.AcceptSuccess, rpc.Mismatch{})
	if err != nil {
		return nil, err
	}
	if err := g.ctx.VerifyMIC(replyPrefix, replyVerf); err != nil {
		return nil, err
	}
	if len(replyBody) < 28 {
		return nil, fmt.Errorf("nfsclient: gss list results too short")
	}
	results := replyBody[:len(replyBody)-28]
	if err := g.ctx.VerifyMIC(results, replyBody[len(replyBody)-28:]); err != nil {
		return nil, err
	}
	d := xdr.NewDecoder(results)
	n, err := d.Uint32()
	if err != nil {
		return nil, err
	}
	var types []uint32
	for range n {
		t, err := d.Uint32()
		if err != nil {
			return nil, err
		}
		types = append(types, t)
	}
	return types, nil
}

// GetXattrNamed reads one named attribute of the file the current
// compound addresses: OPENATTR opens the synthetic directory, LOOKUP
// finds the attribute and READ pulls its value, RFC 8881 section 5.3.
func (c *Client) GetXattrNamed(fh []byte, name string, max int) ([]byte, error) {
	res, bodies, err := c.Compound("xattr", [][]byte{
		nfs4.AppendPutfh(nil, fh),
		nfs4.AppendOpenattrArgs(nil, false),
		nfs4.AppendLookup(nil, name),
		nfs4.AppendReadArgs(nil, nfs4.Stateid{}, 0, uint32(max)),
	})
	if err != nil {
		return nil, err
	}
	if res.Status != nfs4.ErrOK {
		return nil, &NFSError{Status: res.Status}
	}
	if len(bodies) < 4 {
		return nil, fmt.Errorf("nfsclient: short xattr reply of %d results", len(bodies))
	}
	d := xdr.NewDecoder(bodies[3])
	if _, derr := d.Bool(); derr != nil { // eof flag
		return nil, derr
	}
	value, derr := d.VarOpaque()
	return value, derr
}

// SetXattrNamed writes one named attribute of the file the current
// compound addresses, creating it when missing.
func (c *Client) SetXattrNamed(fh []byte, name string, value []byte) error {
	res, _, err := c.Compound("xattr", [][]byte{
		nfs4.AppendPutfh(nil, fh),
		nfs4.AppendOpenattrArgs(nil, false),
		nfs4.AppendCreateArgs(nil, nfs4.NF4Reg, name, "", 0, 0, 0),
		nfs4.AppendSavefh(nil),
		nfs4.AppendWriteArgs(nil, nfs4.Stateid{}, 0, nfs4.StableFileSync, value),
	})
	if err != nil {
		return err
	}
	if res.Status != nfs4.ErrOK {
		return &NFSError{Status: res.Status}
	}
	return nil
}

// NFSError carries an NFS status as an error value.
type NFSError struct{ Status uint32 }

func (e *NFSError) Error() string {
	return fmt.Sprintf("nfsclient: nfs status %d", e.Status)
}