ssh.rs

// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Embedded SSH server for sandbox access.

use crate::child_env;
use crate::policy::SandboxPolicy;
use crate::process::drop_privileges;
use crate::sandbox;
#[cfg(target_os = "linux")]
use crate::{register_managed_child, unregister_managed_child};
use miette::{IntoDiagnostic, Result};
use nix::pty::{Winsize, openpty};
use nix::unistd::setsid;
use rand_core::OsRng;
use russh::keys::{Algorithm, PrivateKey};
use russh::server::{Auth, Handle, Session};
use russh::{ChannelId, CryptoVec};
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::SocketAddr;
use std::os::fd::{AsRawFd, RawFd};
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex, mpsc};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tracing::{info, warn};

const PREFACE_MAGIC: &str = "NSSH1";
#[cfg(test)]
const SSH_HANDSHAKE_SECRET_ENV: &str = "OPENSHELL_SSH_HANDSHAKE_SECRET";

/// A time-bounded set of nonces used to detect replayed NSSH1 handshakes.
/// Each entry records the `Instant` it was inserted; a background reaper task
/// periodically evicts entries older than the handshake skew window.
type NonceCache = Arc<Mutex<HashMap<String, Instant>>>;

/// Perform SSH server initialization: generate a host key, build the config,
/// and bind the TCP listener. Extracted so that startup errors can be forwarded
/// through the readiness channel rather than being silently logged.
async fn ssh_server_init(
    listen_addr: SocketAddr,
    ca_file_paths: &Option<(PathBuf, PathBuf)>,
) -> Result<(
    TcpListener,
    Arc<russh::server::Config>,
    Option<Arc<(PathBuf, PathBuf)>>,
)> {
    let mut rng = OsRng;
    let host_key = PrivateKey::random(&mut rng, Algorithm::Ed25519).into_diagnostic()?;

    let mut config = russh::server::Config {
        auth_rejection_time: Duration::from_secs(1),
        ..Default::default()
    };
    config.keys.push(host_key);

    let config = Arc::new(config);
    let ca_paths = ca_file_paths.as_ref().map(|p| Arc::new(p.clone()));
    let listener = TcpListener::bind(listen_addr).await.into_diagnostic()?;
    info!(addr = %listen_addr, "SSH server listening");

    Ok((listener, config, ca_paths))
}

#[allow(clippy::too_many_arguments)]
pub async fn run_ssh_server(
    listen_addr: SocketAddr,
    ready_tx: tokio::sync::oneshot::Sender<Result<()>>,
    policy: SandboxPolicy,
    workdir: Option<String>,
    handshake_secret: String,
    handshake_skew_secs: u64,
    netns_fd: Option<RawFd>,
    proxy_url: Option<String>,
    ca_file_paths: Option<(PathBuf, PathBuf)>,
    provider_env: HashMap<String, String>,
) -> Result<()> {
    let (listener, config, ca_paths) = match ssh_server_init(listen_addr, &ca_file_paths).await {
        Ok(v) => {
            // Signal that the SSH server has bound the socket and is ready to
            // accept connections. The parent task awaits this before spawning
            // the entrypoint process, ensuring exec requests won't race
            // against server startup.
            let _ = ready_tx.send(Ok(()));
            v
        }
        Err(err) => {
            let _ = ready_tx.send(Err(err));
            return Ok(());
        }
    };

    // Nonce cache for replay detection. Entries are evicted by a background
    // reaper once they exceed the handshake skew window.
    let nonce_cache: NonceCache = Arc::new(Mutex::new(HashMap::new()));

    // Background task that periodically purges expired nonces.
    let reaper_cache = nonce_cache.clone();
    let ttl = Duration::from_secs(handshake_skew_secs);
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(60));
        loop {
            interval.tick().await;
            if let Ok(mut cache) = reaper_cache.lock() {
                cache.retain(|_, inserted| inserted.elapsed() < ttl);
            }
        }
    });

    loop {
        let (stream, peer) = listener.accept().await.into_diagnostic()?;
        stream.set_nodelay(true).into_diagnostic()?;
        let config = config.clone();
        let policy = policy.clone();
        let workdir = workdir.clone();
        let secret = handshake_secret.clone();
        let proxy_url = proxy_url.clone();
        let ca_paths = ca_paths.clone();
        let provider_env = provider_env.clone();
        let nonce_cache = nonce_cache.clone();

        tokio::spawn(async move {
            if let Err(err) = handle_connection(
                stream,
                peer,
                config,
                policy,
                workdir,
                &secret,
                handshake_skew_secs,
                netns_fd,
                proxy_url,
                ca_paths,
                provider_env,
                &nonce_cache,
            )
            .await
            {
                warn!(error = %err, "SSH connection failed");
            }
        });
    }
}

#[allow(clippy::too_many_arguments)]
async fn handle_connection(
    mut stream: tokio::net::TcpStream,
    peer: SocketAddr,
    config: Arc<russh::server::Config>,
    policy: SandboxPolicy,
    workdir: Option<String>,
    secret: &str,
    handshake_skew_secs: u64,
    netns_fd: Option<RawFd>,
    proxy_url: Option<String>,
    ca_file_paths: Option<Arc<(PathBuf, PathBuf)>>,
    provider_env: HashMap<String, String>,
    nonce_cache: &NonceCache,
) -> Result<()> {
    info!(peer = %peer, "SSH connection: reading handshake preface");
    let mut line = String::new();
    read_line(&mut stream, &mut line).await?;
    info!(peer = %peer, preface_len = line.len(), "SSH connection: preface received, verifying");
    if !verify_preface(&line, secret, handshake_skew_secs, nonce_cache)? {
        warn!(peer = %peer, "SSH connection: handshake verification failed");
        let _ = stream.write_all(b"ERR\n").await;
        return Ok(());
    }
    stream.write_all(b"OK\n").await.into_diagnostic()?;
    info!(peer = %peer, "SSH handshake accepted");

    let handler = SshHandler::new(
        policy,
        workdir,
        netns_fd,
        proxy_url,
        ca_file_paths,
        provider_env,
    );
    russh::server::run_stream(config, stream, handler)
        .await
        .map_err(|err| miette::miette!("ssh stream error: {err}"))?;
    Ok(())
}

async fn read_line(stream: &mut tokio::net::TcpStream, buf: &mut String) -> Result<()> {
    let mut bytes = Vec::new();
    loop {
        let mut byte = [0u8; 1];
        let n = stream.read(&mut byte).await.into_diagnostic()?;
        if n == 0 {
            break;
        }
        if byte[0] == b'\n' {
            break;
        }
        bytes.push(byte[0]);
        if bytes.len() > 1024 {
            break;
        }
    }
    *buf = String::from_utf8_lossy(&bytes).to_string();
    Ok(())
}

fn verify_preface(
    line: &str,
    secret: &str,
    handshake_skew_secs: u64,
    nonce_cache: &NonceCache,
) -> Result<bool> {
    let parts: Vec<&str> = line.split_whitespace().collect();
    if parts.len() != 5 || parts[0] != PREFACE_MAGIC {
        return Ok(false);
    }
    let token = parts[1];
    let timestamp: i64 = parts[2].parse().unwrap_or(0);
    let nonce = parts[3];
    let signature = parts[4];

    let now = i64::try_from(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .into_diagnostic()?
            .as_secs(),
    )
    .into_diagnostic()?;
    let skew = (now - timestamp).unsigned_abs();
    if skew > handshake_skew_secs {
        return Ok(false);
    }

    let payload = format!("{token}|{timestamp}|{nonce}");
    let expected = hmac_sha256(secret.as_bytes(), payload.as_bytes());
    if signature != expected {
        return Ok(false);
    }

    // Reject replayed nonces. The cache is bounded by the reaper task which
    // evicts entries older than `handshake_skew_secs`.
    let mut cache = nonce_cache
        .lock()
        .map_err(|_| miette::miette!("nonce cache lock poisoned"))?;
    if cache.contains_key(nonce) {
        warn!(nonce = nonce, "NSSH1 nonce replay detected");
        return Ok(false);
    }
    cache.insert(nonce.to_string(), Instant::now());

    Ok(true)
}

fn hmac_sha256(key: &[u8], data: &[u8]) -> String {
    use hmac::{Hmac, Mac};
    use sha2::Sha256;

    let mut mac = Hmac::<Sha256>::new_from_slice(key).expect("hmac key");
    mac.update(data);
    let result = mac.finalize().into_bytes();
    hex::encode(result)
}

/// Per-channel state for tracking PTY resources and I/O senders.
///
/// Each SSH channel gets its own PTY master (if a PTY was requested) and input
/// sender.  This allows `window_change_request` to resize the correct PTY when
/// multiple channels are open simultaneously (e.g. parallel shells, shell +
/// sftp, etc.).
#[derive(Default)]
struct ChannelState {
    input_sender: Option<mpsc::Sender<Vec<u8>>>,
    pty_master: Option<std::fs::File>,
    pty_request: Option<PtyRequest>,
}

struct SshHandler {
    policy: SandboxPolicy,
    workdir: Option<String>,
    netns_fd: Option<RawFd>,
    proxy_url: Option<String>,
    ca_file_paths: Option<Arc<(PathBuf, PathBuf)>>,
    provider_env: HashMap<String, String>,
    channels: HashMap<ChannelId, ChannelState>,
}

impl SshHandler {
    fn new(
        policy: SandboxPolicy,
        workdir: Option<String>,
        netns_fd: Option<RawFd>,
        proxy_url: Option<String>,
        ca_file_paths: Option<Arc<(PathBuf, PathBuf)>>,
        provider_env: HashMap<String, String>,
    ) -> Self {
        Self {
            policy,
            workdir,
            netns_fd,
            proxy_url,
            ca_file_paths,
            provider_env,
            channels: HashMap::new(),
        }
    }
}

impl russh::server::Handler for SshHandler {
    type Error = anyhow::Error;

    async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {
        Ok(Auth::Accept)
    }

    async fn auth_publickey(
        &mut self,
        _user: &str,
        _public_key: &russh::keys::PublicKey,
    ) -> Result<Auth, Self::Error> {
        Ok(Auth::Accept)
    }

    async fn channel_open_session(
        &mut self,
        channel: russh::Channel<russh::server::Msg>,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.channels.insert(channel.id(), ChannelState::default());
        Ok(true)
    }

    /// Clean up per-channel state when the channel is closed.
    ///
    /// This is the final cleanup and subsumes `channel_eof` — if `channel_close`
    /// fires without a preceding `channel_eof`, all resources (pty_master File,
    /// input_sender) are dropped here.
    async fn channel_close(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.channels.remove(&channel);
        Ok(())
    }

    async fn channel_open_direct_tcpip(
        &mut self,
        channel: russh::Channel<russh::server::Msg>,
        host_to_connect: &str,
        port_to_connect: u32,
        _originator_address: &str,
        _originator_port: u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        // Validate port range before truncating u32 -> u16.  The SSH protocol
        // uses u32 for ports, but valid TCP ports are 0-65535.  Without this
        // check, port 65537 truncates to port 1 (privileged).
        if port_to_connect > u32::from(u16::MAX) {
            warn!(
                host = host_to_connect,
                port = port_to_connect,
                "direct-tcpip rejected: port exceeds valid TCP range (0-65535)"
            );
            return Ok(false);
        }

        // Only allow forwarding to loopback destinations to prevent the
        // sandbox SSH server from being used as a generic proxy.
        if !is_loopback_host(host_to_connect) {
            warn!(
                host = host_to_connect,
                port = port_to_connect,
                "direct-tcpip rejected: non-loopback destination"
            );
            return Ok(false);
        }

        let host = host_to_connect.to_string();
        let port = port_to_connect as u16;
        let netns_fd = self.netns_fd;

        tokio::spawn(async move {
            let addr = format!("{host}:{port}");
            let tcp = match connect_in_netns(&addr, netns_fd).await {
                Ok(stream) => stream,
                Err(err) => {
                    warn!(addr = %addr, error = %err, "direct-tcpip: failed to connect");
                    let _ = channel.close().await;
                    return;
                }
            };

            let mut channel_stream = channel.into_stream();
            let mut tcp_stream = tcp;

            let _ = tokio::io::copy_bidirectional(&mut channel_stream, &mut tcp_stream).await;
        });

        Ok(true)
    }

    async fn pty_request(
        &mut self,
        channel: ChannelId,
        term: &str,
        col_width: u32,
        row_height: u32,
        _pix_width: u32,
        _pix_height: u32,
        _modes: &[(russh::Pty, u32)],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        let state = self
            .channels
            .get_mut(&channel)
            .ok_or_else(|| anyhow::anyhow!("pty_request on unknown channel {channel:?}"))?;
        state.pty_request = Some(PtyRequest {
            term: term.to_string(),
            col_width,
            row_height,
            pixel_width: 0,
            pixel_height: 0,
        });
        session.channel_success(channel)?;
        Ok(())
    }

    async fn window_change_request(
        &mut self,
        channel: ChannelId,
        col_width: u32,
        row_height: u32,
        pixel_width: u32,
        pixel_height: u32,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        let Some(state) = self.channels.get(&channel) else {
            warn!("window_change_request on unknown channel {channel:?}");
            return Ok(());
        };
        if let Some(master) = state.pty_master.as_ref() {
            let winsize = Winsize {
                ws_row: to_u16(row_height.max(1)),
                ws_col: to_u16(col_width.max(1)),
                ws_xpixel: to_u16(pixel_width),
                ws_ypixel: to_u16(pixel_height),
            };
            if let Err(e) = unsafe_pty::set_winsize(master.as_raw_fd(), winsize) {
                warn!("failed to resize PTY for channel {channel:?}: {e}");
            }
        }
        Ok(())
    }

    async fn shell_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        session.channel_success(channel)?;
        // Only allocate a PTY when the client explicitly requested one via
        // pty_request.  VS Code Remote-SSH sends shell_request *without* a
        // preceding pty_request and expects pipe-based I/O with clean LF line
        // endings.  Forcing a PTY here caused CRLF translation which made
        // VS Code misdetect the platform as Windows (and then try to run
        // `powershell`).
        self.start_shell(channel, session.handle(), None)?;
        Ok(())
    }

    async fn exec_request(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        session.channel_success(channel)?;
        let command = String::from_utf8_lossy(data).trim().to_string();
        if command.is_empty() {
            return Ok(());
        }
        self.start_shell(channel, session.handle(), Some(command))?;
        Ok(())
    }

    async fn subsystem_request(
        &mut self,
        channel: ChannelId,
        name: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        if name == "sftp" {
            session.channel_success(channel)?;
            // sftp-server speaks the SFTP binary protocol over stdin/stdout,
            // which is exactly what spawn_pipe_exec wires up.  This enables
            // modern scp (SFTP-based, OpenSSH 9.0+) and SFTP clients to
            // transfer files into and out of the sandbox.
            let input_sender = spawn_pipe_exec(
                &self.policy,
                self.workdir.clone(),
                Some("/usr/lib/openssh/sftp-server".to_string()),
                session.handle(),
                channel,
                self.netns_fd,
                self.proxy_url.clone(),
                self.ca_file_paths.clone(),
                &self.provider_env,
            )?;
            let state = self.channels.get_mut(&channel).ok_or_else(|| {
                anyhow::anyhow!("subsystem_request on unknown channel {channel:?}")
            })?;
            state.input_sender = Some(input_sender);
        } else {
            warn!(subsystem = name, "unsupported subsystem requested");
            session.channel_failure(channel)?;
        }
        Ok(())
    }

    async fn env_request(
        &mut self,
        channel: ChannelId,
        variable_name: &str,
        variable_value: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        // Accept the env request so the client knows we handled it, but we
        // don't actually propagate the variables — the sandbox environment is
        // controlled via policy.  We must reply so VSCode doesn't stall.
        let _ = (variable_name, variable_value);
        session.channel_success(channel)?;
        Ok(())
    }

    async fn data(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        let Some(state) = self.channels.get(&channel) else {
            warn!("data on unknown channel {channel:?}");
            return Ok(());
        };
        if let Some(sender) = state.input_sender.as_ref() {
            let _ = sender.send(data.to_vec());
        }
        Ok(())
    }

    async fn channel_eof(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        // Drop the input sender so the stdin writer thread sees a
        // disconnected channel and closes the child's stdin pipe.  This
        // is essential for commands like `cat | tar xf -` which need
        // stdin EOF to know the input stream is complete.
        if let Some(state) = self.channels.get_mut(&channel) {
            state.input_sender.take();
        } else {
            warn!("channel_eof on unknown channel {channel:?}");
        }
        Ok(())
    }
}

impl SshHandler {
    fn start_shell(
        &mut self,
        channel: ChannelId,
        handle: Handle,
        command: Option<String>,
    ) -> anyhow::Result<()> {
        let state = self
            .channels
            .get_mut(&channel)
            .ok_or_else(|| anyhow::anyhow!("start_shell on unknown channel {channel:?}"))?;
        if let Some(pty) = state.pty_request.take() {
            // PTY was requested — allocate a real PTY (interactive shell or
            // exec that explicitly asked for a terminal).
            let (pty_master, input_sender) = spawn_pty_shell(
                &self.policy,
                self.workdir.clone(),
                command,
                &pty,
                handle,
                channel,
                self.netns_fd,
                self.proxy_url.clone(),
                self.ca_file_paths.clone(),
                &self.provider_env,
            )?;
            state.pty_master = Some(pty_master);
            state.input_sender = Some(input_sender);
        } else {
            // No PTY requested — use plain pipes so stdout/stderr are
            // separate and output has clean LF line endings.  This is the
            // path VSCode Remote-SSH exec commands take.
            let input_sender = spawn_pipe_exec(
                &self.policy,
                self.workdir.clone(),
                command,
                handle,
                channel,
                self.netns_fd,
                self.proxy_url.clone(),
                self.ca_file_paths.clone(),
                &self.provider_env,
            )?;
            state.input_sender = Some(input_sender);
        }
        Ok(())
    }
}

/// Connect a TCP stream to `addr` inside the sandbox network namespace.
///
/// The SSH supervisor runs in the host network namespace while sandbox child
/// processes run in an isolated network namespace (with their own loopback).
/// A plain `TcpStream::connect("127.0.0.1:port")` from the supervisor would
/// hit the host loopback, not the sandbox loopback where services are listening.
///
/// On Linux, we spawn a dedicated OS thread, call `setns` to enter the sandbox
/// namespace, create the socket there, then convert it to a tokio `TcpStream`.
/// We use `std::thread::spawn` (not `spawn_blocking`) because `setns` changes
/// the calling thread's network namespace permanently — a tokio blocking-pool
/// thread could be reused for unrelated tasks and must not be contaminated.
/// On non-Linux platforms (no network namespace support), we connect directly.
async fn connect_in_netns(
    addr: &str,
    netns_fd: Option<RawFd>,
) -> std::io::Result<tokio::net::TcpStream> {
    #[cfg(target_os = "linux")]
    if let Some(fd) = netns_fd {
        let addr = addr.to_string();
        let (tx, rx) = tokio::sync::oneshot::channel();
        std::thread::spawn(move || {
            let result = (|| -> std::io::Result<std::net::TcpStream> {
                // Enter the sandbox network namespace on this dedicated thread.
                // SAFETY: setns is safe to call; this is a dedicated thread that
                // will exit after the connection is established.
                #[allow(unsafe_code)]
                let rc = unsafe { libc::setns(fd, libc::CLONE_NEWNET) };
                if rc != 0 {
                    return Err(std::io::Error::last_os_error());
                }
                std::net::TcpStream::connect(&addr)
            })();
            let _ = tx.send(result);
        });

        let std_stream = rx
            .await
            .map_err(|_| std::io::Error::other("netns connect thread panicked"))??;
        std_stream.set_nonblocking(true)?;
        return tokio::net::TcpStream::from_std(std_stream);
    }

    #[cfg(not(target_os = "linux"))]
    let _ = netns_fd;

    tokio::net::TcpStream::connect(addr).await
}

#[derive(Clone)]
struct PtyRequest {
    term: String,
    col_width: u32,
    row_height: u32,
    pixel_width: u32,
    pixel_height: u32,
}

impl Default for PtyRequest {
    fn default() -> Self {
        Self {
            term: "xterm-256color".to_string(),
            col_width: 80,
            row_height: 24,
            pixel_width: 0,
            pixel_height: 0,
        }
    }
}

/// Derive the session USER and HOME from the policy's `run_as_user`.
///
/// Falls back to `("sandbox", "/sandbox")` when the policy has no explicit user,
/// preserving backward compatibility with images that use the default layout.
fn session_user_and_home(policy: &SandboxPolicy) -> (String, String) {
    match policy.process.run_as_user.as_deref() {
        Some(user) if !user.is_empty() => {
            // Look up the user's home directory from /etc/passwd.
            let home = nix::unistd::User::from_name(user)
                .ok()
                .flatten()
                .map(|u| u.dir.to_string_lossy().into_owned())
                .unwrap_or_else(|| format!("/home/{user}"));
            (user.to_string(), home)
        }
        _ => ("sandbox".to_string(), "/sandbox".to_string()),
    }
}

fn apply_child_env(
    cmd: &mut Command,
    session_home: &str,
    session_user: &str,
    term: &str,
    proxy_url: Option<&str>,
    ca_file_paths: Option<&(PathBuf, PathBuf)>,
    provider_env: &HashMap<String, String>,
) {
    let path = std::env::var("PATH").unwrap_or_else(|_| "/usr/local/bin:/usr/bin:/bin".into());

    cmd.env_clear()
        .env("OPENSHELL_SANDBOX", "1")
        .env("HOME", session_home)
        .env("USER", session_user)
        .env("SHELL", "/bin/bash")
        .env("PATH", &path)
        .env("TERM", term);

    if let Some(url) = proxy_url {
        for (key, value) in child_env::proxy_env_vars(url) {
            cmd.env(key, value);
        }
    }

    if let Some((ca_cert_path, combined_bundle_path)) = ca_file_paths {
        for (key, value) in child_env::tls_env_vars(ca_cert_path, combined_bundle_path) {
            cmd.env(key, value);
        }
    }

    for (key, value) in provider_env {
        cmd.env(key, value);
    }
}

#[allow(clippy::too_many_arguments)]
fn spawn_pty_shell(
    policy: &SandboxPolicy,
    workdir: Option<String>,
    command: Option<String>,
    pty: &PtyRequest,
    handle: Handle,
    channel: ChannelId,
    netns_fd: Option<RawFd>,
    proxy_url: Option<String>,
    ca_file_paths: Option<Arc<(PathBuf, PathBuf)>>,
    provider_env: &HashMap<String, String>,
) -> anyhow::Result<(std::fs::File, mpsc::Sender<Vec<u8>>)> {
    let winsize = Winsize {
        ws_row: to_u16(pty.row_height.max(1)),
        ws_col: to_u16(pty.col_width.max(1)),
        ws_xpixel: to_u16(pty.pixel_width),
        ws_ypixel: to_u16(pty.pixel_height),
    };
    let openpty = openpty(Some(&winsize), None)?;
    let master = std::fs::File::from(openpty.master);
    let slave = std::fs::File::from(openpty.slave);
    let slave_fd = slave.as_raw_fd();

    let stdin = slave.try_clone()?;
    let stdout = slave.try_clone()?;
    let stderr = slave;
    let mut reader = master.try_clone()?;
    let mut writer = master.try_clone()?;

    let mut cmd = command.map_or_else(
        || {
            let mut c = Command::new("/bin/bash");
            c.arg("-i");
            c
        },
        |command| {
            let mut c = Command::new("/bin/bash");
            c.arg("-lc").arg(command);
            c
        },
    );

    let term = if pty.term.is_empty() {
        "xterm-256color"
    } else {
        pty.term.as_str()
    };

    // Derive USER and HOME from the policy's run_as_user when available,
    // falling back to "sandbox" / "/sandbox" for backward compatibility.
    let (session_user, session_home) = session_user_and_home(policy);
    apply_child_env(
        &mut cmd,
        &session_home,
        &session_user,
        term,
        proxy_url.as_deref(),
        ca_file_paths.as_deref(),
        provider_env,
    );
    cmd.stdin(stdin).stdout(stdout).stderr(stderr);

    if let Some(dir) = workdir.as_deref() {
        cmd.current_dir(dir);
    }

    #[cfg(unix)]
    {
        unsafe_pty::install_pre_exec(
            &mut cmd,
            policy.clone(),
            workdir.clone(),
            slave_fd,
            netns_fd,
        );
    }

    let mut child = cmd.spawn()?;
    #[cfg(target_os = "linux")]
    let child_pid = child.id();
    #[cfg(target_os = "linux")]
    register_managed_child(child_pid);
    let master_file = master;

    let (sender, receiver) = mpsc::channel::<Vec<u8>>();
    std::thread::spawn(move || {
        while let Ok(bytes) = receiver.recv() {
            if writer.write_all(&bytes).is_err() {
                break;
            }
            let _ = writer.flush();
        }
    });

    let runtime = tokio::runtime::Handle::current();
    let runtime_reader = runtime.clone();
    let handle_clone = handle.clone();
    // Signal from the reader thread to the exit thread that all output has
    // been forwarded.  The exit thread waits for this before sending the
    // exit-status and closing the channel, ensuring the correct SSH protocol
    // ordering: data → EOF → exit-status → close.
    let (reader_done_tx, reader_done_rx) = mpsc::channel::<()>();
    std::thread::spawn(move || {
        let mut buf = [0u8; 4096];
        loop {
            match reader.read(&mut buf) {
                Ok(0) | Err(_) => break,
                Ok(n) => {
                    let data = CryptoVec::from_slice(&buf[..n]);
                    let handle_clone = handle_clone.clone();
                    let _ = runtime_reader
                        .block_on(async move { handle_clone.data(channel, data).await });
                }
            }
        }
        // Send EOF to indicate no more data will be sent on this channel.
        let eof_handle = handle_clone.clone();
        let _ = runtime_reader.block_on(async move { eof_handle.eof(channel).await });
        // Notify the exit thread that all output has been forwarded.
        let _ = reader_done_tx.send(());
    });

    let handle_exit = handle;
    let runtime_exit = runtime;
    std::thread::spawn(move || {
        let status = child.wait().ok();
        #[cfg(target_os = "linux")]
        unregister_managed_child(child_pid);
        let code = status.and_then(|s| s.code()).unwrap_or(1).unsigned_abs();
        // Wait for the reader thread to finish forwarding all output before
        // sending exit-status and closing the channel.  This prevents the
        // race where close() was called before exit_status_request().
        //
        // Use a timeout because a backgrounded grandchild process (e.g.
        // `nohup daemon &`) may hold the PTY slave open indefinitely,
        // preventing the reader from reaching EOF.  Two seconds is enough
        // for any remaining buffered data to drain.
        let _ = reader_done_rx.recv_timeout(Duration::from_secs(2));
        drop(runtime_exit.spawn(async move {
            let _ = handle_exit.exit_status_request(channel, code).await;
            let _ = handle_exit.close(channel).await;
        }));
    });

    Ok((master_file, sender))
}

/// Spawn a command using plain pipes (no PTY).
///
/// stdout is forwarded as SSH channel data and stderr as SSH extended data
/// (type 1), preserving the separation that clients like `VSCode` Remote-SSH
/// expect.  Output retains clean LF line endings (no CRLF translation).
#[allow(clippy::too_many_arguments)]
fn spawn_pipe_exec(
    policy: &SandboxPolicy,
    workdir: Option<String>,
    command: Option<String>,
    handle: Handle,
    channel: ChannelId,
    netns_fd: Option<RawFd>,
    proxy_url: Option<String>,
    ca_file_paths: Option<Arc<(PathBuf, PathBuf)>>,
    provider_env: &HashMap<String, String>,
) -> anyhow::Result<mpsc::Sender<Vec<u8>>> {
    let mut cmd = command.map_or_else(
        || {
            // No command — read from stdin.  Do *not* pass `-i`; interactive
            // mode reads .bashrc, writes prompts to stderr, and can introduce
            // just enough latency for VS Code Remote-SSH's platform detection
            // to time out and fall back to "windows".  Plain `bash` with piped
            // stdin already reads commands line-by-line (script mode), which is
            // exactly what VS Code's local server expects.
            Command::new("/bin/bash")
        },
        |command| {
            let mut c = Command::new("/bin/bash");
            // Use login shell (-l) so that .profile/.bashrc are sourced and
            // tool-specific env vars (VIRTUAL_ENV, UV_PYTHON_INSTALL_DIR, etc.)
            // are available without hardcoding them here.
            c.arg("-lc").arg(command);
            c
        },
    );

    let (session_user, session_home) = session_user_and_home(policy);
    apply_child_env(
        &mut cmd,
        &session_home,
        &session_user,
        "dumb",
        proxy_url.as_deref(),
        ca_file_paths.as_deref(),
        provider_env,
    );
    cmd.stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());

    if let Some(dir) = workdir.as_deref() {
        cmd.current_dir(dir);
    }

    #[cfg(unix)]
    {
        unsafe_pty::install_pre_exec_no_pty(&mut cmd, policy.clone(), workdir.clone(), netns_fd);
    }

    let mut child = cmd.spawn()?;
    #[cfg(target_os = "linux")]
    let child_pid = child.id();
    #[cfg(target_os = "linux")]
    register_managed_child(child_pid);

    let child_stdin = child.stdin.take();
    let child_stdout = child.stdout.take().expect("stdout must be piped");
    let child_stderr = child.stderr.take().expect("stderr must be piped");

    // stdin writer thread
    let (sender, receiver) = mpsc::channel::<Vec<u8>>();
    std::thread::spawn(move || {
        let Some(mut stdin) = child_stdin else {
            return;
        };
        while let Ok(bytes) = receiver.recv() {
            if stdin.write_all(&bytes).is_err() {
                break;
            }
            let _ = stdin.flush();
        }
    });

    let runtime = tokio::runtime::Handle::current();

    // Signal from the reader threads to the exit thread that all output has
    // been forwarded.
    let (reader_done_tx, reader_done_rx) = mpsc::channel::<()>();

    // stdout reader
    let stdout_handle = handle.clone();
    let stdout_runtime = runtime.clone();
    let reader_done_stdout = reader_done_tx.clone();
    std::thread::spawn(move || {
        let mut reader = child_stdout;
        let mut buf = [0u8; 4096];
        loop {
            match reader.read(&mut buf) {
                Ok(0) | Err(_) => break,
                Ok(n) => {
                    let data = CryptoVec::from_slice(&buf[..n]);
                    let h = stdout_handle.clone();
                    let _ = stdout_runtime.block_on(async move { h.data(channel, data).await });
                }
            }
        }
        let _ = reader_done_stdout.send(());
    });

    // stderr reader — sends as extended data (type 1)
    let stderr_handle = handle.clone();
    let stderr_runtime = runtime.clone();
    std::thread::spawn(move || {
        let mut reader = child_stderr;
        let mut buf = [0u8; 4096];
        loop {
            match reader.read(&mut buf) {
                Ok(0) | Err(_) => break,
                Ok(n) => {
                    let data = CryptoVec::from_slice(&buf[..n]);
                    let h = stderr_handle.clone();
                    let _ = stderr_runtime
                        .block_on(async move { h.extended_data(channel, 1, data).await });
                }
            }
        }
        let _ = reader_done_tx.send(());
    });

    // Exit waiter thread
    let handle_exit = handle;
    let runtime_exit = runtime;
    std::thread::spawn(move || {
        let status = child.wait().ok();
        #[cfg(target_os = "linux")]
        unregister_managed_child(child_pid);
        let code = status.and_then(|s| s.code()).unwrap_or(1).unsigned_abs();
        // Wait for both reader threads.
        let _ = reader_done_rx.recv_timeout(Duration::from_secs(2));
        let _ = reader_done_rx.recv_timeout(Duration::from_secs(1));
        drop(runtime_exit.spawn(async move {
            let _ = handle_exit.eof(channel).await;
            let _ = handle_exit.exit_status_request(channel, code).await;
            let _ = handle_exit.close(channel).await;
        }));
    });

    Ok(sender)
}

mod unsafe_pty {
    use super::{Command, RawFd, SandboxPolicy, Winsize, drop_privileges, sandbox, setsid};
    #[cfg(unix)]
    use std::os::unix::process::CommandExt;

    #[allow(unsafe_code)]
    pub fn set_winsize(fd: RawFd, winsize: Winsize) -> std::io::Result<()> {
        let rc = unsafe { libc::ioctl(fd, libc::TIOCSWINSZ, &winsize) };
        if rc != 0 {
            return Err(std::io::Error::last_os_error());
        }
        Ok(())
    }

    #[allow(unsafe_code)]
    fn set_controlling_tty(fd: RawFd) -> std::io::Result<()> {
        let rc = unsafe { libc::ioctl(fd, libc::TIOCSCTTY.into(), 0) };
        if rc != 0 {
            return Err(std::io::Error::last_os_error());
        }
        Ok(())
    }

    #[allow(unsafe_code)]
    pub fn install_pre_exec(
        cmd: &mut Command,
        policy: SandboxPolicy,
        workdir: Option<String>,
        slave_fd: RawFd,
        netns_fd: Option<RawFd>,
    ) {
        unsafe {
            cmd.pre_exec(move || {
                setsid().map_err(|err| std::io::Error::other(err.to_string()))?;
                set_controlling_tty(slave_fd)?;

                enter_netns_and_sandbox(netns_fd, &policy, workdir.as_deref())
            });
        }
    }

    /// Pre-exec hook for pipe-based (non-PTY) exec.
    ///
    /// Skips `setsid` and `TIOCSCTTY` since there is no controlling terminal.
    #[allow(unsafe_code)]
    pub fn install_pre_exec_no_pty(
        cmd: &mut Command,
        policy: SandboxPolicy,
        workdir: Option<String>,
        netns_fd: Option<RawFd>,
    ) {
        unsafe {
            cmd.pre_exec(move || enter_netns_and_sandbox(netns_fd, &policy, workdir.as_deref()));
        }
    }

    fn enter_netns_and_sandbox(
        netns_fd: Option<RawFd>,
        policy: &SandboxPolicy,
        workdir: Option<&str>,
    ) -> std::io::Result<()> {
        // Enter network namespace before dropping privileges.
        // This ensures SSH shell processes are isolated to the same
        // network namespace as the entrypoint, forcing all traffic
        // through the veth pair and CONNECT proxy.
        #[cfg(target_os = "linux")]
        if let Some(fd) = netns_fd {
            #[allow(unsafe_code)]
            let result = unsafe { libc::setns(fd, libc::CLONE_NEWNET) };
            if result != 0 {
                return Err(std::io::Error::last_os_error());
            }
        }

        #[cfg(not(target_os = "linux"))]
        let _ = netns_fd;

        // Drop privileges before applying sandbox restrictions.
        // initgroups/setgid/setuid need access to /etc/group and /etc/passwd
        // which may be blocked by Landlock.
        drop_privileges(policy).map_err(|err| std::io::Error::other(err.to_string()))?;
        sandbox::apply(policy, workdir).map_err(|err| std::io::Error::other(err.to_string()))?;
        Ok(())
    }
}

fn to_u16(value: u32) -> u16 {
    u16::try_from(value.min(u32::from(u16::MAX))).unwrap_or(u16::MAX)
}

/// Check whether a host string refers to a loopback address.
///
/// Covers all representations that resolve to loopback:
/// - `127.0.0.0/8` (the entire IPv4 loopback range, not just `127.0.0.1`)
/// - `localhost`
/// - `::1` and long-form IPv6 loopback (`0:0:0:0:0:0:0:1`)
/// - `::ffff:127.x.x.x` (IPv4-mapped IPv6 loopback)
/// - Bracketed forms like `[::1]`
fn is_loopback_host(host: &str) -> bool {
    // Strip brackets for IPv6 addresses like [::1]
    let host = host
        .strip_prefix('[')
        .and_then(|h| h.strip_suffix(']'))
        .unwrap_or(host);

    if host.eq_ignore_ascii_case("localhost") {
        return true;
    }

    match host.parse::<std::net::IpAddr>() {
        Ok(std::net::IpAddr::V4(v4)) => v4.is_loopback(), // covers all 127.x.x.x
        Ok(std::net::IpAddr::V6(v6)) => {
            if v6.is_loopback() {
                return true; // covers ::1 and long form
            }
            // Check IPv4-mapped IPv6 addresses like ::ffff:127.0.0.1
            if let Some(v4) = v6.to_ipv4_mapped() {
                return v4.is_loopback();
            }
            false
        }
        Err(_) => false,
    }
}

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

    /// Verify that dropping the input sender (the operation `channel_eof`
    /// performs) causes the stdin writer loop to exit and close the child's
    /// stdin pipe.  Without this, commands like `cat | tar xf -` used by
    /// `sync --up` hang forever waiting for EOF on stdin.
    #[test]
    fn dropping_input_sender_closes_child_stdin() {
        let (sender, receiver) = mpsc::channel::<Vec<u8>>();

        let mut child = Command::new("cat")
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .spawn()
            .expect("failed to spawn cat");

        let child_stdin = child.stdin.take().expect("stdin must be piped");

        // Replicate the stdin writer loop from spawn_pipe_exec.
        std::thread::spawn(move || {
            let mut stdin = child_stdin;
            while let Ok(bytes) = receiver.recv() {
                if stdin.write_all(&bytes).is_err() {
                    break;
                }
                let _ = stdin.flush();
            }
        });

        sender.send(b"hello".to_vec()).unwrap();

        // Simulate what channel_eof does: drop the sender.
        drop(sender);

        // cat should see EOF on stdin and exit.  Use a timeout so the test
        // fails fast instead of hanging if the mechanism is broken.
        let (done_tx, done_rx) = mpsc::channel();
        std::thread::spawn(move || {
            let _ = done_tx.send(child.wait_with_output());
        });
        let output = done_rx
            .recv_timeout(Duration::from_secs(5))
            .expect("cat hung for 5s — stdin was not closed (channel_eof bug)")
            .expect("failed to wait for cat");

        assert!(
            output.status.success(),
            "cat exited with {:?}",
            output.status
        );
        assert_eq!(output.stdout, b"hello");
    }

    /// Verify that the stdin writer delivers all buffered data before exiting
    /// when the sender is dropped.  This ensures channel_eof doesn't cause
    /// data loss — only signals "no more data after this".
    #[test]
    fn stdin_writer_delivers_buffered_data_before_eof() {
        let (sender, receiver) = mpsc::channel::<Vec<u8>>();

        let mut child = Command::new("wc")
            .arg("-c")
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .spawn()
            .expect("failed to spawn wc");

        let child_stdin = child.stdin.take().expect("stdin must be piped");

        std::thread::spawn(move || {
            let mut stdin = child_stdin;
            while let Ok(bytes) = receiver.recv() {
                if stdin.write_all(&bytes).is_err() {
                    break;
                }
                let _ = stdin.flush();
            }
        });

        // Send multiple chunks, then drop the sender.
        for _ in 0..100 {
            sender.send(vec![0u8; 1024]).unwrap();
        }
        drop(sender);

        let (done_tx, done_rx) = mpsc::channel();
        std::thread::spawn(move || {
            let _ = done_tx.send(child.wait_with_output());
        });
        let output = done_rx
            .recv_timeout(Duration::from_secs(5))
            .expect("wc hung for 5s — stdin was not closed")
            .expect("failed to wait for wc");

        let count: usize = String::from_utf8_lossy(&output.stdout)
            .trim()
            .parse()
            .expect("wc output was not a number");
        assert_eq!(
            count,
            100 * 1024,
            "expected all 100 KiB delivered before EOF"
        );
    }

    // -----------------------------------------------------------------------
    // verify_preface tests
    // -----------------------------------------------------------------------

    /// Build a valid NSSH1 preface line with the given parameters.
    fn build_preface(token: &str, secret: &str, nonce: &str, timestamp: i64) -> String {
        let payload = format!("{token}|{timestamp}|{nonce}");
        let signature = hmac_sha256(secret.as_bytes(), payload.as_bytes());
        format!("{PREFACE_MAGIC} {token} {timestamp} {nonce} {signature}")
    }

    fn fresh_nonce_cache() -> NonceCache {
        Arc::new(Mutex::new(HashMap::new()))
    }

    fn current_timestamp() -> i64 {
        i64::try_from(
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs(),
        )
        .unwrap()
    }

    #[test]
    fn verify_preface_accepts_valid_preface() {
        let secret = "test-secret-key";
        let nonce = "unique-nonce-1";
        let ts = current_timestamp();
        let line = build_preface("tok1", secret, nonce, ts);
        let cache = fresh_nonce_cache();

        assert!(verify_preface(&line, secret, 300, &cache).unwrap());
    }

    #[test]
    fn verify_preface_rejects_replayed_nonce() {
        let secret = "test-secret-key";
        let nonce = "replay-nonce";
        let ts = current_timestamp();
        let line = build_preface("tok1", secret, nonce, ts);
        let cache = fresh_nonce_cache();

        // First attempt should succeed.
        assert!(verify_preface(&line, secret, 300, &cache).unwrap());
        // Second attempt with the same nonce should be rejected.
        assert!(!verify_preface(&line, secret, 300, &cache).unwrap());
    }

    #[test]
    fn verify_preface_rejects_expired_timestamp() {
        let secret = "test-secret-key";
        let nonce = "expired-nonce";
        // Timestamp 600 seconds in the past, with a 300-second skew window.
        let ts = current_timestamp() - 600;
        let line = build_preface("tok1", secret, nonce, ts);
        let cache = fresh_nonce_cache();

        assert!(!verify_preface(&line, secret, 300, &cache).unwrap());
    }

    #[test]
    fn verify_preface_rejects_invalid_hmac() {
        let secret = "test-secret-key";
        let nonce = "hmac-nonce";
        let ts = current_timestamp();
        // Build with the correct secret, then verify with the wrong one.
        let line = build_preface("tok1", secret, nonce, ts);
        let cache = fresh_nonce_cache();

        assert!(!verify_preface(&line, "wrong-secret", 300, &cache).unwrap());
    }

    #[test]
    fn verify_preface_rejects_malformed_input() {
        let cache = fresh_nonce_cache();

        // Too few parts.
        assert!(!verify_preface("NSSH1 tok1 123", "s", 300, &cache).unwrap());
        // Wrong magic.
        assert!(!verify_preface("NSSH2 tok1 123 nonce sig", "s", 300, &cache).unwrap());
        // Empty string.
        assert!(!verify_preface("", "s", 300, &cache).unwrap());
    }

    #[test]
    fn verify_preface_distinct_nonces_both_accepted() {
        let secret = "test-secret-key";
        let ts = current_timestamp();
        let cache = fresh_nonce_cache();

        let line1 = build_preface("tok1", secret, "nonce-a", ts);
        let line2 = build_preface("tok1", secret, "nonce-b", ts);

        assert!(verify_preface(&line1, secret, 300, &cache).unwrap());
        assert!(verify_preface(&line2, secret, 300, &cache).unwrap());
    }

    #[test]
    fn apply_child_env_keeps_handshake_secret_out_of_ssh_children() {
        let mut cmd = Command::new("/usr/bin/env");
        cmd.env(SSH_HANDSHAKE_SECRET_ENV, "should-not-leak")
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::null());

        let provider_env = std::iter::once((
            "ANTHROPIC_API_KEY".to_string(),
            "openshell:resolve:env:ANTHROPIC_API_KEY".to_string(),
        ))
        .collect();

        apply_child_env(
            &mut cmd,
            "/sandbox",
            "sandbox",
            "dumb",
            None,
            None,
            &provider_env,
        );

        let output = cmd.output().expect("spawn env");
        let stdout = String::from_utf8(output.stdout).expect("utf8");

        assert!(!stdout.contains(SSH_HANDSHAKE_SECRET_ENV));
        assert!(stdout.contains("ANTHROPIC_API_KEY=openshell:resolve:env:ANTHROPIC_API_KEY"));
    }

    // -----------------------------------------------------------------------
    // SEC-007: is_loopback_host tests
    // -----------------------------------------------------------------------

    #[test]
    fn loopback_host_accepts_standard_ipv4() {
        assert!(is_loopback_host("127.0.0.1"));
    }

    #[test]
    fn loopback_host_accepts_full_ipv4_range() {
        assert!(is_loopback_host("127.0.0.2"));
        assert!(is_loopback_host("127.255.255.255"));
    }

    #[test]
    fn loopback_host_accepts_localhost() {
        assert!(is_loopback_host("localhost"));
        assert!(is_loopback_host("LOCALHOST"));
        assert!(is_loopback_host("Localhost"));
    }

    #[test]
    fn loopback_host_accepts_ipv6_loopback() {
        assert!(is_loopback_host("::1"));
        assert!(is_loopback_host("[::1]"));
        assert!(is_loopback_host("0:0:0:0:0:0:0:1"));
    }

    #[test]
    fn loopback_host_accepts_ipv4_mapped_ipv6() {
        assert!(is_loopback_host("::ffff:127.0.0.1"));
    }

    #[test]
    fn loopback_host_rejects_non_loopback() {
        assert!(!is_loopback_host("10.0.0.1"));
        assert!(!is_loopback_host("192.168.1.1"));
        assert!(!is_loopback_host("8.8.8.8"));
        assert!(!is_loopback_host("example.com"));
        assert!(!is_loopback_host("::ffff:10.0.0.1"));
    }

    #[test]
    fn loopback_host_rejects_empty_and_garbage() {
        assert!(!is_loopback_host(""));
        assert!(!is_loopback_host("not-an-ip"));
        assert!(!is_loopback_host("[]"));
    }

    // -----------------------------------------------------------------------
    // Per-channel PTY state tests (#543)
    // -----------------------------------------------------------------------

    #[test]
    fn set_winsize_applies_to_correct_pty() {
        // Verify that set_winsize applies to a specific PTY master FD,
        // which is the mechanism that per-channel tracking relies on.
        // With the old single-pty_master design, a window_change_request
        // for channel N would resize whatever PTY was stored last —
        // potentially belonging to a different channel.
        let pty_a = openpty(None, None).expect("openpty a");
        let pty_b = openpty(None, None).expect("openpty b");
        let master_a = std::fs::File::from(pty_a.master);
        let master_b = std::fs::File::from(pty_b.master);
        let fd_a = master_a.as_raw_fd();
        let fd_b = master_b.as_raw_fd();
        assert_ne!(fd_a, fd_b, "two PTYs must have distinct FDs");

        // Close the slave ends to avoid leaking FDs in the test.
        drop(std::fs::File::from(pty_a.slave));
        drop(std::fs::File::from(pty_b.slave));

        // Resize only PTY B.
        let winsize_b = Winsize {
            ws_row: 50,
            ws_col: 120,
            ws_xpixel: 0,
            ws_ypixel: 0,
        };
        unsafe_pty::set_winsize(fd_b, winsize_b).expect("set_winsize on PTY B");

        // Resize PTY A to a different size.
        let winsize_a = Winsize {
            ws_row: 24,
            ws_col: 80,
            ws_xpixel: 0,
            ws_ypixel: 0,
        };
        unsafe_pty::set_winsize(fd_a, winsize_a).expect("set_winsize on PTY A");

        // Read back sizes via ioctl to verify independence.
        let mut actual_a: libc::winsize = unsafe { std::mem::zeroed() };
        let mut actual_b: libc::winsize = unsafe { std::mem::zeroed() };
        #[allow(unsafe_code)]
        unsafe {
            libc::ioctl(fd_a, libc::TIOCGWINSZ, &mut actual_a);
            libc::ioctl(fd_b, libc::TIOCGWINSZ, &mut actual_b);
        }

        assert_eq!(actual_a.ws_row, 24, "PTY A should be 24 rows");
        assert_eq!(actual_a.ws_col, 80, "PTY A should be 80 cols");
        assert_eq!(actual_b.ws_row, 50, "PTY B should be 50 rows");
        assert_eq!(actual_b.ws_col, 120, "PTY B should be 120 cols");
    }

    #[test]
    fn channel_state_independent_input_senders() {
        // Verify that each channel gets its own input sender so that
        // data() and channel_eof() affect only the targeted channel.
        let (tx_a, rx_a) = mpsc::channel::<Vec<u8>>();
        let (tx_b, rx_b) = mpsc::channel::<Vec<u8>>();

        let mut state_a = ChannelState {
            input_sender: Some(tx_a),
            ..Default::default()
        };
        let state_b = ChannelState {
            input_sender: Some(tx_b),
            ..Default::default()
        };

        // Send data to channel A only.
        state_a
            .input_sender
            .as_ref()
            .unwrap()
            .send(b"hello-a".to_vec())
            .unwrap();
        // Send data to channel B only.
        state_b
            .input_sender
            .as_ref()
            .unwrap()
            .send(b"hello-b".to_vec())
            .unwrap();

        assert_eq!(rx_a.recv().unwrap(), b"hello-a");
        assert_eq!(rx_b.recv().unwrap(), b"hello-b");

        // EOF on channel A (drop sender) should not affect channel B.
        state_a.input_sender.take();
        assert!(
            rx_a.recv().is_err(),
            "channel A sender dropped, recv should fail"
        );

        // Channel B should still be functional.
        state_b
            .input_sender
            .as_ref()
            .unwrap()
            .send(b"still-alive".to_vec())
            .unwrap();
        assert_eq!(rx_b.recv().unwrap(), b"still-alive");
    }
}