gunnar-server 2.0.0

//! Measure, never assert, for outbound webhooks: the reds each proven before the green.
//!
//! Three properties, three failures seen first:
//!
//! * **a delivery happens and is signed** — a real event reaches a local sink,
//!   and the bytes the sink received VERIFY against the server's key. Red: the
//!   same bytes with the wrong key, and a tampered body, FAIL the verify.
//! * **a dead endpoint does not wedge the bus** — a live endpoint keeps
//!   receiving while a second endpoint is a black hole, and the enqueue path is
//!   bounded and non-blocking (the shape a naive synchronous deliver-in-the-
//!   subscribe-path lacks). Red: the naive shape would block; this one does not.
//! * **config survives reboot** — saved and reloaded, the endpoint is still
//!   there. Red: the in-memory-only config reloads empty.

use super::*;
use std::io::Write as _;
use std::sync::atomic::Ordering;
use std::time::Duration;

use tokio::io::{AsyncReadExt, AsyncWriteExt};

fn a_key() -> Arc<ssh_key::PrivateKey> {
    Arc::new(
        ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().expect("mint"))
            .expect("parse"),
    )
}

/// What a sink captured off one delivery.
#[derive(Debug, Clone)]
struct Captured {
    body: Vec<u8>,
    signature_header: String,
    fingerprint_header: String,
    namespace_header: String,
    event_header: String,
}

/// A one-shot local HTTP sink. Accepts one connection, reads the request,
/// answers 200, and reports what it received.
async fn spawn_sink() -> (String, tokio::sync::oneshot::Receiver<Captured>) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (tx, rx) = tokio::sync::oneshot::channel();
    tokio::spawn(async move {
        let (mut socket, _) = listener.accept().await.unwrap();
        let mut buf = Vec::new();
        let mut tmp = [0u8; 4096];
        // Read until we have the whole request (headers + Content-Length body).
        let mut header_end = None;
        let mut content_len = 0usize;
        loop {
            if header_end.is_none() {
                if let Some(pos) = find_subslice(&buf, b"\r\n\r\n") {
                    header_end = Some(pos + 4);
                    let head = String::from_utf8_lossy(&buf[..pos]).to_lowercase();
                    for line in head.lines() {
                        if let Some(v) = line.strip_prefix("content-length:") {
                            content_len = v.trim().parse().unwrap_or(0);
                        }
                    }
                }
            }
            if let Some(start) = header_end {
                if buf.len() >= start + content_len {
                    break;
                }
            }
            let n = socket.read(&mut tmp).await.unwrap();
            if n == 0 {
                break;
            }
            buf.extend_from_slice(&tmp[..n]);
        }
        let start = header_end.unwrap_or(buf.len());
        let body = buf[start..start + content_len.min(buf.len() - start)].to_vec();
        // Case-preserving header extraction from the original bytes.
        let head = String::from_utf8_lossy(&buf[..start]).to_string();
        let get = |name: &str| -> String {
            head.lines()
                .find(|l| l.to_lowercase().starts_with(&format!("{}:", name.to_lowercase())))
                .and_then(|l| l.split_once(':'))
                .map(|(_, v)| v.trim().to_owned())
                .unwrap_or_default()
        };
        let captured = Captured {
            body,
            signature_header: get(gunnar_core::webhook::SIGNATURE_HEADER),
            fingerprint_header: get(gunnar_core::webhook::KEY_FINGERPRINT_HEADER),
            namespace_header: get(gunnar_core::webhook::NAMESPACE_HEADER),
            event_header: get(gunnar_core::webhook::EVENT_HEADER),
        };
        let _ = socket
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
            .await;
        let _ = tx.send(captured);
    });
    (format!("http://{addr}/hook"), rx)
}

fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack.windows(needle.len()).position(|w| w == needle)
}

fn repo_deleted_event(seq: u64) -> Arc<Event> {
    Arc::new(Event::new(
        seq,
        1,
        event::Payload::RepoDeleted(gunnar_proto::RepoDeleted {
            repo: Some(RepoPath {
                namespace: "alice".into(),
                repo: "edda".into(),
            }),
            principal_id: "p-1".into(),
        }),
    ))
}

fn endpoint(name: &str, url: &str) -> Endpoint {
    Endpoint {
        name: name.into(),
        url: url.into(),
        kinds: Vec::new(),
        namespace: String::new(),
        repo: None,
        secret: None,
        added_at_unix_ms: 0,
    }
}

// ── 1. delivery happens and is signed ───────────────────────────────────────

#[tokio::test]
async fn a_real_event_reaches_the_sink_and_the_received_body_verifies() {
    let (url, rx) = spawn_sink().await;
    let key = a_key();
    let bus = Arc::new(EventBus::new(64));
    let config = WebhookConfig {
        endpoints: vec![endpoint("ci", &url)],
    };
    let delivery = WebhookDelivery::new(Arc::clone(&bus), Arc::clone(&key), config);
    let _tasks = delivery.spawn();
    // Let the subscriber attach before the event is published, so the live tail
    // carries it (the bus is live-only for an in-process subscriber).
    tokio::time::sleep(Duration::from_millis(50)).await;

    bus.publish(event::Payload::RepoDeleted(gunnar_proto::RepoDeleted {
        repo: Some(RepoPath {
            namespace: "alice".into(),
            repo: "edda".into(),
        }),
        principal_id: "p-1".into(),
    }));

    let captured = tokio::time::timeout(Duration::from_secs(5), rx)
        .await
        .expect("the sink must receive a delivery")
        .expect("sink channel");

    // The delivery is what we expected.
    assert_eq!(captured.event_header, "repo.deleted");
    assert_eq!(
        captured.namespace_header,
        gunnar_core::webhook::WEBHOOK_NAMESPACE
    );
    let body: serde_json::Value = serde_json::from_slice(&captured.body).expect("body is json");
    assert_eq!(body["kind"], "repo.deleted");
    assert_eq!(body["repo"], "edda");

    // GREEN: the RECEIVED bytes verify against the server's key.
    gunnar_core::webhook::verify_delivery(
        key.public_key(),
        &captured.body,
        &captured.signature_header,
        &captured.fingerprint_header,
    )
    .expect("the received body must verify against the server's published key");

    // RED 1: the same received bytes must FAIL against a different key.
    let attacker = a_key();
    let err = gunnar_core::webhook::verify_delivery(
        attacker.public_key(),
        &captured.body,
        &captured.signature_header,
        &attacker
            .public_key()
            .fingerprint(ssh_key::HashAlg::Sha256)
            .to_string(),
    )
    .expect_err("a delivery signed by gunnar must not verify against an attacker's key");
    assert_eq!(err, gunnar_core::webhook::DeliveryProblem::NotVerified);

    // RED 2: a single flipped byte in the received body must FAIL.
    let mut tampered = captured.body.clone();
    tampered[10] ^= 0x01;
    let err = gunnar_core::webhook::verify_delivery(
        key.public_key(),
        &tampered,
        &captured.signature_header,
        &captured.fingerprint_header,
    )
    .expect_err("a tampered body must not verify");
    assert_eq!(err, gunnar_core::webhook::DeliveryProblem::NotVerified);
}

// ── 2. a dead endpoint does not wedge the bus ───────────────────────────────

#[tokio::test]
async fn a_dead_endpoint_does_not_stop_a_live_one() {
    let (url, rx) = spawn_sink().await;
    let key = a_key();
    let bus = Arc::new(EventBus::new(64));
    // "dead" is a refusing address (port 1); its worker fails fast and backs
    // off, and — the point — the subscribe task never awaits it.
    let config = WebhookConfig {
        endpoints: vec![
            endpoint("dead", "http://127.0.0.1:1/hook"),
            endpoint("live", &url),
        ],
    };
    let delivery = WebhookDelivery::new(Arc::clone(&bus), key, config);
    let _tasks = delivery.spawn();
    tokio::time::sleep(Duration::from_millis(50)).await;

    bus.publish(event::Payload::RepoDeleted(gunnar_proto::RepoDeleted {
        repo: Some(RepoPath {
            namespace: "alice".into(),
            repo: "edda".into(),
        }),
        principal_id: "p-1".into(),
    }));

    // The live endpoint still receives, promptly, even though a sibling is dead.
    tokio::time::timeout(Duration::from_secs(5), rx)
        .await
        .expect("the live endpoint must deliver despite the dead sibling")
        .expect("sink channel");
}

#[tokio::test]
async fn enqueue_is_bounded_and_never_blocks_when_nothing_drains() {
    // The property a naive synchronous deliver-in-the-subscribe-path lacks: the
    // enqueue path is O(1), non-blocking, and bounded. Nothing drains this
    // runner (no worker spawned), yet flooding it far past capacity neither
    // blocks nor grows memory — it drops-with-a-gap.
    let runner = Runner::new(endpoint("x", "http://127.0.0.1:1/hook"));
    let flood = DEFAULT_QUEUE_CAPACITY + 500;
    let start = std::time::Instant::now();
    for seq in 0..flood as u64 {
        runner.enqueue(repo_deleted_event(seq));
    }
    assert!(
        start.elapsed() < Duration::from_secs(2),
        "enqueue must not block — a synchronous deliver would have here"
    );
    assert_eq!(
        runner.queue_len(),
        DEFAULT_QUEUE_CAPACITY,
        "the queue is bounded: the overflow is dropped, not buffered"
    );
    assert_eq!(
        runner.counters.dropped.load(Ordering::Relaxed),
        500,
        "every dropped event is counted, so the receiver can see it fell behind"
    );
}

// ── 3. config survives reboot ───────────────────────────────────────────────

#[test]
fn config_survives_a_reboot_but_an_in_memory_one_does_not() {
    let dir = tempdir();

    // RED: an in-memory-only config — never written — reloads as empty. This is
    // the store the durable one is being distinguished from.
    let ephemeral = WebhookConfig {
        endpoints: vec![endpoint("ci", "https://ci.example/hook")],
    };
    assert_eq!(ephemeral.endpoints.len(), 1);
    let reloaded_without_saving = WebhookConfig::load(&dir).expect("load");
    assert!(
        reloaded_without_saving.endpoints.is_empty(),
        "a config that was only ever in memory is gone after a reboot"
    );

    // GREEN: saved to webhooks.json, it survives.
    ephemeral.save(&dir).expect("save");
    assert!(
        dir.join(WEBHOOK_CONFIG).is_file(),
        "the config must be a real file on the data path"
    );
    let reloaded = WebhookConfig::load(&dir).expect("load after save");
    assert_eq!(reloaded.endpoints.len(), 1);
    assert_eq!(reloaded.endpoints[0].name, "ci");
    assert_eq!(reloaded.endpoints[0].url, "https://ci.example/hook");
}

#[test]
fn an_empty_config_file_loads_as_no_webhooks_not_a_boot_failure() {
    // A power cut during save can leave a 0-byte file. It must come up
    // "no webhooks" and serving, never fail the parse (which as PID 1 panics the
    // kernel). A NON-empty malformed file, by contrast, is an operator typo and
    // stays a hard error.
    let dir = tempdir();
    let mut f = std::fs::File::create(dir.join(WEBHOOK_CONFIG)).unwrap();
    f.write_all(b"   \n").unwrap();
    drop(f);
    assert!(WebhookConfig::load(&dir).expect("empty loads").endpoints.is_empty());

    std::fs::write(dir.join(WEBHOOK_CONFIG), b"{ not json").unwrap();
    assert!(
        WebhookConfig::load(&dir).is_err(),
        "a non-empty malformed config is an operator typo and must be refused"
    );
}

#[test]
fn a_kinds_typo_is_refused_at_add_time() {
    assert!(kind_from_name("ref.updated").is_some());
    assert!(
        kind_from_name("ref.updatd").is_none(),
        "a typo'd kind name selects nothing and must be catchable, not silently deliver nothing"
    );
}

/// A throwaway directory under the scratch cargo target, removed on drop.
struct TempDir(std::path::PathBuf);
impl std::ops::Deref for TempDir {
    type Target = std::path::Path;
    fn deref(&self) -> &std::path::Path {
        &self.0
    }
}
impl AsRef<std::path::Path> for TempDir {
    fn as_ref(&self) -> &std::path::Path {
        &self.0
    }
}
impl Drop for TempDir {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.0);
    }
}
fn tempdir() -> TempDir {
    let base = std::env::temp_dir().join(format!(
        "gunnar-webhook-test-{}-{}",
        std::process::id(),
        NEXT.fetch_add(1, Ordering::Relaxed)
    ));
    std::fs::create_dir_all(&base).unwrap();
    TempDir(base)
}
static NEXT: AtomicU64 = AtomicU64::new(0);