gunnar-ssh 2.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:?}");
}