gunnar-server 3.0.0

//! Outbound webhooks, end to end against a real [`ServerStores`] and its
//! real minted signing key — the reds the module unit tests prove in miniature,
//! proven again through the whole server-side slot (measured, not asserted).
//!
//! * **a delivery is signed and verifies** against the key the store actually
//!   minted, off a real event on a real bus. Red: the wrong key fails.
//! * **config survives a reboot**: an endpoint added through the slot is still
//!   listed after the slot (and the delivery) is torn down and rebuilt over the
//!   same data root — a real restart. Red: a slot over an empty root lists none.
//! * **the signing key is stable across a reboot**, so a receiver's pin does
//!   not have to change every time the server restarts.

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

use std::sync::Arc;
use std::time::Duration;

use gunnar_grpc::EventBus;
use gunnar_proto::{event, RepoPath};
use gunnar_server::stores::ServerStores;
use gunnar_server::webhook::WebhookSlot;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

fn stores(dir: &std::path::Path) -> Arc<ServerStores> {
    Arc::new(ServerStores::open(dir).expect("stores"))
}

/// A one-shot local HTTP sink returning the request body and signature headers.
async fn spawn_sink() -> (
    String,
    tokio::sync::oneshot::Receiver<(Vec<u8>, String, String)>,
) {
    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];
        let mut header_end = None;
        let mut clen = 0usize;
        loop {
            if header_end.is_none() {
                if let Some(p) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                    header_end = Some(p + 4);
                    let head = String::from_utf8_lossy(&buf[..p]).to_lowercase();
                    for l in head.lines() {
                        if let Some(v) = l.strip_prefix("content-length:") {
                            clen = v.trim().parse().unwrap_or(0);
                        }
                    }
                }
            }
            if let Some(s) = header_end {
                if buf.len() >= s + clen {
                    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 + clen.min(buf.len() - start)].to_vec();
        let head = String::from_utf8_lossy(&buf[..start]).to_string();
        let get = |name: &str| {
            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 sig = get(gunnar_core::webhook::SIGNATURE_HEADER);
        let fp = get(gunnar_core::webhook::KEY_FINGERPRINT_HEADER);
        let _ = socket
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
            .await;
        let _ = tx.send((body, sig, fp));
    });
    (format!("http://{addr}/hook"), rx)
}

fn add(slot: &Arc<WebhookSlot>, name: &str, url: &str) {
    slot.add(gunnar_server::webhook::Endpoint {
        name: name.into(),
        url: url.into(),
        kinds: Vec::new(),
        namespace: String::new(),
        repo: None,
        secret: None,
        added_at_unix_ms: 0,
    })
    .expect("add endpoint");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_delivery_through_the_server_slot_is_signed_and_verifies() {
    let dir = tempfile::tempdir().unwrap();
    let stores = stores(dir.path());
    let bus = Arc::new(EventBus::new(64));
    let slot = WebhookSlot::new(
        Arc::clone(&stores),
        Arc::clone(&bus),
        tokio::runtime::Handle::current(),
    );
    let (url, rx) = spawn_sink().await;
    add(&slot, "ci", &url);
    tokio::time::sleep(Duration::from_millis(50)).await;

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

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

    // The key the RECEIVER pins is exactly the one the control plane publishes.
    let published = slot.signing_public_key().expect("public key");
    let pubkey = gunnar_core::ssh_key::PublicKey::from_openssh(&published).expect("parse pub");

    // GREEN: the received body verifies against the published key.
    gunnar_core::webhook::verify_delivery(&pubkey, &body, &sig, &fp)
        .expect("the received delivery must verify against the server's published key");

    // RED: it must NOT verify against an unrelated key.
    let other =
        gunnar_core::ssh_key::PrivateKey::from_openssh(gunnar_core::mint_ed25519().unwrap())
            .unwrap();
    let err = gunnar_core::webhook::verify_delivery(
        other.public_key(),
        &body,
        &sig,
        &other
            .public_key()
            .fingerprint(gunnar_core::ssh_key::HashAlg::Sha256)
            .to_string(),
    )
    .expect_err("a delivery signed by the server must not verify against a stranger's key");
    assert_eq!(err, gunnar_core::webhook::DeliveryProblem::NotVerified);

    slot.stop();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn config_and_signing_key_survive_a_reboot() {
    let dir = tempfile::tempdir().unwrap();

    // RED: a fresh slot over an untouched root lists no endpoint.
    {
        let stores = stores(dir.path());
        let bus = Arc::new(EventBus::new(16));
        let slot = WebhookSlot::new(stores, bus, tokio::runtime::Handle::current());
        assert!(
            slot.list_response().unwrap().endpoints.is_empty(),
            "an untouched data root has no webhooks"
        );
    }

    // Boot 1: add an endpoint, capture the signing key, then tear the slot down.
    let key_before;
    {
        let stores = stores(dir.path());
        let bus = Arc::new(EventBus::new(16));
        let slot = WebhookSlot::new(stores, bus, tokio::runtime::Handle::current());
        add(&slot, "ci", "https://ci.example/hook");
        key_before = slot.signing_public_key().expect("key");
        slot.stop();
    }

    // Boot 2: a brand-new slot over the SAME root — the reboot — still lists the
    // endpoint, and the signing key a receiver pinned is unchanged.
    {
        let stores = stores(dir.path());
        let bus = Arc::new(EventBus::new(16));
        let slot = WebhookSlot::new(stores, bus, tokio::runtime::Handle::current());
        let listed = slot.list_response().expect("list");
        assert_eq!(
            listed.endpoints.len(),
            1,
            "the endpoint survived the reboot"
        );
        assert_eq!(listed.endpoints[0].name, "ci");
        assert_eq!(listed.endpoints[0].url, "https://ci.example/hook");
        assert_eq!(
            slot.signing_public_key().expect("key"),
            key_before,
            "the signing key is stable across a reboot, so a receiver's pin still holds"
        );
        slot.stop();
    }
}