forjar 1.23.0

//! Integration tests for [`crate::core::webhook_server`] over real sockets.
//!
//! These drive the server the way a sender does — connect, write bytes, read the
//! response — because that is the only layer where the framing, the MAC over wire
//! bytes, the status codes and the concurrency behaviour are all simultaneously
//! true or false. The old suite's server tests wrote a small body in one
//! `write_all` and asserted `resp.contains("200 OK")`, which could not observe any
//! of it.

use super::webhook_server::run_webhook_server;
use super::webhook_sig::{canonical_payload, compute_hmac_hex, unix_now};
use super::webhook_source::WebhookConfig;
use crate::core::types::InfraEvent;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver};
use std::sync::Arc;
use std::time::{Duration, Instant};

const SECRET: &str = "test-secret";

/// Serialises every test that binds a real port.
///
/// `free_port` asks the OS for an ephemeral port and then DROPS the listener, so
/// the number it returns is reserved only until it returns. Two tests running in
/// parallel can be handed the same port, and the loser fails with a bind error
/// that has nothing to do with what it was testing. Serialising makes that
/// window unreachable rather than merely narrow.
///
/// The readiness probe below already CONNECTS rather than binding, which is the
/// other half of this race and the half that mattered most — a probe that binds
/// competes with the server it is waiting for, holds the port when it wins, and
/// starves the server until the timeout (forjar#276).
static PORT_GUARD: std::sync::Mutex<()> = std::sync::Mutex::new(());

fn free_port() -> u16 {
    let l = TcpListener::bind("127.0.0.1:0").unwrap();
    l.local_addr().unwrap().port()
}

struct Server {
    port: u16,
    rx: Receiver<InfraEvent>,
    shutdown: Arc<AtomicBool>,
}

impl Drop for Server {
    fn drop(&mut self) {
        self.shutdown.store(true, Ordering::Relaxed);
    }
}

fn start(config: WebhookConfig) -> Server {
    // Held until the returned Server is dropped, i.e. for the whole test.
    let _guard = PORT_GUARD.lock().unwrap_or_else(|e| e.into_inner());
    let port = config.port;
    let (tx, rx) = mpsc::channel();
    let shutdown = Arc::new(AtomicBool::new(false));
    let s2 = Arc::clone(&shutdown);
    std::thread::spawn(move || {
        let _ = run_webhook_server(&config, tx, s2);
    });
    // Wait for the listener rather than sleeping a fixed amount.
    let deadline = Instant::now() + Duration::from_secs(5);
    while Instant::now() < deadline {
        if TcpStream::connect(("127.0.0.1", port)).is_ok() {
            break;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    Server { port, rx, shutdown }
}

fn config_for(port: u16) -> WebhookConfig {
    WebhookConfig {
        port,
        secret: Some(SECRET.to_string()),
        ..WebhookConfig::default()
    }
}

/// Sign `body` for `path` and return the header value.
fn sign(path: &str, body: &[u8], t: i64) -> String {
    let signed = canonical_payload(t, "POST", path, body);
    format!("t={t},v1={}", compute_hmac_hex(SECRET.as_bytes(), &signed))
}

/// Send raw bytes, optionally in two writes separated by a pause.
fn send(port: u16, first: &[u8], second: Option<&[u8]>) -> String {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.write_all(first).unwrap();
    s.flush().unwrap();
    if let Some(rest) = second {
        std::thread::sleep(Duration::from_millis(150));
        s.write_all(rest).unwrap();
        s.flush().unwrap();
    }
    s.set_read_timeout(Some(Duration::from_secs(5))).ok();
    let mut out = Vec::new();
    let _ = s.read_to_end(&mut out);
    String::from_utf8_lossy(&out).to_string()
}

fn request_bytes(path: &str, body: &[u8], sig: Option<&str>) -> (Vec<u8>, Vec<u8>) {
    let mut head = format!(
        "POST {path} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\n",
        body.len()
    );
    if let Some(s) = sig {
        head.push_str(&format!("X-Forjar-Signature: {s}\r\n"));
    }
    head.push_str("\r\n");
    (head.into_bytes(), body.to_vec())
}

// ── Happy path, including split delivery ─────────────────────────────────────

#[test]
fn signed_delivery_is_accepted() {
    let srv = start(config_for(free_port()));
    let body = br#"{"action":"deploy"}"#;
    let sig = sign("/webhook", body, unix_now());
    let (head, b) = request_bytes("/webhook", body, Some(&sig));
    let resp = send(srv.port, &[head, b].concat(), None);

    assert!(resp.starts_with("HTTP/1.1 200 OK"), "resp: {resp}");
    let ev = srv.rx.recv_timeout(Duration::from_secs(2)).expect("event");
    assert_eq!(ev.payload.get("action").map(String::as_str), Some("deploy"));
}

/// A correctly-signed delivery whose body arrives in a SECOND segment. Measured
/// against the old server: 400, and the event never reached the channel.
#[test]
fn signed_delivery_split_across_segments_is_accepted() {
    let srv = start(config_for(free_port()));
    let body = br#"{"action":"deploy"}"#;
    let sig = sign("/webhook", body, unix_now());
    let (head, b) = request_bytes("/webhook", body, Some(&sig));
    let resp = send(srv.port, &head, Some(&b));

    assert!(resp.starts_with("HTTP/1.1 200 OK"), "resp: {resp}");
    assert!(
        srv.rx.recv_timeout(Duration::from_secs(2)).is_ok(),
        "event must reach the channel on a split delivery"
    );
}

/// The MAC must cover the exact octets, including bytes that are not valid UTF-8.
/// The old path ran the body through `from_utf8_lossy` first, so this could never
/// verify.
#[test]
fn non_utf8_body_signed_over_wire_bytes_is_accepted() {
    let mut cfg = config_for(free_port());
    cfg.allowed_paths = vec!["/webhook".into()];
    let srv = start(cfg);
    // Valid JSON except for a raw 0xFF inside a string — deliberately non-UTF-8.
    let body: Vec<u8> = b"{\"k\":\"a\xffb\"}".to_vec();
    assert!(std::str::from_utf8(&body).is_err());
    let sig = sign("/webhook", &body, unix_now());
    let (head, b) = request_bytes("/webhook", &body, Some(&sig));
    let resp = send(srv.port, &[head, b].concat(), None);

    // Signature verifies over the raw bytes, so this is NOT 401. The body then
    // fails UTF-8/JSON parsing, which is a 400 — the point is that authentication
    // succeeded on the exact octets.
    assert!(
        !resp.contains(" 401 "),
        "signature must verify over raw bytes; got {resp}"
    );
    assert!(
        resp.contains(" 400 "),
        "expected a body-parse 400; got {resp}"
    );
}

// ── Authentication ───────────────────────────────────────────────────────────

#[test]
fn unsigned_request_is_401_when_a_secret_is_configured() {
    let srv = start(config_for(free_port()));
    let (head, b) = request_bytes("/webhook", b"{}", None);
    let resp = send(srv.port, &[head, b].concat(), None);
    assert!(resp.starts_with("HTTP/1.1 401"), "resp: {resp}");
    assert!(resp.contains("signature_missing"));
}

/// A signature minted for one allowed path must not verify at another. With the
/// old body-only MAC it did — and `_path` reads like an authorization input.
#[test]
fn signature_for_one_path_is_rejected_at_another() {
    let mut cfg = config_for(free_port());
    cfg.allowed_paths = vec!["/hooks/deploy".into(), "/hooks/destroy".into()];
    let srv = start(cfg);

    let body = br#"{"go":true}"#;
    let sig = sign("/hooks/deploy", body, unix_now());
    let (head, b) = request_bytes("/hooks/destroy", body, Some(&sig));
    let resp = send(srv.port, &[head, b].concat(), None);

    assert!(resp.starts_with("HTTP/1.1 401"), "resp: {resp}");
    assert!(resp.contains("signature_invalid"));
}

/// An old capture must not replay, even with a valid MAC.
#[test]
fn stale_timestamp_is_rejected() {
    let srv = start(config_for(free_port()));
    let body = br#"{"a":1}"#;
    let sig = sign("/webhook", body, unix_now() - 3600);
    let (head, b) = request_bytes("/webhook", body, Some(&sig));
    let resp = send(srv.port, &[head, b].concat(), None);
    assert!(resp.starts_with("HTTP/1.1 401"), "resp: {resp}");
    assert!(resp.contains("signature_stale"));
}

// ── Idempotency ──────────────────────────────────────────────────────────────

/// The identical signed request twice: both answered 200, exactly ONE event.
/// A retrying sender did nothing wrong, so a 4xx would only make it retry harder.
#[test]
fn duplicate_delivery_yields_exactly_one_event() {
    let srv = start(config_for(free_port()));
    let body = br#"{"action":"once"}"#;
    let sig = sign("/webhook", body, unix_now());
    let (head, b) = request_bytes("/webhook", body, Some(&sig));
    let raw = [head, b].concat();

    let first = send(srv.port, &raw, None);
    let second = send(srv.port, &raw, None);

    assert!(first.starts_with("HTTP/1.1 200"), "first: {first}");
    assert!(second.starts_with("HTTP/1.1 200"), "second: {second}");
    assert!(second.contains("duplicate_ignored"), "second: {second}");

    assert!(srv.rx.recv_timeout(Duration::from_secs(2)).is_ok());
    assert!(
        srv.rx.recv_timeout(Duration::from_millis(300)).is_err(),
        "a duplicate must not produce a second event"
    );
}

// ── Method and path ──────────────────────────────────────────────────────────

#[test]
fn get_is_405_with_allow_header() {
    let srv = start(config_for(free_port()));
    let raw = b"GET /webhook HTTP/1.1\r\nHost: x\r\n\r\n";
    let resp = send(srv.port, raw, None);
    assert!(resp.starts_with("HTTP/1.1 405"), "resp: {resp}");
    assert!(resp.contains("Allow: POST"), "resp: {resp}");
}

#[test]
fn unknown_path_is_404() {
    let srv = start(config_for(free_port()));
    let body = b"{}";
    let sig = sign("/nope", body, unix_now());
    let (head, b) = request_bytes("/nope", body, Some(&sig));
    let resp = send(srv.port, &[head, b].concat(), None);
    assert!(resp.starts_with("HTTP/1.1 404"), "resp: {resp}");
    // Must NOT echo the requested path back.
    assert!(!resp.contains("/nope"), "response reflected input: {resp}");
}

/// A query string on an allowed path is still that path. Exact matching on the
/// raw target made any query string a rejection.
#[test]
fn query_string_does_not_break_path_matching() {
    let srv = start(config_for(free_port()));
    let body = br#"{"a":1}"#;
    let sig = sign("/webhook?src=gh", body, unix_now());
    let (head, b) = request_bytes("/webhook?src=gh", body, Some(&sig));
    let resp = send(srv.port, &[head, b].concat(), None);
    assert!(resp.starts_with("HTTP/1.1 200"), "resp: {resp}");
}

// ── Pre-authentication denial of service ─────────────────────────────────────

/// Two sockets that connect and send NOTHING must not delay a real delivery.
/// Measured against the old inline handler: one idle socket delayed a legitimate
/// signed request by 5383ms, entirely upstream of signature checking.
#[test]
fn idle_connections_do_not_delay_a_real_delivery() {
    let srv = start(config_for(free_port()));

    let mut idle = Vec::new();
    for _ in 0..2 {
        idle.push(TcpStream::connect(("127.0.0.1", srv.port)).unwrap());
    }

    let body = br#"{"action":"urgent"}"#;
    let sig = sign("/webhook", body, unix_now());
    let (head, b) = request_bytes("/webhook", body, Some(&sig));

    let started = Instant::now();
    let resp = send(srv.port, &[head, b].concat(), None);
    let elapsed = started.elapsed();

    assert!(resp.starts_with("HTTP/1.1 200"), "resp: {resp}");
    assert!(
        elapsed < Duration::from_secs(2),
        "idle sockets delayed a real delivery by {elapsed:?}"
    );
    drop(idle);
}

// ── Fail closed at startup ───────────────────────────────────────────────────

#[test]
fn refuses_to_start_without_a_secret() {
    let cfg = WebhookConfig {
        port: free_port(),
        secret: None,
        ..WebhookConfig::default()
    };
    let (tx, _rx) = mpsc::channel();
    let err = run_webhook_server(&cfg, tx, Arc::new(AtomicBool::new(true)))
        .expect_err("must refuse to start unauthenticated");
    assert!(err.contains("allow_unauthenticated"), "err: {err}");
}

#[test]
fn starts_without_a_secret_when_explicitly_allowed() {
    let cfg = WebhookConfig {
        port: free_port(),
        secret: None,
        allow_unauthenticated: true,
        ..WebhookConfig::default()
    };
    assert!(cfg.validate_startup().is_ok());
}

/// An empty allow-list denies everything, so starting with one is a
/// misconfiguration rather than a wide-open endpoint.
#[test]
fn refuses_to_start_with_an_empty_allow_list() {
    let cfg = WebhookConfig {
        allowed_paths: vec![],
        secret: Some(SECRET.into()),
        ..WebhookConfig::default()
    };
    let err = cfg
        .validate_startup()
        .expect_err("empty allow-list must fail");
    assert!(err.contains("allowed_paths"), "err: {err}");
}

/// Signatures authenticate a sender; they do not encrypt. A non-loopback bind
/// needs the operator to say TLS is terminated in front.
#[test]
fn refuses_a_non_loopback_bind_without_upstream_tls() {
    let cfg = WebhookConfig {
        bind: "0.0.0.0".into(),
        secret: Some(SECRET.into()),
        ..WebhookConfig::default()
    };
    let err = cfg
        .validate_startup()
        .expect_err("must refuse plaintext exposure");
    assert!(err.contains("tls_terminated_upstream"), "err: {err}");

    let ok = WebhookConfig {
        bind: "0.0.0.0".into(),
        secret: Some(SECRET.into()),
        tls_terminated_upstream: true,
        ..WebhookConfig::default()
    };
    assert!(ok.validate_startup().is_ok());
}

/// The secret must never reach a log line or a panic message.
#[test]
fn debug_redacts_the_secret() {
    let cfg = WebhookConfig {
        secret: Some("super-secret-value".into()),
        ..WebhookConfig::default()
    };
    let shown = format!("{cfg:?}");
    assert!(!shown.contains("super-secret-value"), "leaked: {shown}");
    assert!(shown.contains("<redacted>"), "shown: {shown}");
}

/// Startup validation runs BEFORE the listener binds, so a bad config cannot
/// leave a socket open.
#[test]
fn bad_config_does_not_bind() {
    let port = free_port();
    let cfg = WebhookConfig {
        port,
        secret: None,
        ..WebhookConfig::default()
    };
    let (tx, _rx) = mpsc::channel();
    assert!(run_webhook_server(&cfg, tx, Arc::new(AtomicBool::new(true))).is_err());
    // The port must still be bindable, proving nothing was left listening.
    //
    // RETRIED, because the assertion is about OUR server and the port number is
    // a SHARED, CONTENDED resource. `free_port()` binds :0, reads the number
    // and drops the listener, so between that and this line the kernel is free
    // to hand the same port to anything else on the machine — another test in
    // this binary, or anything else on the host. A single immediate bind makes
    // the test's verdict depend on winning that race, which is why it passed
    // 15/15 locally and failed inside the clean-room container, where the
    // scheduling differs.
    //
    // This does NOT weaken the property. If our server had left a listener, the
    // port stays bound for the whole window and every attempt fails. Retrying
    // only tolerates a transient unrelated user of the same number.
    let mut bound = false;
    for _ in 0..50 {
        if TcpListener::bind(("127.0.0.1", port)).is_ok() {
            bound = true;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
    assert!(
        bound,
        "port {port} was still bound 1s after a rejected config — the server left a listener"
    );
}