gunnar-ssh 3.0.0

//! **A client's protocol misuse is not the server's ERROR; an operator's fire drill is**
//! (stinkies ssh-double-exec-error, 2026-09-30).
//!
//! A second `exec` on one SSH channel made the primary log `ERROR exec request for a
//! channel that was never opened` — any caller that may run one git command (anonymous on
//! a public repository too) could raise main's ERROR, and the watch's alert, at will. It is
//! the CLIENT's misuse: said at WARN once per connection, never at ERROR.
//!
//! The estate's watch still needs an ERROR it can raise on purpose to prove its alert path
//! end to end (estate-box 8A): `gunnar-fire-drill`, one exec verb held to server-wide Admin
//! like `gunnar-tunnel-address`, says ONE ERROR naming itself a drill. Anyone else is
//! refused before it says anything.

mod common;

use std::sync::{Arc, Mutex, OnceLock};

use common::*;
use gunnar_core::{Grant, Level};
use gunnar_ssh::seam::BannerTransport;
use gunnar_ssh::{Access, GunnarSsh, SshConfig, FIRE_DRILL_VERB};

/// What the capture holds: (level, message and fields) per event.
type Said = Arc<Mutex<Vec<(tracing::Level, String)>>>;

/// Every event this test binary's server says: (level, message and fields).
fn said() -> &'static Said {
    static SAID: OnceLock<Said> = OnceLock::new();
    SAID.get_or_init(|| {
        use tracing_subscriber::layer::SubscriberExt;
        let lines: Said = Arc::default();
        struct Capture(Said);
        struct Text(String);
        impl tracing::field::Visit for Text {
            fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
                self.0.push_str(&format!(" {}={:?}", field.name(), value));
            }
            fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
                self.0.push_str(&format!(" {}={}", field.name(), value));
            }
        }
        impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Capture {
            fn on_event(
                &self,
                event: &tracing::Event<'_>,
                _: tracing_subscriber::layer::Context<'_, S>,
            ) {
                let mut t = Text(String::new());
                event.record(&mut t);
                self.0
                    .lock()
                    .unwrap()
                    .push((*event.metadata().level(), t.0));
            }
        }
        let _ = tracing::subscriber::set_global_default(
            tracing_subscriber::registry().with(Capture(Arc::clone(&lines))),
        );
        lines
    })
}

fn lines_with(needle: &str) -> Vec<(tracing::Level, String)> {
    said()
        .lock()
        .unwrap()
        .iter()
        .filter(|(_, l)| l.contains(needle))
        .cloned()
        .collect()
}

async fn serve(plane: Plane) -> gunnar_ssh::BoundServer {
    let transport = Arc::new(BannerTransport::new());
    GunnarSsh::new(SshConfig::ephemeral_loopback(), plane.control, transport)
        .bind(mint_key())
        .await
        .expect("bind")
}

/// Three execs on one channel, on each of two connections: one WARN per connection
/// naming the misuse, no ERROR at all.
///
/// **Deterministic, never a sleep** (r4 2026-10-06: it failed 2-4 times in 6 at load
/// 20-30). The banner transport used to finish the first transfer in milliseconds and
/// close the channel, and a second exec that arrived after that close was dropped by
/// the SSH layer before gunnar saw it — no misuse, so no WARN: a scheduling race in the
/// TEST, not a server defect (a closed channel has nothing left to misuse). Now the
/// first transfer is HELD (`BannerTransport::holding`) and the test waits for its
/// banner — the channel is open and taken — then sends the two misuses and waits for
/// the server's own answer to each (`channel_failure`), which it sends right after it
/// has said the WARN.
#[tokio::test]
async fn a_second_exec_on_one_channel_is_a_warn_once_per_connection_never_an_error() {
    said();
    let key = mint_key();
    let plane = Plane::empty()
        .with_the_repo()
        .with_principal("reader")
        .with_key("reader", &key)
        .with_grant("reader", Access::Read);
    let (transport, hold) = BannerTransport::holding();
    let server = GunnarSsh::new(
        SshConfig::ephemeral_loopback(),
        plane.control,
        Arc::new(transport),
    )
    .bind(mint_key())
    .await
    .expect("bind");
    let command = format!("git-upload-pack '{}'", the_repo());
    let within = std::time::Duration::from_secs(30);
    for connection in 0..2 {
        let s = authenticate(server.local_addr(), &key, "git")
            .await
            .unwrap()
            .expect("key accepted");
        let mut ch = s.channel_open_session().await.expect("session");
        ch.exec(true, command.as_bytes()).await.expect("exec sent");
        // The first exec is accepted and its transfer is IN FLIGHT: its banner arrived,
        // and the held transport keeps the channel open and taken.
        let (mut accepted, mut banner) = (false, false);
        while !(accepted && banner) {
            match tokio::time::timeout(within, ch.wait())
                .await
                .expect("the first exec answered")
            {
                Some(russh::ChannelMsg::Success) => accepted = true,
                Some(russh::ChannelMsg::Data { .. }) => banner = true,
                Some(russh::ChannelMsg::Failure) => {
                    panic!("connection {connection}: the first exec was refused")
                }
                Some(_) => {}
                None => panic!("connection {connection}: the channel closed under the first exec"),
            }
        }
        // Two misuses on the same channel; the server answers each with a failure.
        for _ in 0..2 {
            ch.exec(true, command.as_bytes()).await.expect("exec sent");
        }
        let mut refused = 0;
        while refused < 2 {
            match tokio::time::timeout(within, ch.wait())
                .await
                .expect("the misuses answered")
            {
                Some(russh::ChannelMsg::Failure) => refused += 1,
                Some(russh::ChannelMsg::Success) => {
                    panic!("connection {connection}: a second exec on one channel was accepted")
                }
                Some(_) => {}
                None => panic!(
                    "connection {connection}: the channel closed before the misuses were answered"
                ),
            }
        }
        let _ = s
            .disconnect(russh::Disconnect::ByApplication, "", "en")
            .await;
    }
    hold.release();
    let misuse = lines_with("exec request for a channel");
    let errors: Vec<_> = misuse
        .iter()
        .filter(|(l, _)| *l == tracing::Level::ERROR)
        .collect();
    assert!(
        errors.is_empty(),
        "a client's misuse raised the server's ERROR: {errors:?}"
    );
    let warns: Vec<_> = misuse
        .iter()
        .filter(|(l, _)| *l == tracing::Level::WARN)
        .collect();
    assert_eq!(
        warns.len(),
        2,
        "one WARN per connection (2 connections × 2 misuses each): {misuse:?}"
    );
}

/// The drill: a server admin raises exactly ONE ERROR that names itself a drill; a
/// principal without server-wide Admin is refused and raises none.
#[tokio::test]
async fn a_server_admin_raises_one_fire_drill_error_and_nobody_else_can() {
    said();
    let admin = mint_key();
    let reader = mint_key();
    let plane = Plane::empty()
        .with_the_repo()
        .with_principal("admin")
        .with_key("admin", &admin)
        .with_principal("reader")
        .with_key("reader", &reader)
        .with_grant("reader", Access::Admin);
    plane
        .auth()
        .grant(Grant::server("admin", Level::Admin))
        .expect("server admin");
    let server = serve(plane).await;

    let r = authenticate(server.local_addr(), &reader, "git")
        .await
        .unwrap()
        .expect("reader");
    match exec(&r, FIRE_DRILL_VERB).await.unwrap() {
        ExecOutcome::Accepted {
            stdout,
            exit_status,
        } => {
            assert_eq!(
                exit_status,
                Some(gunnar_ssh::server::EXIT_DENIED),
                "{stdout}"
            );
            assert!(stdout.contains("server-wide Admin"), "{stdout}");
        }
        ExecOutcome::Refused => {}
    }
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let drills = || {
        lines_with("fire drill")
            .into_iter()
            .filter(|(l, _)| *l == tracing::Level::ERROR)
            .collect::<Vec<_>>()
    };
    assert!(
        drills().is_empty(),
        "a repository admin raised the drill: {:?}",
        drills()
    );

    let s = authenticate(server.local_addr(), &admin, "git")
        .await
        .unwrap()
        .expect("admin");
    match exec(&s, FIRE_DRILL_VERB).await.unwrap() {
        ExecOutcome::Accepted {
            stdout,
            exit_status,
        } => {
            assert_eq!(exit_status, Some(0), "{stdout}");
            assert!(stdout.contains("fire drill"), "{stdout}");
        }
        other => panic!("a server admin must reach the drill: {other:?}"),
    }
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    let d = drills();
    assert_eq!(d.len(), 1, "exactly one ERROR for one drill: {d:?}");
    assert!(d[0].1.contains("admin"), "the drill names who asked: {d:?}");
}