ferrix/crates/integration-tests/tests/http_api.rs

//! Drives the HTTP API against a real server with a real BGP peer attached.

use std::net::Ipv4Addr;
use std::time::Duration;

use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::message::{
    AsPathAttribute, CommunitiesPathAttribute, Message, NextHopPathAttribute,
    OriginPathAttribute, PathAttribute, UpdateMessage,
};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_primitives::parser::ParserContext;
use bgp_rib::LocalRibPolicy;
use eyre::Result;
use serde_json::{Value, json};

const LOCAL_ASN: u32 = 65000;
const PEER_ASN: u32 = 65001;
const TIMEOUT: Duration = Duration::from_secs(5);

fn prefix(s: &str) -> IpPrefix {
    IpPrefix::try_from(s).unwrap()
}

async fn established_peer(server: &TestServer) -> Result<TestPeer> {
    let mut peer = TestPeer::connect(server.addr).await?;
    peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).await?;
    tokio::time::timeout(TIMEOUT, async {
        while server.registry.list().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await?;
    Ok(peer)
}

async fn get_json(client: &reqwest::Client, url: &str) -> Result<Value> {
    Ok(client.get(url).send().await?.error_for_status()?.json().await?)
}

#[tokio::test]
async fn test_status_and_peers() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let status = get_json(&client, &format!("{base}/api/status")).await?;
    assert_eq!(status["local_asn"], LOCAL_ASN);
    assert_eq!(status["router_id"], "127.0.0.1");
    assert_eq!(status["sessions"], 0);
    assert!(status["started_at_unix"].as_u64().unwrap() > 1_700_000_000);
    assert!(!status["hostname"].as_str().unwrap().is_empty());

    let _peer = established_peer(&server).await?;

    let status = get_json(&client, &format!("{base}/api/status")).await?;
    assert_eq!(status["sessions"], 1);

    let peers = get_json(&client, &format!("{base}/api/peers")).await?;
    assert_eq!(peers.as_array().unwrap().len(), 1);
    assert_eq!(peers[0]["peer_asn"], PEER_ASN);
    assert_eq!(peers[0]["local_addr"], server.addr.to_string());
    assert_eq!(peers[0]["four_octet_asn"], true);
    Ok(())
}

/// The full loop: POST an announcement, the BGP peer receives the UPDATE,
/// the local table lists it, the wire view matches the bytes, DELETE
/// withdraws it.
#[tokio::test]
async fn test_announce_over_http_reaches_peer() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();
    let mut peer = established_peer(&server).await?;

    let resp = client
        .post(format!("{base}/api/routes/local"))
        .json(&json!({
            "prefix": "10.100.1.0/24",
            "owner": "alice",
            "prepend": 1,
            "med": 10,
            "communities": ["65000:100", "NO_EXPORT"],
            "large_communities": ["65000:1:2"],
        }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200, "{}", resp.text().await?);
    let body: Value = resp.json().await?;
    assert_eq!(body["prefix"], "10.100.1.0/24");
    assert_eq!(body["origin"], "IGP");
    assert_eq!(body["communities"], json!(["65000:100", "65535:65281"]));
    assert_eq!(body["large_communities"], json!(["65000:1:2"]));

    // The peer sees the UPDATE.
    let update: UpdateMessage = tokio::time::timeout(TIMEOUT, peer.recv_update()).await??;
    assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
    assert_eq!(update.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN, LOCAL_ASN]));
    assert_eq!(update.path_attributes[2], PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::LOCALHOST)));
    assert!(update.path_attributes.contains(&PathAttribute::Communities(CommunitiesPathAttribute(vec![
        (65000, 100),
        (65535, 65281)
    ]))));

    // Local table lists it.
    let local = get_json(&client, &format!("{base}/api/routes/local")).await?;
    assert_eq!(local.as_array().unwrap().len(), 1);
    assert_eq!(local[0]["owner"], "alice");

    // Wire view reproduces exactly what went over the socket.
    let wire = get_json(&client, &format!("{base}/api/routes/local/wire?prefix=10.100.1.0/24")).await?;
    assert_eq!(wire.as_array().unwrap().len(), 1);
    assert_eq!(wire[0]["peer_asn"], PEER_ASN);
    assert_eq!(wire[0]["four_octet_asn"], true);
    assert_eq!(wire[0]["summary"]["as_path"], json!([LOCAL_ASN, LOCAL_ASN]));
    let hex = wire[0]["hex"].as_str().unwrap();
    let ctx = ParserContext { four_octet_asn: peer.ctx.four_octet_asn, address_family: None };
    let mut expected = bytes::BytesMut::new();
    Message::Update(update).to_wire(&ctx, &mut expected)?;
    let expected_hex: String = expected.iter().map(|b| format!("{b:02x}")).collect();
    assert_eq!(hex, expected_hex);
    assert!(hex.starts_with(&"ff".repeat(16)), "marker missing");

    // Withdraw.
    let resp = client
        .delete(format!("{base}/api/routes/local"))
        .json(&json!({ "prefix": "10.100.1.0/24", "owner": "alice" }))
        .send()
        .await?;
    assert_eq!(resp.status(), 200);
    let withdrawn: UpdateMessage = tokio::time::timeout(TIMEOUT, peer.recv_update()).await??;
    assert_eq!(withdrawn.withdrawn_routes, vec![prefix("10.100.1.0/24")]);

    let local = get_json(&client, &format!("{base}/api/routes/local")).await?;
    assert!(local.as_array().unwrap().is_empty());
    Ok(())
}

#[tokio::test]
async fn test_validation_and_policy_status_codes() -> Result<()> {
    let server = TestServer::start_with_policy(
        LOCAL_ASN,
        LocalRibPolicy { allowed_prefixes: vec![prefix("10.100.0.0/16")], max_routes_per_owner: 1 },
    )
    .await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();
    let url = format!("{base}/api/routes/local");

    let post = |body: Value| client.post(&url).json(&body).send();

    // Malformed input → 400 with an error message.
    let resp = post(json!({ "prefix": "not-a-prefix", "owner": "alice" })).await?;
    assert_eq!(resp.status(), 400);
    assert!(resp.json::<Value>().await?["error"].as_str().unwrap().contains("invalid prefix"));

    let resp = post(json!({ "prefix": "10.100.1.0/24", "owner": "" })).await?;
    assert_eq!(resp.status(), 400);

    let resp = post(json!({ "prefix": "10.100.1.0/24", "owner": "alice", "communities": ["bogus"] })).await?;
    assert_eq!(resp.status(), 400);

    let resp = post(json!({ "prefix": "10.100.1.0/24", "owner": "alice", "origin": "weird" })).await?;
    assert_eq!(resp.status(), 400);

    // Outside allowlist → 403.
    let resp = post(json!({ "prefix": "192.0.2.0/24", "owner": "alice" })).await?;
    assert_eq!(resp.status(), 403);

    // First allowed route is fine.
    assert_eq!(post(json!({ "prefix": "10.100.1.0/24", "owner": "alice" })).await?.status(), 200);

    // Per-owner limit → 422.
    let resp = post(json!({ "prefix": "10.100.2.0/24", "owner": "alice" })).await?;
    assert_eq!(resp.status(), 422);

    // Someone else's prefix → 409, for both replace and withdraw.
    let resp = post(json!({ "prefix": "10.100.1.0/24", "owner": "bob" })).await?;
    assert_eq!(resp.status(), 409);
    let resp = client
        .delete(&url)
        .json(&json!({ "prefix": "10.100.1.0/24", "owner": "bob" }))
        .send()
        .await?;
    assert_eq!(resp.status(), 409);

    // Withdrawing an unknown prefix is a 200 with null.
    let resp = client.delete(&url).json(&json!({ "prefix": "10.100.7.0/24" })).send().await?;
    assert_eq!(resp.status(), 200);
    assert_eq!(resp.json::<Value>().await?, Value::Null);

    // Wire view for an unannounced prefix → 404.
    let resp = client.get(format!("{base}/api/routes/local/wire?prefix=10.100.7.0/24")).send().await?;
    assert_eq!(resp.status(), 404);
    Ok(())
}

/// Routes the peer sends us show up in the received table with a summary.
#[tokio::test]
async fn test_received_routes_summary() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();
    let mut peer = established_peer(&server).await?;

    peer.send(&Message::Update(UpdateMessage {
        withdrawn_routes: vec![],
        path_attributes: vec![
            PathAttribute::Origin(OriginPathAttribute::IGP),
            AsPathAttribute::from_asns(vec![PEER_ASN, 65002]),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 1))),
            PathAttribute::Communities(CommunitiesPathAttribute(vec![(65001, 7)])),
        ],
        announced_routes: vec![prefix("203.0.113.0/24")],
    }))
    .await?;

    let received = tokio::time::timeout(TIMEOUT, async {
        loop {
            let v = get_json(&client, &format!("{base}/api/routes/received")).await?;
            if !v.as_array().unwrap().is_empty() {
                return Ok::<_, eyre::Error>(v);
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    })
    .await??;
    assert_eq!(received[0]["prefix"], "203.0.113.0/24");
    assert_eq!(received[0]["peer_asn"], PEER_ASN);
    assert_eq!(received[0]["summary"]["as_path"], json!([PEER_ASN, 65002]));
    assert_eq!(received[0]["summary"]["next_hop"], "10.0.0.1");
    assert_eq!(received[0]["summary"]["communities"], json!(["65001:7"]));
    Ok(())
}

/// The SSE stream reports session and route changes as they happen.
#[tokio::test]
async fn test_event_stream() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let mut resp = client.get(format!("{base}/api/events")).send().await?.error_for_status()?;
    assert!(resp.headers()["content-type"].to_str()?.starts_with("text/event-stream"));

    let _peer = established_peer(&server).await?;
    server
        .local_rib
        .announce(bgp_rib::LocalRoute::new(prefix("10.100.1.0/24"), "alice", Default::default()))
        .await?;

    // Collect SSE frames until we have seen both events.
    let mut buf = String::new();
    let mut seen_up = false;
    let mut seen_announce = false;
    tokio::time::timeout(TIMEOUT, async {
        while !(seen_up && seen_announce) {
            let chunk = resp.chunk().await?.ok_or_else(|| eyre::eyre!("stream ended"))?;
            buf.push_str(std::str::from_utf8(&chunk)?);
            for line in buf.lines().filter_map(|l| l.strip_prefix("data:")) {
                let ev: Value = serde_json::from_str(line.trim())?;
                match ev["type"].as_str() {
                    Some("session_up") if ev["peer"]["peer_asn"] == PEER_ASN => seen_up = true,
                    Some("local_announced") if ev["route"]["prefix"] == "10.100.1.0/24" => seen_announce = true,
                    _ => {}
                }
            }
        }
        Ok::<_, eyre::Error>(())
    })
    .await??;
    Ok(())
}

#[tokio::test]
async fn test_statusz_page_is_served() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let resp = client.get(format!("{base}/statusz")).send().await?;
    assert_eq!(resp.status(), 200);
    assert!(resp.headers()["content-type"].to_str()?.starts_with("text/html"));
    let body = resp.text().await?;
    assert!(body.contains("<title>Status for ferrix_bgp</title>"));
    assert!(body.contains("/api/events"));

    // The root redirects to the page.
    let no_redirect = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).build()?;
    let resp = no_redirect.get(format!("{base}/")).send().await?;
    assert_eq!(resp.status(), 307);
    assert_eq!(resp.headers()["location"], "/statusz");
    Ok(())
}

#[tokio::test]
async fn test_metricz_and_well_known() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();
    let _peer = established_peer(&server).await?;

    let text = client.get(format!("{base}/metricz")).send().await?.error_for_status()?.text().await?;
    // Metrics are process-global, so other tests in this binary contribute
    // too; assert on presence and a lower bound rather than exact values.
    let metric = |name: &str| -> Option<i64> {
        text.lines()
            .find(|l| l.starts_with(name) && l[name.len()..].starts_with(' '))
            .and_then(|l| l.rsplit(' ').next()?.parse().ok())
    };
    assert!(metric("bgp_sessions_established").unwrap_or(0) >= 1, "{text}");
    assert!(metric("bgp_messages_sent_total{type=\"open\"}").unwrap_or(0) >= 1, "{text}");
    assert!(metric("bgp_messages_received_total{type=\"keepalive\"}").unwrap_or(0) >= 1, "{text}");
    assert!(metric("bgp_local_routes").is_some(), "{text}");
    assert!(metric("bgp_adj_rib_in_routes").is_some(), "{text}");

    let wk = get_json(&client, &format!("{base}/api/communities/well-known")).await?;
    assert!(wk.as_array().unwrap().iter().any(|c| c["name"] == "NO_EXPORT" && c["value"] == "65535:65281"));
    Ok(())
}