puressh 0.2.4

//! High-level synchronous SSH client over `std::net::TcpStream`.
//!
//! ```ignore
//! use puressh::client::{Client, Config};
//!
//! let mut c = Client::connect("example.com:22", Config::insecure())?;
//! c.authenticate_password("alice", "hunter2")?;
//! let out = c.exec("uname -a")?;
//! print!("{}", String::from_utf8_lossy(&out.stdout));
//! ```

#![cfg(feature = "std")]

use std::collections::{BTreeMap, VecDeque};
use std::io::{ErrorKind, Read, Write};
use std::net::{TcpStream, ToSocketAddrs};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, Sender, SyncSender, TryRecvError};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use purecrypto::hash::{Digest, Sha256};
use purecrypto::rng::RngCore;

use crate::auth::{ClientAuth, ClientCredential, ClientStep};
use crate::channel::{
    ChannelEvent, ChannelOpen, ChannelRequest, ConnectionState, SSH_EXTENDED_DATA_STDERR,
    SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
};
use crate::driver::{ClientDriver, Event};
use crate::error::{Error, Result};
use crate::hostkey::{HostKey, HostKeyVerify, host_key_verify_by_name};
use crate::known_hosts::{KnownHosts, LookupResult};
use crate::sftp::SftpClient;
pub use crate::stream::{ChannelEgress, ChannelStream};
use crate::transport::kex::{defaults, is_strict_kex_marker};
#[cfg(any(test, feature = "multichannel"))]
use crate::transport::ping::encode_ping;
use crate::transport::{KexAlgorithmsOwned, KexInit, KexRunner};

/// Abstraction over the byte transport a [`Client`] runs on. The default is
/// a plain [`TcpStream`], but a client can also run over a proxied channel
/// (ProxyJump's `direct-tcpip` stream) or a spawned helper process's pipes
/// (ProxyCommand) — anything that is `Read + Write + Send` and can toggle a
/// read timeout.
///
/// `set_read_timeout` mirrors [`TcpStream::set_read_timeout`]: `None` clears
/// it (blocking reads). Transports that cannot honour a timeout (e.g. a pipe
/// to a child process) may implement it as a no-op `Ok(())` — callers that
/// rely on a real timeout (the serve / forwarding poll loops) must not be
/// driven over such a transport.
pub trait Transport: Read + Write + Send {
    /// Set the read timeout on the underlying transport. See
    /// [`TcpStream::set_read_timeout`].
    fn set_read_timeout(&mut self, t: Option<Duration>) -> std::io::Result<()>;

    /// Set the write timeout on the underlying transport. Bounds how long a
    /// single socket write may block before returning `WouldBlock` /
    /// `TimedOut`, which the frontend treats as backpressure: the unsent
    /// bytes stay buffered and are retried by the next pump instead of
    /// wedging the shared transport lock across a stalled write. This is what
    /// prevents the bidirectional-saturation deadlock where both peers block
    /// in `write` and neither side ever reads.
    ///
    /// Transports that cannot honour a timeout (e.g. a pipe to a child
    /// process, or a channel running over another connection) leave the
    /// default no-op — they are never the multi-channel concurrent-pump case
    /// the timeout exists to protect.
    fn set_write_timeout(&mut self, _t: Option<Duration>) -> std::io::Result<()> {
        Ok(())
    }

    /// A second, independent handle on the read side of the transport,
    /// which [`crate::shared::SharedClient`] reads from *without* holding
    /// its connection lock, so writers are never stuck behind a blocking
    /// read. Once it is taken, every read goes through it and the transport
    /// itself is only written to. `None` (the default) falls back to timed
    /// reads under the lock.
    fn split_reader(&self) -> Option<Box<dyn TransportReader>> {
        None
    }
}

/// The read half handed out by [`Transport::split_reader`].
pub trait TransportReader: Send {
    /// Read what is available into `buf`, waiting at most `timeout` for it
    /// (`None`: until data or EOF). An expired wait is
    /// [`ErrorKind::WouldBlock`] or [`ErrorKind::TimedOut`]; `Ok(0)` is EOF.
    fn read_within(&mut self, buf: &mut [u8], timeout: Option<Duration>) -> std::io::Result<usize>;
}

impl Transport for TcpStream {
    fn set_read_timeout(&mut self, t: Option<Duration>) -> std::io::Result<()> {
        TcpStream::set_read_timeout(self, t)
    }
    fn set_write_timeout(&mut self, t: Option<Duration>) -> std::io::Result<()> {
        TcpStream::set_write_timeout(self, t)
    }
    fn split_reader(&self) -> Option<Box<dyn TransportReader>> {
        let stream = self.try_clone().ok()?;
        let applied = stream.read_timeout().ok();
        Some(Box::new(TcpReader { stream, applied }))
    }
}

/// [`TransportReader`] over a clone of a [`TcpStream`]. The read timeout is
/// a property of the socket, shared with the original, which is why the
/// owner stops setting it once the reader is split off.
struct TcpReader {
    stream: TcpStream,
    /// The timeout last applied to the socket, to skip redundant
    /// `setsockopt` calls; `None` when unknown.
    applied: Option<Option<Duration>>,
}

impl TransportReader for TcpReader {
    fn read_within(&mut self, buf: &mut [u8], timeout: Option<Duration>) -> std::io::Result<usize> {
        if self.applied != Some(timeout) {
            self.applied = None;
            self.stream.set_read_timeout(timeout)?;
            self.applied = Some(timeout);
        }
        self.stream.read(buf)
    }
}

/// The reader [`crate::shared::SharedClient`] splits off its transport,
/// shared between the pumper — which waits for input on it with the
/// connection lock released ([`Self::fill`]) — and the [`Client`], which
/// reads through it under the lock. Input the pumper read is stashed until
/// the client consumes it, so bytes always reach the driver in wire order.
#[cfg(feature = "multichannel")]
pub(crate) struct SplitReader {
    state: Mutex<SplitState>,
}

#[cfg(feature = "multichannel")]
struct SplitState {
    src: Box<dyn TransportReader>,
    pending: Vec<u8>,
    eof: bool,
}

#[cfg(feature = "multichannel")]
impl SplitReader {
    /// Unless input is already stashed, wait up to `timeout` for some and
    /// stash it. A timeout is not an error.
    pub(crate) fn fill(&self, timeout: Duration) -> std::io::Result<()> {
        let mut s = self.state.lock().map_err(|_| poisoned_reader())?;
        if !s.pending.is_empty() || s.eof {
            return Ok(());
        }
        let mut tmp = [0u8; 16 * 1024];
        match s.src.read_within(&mut tmp, Some(timeout)) {
            Ok(0) => s.eof = true,
            Ok(n) => s.pending.extend_from_slice(&tmp[..n]),
            Err(e)
                if matches!(
                    e.kind(),
                    ErrorKind::WouldBlock | ErrorKind::TimedOut | ErrorKind::Interrupted
                ) => {}
            Err(e) => return Err(e),
        }
        Ok(())
    }

    /// Stashed input, else (unless `stashed_only`, which reports
    /// [`ErrorKind::WouldBlock`] instead) a read waiting up to `timeout`.
    fn read(
        &self,
        buf: &mut [u8],
        timeout: Option<Duration>,
        stashed_only: bool,
    ) -> std::io::Result<usize> {
        let mut s = self.state.lock().map_err(|_| poisoned_reader())?;
        if !s.pending.is_empty() {
            let n = s.pending.len().min(buf.len());
            buf[..n].copy_from_slice(&s.pending[..n]);
            s.pending.drain(..n);
            return Ok(n);
        }
        if s.eof {
            return Ok(0);
        }
        if stashed_only {
            return Err(ErrorKind::WouldBlock.into());
        }
        s.src.read_within(buf, timeout)
    }
}

#[cfg(feature = "multichannel")]
fn poisoned_reader() -> std::io::Error {
    std::io::Error::new(ErrorKind::BrokenPipe, "transport reader mutex poisoned")
}

/// Number of banner lines the handshake will skim through before giving up
/// (used to bound the handshake pump's iteration count). The driver does the
/// actual banner-scanning and enforces its own byte caps.
const MAX_BANNER_LINES: usize = 32;
/// Hard cap on accumulated exec stdout+stderr.
const MAX_EXEC_OUTPUT: usize = 64 * 1024 * 1024;
/// Maximum iterations for the KEX driver loop.
const MAX_KEX_STEPS: usize = 32;
/// Maximum iterations for the userauth loop.
const MAX_AUTH_STEPS: usize = 64;
/// Maximum iterations for the exec drain loop.
const MAX_EXEC_ITER: usize = 1_000_000;
/// Cap on the per-channel egress mpsc inside [`Client::serve`]. Mirrors the
/// server's `SUBSYSTEM_EGRESS_BACKLOG`: small enough to bound memory, large
/// enough to absorb a few full-window writes before the handler thread has
/// to block on its `Write::write`.
const SERVE_EGRESS_BACKLOG: usize = 32;
/// Cap on the per-channel *ingress* mpsc inside [`Client::serve`] (chunks
/// handed to a handler thread but not yet read by it). Chunks beyond this
/// are parked dispatcher-side without crediting the SSH receive window, so
/// a fast peer is throttled by flow control instead of growing an unbounded
/// queue. See [`ServeRuntime`].
const SERVE_INGRESS_BACKLOG: usize = 32;
/// Cap on the stderr (extended-data) bytes a [`ClientChannelStream`] keeps
/// for [`ClientChannelStream::take_stderr`]. Nothing drains that buffer
/// while the channel is in use, so rather than withholding window credit
/// (which would stall the channel forever on a chatty stderr) the buffer
/// keeps only the most recent `STREAM_STDERR_CAP` bytes.
const STREAM_STDERR_CAP: usize = 1024 * 1024;
/// Outer-loop step cap for [`Client::serve`] — generous because each step is
/// either one packet or one 50 ms timeout tick.
const MAX_SERVE_STEPS: usize = 100_000_000;
/// Read timeout the serve loop installs on the socket while it has at least
/// one live channel runtime to drain. Reverts to blocking when idle.
const SERVE_POLL_INTERVAL: Duration = Duration::from_millis(50);

/// Error text used when a credential or feature is refused because the
/// session was established through a *changed* host key that
/// [`TofuAction::AcceptWithWarning`] (StrictHostKeyChecking=no) let
/// through. Mirrors OpenSSH, which in that situation disables password and
/// keyboard-interactive authentication plus agent / X11 / port forwarding
/// so a man-in-the-middle cannot harvest a password or piggyback on the
/// user's agent and displays.
pub const HOST_KEY_MISMATCH_DISABLED: &str = "password/keyboard-interactive authentication and \
     agent/X11/port forwarding are disabled because the host key changed and \
     StrictHostKeyChecking=no accepted it (man-in-the-middle protection)";

/// Drop the credentials OpenSSH refuses to use over a changed host key
/// (`password`, the re-promptable password closure, and
/// `keyboard-interactive`), warning on stderr for each one removed.
/// Public-key credentials (including agent-backed signers) are kept: a
/// signature never reveals the secret to an eavesdropper. Fails with
/// [`Error::Config`] if nothing usable remains.
pub(crate) fn filter_credentials_after_mismatch(
    credentials: Vec<ClientCredential>,
) -> Result<Vec<ClientCredential>> {
    let mut kept = Vec::with_capacity(credentials.len());
    for c in credentials {
        match c {
            ClientCredential::Password(_) | ClientCredential::PasswordPrompt(_) => {
                eprintln!(
                    "Password authentication is disabled to avoid man-in-the-middle attacks."
                );
            }
            ClientCredential::KeyboardInteractive(_) => {
                eprintln!(
                    "Keyboard-interactive authentication is disabled to avoid \
                     man-in-the-middle attacks."
                );
            }
            other => kept.push(other),
        }
    }
    if kept.iter().all(|c| matches!(c, ClientCredential::None)) {
        return Err(Error::Config(HOST_KEY_MISMATCH_DISABLED));
    }
    Ok(kept)
}

/// Is `payload` an outbound userauth packet that must not go out over a
/// changed host key? True for `SSH_MSG_USERAUTH_REQUEST` with method
/// `password` or `keyboard-interactive`, and for any
/// `SSH_MSG_USERAUTH_INFO_RESPONSE` (which carries keyboard-interactive
/// answers). This is the last-line gate in `run_auth`, covering
/// [`ClientAuth`] drivers the caller assembled themselves.
pub(crate) fn auth_packet_blocked_after_mismatch(payload: &[u8]) -> bool {
    use crate::auth::message::{SSH_MSG_USERAUTH_INFO_RESPONSE, SSH_MSG_USERAUTH_REQUEST};
    match payload.first() {
        Some(&SSH_MSG_USERAUTH_INFO_RESPONSE) => true,
        Some(&SSH_MSG_USERAUTH_REQUEST) => {
            let mut r = crate::format::Reader::new(&payload[1..]);
            let method = r
                .read_string()
                .and_then(|_user| r.read_string())
                .and_then(|_service| r.read_string());
            matches!(method, Ok(b"password") | Ok(b"keyboard-interactive"))
        }
        _ => false,
    }
}

/// The reply the connection layer owes the peer for an event that a
/// single-channel loop would otherwise ignore, if any:
///
/// - a peer-initiated `CHANNEL_OPEN` we have no handler for →
///   `CHANNEL_OPEN_FAILURE` (`SSH_OPEN_ADMINISTRATIVELY_PROHIBITED`), which
///   also drops the phantom [`ChannelState`](crate::channel::ChannelState)
///   `on_packet` allocated (RFC 4254 §5.1 requires a reply);
/// - an open the connection layer already refused
///   ([`ChannelEvent::OpenRejected`]) → its pre-built failure payload;
/// - a `GLOBAL_REQUEST` with `want_reply` (e.g. sshd's
///   `keepalive@openssh.com` probe) → `REQUEST_FAILURE`, without which a
///   `ClientAliveInterval` server drops long transfers.
///
/// Shared by every frontend (blocking, [`Client::serve`], `SharedClient`,
/// async, mio) so no dispatcher forgets one of these.
pub(crate) fn unsolicited_reply(
    conn: &mut ConnectionState,
    ev: &ChannelEvent,
) -> Result<Option<Vec<u8>>> {
    match ev {
        ChannelEvent::OpenRequest { channel, .. } => Ok(Some(conn.reject_open(
            *channel,
            SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
            "channel type not supported",
            "",
        )?)),
        ChannelEvent::OpenRejected { payload, .. } => Ok(Some(payload.clone())),
        ChannelEvent::GlobalRequest {
            want_reply: true, ..
        } => Ok(Some(conn.send_global_failure())),
        _ => Ok(None),
    }
}

/// Policy for accepting (or rejecting) a server's host key.
#[non_exhaustive]
pub enum HostKeyPolicy {
    /// Trust whatever the server presents — equivalent to OpenSSH's
    /// `StrictHostKeyChecking=no`. Insecure; do not use against untrusted peers.
    AcceptAny,
    /// Accept only if the server's host-key SHA-256 fingerprint (the raw 32
    /// bytes, exactly as `ssh-keygen -lf` reports them) matches.
    AcceptFingerprint([u8; 32]),
    /// Verify against an OpenSSH-format `known_hosts` store. Requires the
    /// host name + port to be threaded in via [`Client::connect_to_host`]
    /// (or [`Client::connect_via`]). Constructors that take a bare socket
    /// address (`Client::connect`) have no address-independent name to look
    /// up, so with this variant configured they **fail closed** — the
    /// handshake is refused with [`Error::Config`] rather than silently
    /// downgrading to `AcceptAny`.
    ///
    /// See [`KnownHostsPolicy`] for the knobs.
    KnownHosts(KnownHostsPolicy),
}

/// Configuration for [`HostKeyPolicy::KnownHosts`].
///
/// `store` is shared (Arc + Mutex) so the binary can keep a handle to
/// inspect / persist independently of the client; the verifier itself
/// locks it just long enough to look up and optionally append a TOFU
/// entry. `save_path` (if set) is the file rewritten when a TOFU accept
/// adds a new entry; absent it, accepts stay in memory only.
#[non_exhaustive]
pub struct KnownHostsPolicy {
    /// The in-memory store. Locked just long enough for lookup / add.
    pub store: Arc<Mutex<KnownHosts>>,
    /// File path the store is persisted to on TOFU-accept. If `None`,
    /// accepts stay in memory only (useful for tests).
    pub save_path: Option<PathBuf>,
    /// Hash new TOFU entries (OpenSSH's `HashKnownHosts yes`).
    pub hash_new: bool,
    /// What to do when the host is *unknown* (no entry matches).
    pub on_unknown: TofuAction,
    /// What to do when the host **is** known but the key does not match
    /// any stored entry. The OpenSSH-safe default (when constructed via
    /// [`KnownHostsPolicy::strict`]) is [`TofuAction::Reject`]; binaries
    /// honouring `StrictHostKeyChecking=no` can opt into
    /// [`TofuAction::AcceptWithWarning`] to keep insecure-but-loud
    /// parity with OpenSSH.
    pub on_mismatch: TofuAction,
    /// Read-only system files (`GlobalKnownHostsFile`, by default
    /// `/etc/ssh/ssh_known_hosts{,2}`), evaluated together with `store`
    /// as OpenSSH does: a key listed in any file is accepted, a host listed
    /// with other keys in any file is a mismatch. Never written.
    pub global: Vec<Arc<KnownHosts>>,
    /// `HostKeyAlias`: look the host key up (and record it) under this name,
    /// on the default port, instead of the connection's host and port.
    pub host_key_alias: Option<String>,
    /// `RevokedHostKeys`: host keys (and certificates) refused outright.
    pub revoked: Option<Arc<crate::known_hosts::RevokedHostKeys>>,
    /// `KnownHostsCommand`: given the server's key type and key blob, the
    /// `known_hosts` lines to trust for it (read with the global files), or
    /// `None` when the command failed.
    pub command: Option<Arc<KnownHostsCommandFn>>,
}

/// Callback type for [`KnownHostsPolicy::command`] — `(key_type,
/// key_blob) → known_hosts text`.
pub type KnownHostsCommandFn = dyn Fn(&str, &[u8]) -> Option<Vec<u8>> + Send + Sync;

impl KnownHostsPolicy {
    /// Build a policy with the OpenSSH-safe defaults: reject unknown
    /// hosts, reject mismatched keys.
    pub fn strict(store: Arc<Mutex<KnownHosts>>) -> Self {
        Self {
            store,
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::Reject,
            on_mismatch: TofuAction::Reject,
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        }
    }
}

/// See [`Client::session_binding`].
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct SessionBinding {
    /// The server's host-key blob from the first key exchange.
    pub host_key: Vec<u8>,
    /// The session identifier.
    pub session_id: Vec<u8>,
    /// The host's signature blob over the session identifier.
    pub signature: Vec<u8>,
}

/// Callback type for [`TofuAction::Prompt`] — `(host, port, key_type,
/// key_blob) → accept?`.
pub type TofuPromptFn = dyn Fn(&str, u16, &str, &[u8]) -> bool + Send + Sync;

/// Why a host key needs an interactive decision, passed in [`HostKeyPrompt`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum HostKeyChange {
    /// The host is not in `known_hosts` at all — a first-contact (TOFU) add.
    Unknown,
    /// The host **is** known but presented a key that matches no stored
    /// entry — an override of an existing trust record.
    Changed,
}

/// Full context handed to a [`TofuAction::PromptDetailed`] callback so it can
/// render its own host-key prompt (TUI/GUI/stdin) instead of relying on the
/// built-in stderr banner.
///
/// On [`HostKeyChange::Changed`], [`existing`](Self::existing) holds the
/// previously-trusted keys; [`existing_fingerprints`](Self::existing_fingerprints)
/// formats them as `SHA256:…` strings for an "old key" line.
#[non_exhaustive]
pub struct HostKeyPrompt<'a> {
    /// Whether this is a first-contact add or a changed-key override.
    pub change: HostKeyChange,
    /// Target host name (as passed to [`Client::connect_to_host`]).
    pub host: &'a str,
    /// Target port.
    pub port: u16,
    /// Key type the peer presented, e.g. `"ssh-ed25519"`.
    pub key_type: &'a str,
    /// Raw host-key blob the peer presented (SSH wire format).
    pub key_blob: &'a [u8],
    /// `SHA256:…` fingerprint of [`key_blob`](Self::key_blob).
    pub fingerprint: String,
    /// Previously-stored `(key_type, key_blob)` entries for this host.
    /// Empty when [`change`](Self::change) is [`HostKeyChange::Unknown`].
    pub existing: &'a [(String, Vec<u8>)],
}

impl HostKeyPrompt<'_> {
    /// `SHA256:…` fingerprints of the [`existing`](Self::existing) stored
    /// keys, for displaying the "old key" being replaced on an override.
    pub fn existing_fingerprints(&self) -> Vec<String> {
        self.existing
            .iter()
            .map(|(_, blob)| host_key_fingerprint(blob))
            .collect()
    }
}

/// Callback type for [`TofuAction::PromptDetailed`] — receives full
/// [`HostKeyPrompt`] context and returns `true` to accept.
pub type HostKeyPromptFn = dyn Fn(&HostKeyPrompt<'_>) -> bool + Send + Sync;

/// What to do when [`HostKeyPolicy::KnownHosts`] encounters an unknown
/// host or a key that doesn't match any stored entry.
///
/// `#[non_exhaustive]`: new decision variants may be added in future minor
/// releases, so downstream `match`es must include a wildcard arm.
#[non_exhaustive]
pub enum TofuAction {
    /// Refuse the connection — equivalent to OpenSSH's
    /// `StrictHostKeyChecking=yes` against an empty `known_hosts`.
    Reject,
    /// Accept silently and add the entry — equivalent to
    /// `StrictHostKeyChecking=accept-new`.
    Accept,
    /// Ask the user. See [`TofuPromptFn`] for the callback signature.
    Prompt(Arc<TofuPromptFn>),
    /// Accept the connection but emit a loud warning to stderr. This is
    /// the only variant that makes sense for `on_mismatch` outside of
    /// `Reject`: it mirrors `StrictHostKeyChecking=no` in OpenSSH, which
    /// proceeds anyway after printing the
    /// `WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!` banner.
    /// Used for `on_unknown` it's silent like `Accept` but is included
    /// here for symmetry.
    AcceptWithWarning,
    /// Like [`Prompt`](Self::Prompt), but the callback receives full
    /// [`HostKeyPrompt`] context (change kind, ready `SHA256:…` fingerprint,
    /// and the previously-stored keys on an override) and owns all
    /// user-facing display. When used for `on_mismatch`, the built-in
    /// stderr `REMOTE HOST IDENTIFICATION HAS CHANGED` banner is
    /// **suppressed** so the callback can render its own warning.
    ///
    /// On accept, the new key is added (and rotated in over any prior keys
    /// on a mismatch) exactly as for [`Prompt`](Self::Prompt).
    ///
    /// ```ignore
    /// use std::io::{self, Write};
    /// use std::sync::Arc;
    /// use puressh::client::{HostKeyChange, HostKeyPrompt, TofuAction};
    ///
    /// let cb: Arc<puressh::client::HostKeyPromptFn> = Arc::new(|p: &HostKeyPrompt| {
    ///     match p.change {
    ///         HostKeyChange::Unknown => {
    ///             println!("The authenticity of host '{}:{}' can't be established.", p.host, p.port);
    ///             println!("{} key fingerprint is {}.", p.key_type, p.fingerprint);
    ///         }
    ///         HostKeyChange::Changed => {
    ///             println!("WARNING: host key for {}:{} has CHANGED.", p.host, p.port);
    ///             for old in p.existing_fingerprints() {
    ///                 println!("  old: {old}");
    ///             }
    ///             println!("  new: {}", p.fingerprint);
    ///         }
    ///     }
    ///     print!("Accept? [y/N] ");
    ///     io::stdout().flush().ok();
    ///     let mut line = String::new();
    ///     io::stdin().read_line(&mut line).ok();
    ///     matches!(line.trim(), "y" | "yes")
    /// });
    /// let _ = TofuAction::PromptDetailed(cb);
    /// ```
    PromptDetailed(Arc<HostKeyPromptFn>),
}

/// Client configuration knobs.
///
/// **Note**: `Config` deliberately does not implement [`Default`]. The
/// only sane default for `host_key_policy` would be
/// [`HostKeyPolicy::AcceptAny`], which is insecure (any peer with the
/// right port is trusted). Forcing callers to spell the policy out makes
/// the trust decision explicit at the call site — see [`Config::insecure`]
/// for an explicit opt-in equivalent of the old default, or
/// [`Config::with_known_hosts`] for the OpenSSH-style strict policy.
#[non_exhaustive]
pub struct Config {
    /// How to decide whether a server's host key is acceptable.
    pub host_key_policy: HostKeyPolicy,
    /// Optional per-operation socket timeout.
    pub timeout: Option<Duration>,
    /// Optional overrides for the advertised crypto-algorithm preference
    /// lists, populated from `ssh_config` (`Ciphers`, `MACs`,
    /// `KexAlgorithms`, `HostKeyAlgorithms`, `PubkeyAcceptedAlgorithms`).
    /// `None` everywhere ⇒ use the built-in defaults.
    pub algorithms: AlgoOverrides,
    /// Host key the server must present, as an SSH wire blob, checked in
    /// addition to `host_key_policy`. Set by a knock rendezvous
    /// ([`KnockAccepted::pin`](crate::knock::KnockAccepted::pin)) so the
    /// session runs over the key that signed the ACCEPT; it also narrows the
    /// advertised host-key algorithms to that key's. `None` ⇒ no pin.
    pub pinned_host_key: Option<Vec<u8>>,
}

/// Per-connection overrides for the advertised algorithm preference lists.
///
/// Each `None` field falls back to the built-in default for that category.
/// The strict-kex signalling markers are NOT represented here — they are
/// re-appended unconditionally by the KEXINIT builder after override
/// resolution, so a `KexAlgorithms` override can never disable the Terrapin
/// (CVE-2023-48795) mitigation.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct AlgoOverrides {
    /// `Ciphers` — applied to both directions.
    pub ciphers: Option<Vec<String>>,
    /// `MACs` — applied to both directions.
    pub macs: Option<Vec<String>>,
    /// `KexAlgorithms` — real (non-marker) kex methods.
    pub kex_algorithms: Option<Vec<String>>,
    /// `HostKeyAlgorithms` — server host-key algorithms we will accept.
    pub host_key_algorithms: Option<Vec<String>>,
    /// `PubkeyAcceptedAlgorithms` — signature algorithms for our own
    /// publickey credentials. Consumed by the userauth driver, not KEXINIT.
    pub pubkey_accepted_algorithms: Option<Vec<String>>,
    /// `CASignatureAlgorithms` — signature algorithms accepted from a CA when
    /// verifying a host certificate. `None` ⇒ built-in default
    /// (`config::algos::CA_SIGNATURE_DEFAULTS`).
    pub ca_signature_algorithms: Option<Vec<String>>,
    /// `Compression` — `Some(true)` advertises `zlib@openssh.com` ahead of
    /// `none` in both directions; `Some(false)` / `None` advertises `none`
    /// only. Honouring `Some(true)` requires the `compress` feature; the
    /// `ssh` binary rejects the keyword up front when the feature is absent,
    /// and the builder degrades to `none` defensively if it somehow slips
    /// through.
    pub compression: Option<bool>,
}

impl Config {
    /// Build a config with an explicit host-key policy, no socket timeout and
    /// the built-in algorithm preference lists. Adjust `timeout` /
    /// `algorithms` by field assignment afterwards; the struct is
    /// `#[non_exhaustive]` so it cannot be built with a struct literal from
    /// outside the crate.
    pub fn new(host_key_policy: HostKeyPolicy) -> Self {
        Self {
            host_key_policy,
            timeout: None,
            algorithms: AlgoOverrides::default(),
            pinned_host_key: None,
        }
    }

    /// Set the per-operation socket timeout (chainable).
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = Some(timeout);
        self
    }

    /// Replace the algorithm preference overrides (chainable).
    pub fn with_algorithms(mut self, algorithms: AlgoOverrides) -> Self {
        self.algorithms = algorithms;
        self
    }

    /// Explicit, audit-friendly constructor for "trust whatever the peer
    /// presents" — the old behaviour of `Config::default()`. Replaces
    /// the removed `Default` impl so the trust decision shows up in
    /// `git grep` for `Config::insecure`.
    pub fn insecure() -> Self {
        Self {
            host_key_policy: HostKeyPolicy::AcceptAny,
            timeout: None,
            algorithms: AlgoOverrides::default(),
            pinned_host_key: None,
        }
    }

    /// Build a config that verifies the host key against `store` with
    /// OpenSSH-style strict semantics (reject unknown, reject mismatch).
    /// The caller is still responsible for routing connects through
    /// [`Client::connect_to_host`] so the verifier has a host name to
    /// look up.
    pub fn with_known_hosts(store: Arc<Mutex<KnownHosts>>) -> Self {
        Self {
            host_key_policy: HostKeyPolicy::KnownHosts(KnownHostsPolicy::strict(store)),
            timeout: None,
            algorithms: AlgoOverrides::default(),
            pinned_host_key: None,
        }
    }
}

/// Result of running `exec`.
#[non_exhaustive]
pub struct ExecOutput {
    /// Captured stdout bytes.
    pub stdout: Vec<u8>,
    /// Captured stderr bytes (extended-data channel, code 1).
    pub stderr: Vec<u8>,
    /// Process exit status (POSIX exit code), if the server sent `exit-status`.
    pub exit_status: Option<u32>,
    /// Signal name (no `SIG` prefix), if the server sent `exit-signal`.
    pub exit_signal: Option<String>,
}

/// Origin info accompanying a server-initiated `forwarded-tcpip` channel
/// (RFC 4254 §7.2). Passed to [`ClientHandlers::on_forwarded_tcpip`] so the
/// callback can decide which local destination to splice the channel to.
///
/// The `bound_*` fields echo the address+port the server is listening on
/// (per a previous `tcpip-forward` global request); `orig_*` are the
/// remote-side socket coordinates of the peer that just connected to it.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ForwardedTcpipOrigin {
    /// Address the server is listening on (typically what the client passed
    /// to [`Client::request_tcpip_forward`]).
    pub bound_address: String,
    /// Port the server is listening on.
    pub bound_port: u16,
    /// Originating peer's address (the side that just connected to the
    /// server's listener).
    pub orig_address: String,
    /// Originating peer's port.
    pub orig_port: u16,
}

/// Type alias for the [`ClientHandlers::on_forwarded_tcpip`] callback.
///
/// The handler takes ownership of a fresh [`ChannelStream`] wired to the
/// incoming channel. When it returns (or the stream drops), the channel is
/// torn down by the serve loop. Implementations typically spawn their own
/// thread or splice the stream against a local TCP destination — see
/// [`crate::forwarding::reverse`] for the server-side counterpart.
pub type ForwardedTcpipCallback =
    dyn Fn(ForwardedTcpipOrigin, ChannelStream) + Send + Sync + 'static;

/// Origin info accompanying a server-initiated `forwarded-streamlocal@openssh.com`
/// channel (OpenSSH extension; the Unix-socket analog of
/// [`ForwardedTcpipOrigin`]). Passed to
/// [`ClientHandlers::on_forwarded_streamlocal`] so the callback knows which
/// bound socket the connection arrived on.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ForwardedStreamlocalOrigin {
    /// Path the server is listening on (typically what the client passed to
    /// [`Client::request_streamlocal_forward`]).
    pub socket_path: String,
}

/// Type alias for the [`ClientHandlers::on_forwarded_streamlocal`] callback.
///
/// The handler takes ownership of a fresh [`ChannelStream`] wired to the
/// incoming channel. When it returns (or the stream drops), the channel is
/// torn down by the serve loop. Implementations typically splice the stream
/// against a local Unix socket — see
/// [`crate::forwarding::streamlocal::splice_to_unix_socket_callback`].
pub type ForwardedStreamlocalCallback =
    dyn Fn(ForwardedStreamlocalOrigin, ChannelStream) + Send + Sync + 'static;

/// Shows the server's pre-authentication banner
/// ([`Client::set_banner_handler`]).
type BannerHandler = Box<dyn Fn(&str) + Send + Sync>;

/// See [`Client::set_global_request_hook`].
type GlobalRequestHook = Arc<dyn Fn(&str, &[u8]) + Send + Sync>;

/// Type alias for the [`ClientHandlers::on_auth_agent`] callback.
///
/// Fires on every server-initiated `auth-agent@openssh.com` channel — one
/// per peer-side `SSH_AUTH_SOCK` connection. The handler typically splices
/// the stream against the local `$SSH_AUTH_SOCK` Unix socket, completing
/// the agent-forwarding round-trip (OpenSSH's `ssh -A`).
pub type AuthAgentCallback = dyn Fn(ChannelStream) + Send + Sync + 'static;

/// Type alias for the [`ClientHandlers::on_x11`] callback.
///
/// Fires on every server-initiated `x11` channel — one per peer-side X11
/// client connection that landed on the forwarded display. The handler
/// typically splices the stream against the local `$DISPLAY` (Unix domain
/// `/tmp/.X11-unix/X<N>` or TCP `host:6000+N`), completing the X11
/// forwarding round-trip (OpenSSH's `ssh -X` / `ssh -Y`).
pub type X11Callback = dyn Fn(ChannelStream) + Send + Sync + 'static;

/// Set of callbacks driving [`Client::serve`].
///
/// Each `on_*` field accepts a peer-initiated channel-open of the matching
/// type; unset callbacks make the serve loop reject opens of that type with
/// `SSH_OPEN_ADMINISTRATIVELY_PROHIBITED`. Setting [`Self::stop`] to `true`
/// asks the loop to exit at its next opportunity, after waiting for any
/// channels still active to drain naturally.
pub struct ClientHandlers {
    /// Callback for `"forwarded-tcpip"` channel opens (RFC 4254 §7.2, the
    /// inbound bookend of `ssh -R`). `None` ⇒ reject.
    pub on_forwarded_tcpip: Option<Arc<ForwardedTcpipCallback>>,
    /// Callback for `"forwarded-streamlocal@openssh.com"` channel opens
    /// (OpenSSH extension, the inbound bookend of `ssh -R /remote.sock:...`).
    /// `None` ⇒ reject.
    pub on_forwarded_streamlocal: Option<Arc<ForwardedStreamlocalCallback>>,
    /// Callback for `"auth-agent@openssh.com"` channel opens (server-side
    /// half of agent forwarding, `ssh -A`). `None` ⇒ reject.
    pub on_auth_agent: Option<Arc<AuthAgentCallback>>,
    /// Callback for `"x11"` channel opens (server-side half of X11
    /// forwarding, `ssh -X` / `ssh -Y`). `None` ⇒ reject.
    pub on_x11: Option<Arc<X11Callback>>,
    /// Cooperative stop signal. The loop polls this flag once per tick and
    /// returns `Ok(())` as soon as it's `true` AND no channels are open.
    pub stop: Arc<AtomicBool>,
    /// Inbound command channel — external threads (e.g. an `ssh -L`
    /// listener) push [`ServeCommand`]s here to ask the serve loop to open
    /// outbound channels. `None` when the loop has no associated
    /// [`ServeContext`]; the loop simply doesn't poll for outbound opens.
    cmd_rx: Option<Receiver<ServeCommand>>,
}

impl Default for ClientHandlers {
    fn default() -> Self {
        Self::new()
    }
}

impl ClientHandlers {
    /// Build an empty handler set with all callbacks unset.
    pub fn new() -> Self {
        Self {
            on_forwarded_tcpip: None,
            on_forwarded_streamlocal: None,
            on_auth_agent: None,
            on_x11: None,
            stop: Arc::new(AtomicBool::new(false)),
            cmd_rx: None,
        }
    }

    /// Install a `forwarded-tcpip` handler (`ssh -R` server-initiated opens).
    pub fn with_forwarded_tcpip(mut self, cb: Arc<ForwardedTcpipCallback>) -> Self {
        self.on_forwarded_tcpip = Some(cb);
        self
    }

    /// Install a `forwarded-streamlocal@openssh.com` handler
    /// (`ssh -R /remote.sock:...` server-initiated opens). For the default
    /// behaviour (splice each incoming channel against a local Unix socket),
    /// see [`crate::forwarding::streamlocal::splice_to_unix_socket_callback`].
    pub fn with_forwarded_streamlocal(mut self, cb: Arc<ForwardedStreamlocalCallback>) -> Self {
        self.on_forwarded_streamlocal = Some(cb);
        self
    }

    /// Install an `auth-agent@openssh.com` handler (`ssh -A` server-initiated
    /// opens). For the default behaviour (bind each incoming channel's agent
    /// connection to the session and splice it against the local
    /// `$SSH_AUTH_SOCK`), see
    /// [`crate::forwarding::agent::bound_splice_to_local_agent_callback`].
    pub fn with_auth_agent(mut self, cb: Arc<AuthAgentCallback>) -> Self {
        self.on_auth_agent = Some(cb);
        self
    }

    /// Install an `x11` handler (`ssh -X` / `ssh -Y` server-initiated opens).
    /// For the default behaviour (splice each incoming channel against the
    /// local `$DISPLAY`), see
    /// [`crate::forwarding::x11::splice_to_local_display_callback`].
    pub fn with_x11(mut self, cb: Arc<X11Callback>) -> Self {
        self.on_x11 = Some(cb);
        self
    }

    /// Attach an outbound-command receiver and hand back a paired
    /// [`ServeContext`] that external threads can clone to request
    /// `direct-tcpip` opens against the running serve loop. Used by
    /// `ssh -L`'s per-listener accept thread.
    pub fn with_serve_context(mut self) -> (Self, ServeContext) {
        let (tx, rx) = mpsc::channel();
        self.cmd_rx = Some(rx);
        (self, ServeContext { cmd_tx: tx })
    }
}

/// Command an external thread sends to [`Client::serve`] via a
/// [`ServeContext`]. Currently the only variant is outbound `direct-tcpip`
/// open; future ones (e.g. agent-channel open for Phase 8) plug in the same
/// way.
#[non_exhaustive]
pub enum ServeCommand {
    /// Open a `direct-tcpip` channel; on confirmation, send the resulting
    /// [`ChannelStream`] back through `reply`. On peer rejection, send
    /// `Err`.
    OpenDirectTcpip {
        /// Where the *server* should connect to (the `dest_host:dest_port`
        /// pair on the wire). For `ssh -L LPORT:RHOST:RPORT`, this is
        /// `RHOST:RPORT`.
        dest_host: String,
        /// Destination port the server should dial.
        dest_port: u16,
        /// Informational source-address echoed in the open. For
        /// `ssh -L`-style use, the client's accept address (e.g.
        /// `127.0.0.1`).
        orig_host: String,
        /// Informational source-port echoed in the open.
        orig_port: u16,
        /// Where to deliver the open's outcome — `Ok(stream)` on
        /// `OpenConfirmed`, `Err` on `OpenFailed` or loop teardown.
        reply: mpsc::SyncSender<Result<ChannelStream>>,
    },
    /// Open a `direct-streamlocal@openssh.com` channel; on confirmation, send
    /// the resulting [`ChannelStream`] back through `reply`. On peer
    /// rejection, send `Err`.
    OpenDirectStreamlocal {
        /// Unix-socket path the *server* should connect to.
        socket_path: String,
        /// Where to deliver the open's outcome.
        reply: mpsc::SyncSender<Result<ChannelStream>>,
    },
}

/// Handle external threads use to drive the running [`Client::serve`] loop.
///
/// Returned by [`ClientHandlers::with_serve_context`] and cloned freely
/// across the listener / accept threads that need to open outbound
/// `direct-tcpip` channels (`ssh -L`).
#[derive(Clone)]
pub struct ServeContext {
    cmd_tx: Sender<ServeCommand>,
}

impl ServeContext {
    /// Request a `direct-tcpip` channel from inside a running serve loop
    /// and block until the peer either confirms or rejects. Mirrors
    /// [`Client::open_direct_tcpip`] but works while `serve` is driving
    /// the socket — there's no other way to interleave the open with
    /// inbound traffic on the same client.
    ///
    /// Returns `Err(Error::Protocol(_))` if the serve loop has already
    /// returned (the receiver hung up).
    pub fn open_direct_tcpip(
        &self,
        dest_host: &str,
        dest_port: u16,
        orig_host: &str,
        orig_port: u16,
    ) -> Result<ChannelStream> {
        let (reply_tx, reply_rx) = mpsc::sync_channel::<Result<ChannelStream>>(1);
        self.cmd_tx
            .send(ServeCommand::OpenDirectTcpip {
                dest_host: dest_host.to_string(),
                dest_port,
                orig_host: orig_host.to_string(),
                orig_port,
                reply: reply_tx,
            })
            .map_err(|_| Error::Protocol("serve loop terminated"))?;
        reply_rx
            .recv()
            .map_err(|_| Error::Protocol("serve loop terminated"))?
    }

    /// Request a `direct-streamlocal@openssh.com` channel from inside a
    /// running serve loop and block until the peer confirms or rejects.
    /// Mirrors [`Self::open_direct_tcpip`] but for a Unix-socket destination
    /// (the wire side of `ssh -L local:/remote.sock`).
    ///
    /// Returns `Err(Error::Protocol(_))` if the serve loop has already
    /// returned (the receiver hung up).
    pub fn open_direct_streamlocal(&self, socket_path: &str) -> Result<ChannelStream> {
        let (reply_tx, reply_rx) = mpsc::sync_channel::<Result<ChannelStream>>(1);
        self.cmd_tx
            .send(ServeCommand::OpenDirectStreamlocal {
                socket_path: socket_path.to_string(),
                reply: reply_tx,
            })
            .map_err(|_| Error::Protocol("serve loop terminated"))?;
        reply_rx
            .recv()
            .map_err(|_| Error::Protocol("serve loop terminated"))?
    }
}

/// Per-channel state for an outbound channel open we've asked the serve
/// loop to drive: waiting for `OpenConfirmed` / `OpenFailed`.
struct PendingOutboundOpen {
    /// Pre-built stream handed to the caller on `OpenConfirmed`.
    stream: Option<ChannelStream>,
    /// Pre-built ingress sender; moves into the [`ServeRuntime`] on
    /// `OpenConfirmed`.
    ingress_tx: SyncSender<Option<Vec<u8>>>,
    /// Pre-built egress receiver; moves into the [`ServeRuntime`] on
    /// `OpenConfirmed`.
    egress_rx: Option<Receiver<ChannelEgress>>,
    /// Where to deliver the final result.
    reply: mpsc::SyncSender<Result<ChannelStream>>,
}

/// Per-channel state for an in-process handler running underneath
/// [`Client::serve`]. Parallel to the server's `SubsystemRuntime` —
/// dispatcher-side mailbox for one open channel.
///
/// # Inbound memory bound
///
/// Peer bytes are handed to the handler thread through a **bounded**
/// `sync_channel` of [`SERVE_INGRESS_BACKLOG`] chunks. The dispatcher never
/// blocks on it: chunks that don't fit are parked on `ingress_backlog` and
/// retried every tick. Crucially, the SSH receive window is credited back
/// to the peer only for chunks that actually reached the handler's queue —
/// bytes sitting in `ingress_backlog` stay uncredited, so a peer that
/// streams faster than the handler drains runs out of window and stalls.
/// Per channel that caps inbound memory at roughly one receive window
/// (backlog) plus `SERVE_INGRESS_BACKLOG` × max-packet (queued) plus one
/// chunk (the handler's own buffer), whatever the peer does.
struct ServeRuntime {
    /// Push peer-sent bytes (or `None` for EOF) to the handler thread.
    /// Bounded; see the type-level note. Only ever used with `try_send`.
    ingress_tx: SyncSender<Option<Vec<u8>>>,
    /// Inbound chunks (and a trailing `None` EOF marker) that did not fit
    /// in `ingress_tx` yet. Window credit for these is withheld until they
    /// are delivered, so their total size never exceeds the receive window.
    ingress_backlog: VecDeque<Option<Vec<u8>>>,
    /// Drain bytes the handler wants to ship out.
    egress_rx: Receiver<ChannelEgress>,
    /// Egress data that didn't fit in the remote window on the last tick;
    /// re-tried before pulling more from `egress_rx`.
    pending_data: Vec<u8>,
    /// EOF has been pulled from the handler but not yet emitted on the wire.
    pending_eof: bool,
    /// Close has been pulled from the handler but not yet emitted on the wire.
    pending_close: bool,
    /// Whether we've already sent `CHANNEL_EOF` on the wire.
    eof_sent: bool,
    /// Whether we've already sent `CHANNEL_CLOSE` on the wire.
    close_sent: bool,
}

impl ServeRuntime {
    fn new(ingress_tx: SyncSender<Option<Vec<u8>>>, egress_rx: Receiver<ChannelEgress>) -> Self {
        Self {
            ingress_tx,
            ingress_backlog: VecDeque::new(),
            egress_rx,
            pending_data: Vec::new(),
            pending_eof: false,
            pending_close: false,
            eof_sent: false,
            close_sent: false,
        }
    }

    /// Hand `item` to the handler thread if its queue has room, otherwise
    /// park it. Returns the number of payload bytes that were **delivered**
    /// (and may therefore be credited back to the peer's window): `0` when
    /// the item was parked, when it was an EOF marker, or when the handler
    /// is gone (its backlog is discarded — the channel is on its way to
    /// Close and no further credit is owed).
    fn offer_ingress(&mut self, item: Option<Vec<u8>>) -> usize {
        if !self.ingress_backlog.is_empty() {
            self.ingress_backlog.push_back(item);
            return 0;
        }
        let len = item.as_ref().map_or(0, Vec::len);
        match self.ingress_tx.try_send(item) {
            Ok(()) => len,
            Err(mpsc::TrySendError::Full(item)) => {
                self.ingress_backlog.push_back(item);
                0
            }
            Err(mpsc::TrySendError::Disconnected(_)) => {
                self.ingress_backlog.clear();
                0
            }
        }
    }

    /// Retry parked ingress. Returns the number of payload bytes delivered
    /// this call, for window credit.
    fn flush_ingress(&mut self) -> usize {
        let mut delivered = 0usize;
        while let Some(item) = self.ingress_backlog.pop_front() {
            let len = item.as_ref().map_or(0, Vec::len);
            match self.ingress_tx.try_send(item) {
                Ok(()) => delivered += len,
                Err(mpsc::TrySendError::Full(item)) => {
                    self.ingress_backlog.push_front(item);
                    break;
                }
                Err(mpsc::TrySendError::Disconnected(_)) => {
                    self.ingress_backlog.clear();
                    break;
                }
            }
        }
        delivered
    }
}

/// A blocking SSH client.
pub struct Client {
    stream: Box<dyn Transport>,
    /// Set once a [`crate::shared::SharedClient`] split the read side off
    /// `stream` ([`Self::split_reader`]); all reads then go through it.
    #[cfg(feature = "multichannel")]
    split_reader: Option<Arc<SplitReader>>,
    /// The read timeout last asked for, which reads through
    /// [`Self::split_reader`] apply themselves.
    #[cfg(feature = "multichannel")]
    read_timeout: Option<Duration>,
    /// Encoded transport bytes the driver has produced but the socket has not
    /// yet accepted. Normally drained to empty inside [`Self::pump_out`]; only
    /// holds bytes across calls when the socket's send buffer is full and a
    /// write timeout is configured, in which case [`Self::flush_out_buf`]
    /// reports backpressure rather than blocking the shared transport lock.
    /// Bounded in practice by SSH per-channel send windows (we stop encoding
    /// `CHANNEL_DATA` once the peer's window is exhausted).
    out_buf: VecDeque<u8>,
    /// Connection multiplexer (channel bookkeeping). The frontend owns this:
    /// the sans-IO [`ClientDriver`] surfaces decoded application payloads
    /// (post-auth), which we feed into `conn`, and we hand `conn`'s
    /// channel-protocol output back to the driver to encode and queue.
    pub(crate) conn: ConnectionState,
    /// Transport + authentication engine: version exchange, key exchange,
    /// re-key, EXT_INFO/PING handling, and the userauth state machine, all
    /// sans-IO. The frontend pumps it via `write_payload` / `read_one_packet`.
    driver: ClientDriver,
    /// Algorithm-preference overrides from config. Retained for the auth
    /// driver's PubkeyAcceptedAlgorithms / server-sig-algs policy.
    algo_overrides: AlgoOverrides,
    /// When `true`, every session-channel helper ([`Self::exec`],
    /// [`Self::exec_stream`], [`Self::shell_with_stdin`], [`Self::sftp`])
    /// issues `auth-agent-req@openssh.com` immediately after the open and
    /// before the matching shell/exec/subsystem request. Toggle with
    /// [`Self::set_request_auth_agent_forwarding`].
    pub(crate) request_auth_agent: bool,
    /// When `Some`, every session-channel helper issues `x11-req`
    /// immediately after the open and before the matching
    /// shell/exec/subsystem request. Toggle with
    /// [`Self::set_request_x11_forwarding`]. Carries the wire arguments —
    /// `single_connection`, `auth_protocol`, `auth_cookie`, `screen` —
    /// captured at toggle time.
    request_x11: Option<X11ReqArgs>,
    /// Environment variables to emit as `env` channel requests on every
    /// subsequent session-channel helper, between the open and the matching
    /// shell/exec request. Populated from `SetEnv` / `SendEnv`. Each pair is
    /// sent with `want_reply = false` (matching OpenSSH); the server applies
    /// its `AcceptEnv` filter silently.
    session_env: Vec<(String, String)>,
    /// Keepalive policy for [`Self::serve`]: `Some((interval, count_max))`
    /// sends a `keepalive@openssh.com` global request (with `want_reply`)
    /// after `interval` of silence, tearing the connection down once
    /// `count_max` probes go unanswered (`ServerAliveInterval` /
    /// `ServerAliveCountMax`). `None` ⇒ no keepalive.
    keepalive: Option<(Duration, u32)>,
    /// When `Some`, the next `exec` helper allocates a PTY before issuing the
    /// `exec` request — used by `RequestTTY force`/`yes` so a remote command
    /// runs under a terminal. Carries the `pty-req` wire arguments. The
    /// interactive shell path passes its PTY parameters directly, so this
    /// toggle is only consulted by [`Self::exec`].
    request_pty: Option<PtyReqArgs>,
    /// Outstanding reverse-forward grants: the set of `(bind_address,
    /// bind_port)` pairs for which [`Self::request_tcpip_forward`] succeeded
    /// and which have not been cancelled via [`Self::cancel_tcpip_forward`].
    ///
    /// Used as a defence-in-depth filter in [`serve`](Self::serve): a
    /// server-initiated `forwarded-tcpip` channel-open must correlate to a
    /// forward the client actually requested. Because the server may echo a
    /// bind address that differs textually from the one requested (e.g.
    /// `0.0.0.0` for an empty/`localhost` request) and `bind_port == 0`
    /// resolves to a server-assigned port, the matching here is
    /// deliberately conservative: the library only *rejects* a
    /// `forwarded-tcpip` open when there are **zero** outstanding grants —
    /// i.e. an unsolicited forward when the client never asked for any.
    /// Callers that need exact per-binding correlation (as the shipped
    /// `ssh` binary does) should keep doing it in their
    /// `on_forwarded_tcpip` callback.
    pub(crate) tcpip_forward_grants: Vec<(String, u16)>,
    /// Outstanding reverse streamlocal-forward grants: socket paths for which
    /// [`Self::request_streamlocal_forward`] succeeded and which have not been
    /// cancelled via [`Self::cancel_streamlocal_forward`]. Used as the same
    /// conservative defence-in-depth filter as [`Self::tcpip_forward_grants`]:
    /// a `forwarded-streamlocal@openssh.com` open is rejected only when there
    /// are **zero** outstanding grants.
    pub(crate) streamlocal_forward_grants: Vec<String>,
    /// Set by the host-key verifier when a *changed* key was accepted via
    /// [`TofuAction::AcceptWithWarning`]. Shared with the verifier closure
    /// (which runs on every KEX, including re-keys). See
    /// [`Self::host_key_mismatched`].
    host_key_mismatched: Arc<AtomicBool>,
    /// The methods the server last said could continue, when
    /// authentication failed ([`Self::auth_methods_left`]).
    auth_methods_left: Vec<String>,
    /// Shows the server's pre-authentication banner
    /// ([`Self::set_banner_handler`]).
    banner_handler: Option<BannerHandler>,
    /// See [`Self::set_global_request_hook`].
    global_request_hook: Option<GlobalRequestHook>,
}

/// Wire arguments captured by [`Client::set_request_x11_forwarding`] and
/// emitted as the body of each `x11-req` channel request.
#[derive(Clone)]
struct X11ReqArgs {
    single_connection: bool,
    auth_protocol: String,
    auth_cookie: String,
    screen: u32,
}

/// Wire arguments captured by [`Client::set_request_pty`] and emitted as the
/// body of a `pty-req` channel request ahead of an `exec` (RequestTTY force).
#[derive(Clone)]
struct PtyReqArgs {
    term: String,
    cols: u32,
    rows: u32,
    px_w: u32,
    px_h: u32,
    modes: Vec<u8>,
}

impl Client {
    /// Connect, complete version exchange + KEX + NEWKEYS, leave the codec keyed
    /// and ready for userauth.
    pub fn connect<A: ToSocketAddrs>(addr: A, cfg: Config) -> Result<Self> {
        let stream = TcpStream::connect(addr)?;
        if let Some(t) = cfg.timeout {
            stream.set_read_timeout(Some(t))?;
            stream.set_write_timeout(Some(t))?;
        }
        stream.set_nodelay(true)?;
        #[cfg(feature = "multichannel")]
        let read_timeout = cfg.timeout;
        #[allow(unused_mut)]
        let mut me = Self::from_transport(Box::new(stream), "", 0, cfg)?;
        #[cfg(feature = "multichannel")]
        {
            me.read_timeout = read_timeout;
        }
        Ok(me)
    }

    /// Like [`Self::connect`], but threads the user-supplied host name and
    /// port through so [`HostKeyPolicy::KnownHosts`] has something stable
    /// to look up. Use this instead of `connect` whenever the host key
    /// policy reads `known_hosts` — `connect` accepts any `ToSocketAddrs`
    /// (including `IpAddr`) and so cannot recover the original hostname.
    pub fn connect_to_host(host: &str, port: u16, cfg: Config) -> Result<Self> {
        let stream = TcpStream::connect((host, port))?;
        if let Some(t) = cfg.timeout {
            stream.set_read_timeout(Some(t))?;
            stream.set_write_timeout(Some(t))?;
        }
        stream.set_nodelay(true)?;
        #[cfg(feature = "multichannel")]
        let read_timeout = cfg.timeout;
        #[allow(unused_mut)]
        let mut me = Self::from_transport(Box::new(stream), host, port, cfg)?;
        #[cfg(feature = "multichannel")]
        {
            me.read_timeout = read_timeout;
        }
        Ok(me)
    }

    /// Run a client over an arbitrary [`Transport`] (rather than a fresh
    /// [`TcpStream`]). Used by ProxyJump (a `direct-tcpip` channel to the
    /// next hop) and ProxyCommand (pipes to a spawned helper). `host`/`port`
    /// name the *target reached through* the transport so per-hop host-key
    /// checks (`KnownHosts`) key off the right name. The caller is
    /// responsible for any timeouts on the transport.
    pub fn connect_via(
        stream: Box<dyn Transport>,
        host: &str,
        port: u16,
        cfg: Config,
    ) -> Result<Self> {
        Self::from_transport(stream, host, port, cfg)
    }

    /// Shared construction: build the [`Client`] struct over `stream`,
    /// recording `host`/`port` for host-key lookups, then run version
    /// exchange + KEX. `host` may be empty (e.g. plain `connect`), in which
    /// case a `KnownHosts` policy fails closed with [`Error::Config`] (see
    /// [`HostKeyPolicy::KnownHosts`]).
    fn from_transport(
        stream: Box<dyn Transport>,
        host: &str,
        port: u16,
        cfg: Config,
    ) -> Result<Self> {
        // Host-key verification is injected into the sans-IO driver as a
        // closure, capturing the policy + target so the driver stays free of
        // known-hosts I/O / prompting. Called on each `SSH_MSG_KEX_ECDH_REPLY`
        // (initial and re-key).
        let host_key_policy = cfg.host_key_policy;
        let mut algorithms = cfg.algorithms;
        let pinned = cfg.pinned_host_key;
        if let Some(pin) = &pinned {
            algorithms.host_key_algorithms = Some(pinned_host_key_algorithms(
                pin,
                algorithms.host_key_algorithms.as_deref(),
            )?);
        }
        // Bring the host-key types we already trust in `known_hosts` to the
        // front of the offer (OpenSSH's `order_hostkeyalgs()`), so a server
        // with several host keys presents the one we recorded rather than a
        // higher-preference type that would look like a changed key. Done here,
        // before the policy is moved into the verifier closure.
        prioritize_known_host_key_algorithms(&mut algorithms, &host_key_policy, host, port);
        let target_host = host.to_string();
        let ca_sig_algos = algorithms.ca_signature_algorithms.clone();
        let host_key_mismatched = Arc::new(AtomicBool::new(false));
        let mismatched_flag = host_key_mismatched.clone();
        let verifier_factory: crate::driver::client::VerifierFactory =
            Box::new(move |reply: &[u8], runner: &KexRunner| {
                if let Some(pin) = &pinned
                    && reply_host_key(reply)? != pin.as_slice()
                {
                    return Err(Error::HostKeyRejected);
                }
                build_verifier(
                    reply,
                    &host_key_policy,
                    runner,
                    &target_host,
                    port,
                    ca_sig_algos.as_deref(),
                    unix_now(),
                    &mismatched_flag,
                )
            });
        let driver = ClientDriver::new(algorithms.clone(), verifier_factory);
        let mut me = Self {
            stream,
            #[cfg(feature = "multichannel")]
            split_reader: None,
            #[cfg(feature = "multichannel")]
            read_timeout: None,
            out_buf: VecDeque::new(),
            conn: ConnectionState::new(),
            driver,
            algo_overrides: algorithms,
            request_auth_agent: false,
            request_x11: None,
            session_env: Vec::new(),
            keepalive: None,
            request_pty: None,
            tcpip_forward_grants: Vec::new(),
            streamlocal_forward_grants: Vec::new(),
            host_key_mismatched,
            auth_methods_left: Vec::new(),
            banner_handler: None,
            global_request_hook: None,
        };
        me.driver.start(Instant::now())?;
        me.drive_handshake()?;
        Ok(me)
    }

    /// Whether X11 forwarding was requested ([`Self::set_request_x11_forwarding`]).
    pub(crate) fn wants_x11(&self) -> bool {
        self.request_x11.is_some()
    }

    /// `true` if this session was established over a host key that did
    /// **not** match the `known_hosts` entry and was let through only by
    /// [`TofuAction::AcceptWithWarning`] (OpenSSH's
    /// `StrictHostKeyChecking=no`). While set, the client — like OpenSSH —
    /// refuses to send password or keyboard-interactive credentials
    /// (`authenticate` / `run_auth` fail with [`Error::Config`]), does not
    /// request agent or X11 forwarding, and refuses to set up port
    /// forwards, so a man-in-the-middle cannot harvest a password or reach
    /// the local agent / display. Public-key authentication still works.
    pub fn host_key_mismatched(&self) -> bool {
        self.host_key_mismatched.load(Ordering::SeqCst)
    }

    /// Send whatever reply [`unsolicited_reply`] says the peer is owed for
    /// `ev`; the catch-all arm of every single-channel packet loop.
    pub(crate) fn handle_unsolicited(&mut self, ev: &ChannelEvent) -> Result<()> {
        self.notify_global_request(ev);
        if let Some(p) = unsolicited_reply(&mut self.conn, ev)? {
            self.write_payload(&p)?;
        }
        Ok(())
    }

    /// Call `f` with the name and body of each global request the server
    /// sends that the library does not handle itself — OpenSSH's
    /// `hostkeys-00@openssh.com` (its host keys, for `UpdateHostKeys`), say.
    /// Requests wanting a reply are still refused. Stays in effect once the
    /// connection is shared.
    pub fn set_global_request_hook(&mut self, f: impl Fn(&str, &[u8]) + Send + Sync + 'static) {
        self.global_request_hook = Some(Arc::new(f));
    }

    /// Pass an unhandled server global request to the hook, if any.
    pub(crate) fn notify_global_request(&self, ev: &ChannelEvent) {
        if let (
            Some(f),
            ChannelEvent::GlobalRequest {
                request: crate::channel::GlobalRequest::Other { name, raw },
                ..
            },
        ) = (&self.global_request_hook, ev)
        {
            f(name, raw);
        }
    }

    /// Override the re-key thresholds (defaults to
    /// [`RekeyPolicy::default`](crate::transport::RekeyPolicy), i.e. 1 GiB /
    /// 1 hour). Mirrors the server-side knob; mainly useful for tests that
    /// need to force frequent client-initiated re-keys.
    pub fn set_rekey_policy(&mut self, policy: crate::transport::RekeyPolicy) {
        self.driver.set_rekey_policy(policy);
    }

    /// Try every credential in order until one succeeds or all are refused.
    ///
    /// Builds a single [`ClientAuth`] driver carrying every credential and
    /// runs it to completion on this one connection — the SERVICE_REQUEST is
    /// sent exactly once. Multiple credentials (e.g. several keys, a password,
    /// a keyboard-interactive responder) are tried in order *within the same
    /// userauth exchange*, which is what a multi-factor server expects (the
    /// driver advances on each USERAUTH_FAILURE / partial-success).
    pub fn authenticate(&mut self, user: &str, credentials: Vec<ClientCredential>) -> Result<()> {
        // Over a changed host key only public-key credentials may be used
        // (see `host_key_mismatched`); this fails outright if nothing
        // usable remains.
        let credentials = if self.host_key_mismatched() {
            filter_credentials_after_mismatch(credentials)?
        } else {
            credentials
        };
        let mut auth = ClientAuth::new(user, self.driver.session_id().to_vec());
        // Local PubkeyAcceptedAlgorithms policy from ssh_config, applied
        // before any server-sig-algs filtering.
        if let Some(accepted) = self.algo_overrides.pubkey_accepted_algorithms.clone() {
            auth.set_pubkey_accepted(accepted);
        }
        // RFC 8308 §3.1: if the server told us which signature algorithms
        // it will accept on a publickey auth, propagate that to the auth
        // driver so we skip credentials it would reject anyway.
        if let Some(ext) = self.driver.peer_ext_info()
            && let Some(algs) = ext.server_sig_algs.as_deref()
        {
            auth.set_server_sig_algs(algs);
        }
        for c in credentials {
            auth.add_credential(c);
        }
        self.run_auth(auth)
    }

    /// Drive a fully-configured [`ClientAuth`] to a verdict on this connection.
    /// Hands the driver the auth state machine (which emits the single
    /// SERVICE_REQUEST) and pumps until it reports success or exhausts its
    /// credentials. Callers that need to inject policy (server-sig-algs,
    /// pubkey-accepted, a re-promptable password closure, a keyboard-interactive
    /// responder) build the driver themselves and hand it here — there is
    /// exactly one driver, and thus exactly one SERVICE_REQUEST, per connection.
    pub fn run_auth(&mut self, mut auth: ClientAuth) -> Result<()> {
        // A server that announces it takes `publickey-hostbound-v00`: bind
        // each signature to its host key (as OpenSSH's client does).
        let hostbound = self
            .driver
            .peer_ext_info()
            .and_then(|e| e.extension("publickey-hostbound@openssh.com"))
            == Some("0");
        if hostbound && let Some(b) = self.session_binding() {
            auth.set_hostbound(Some(b.host_key));
        }
        let first = auth.start();
        self.write_payload(&first)?;

        let mismatched = self.host_key_mismatched();
        let sig_algs = |c: &Self| {
            c.driver
                .peer_ext_info()
                .and_then(|e| e.server_sig_algs.clone())
        };
        let mut known_sig_algs = sig_algs(self);
        for _ in 0..MAX_AUTH_STEPS {
            // `read_one_packet` surfaces post-NEWKEYS payloads (userauth here)
            // from the driver; transport concerns stay inside it.
            let payload = self.read_one_packet()?;
            // A server told our `ext-info-in-auth@openssh.com` sends its
            // `server-sig-algs` again once it knows the user.
            let now_sig_algs = sig_algs(self);
            if now_sig_algs != known_sig_algs {
                if let Some(algs) = now_sig_algs.as_deref() {
                    auth.set_server_sig_algs(algs);
                }
                known_sig_algs = now_sig_algs;
            }
            match auth.on_packet(&payload)? {
                // Never let a password / keyboard-interactive answer out
                // over a changed host key, whatever the caller put in the
                // driver. `authenticate` filters these up front; this is
                // the backstop for hand-built `ClientAuth` drivers.
                ClientStep::Send(p) if mismatched && auth_packet_blocked_after_mismatch(&p) => {
                    return Err(Error::Config(HOST_KEY_MISMATCH_DISABLED));
                }
                ClientStep::Send(p) => self.write_payload(&p)?,
                ClientStep::Success => {
                    // Hand the post-auth transitions (compression + EXT_INFO
                    // re-arm) to the driver, which owns the codec/runner.
                    self.driver.notify_auth_success();
                    return Ok(());
                }
                ClientStep::Failed { continuations, .. } => {
                    self.auth_methods_left = continuations;
                    return Err(Error::AuthFailed);
                }
                ClientStep::Banner { message, .. } => {
                    if let Some(show) = &self.banner_handler {
                        show(&message);
                    }
                }
                ClientStep::Idle => {}
            }
        }
        Err(Error::Protocol("auth: too many steps without termination"))
    }

    /// Packet and byte totals of the connection so far.
    pub fn traffic(&self) -> crate::transport::TrafficStats {
        self.driver.traffic()
    }

    /// The algorithms the most recent key exchange chose (kex, host key,
    /// ciphers, MACs, compression); `None` before the first one.
    pub fn negotiated(&self) -> Option<crate::transport::Negotiated> {
        self.driver.negotiated()
    }

    /// Call `f` after each key exchange completes — the first one and every
    /// re-key, including those run once the connection is shared — with the
    /// algorithms it chose.
    pub fn set_kex_observer(
        &mut self,
        f: impl Fn(&crate::transport::Negotiated) + Send + Sync + 'static,
    ) {
        self.driver.set_kex_observer(std::sync::Arc::new(f));
    }

    /// Show the server's `SSH_MSG_USERAUTH_BANNER` (sshd's `Banner`, sent
    /// before authentication) with `show`, as [`Self::run_auth`] receives
    /// it. Without a handler, banners are ignored.
    pub fn set_banner_handler(&mut self, show: impl Fn(&str) + Send + Sync + 'static) {
        self.banner_handler = Some(Box::new(show));
    }

    /// After [`Self::run_auth`] failed: the methods the server last said
    /// could continue (OpenSSH prints them: `Permission denied (publickey).`).
    pub fn auth_methods_left(&self) -> &[String] {
        &self.auth_methods_left
    }

    /// Build a [`ClientAuth`] for `user` with this connection's resolved
    /// ssh_config policy (PubkeyAcceptedAlgorithms + server-sig-algs)
    /// pre-installed, ready for the caller to push credentials onto and then
    /// hand to [`Self::run_auth`]. This is the one-driver-per-connection entry
    /// point for callers (e.g. the `ssh` binary) that need a re-promptable
    /// password or a keyboard-interactive responder.
    ///
    /// The returned [`ClientAuth`] is a [`crate::hazmat::auth`] type: it is
    /// exposed so the credential list can be assembled incrementally, but it
    /// carries no stability promise beyond the crate version.
    pub fn new_auth_driver(&self, user: &str) -> ClientAuth {
        let mut auth = ClientAuth::new(user, self.driver.session_id().to_vec());
        if let Some(accepted) = self.algo_overrides.pubkey_accepted_algorithms.clone() {
            auth.set_pubkey_accepted(accepted);
        }
        if let Some(ext) = self.driver.peer_ext_info()
            && let Some(algs) = ext.server_sig_algs.as_deref()
        {
            auth.set_server_sig_algs(algs);
        }
        auth
    }

    /// Convenience: try password authentication only.
    pub fn authenticate_password(&mut self, user: &str, password: &str) -> Result<()> {
        self.authenticate(user, vec![ClientCredential::Password(password.into())])
    }

    /// Session identifier for this connection — the exchange hash `H` of the
    /// *first* key exchange (RFC 4253 §7.2). Stable across re-keys.
    pub fn session_id(&self) -> &[u8] {
        self.driver.session_id()
    }

    /// What binds an `ssh-agent` connection to this session
    /// (`session-bind@openssh.com`, for keys added with destination
    /// constraints): the server's host key, the session id and the host's
    /// signature over it. `None` before the key exchange.
    pub fn session_binding(&self) -> Option<SessionBinding> {
        let (host_key, signature) = self.driver.initial_host_key_sig()?;
        Some(SessionBinding {
            host_key: host_key.to_vec(),
            session_id: self.driver.session_id().to_vec(),
            signature: signature.to_vec(),
        })
    }

    /// Most recent `SSH_MSG_EXT_INFO` (RFC 8308) received from the server,
    /// if any. Carries `server-sig-algs` and any forward-compatible
    /// extensions the server advertised. Returns `None` when neither side
    /// advertised the `ext-info-{c,s}` markers or the server simply
    /// declined to send one.
    pub fn peer_ext_info(&self) -> Option<&crate::transport::ExtInfo> {
        self.driver.peer_ext_info()
    }

    /// Compression algorithms chosen by the most recent key exchange, as
    /// `(client→server, server→client)`. Both are `"none"` on an
    /// uncompressed connection; requesting compression (`Config.compression =
    /// Some(true)`, the `compress` feature enabled) and connecting to a peer
    /// that offers it yields `"zlib@openssh.com"`. Returns `None` before the
    /// first KEX completes.
    ///
    /// Note `zlib@openssh.com` is *delayed*: the codec advertises this name at
    /// KEX but only starts compressing after `USERAUTH_SUCCESS`, so this
    /// reflects the negotiated intent, not whether bytes are being compressed
    /// yet.
    pub fn negotiated_compression(&self) -> Option<(String, String)> {
        self.driver.negotiated_compression()
    }

    /// Convenience: try publickey authentication only.
    pub fn authenticate_publickey(
        &mut self,
        user: &str,
        key: Box<dyn HostKey + Send>,
    ) -> Result<()> {
        self.authenticate(user, vec![ClientCredential::PublicKey(key)])
    }

    /// Open a session channel that exists solely to host an
    /// `auth-agent-req@openssh.com` request, then send that request and
    /// return the channel's local id. The caller is expected to keep this
    /// channel open for the lifetime of any agent-forwarding work
    /// ([`Self::serve`] with [`ClientHandlers::on_auth_agent`] installed),
    /// then tear it down with [`Self::close_session`].
    ///
    /// On the server side, the matching `auth-agent-req` arms a
    /// per-session-channel Unix-socket listener; closing this channel
    /// unlinks the socket and stops accepting agent calls. See
    /// [`crate::forwarding::agent`] for the server side.
    pub fn open_session_for_agent_forward(&mut self) -> Result<u32> {
        if self.host_key_mismatched() {
            return Err(Error::Config(HOST_KEY_MISMATCH_DISABLED));
        }
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        let mut iter_guard = 0usize;
        loop {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("agent-forward: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => break,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("agent-forward: channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        let p = self
            .conn
            .send_request(local_id, ChannelRequest::AuthAgentReq, false)?;
        self.write_payload(&p)?;
        Ok(local_id)
    }

    /// Open a session channel that exists solely to host an `x11-req`
    /// request, then send that request and return the channel's local id.
    /// The caller is expected to keep this channel open for the lifetime
    /// of any X11-forwarding work ([`Self::serve`] with
    /// [`ClientHandlers::on_x11`] installed), then tear it down with
    /// [`Self::close_session`].
    ///
    /// On the server side, the matching `x11-req` arms a per-session
    /// TCP display listener (`127.0.0.1:6000+N`); closing this channel
    /// stops it. See [`crate::forwarding::x11`] for the server side.
    pub fn open_session_for_x11_forward(
        &mut self,
        single_connection: bool,
        auth_protocol: &str,
        auth_cookie: &str,
        screen: u32,
    ) -> Result<u32> {
        if self.host_key_mismatched() {
            return Err(Error::Config(HOST_KEY_MISMATCH_DISABLED));
        }
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        let mut iter_guard = 0usize;
        loop {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("x11-forward: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => break,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("x11-forward: channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        let p = self.conn.send_request(
            local_id,
            ChannelRequest::X11Req {
                single_connection,
                auth_protocol: auth_protocol.to_string(),
                auth_cookie: auth_cookie.to_string(),
                screen,
            },
            false,
        )?;
        self.write_payload(&p)?;
        Ok(local_id)
    }

    /// Set the underlying TCP socket's read timeout. `None` clears it
    /// (blocking reads, the default). Used by the SharedClient pump in
    /// interactive-shell mode to release its inner mutex periodically so
    /// sibling write threads can make progress; you generally don't need
    /// it for one-shot exec / SFTP flows where there's only one waiter.
    ///
    /// Callers that set a non-`None` value MUST tolerate
    /// `ErrorKind::WouldBlock` / `ErrorKind::TimedOut` on subsequent
    /// reads (or only use code paths that fold those into a no-op).
    pub fn set_read_timeout(&mut self, t: Option<core::time::Duration>) -> std::io::Result<()> {
        #[cfg(feature = "multichannel")]
        {
            self.read_timeout = t;
        }
        // A split-off reader applies the timeout on each read (the socket
        // option is shared between the two handles).
        #[cfg(feature = "multichannel")]
        let split = self.split_reader.is_some();
        #[cfg(not(feature = "multichannel"))]
        let split = false;
        if !split {
            self.stream.set_read_timeout(t)?;
        }
        // Bound blocking writes by the same interval. A consumer that asks for
        // a read timeout is, by construction, the concurrent-pump case: it
        // releases and re-acquires the shared transport lock between pumps so
        // siblings can make progress. A write that wedged the lock (full send
        // buffer, peer not reading) would defeat that and deadlock both peers.
        // With a write timeout set, `pump_out` buffers the unsent remainder
        // and returns, so the lock is always released promptly. Clearing the
        // read timeout (`None`) restores fully-blocking writes for the
        // single-waiter path, where there is no sibling to starve.
        self.stream.set_write_timeout(t)
    }

    /// Send `SSH_MSG_CHANNEL_CLOSE` for `channel`. Used to tear down a
    /// session channel opened by
    /// [`Self::open_session_for_agent_forward`] once the matching serve
    /// loop has returned. Best-effort: silently swallows codec errors for
    /// channels already torn down by the peer.
    pub fn close_session(&mut self, channel: u32) -> Result<()> {
        let payload = match self.conn.send_close(channel) {
            Ok(p) => p,
            Err(_) => return Ok(()),
        };
        self.write_payload(&payload)?;
        Ok(())
    }

    /// Toggle whether the next session-channel helper ([`Self::exec`],
    /// [`Self::exec_stream`], [`Self::shell_with_stdin`], [`Self::sftp`])
    /// prefaces its shell/exec/subsystem request with
    /// `auth-agent-req@openssh.com` (the channel request that asks the
    /// server to set up an `SSH_AUTH_SOCK` Unix socket for this session
    /// channel and call back via `auth-agent@openssh.com` channels —
    /// OpenSSH's `ssh -A`).
    ///
    /// Sticky: stays set until cleared. The agent-req is sent with
    /// `want_reply = false`, matching OpenSSH; the server's response
    /// (acceptance / refusal) is observable only via whether a callback
    /// channel ever lands on a peer-installed [`ClientHandlers::on_auth_agent`].
    pub fn set_request_auth_agent_forwarding(&mut self, on: bool) {
        self.request_auth_agent = on;
    }

    /// Internal: emit `auth-agent-req@openssh.com` if the toggle is set.
    /// Called by every session-channel helper between OpenConfirmed and
    /// the matching shell/exec/subsystem request.
    pub(crate) fn maybe_send_auth_agent_req(&mut self, channel: u32) -> Result<()> {
        // Never expose the local agent to a server whose key changed
        // (`host_key_mismatched`) — OpenSSH clears ForwardAgent there too.
        if self.request_auth_agent && !self.host_key_mismatched() {
            let p = self
                .conn
                .send_request(channel, ChannelRequest::AuthAgentReq, false)?;
            self.write_payload(&p)?;
        }
        Ok(())
    }

    /// Arm `x11-req` to be emitted on every subsequent session-channel
    /// helper ([`Self::exec`], [`Self::exec_stream`],
    /// [`Self::shell_with_stdin`], [`Self::sftp`]) between OpenConfirmed
    /// and the matching shell/exec/subsystem request.
    ///
    /// `single_connection` follows RFC 4254 §6.3.1: `true` accepts exactly
    /// one X-client connection on the forwarded display, `false` accepts
    /// any number (the OpenSSH default for both `-X` and `-Y`).
    ///
    /// `auth_protocol` is typically `"MIT-MAGIC-COOKIE-1"`; `auth_cookie`
    /// is the matching cookie as a hex string. The server forwards both
    /// verbatim into [`crate::server::X11ForwardHandler::setup`] — cookie
    /// substitution at the X-protocol level is the responsibility of the
    /// `on_x11` handler. `screen` is the X screen number (0 in nearly all
    /// uses). Pass `None` to clear the toggle.
    pub fn set_request_x11_forwarding(&mut self, args: Option<(bool, String, String, u32)>) {
        self.request_x11 =
            args.map(
                |(single_connection, auth_protocol, auth_cookie, screen)| X11ReqArgs {
                    single_connection,
                    auth_protocol,
                    auth_cookie,
                    screen,
                },
            );
    }

    /// Internal: emit `x11-req` if the toggle is set. Called by every
    /// session-channel helper between OpenConfirmed and the matching
    /// shell/exec/subsystem request, right after
    /// [`Self::maybe_send_auth_agent_req`].
    pub(crate) fn maybe_send_x11_req(&mut self, channel: u32) -> Result<()> {
        // No X11 forwarding over a changed host key (`host_key_mismatched`).
        if self.host_key_mismatched() {
            return Ok(());
        }
        if let Some(args) = self.request_x11.clone() {
            let p = self.conn.send_request(
                channel,
                ChannelRequest::X11Req {
                    single_connection: args.single_connection,
                    auth_protocol: args.auth_protocol,
                    auth_cookie: args.auth_cookie,
                    screen: args.screen,
                },
                false,
            )?;
            self.write_payload(&p)?;
        }
        Ok(())
    }

    /// Set the environment variables emitted as `env` channel requests on
    /// every subsequent session-channel helper ([`Self::exec`],
    /// [`Self::exec_stream`], [`Self::shell_with_stdin`]).
    ///
    /// Populated from `SetEnv` (literal pairs) and `SendEnv` (local env vars
    /// matched by pattern, resolved by the caller). Each pair is sent with
    /// `want_reply = false`, matching OpenSSH; whether the server accepts a
    /// given variable is governed by its `AcceptEnv` policy and is not
    /// observable here. Sticky: stays set until replaced.
    pub fn set_session_env(&mut self, env: Vec<(String, String)>) {
        self.session_env = env;
    }

    /// Internal: emit one `env` request per [`Self::set_session_env`] entry.
    /// Called by every session-channel helper between OpenConfirmed and the
    /// matching shell/exec request.
    pub(crate) fn maybe_send_env(&mut self, channel: u32) -> Result<()> {
        for (name, value) in self.session_env.clone() {
            let p = self
                .conn
                .send_request(channel, ChannelRequest::Env { name, value }, false)?;
            self.write_payload(&p)?;
        }
        Ok(())
    }

    /// Configure server-keepalive probing for [`Self::serve`]
    /// (`ServerAliveInterval` / `ServerAliveCountMax`).
    ///
    /// `interval_secs == 0` disables keepalive. Otherwise the serve loop
    /// sends a `keepalive@openssh.com` global request (with `want_reply`)
    /// after `interval_secs` of socket silence and tears the connection down
    /// once `count_max` consecutive probes go unanswered. Has no effect on
    /// the one-shot [`Self::exec`] path (which doesn't run the serve loop).
    pub fn set_keepalive(&mut self, interval_secs: u32, count_max: u32) {
        self.keepalive = if interval_secs == 0 {
            None
        } else {
            Some((Duration::from_secs(interval_secs as u64), count_max.max(1)))
        };
    }

    /// Arm a `pty-req` to be emitted ahead of the next [`Self::exec`]'s
    /// `exec` request, so a remote command runs under a pseudo-terminal
    /// (`RequestTTY force` / `yes`). Pass `None` to clear.
    ///
    /// The interactive shell path allocates its PTY directly via
    /// [`crate::shared::SharedClient::shell_stream`]; this toggle exists for
    /// the one-shot `exec` path only.
    pub fn set_request_pty(&mut self, args: Option<(String, u32, u32, u32, u32, Vec<u8>)>) {
        self.request_pty = args.map(|(term, cols, rows, px_w, px_h, modes)| PtyReqArgs {
            term,
            cols,
            rows,
            px_w,
            px_h,
            modes,
        });
    }

    /// Internal: emit `pty-req` if [`Self::set_request_pty`] armed one.
    pub(crate) fn maybe_send_pty_req(&mut self, channel: u32) -> Result<()> {
        if let Some(args) = self.request_pty.clone() {
            let p = self.conn.send_request(
                channel,
                ChannelRequest::PtyReq {
                    term: args.term,
                    cols: args.cols,
                    rows: args.rows,
                    px_w: args.px_w,
                    px_h: args.px_h,
                    modes: args.modes,
                },
                false,
            )?;
            self.write_payload(&p)?;
        }
        Ok(())
    }

    /// Run a remote command, draining stdout/stderr and capturing the exit
    /// status (or signal). Returns once the server has closed the channel.
    pub fn exec(&mut self, command: &str) -> Result<ExecOutput> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        let mut opened = false;
        let mut iter_guard = 0usize;
        while !opened {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("exec: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => {
                    opened = true;
                }
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        self.maybe_send_auth_agent_req(local_id)?;
        self.maybe_send_x11_req(local_id)?;
        self.maybe_send_env(local_id)?;
        self.maybe_send_pty_req(local_id)?;

        let exec_req = self.conn.send_request(
            local_id,
            ChannelRequest::Exec {
                command: command.into(),
            },
            true,
        )?;
        self.write_payload(&exec_req)?;

        let mut exec_accepted = false;
        iter_guard = 0;
        while !exec_accepted {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("exec: request loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::Success { channel } if channel == local_id => exec_accepted = true,
                ChannelEvent::Failure { channel } if channel == local_id => {
                    return Err(Error::Protocol("exec request denied"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        let mut out = ExecOutput {
            stdout: Vec::new(),
            stderr: Vec::new(),
            exit_status: None,
            exit_signal: None,
        };
        let mut local_eof_sent = false;
        let mut local_close_sent = false;
        let mut remote_close_seen = false;

        for _ in 0..MAX_EXEC_ITER {
            if remote_close_seen && local_close_sent {
                break;
            }
            let payload = self.read_one_packet()?;
            let ev = self.conn.on_packet(&payload)?;
            match ev {
                ChannelEvent::Data { channel, data } if channel == local_id => {
                    if out.stdout.len() + out.stderr.len() + data.len() > MAX_EXEC_OUTPUT {
                        return Err(Error::Protocol("exec output too large"));
                    }
                    let n = data.len() as u32;
                    out.stdout.extend_from_slice(&data);
                    if let Some(adj) = self.conn.replenish_window(local_id, n)? {
                        self.write_payload(&adj)?;
                    }
                }
                ChannelEvent::ExtendedData {
                    channel,
                    code,
                    data,
                } if channel == local_id => {
                    if out.stdout.len() + out.stderr.len() + data.len() > MAX_EXEC_OUTPUT {
                        return Err(Error::Protocol("exec output too large"));
                    }
                    let n = data.len() as u32;
                    if code == SSH_EXTENDED_DATA_STDERR {
                        out.stderr.extend_from_slice(&data);
                    } else {
                        out.stdout.extend_from_slice(&data);
                    }
                    if let Some(adj) = self.conn.replenish_window(local_id, n)? {
                        self.write_payload(&adj)?;
                    }
                }
                ChannelEvent::Request {
                    channel,
                    request,
                    want_reply,
                } if channel == local_id => {
                    match request {
                        ChannelRequest::ExitStatus { code } => out.exit_status = Some(code),
                        ChannelRequest::ExitSignal { name, .. } => out.exit_signal = Some(name),
                        _ => {}
                    }
                    if want_reply {
                        let p = self.conn.send_request_failure(local_id)?;
                        self.write_payload(&p)?;
                    }
                }
                ChannelEvent::Eof { channel } if channel == local_id && !local_eof_sent => {
                    let p = self.conn.send_eof(local_id)?;
                    self.write_payload(&p)?;
                    local_eof_sent = true;
                }
                ChannelEvent::Close { channel } if channel == local_id => {
                    remote_close_seen = true;
                    if !local_close_sent {
                        let p = self.conn.send_close(local_id)?;
                        self.write_payload(&p)?;
                        local_close_sent = true;
                    }
                }
                ChannelEvent::WindowAdjust { .. } => {}
                other => self.handle_unsolicited(&other)?,
            }
        }

        if !(remote_close_seen && local_close_sent) {
            return Err(Error::Protocol("exec: drain loop exceeded iteration cap"));
        }
        Ok(out)
    }

    /// Open an interactive `"shell"` session with an allocated PTY, push
    /// `stdin` into the channel once, EOF, and drain the response until
    /// the server closes the channel.
    ///
    /// This is a one-shot helper aimed at scripted tests and simple
    /// interop checks — a real interactive client would interleave reads
    /// and writes with the terminal. It exists primarily so the server's
    /// `pty-req` / `shell` wiring can be exercised end-to-end.
    pub fn shell_with_stdin(
        &mut self,
        term: &str,
        cols: u32,
        rows: u32,
        stdin: &[u8],
    ) -> Result<ExecOutput> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        // Wait for OPEN_CONFIRMATION.
        let mut opened = false;
        let mut iter_guard = 0usize;
        while !opened {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("shell: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => {
                    opened = true;
                }
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        self.maybe_send_auth_agent_req(local_id)?;
        self.maybe_send_x11_req(local_id)?;

        // pty-req with want_reply=true.
        let pty_req = self.conn.send_request(
            local_id,
            ChannelRequest::PtyReq {
                term: term.into(),
                cols,
                rows,
                px_w: 0,
                px_h: 0,
                modes: Vec::new(),
            },
            true,
        )?;
        self.write_payload(&pty_req)?;
        self.await_request_reply(local_id, "pty-req")?;

        // shell with want_reply=true.
        let shell_req = self
            .conn
            .send_request(local_id, ChannelRequest::Shell, true)?;
        self.write_payload(&shell_req)?;
        self.await_request_reply(local_id, "shell")?;

        // Push stdin (if any), then EOF.
        if !stdin.is_empty() {
            let mut off = 0usize;
            iter_guard = 0;
            while off < stdin.len() {
                iter_guard += 1;
                if iter_guard > MAX_EXEC_ITER {
                    return Err(Error::Protocol("shell: stdin drain loop did not converge"));
                }
                let (payload, taken) = self.conn.send_data(local_id, &stdin[off..])?;
                if taken == 0 {
                    // Window full — read until we get a WINDOW_ADJUST.
                    let pkt = self.read_one_packet()?;
                    match self.conn.on_packet(&pkt)? {
                        ChannelEvent::WindowAdjust { channel, .. } if channel == local_id => {}
                        ChannelEvent::Close { channel } if channel == local_id => {
                            return Err(Error::Protocol(
                                "shell: peer closed channel before stdin drain",
                            ));
                        }
                        _ => {}
                    }
                    continue;
                }
                self.write_payload(&payload)?;
                off += taken;
            }
        }
        let eof = self.conn.send_eof(local_id)?;
        self.write_payload(&eof)?;

        // Drain until both sides have CLOSEd.
        let mut out = ExecOutput {
            stdout: Vec::new(),
            stderr: Vec::new(),
            exit_status: None,
            exit_signal: None,
        };
        let mut local_close_sent = false;
        let mut remote_close_seen = false;

        for _ in 0..MAX_EXEC_ITER {
            if remote_close_seen && local_close_sent {
                break;
            }
            let payload = self.read_one_packet()?;
            let ev = self.conn.on_packet(&payload)?;
            match ev {
                ChannelEvent::Data { channel, data } if channel == local_id => {
                    if out.stdout.len() + out.stderr.len() + data.len() > MAX_EXEC_OUTPUT {
                        return Err(Error::Protocol("shell output too large"));
                    }
                    let n = data.len() as u32;
                    out.stdout.extend_from_slice(&data);
                    if let Some(adj) = self.conn.replenish_window(local_id, n)? {
                        self.write_payload(&adj)?;
                    }
                }
                ChannelEvent::ExtendedData {
                    channel,
                    code,
                    data,
                } if channel == local_id => {
                    if out.stdout.len() + out.stderr.len() + data.len() > MAX_EXEC_OUTPUT {
                        return Err(Error::Protocol("shell output too large"));
                    }
                    let n = data.len() as u32;
                    if code == SSH_EXTENDED_DATA_STDERR {
                        out.stderr.extend_from_slice(&data);
                    } else {
                        out.stdout.extend_from_slice(&data);
                    }
                    if let Some(adj) = self.conn.replenish_window(local_id, n)? {
                        self.write_payload(&adj)?;
                    }
                }
                ChannelEvent::Request {
                    channel,
                    request,
                    want_reply,
                } if channel == local_id => {
                    match request {
                        ChannelRequest::ExitStatus { code } => out.exit_status = Some(code),
                        ChannelRequest::ExitSignal { name, .. } => out.exit_signal = Some(name),
                        _ => {}
                    }
                    if want_reply {
                        let p = self.conn.send_request_failure(local_id)?;
                        self.write_payload(&p)?;
                    }
                }
                ChannelEvent::Eof { channel } if channel == local_id => {}
                ChannelEvent::Close { channel } if channel == local_id => {
                    remote_close_seen = true;
                    if !local_close_sent {
                        let p = self.conn.send_close(local_id)?;
                        self.write_payload(&p)?;
                        local_close_sent = true;
                    }
                }
                ChannelEvent::WindowAdjust { .. } => {}
                other => self.handle_unsolicited(&other)?,
            }
        }

        if !(remote_close_seen && local_close_sent) {
            return Err(Error::Protocol("shell: drain loop exceeded iteration cap"));
        }
        Ok(out)
    }

    /// One-shot subsystem helper: open a session channel, send a
    /// `subsystem` request with the given `name`, push `stdin` once, send
    /// EOF, then drain the response until the peer CLOSEs. Returns the
    /// accumulated channel data (the stream's stdout half).
    ///
    /// This is the subsystem analogue of [`exec`]: it doesn't expose the
    /// channel for streaming use (a `Client::sftp` streaming wrapper will
    /// land in a follow-up phase), but it's enough to exercise the
    /// server's subsystem dispatch end-to-end and to drive small
    /// request/response protocols.
    ///
    /// [`exec`]: Self::exec
    pub fn subsystem_once(&mut self, name: &str, stdin: &[u8]) -> Result<Vec<u8>> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        let mut opened = false;
        let mut iter_guard = 0usize;
        while !opened {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("subsystem: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => opened = true,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        let sub_req = self.conn.send_request(
            local_id,
            ChannelRequest::Subsystem { name: name.into() },
            true,
        )?;
        self.write_payload(&sub_req)?;
        self.await_request_reply(local_id, "subsystem")?;

        // Push stdin (if any), then EOF.
        if !stdin.is_empty() {
            let mut off = 0usize;
            iter_guard = 0;
            while off < stdin.len() {
                iter_guard += 1;
                if iter_guard > MAX_EXEC_ITER {
                    return Err(Error::Protocol(
                        "subsystem: stdin drain loop did not converge",
                    ));
                }
                let (payload, taken) = self.conn.send_data(local_id, &stdin[off..])?;
                if taken == 0 {
                    let pkt = self.read_one_packet()?;
                    match self.conn.on_packet(&pkt)? {
                        ChannelEvent::WindowAdjust { channel, .. } if channel == local_id => {}
                        ChannelEvent::Close { channel } if channel == local_id => {
                            return Err(Error::Protocol(
                                "subsystem: peer closed channel before stdin drain",
                            ));
                        }
                        _ => {}
                    }
                    continue;
                }
                self.write_payload(&payload)?;
                off += taken;
            }
        }
        let eof = self.conn.send_eof(local_id)?;
        self.write_payload(&eof)?;

        let mut out = Vec::<u8>::new();
        let mut local_close_sent = false;
        let mut remote_close_seen = false;

        for _ in 0..MAX_EXEC_ITER {
            if remote_close_seen && local_close_sent {
                break;
            }
            let payload = self.read_one_packet()?;
            let ev = self.conn.on_packet(&payload)?;
            match ev {
                ChannelEvent::Data { channel, data } if channel == local_id => {
                    if out.len() + data.len() > MAX_EXEC_OUTPUT {
                        return Err(Error::Protocol("subsystem output too large"));
                    }
                    let n = data.len() as u32;
                    out.extend_from_slice(&data);
                    if let Some(adj) = self.conn.replenish_window(local_id, n)? {
                        self.write_payload(&adj)?;
                    }
                }
                ChannelEvent::ExtendedData {
                    channel,
                    code: _,
                    data,
                } if channel == local_id => {
                    // Subsystems shouldn't send stderr by RFC convention, but
                    // if they do, treat it like stdout for window accounting.
                    let n = data.len() as u32;
                    if let Some(adj) = self.conn.replenish_window(local_id, n)? {
                        self.write_payload(&adj)?;
                    }
                }
                ChannelEvent::Eof { channel } if channel == local_id => {}
                ChannelEvent::Close { channel } if channel == local_id => {
                    remote_close_seen = true;
                    if !local_close_sent {
                        let p = self.conn.send_close(local_id)?;
                        self.write_payload(&p)?;
                        local_close_sent = true;
                    }
                }
                ChannelEvent::WindowAdjust { .. } => {}
                other => self.handle_unsolicited(&other)?,
            }
        }

        if !(remote_close_seen && local_close_sent) {
            return Err(Error::Protocol(
                "subsystem: drain loop exceeded iteration cap",
            ));
        }
        Ok(out)
    }

    /// Open an SFTP session: opens a session channel, requests the `sftp`
    /// subsystem, performs the SFTP `INIT`/`VERSION` handshake, and returns
    /// a [`SftpClient`] borrowing the channel for its lifetime.
    ///
    /// The returned client serialises one request/response at a time. When
    /// it's dropped, the channel is closed.
    #[cfg_attr(
        feature = "multichannel",
        deprecated(
            since = "0.0.2",
            note = "Use SharedClient::sftp instead; the borrow-based API \
                    prevents multiple concurrent channels on one connection."
        )
    )]
    pub fn sftp(&mut self) -> Result<SftpClient<ClientChannelStream<'_>>> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        let mut opened = false;
        let mut iter_guard = 0usize;
        while !opened {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("sftp: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => opened = true,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        self.maybe_send_auth_agent_req(local_id)?;
        self.maybe_send_x11_req(local_id)?;

        let sub_req = self.conn.send_request(
            local_id,
            ChannelRequest::Subsystem {
                name: "sftp".into(),
            },
            true,
        )?;
        self.write_payload(&sub_req)?;
        self.await_request_reply(local_id, "subsystem")?;

        let stream = ClientChannelStream {
            client: self,
            channel: local_id,
            read_buf: Vec::new(),
            stderr_buf: Vec::new(),
            remote_eof: false,
            local_close_sent: false,
        };
        // SftpClient::new performs INIT/VERSION; on error we still want to
        // try to close the channel. Wrap with a manual try-catch.
        match SftpClient::new(stream) {
            Ok(c) => Ok(c),
            Err(e) => Err(Error::Protocol(match e {
                crate::sftp::SftpError::Protocol(s) => s,
                _ => "sftp: handshake failed",
            })),
        }
    }

    /// Open a session channel and ask the server to execute `command`,
    /// returning a [`ClientChannelStream`] borrowing the client for the
    /// channel's lifetime. The stream's `Read` half delivers the command's
    /// stdout, `Write` feeds stdin, and the channel is closed on drop.
    /// Stderr is buffered and can be drained with
    /// [`ClientChannelStream::take_stderr`].
    ///
    /// Mirrors [`Client::sftp`] but issues `ChannelRequest::Exec` instead
    /// of `Subsystem`. Used by [`Client::scp_send_to`] /
    /// [`Client::scp_recv_from`] to drive the remote `scp -t` / `scp -f`
    /// helper over the channel. For one-shot commands whose output you just
    /// want to collect, use [`Client::exec`] instead.
    #[cfg_attr(
        feature = "multichannel",
        deprecated(
            since = "0.0.2",
            note = "Use SharedClient::exec_stream for multi-channel support."
        )
    )]
    pub fn exec_stream(&mut self, command: &str) -> Result<ClientChannelStream<'_>> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::Session)?;
        self.write_payload(&open_payload)?;

        let mut opened = false;
        let mut iter_guard = 0usize;
        while !opened {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("exec_stream: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => opened = true,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("exec_stream: channel open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        self.maybe_send_auth_agent_req(local_id)?;
        self.maybe_send_x11_req(local_id)?;
        self.maybe_send_env(local_id)?;
        self.maybe_send_pty_req(local_id)?;

        let exec_req = self.conn.send_request(
            local_id,
            ChannelRequest::Exec {
                command: command.into(),
            },
            true,
        )?;
        self.write_payload(&exec_req)?;
        self.await_request_reply(local_id, "exec")?;

        Ok(ClientChannelStream {
            client: self,
            channel: local_id,
            read_buf: Vec::new(),
            stderr_buf: Vec::new(),
            remote_eof: false,
            local_close_sent: false,
        })
    }

    /// Open a `direct-tcpip` channel (RFC 4254 §7.2) that asks the server to
    /// connect to `dest_host:dest_port` and proxy bytes. Returns a
    /// `Read + Write` stream borrowing the client for the channel's
    /// lifetime; dropping it closes the channel.
    ///
    /// `orig_host`/`orig_port` are informational (the server logs them but
    /// makes no routing decision on them); pass `("127.0.0.1", 0)` if you
    /// don't have a meaningful source.
    ///
    /// Like [`Client::sftp`] / [`Client::exec`], this is a single-channel
    /// helper: while the returned stream is alive, the client cannot be
    /// used for anything else. Multi-channel multiplexing comes later via
    /// the `Client::serve` event loop.
    #[cfg_attr(
        feature = "multichannel",
        deprecated(
            since = "0.0.2",
            note = "Use SharedClient::open_direct_tcpip for multi-channel support."
        )
    )]
    pub fn open_direct_tcpip(
        &mut self,
        dest_host: &str,
        dest_port: u16,
        orig_host: &str,
        orig_port: u16,
    ) -> Result<ClientChannelStream<'_>> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::DirectTcpip {
            dest_host: dest_host.to_string(),
            dest_port: dest_port as u32,
            orig_host: orig_host.to_string(),
            orig_port: orig_port as u32,
        })?;
        self.write_payload(&open_payload)?;

        let mut iter_guard = 0usize;
        loop {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol("direct-tcpip: open loop did not converge"));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => break,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("direct-tcpip: open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        Ok(ClientChannelStream {
            client: self,
            channel: local_id,
            read_buf: Vec::new(),
            stderr_buf: Vec::new(),
            remote_eof: false,
            local_close_sent: false,
        })
    }

    /// Open a `direct-streamlocal@openssh.com` channel (OpenSSH extension; the
    /// Unix-socket analog of [`Self::open_direct_tcpip`], the wire side of
    /// `ssh -L local:/remote.sock`). Asks the server to connect to
    /// `socket_path` and proxies bytes over the returned channel.
    ///
    /// Single-channel helper: while the returned stream is alive the client
    /// cannot be used for anything else. For multi-channel use, drive
    /// [`Self::serve`] and call
    /// [`ServeContext::open_direct_streamlocal`] from another thread.
    #[cfg_attr(
        feature = "multichannel",
        deprecated(
            since = "0.0.7",
            note = "Use ServeContext::open_direct_streamlocal for multi-channel support."
        )
    )]
    pub fn open_direct_streamlocal(
        &mut self,
        socket_path: &str,
    ) -> Result<ClientChannelStream<'_>> {
        let (local_id, open_payload) = self.conn.open(ChannelOpen::DirectStreamlocal {
            socket_path: socket_path.to_string(),
        })?;
        self.write_payload(&open_payload)?;

        let mut iter_guard = 0usize;
        loop {
            iter_guard += 1;
            if iter_guard > MAX_EXEC_ITER {
                return Err(Error::Protocol(
                    "direct-streamlocal: open loop did not converge",
                ));
            }
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::OpenConfirmed { channel } if channel == local_id => break,
                ChannelEvent::OpenFailed { channel, .. } if channel == local_id => {
                    return Err(Error::Protocol("direct-streamlocal: open failed"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }

        Ok(ClientChannelStream {
            client: self,
            channel: local_id,
            read_buf: Vec::new(),
            stderr_buf: Vec::new(),
            remote_eof: false,
            local_close_sent: false,
        })
    }

    /// Upload one or more local sources to a remote destination via SCP.
    ///
    /// Issues `scp -t [-r] [-p] -- <quoted dest>` on the peer (a remote
    /// `scp` binary, or our own `sshd`'s in-process [`crate::server::ExecStreamHandler`]
    /// reading the SCP protocol). For each source we then drive
    /// [`crate::scp::Sender::send_path`] over the channel.
    ///
    /// Remote-path quoting refuses `'`, `\n`, and `\0` to prevent shell
    /// injection on the remote side. On error, any buffered server-side
    /// stderr is appended to the message via
    /// [`ClientChannelStream::take_stderr`].
    pub fn scp_send_to(
        &mut self,
        sources: &[&std::path::Path],
        remote_dest: &str,
        opts: crate::scp::ScpSendOptions,
    ) -> Result<()> {
        let cmd = build_scp_to_cmd(remote_dest, &opts)?;
        // Internal scp-wrapping path: uses the borrow-based exec_stream
        // because the helper itself is single-channel anyway. A future
        // SharedClient::scp_send_to could replace this; for now we
        // suppress the in-crate deprecation warning here.
        #[allow(deprecated)]
        let mut stream = self.exec_stream(&cmd)?;
        let result = (|| -> Result<()> {
            let mut sender = crate::scp::Sender::new(&mut stream)
                .map_err(|e| scp_proto(e, "scp_send_to: handshake"))?;
            for src in sources {
                sender
                    .send_path(src, &opts)
                    .map_err(|e| scp_proto(e, "scp_send_to: send_path"))?;
            }
            Ok(())
        })();
        // Drain any stderr the remote scp printed; surface it on error.
        let stderr = stream.take_stderr();
        match result {
            Ok(()) => Ok(()),
            Err(e) => {
                if !stderr.is_empty() {
                    let msg = sanitize_for_terminal(&stderr);
                    eprintln!("scp_send_to: remote stderr: {}", msg.trim());
                }
                Err(e)
            }
        }
    }

    /// Download from a remote source to a local destination via SCP.
    ///
    /// Issues `scp -f [-r] [-p] -- <quoted source>` on the peer, then
    /// drives [`crate::scp::Receiver::run`] over the channel.
    ///
    /// `local_dest` is interpreted as a target *directory* unless
    /// `opts.recursive` is false AND `local_dest` doesn't exist as a
    /// directory — in which case it's treated as the literal file
    /// path (matches `scp remote:foo /tmp/bar`). To force file-target
    /// behaviour set the recv options' `target_is_file` flag.
    pub fn scp_recv_from(
        &mut self,
        remote_source: &str,
        local_dest: &std::path::Path,
        mut opts: crate::scp::ScpRecvOptions,
    ) -> Result<()> {
        let cmd = build_scp_from_cmd(remote_source, &opts)?;
        // If the local target is not an existing directory and the
        // caller hasn't said otherwise, treat it as a file target.
        if !opts.target_is_file && !opts.recursive {
            if let Ok(md) = std::fs::metadata(local_dest) {
                if !md.is_dir() {
                    opts.target_is_file = true;
                }
            } else {
                opts.target_is_file = true;
            }
        }
        // Residual CVE-2019-6111: when this is a single, non-recursive
        // fetch of a literal path (no glob metacharacters), we know the
        // exact basename the server is supposed to return. Pass it to the
        // receiver so it can reject a server that pushes extra or renamed
        // files. For recursive or glob requests the basename set is
        // unpredictable, so we leave the receiver in its confinement-only
        // mode (validate_name + guard_path).
        let expected_name: Option<String> = if opts.recursive {
            None
        } else {
            let base = remote_source.rsplit('/').next().unwrap_or(remote_source);
            // Glob metacharacters mean the server legitimately chooses the
            // names — don't pin a single expected basename in that case.
            let is_glob = base.contains(['*', '?', '[']);
            if base.is_empty() || is_glob {
                None
            } else {
                Some(base.to_string())
            }
        };
        // Matches scp_send_to: single-channel helper, internal use.
        #[allow(deprecated)]
        let mut stream = self.exec_stream(&cmd)?;
        let result = (|| -> Result<()> {
            let mut recv = crate::scp::Receiver::new(&mut stream, local_dest, opts)
                .map_err(|e| scp_proto(e, "scp_recv_from: handshake"))?
                .with_expected_name(expected_name.as_deref());
            recv.run().map_err(|e| scp_proto(e, "scp_recv_from: run"))?;
            Ok(())
        })();
        let stderr = stream.take_stderr();
        match result {
            Ok(()) => Ok(()),
            Err(e) => {
                if !stderr.is_empty() {
                    let msg = sanitize_for_terminal(&stderr);
                    eprintln!("scp_recv_from: remote stderr: {}", msg.trim());
                }
                Err(e)
            }
        }
    }

    /// Send a `tcpip-forward` global request (RFC 4254 §7.1; the
    /// outbound bookend of `ssh -R`) and block until the server replies.
    ///
    /// `bind_port == 0` asks the server to pick a port; the actual port
    /// is returned. Any other value asks the server to bind that exact
    /// port and is returned verbatim on success.
    ///
    /// Returns `Err(Error::Protocol(_))` if the server replies
    /// `REQUEST_FAILURE` or if the reply tail is malformed.
    ///
    /// Note: at the time of writing the server side is bind-and-drop —
    /// connections accepted on the bound port are closed immediately.
    /// End-to-end byte forwarding (server-initiated `forwarded-tcpip`
    /// channel-opens back to the client) lands in a follow-up commit.
    pub fn request_tcpip_forward(&mut self, bind_address: &str, bind_port: u16) -> Result<u16> {
        use crate::channel::GlobalRequest;
        if self.host_key_mismatched() {
            return Err(Error::Config(HOST_KEY_MISMATCH_DISABLED));
        }
        let payload = self.conn.send_global_request(
            GlobalRequest::TcpipForward {
                bind_address: bind_address.to_string(),
                bind_port: bind_port as u32,
            },
            true,
        );
        self.write_payload(&payload)?;
        let data = self.await_global_reply("tcpip-forward")?;
        let granted_port = if bind_port == 0 {
            let mut r = crate::format::Reader::new(&data);
            let p = r
                .read_u32()
                .map_err(|_| Error::Protocol("tcpip-forward: server omitted assigned-port tail"))?;
            if p > u16::MAX as u32 {
                return Err(Error::Protocol(
                    "tcpip-forward: server returned out-of-range port",
                ));
            }
            p as u16
        } else {
            bind_port
        };
        // Record the grant so `serve` can refuse unsolicited
        // `forwarded-tcpip` opens (see `tcpip_forward_grants`).
        self.tcpip_forward_grants
            .push((bind_address.to_string(), granted_port));
        Ok(granted_port)
    }

    /// Send a `cancel-tcpip-forward` global request (RFC 4254 §7.1) and
    /// block until the server replies. The `(bind_address, bind_port)`
    /// pair must match a previous successful `request_tcpip_forward`.
    pub fn cancel_tcpip_forward(&mut self, bind_address: &str, bind_port: u16) -> Result<()> {
        use crate::channel::GlobalRequest;
        let payload = self.conn.send_global_request(
            GlobalRequest::CancelTcpipForward {
                bind_address: bind_address.to_string(),
                bind_port: bind_port as u32,
            },
            true,
        );
        self.write_payload(&payload)?;
        let _ = self.await_global_reply("cancel-tcpip-forward")?;
        // Drop the matching grant so a later unsolicited forward is refused.
        // Match on (address, port); if the exact pair isn't found (e.g. the
        // caller cancelled by assigned port under a different address
        // spelling), fall back to dropping one grant for the same port.
        if let Some(idx) = self
            .tcpip_forward_grants
            .iter()
            .position(|(a, p)| a == bind_address && *p == bind_port)
            .or_else(|| {
                self.tcpip_forward_grants
                    .iter()
                    .position(|(_, p)| *p == bind_port)
            })
        {
            self.tcpip_forward_grants.swap_remove(idx);
        }
        Ok(())
    }

    /// Send a `streamlocal-forward@openssh.com` global request (OpenSSH
    /// extension; the Unix-socket analog of [`Self::request_tcpip_forward`],
    /// the outbound bookend of `ssh -R /remote.sock:...`) and block until the
    /// server replies.
    ///
    /// On success the server binds a Unix-domain listener at `socket_path` and
    /// will open `forwarded-streamlocal@openssh.com` channels back for each
    /// accepted connection — handle them via
    /// [`ClientHandlers::on_forwarded_streamlocal`] in [`Self::serve`].
    ///
    /// Returns `Err(Error::Protocol(_))` if the server replies
    /// `REQUEST_FAILURE`.
    pub fn request_streamlocal_forward(&mut self, socket_path: &str) -> Result<()> {
        use crate::channel::GlobalRequest;
        if self.host_key_mismatched() {
            return Err(Error::Config(HOST_KEY_MISMATCH_DISABLED));
        }
        let payload = self.conn.send_global_request(
            GlobalRequest::StreamlocalForward {
                socket_path: socket_path.to_string(),
            },
            true,
        );
        self.write_payload(&payload)?;
        let _ = self.await_global_reply("streamlocal-forward@openssh.com")?;
        self.streamlocal_forward_grants
            .push(socket_path.to_string());
        Ok(())
    }

    /// Send a `cancel-streamlocal-forward@openssh.com` global request (OpenSSH
    /// extension) and block until the server replies. `socket_path` must match
    /// a previous successful [`Self::request_streamlocal_forward`].
    pub fn cancel_streamlocal_forward(&mut self, socket_path: &str) -> Result<()> {
        use crate::channel::GlobalRequest;
        let payload = self.conn.send_global_request(
            GlobalRequest::CancelStreamlocalForward {
                socket_path: socket_path.to_string(),
            },
            true,
        );
        self.write_payload(&payload)?;
        let _ = self.await_global_reply("cancel-streamlocal-forward@openssh.com")?;
        if let Some(idx) = self
            .streamlocal_forward_grants
            .iter()
            .position(|p| p == socket_path)
        {
            self.streamlocal_forward_grants.swap_remove(idx);
        }
        Ok(())
    }

    /// Multi-channel event loop. Run after a [`Self::request_tcpip_forward`]
    /// so server-initiated `forwarded-tcpip` channel opens land in
    /// [`ClientHandlers::on_forwarded_tcpip`].
    ///
    /// The loop polls the socket with a small read timeout
    /// (`SERVE_POLL_INTERVAL`, currently 50ms) so it can interleave wire reads
    /// with per-channel egress draining. Returns:
    ///
    /// - `Ok(())` once `handlers.stop` has been set AND every accepted
    ///   channel has been torn down (matching the server's
    ///   `do_connection_loop` exit condition).
    /// - `Err(Error::Protocol(_))` on protocol violation.
    /// - `Err(Error::Io(_))` if the peer hangs up the socket.
    ///
    /// Channel opens whose handler is unset are rejected with
    /// `SSH_OPEN_ADMINISTRATIVELY_PROHIBITED`; unrelated channel-events
    /// (e.g. for `direct-tcpip` channels the user opened via
    /// [`Self::open_direct_tcpip`] before calling `serve`) are NOT
    /// dispatched — those types own their own
    /// [`ClientChannelStream`] which would also try to drain the socket.
    /// In other words: while `serve` is running, no other Client method
    /// may be used on this client.
    pub fn serve(&mut self, handlers: ClientHandlers) -> Result<()> {
        let mut runtimes: BTreeMap<u32, ServeRuntime> = BTreeMap::new();
        let mut pending_opens: BTreeMap<u32, PendingOutboundOpen> = BTreeMap::new();
        // Always poll with a small read timeout so the loop responds to
        // `handlers.stop` even when no channels are open (e.g. between
        // forwarded connections). Reverted on return.
        let _ = self.stream.set_read_timeout(Some(SERVE_POLL_INTERVAL));
        let mut steps = 0usize;
        // Keepalive bookkeeping (ServerAliveInterval / ServerAliveCountMax).
        // `last_activity` advances on every inbound packet; `probes_pending`
        // counts unanswered keepalives. Both are inert when keepalive is off.
        let mut last_activity = Instant::now();
        let mut probes_pending: u32 = 0;
        let result = loop {
            steps += 1;
            if steps > MAX_SERVE_STEPS {
                break Err(Error::Protocol("serve: step cap exceeded"));
            }

            // Keepalive: after `interval` of silence, probe the peer. After
            // `count_max` unanswered probes, treat the connection as dead.
            if let Some((interval, count_max)) = self.keepalive
                && !self.driver.is_kexing()
                && last_activity.elapsed() >= interval
            {
                if probes_pending >= count_max {
                    break Err(Error::Protocol(
                        "serve: server keepalive timed out (ServerAliveCountMax exceeded)",
                    ));
                }
                let probe = self
                    .conn
                    .send_global_request(crate::channel::GlobalRequest::Keepalive, true);
                if let Err(e) = self.write_payload(&probe) {
                    break Err(e);
                }
                probes_pending += 1;
                // Reset the silence timer so we space probes by `interval`
                // rather than spamming once the threshold is first crossed.
                last_activity = Instant::now();
            }

            // Per-tick: process any outbound open requests from external
            // threads (ssh -L listener accepts, etc.). The driver buffers and
            // replays app payloads received during a re-KEX itself, so the
            // frontend no longer tracks a deferred queue.
            if !self.driver.is_kexing()
                && let Some(rx) = handlers.cmd_rx.as_ref()
                && let Err(e) = serve_drain_commands(self, rx, &mut pending_opens)
            {
                break Err(e);
            }

            // Per-tick: retry parked ingress (crediting the receive window
            // only for what the handlers actually took), ship pending egress
            // from each runtime onto the wire, then reap any runtime whose
            // close has been fully emitted.
            if !self.driver.is_kexing() {
                if let Err(e) = serve_flush_ingress(self, &mut runtimes) {
                    break Err(e);
                }
                if let Err(e) = serve_drain_runtimes(self, &mut runtimes) {
                    break Err(e);
                }
                runtimes.retain(|_, rt| !rt.close_sent);
            }

            // Exit when caller asked us to stop AND every channel has been
            // torn down. Without this guard a `stop` mid-handshake would
            // leave the channel in a half-closed state on the peer.
            if handlers.stop.load(Ordering::SeqCst)
                && runtimes.is_empty()
                && pending_opens.is_empty()
            {
                break Ok(());
            }

            // RFC 4253 §9 re-key is driven inside the driver's `handle_timeout`
            // (invoked by `read_one_packet_maybe_timeout` below), so the
            // frontend no longer initiates it explicitly.
            let payload = match self.read_one_packet_maybe_timeout() {
                Ok(Some(p)) => p,
                Ok(None) => continue, // tick; re-enter drain/stop checks
                Err(e) => break Err(e),
            };
            // Any inbound packet — including the keepalive's own
            // REQUEST_SUCCESS/FAILURE reply — proves the peer is alive, so
            // clear the probe counter and restart the silence timer.
            if self.keepalive.is_some() {
                probes_pending = 0;
                last_activity = Instant::now();
            }
            if let Err(e) =
                serve_dispatch_packet(self, &handlers, &mut runtimes, &mut pending_opens, &payload)
            {
                break Err(e);
            }
        };

        // Cleanup: drop any remaining runtimes (closes their ingress mpsc so
        // handler threads see EOF and exit), revert the socket timeout.
        // Pending outbound opens get an Err reply so callers don't hang.
        let stale_opens = core::mem::take(&mut pending_opens);
        for (_ch, po) in stale_opens {
            let _ = po.reply.send(Err(Error::Protocol("serve loop terminated")));
        }
        runtimes.clear();
        let _ = self.stream.set_read_timeout(None);
        result
    }

    /// Whether a packet is already decoded and waiting (so there is no need
    /// to wait for the transport).
    #[cfg(feature = "multichannel")]
    pub(crate) fn has_buffered_input(&self) -> bool {
        self.driver.has_pending_events()
    }

    /// Split the read side off the transport ([`Transport::split_reader`]),
    /// from then on reading through it; `None` when the transport cannot.
    #[cfg(feature = "multichannel")]
    pub(crate) fn split_reader(&mut self) -> Option<Arc<SplitReader>> {
        if self.split_reader.is_none() {
            let src = self.stream.split_reader()?;
            self.split_reader = Some(Arc::new(SplitReader {
                state: Mutex::new(SplitState {
                    src,
                    pending: Vec::new(),
                    eof: false,
                }),
            }));
        }
        self.split_reader.clone()
    }

    /// Like [`Self::read_one_packet`] but returns `Ok(None)` on a read
    /// timeout (the socket's `set_read_timeout` having elapsed) instead of
    /// erroring. Used by [`Self::serve`] to interleave wire reads with
    /// per-channel egress draining, and by the `SharedClient` pump in
    /// short-timeout (interactive) mode to release its inner mutex
    /// between reads so siblings can write.
    pub(crate) fn read_one_packet_maybe_timeout(&mut self) -> Result<Option<Vec<u8>>> {
        timeout_as_none(self.read_one_packet())
    }

    /// Like [`Self::read_one_packet_maybe_timeout`], but only consumes input
    /// the split-off reader already stashed ([`SplitReader::fill`]): never
    /// waits on the transport, so it is safe under the `SharedClient` lock.
    #[cfg(feature = "multichannel")]
    pub(crate) fn read_stashed_packet(&mut self) -> Result<Option<Vec<u8>>> {
        timeout_as_none(self.next_packet(true))
    }

    /// Block until the server answers a `GLOBAL_REQUEST` with
    /// `want_reply = true`. Returns the request-specific tail bytes
    /// (empty for requests without a payload).
    fn await_global_reply(&mut self, what: &'static str) -> Result<Vec<u8>> {
        for _ in 0..MAX_EXEC_ITER {
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::GlobalSuccess { data } => return Ok(data),
                ChannelEvent::GlobalFailure => {
                    let _ = what; // for future tracing
                    return Err(Error::Protocol("global request denied"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }
        Err(Error::Protocol(
            "global request: reply loop did not converge",
        ))
    }

    /// Block until the peer answers a single `CHANNEL_REQUEST` we sent
    /// with `want_reply = true`. Used by [`shell_with_stdin`] to gate the
    /// pty-req → shell handoff.
    ///
    /// [`shell_with_stdin`]: Self::shell_with_stdin
    pub(crate) fn await_request_reply(&mut self, channel: u32, what: &'static str) -> Result<()> {
        for _ in 0..MAX_EXEC_ITER {
            let payload = self.read_one_packet()?;
            match self.conn.on_packet(&payload)? {
                ChannelEvent::Success { channel: c } if c == channel => return Ok(()),
                ChannelEvent::Failure { channel: c } if c == channel => {
                    let _ = what; // for future tracing
                    return Err(Error::Protocol("shell: channel request denied"));
                }
                other => self.handle_unsolicited(&other)?,
            }
        }
        Err(Error::Protocol(
            "shell: request-reply loop did not converge",
        ))
    }

    /// Move every frame the driver has queued into the outbound buffer, then
    /// flush as much of that buffer to the wire as the transport will accept.
    pub(crate) fn pump_out(&mut self) -> Result<()> {
        while let Some(frame) = self.driver.poll_transmit() {
            self.out_buf.extend(frame);
        }
        self.flush_out_buf()
    }

    /// Push buffered outbound bytes to the transport until either the buffer
    /// drains or the transport refuses more.
    ///
    /// With no write timeout configured (the default, single-channel blocking
    /// path) a socket `write` never reports `WouldBlock`, so this drains the
    /// buffer completely — behaviourally identical to the old
    /// `write_all`-per-frame loop.
    ///
    /// With a write timeout configured (set alongside the read timeout by the
    /// concurrent-pump consumers — `SharedClient`'s mux splice and the serve
    /// loop) a congested write returns `WouldBlock` / `TimedOut` once the
    /// kernel send buffer is full. We treat that as backpressure and leave the
    /// remainder queued: the caller releases the shared transport lock, a
    /// sibling thread pumps inbound (the only way the peer's receive window —
    /// and thus our send buffer — reopens), and the next `pump_out` flushes
    /// the backlog. Without this a bidirectionally-saturated connection
    /// deadlocks, both peers blocked in `write` with neither reading.
    fn flush_out_buf(&mut self) -> Result<()> {
        while !self.out_buf.is_empty() {
            // `as_slices().0` is the leading contiguous run; non-empty
            // whenever the deque is non-empty.
            let front_len = self.out_buf.as_slices().0.len();
            let n = match self.stream.write(self.out_buf.as_slices().0) {
                Ok(0) => return Err(Error::Protocol("connection closed")),
                Ok(n) => n,
                Err(ref e) if e.kind() == ErrorKind::Interrupted => continue,
                Err(ref e)
                    if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut =>
                {
                    // Send buffer full: keep the remainder for the next pump.
                    return Ok(());
                }
                Err(e) => return Err(Error::Io(e)),
            };
            debug_assert!(n <= front_len);
            self.out_buf.drain(..n);
        }
        Ok(())
    }

    /// Read one chunk off the transport and feed it to the driver. Propagates
    /// the socket's `WouldBlock`/`TimedOut` error so timeout-aware callers can
    /// convert it to `Ok(None)`.
    fn read_into_driver(&mut self, stashed_only: bool) -> Result<()> {
        let mut tmp = [0u8; 16 * 1024];
        #[cfg(feature = "multichannel")]
        let n = match &self.split_reader {
            Some(r) => r.read(&mut tmp, self.read_timeout, stashed_only)?,
            None => self.stream.read(&mut tmp)?,
        };
        #[cfg(not(feature = "multichannel"))]
        let n = {
            let _ = stashed_only;
            self.stream.read(&mut tmp)?
        };
        if n == 0 {
            return Err(Error::Protocol("connection closed"));
        }
        self.driver.handle_input(&tmp[..n], Instant::now())?;
        Ok(())
    }

    /// Pump the transport until the handshake (version exchange + first KEX)
    /// completes.
    fn drive_handshake(&mut self) -> Result<()> {
        for _ in 0..MAX_KEX_STEPS.saturating_mul(MAX_BANNER_LINES + 4) {
            self.pump_out()?;
            while let Some(ev) = self.driver.poll_event() {
                if matches!(ev, Event::HandshakeComplete) {
                    self.pump_out()?;
                    return Ok(());
                }
            }
            self.read_into_driver(false)?;
        }
        Err(Error::Protocol("kex: too many steps"))
    }

    /// Block until the driver yields the next application payload (post-auth
    /// connection-protocol packet). Transport concerns (KEX, re-key, EXT_INFO,
    /// PING/PONG) are handled inside the driver and never surface here. The
    /// returned bytes are fed to `self.conn` by the caller, exactly as before.
    pub(crate) fn read_one_packet(&mut self) -> Result<Vec<u8>> {
        self.next_packet(false)
    }

    /// [`Self::read_one_packet`]; `stashed_only` as for [`Self::read_stashed_packet`].
    fn next_packet(&mut self, stashed_only: bool) -> Result<Vec<u8>> {
        loop {
            // Tick re-key / keepalive timers before each (possibly blocking or
            // timing-out) read so they advance even when the wire is idle.
            self.driver.handle_timeout(Instant::now())?;
            self.pump_out()?;
            while let Some(ev) = self.driver.poll_event() {
                if let Event::AppData(payload) = ev {
                    self.pump_out()?;
                    return Ok(payload);
                }
                // HandshakeComplete doesn't occur on this (post-handshake) path.
            }
            self.read_into_driver(stashed_only)?;
        }
    }

    /// Encode `payload` and send it. The sans-IO driver owns the codec, so this
    /// queues the frame and flushes the outbound queue immediately (preserving
    /// the old eager-send semantics).
    /// Start a key re-exchange now (`ssh`'s `~R`), unless one is running.
    pub fn rekey(&mut self) -> Result<()> {
        self.driver.request_rekey()?;
        self.pump_out()
    }

    pub(crate) fn write_payload(&mut self, payload: &[u8]) -> Result<()> {
        self.driver.enqueue_payload(payload)?;
        self.pump_out()
    }

    /// Send a `ping@openssh.com` `SSH2_MSG_PING` carrying `data` over the
    /// transport. The peer answers with a `SSH2_MSG_PONG` echoing `data`,
    /// which the read loop drops. Used as constant-rate "chaff" by the
    /// `ObscureKeystrokeTiming` sender in the `ssh` binary. Must not be
    /// called while a KEX is in flight.
    #[cfg(any(test, feature = "multichannel"))]
    pub(crate) fn send_transport_ping(&mut self, data: &[u8]) -> Result<()> {
        let ping = encode_ping(data);
        self.write_payload(&ping)
    }
}

/// Read+Write adapter wrapping a single open channel on a [`Client`],
/// driving the underlying SSH packet loop on every `read` / `write`. Used
/// by [`Client::sftp`] to feed [`SftpClient`] a synchronous transport,
/// and by [`Client::open_direct_tcpip`] to expose the channel as a plain
/// byte stream.
///
/// On `Drop` the channel is closed (CHANNEL_CLOSE is sent; the matching
/// peer CLOSE is best-effort drained).
///
/// Extended-data (stderr) is **buffered** rather than dropped — pump_one
/// appends each `SSH_EXTENDED_DATA_STDERR` payload into an internal buffer
/// (still replenishing the window). Callers that want server-side
/// diagnostic output (notably [`Client::scp_send_to`] /
/// [`Client::scp_recv_from`]) drain it via [`Self::take_stderr`]. The SFTP
/// subsystem doesn't emit extended data, so the buffer stays empty there.
/// The buffer is capped at 1 MiB: once full, the **oldest** bytes are
/// discarded so the tail (where a remote error message lands) is kept and
/// a peer cannot exhaust memory through stderr.
pub struct ClientChannelStream<'a> {
    client: &'a mut Client,
    channel: u32,
    read_buf: Vec<u8>,
    stderr_buf: Vec<u8>,
    remote_eof: bool,
    local_close_sent: bool,
}

impl ClientChannelStream<'_> {
    /// Take ownership of any accumulated server-side stderr (extended-data
    /// channel) bytes. Leaves the buffer empty. Used by SCP error paths to
    /// surface the remote `scp` binary's diagnostic output back to the
    /// caller; for SFTP / direct-tcpip channels the buffer is normally
    /// empty.
    pub fn take_stderr(&mut self) -> Vec<u8> {
        core::mem::take(&mut self.stderr_buf)
    }

    /// Drive the SSH packet loop until either `read_buf` has bytes available
    /// or the peer closes the channel. Window-adjust packets are handled
    /// transparently; extended-data (any code) is buffered into
    /// [`Self::stderr_buf`].
    fn pump_one(&mut self) -> std::io::Result<()> {
        let payload = self.client.read_one_packet().map_err(io_err)?;
        let ev = self.client.conn.on_packet(&payload).map_err(io_err)?;
        match ev {
            ChannelEvent::Data { channel, data } if channel == self.channel => {
                let n = data.len() as u32;
                self.read_buf.extend_from_slice(&data);
                if let Some(adj) = self
                    .client
                    .conn
                    .replenish_window(self.channel, n)
                    .map_err(io_err)?
                {
                    self.client.write_payload(&adj).map_err(io_err)?;
                }
            }
            ChannelEvent::ExtendedData {
                channel,
                code: _,
                data,
            } if channel == self.channel => {
                let n = data.len() as u32;
                // Bounded: keep only the newest `STREAM_STDERR_CAP` bytes
                // (the tail carries the useful diagnostics) so a chatty or
                // malicious peer cannot grow this buffer without limit. The
                // window is still credited so the channel keeps flowing.
                self.stderr_buf.extend_from_slice(&data);
                if self.stderr_buf.len() > STREAM_STDERR_CAP {
                    let excess = self.stderr_buf.len() - STREAM_STDERR_CAP;
                    self.stderr_buf.drain(..excess);
                }
                if let Some(adj) = self
                    .client
                    .conn
                    .replenish_window(self.channel, n)
                    .map_err(io_err)?
                {
                    self.client.write_payload(&adj).map_err(io_err)?;
                }
            }
            ChannelEvent::Eof { channel } if channel == self.channel => {
                self.remote_eof = true;
            }
            ChannelEvent::Close { channel } if channel == self.channel => {
                self.remote_eof = true;
                if !self.local_close_sent {
                    let p = self.client.conn.send_close(self.channel).map_err(io_err)?;
                    self.client.write_payload(&p).map_err(io_err)?;
                    self.local_close_sent = true;
                }
            }
            other => self.client.handle_unsolicited(&other).map_err(io_err)?,
        }
        Ok(())
    }
}

impl Read for ClientChannelStream<'_> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        // Block until we have something in the buffer, or the peer is done.
        while self.read_buf.is_empty() && !self.remote_eof {
            self.pump_one()?;
        }
        if self.read_buf.is_empty() {
            return Ok(0);
        }
        let n = core::cmp::min(buf.len(), self.read_buf.len());
        buf[..n].copy_from_slice(&self.read_buf[..n]);
        self.read_buf.drain(..n);
        Ok(n)
    }
}

impl Write for ClientChannelStream<'_> {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        if buf.is_empty() {
            return Ok(0);
        }
        // Loop until we manage to put something on the wire. `send_data`
        // returns `taken == 0` when the remote window is full; we then read
        // packets (which will eventually arrive with a WindowAdjust) until
        // it opens.
        loop {
            let (payload, taken) = self
                .client
                .conn
                .send_data(self.channel, buf)
                .map_err(io_err)?;
            if taken > 0 {
                self.client.write_payload(&payload).map_err(io_err)?;
                return Ok(taken);
            }
            // Window full — pump one packet. If the peer closed we can't
            // make progress; surface as broken pipe.
            if self.remote_eof {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "channel closed by peer mid-write",
                ));
            }
            self.pump_one()?;
        }
    }

    fn flush(&mut self) -> std::io::Result<()> {
        // All writes are flushed inside `write_payload` via `write_all`.
        Ok(())
    }
}

impl Drop for ClientChannelStream<'_> {
    fn drop(&mut self) {
        // Best-effort tear-down: send EOF + CLOSE, then drain a few packets
        // so the peer's matching CLOSE doesn't get stranded in the inbox.
        if !self.local_close_sent {
            if let Ok(p) = self.client.conn.send_eof(self.channel) {
                let _ = self.client.write_payload(&p);
            }
            if let Ok(p) = self.client.conn.send_close(self.channel) {
                let _ = self.client.write_payload(&p);
            }
            self.local_close_sent = true;
        }
        // Drain up to ~MAX_DRAIN packets so the peer's CLOSE is acknowledged.
        const MAX_DRAIN: usize = 128;
        for _ in 0..MAX_DRAIN {
            if self.remote_eof {
                break;
            }
            if self.pump_one().is_err() {
                break;
            }
        }
    }
}

pub(crate) fn io_err(e: Error) -> std::io::Error {
    match e {
        Error::Io(io) => io,
        // Keep the error itself (callers can downcast it, e.g. to report a
        // server's SSH_MSG_DISCONNECT the way OpenSSH does).
        other => std::io::Error::other(other),
    }
}

/// Encode a local Unix termios into the RFC 4254 §8 terminal-modes
/// blob expected by the `pty-req` channel request. Each significant
/// opcode is followed by a u32 value; the blob is terminated by
/// `TTY_OP_END = 0`. Pass the result through to
/// the `SharedClient::shell_stream` mirror as the `modes` parameter so
/// the remote PTY mirrors the local control characters (e.g. `^C` to
/// interrupt) and flag set (canonical mode, echo, etc.).
///
/// The full opcode list and wire format live in RFC 4254 §8. We emit
/// the opcodes OpenSSH considers core: control characters
/// `VINTR`/`VQUIT`/…/`VEOL2`, IXON/IXOFF/IXANY, ICANON/ECHO/ISIG and
/// friends, OPOST/ONLCR/OCRNL/ONLRET, CS7/CS8/PARENB/PARODD, plus the
/// two `TTY_OP_*SPEED` bauds. Unknown / non-portable opcodes are
/// skipped — the peer ignores opcodes it doesn't recognise per the
/// RFC.
///
/// Only available on Unix builds with the default `std` feature, since
/// the source `termios` type is itself a libc construct.
#[cfg(all(feature = "std", unix))]
pub fn encode_termios_modes(t: &libc::termios) -> Vec<u8> {
    // Opcode numbers per RFC 4254 §8.
    const TTY_OP_END: u8 = 0;
    const VINTR: u8 = 1;
    const VQUIT: u8 = 2;
    const VERASE: u8 = 3;
    const VKILL: u8 = 4;
    const VEOF: u8 = 5;
    const VEOL: u8 = 6;
    const VEOL2: u8 = 7;
    const VSTART: u8 = 8;
    const VSTOP: u8 = 9;
    const VSUSP: u8 = 10;
    const VREPRINT: u8 = 12;
    const VWERASE: u8 = 13;
    const VLNEXT: u8 = 14;
    const IGNPAR: u8 = 30;
    const PARMRK: u8 = 31;
    const INPCK: u8 = 32;
    const ISTRIP: u8 = 33;
    const INLCR: u8 = 34;
    const IGNCR: u8 = 35;
    const ICRNL: u8 = 36;
    #[cfg(any(target_os = "linux", target_os = "android"))]
    const IUCLC: u8 = 37;
    const IXON: u8 = 38;
    const IXANY: u8 = 39;
    const IXOFF: u8 = 40;
    const IMAXBEL: u8 = 41;
    // RFC 8160.
    #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos"))]
    const IUTF8: u8 = 42;
    const ISIG: u8 = 50;
    const ICANON: u8 = 51;
    const ECHO: u8 = 53;
    const ECHOE: u8 = 54;
    const ECHOK: u8 = 55;
    const ECHONL: u8 = 56;
    const NOFLSH: u8 = 57;
    const TOSTOP: u8 = 58;
    const IEXTEN: u8 = 59;
    const ECHOCTL: u8 = 60;
    const ECHOKE: u8 = 61;
    const OPOST: u8 = 70;
    const ONLCR: u8 = 72;
    const OCRNL: u8 = 73;
    const ONOCR: u8 = 74;
    const ONLRET: u8 = 75;
    const CS7: u8 = 90;
    const CS8: u8 = 91;
    const PARENB: u8 = 92;
    const PARODD: u8 = 93;
    const TTY_OP_ISPEED: u8 = 128;
    const TTY_OP_OSPEED: u8 = 129;

    let mut out = Vec::with_capacity(128);
    let mut push = |op: u8, val: u32| {
        out.push(op);
        out.extend_from_slice(&val.to_be_bytes());
    };

    // Control characters. Each is one byte in `c_cc`; the wire field
    // is u32 so we zero-extend. Skip if the slot is at the libc
    // sentinel (0xff means "disabled" on Linux/macOS, but we just
    // forward whatever is there — the remote can interpret it).
    let cc = |i: usize| -> u32 { t.c_cc[i] as u32 };
    push(VINTR, cc(libc::VINTR));
    push(VQUIT, cc(libc::VQUIT));
    push(VERASE, cc(libc::VERASE));
    push(VKILL, cc(libc::VKILL));
    push(VEOF, cc(libc::VEOF));
    push(VEOL, cc(libc::VEOL));
    push(VEOL2, cc(libc::VEOL2));
    push(VSTART, cc(libc::VSTART));
    push(VSTOP, cc(libc::VSTOP));
    push(VSUSP, cc(libc::VSUSP));
    push(VREPRINT, cc(libc::VREPRINT));
    push(VWERASE, cc(libc::VWERASE));
    push(VLNEXT, cc(libc::VLNEXT));

    // Input modes. Wire value is 0 (clear) or non-zero (set) per the
    // RFC; we use 1 for set so the peer sees a tidy boolean.
    let iflag = t.c_iflag;
    let bit_i = |mask: libc::tcflag_t| -> u32 { if iflag & mask != 0 { 1 } else { 0 } };
    push(IGNPAR, bit_i(libc::IGNPAR));
    push(PARMRK, bit_i(libc::PARMRK));
    push(INPCK, bit_i(libc::INPCK));
    push(ISTRIP, bit_i(libc::ISTRIP));
    push(INLCR, bit_i(libc::INLCR));
    push(IGNCR, bit_i(libc::IGNCR));
    push(ICRNL, bit_i(libc::ICRNL));
    push(IXON, bit_i(libc::IXON));
    push(IXANY, bit_i(libc::IXANY));
    push(IXOFF, bit_i(libc::IXOFF));
    push(IMAXBEL, bit_i(libc::IMAXBEL));
    #[cfg(any(target_os = "linux", target_os = "android"))]
    push(IUCLC, bit_i(libc::IUCLC));
    #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos"))]
    push(IUTF8, bit_i(libc::IUTF8));

    // Local modes.
    let lflag = t.c_lflag;
    let bit_l = |mask: libc::tcflag_t| -> u32 { if lflag & mask != 0 { 1 } else { 0 } };
    push(ISIG, bit_l(libc::ISIG));
    push(ICANON, bit_l(libc::ICANON));
    push(ECHO, bit_l(libc::ECHO));
    push(ECHOE, bit_l(libc::ECHOE));
    push(ECHOK, bit_l(libc::ECHOK));
    push(ECHONL, bit_l(libc::ECHONL));
    push(NOFLSH, bit_l(libc::NOFLSH));
    push(TOSTOP, bit_l(libc::TOSTOP));
    push(IEXTEN, bit_l(libc::IEXTEN));
    push(ECHOCTL, bit_l(libc::ECHOCTL));
    push(ECHOKE, bit_l(libc::ECHOKE));

    // Output modes.
    let oflag = t.c_oflag;
    let bit_o = |mask: libc::tcflag_t| -> u32 { if oflag & mask != 0 { 1 } else { 0 } };
    push(OPOST, bit_o(libc::OPOST));
    push(ONLCR, bit_o(libc::ONLCR));
    push(OCRNL, bit_o(libc::OCRNL));
    push(ONOCR, bit_o(libc::ONOCR));
    push(ONLRET, bit_o(libc::ONLRET));

    // Control modes (character size, parity).
    let cflag = t.c_cflag;
    let cs = cflag & libc::CSIZE;
    push(CS7, if cs == libc::CS7 { 1 } else { 0 });
    push(CS8, if cs == libc::CS8 { 1 } else { 0 });
    push(PARENB, if cflag & libc::PARENB != 0 { 1 } else { 0 });
    push(PARODD, if cflag & libc::PARODD != 0 { 1 } else { 0 });

    // Line speeds. The cf*speed accessors require unsafe FFI and the
    // value is cosmetic for a virtual PTY anyway — pin to the RFC's
    // illustrative 38400 baud so no caller has to think about it.
    // (`forbid(unsafe_code)` outside `ffi` rules out the libc path.)
    let _ = t; // suppress unused-binding lint once the cf*speed calls are gone
    push(TTY_OP_ISPEED, 38_400);
    push(TTY_OP_OSPEED, 38_400);

    out.push(TTY_OP_END);
    out
}

/// Build `scp -t [-r] [-p] -- '<dest>'` for [`Client::scp_send_to`].
fn build_scp_to_cmd(remote_dest: &str, opts: &crate::scp::ScpSendOptions) -> Result<String> {
    let quoted = single_quote_for_remote(remote_dest)?;
    let mut s = String::from("scp -t");
    if opts.recursive {
        s.push_str(" -r");
    }
    if opts.preserve_times {
        s.push_str(" -p");
    }
    s.push_str(" -- ");
    s.push_str(&quoted);
    Ok(s)
}

/// Build `scp -f [-r] [-p] -- '<source>'` for [`Client::scp_recv_from`].
fn build_scp_from_cmd(remote_source: &str, opts: &crate::scp::ScpRecvOptions) -> Result<String> {
    let quoted = single_quote_for_remote(remote_source)?;
    let mut s = String::from("scp -f");
    if opts.recursive {
        s.push_str(" -r");
    }
    if opts.preserve_times {
        s.push_str(" -p");
    }
    s.push_str(" -- ");
    s.push_str(&quoted);
    Ok(s)
}

/// Single-quote a remote path for the SSH server's command string.
/// Rejects `'`, `\n`, `\0` — these are characters our own sshd parser
/// can't safely unquote (`'` would terminate the quote; the others can
/// desync most shell parsers). Matches the validation in
/// [`crate::scp::protocol::validate_name`] but on full paths.
fn single_quote_for_remote(p: &str) -> Result<String> {
    if p.contains('\'') {
        return Err(Error::Protocol("scp: remote path contains single quote"));
    }
    if p.contains('\n') {
        return Err(Error::Protocol("scp: remote path contains newline"));
    }
    if p.contains('\0') {
        return Err(Error::Protocol("scp: remote path contains NUL"));
    }
    if p.starts_with('-') {
        return Err(Error::Protocol("scp: remote path starts with '-'"));
    }
    let mut q = String::with_capacity(p.len() + 2);
    q.push('\'');
    q.push_str(p);
    q.push('\'');
    Ok(q)
}

/// Map a [`crate::scp::ScpError`] into the lib's [`Error`] type. Drops
/// the dynamic message string into `Error::Protocol`; the caller has
/// already taken `stream.take_stderr()` so the precise wire error
/// surfaces in `eprintln` rather than the typed error.
fn scp_proto(e: crate::scp::ScpError, _stage: &'static str) -> Error {
    match e {
        crate::scp::ScpError::Io(io) => Error::Io(io),
        crate::scp::ScpError::Remote(_) => Error::Protocol("scp: remote fatal frame"),
        crate::scp::ScpError::Warning(_) => Error::Protocol("scp: remote warning frame"),
        crate::scp::ScpError::BadHeader(_) => Error::Protocol("scp: malformed header"),
        crate::scp::ScpError::BadName(_) => Error::Protocol("scp: invalid name"),
        crate::scp::ScpError::PathEscape => Error::Protocol("scp: path escapes base"),
        crate::scp::ScpError::Unexpected(_) => Error::Protocol("scp: unexpected protocol state"),
    }
}

/// `SHA256:<base64>` fingerprint of a host-key blob, matching
/// `ssh-keygen -lf`. Useful for rendering host-key prompts (see
/// [`HostKeyPrompt`]) or cross-checking a peer's key against
/// `known_hosts` by hand.
pub fn host_key_fingerprint(blob: &[u8]) -> String {
    const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let d = Sha256::digest(blob);
    let bytes: &[u8] = d.as_ref();
    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4 + 7);
    out.push_str("SHA256:");
    let mut i = 0;
    while i + 3 <= bytes.len() {
        let b = ((bytes[i] as u32) << 16) | ((bytes[i + 1] as u32) << 8) | (bytes[i + 2] as u32);
        out.push(ALPHABET[((b >> 18) & 0x3F) as usize] as char);
        out.push(ALPHABET[((b >> 12) & 0x3F) as usize] as char);
        out.push(ALPHABET[((b >> 6) & 0x3F) as usize] as char);
        out.push(ALPHABET[(b & 0x3F) as usize] as char);
        i += 3;
    }
    let rem = bytes.len() - i;
    if rem == 1 {
        let b = (bytes[i] as u32) << 16;
        out.push(ALPHABET[((b >> 18) & 0x3F) as usize] as char);
        out.push(ALPHABET[((b >> 12) & 0x3F) as usize] as char);
    } else if rem == 2 {
        let b = ((bytes[i] as u32) << 16) | ((bytes[i + 1] as u32) << 8);
        out.push(ALPHABET[((b >> 18) & 0x3F) as usize] as char);
        out.push(ALPHABET[((b >> 12) & 0x3F) as usize] as char);
        out.push(ALPHABET[((b >> 6) & 0x3F) as usize] as char);
    }
    out
}

/// Emit OpenSSH's `WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!`
/// banner to stderr, listing every previously-stored fingerprint we
/// would have accepted next to the one the peer is presenting now.
///
/// This is what the user sees on a host-key mismatch regardless of the
/// `on_mismatch` policy: `Reject` then refuses, `AcceptWithWarning`
/// proceeds, `Prompt` asks. The banner itself never changes — only the
/// follow-up wording does.
fn print_mismatch_banner(
    _host: &str,
    _port: u16,
    _expected: &[(String, Vec<u8>)],
    _new_key_type: &str,
    new_key_blob: &[u8],
) {
    // OpenSSH's wording, line for line.
    eprintln!("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
    eprintln!("@    WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED!     @");
    eprintln!("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
    eprintln!("IT IS POSSIBLE THAT SOMEONE IS DOING SOMETHING NASTY!");
    eprintln!("Someone could be eavesdropping on you right now (man-in-the-middle attack)!");
    eprintln!("It is also possible that a host key has just been changed.");
    eprintln!(
        "The fingerprint for the {} key sent by the remote host is\n{}.",
        host_key_type_label(new_key_blob),
        host_key_fingerprint(new_key_blob),
    );
    eprintln!("Please contact your system administrator.");
}

/// A host key's type as OpenSSH names it in messages: `ED25519`, `RSA`,
/// `ECDSA`, `ED25519-SK`, `ECDSA-SK`, `DSA`, with `-CERT` for
/// certificates.
pub fn host_key_type_label(key_blob: &[u8]) -> String {
    let name = crate::format::Reader::new(key_blob)
        .read_string()
        .ok()
        .and_then(|n| core::str::from_utf8(n).ok())
        .unwrap_or("");
    let base = if name.starts_with("sk-ssh-ed25519") {
        "ED25519-SK"
    } else if name.starts_with("ssh-ed25519") {
        "ED25519"
    } else if name.starts_with("sk-ecdsa") {
        "ECDSA-SK"
    } else if name.starts_with("ecdsa-sha2") {
        "ECDSA"
    } else if name.starts_with("ssh-rsa") || name.starts_with("rsa-sha2") {
        "RSA"
    } else if name.starts_with("ssh-dss") {
        "DSA"
    } else if name.starts_with("ssh-mldsa44-ed25519") {
        "MLDSA44-ED25519"
    } else {
        return name.to_string();
    };
    if name.contains("-cert-v01@") {
        format!("{base}-CERT")
    } else {
        base.to_string()
    }
}

/// Helper: turn an `Option<Vec<String>>` override into an owned list,
/// falling back to the given default slice when `None`.
fn owned_or_default(over: &Option<Vec<String>>, default: &[&str]) -> Vec<String> {
    match over {
        Some(v) => v.clone(),
        None => default.iter().map(|s| s.to_string()).collect(),
    }
}

/// Build the client's advertised KEXINIT, applying any config algorithm
/// The default host-key algorithm offer, with the certificate key-types ahead
/// of the matching plain keys (matching OpenSSH), used when no explicit
/// `HostKeyAlgorithms` override is given.
fn default_host_key_algorithms() -> Vec<String> {
    let mut v: Vec<String> = crate::cert::CERT_KEY_NAMES
        .iter()
        .map(|s| s.to_string())
        .collect();
    v.extend(defaults::HOST_KEY.iter().map(|s| s.to_string()));
    v
}

/// The base host-key type an algorithm name verifies against, ignoring the
/// `-cert-v01@openssh.com` certificate suffix and collapsing the RSA signature
/// variants (`rsa-sha2-256`, `rsa-sha2-512`, `ssh-rsa`) to one key. Two names
/// with the same base are backed by the same stored `known_hosts` key.
fn base_host_key_type(name: &str) -> &str {
    let n = name.strip_suffix("-cert-v01@openssh.com").unwrap_or(name);
    match n {
        "rsa-sha2-256" | "rsa-sha2-512" | "ssh-rsa" => "ssh-rsa",
        other => other,
    }
}

/// Reorder a host-key algorithm offer so the key types already recorded in
/// `known_hosts` for the target come first, preserving the default relative
/// order within each group and dropping nothing. This is OpenSSH's
/// `order_hostkeyalgs()` behaviour: it makes a server that has several host
/// keys present the one we already trust, so a connection to a host we only
/// have (say) an ECDSA entry for does not trip a spurious
/// "REMOTE HOST IDENTIFICATION HAS CHANGED" when the server also has a
/// newer-preferred Ed25519 key. A brand-new key type still appears (in the
/// back group) so first-contact and genuine changes are handled as before.
fn order_host_key_algorithms(defaults: Vec<String>, present_types: &[String]) -> Vec<String> {
    if present_types.is_empty() {
        return defaults;
    }
    let present: std::collections::HashSet<&str> = present_types
        .iter()
        .map(|t| base_host_key_type(t))
        .collect();
    let (mut front, mut back): (Vec<String>, Vec<String>) = defaults
        .into_iter()
        .partition(|a| present.contains(base_host_key_type(a)));
    front.append(&mut back);
    front
}

/// If no explicit `HostKeyAlgorithms` override is set, reorder the default
/// host-key offer so the key types already recorded in `known_hosts` for the
/// target come first (OpenSSH's `order_hostkeyalgs()`; see
/// [`order_host_key_algorithms`]). Shared by every client frontend so the
/// behaviour does not depend on which one connects. Call it *after* any
/// host-key pin has been applied (a pin already fixes the algorithm list, so
/// the `is_some()` check leaves it untouched) and before the KEXINIT is built.
/// The lookup mirrors the verifier: `HostKeyAlias` on port 22, otherwise the
/// connection's host and port, across the writable store and the read-only
/// global files, counting plain (non-marker) entries only.
pub(crate) fn prioritize_known_host_key_algorithms(
    algorithms: &mut AlgoOverrides,
    policy: &HostKeyPolicy,
    host: &str,
    port: u16,
) {
    if algorithms.host_key_algorithms.is_some() {
        return;
    }
    let HostKeyPolicy::KnownHosts(kh) = policy else {
        return;
    };
    let (lh, lp) = match &kh.host_key_alias {
        Some(alias) => (alias.as_str(), 22),
        None => (host, port),
    };
    let mut present: Vec<String> = Vec::new();
    if let Ok(store) = kh.store.lock() {
        present.extend(
            store
                .find(lh, lp)
                .into_iter()
                .filter(|e| e.marker.is_none())
                .map(|e| e.key_type.clone()),
        );
    }
    for g in &kh.global {
        present.extend(
            g.find(lh, lp)
                .into_iter()
                .filter(|e| e.marker.is_none())
                .map(|e| e.key_type.clone()),
        );
    }
    if !present.is_empty() {
        algorithms.host_key_algorithms = Some(order_host_key_algorithms(
            default_host_key_algorithms(),
            &present,
        ));
    }
}

/// overrides over the built-in defaults.
///
/// The client's strict-kex marker (`kex-strict-c-v00@openssh.com`; only the
/// server sends the `-s` one, as with OpenSSH) is re-appended here, *after* the (possibly user-supplied) kex list, so a
/// `KexAlgorithms` override can never strip the Terrapin (CVE-2023-48795)
/// mitigation. The non-marker default order is taken from `defaults::KEX`.
pub(crate) fn build_default_kexinit<R: RngCore>(rng: &mut R, over: &AlgoOverrides) -> KexInit {
    // Real (non-marker) kex default order.
    let default_kex: Vec<&str> = defaults::KEX
        .iter()
        .copied()
        .filter(|n| !is_strict_kex_marker(n))
        .collect();

    let mut kex = owned_or_default(&over.kex_algorithms, &default_kex);
    // Re-append the client's strict-kex marker. Offering the server's too
    // would let a peer with no real KEX in common "choose" it (OpenSSH then
    // fails with an internal error instead of "no matching key exchange").
    kex.retain(|k| !is_strict_kex_marker(k));
    kex.push(crate::transport::kex::STRICT_KEX_CLIENT_MARKER.to_string());

    let ciphers = owned_or_default(&over.ciphers, defaults::CIPHERS);
    let macs = owned_or_default(&over.macs, defaults::MACS);
    // Host-key algorithms. With no explicit `HostKeyAlgorithms` override we
    // advertise the certificate key-types ahead of the matching plain keys
    // (matching OpenSSH), so a server with a host certificate can offer it.
    // Accepting an offered cert still requires a trusted CA at verify time
    // (`build_verifier` → `verify_host_cert`); merely advertising the name does
    // not weaken host verification. An explicit override is taken verbatim.
    let host_key = match &over.host_key_algorithms {
        Some(v) => v.clone(),
        None => default_host_key_algorithms(),
    };
    // Compression preference. `Compression yes` advertises
    // `zlib@openssh.com` ahead of `none` so we still negotiate a session
    // with a server that lacks zlib. The delayed-zlib variant only starts
    // compressing post-auth (see ConnectionState::activate_compress), which
    // matches OpenSSH. Honoured only with the `compress` feature compiled
    // in; without it, even an explicit request degrades to `none` because
    // the codec has no zlib implementation to install.
    let comp: Vec<String> = if cfg!(feature = "compress") && over.compression == Some(true) {
        vec!["zlib@openssh.com".to_string(), "none".to_string()]
    } else {
        defaults::COMP.iter().map(|s| s.to_string()).collect()
    };

    let algs = KexAlgorithmsOwned {
        kex,
        server_host_key: host_key,
        ciphers_c2s: ciphers.clone(),
        ciphers_s2c: ciphers,
        macs_c2s: macs.clone(),
        macs_s2c: macs,
        comp_c2s: comp.clone(),
        comp_s2c: comp,
        lang_c2s: Vec::new(),
        lang_s2c: Vec::new(),
    };
    let mut cookie = [0u8; 16];
    rng.fill_bytes(&mut cookie);
    KexInit::from_algorithms_owned(algs, cookie)
}

/// Current wall-clock time as Unix seconds, injected into certificate
/// validity checks at the std edge. Falls back to 0 (which fails every
/// not-yet-valid check and is harmless for already-valid certs) on the
/// impossible pre-epoch case.
/// Fold a read timeout (`WouldBlock` / `TimedOut`) into `Ok(None)`.
fn timeout_as_none(r: Result<Vec<u8>>) -> Result<Option<Vec<u8>>> {
    match r {
        Ok(p) => Ok(Some(p)),
        Err(Error::Io(e))
            if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut =>
        {
            Ok(None)
        }
        Err(e) => Err(e),
    }
}

pub(crate) fn unix_now() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Persist an accepted host key. Like OpenSSH, failing to write the file
/// only warns: the key was accepted for this connection either way, and a
/// `UserKnownHostsFile` of `/dev/null` (the usual "don't remember" idiom) is
/// not written at all.
fn save_known_hosts(store: &KnownHosts, path: &std::path::Path) {
    if path == std::path::Path::new("/dev/null") {
        return;
    }
    if let Err(e) = store.save(path) {
        eprintln!(
            "Warning: failed to add the host to the list of known hosts ({}): {e}",
            path.display()
        );
    }
}

/// `K_S` (the host key blob) of an `SSH_MSG_KEX_ECDH_REPLY` payload.
fn reply_host_key(reply_payload: &[u8]) -> Result<&[u8]> {
    let mut r = crate::format::Reader::new(reply_payload);
    r.read_u8()?;
    r.read_string()
}

/// Host-key algorithms that can present `pin`, intersected with the
/// caller's `HostKeyAlgorithms` override (if any), in that override's order.
fn pinned_host_key_algorithms(pin: &[u8], over: Option<&[String]>) -> Result<Vec<String>> {
    let key_alg = crate::format::Reader::new(pin).read_string()?;
    let algs: &[&str] = match key_alg {
        b"ssh-rsa" => &["rsa-sha2-512", "rsa-sha2-256"],
        b"ssh-ed25519" => &["ssh-ed25519"],
        b"ecdsa-sha2-nistp256" => &["ecdsa-sha2-nistp256"],
        b"ecdsa-sha2-nistp384" => &["ecdsa-sha2-nistp384"],
        b"ecdsa-sha2-nistp521" => &["ecdsa-sha2-nistp521"],
        b"ssh-mldsa44-ed25519@openssh.com" => &["ssh-mldsa44-ed25519@openssh.com"],
        _ => return Err(Error::Unsupported("pinned host key type")),
    };
    let out: Vec<String> = match over {
        Some(list) => list
            .iter()
            .filter(|a| algs.contains(&a.as_str()))
            .cloned()
            .collect(),
        None => algs.iter().map(|a| a.to_string()).collect(),
    };
    if out.is_empty() {
        return Err(Error::Config(
            "HostKeyAlgorithms excludes the pinned host key's algorithm",
        ));
    }
    Ok(out)
}

/// Build the exchange-hash verifier for the host key carried in
/// `reply_payload`, making the trust decision per `policy` first. On a
/// re-key (`runner` already holds the first KEX's host key) the policy is
/// skipped and the key must equal that first one.
///
/// `host_key_mismatched` is set to `true` (never cleared) when a *changed*
/// key was let through by [`TofuAction::AcceptWithWarning`]; the frontends
/// capture the same flag so they can disable password / keyboard-interactive
/// auth and forwarding for the session (see [`Client::host_key_mismatched`]).
#[allow(clippy::too_many_arguments)]
pub(crate) fn build_verifier(
    reply_payload: &[u8],
    policy: &HostKeyPolicy,
    runner: &KexRunner,
    target_host: &str,
    target_port: u16,
    ca_signature_algorithms: Option<&[String]>,
    now: u64,
    host_key_mismatched: &AtomicBool,
) -> Result<Box<dyn HostKeyVerify>> {
    if reply_payload.len() < 5 {
        return Err(Error::Format("kex-ecdh-reply too short"));
    }
    let k_s_len = u32::from_be_bytes([
        reply_payload[1],
        reply_payload[2],
        reply_payload[3],
        reply_payload[4],
    ]) as usize;
    if reply_payload.len() < 5 + k_s_len {
        return Err(Error::Format("kex-ecdh-reply truncated"));
    }
    let k_s = &reply_payload[5..5 + k_s_len];

    // Re-key: the host key is pinned to the one the first KEX verified
    // (OpenSSH: "Server host key changed during re-key"). The known_hosts
    // policy is never re-run — no prompt, no store update — so a changed key
    // is refused and an unchanged one only needs the exchange-hash verifier.
    if let Some((initial, _)) = runner.initial_host_key_sig() {
        if k_s != initial {
            return Err(Error::HostKeyRejected);
        }
        let neg = runner
            .negotiated()
            .ok_or(Error::Protocol("kex: no negotiated algorithms"))?;
        return host_key_verify_by_name(&neg.host_key, k_s);
    }

    // Reject misconfigured KnownHosts callers *before* we touch the KEX
    // runner state — the missing-host case is a config error, not a
    // protocol error, regardless of where in the connect we are.
    if matches!(policy, HostKeyPolicy::KnownHosts(_))
        && (target_host.is_empty() || target_port == 0)
    {
        return Err(Error::Config(
            "HostKeyPolicy::KnownHosts requires Client::connect_to_host",
        ));
    }

    let neg = runner
        .negotiated()
        .ok_or(Error::Protocol("kex: no negotiated algorithms"))?;

    // Parse `K_S` as a key (or certificate) of the negotiated host-key
    // algorithm *before* any known_hosts lookup, prompt or save. Everything
    // below — and above all the type string written to known_hosts on a
    // TOFU / accepted-change add — then describes a well-formed key of the
    // family the KEX agreed on, never raw attacker-chosen bytes.
    let verifier = host_key_verify_by_name(&neg.host_key, k_s)?;

    // Certificate host keys: `k_s` is an OpenSSH certificate, not a plain key.
    // Parse it, enforce type/validity/critical-options, then make the trust
    // decision per policy. The signature over `H` is verified against the
    // certificate's EMBEDDED key (built below); the CA signature ties the
    // embedded key to a trusted CA.
    // `KnownHostsCommand`: the known_hosts lines it prints for this key,
    // read with the global files.
    let command_hosts = |blob: &[u8]| -> Option<Arc<KnownHosts>> {
        let HostKeyPolicy::KnownHosts(kh) = policy else {
            return None;
        };
        let f = kh.command.as_ref()?;
        let key_type = crate::format::Reader::new(blob)
            .read_string()
            .ok()
            .and_then(|t| core::str::from_utf8(t).ok())?
            .to_string();
        f(&key_type, blob).map(|text| Arc::new(KnownHosts::from_bytes(&text)))
    };
    // A certificate whose CA known_hosts does not list is checked as the
    // plain key it certifies (OpenSSH: "No matching CA found. Retry with
    // plain key").
    let mut plain_from_cert: Option<Vec<u8>> = None;
    if crate::cert::is_cert_name(&neg.host_key) {
        let cert = crate::cert::Certificate::parse(k_s)?;
        let ca_algos: Vec<&str> = match ca_signature_algorithms {
            Some(list) => list.iter().map(|s| s.as_str()).collect(),
            None => crate::config::algos::CA_SIGNATURE_DEFAULTS.to_vec(),
        };
        match policy {
            // No host verification requested: still parse and enforce the
            // cert's own constraints (type/validity/critical-options) but do
            // not require CA trust. `H` is verified against the embedded key.
            HostKeyPolicy::AcceptAny => {
                cert.check_type(crate::cert::CertType::Host)?;
                cert.check_validity(now)?;
                cert.require_known_critical_options()?;
            }
            // Pin the whole certificate blob by fingerprint.
            HostKeyPolicy::AcceptFingerprint(fp) => {
                let digest = Sha256::digest(k_s);
                if digest.as_ref() != fp {
                    return Err(Error::HostKeyRejected);
                }
                cert.check_type(crate::cert::CertType::Host)?;
                cert.check_validity(now)?;
                cert.require_known_critical_options()?;
            }
            // Trust via known_hosts `@cert-authority` (and `@revoked`), in
            // the user file or a global one.
            HostKeyPolicy::KnownHosts(kh) => {
                let from_command = command_hosts(k_s);
                let (host, port) = match &kh.host_key_alias {
                    Some(alias) => (alias.as_str(), 22),
                    None => (target_host, target_port),
                };
                if kh
                    .revoked
                    .as_ref()
                    .is_some_and(|r| r.is_revoked_cert(&cert))
                {
                    print_revoked_banner(&neg.host_key, host, port);
                    return Err(Error::HostKeyRejected);
                }
                let store = kh.store.lock().map_err(|_| Error::HostKeyRejected)?;
                if store.revokes_cert(host, port, &cert)
                    || kh
                        .global
                        .iter()
                        .chain(from_command.iter())
                        .any(|g| g.revokes_cert(host, port, &cert))
                {
                    return Err(Error::HostKeyRejected);
                }
                let has_ca = store.has_ca_for(host, port, &cert)
                    || kh
                        .global
                        .iter()
                        .chain(from_command.iter())
                        .any(|g| g.has_ca_for(host, port, &cert));
                if !has_ca {
                    plain_from_cert = Some(cert.embedded_pubkey_blob.clone());
                }
                let user = if has_ca {
                    store.verify_host_cert(host, port, &cert, &ca_algos, now)
                } else {
                    Ok(())
                };
                if user.is_err()
                    && kh.global.iter().chain(from_command.iter()).any(|g| {
                        g.verify_host_cert(host, port, &cert, &ca_algos, now)
                            .is_ok()
                    })
                {
                    // Trusted by a system file.
                } else {
                    user?;
                }
            }
        }
        // Build the exchange-hash verifier from the cert's embedded key, keyed
        // on the negotiated cert name (which pins the RSA hash).
        if plain_from_cert.is_none() {
            return Ok(verifier);
        }
    }
    // The key known_hosts is checked for: the host key, or the key a
    // certificate from an unlisted CA certifies.
    let k_s: &[u8] = plain_from_cert.as_deref().unwrap_or(k_s);

    match policy {
        HostKeyPolicy::AcceptAny => {}
        HostKeyPolicy::AcceptFingerprint(fp) => {
            let digest = Sha256::digest(k_s);
            if digest.as_ref() != fp {
                return Err(Error::HostKeyRejected);
            }
        }
        HostKeyPolicy::KnownHosts(kh) => {
            let from_command = command_hosts(k_s);
            // known_hosts records the key's type (`ssh-rsa`), not the
            // signature algorithm negotiated for it (`rsa-sha2-512`). Derive
            // it from the negotiated name — the blob was already parsed as
            // that family above — and insist the blob agrees.
            let key_type = known_hosts_key_type(&neg.host_key).to_string();
            if crate::format::Reader::new(k_s).read_string().ok() != Some(key_type.as_bytes()) {
                return Err(Error::HostKeyRejected);
            }
            // Host-empty / port-zero already rejected above. Past this
            // point we have a real (host, port) pair, so the lookup can
            // proceed. `HostKeyAlias` replaces both for the lookup.
            let (target_host, target_port) = match &kh.host_key_alias {
                Some(alias) => (alias.as_str(), 22),
                None => (target_host, target_port),
            };
            if kh.revoked.as_ref().is_some_and(|r| r.is_revoked_key(k_s)) {
                print_revoked_banner(&key_type, target_host, target_port);
                return Err(Error::HostKeyRejected);
            }
            let mut store = kh.store.lock().map_err(|_| Error::HostKeyRejected)?;
            let mut lookup = store.lookup(target_host, target_port, &key_type, k_s);
            for g in kh.global.iter().chain(from_command.iter()) {
                lookup =
                    combine_lookups(lookup, g.lookup(target_host, target_port, &key_type, k_s));
            }
            match lookup {
                LookupResult::Match => {}
                LookupResult::Revoked => {
                    // A `@revoked` host key is a hard failure regardless of
                    // `on_mismatch` — no `StrictHostKeyChecking=no` accept,
                    // no prompt, no rotation of the store. OpenSSH aborts
                    // the connection here too.
                    print_revoked_banner(&key_type, target_host, target_port);
                    return Err(Error::HostKeyRejected);
                }
                LookupResult::Mismatch { expected } => {
                    // Print the OpenSSH-style loud banner on mismatch,
                    // before any policy decision. Shows both the
                    // previously-stored ("old") fingerprint(s) and the new
                    // one the peer just presented so the user can spot the
                    // change without digging through known_hosts manually.
                    // Skipped for `PromptDetailed`: that callback receives
                    // the same information via `HostKeyPrompt` and owns its
                    // own user-facing display.
                    if !matches!(&kh.on_mismatch, TofuAction::PromptDetailed(_)) {
                        print_mismatch_banner(target_host, target_port, &expected, &key_type, k_s);
                    }

                    let accept = match &kh.on_mismatch {
                        TofuAction::Reject => false,
                        TofuAction::Accept => true,
                        TofuAction::AcceptWithWarning => {
                            // StrictHostKeyChecking=no: OpenSSH accepts
                            // for the current session but does NOT
                            // rotate the stored key. We follow suit —
                            // the warning was printed above; nothing
                            // touches the store.
                            // OpenSSH's notice; the trusted entry is not
                            // updated either.
                            eprintln!(
                                "Password authentication is disabled to avoid \
                                 man-in-the-middle attacks."
                            );
                            eprintln!(
                                "Keyboard-interactive authentication is disabled to avoid \
                                 man-in-the-middle attacks."
                            );
                            // Remember that this session rides on an
                            // unverified key so the frontends can refuse
                            // password / keyboard-interactive auth and
                            // forwarding, exactly as OpenSSH does.
                            host_key_mismatched.store(true, Ordering::SeqCst);
                            true
                        }
                        TofuAction::Prompt(cb) => {
                            // Drop the lock for the duration of the
                            // callback — it may block on stdin and
                            // shouldn't hold up other policy users.
                            // The loud banner was already printed above
                            // so the callback only needs to ask the
                            // accept/refuse question.
                            drop(store);
                            let ok = cb(target_host, target_port, &key_type, k_s);
                            store = kh.store.lock().map_err(|_| Error::HostKeyRejected)?;
                            ok
                        }
                        TofuAction::PromptDetailed(cb) => {
                            // Hand the callback full context (no banner was
                            // printed above) so it can render its own
                            // changed-key warning, including the old keys.
                            drop(store);
                            let prompt = HostKeyPrompt {
                                change: HostKeyChange::Changed,
                                host: target_host,
                                port: target_port,
                                key_type: &key_type,
                                key_blob: k_s,
                                fingerprint: host_key_fingerprint(k_s),
                                existing: &expected,
                            };
                            let ok = cb(&prompt);
                            store = kh.store.lock().map_err(|_| Error::HostKeyRejected)?;
                            ok
                        }
                    };
                    if !accept {
                        return Err(Error::HostKeyRejected);
                    }
                    // Only rotate the stored entry when the policy is a
                    // prompt that explicitly returned `true` — i.e. the
                    // user typed `yes`. `AcceptWithWarning` (the
                    // StrictHostKeyChecking=no path) deliberately
                    // leaves the store untouched: OpenSSH does the same.
                    if matches!(
                        &kh.on_mismatch,
                        TofuAction::Accept | TofuAction::Prompt(_) | TofuAction::PromptDetailed(_)
                    ) {
                        // Replace the existing entries so future
                        // connects don't keep tripping the mismatch
                        // path. Honours the same hash-new / save-path
                        // knobs as the Unknown path.
                        //
                        // `remove_exact` only drops marker-less lines
                        // naming exactly this host[:port] (literal or
                        // hashed). Wildcard / multi-host patterns and
                        // `@cert-authority` / `@revoked` markers are
                        // shared with other hosts and must survive a
                        // single host's key rotation — the broad
                        // `remove` would silently delete them.
                        let _ = store.remove_exact(target_host, target_port);
                        store
                            .try_add(target_host, target_port, &key_type, k_s, kh.hash_new)
                            .map_err(|_| Error::HostKeyRejected)?;
                        if let Some(path) = &kh.save_path {
                            save_known_hosts(&store, path);
                        }
                    }
                }
                LookupResult::Unknown => {
                    let accept = match &kh.on_unknown {
                        TofuAction::Reject => false,
                        TofuAction::Accept | TofuAction::AcceptWithWarning => true,
                        TofuAction::Prompt(cb) => {
                            // Drop the lock for the duration of the
                            // callback — it may block on stdin and
                            // shouldn't hold up other policy users.
                            drop(store);
                            let ok = cb(target_host, target_port, &key_type, k_s);
                            store = kh.store.lock().map_err(|_| Error::HostKeyRejected)?;
                            ok
                        }
                        TofuAction::PromptDetailed(cb) => {
                            // First-contact (TOFU) add: no prior keys, so
                            // `existing` is empty.
                            drop(store);
                            let prompt = HostKeyPrompt {
                                change: HostKeyChange::Unknown,
                                host: target_host,
                                port: target_port,
                                key_type: &key_type,
                                key_blob: k_s,
                                fingerprint: host_key_fingerprint(k_s),
                                existing: &[],
                            };
                            let ok = cb(&prompt);
                            store = kh.store.lock().map_err(|_| Error::HostKeyRejected)?;
                            ok
                        }
                    };
                    if !accept {
                        return Err(Error::HostKeyRejected);
                    }
                    store
                        .try_add(target_host, target_port, &key_type, k_s, kh.hash_new)
                        .map_err(|_| Error::HostKeyRejected)?;
                    if let Some(path) = &kh.save_path {
                        save_known_hosts(&store, path);
                    }
                }
            }
        }
    }

    Ok(verifier)
}

/// The key type a `known_hosts` line records for a host key negotiated as
/// `alg`: the plain key name, with certificate names mapped to the key they
/// embed and RSA signature algorithms (`rsa-sha2-*`) to `ssh-rsa`.
fn known_hosts_key_type(alg: &str) -> &str {
    match crate::cert::cert_name_to_plain(alg).unwrap_or(alg) {
        "rsa-sha2-256" | "rsa-sha2-512" => "ssh-rsa",
        plain => plain,
    }
}

/// OpenSSH's banner for a revoked host key.
fn print_revoked_banner(key_type: &str, host: &str, port: u16) {
    eprintln!(
        "@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\n\
         @       WARNING: REVOKED HOST KEY DETECTED!               @\n\
         @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\n\
         The {key_type} host key for {} is revoked.\n\
         This could mean that a stolen key is being used to \
         impersonate this host.\n\
         Host key verification failed: host key is revoked.",
        if port == 22 {
            host.to_string()
        } else {
            format!("[{host}]:{port}")
        },
    );
}

/// Combine the lookups of the user and a global known_hosts file the way
/// OpenSSH evaluates them together: a revocation anywhere wins, then a
/// match anywhere, then a mismatch anywhere (listing every expected key).
fn combine_lookups(a: LookupResult, b: LookupResult) -> LookupResult {
    match (a, b) {
        (LookupResult::Revoked, _) | (_, LookupResult::Revoked) => LookupResult::Revoked,
        (LookupResult::Match, _) | (_, LookupResult::Match) => LookupResult::Match,
        (LookupResult::Mismatch { mut expected }, LookupResult::Mismatch { expected: more }) => {
            expected.extend(more);
            LookupResult::Mismatch { expected }
        }
        (m @ LookupResult::Mismatch { .. }, _) | (_, m @ LookupResult::Mismatch { .. }) => m,
        _ => LookupResult::Unknown,
    }
}

/// Pull every queued [`ServeCommand`] from `cmd_rx` (non-blocking) and
/// kick off the matching outbound open. The serve loop then waits for
/// `OpenConfirmed` / `OpenFailed` in [`serve_dispatch_packet`].
///
/// Errors here propagate out of [`Client::serve`] — they indicate a
/// failure to encode/transmit a CHANNEL_OPEN, not a peer rejection
/// (which arrives as a normal `OpenFailed` event).
fn serve_drain_commands(
    client: &mut Client,
    cmd_rx: &Receiver<ServeCommand>,
    pending_opens: &mut BTreeMap<u32, PendingOutboundOpen>,
) -> Result<()> {
    loop {
        match cmd_rx.try_recv() {
            // Local port forwards (`-L` / `-D`) are refused over a changed
            // host key, as OpenSSH does; the requester gets a clear error
            // instead of a channel.
            Ok(ServeCommand::OpenDirectTcpip { reply, .. })
            | Ok(ServeCommand::OpenDirectStreamlocal { reply, .. })
                if client.host_key_mismatched() =>
            {
                let _ = reply.send(Err(Error::Config(HOST_KEY_MISMATCH_DISABLED)));
            }
            Ok(ServeCommand::OpenDirectTcpip {
                dest_host,
                dest_port,
                orig_host,
                orig_port,
                reply,
            }) => {
                let (local_id, open_payload) = client.conn.open(ChannelOpen::DirectTcpip {
                    dest_host,
                    dest_port: dest_port as u32,
                    orig_host,
                    orig_port: orig_port as u32,
                })?;
                client.write_payload(&open_payload)?;
                let (ingress_tx, ingress_rx) =
                    mpsc::sync_channel::<Option<Vec<u8>>>(SERVE_INGRESS_BACKLOG);
                let (egress_tx, egress_rx) =
                    mpsc::sync_channel::<ChannelEgress>(SERVE_EGRESS_BACKLOG);
                let stream = ChannelStream::new(ingress_rx, egress_tx);
                pending_opens.insert(
                    local_id,
                    PendingOutboundOpen {
                        stream: Some(stream),
                        ingress_tx,
                        egress_rx: Some(egress_rx),
                        reply,
                    },
                );
            }
            Ok(ServeCommand::OpenDirectStreamlocal { socket_path, reply }) => {
                let (local_id, open_payload) = client
                    .conn
                    .open(ChannelOpen::DirectStreamlocal { socket_path })?;
                client.write_payload(&open_payload)?;
                let (ingress_tx, ingress_rx) =
                    mpsc::sync_channel::<Option<Vec<u8>>>(SERVE_INGRESS_BACKLOG);
                let (egress_tx, egress_rx) =
                    mpsc::sync_channel::<ChannelEgress>(SERVE_EGRESS_BACKLOG);
                let stream = ChannelStream::new(ingress_rx, egress_tx);
                pending_opens.insert(
                    local_id,
                    PendingOutboundOpen {
                        stream: Some(stream),
                        ingress_tx,
                        egress_rx: Some(egress_rx),
                        reply,
                    },
                );
            }
            Err(TryRecvError::Empty) => break,
            Err(TryRecvError::Disconnected) => break,
        }
    }
    Ok(())
}

/// Credit `n` consumed bytes back to the peer's window on `channel`,
/// emitting a `CHANNEL_WINDOW_ADJUST` when the threshold is reached. A
/// channel we have already closed (or that is gone) is silently skipped:
/// deferred credit can legitimately outlive the channel.
fn serve_credit_window(client: &mut Client, channel: u32, n: usize) -> Result<()> {
    if n == 0 {
        return Ok(());
    }
    let n = u32::try_from(n).unwrap_or(u32::MAX);
    match client.conn.replenish_window(channel, n) {
        Ok(Some(adj)) => client.write_payload(&adj),
        Ok(None) | Err(Error::BadChannelState) => Ok(()),
        Err(e) => Err(e),
    }
}

/// Per-tick: retry every runtime's parked ingress and credit the receive
/// window for whatever the handlers accepted this time.
fn serve_flush_ingress(
    client: &mut Client,
    runtimes: &mut BTreeMap<u32, ServeRuntime>,
) -> Result<()> {
    let channels: Vec<u32> = runtimes
        .iter()
        .filter(|(_, rt)| !rt.ingress_backlog.is_empty())
        .map(|(ch, _)| *ch)
        .collect();
    for ch in channels {
        let delivered = match runtimes.get_mut(&ch) {
            Some(rt) => rt.flush_ingress(),
            None => 0,
        };
        serve_credit_window(client, ch, delivered)?;
    }
    Ok(())
}

fn serve_drain_runtimes(
    client: &mut Client,
    runtimes: &mut BTreeMap<u32, ServeRuntime>,
) -> Result<()> {
    let channels: Vec<u32> = runtimes.keys().copied().collect();
    for ch in channels {
        let Some(rt) = runtimes.get_mut(&ch) else {
            continue;
        };
        if rt.close_sent {
            continue;
        }

        // 1) Re-attempt any leftover bytes from last tick.
        if !rt.pending_data.is_empty() {
            let leftover = core::mem::take(&mut rt.pending_data);
            emit_serve_data(client, ch, &leftover, rt)?;
            if !rt.pending_data.is_empty() {
                // Still window-blocked; skip this tick's drain entirely.
                continue;
            }
        }

        // 2) Pull as many egress messages as we can without blocking.
        loop {
            if !rt.pending_data.is_empty() {
                break;
            }
            match rt.egress_rx.try_recv() {
                Ok(ChannelEgress::Data(bytes)) => {
                    emit_serve_data(client, ch, &bytes, rt)?;
                }
                Ok(ChannelEgress::Datagram(bytes)) => {
                    if let Some(p) = client.conn.send_datagram(ch, &bytes)? {
                        client.write_payload(&p)?;
                    }
                }
                // Session-only egress (stderr, exit status) has no meaning
                // on the forwarded channels a client serves.
                Ok(ChannelEgress::Stderr(_) | ChannelEgress::ExitStatus(_)) => {}
                Ok(ChannelEgress::Eof) => {
                    rt.pending_eof = true;
                    break;
                }
                Ok(ChannelEgress::Close) => {
                    rt.pending_close = true;
                    break;
                }
                Err(TryRecvError::Empty) => break,
                Err(TryRecvError::Disconnected) => {
                    // Handler thread vanished without an explicit Close;
                    // synthesise one so the channel still tears down cleanly.
                    rt.pending_close = true;
                    break;
                }
            }
        }

        // 3) Emit EOF / Close if we have them pending and all data shipped.
        if rt.pending_data.is_empty() {
            if rt.pending_eof && !rt.eof_sent {
                let p = client.conn.send_eof(ch)?;
                client.write_payload(&p)?;
                rt.eof_sent = true;
            }
            if rt.pending_close && !rt.close_sent {
                if !rt.eof_sent {
                    let p = client.conn.send_eof(ch)?;
                    client.write_payload(&p)?;
                    rt.eof_sent = true;
                }
                let p = client.conn.send_close(ch)?;
                client.write_payload(&p)?;
                rt.close_sent = true;
            }
        }
    }
    Ok(())
}

/// Ship `bytes` over `CHANNEL_DATA`, stashing anything the remote window
/// can't accept onto `rt.pending_data` for next tick.
fn emit_serve_data(
    client: &mut Client,
    channel: u32,
    bytes: &[u8],
    rt: &mut ServeRuntime,
) -> Result<()> {
    let mut off = 0usize;
    while off < bytes.len() {
        let (payload, taken) = client.conn.send_data(channel, &bytes[off..])?;
        if taken == 0 {
            rt.pending_data.extend_from_slice(&bytes[off..]);
            return Ok(());
        }
        client.write_payload(&payload)?;
        off += taken;
    }
    Ok(())
}

/// Process one inbound app-layer packet for [`Client::serve`].
///
/// Mirrors the server-side `dispatch_app_packet`: routes peer-initiated
/// channel opens to the matching [`ClientHandlers`] callback (or rejects),
/// fans Data/Eof/Close into the matching runtime, replenishes the SSH
/// receive window, and emits our half-close in response to peer Close.
fn serve_dispatch_packet(
    client: &mut Client,
    handlers: &ClientHandlers,
    runtimes: &mut BTreeMap<u32, ServeRuntime>,
    pending_opens: &mut BTreeMap<u32, PendingOutboundOpen>,
    payload: &[u8],
) -> Result<()> {
    let ev = client.conn.on_packet(payload)?;
    match ev {
        ChannelEvent::OpenConfirmed { channel } => {
            // Caller-initiated open succeeded: hand the pre-built stream
            // to the requester, and promote bookkeeping to a ServeRuntime.
            if let Some(mut po) = pending_opens.remove(&channel) {
                let stream = po
                    .stream
                    .take()
                    .ok_or(Error::Protocol("pending open: stream taken twice"))?;
                let egress_rx = po
                    .egress_rx
                    .take()
                    .ok_or(Error::Protocol("pending open: egress taken twice"))?;
                // If the caller hung up before we got here, just close.
                if po.reply.send(Ok(stream)).is_err() {
                    let p = client.conn.send_close(channel)?;
                    client.write_payload(&p)?;
                    return Ok(());
                }
                runtimes.insert(channel, ServeRuntime::new(po.ingress_tx, egress_rx));
            }
        }
        ChannelEvent::OpenFailed { channel, .. } => {
            if let Some(po) = pending_opens.remove(&channel) {
                let _ = po
                    .reply
                    .send(Err(Error::Protocol("direct-tcpip: open failed")));
            }
        }
        ChannelEvent::OpenRequest { channel, kind } => match kind {
            ChannelOpen::ForwardedTcpip {
                dest_host,
                dest_port,
                orig_host,
                orig_port,
            } => {
                // Defence-in-depth: a `forwarded-tcpip` open must correlate
                // to a forward the client actually requested. We reject the
                // open if there are zero outstanding `tcpip-forward` grants
                // (an unsolicited forward when none were requested). This is
                // deliberately conservative — the server may echo a bind
                // address/port that doesn't textually match the request, so
                // a stricter per-binding match could reject legitimate
                // forwards. Callers needing exact correlation should also
                // check `origin` in their callback.
                // Never over a host key that did not match known_hosts.
                if client.tcpip_forward_grants.is_empty() || client.host_key_mismatched() {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "no tcpip-forward requested",
                        "",
                    )?;
                    client.write_payload(&p)?;
                    return Ok(());
                }
                // The wire carries ports as u32; anything above 65535 is not
                // a TCP port. Reject rather than clamp, so a handler never
                // sees a fabricated `65535` origin.
                let (Ok(bound_port), Ok(orig_port)) =
                    (u16::try_from(dest_port), u16::try_from(orig_port))
                else {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "port out of range",
                        "",
                    )?;
                    client.write_payload(&p)?;
                    return Ok(());
                };
                if let Some(cb) = handlers.on_forwarded_tcpip.clone() {
                    let p = client.conn.accept_open(channel)?;
                    client.write_payload(&p)?;

                    let (ingress_tx, ingress_rx) =
                        mpsc::sync_channel::<Option<Vec<u8>>>(SERVE_INGRESS_BACKLOG);
                    let (egress_tx, egress_rx) =
                        mpsc::sync_channel::<ChannelEgress>(SERVE_EGRESS_BACKLOG);
                    let cs = ChannelStream::new(ingress_rx, egress_tx);
                    let origin = ForwardedTcpipOrigin {
                        bound_address: dest_host,
                        bound_port,
                        orig_address: orig_host,
                        orig_port,
                    };
                    thread::spawn(move || {
                        cb(origin, cs);
                    });
                    runtimes.insert(channel, ServeRuntime::new(ingress_tx, egress_rx));
                } else {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "forwarded-tcpip not enabled",
                        "",
                    )?;
                    client.write_payload(&p)?;
                }
            }
            ChannelOpen::ForwardedStreamlocal { socket_path } => {
                // Same conservative defence-in-depth as `forwarded-tcpip`: a
                // `forwarded-streamlocal@openssh.com` open must correlate to a
                // streamlocal-forward the client actually requested. Reject if
                // there are zero outstanding grants.
                if client.streamlocal_forward_grants.is_empty() || client.host_key_mismatched() {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "no streamlocal-forward requested",
                        "",
                    )?;
                    client.write_payload(&p)?;
                    return Ok(());
                }
                if let Some(cb) = handlers.on_forwarded_streamlocal.clone() {
                    let p = client.conn.accept_open(channel)?;
                    client.write_payload(&p)?;

                    let (ingress_tx, ingress_rx) =
                        mpsc::sync_channel::<Option<Vec<u8>>>(SERVE_INGRESS_BACKLOG);
                    let (egress_tx, egress_rx) =
                        mpsc::sync_channel::<ChannelEgress>(SERVE_EGRESS_BACKLOG);
                    let cs = ChannelStream::new(ingress_rx, egress_tx);
                    let origin = ForwardedStreamlocalOrigin { socket_path };
                    thread::spawn(move || {
                        cb(origin, cs);
                    });
                    runtimes.insert(channel, ServeRuntime::new(ingress_tx, egress_rx));
                } else {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "forwarded-streamlocal not enabled",
                        "",
                    )?;
                    client.write_payload(&p)?;
                }
            }
            ChannelOpen::AuthAgent => {
                // Only when agent forwarding was requested, and never over a
                // host key that did not match known_hosts (mirrors
                // `shared::incoming_kind`): a handler being installed is not
                // consent to a server-initiated agent channel.
                let allowed = client.request_auth_agent && !client.host_key_mismatched();
                if let Some(cb) = handlers.on_auth_agent.clone().filter(|_| allowed) {
                    let p = client.conn.accept_open(channel)?;
                    client.write_payload(&p)?;

                    let (ingress_tx, ingress_rx) =
                        mpsc::sync_channel::<Option<Vec<u8>>>(SERVE_INGRESS_BACKLOG);
                    let (egress_tx, egress_rx) =
                        mpsc::sync_channel::<ChannelEgress>(SERVE_EGRESS_BACKLOG);
                    let cs = ChannelStream::new(ingress_rx, egress_tx);
                    thread::spawn(move || {
                        cb(cs);
                    });
                    runtimes.insert(channel, ServeRuntime::new(ingress_tx, egress_rx));
                } else {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "auth-agent not enabled",
                        "",
                    )?;
                    client.write_payload(&p)?;
                }
            }
            ChannelOpen::X11 {
                orig_host: _,
                orig_port: _,
            } => {
                // Only when X11 forwarding was requested, and never over a
                // mismatched host key (as for `auth-agent` above).
                let allowed = client.wants_x11() && !client.host_key_mismatched();
                if let Some(cb) = handlers.on_x11.clone().filter(|_| allowed) {
                    let p = client.conn.accept_open(channel)?;
                    client.write_payload(&p)?;

                    let (ingress_tx, ingress_rx) =
                        mpsc::sync_channel::<Option<Vec<u8>>>(SERVE_INGRESS_BACKLOG);
                    let (egress_tx, egress_rx) =
                        mpsc::sync_channel::<ChannelEgress>(SERVE_EGRESS_BACKLOG);
                    let cs = ChannelStream::new(ingress_rx, egress_tx);
                    thread::spawn(move || {
                        cb(cs);
                    });
                    runtimes.insert(channel, ServeRuntime::new(ingress_tx, egress_rx));
                } else {
                    let p = client.conn.reject_open(
                        channel,
                        SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                        "x11 not enabled",
                        "",
                    )?;
                    client.write_payload(&p)?;
                }
            }
            _ => {
                let p = client.conn.reject_open(
                    channel,
                    SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                    "channel type not supported",
                    "",
                )?;
                client.write_payload(&p)?;
            }
        },
        ChannelEvent::Data { channel, data } => {
            // Credit the window only for bytes the handler's (bounded)
            // queue accepted; anything parked in the backlog is credited
            // when `serve_flush_ingress` later delivers it. Data for a
            // channel with no runtime (already torn down) is dropped and
            // credited so the peer isn't left with a dead window.
            let delivered = match runtimes.get_mut(&channel) {
                Some(rt) => rt.offer_ingress(Some(data)),
                None => data.len(),
            };
            serve_credit_window(client, channel, delivered)?;
        }
        ChannelEvent::ExtendedData { channel, data, .. } => {
            // forwarded-tcpip channels shouldn't carry extended-data, but if
            // we get any, just credit the window and drop the bytes.
            if let Some(adj) = client.conn.replenish_window(channel, data.len() as u32)? {
                client.write_payload(&adj)?;
            }
        }
        ChannelEvent::Eof { channel } => {
            if let Some(rt) = runtimes.get_mut(&channel) {
                // None = EOF marker; the handler's `Read::read` returns
                // `Ok(0)` next time it drains its buffer. Queued behind any
                // parked data so ordering is preserved.
                let _ = rt.offer_ingress(None);
            }
        }
        ChannelEvent::Close { channel } => {
            if let Some(ch) = client.conn.channel(channel)
                && !ch.local_closed
            {
                let p = client.conn.send_close(channel)?;
                client.write_payload(&p)?;
            }
            // Dropping the runtime closes `ingress_tx`; the handler thread's
            // next `Read` returns `Ok(0)` and the thread exits.
            runtimes.remove(&channel);
        }
        // The peer may send its own keepalive (or any other global request)
        // at us; per RFC 4254 §4 a request we don't act on still needs a
        // REQUEST_FAILURE when want_reply is set, which is exactly what
        // OpenSSH's keepalive probe expects back. Reading the packet already
        // refreshed our own keepalive timer above.
        ChannelEvent::GlobalRequest {
            want_reply: true, ..
        } => {
            let p = client.conn.send_global_failure();
            client.write_payload(&p)?;
        }
        // An open the connection layer already refused (channel cap):
        // forward its pre-built OPEN_FAILURE so the peer gets its reply.
        ChannelEvent::OpenRejected { payload, .. } => client.write_payload(&payload)?,
        // Other events (OpenConfirmed/Failed for opens *we* initiated,
        // Request, GlobalSuccess/GlobalFailure for our keepalive probes,
        // etc.) need no action here — the keepalive bookkeeping in `serve`
        // already cleared on the inbound read. Silently drop.
        _ => {}
    }
    Ok(())
}

/// Make peer-supplied bytes safe to print on a terminal: lossy UTF-8, with
/// every control character (C0, DEL and the C1 range — so ESC / CSI escape
/// sequences and OSC hyperlinks included) other than `\n` and `\t` replaced
/// by `?`. Used for the remote `scp`'s stderr, which an attacker-controlled
/// server can fill with terminal escape sequences.
fn sanitize_for_terminal(bytes: &[u8]) -> String {
    String::from_utf8_lossy(bytes)
        .chars()
        .map(|c| {
            if c.is_control() && c != '\n' && c != '\t' {
                '?'
            } else {
                c
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hostkey::Ed25519HostKey;
    use crate::transport::version::LOCAL_VERSION;
    use crate::transport::{PacketCodec, Role, VersionExchange};
    use purecrypto::rng::OsRng;
    use std::io::{Read, Write};
    use std::net::TcpListener;
    use std::thread;

    // Transport-engine details the in-process fake-server harness below needs.
    // (The driver owns these in production; the tests hand-roll a server.)
    const SSH_MSG_KEX_ECDH_REPLY: u8 = 31;
    const MAX_INBOX_BYTES: usize = 8 * 1024 * 1024;

    /// Read one `\n`-terminated line from `stream` into `buf` (test helper for
    /// the fake server's version-exchange step).
    fn read_line<S: Read>(stream: &mut S, buf: &mut Vec<u8>, max_len: usize) -> Result<()> {
        let mut byte = [0u8; 1];
        loop {
            let n = stream.read(&mut byte)?;
            if n == 0 {
                return Err(Error::Protocol("connection closed before newline"));
            }
            buf.push(byte[0]);
            if byte[0] == b'\n' {
                return Ok(());
            }
            if buf.len() >= max_len {
                return Err(Error::Protocol("banner line too long"));
            }
        }
    }

    #[test]
    fn host_key_order_prioritises_known_type() {
        // A host we only have an ECDSA entry for: the ecdsa algorithm (and its
        // cert variant) must come before the otherwise-preferred Ed25519, so
        // the server presents the key we already trust instead of a changed
        // one. Nothing is dropped.
        let ordered = order_host_key_algorithms(
            default_host_key_algorithms(),
            &["ecdsa-sha2-nistp256".to_string()],
        );
        let ed = ordered.iter().position(|a| a == "ssh-ed25519").unwrap();
        let ec = ordered
            .iter()
            .position(|a| a == "ecdsa-sha2-nistp256")
            .unwrap();
        assert!(ec < ed, "ecdsa must precede ed25519: {ordered:?}");
        assert_eq!(
            ordered.len(),
            default_host_key_algorithms().len(),
            "reordering must not drop or add algorithms"
        );
        // The cert variant of the known type is prioritised too.
        let ec_cert = ordered
            .iter()
            .position(|a| a == "ecdsa-sha2-nistp256-cert-v01@openssh.com");
        assert!(ec_cert.map(|p| p < ed).unwrap_or(true));
    }

    #[test]
    fn host_key_order_collapses_rsa_variants() {
        // A stored `ssh-rsa` key backs every RSA signature algorithm, so all of
        // them move forward.
        let ordered =
            order_host_key_algorithms(default_host_key_algorithms(), &["ssh-rsa".to_string()]);
        let ed = ordered.iter().position(|a| a == "ssh-ed25519").unwrap();
        for rsa in ["rsa-sha2-512", "rsa-sha2-256"] {
            let p = ordered.iter().position(|a| a == rsa).unwrap();
            assert!(p < ed, "{rsa} must precede ed25519: {ordered:?}");
        }
    }

    #[test]
    fn host_key_order_noop_without_entries() {
        // No known_hosts entries → the default order is returned unchanged.
        let def = default_host_key_algorithms();
        assert_eq!(order_host_key_algorithms(def.clone(), &[]), def);
    }

    #[test]
    fn config_insecure_constructor_is_accept_any() {
        // `Config::insecure()` replaces the old `Default` impl. The trust
        // decision now has to be spelled out explicitly at the call site;
        // see `Config::with_known_hosts` for the strict alternative.
        let cfg = Config::insecure();
        assert!(matches!(cfg.host_key_policy, HostKeyPolicy::AcceptAny));
        assert!(cfg.timeout.is_none());
    }

    #[test]
    fn known_hosts_strict_constructor_defaults_reject_reject() {
        // Sanity-check that the convenience constructor doesn't pick up
        // an `Accept*` variant accidentally.
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        let p = KnownHostsPolicy::strict(store);
        assert!(matches!(p.on_unknown, TofuAction::Reject));
        assert!(matches!(p.on_mismatch, TofuAction::Reject));
        assert!(!p.hash_new);
        assert!(p.save_path.is_none());
    }

    #[test]
    fn build_verifier_fails_hard_on_empty_host() {
        // Synthesise the minimum to drive build_verifier with a
        // KnownHosts policy and an empty target_host. The KEX runner /
        // reply payload don't get inspected because the host check
        // fires first.
        use crate::transport::kex::KexAlgorithms;
        use crate::transport::{KexInit, KexRunner};
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        let policy = HostKeyPolicy::KnownHosts(KnownHostsPolicy::strict(store));
        // Drive runner just enough to have a negotiated()-returning state
        // is not actually needed for this branch — the empty-host check
        // fires first. Build a dummy runner that we never call into.
        let runner = KexRunner::new(
            Role::Client,
            KexInit::from_algorithms(
                &KexAlgorithms {
                    kex: defaults::KEX,
                    server_host_key: defaults::HOST_KEY,
                    ciphers_c2s: defaults::CIPHERS,
                    ciphers_s2c: defaults::CIPHERS,
                    macs_c2s: defaults::MACS,
                    macs_s2c: defaults::MACS,
                    comp_c2s: defaults::COMP,
                    comp_s2c: defaults::COMP,
                    lang_c2s: &[],
                    lang_s2c: &[],
                },
                [0u8; 16],
            ),
        );
        // Provide a reply payload of the minimum shape (5 bytes header +
        // a 0-byte K_S). The host-empty branch fires before negotiated()
        // is consulted, so we don't need a real KEX outcome.
        let mut reply = vec![SSH_MSG_KEX_ECDH_REPLY];
        reply.extend_from_slice(&0u32.to_be_bytes());
        let flag = AtomicBool::new(false);
        let err = build_verifier(&reply, &policy, &runner, "", 22, None, 0, &flag);
        assert!(matches!(err, Err(Error::Config(_))));
        let err = build_verifier(&reply, &policy, &runner, "host", 0, None, 0, &flag);
        assert!(matches!(err, Err(Error::Config(_))));
    }

    /// A client-side `KexRunner` that has seen an ed25519-only server
    /// KEXINIT, so `negotiated()` reports `ssh-ed25519`.
    fn negotiated_ed25519_runner() -> KexRunner {
        let mut rng = OsRng;
        let over = AlgoOverrides {
            host_key_algorithms: Some(vec!["ssh-ed25519".to_string()]),
            ..Default::default()
        };
        let mut server = KexRunner::new(Role::Server, build_default_kexinit(&mut rng, &over));
        let server_kexinit = server.start(&mut rng).unwrap().outbound.remove(0);
        let mut client = KexRunner::new(Role::Client, build_default_kexinit(&mut rng, &over));
        client.start(&mut rng).unwrap();
        let mut codec = PacketCodec::new();
        client
            .on_packet(
                &mut rng,
                &mut codec,
                &server_kexinit,
                None,
                None,
                LOCAL_VERSION.as_bytes(),
                LOCAL_VERSION.as_bytes(),
            )
            .unwrap();
        assert_eq!(client.negotiated().unwrap().host_key, "ssh-ed25519");
        client
    }

    /// CLI-5a: once the first KEX has completed, a re-key's host key must be
    /// the same one — a different key is refused without consulting the
    /// known_hosts policy (no prompt, no save), the same key is accepted.
    #[test]
    fn build_verifier_pins_host_key_across_rekey() {
        let mut rng = OsRng;
        let over = AlgoOverrides {
            host_key_algorithms: Some(vec!["ssh-ed25519".to_string()]),
            ..Default::default()
        };
        let server_hk = Ed25519HostKey::from_seed([5u8; 32]);
        let v = LOCAL_VERSION.as_bytes();
        let mut server = KexRunner::new(Role::Server, build_default_kexinit(&mut rng, &over));
        let mut client = KexRunner::new(Role::Client, build_default_kexinit(&mut rng, &over));
        let (mut s_codec, mut c_codec) = (PacketCodec::new(), PacketCodec::new());
        let mut to_client: VecDeque<Vec<u8>> = server.start(&mut rng).unwrap().outbound.into();
        let mut to_server: VecDeque<Vec<u8>> = client.start(&mut rng).unwrap().outbound.into();
        let mut done = (false, false);
        for _ in 0..16 {
            while let Some(p) = to_server.pop_front() {
                let adv = server
                    .on_packet(&mut rng, &mut s_codec, &p, Some(&server_hk), None, v, v)
                    .unwrap();
                done.0 |= adv.completed;
                to_client.extend(adv.outbound);
            }
            while let Some(p) = to_client.pop_front() {
                let verifier = if p[0] == SSH_MSG_KEX_ECDH_REPLY {
                    Some(
                        host_key_verify_by_name("ssh-ed25519", reply_host_key(&p).unwrap())
                            .unwrap(),
                    )
                } else {
                    None
                };
                let adv = client
                    .on_packet(&mut rng, &mut c_codec, &p, None, verifier.as_deref(), v, v)
                    .unwrap();
                done.1 |= adv.completed;
                to_server.extend(adv.outbound);
            }
            if done == (true, true) {
                break;
            }
        }
        assert_eq!(done, (true, true), "kex did not complete");
        assert!(client.initial_host_key_sig().is_some());

        let prompts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let p2 = prompts.clone();
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        let mut kh = KnownHostsPolicy::strict(store.clone());
        kh.on_unknown = TofuAction::Prompt(Arc::new(move |_, _, _, _| {
            p2.fetch_add(1, Ordering::SeqCst);
            true
        }));
        let policy = HostKeyPolicy::KnownHosts(kh);
        let flag = AtomicBool::new(false);
        let other = Ed25519HostKey::from_seed([6u8; 32]).public_blob();
        let r = build_verifier(
            &reply_with_k_s(&other),
            &policy,
            &client,
            "h.example",
            22,
            None,
            0,
            &flag,
        );
        assert!(matches!(r, Err(Error::HostKeyRejected)));
        let same = server_hk.public_blob();
        build_verifier(
            &reply_with_k_s(&same),
            &policy,
            &client,
            "h.example",
            22,
            None,
            0,
            &flag,
        )
        .unwrap();
        assert_eq!(prompts.load(Ordering::SeqCst), 0, "no re-prompt on re-key");
        assert!(store.lock().unwrap().to_bytes().is_empty(), "no re-save");
    }

    fn reply_with_k_s(k_s: &[u8]) -> Vec<u8> {
        let mut reply = vec![SSH_MSG_KEX_ECDH_REPLY];
        reply.extend_from_slice(&(k_s.len() as u32).to_be_bytes());
        reply.extend_from_slice(k_s);
        reply
    }

    /// CLI-1: the key type written to known_hosts on a TOFU add comes from
    /// the negotiated algorithm, and a `K_S` whose type string is not a
    /// well-formed key of that algorithm (here one smuggling a newline and
    /// a wildcard line) is rejected before anything is saved.
    #[test]
    fn build_verifier_rejects_injected_key_type_before_saving() {
        let runner = negotiated_ed25519_runner();
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        let mut kh = KnownHostsPolicy::strict(store.clone());
        kh.on_unknown = TofuAction::Accept;
        let policy = HostKeyPolicy::KnownHosts(kh);
        let flag = AtomicBool::new(false);

        let mut evil = Vec::new();
        for part in [
            b"ssh-ed25519 AAAA\n* ssh-ed25519 YXR0YWNrZXJrZXk=\n#".as_slice(),
            &[7u8; 32],
        ] {
            evil.extend_from_slice(&(part.len() as u32).to_be_bytes());
            evil.extend_from_slice(part);
        }
        let r = build_verifier(
            &reply_with_k_s(&evil),
            &policy,
            &runner,
            "victim.example",
            22,
            None,
            0,
            &flag,
        );
        assert!(r.is_err());
        assert!(store.lock().unwrap().to_bytes().is_empty(), "nothing saved");

        // A real ed25519 key is TOFU-added under its plain type.
        let good = Ed25519HostKey::from_seed([3u8; 32]).public_blob();
        build_verifier(
            &reply_with_k_s(&good),
            &policy,
            &runner,
            "victim.example",
            22,
            None,
            0,
            &flag,
        )
        .unwrap();
        let saved = String::from_utf8(store.lock().unwrap().to_bytes()).unwrap();
        assert!(saved.starts_with("victim.example ssh-ed25519 "), "{saved}");
        assert_eq!(saved.lines().count(), 1);
    }

    /// The serve-loop ingress path is bounded: chunks beyond the handler
    /// queue's capacity are parked (uncredited) and only delivered — and
    /// credited — once the handler drains, with EOF kept in order behind
    /// parked data.
    #[test]
    fn serve_runtime_ingress_is_bounded_and_credits_on_delivery() {
        let (ingress_tx, ingress_rx) = mpsc::sync_channel::<Option<Vec<u8>>>(2);
        let (_egress_tx, egress_rx) = mpsc::sync_channel::<ChannelEgress>(1);
        let mut rt = ServeRuntime::new(ingress_tx, egress_rx);

        assert_eq!(rt.offer_ingress(Some(vec![1; 10])), 10);
        assert_eq!(rt.offer_ingress(Some(vec![2; 20])), 20);
        // Queue full: parked, no credit.
        assert_eq!(rt.offer_ingress(Some(vec![3; 30])), 0);
        assert_eq!(rt.offer_ingress(None), 0);
        assert_eq!(rt.ingress_backlog.len(), 2);
        // Still full: nothing moves.
        assert_eq!(rt.flush_ingress(), 0);

        // Handler drains one chunk → one parked chunk fits and is credited;
        // the EOF marker still waits behind nothing else but a full queue.
        assert_eq!(ingress_rx.try_recv().unwrap().unwrap().len(), 10);
        assert_eq!(rt.flush_ingress(), 30);
        assert_eq!(rt.ingress_backlog.len(), 1);
        assert_eq!(ingress_rx.try_recv().unwrap().unwrap().len(), 20);
        assert_eq!(rt.flush_ingress(), 0); // EOF delivered, 0 bytes
        assert!(rt.ingress_backlog.is_empty());
        assert_eq!(ingress_rx.try_recv().unwrap().unwrap().len(), 30);
        assert!(ingress_rx.try_recv().unwrap().is_none());

        // Handler gone: parked data is discarded rather than retained.
        drop(ingress_rx);
        assert_eq!(rt.offer_ingress(Some(vec![4; 40])), 0);
        assert!(rt.ingress_backlog.is_empty());
    }

    /// Every dispatcher answers the peer's unsolicited events through
    /// `unsolicited_reply`: a server-initiated open we don't handle gets
    /// OPEN_FAILURE (and its phantom channel state is dropped), an open the
    /// connection layer already refused has its payload forwarded, and a
    /// `want_reply` global request (sshd keepalive) gets REQUEST_FAILURE.
    #[test]
    fn unsolicited_reply_answers_opens_and_global_requests() {
        use crate::channel::{
            GlobalRequest, MSG_CHANNEL_OPEN_FAILURE, MSG_REQUEST_FAILURE,
            SSH_OPEN_RESOURCE_SHORTAGE,
        };
        let mut peer = ConnectionState::new();
        let mut conn = ConnectionState::new();

        // Peer opens a channel at us.
        let (_peer_id, open) = peer
            .open(ChannelOpen::ForwardedTcpip {
                dest_host: "h".into(),
                dest_port: 1,
                orig_host: "o".into(),
                orig_port: 2,
            })
            .unwrap();
        let ev = conn.on_packet(&open).unwrap();
        let ChannelEvent::OpenRequest { channel, .. } = &ev else {
            panic!("expected OpenRequest, got {ev:?}");
        };
        let local_id = *channel;
        assert!(
            conn.channel(local_id).is_some(),
            "state allocated by on_packet"
        );
        let reply = unsolicited_reply(&mut conn, &ev)
            .unwrap()
            .expect("open must be answered");
        assert_eq!(reply[0], MSG_CHANNEL_OPEN_FAILURE);
        assert!(conn.channel(local_id).is_none(), "phantom state dropped");
        // The peer understands the reply as a rejection of its open.
        assert!(matches!(
            peer.on_packet(&reply).unwrap(),
            ChannelEvent::OpenFailed {
                reason: SSH_OPEN_ADMINISTRATIVELY_PROHIBITED,
                ..
            }
        ));

        // Pre-rejected open: payload is passed through verbatim.
        let ev = ChannelEvent::OpenRejected {
            payload: vec![MSG_CHANNEL_OPEN_FAILURE, 9],
            reason: SSH_OPEN_RESOURCE_SHORTAGE,
        };
        assert_eq!(
            unsolicited_reply(&mut conn, &ev).unwrap().unwrap(),
            vec![MSG_CHANNEL_OPEN_FAILURE, 9]
        );

        // Keepalive-style global request with want_reply → REQUEST_FAILURE;
        // without want_reply → nothing.
        let probe = peer.send_global_request(GlobalRequest::Keepalive, true);
        let ev = conn.on_packet(&probe).unwrap();
        let reply = unsolicited_reply(&mut conn, &ev).unwrap().unwrap();
        assert_eq!(reply, vec![MSG_REQUEST_FAILURE]);
        assert!(matches!(
            peer.on_packet(&reply).unwrap(),
            ChannelEvent::GlobalFailure
        ));
        let silent = peer.send_global_request(GlobalRequest::Keepalive, false);
        let ev = conn.on_packet(&silent).unwrap();
        assert!(unsolicited_reply(&mut conn, &ev).unwrap().is_none());

        // Events that need no reply.
        assert!(
            unsolicited_reply(&mut conn, &ChannelEvent::GlobalFailure)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn sanitize_for_terminal_strips_escape_sequences() {
        // ESC-based CSI, a C1 CSI (U+009B), OSC/BEL, DEL, CR and a NUL are
        // all neutralised; newline / tab and ordinary text survive.
        let raw = b"ok\x1b[31mred\x1b[0m \xc2\x9b2J \x1b]8;;http://x\x07link \x7f\r\x00\nnext\tcol";
        let out = sanitize_for_terminal(raw);
        assert_eq!(out, "ok?[31mred?[0m ?2J ?]8;;http://x?link ???\nnext\tcol");
        // Invalid UTF-8 is replaced rather than dropped or passed through.
        assert_eq!(sanitize_for_terminal(b"a\xffb"), "a\u{fffd}b");
    }

    #[test]
    fn exec_output_constructible() {
        let _ = ExecOutput {
            stdout: Vec::new(),
            stderr: Vec::new(),
            exit_status: Some(0),
            exit_signal: None,
        };
    }

    fn run_server(
        listener: TcpListener,
        host_key_seed: [u8; 32],
    ) -> thread::JoinHandle<std::result::Result<Vec<u8>, String>> {
        thread::spawn(move || -> std::result::Result<Vec<u8>, String> {
            let (mut s, _) = listener.accept().map_err(|e| e.to_string())?;
            let server_hk = Ed25519HostKey::from_seed(host_key_seed);

            s.write_all(&VersionExchange::outgoing_bytes())
                .map_err(|e| e.to_string())?;
            let mut line = Vec::new();
            let v_c: Vec<u8> = {
                read_line(&mut s, &mut line, 1024).map_err(|e| format!("{e:?}"))?;
                if !line.starts_with(b"SSH-") {
                    return Err("client did not send SSH banner".into());
                }
                let parsed = VersionExchange::parse_remote(&line).map_err(|e| format!("{e:?}"))?;
                parsed.into_bytes()
            };
            let v_s = LOCAL_VERSION.as_bytes().to_vec();

            let mut codec = PacketCodec::new();
            // This minimal test server holds a single plain Ed25519 host key, so
            // it must advertise only that algorithm — not the certificate
            // key-types the client builder lists by default (which would let
            // negotiation pick a cert name this server cannot satisfy).
            let server_over = AlgoOverrides {
                host_key_algorithms: Some(vec!["ssh-ed25519".to_string()]),
                ..Default::default()
            };
            let advert = build_default_kexinit(&mut OsRng, &server_over);
            let mut runner = KexRunner::new(Role::Server, advert);

            let mut inbox: Vec<u8> = Vec::new();
            let mut rng = OsRng;

            let initial = runner.start(&mut rng).map_err(|e| format!("{e:?}"))?;
            for p in initial.outbound {
                let frame = codec.encode(&p, &mut rng).map_err(|e| format!("{e:?}"))?;
                s.write_all(&frame).map_err(|e| e.to_string())?;
            }

            let mut steps = 0;
            loop {
                steps += 1;
                if steps > MAX_KEX_STEPS {
                    return Err("server kex did not converge".into());
                }
                let payload = read_one_packet_local(&mut s, &mut codec, &mut inbox)
                    .map_err(|e| format!("{e:?}"))?;
                let adv = runner
                    .on_packet(
                        &mut rng,
                        &mut codec,
                        &payload,
                        Some(&server_hk),
                        None,
                        &v_c,
                        &v_s,
                    )
                    .map_err(|e| format!("{e:?}"))?;
                for p in adv.outbound {
                    let frame = codec.encode(&p, &mut rng).map_err(|e| format!("{e:?}"))?;
                    s.write_all(&frame).map_err(|e| e.to_string())?;
                }
                if adv.completed {
                    break;
                }
            }

            let sid = runner.session_id().unwrap().to_vec();
            Ok(sid)
        })
    }

    fn read_one_packet_local(
        s: &mut TcpStream,
        codec: &mut PacketCodec,
        inbox: &mut Vec<u8>,
    ) -> Result<Vec<u8>> {
        loop {
            if let Some((payload, consumed)) = codec.decode(inbox)? {
                inbox.drain(..consumed);
                return Ok(payload);
            }
            let mut tmp = [0u8; 4096];
            let n = s.read(&mut tmp)?;
            if n == 0 {
                return Err(Error::Protocol("connection closed"));
            }
            inbox.extend_from_slice(&tmp[..n]);
            if inbox.len() > MAX_INBOX_BYTES {
                return Err(Error::Protocol("inbound buffer too large"));
            }
        }
    }

    #[test]
    fn handshake_over_real_loopback_socket() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        let client = Client::connect(addr, Config::insecure()).expect("client connect");
        let server_sid = server.join().unwrap().expect("server handshake");
        assert_eq!(client.session_id(), server_sid.as_slice());
        assert!(!client.session_id().is_empty());
    }

    /// Drive the *client* read loop to answer a `ping@openssh.com` PING with
    /// a PONG echoing the data. A minimal loopback server completes KEX, sends
    /// an encrypted `SSH2_MSG_PING`, and asserts the client replies with the
    /// matching `SSH2_MSG_PONG`. Also confirms the client's read loop drops a
    /// stray inbound PONG (we send one before the PING and the client never
    /// surfaces it).
    #[test]
    fn client_answers_ping_with_pong() {
        use crate::transport::ping::{SSH_MSG_PONG, encode_ping, encode_pong};

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);

        let server = thread::spawn(move || -> std::result::Result<Vec<u8>, String> {
            let (mut s, _) = listener.accept().map_err(|e| e.to_string())?;
            let server_hk = Ed25519HostKey::from_seed(seed);

            s.write_all(&VersionExchange::outgoing_bytes())
                .map_err(|e| e.to_string())?;
            let mut line = Vec::new();
            let v_c: Vec<u8> = {
                read_line(&mut s, &mut line, 1024).map_err(|e| format!("{e:?}"))?;
                let parsed = VersionExchange::parse_remote(&line).map_err(|e| format!("{e:?}"))?;
                parsed.into_bytes()
            };
            let v_s = LOCAL_VERSION.as_bytes().to_vec();

            let mut codec = PacketCodec::new();
            let server_over = AlgoOverrides {
                host_key_algorithms: Some(vec!["ssh-ed25519".to_string()]),
                ..Default::default()
            };
            let advert = build_default_kexinit(&mut OsRng, &server_over);
            let mut runner = KexRunner::new(Role::Server, advert);
            let mut inbox: Vec<u8> = Vec::new();
            let mut rng = OsRng;

            let initial = runner.start(&mut rng).map_err(|e| format!("{e:?}"))?;
            for p in initial.outbound {
                let frame = codec.encode(&p, &mut rng).map_err(|e| format!("{e:?}"))?;
                s.write_all(&frame).map_err(|e| e.to_string())?;
            }
            let mut steps = 0;
            loop {
                steps += 1;
                if steps > MAX_KEX_STEPS {
                    return Err("server kex did not converge".into());
                }
                let payload = read_one_packet_local(&mut s, &mut codec, &mut inbox)
                    .map_err(|e| format!("{e:?}"))?;
                let adv = runner
                    .on_packet(
                        &mut rng,
                        &mut codec,
                        &payload,
                        Some(&server_hk),
                        None,
                        &v_c,
                        &v_s,
                    )
                    .map_err(|e| format!("{e:?}"))?;
                for p in adv.outbound {
                    let frame = codec.encode(&p, &mut rng).map_err(|e| format!("{e:?}"))?;
                    s.write_all(&frame).map_err(|e| e.to_string())?;
                }
                if adv.completed {
                    break;
                }
            }

            // Post-KEX: send a stray PONG (client must drop it), then a PING.
            let stray = encode_pong(b"unsolicited");
            let frame = codec
                .encode(&stray, &mut rng)
                .map_err(|e| format!("{e:?}"))?;
            s.write_all(&frame).map_err(|e| e.to_string())?;
            let ping = encode_ping(b"chaff-1234");
            let frame = codec
                .encode(&ping, &mut rng)
                .map_err(|e| format!("{e:?}"))?;
            s.write_all(&frame).map_err(|e| e.to_string())?;

            // Expect the client's PONG echoing our PING data.
            let reply = read_one_packet_local(&mut s, &mut codec, &mut inbox)
                .map_err(|e| format!("{e:?}"))?;
            if reply.first().copied() != Some(SSH_MSG_PONG) {
                return Err(format!("expected PONG, got msg {:?}", reply.first()));
            }
            let mut r = crate::format::Reader::new(&reply);
            r.read_u8().map_err(|e| format!("{e:?}"))?;
            let data = r.read_string().map_err(|e| format!("{e:?}"))?;
            if data != b"chaff-1234" {
                return Err(format!("PONG echoed wrong data: {data:?}"));
            }
            Ok(b"ok".to_vec())
        });

        let mut client = Client::connect(addr, Config::insecure()).expect("client connect");
        // Pump the client read loop: it must drop the stray PONG and answer
        // the PING by writing a PONG back (which our server thread asserts).
        // The server closes after reading our PONG, so this eventually errors
        // with a closed connection — that's the success path here.
        let _ = client.read_one_packet();

        let res = server.join().unwrap();
        assert_eq!(res.expect("server PONG assertions"), b"ok");
    }

    #[test]
    fn fingerprint_mismatch_rejected() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();

        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        let cfg = Config {
            host_key_policy: HostKeyPolicy::AcceptFingerprint([0xffu8; 32]),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };
        let err = Client::connect(addr, cfg).err().expect("must fail");
        assert!(matches!(err, Error::HostKeyRejected));
        // The server thread may have errored after our connect dropped — that's fine.
        let _ = server.join();
    }

    /// `PromptDetailed` on an unknown host: the callback sees
    /// `HostKeyChange::Unknown`, a `SHA256:` fingerprint and an empty
    /// `existing` slice; returning `true` adds the key (a later lookup of
    /// the real server key matches).
    #[test]
    fn known_hosts_promptdetailed_unknown_adds_entry() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        // (change, fingerprint, existing.len()) recorded by the callback.
        type SeenUnknown = Arc<Mutex<Option<(HostKeyChange, String, usize)>>>;
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        let seen: SeenUnknown = Arc::new(Mutex::new(None));
        let seen_cb = seen.clone();
        let cb: Arc<HostKeyPromptFn> = Arc::new(move |p: &HostKeyPrompt<'_>| {
            *seen_cb.lock().unwrap() = Some((p.change, p.fingerprint.clone(), p.existing.len()));
            true
        });
        let policy = KnownHostsPolicy {
            store: store.clone(),
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::PromptDetailed(cb),
            on_mismatch: TofuAction::Reject,
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        };
        let cfg = Config {
            host_key_policy: HostKeyPolicy::KnownHosts(policy),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };

        let _client =
            Client::connect_to_host("127.0.0.1", addr.port(), cfg).expect("connect should succeed");
        let _ = server.join();

        let (change, fp, existing_len) = seen.lock().unwrap().take().expect("callback fired");
        assert_eq!(change, HostKeyChange::Unknown);
        assert!(fp.starts_with("SHA256:"), "fingerprint was {fp}");
        assert_eq!(existing_len, 0);

        // The real server key is now trusted.
        let real_blob = Ed25519HostKey::from_seed(seed).public_blob();
        assert!(matches!(
            store
                .lock()
                .unwrap()
                .lookup("127.0.0.1", addr.port(), "ssh-ed25519", &real_blob),
            LookupResult::Match
        ));
    }

    /// `PromptDetailed` returning `false` refuses the connection and adds
    /// nothing to the store.
    #[test]
    fn known_hosts_promptdetailed_unknown_reject() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        let store = Arc::new(Mutex::new(KnownHosts::new()));
        let cb: Arc<HostKeyPromptFn> = Arc::new(|_p: &HostKeyPrompt<'_>| false);
        let policy = KnownHostsPolicy {
            store: store.clone(),
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::PromptDetailed(cb),
            on_mismatch: TofuAction::Reject,
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        };
        let cfg = Config {
            host_key_policy: HostKeyPolicy::KnownHosts(policy),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };

        let err = Client::connect_to_host("127.0.0.1", addr.port(), cfg)
            .err()
            .expect("connect must fail");
        assert!(matches!(err, Error::HostKeyRejected));
        let _ = server.join();

        let real_blob = Ed25519HostKey::from_seed(seed).public_blob();
        assert!(matches!(
            store
                .lock()
                .unwrap()
                .lookup("127.0.0.1", addr.port(), "ssh-ed25519", &real_blob),
            LookupResult::Unknown
        ));
    }

    /// `PromptDetailed` on a changed key: the callback sees
    /// `HostKeyChange::Changed` plus the previously-stored key in
    /// `existing`; returning `true` rotates the stored entry to the new key.
    #[test]
    fn known_hosts_promptdetailed_mismatch_rotates() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        // Pre-seed the store with a *different* key for this host so the
        // lookup trips the mismatch path.
        let bogus_blob = Ed25519HostKey::from_seed([0x11u8; 32]).public_blob();
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        store
            .lock()
            .unwrap()
            .add("127.0.0.1", addr.port(), "ssh-ed25519", &bogus_blob, false);

        // (change, existing fingerprints) recorded by the callback.
        type SeenChanged = Arc<Mutex<Option<(HostKeyChange, Vec<String>)>>>;
        let seen: SeenChanged = Arc::new(Mutex::new(None));
        let seen_cb = seen.clone();
        let cb: Arc<HostKeyPromptFn> = Arc::new(move |p: &HostKeyPrompt<'_>| {
            *seen_cb.lock().unwrap() = Some((p.change, p.existing_fingerprints()));
            true
        });
        let policy = KnownHostsPolicy {
            store: store.clone(),
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::Reject,
            on_mismatch: TofuAction::PromptDetailed(cb),
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        };
        let cfg = Config {
            host_key_policy: HostKeyPolicy::KnownHosts(policy),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };

        let _client =
            Client::connect_to_host("127.0.0.1", addr.port(), cfg).expect("connect should succeed");
        let _ = server.join();

        let (change, old_fps) = seen.lock().unwrap().take().expect("callback fired");
        assert_eq!(change, HostKeyChange::Changed);
        assert_eq!(old_fps, vec![host_key_fingerprint(&bogus_blob)]);

        // New key trusted, old key rotated out.
        let real_blob = Ed25519HostKey::from_seed(seed).public_blob();
        let guard = store.lock().unwrap();
        assert!(matches!(
            guard.lookup("127.0.0.1", addr.port(), "ssh-ed25519", &real_blob),
            LookupResult::Match
        ));
        assert!(matches!(
            guard.lookup("127.0.0.1", addr.port(), "ssh-ed25519", &bogus_blob),
            LookupResult::Mismatch { .. }
        ));
    }

    /// A `@revoked` host key is refused no matter how lenient `on_mismatch`
    /// is — here `AcceptWithWarning` (StrictHostKeyChecking=no) — and the
    /// store is left exactly as it was (no rotation, no deleted marker).
    #[test]
    fn known_hosts_revoked_key_fails_closed_regardless_of_policy() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        let real_blob = Ed25519HostKey::from_seed(seed).public_blob();
        let line = format!(
            "@revoked [127.0.0.1]:{} ssh-ed25519 {}\n",
            addr.port(),
            crate::key::base64::encode(&real_blob)
        );
        let store = Arc::new(Mutex::new(KnownHosts::from_bytes(line.as_bytes())));
        let policy = KnownHostsPolicy {
            store: store.clone(),
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::Accept,
            on_mismatch: TofuAction::AcceptWithWarning,
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        };
        let cfg = Config {
            host_key_policy: HostKeyPolicy::KnownHosts(policy),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };

        let err = Client::connect_to_host("127.0.0.1", addr.port(), cfg)
            .err()
            .expect("revoked host key must be refused");
        assert!(matches!(err, Error::HostKeyRejected), "got {err:?}");
        let _ = server.join();

        // Store untouched: the revoke line survives and still fires.
        let guard = store.lock().unwrap();
        assert_eq!(guard.to_bytes(), line.as_bytes());
        assert!(matches!(
            guard.lookup("127.0.0.1", addr.port(), "ssh-ed25519", &real_blob),
            LookupResult::Revoked
        ));
    }

    /// Accepting a changed key rotates only the literal entry for this
    /// host; a `@cert-authority` wildcard (and a plain wildcard) shared
    /// with other hosts must survive the rotation.
    #[test]
    fn known_hosts_rotation_spares_shared_marker_and_wildcard_lines() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        let bogus_blob = Ed25519HostKey::from_seed([0x22u8; 32]).public_blob();
        // Both wildcard lines genuinely match this host:port (a bare `*`
        // only covers port 22), so the broad `remove` would have deleted
        // them along with the literal entry.
        let shared = format!(
            "@cert-authority [*]:{p} ssh-ed25519 AAAA\n[*]:{p} ssh-rsa AAAB\n",
            p = addr.port()
        );
        let store = Arc::new(Mutex::new(KnownHosts::from_bytes(shared.as_bytes())));
        store
            .lock()
            .unwrap()
            .add("127.0.0.1", addr.port(), "ssh-ed25519", &bogus_blob, false);
        let policy = KnownHostsPolicy {
            store: store.clone(),
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::Reject,
            on_mismatch: TofuAction::Accept,
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        };
        let cfg = Config {
            host_key_policy: HostKeyPolicy::KnownHosts(policy),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };

        let _client =
            Client::connect_to_host("127.0.0.1", addr.port(), cfg).expect("connect should succeed");
        let _ = server.join();

        let real_blob = Ed25519HostKey::from_seed(seed).public_blob();
        let guard = store.lock().unwrap();
        let out = String::from_utf8(guard.to_bytes()).unwrap();
        assert!(out.starts_with(&shared), "shared lines must survive: {out}");
        assert!(matches!(
            guard.lookup("127.0.0.1", addr.port(), "ssh-ed25519", &real_blob),
            LookupResult::Match
        ));
        assert!(!out.contains(&crate::key::base64::encode(&bogus_blob)));
    }

    /// StrictHostKeyChecking=no over a *changed* key connects, but the
    /// session is flagged and — as in OpenSSH — password authentication,
    /// agent / X11 forwarding and port forwarding are refused with a
    /// clear error before anything hits the wire.
    #[test]
    fn accept_with_warning_on_changed_key_disables_password_and_forwarding() {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let mut seed = [0u8; 32];
        OsRng.fill_bytes(&mut seed);
        let server = run_server(listener, seed);

        let bogus_blob = Ed25519HostKey::from_seed([0x33u8; 32]).public_blob();
        let store = Arc::new(Mutex::new(KnownHosts::new()));
        store
            .lock()
            .unwrap()
            .add("127.0.0.1", addr.port(), "ssh-ed25519", &bogus_blob, false);
        let policy = KnownHostsPolicy {
            store: store.clone(),
            save_path: None,
            hash_new: false,
            on_unknown: TofuAction::Reject,
            on_mismatch: TofuAction::AcceptWithWarning,
            global: Vec::new(),
            host_key_alias: None,
            revoked: None,
            command: None,
        };
        let cfg = Config {
            host_key_policy: HostKeyPolicy::KnownHosts(policy),
            timeout: None,
            algorithms: Default::default(),
            pinned_host_key: None,
        };

        let mut client =
            Client::connect_to_host("127.0.0.1", addr.port(), cfg).expect("connect should succeed");
        let _ = server.join();
        assert!(client.host_key_mismatched());

        let blocked =
            |r: Result<()>| matches!(r, Err(Error::Config(m)) if m == HOST_KEY_MISMATCH_DISABLED);
        assert!(blocked(client.authenticate_password("alice", "hunter2")));
        assert!(blocked(
            client.authenticate("alice", vec![ClientCredential::None])
        ));
        assert!(blocked(
            client.request_tcpip_forward("127.0.0.1", 0).map(drop)
        ));
        assert!(blocked(client.request_streamlocal_forward("/tmp/x.sock")));
        assert!(blocked(client.open_session_for_agent_forward().map(drop)));
        assert!(blocked(
            client
                .open_session_for_x11_forward(false, "MIT-MAGIC-COOKIE-1", "00", 0)
                .map(drop)
        ));
        // The same over a SharedClient (`-L`, `-D`, `ssh -O forward`).
        #[cfg(feature = "multichannel")]
        {
            let shared: crate::shared::SharedClient = client.into();
            assert!(shared.host_key_mismatched());
            assert!(blocked(
                shared.open_direct_tcpip("h", 1, "127.0.0.1", 0).map(drop)
            ));
            assert!(blocked(shared.open_direct_streamlocal("/x").map(drop)));
            assert!(blocked(
                shared.request_tcpip_forward("127.0.0.1", 0).map(drop)
            ));
            assert!(blocked(shared.request_streamlocal_forward("/x")));
        }
        // The stored (old) key was NOT rotated.
        assert!(matches!(
            store
                .lock()
                .unwrap()
                .lookup("127.0.0.1", addr.port(), "ssh-ed25519", &bogus_blob),
            LookupResult::Match
        ));
    }

    #[test]
    fn filter_credentials_after_mismatch_keeps_only_publickey() {
        let key: Box<dyn HostKey> = Box::new(Ed25519HostKey::from_seed([1u8; 32]));
        let kept = filter_credentials_after_mismatch(vec![
            ClientCredential::Password("pw".into()),
            ClientCredential::None,
            ClientCredential::PublicKey(key),
            ClientCredential::PasswordPrompt(Box::new(|_| None)),
        ])
        .expect("publickey survives");
        assert_eq!(kept.len(), 2);
        assert!(matches!(kept[0], ClientCredential::None));
        assert!(matches!(kept[1], ClientCredential::PublicKey(_)));

        let err = filter_credentials_after_mismatch(vec![
            ClientCredential::Password("pw".into()),
            ClientCredential::None,
        ])
        .err()
        .expect("nothing usable left");
        assert!(matches!(err, Error::Config(m) if m == HOST_KEY_MISMATCH_DISABLED));
    }

    #[test]
    fn auth_packet_gate_blocks_password_and_kbdint_only() {
        use crate::auth::message::{SSH_MSG_USERAUTH_INFO_RESPONSE, SSH_MSG_USERAUTH_REQUEST};
        use crate::format::Writer;
        let req = |method: &str| {
            let mut w = Writer::new();
            w.write_u8(SSH_MSG_USERAUTH_REQUEST);
            w.write_string(b"alice");
            w.write_string(b"ssh-connection");
            w.write_string(method.as_bytes());
            w.into_vec()
        };
        assert!(auth_packet_blocked_after_mismatch(&req("password")));
        assert!(auth_packet_blocked_after_mismatch(&req(
            "keyboard-interactive"
        )));
        assert!(!auth_packet_blocked_after_mismatch(&req("publickey")));
        assert!(!auth_packet_blocked_after_mismatch(&req("none")));
        assert!(auth_packet_blocked_after_mismatch(&[
            SSH_MSG_USERAUTH_INFO_RESPONSE,
            0,
            0,
            0,
            0
        ]));
        // Truncated / unrelated packets are not blocked (they are not
        // credentials); the auth driver rejects garbage on its own.
        assert!(!auth_packet_blocked_after_mismatch(&[
            SSH_MSG_USERAUTH_REQUEST
        ]));
        assert!(!auth_packet_blocked_after_mismatch(&[5, 0, 0, 0, 0]));
        assert!(!auth_packet_blocked_after_mismatch(&[]));
    }

    #[test]
    fn client_advert_no_overrides_matches_defaults() {
        let advert = build_default_kexinit(&mut OsRng, &AlgoOverrides::default());
        // Ciphers/MACs/host-key advertise the built-in defaults.
        let want_ciphers: Vec<String> = defaults::CIPHERS.iter().map(|s| s.to_string()).collect();
        assert_eq!(advert.ciphers_c2s, want_ciphers);
        assert_eq!(advert.ciphers_s2c, want_ciphers);
        // The kex list ends with the client's strict-kex marker only.
        let n = advert.kex.len();
        assert_eq!(
            advert.kex[n - 1],
            crate::transport::kex::STRICT_KEX_CLIENT_MARKER
        );
        assert!(!crate::transport::kex::is_strict_kex_marker(
            &advert.kex[n - 2]
        ));
    }

    #[test]
    fn client_cipher_override_replaces_and_keeps_markers() {
        let over = AlgoOverrides {
            ciphers: Some(vec!["aes128-ctr".to_string()]),
            ..Default::default()
        };
        let advert = build_default_kexinit(&mut OsRng, &over);
        assert_eq!(advert.ciphers_c2s, vec!["aes128-ctr".to_string()]);
        assert_eq!(advert.ciphers_s2c, vec!["aes128-ctr".to_string()]);
        // Markers regained even though we overrode an unrelated category.
        assert!(
            advert
                .kex
                .iter()
                .any(|k| crate::transport::kex::is_strict_kex_marker(k))
        );
    }

    #[test]
    fn client_kex_override_reappends_markers() {
        let over = AlgoOverrides {
            // A single real kex; markers must still come back.
            kex_algorithms: Some(vec!["curve25519-sha256".to_string()]),
            ..Default::default()
        };
        let advert = build_default_kexinit(&mut OsRng, &over);
        assert_eq!(advert.kex[0], "curve25519-sha256");
        let markers = advert
            .kex
            .iter()
            .filter(|k| crate::transport::kex::is_strict_kex_marker(k))
            .count();
        assert_eq!(markers, 1, "the client's strict-kex marker is re-appended");
    }

    #[test]
    fn restricted_client_ciphers_negotiate_against_default_server() {
        use crate::transport::kexinit::negotiate;
        // Client restricts ciphers to a single CTR suite; server advertises
        // defaults. Negotiation must pick the client's restricted choice.
        let client_over = AlgoOverrides {
            ciphers: Some(vec!["aes128-ctr".to_string()]),
            ..Default::default()
        };
        let client =
            build_default_kexinit(&mut OsRng, &client_over).with_ext_info_marker(Role::Client);
        let mut server = build_default_kexinit(&mut OsRng, &AlgoOverrides::default());
        // (A client-built proposal: give it the server's strict-kex marker.)
        server
            .kex
            .retain(|k| !crate::transport::kex::is_strict_kex_marker(k));
        server
            .kex
            .push(crate::transport::kex::STRICT_KEX_SERVER_MARKER.to_string());
        let neg = negotiate(&client, &server).expect("should negotiate");
        assert_eq!(neg.cipher_c2s, "aes128-ctr");
        assert_eq!(neg.cipher_s2c, "aes128-ctr");
        // Strict-kex survives an unrelated cipher override on the client.
        assert!(neg.strict_kex_enabled);
    }

    #[test]
    fn disjoint_ciphers_fail_with_no_common_algorithm() {
        use crate::transport::kexinit::negotiate;
        let client_over = AlgoOverrides {
            ciphers: Some(vec!["aes128-ctr".to_string()]),
            ..Default::default()
        };
        let server_over = AlgoOverrides {
            ciphers: Some(vec!["aes256-ctr".to_string()]),
            ..Default::default()
        };
        let client = build_default_kexinit(&mut OsRng, &client_over);
        let server = build_default_kexinit(&mut OsRng, &server_over);
        assert!(matches!(
            negotiate(&client, &server),
            Err(Error::NoMatch { .. })
        ));
    }
}