feature: HTTP API, event stream and Prometheus metrics (bgp-web)

Phase 2 of the teaching demo: everything the status page will need is now
reachable over HTTP.

- New crate crates/web (axum). JSON endpoints for status, peers, local
  routes (announce / withdraw), received routes, the exact UPDATE bytes a
  local route produces per session, and well-known community names.
  Communities are accepted as ASN:VALUE or by name (NO_EXPORT, BLACKHOLE,
  ...); large communities as A:B:C. Policy errors map to 403/409/422.
- GET /api/events is a server-sent event stream merging session up/down,
  local route and Adj-RIB-In changes, with lag notifications.
- GET /metricz serves Prometheus metrics; the server crate now counts
  messages sent/received by type and sessions established.
- The runnable main moves from crates/server to bin/bgp so it can depend
  on both server and web without a cycle. `cargo run -- config.textpb`.
- Config gains http_endpoint; unset disables the API.
- Integration tests drive the API against a live server with a real BGP
  peer, including verifying the wire view matches the bytes the peer got.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Cargo.toml

@@ -19,6 +19,7 @@ bgp-packet = { path = "crates/packet" }
 bgp-primitives = { path = "crates/primitives" }
 bgp-rib = { path = "crates/rib" }
 bgp-server = { path = "crates/server" }
+bgp-web = { path = "crates/web" }
 bgp-integration-tests = { path = "crates/integration-tests" }
 
 # --- General ---
@@ -39,3 +40,6 @@ protobuf = { package = "protobuf", version = "3.7.2" }
 protobuf-codegen = { package = "protobuf-codegen", version = "3.7.2" }
 clap = { version = "4.5.53", features = ["derive"] }
 futures = "0.3"
+axum = "0.8.9"
+tokio-stream = { version = "0.1.19", features = ["sync"] }
+reqwest = { version = "0.13.5", default-features = false, features = ["json"] }

bin/bgp/Cargo.toml

@@ -7,5 +7,10 @@ name               = "bgp"
 version.workspace  = true
 
 [dependencies]
-eyre.workspace  = true
-tokio.workspace = true
+bgp-server.workspace = true
+bgp-web.workspace    = true
+clap.workspace       = true
+eyre.workspace       = true
+protobuf.workspace   = true
+tokio.workspace      = true
+tracing.workspace    = true

bin/bgp/README.md

@@ -1,9 +1,48 @@
 # BGP
 
-This crate provides a binary which runs the Border Gateway Protocol (BGP).
+This crate provides the `bgp` binary which runs the Border Gateway Protocol
+daemon together with its HTTP API.
 
-Please refer to the other implementation crates part of this repository where most of the actual implemntation lives.
+```
+cargo run -- path/to/config.textpb
+```
 
-Warning: This is still very much a work in progress and is not yet ready for use.
+The configuration format is defined in `crates/server/proto/config.proto`;
+`testdata/server.textpb` is a worked example. When `http_endpoint` is set the
+binary also serves:
 
-The main ongoing work is happening currently in the BGP packet crate.
+| Path | Purpose |
+|---|---|
+| `GET /api/status` | Router ID, ASN, uptime and table sizes |
+| `GET /api/peers` | Established sessions |
+| `GET /api/routes/local` | Locally originated routes |
+| `POST /api/routes/local` | Announce a route (JSON body, see below) |
+| `DELETE /api/routes/local` | Withdraw a route (JSON body: `prefix`, optional `owner`) |
+| `GET /api/routes/local/wire?prefix=…` | The exact UPDATE bytes sent to each peer |
+| `GET /api/routes/received` | Adj-RIB-In contents |
+| `GET /api/communities/well-known` | Names accepted in place of numeric communities |
+| `GET /api/events` | Server-sent events for session and route changes |
+| `GET /metricz` | Prometheus metrics |
+
+Announce body:
+
+```json
+{
+  "prefix": "10.100.1.0/24",
+  "owner": "alice",
+  "origin": "IGP",
+  "prepend": 2,
+  "med": 100,
+  "communities": ["65000:100", "NO_EXPORT"],
+  "large_communities": ["65000:1:2"]
+}
+```
+
+Only `prefix` and `owner` are required. Routes are subject to the
+`local_route_policy` in the configuration.
+
+Please refer to the other implementation crates in this repository where most
+of the actual implementation lives.
+
+Warning: This is still very much a work in progress and is not yet ready for
+production use.

bin/bgp/src/main.rs

@@ -1,6 +1,55 @@
-use eyre::Result;
+use std::sync::Arc;
+use std::time::Instant;
+
+use bgp_server::config::Config;
+use bgp_server::proto::config::BgpConfig;
+use bgp_server::server::BgpServerBuilder;
+use bgp_server::tracing::TracingArgs;
+use bgp_web::{AppState, RouterInfo};
+use clap::Parser;
+use eyre::{Context, Result};
+use tracing::error;
+
+#[derive(Parser, Debug)]
+struct ServerArgs {
+    /// Path to the textproto configuration file.
+    config: String,
+
+    #[command(flatten)]
+    tracing: TracingArgs,
+}
 
 #[tokio::main]
-pub async fn main() -> Result<()> {
-    todo!("Implement the server");
+async fn main() -> Result<()> {
+    let args = ServerArgs::parse();
+    args.tracing.install()?;
+
+    let config_string =
+        std::fs::read_to_string(&args.config).wrap_err("Failed to open config file")?;
+    let proto_config = protobuf::text_format::parse_from_str::<BgpConfig>(&config_string)
+        .wrap_err("Failed to parse config file")?;
+    let config = Config::from_proto(proto_config).wrap_err("Invalid configuration")?;
+
+    let server = BgpServerBuilder::new(config).bind().await?;
+
+    if let Some(http_endpoint) = server.config().http_endpoint {
+        let state = AppState {
+            info: Arc::new(RouterInfo {
+                router_id: server.config().router_id,
+                local_asn: server.config().local_asn,
+                version: env!("CARGO_PKG_VERSION").to_string(),
+                started: Instant::now(),
+            }),
+            registry: server.registry.clone(),
+            local_rib: server.local_rib.clone(),
+            adj_rib_in: server.adj_rib_in.clone(),
+        };
+        tokio::spawn(async move {
+            if let Err(e) = bgp_web::serve(state, http_endpoint).await {
+                error!(err = %e, "HTTP server exited");
+            }
+        });
+    }
+
+    server.run().await
 }

crates/integration-tests/Cargo.toml

@@ -8,6 +8,9 @@ bgp-packet.workspace = true
 bgp-primitives.workspace = true
 bgp-rib.workspace = true
 bgp-server.workspace = true
+bgp-web.workspace = true
 bytes.workspace = true
 eyre.workspace = true
+reqwest.workspace = true
+serde_json.workspace = true
 tokio.workspace = true

crates/integration-tests/src/test_server.rs

@@ -4,7 +4,9 @@ use bgp_rib::{AdjRibIn, LocalRib, LocalRibPolicy};
 use bgp_server::config::Config;
 use bgp_server::server::BgpServerBuilder;
 use bgp_server::session::SessionRegistry;
+use bgp_web::{AppState, RouterInfo};
 use eyre::Result;
+use tokio::net::TcpListener;
 use tokio::task::JoinHandle;
 
 /// A BGP server started on an ephemeral port for testing.
@@ -19,6 +21,7 @@ pub struct TestServer {
     pub local_rib: LocalRib,
     /// Live view of the server's established sessions.
     pub registry: SessionRegistry,
+    pub local_asn: u32,
 }
 
 impl TestServer {
@@ -45,6 +48,7 @@ impl TestServer {
             listen_endpoints: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)],
             peers: vec![],
             local_route_policy: policy,
+            http_endpoint: None,
         };
 
         let bound = BgpServerBuilder::new(config).bind().await?;
@@ -58,7 +62,41 @@ impl TestServer {
         let registry = bound.registry.clone();
 
         let handle = tokio::spawn(bound.run());
-        Ok(Self { handle, addr, adj_rib_in, local_rib, registry })
+        Ok(Self { handle, addr, adj_rib_in, local_rib, registry, local_asn })
+    }
+}
+
+impl TestServer {
+    /// Starts the HTTP API on an ephemeral loopback port and returns its
+    /// base URL (e.g. `http://127.0.0.1:12345`). The API task is aborted
+    /// when the returned guard drops.
+    pub async fn serve_http(&self) -> Result<(String, HttpGuard)> {
+        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
+        let base_url = format!("http://{}", listener.local_addr()?);
+        let state = AppState {
+            info: std::sync::Arc::new(RouterInfo {
+                router_id: Ipv4Addr::new(127, 0, 0, 1),
+                local_asn: self.local_asn,
+                version: "test".into(),
+                started: std::time::Instant::now(),
+            }),
+            registry: self.registry.clone(),
+            local_rib: self.local_rib.clone(),
+            adj_rib_in: self.adj_rib_in.clone(),
+        };
+        let handle = tokio::spawn(bgp_web::serve_on(state, listener));
+        Ok((base_url, HttpGuard { handle }))
+    }
+}
+
+/// Aborts the HTTP API task when dropped.
+pub struct HttpGuard {
+    handle: JoinHandle<Result<()>>,
+}
+
+impl Drop for HttpGuard {
+    fn drop(&mut self) {
+        self.handle.abort();
     }
 }
 

crates/integration-tests/tests/http_api.rs

@@ -0,0 +1,298 @@
+//! 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);
+
+    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]["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_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(())
+}

crates/server/proto/config.proto

@@ -20,6 +20,10 @@ message BgpConfig {
 
   // Limits on routes originated locally (e.g. via the demo UI).
   LocalRoutePolicy local_route_policy = 5;
+
+  // Where to serve the HTTP API, status page and metrics. The port is
+  // required. Unset disables the HTTP server.
+  TcpEndpoint http_endpoint = 6;
 }
 
 // Limits on locally originated routes.

crates/server/src/config.rs

@@ -20,6 +20,8 @@ pub struct Config {
     pub peers: Vec<PeerConfig>,
     /// Limits on locally originated routes.
     pub local_route_policy: LocalRibPolicy,
+    /// Address for the HTTP API; `None` disables it.
+    pub http_endpoint: Option<SocketAddr>,
 }
 
 pub struct PeerConfig {
@@ -90,6 +92,16 @@ impl Config {
             None => LocalRibPolicy::default(),
         };
 
+        let http_endpoint = match proto.http_endpoint.into_option() {
+            Some(ep) => {
+                if ep.port == 0 {
+                    bail!("http_endpoint.port is required");
+                }
+                Some(parse_endpoint(&ep.address, ep.port).wrap_err("Invalid http_endpoint")?)
+            }
+            None => None,
+        };
+
         Ok(Self {
             router_id,
             local_asn: proto.local_asn,
@@ -97,6 +109,7 @@ impl Config {
             listen_endpoints,
             peers,
             local_route_policy,
+            http_endpoint,
         })
     }
 }

crates/server/src/fsm.rs

@@ -18,6 +18,7 @@ use tracing::{info, warn};
 
 use bgp_packet::message::{AS_TRANS, BGP4_VERSION};
 
+use crate::metrics::{MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, message_type_name};
 use crate::session::{SessionHandle, SessionRegistry};
 
 /// Number of outbound messages that may be queued for a session before
@@ -141,6 +142,7 @@ pub async fn run_session(
     }
 
     info!(%peer_addr, peer_asn, "Session established");
+    SESSIONS_ESTABLISHED_TOTAL.inc();
 
     let peer_id = PeerId { addr: peer_addr, asn: peer_asn };
 
@@ -266,6 +268,7 @@ async fn read_message(reader: &mut OwnedReadHalf, ctx: &ParserContext) -> Result
 
     let (_, msg) =
         Message::from_wire(ctx, &buf).map_err(|e| eyre!("Failed to parse BGP message: {e:?}"))?;
+    MESSAGES_RECEIVED.with_label_values(&[message_type_name(&msg)]).inc();
     Ok(msg)
 }
 
@@ -287,6 +290,7 @@ async fn send_message(
     let mut buf = BytesMut::new();
     msg.to_wire(ctx, &mut buf)?;
     writer.write_all(&buf).await?;
+    MESSAGES_SENT.with_label_values(&[message_type_name(msg)]).inc();
     Ok(())
 }
 

crates/server/src/lib.rs

@@ -2,6 +2,7 @@ pub mod advertise;
 pub mod config;
 pub mod connection;
 pub mod fsm;
+pub mod metrics;
 pub mod server;
 pub mod session;
 pub mod tracing;

crates/server/src/main.rs

@@ -1,29 +0,0 @@
-use bgp_server::config::Config;
-use bgp_server::proto::config::BgpConfig;
-use bgp_server::server::BgpServerBuilder;
-use bgp_server::tracing::TracingArgs;
-use clap::Parser;
-use eyre::{Context, Result};
-
-#[derive(Parser, Debug)]
-struct ServerArgs {
-    /// Path to the textproto configuration file.
-    config: String,
-
-    #[command(flatten)]
-    tracing: TracingArgs,
-}
-
-#[tokio::main]
-async fn main() -> Result<()> {
-    let args = ServerArgs::parse();
-    args.tracing.install()?;
-
-    let config_string =
-        std::fs::read_to_string(&args.config).wrap_err("Failed to open config file")?;
-    let proto_config = protobuf::text_format::parse_from_str::<BgpConfig>(&config_string)
-        .wrap_err("Failed to parse config file")?;
-    let config = Config::from_proto(proto_config).wrap_err("Invalid configuration")?;
-
-    BgpServerBuilder::new(config).run().await
-}

crates/server/src/metrics.rs

@@ -0,0 +1,59 @@
+//! Process-wide Prometheus metrics, registered on the default registry.
+
+use std::sync::LazyLock;
+
+use bgp_packet::message::Message;
+use prometheus::{
+    IntCounter, IntCounterVec, IntGauge, register_int_counter, register_int_counter_vec,
+    register_int_gauge,
+};
+
+pub static MESSAGES_SENT: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!("bgp_messages_sent_total", "BGP messages sent, by type", &["type"])
+        .expect("metric registration")
+});
+
+pub static MESSAGES_RECEIVED: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!("bgp_messages_received_total", "BGP messages received, by type", &["type"])
+        .expect("metric registration")
+});
+
+pub static SESSIONS_ESTABLISHED_TOTAL: LazyLock<IntCounter> = LazyLock::new(|| {
+    register_int_counter!("bgp_sessions_established_total", "Sessions that reached Established")
+        .expect("metric registration")
+});
+
+pub static SESSIONS_ESTABLISHED: LazyLock<IntGauge> = LazyLock::new(|| {
+    register_int_gauge!("bgp_sessions_established", "Sessions currently in Established")
+        .expect("metric registration")
+});
+
+pub static LOCAL_ROUTES: LazyLock<IntGauge> = LazyLock::new(|| {
+    register_int_gauge!("bgp_local_routes", "Locally originated routes").expect("metric registration")
+});
+
+pub static ADJ_RIB_IN_ROUTES: LazyLock<IntGauge> = LazyLock::new(|| {
+    register_int_gauge!("bgp_adj_rib_in_routes", "Routes held in Adj-RIB-In across all peers")
+        .expect("metric registration")
+});
+
+/// Forces registration of every metric so they appear in scrapes before
+/// their first observation.
+pub fn init() {
+    LazyLock::force(&MESSAGES_SENT);
+    LazyLock::force(&MESSAGES_RECEIVED);
+    LazyLock::force(&SESSIONS_ESTABLISHED_TOTAL);
+    LazyLock::force(&SESSIONS_ESTABLISHED);
+    LazyLock::force(&LOCAL_ROUTES);
+    LazyLock::force(&ADJ_RIB_IN_ROUTES);
+}
+
+pub fn message_type_name(msg: &Message) -> &'static str {
+    match msg {
+        Message::Open(_) => "open",
+        Message::Update(_) => "update",
+        Message::Notification(_) => "notification",
+        Message::KeepAlive => "keepalive",
+        Message::RouteRefresh => "route_refresh",
+    }
+}

crates/server/src/server.rs

@@ -67,12 +67,17 @@ impl BgpServerBuilder {
 }
 
 impl BgpServer {
+    pub fn config(&self) -> &Config {
+        &self.config
+    }
+
     /// The addresses actually bound (useful when the config uses port 0).
     pub fn local_addrs(&self) -> Vec<SocketAddr> {
         self.listeners.iter().map(|(addr, _)| *addr).collect()
     }
 
     pub async fn run(self) -> Result<()> {
+        crate::metrics::init();
         info!(
             router_id = %self.config.router_id,
             local_asn = self.config.local_asn,

crates/server/src/session.rs

@@ -14,6 +14,8 @@ use bgp_packet::message::Message;
 use bgp_rib::PeerId;
 use tokio::sync::{broadcast, mpsc};
 
+use crate::metrics::SESSIONS_ESTABLISHED;
+
 const EVENT_CHANNEL_CAPACITY: usize = 256;
 
 /// Handle to one established session. Cloning is cheap.
@@ -80,6 +82,7 @@ impl SessionRegistry {
     pub fn register(&self, handle: SessionHandle) -> SessionGuard {
         let peer = handle.peer;
         self.inner.write().expect("session registry poisoned").insert(peer, handle.clone());
+        SESSIONS_ESTABLISHED.inc();
         let _ = self.events.send(SessionEvent::Up(handle));
         SessionGuard { registry: self.clone(), peer }
     }
@@ -100,6 +103,7 @@ impl SessionRegistry {
     fn remove(&self, peer: &PeerId) {
         let removed = self.inner.write().expect("session registry poisoned").remove(peer);
         if removed.is_some() {
+            SESSIONS_ESTABLISHED.dec();
             let _ = self.events.send(SessionEvent::Down(*peer));
         }
     }

crates/server/src/tracing.rs

@@ -4,7 +4,7 @@ use tracing::Level;
 use tracing_subscriber::EnvFilter;
 
 /// Default targets that we enable logging for.
-const DEFAULT_TARGETS: [&str; 1] = ["bgp_server"];
+const DEFAULT_TARGETS: [&str; 3] = ["bgp", "bgp_server", "bgp_web"];
 
 #[derive(Debug, Args)]
 pub struct TracingArgs {

crates/web/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "bgp-web"
+description = "HTTP API, event stream and metrics for the BGP daemon"
+edition.workspace = true
+
+[dependencies]
+axum.workspace = true
+bgp-packet.workspace = true
+bgp-primitives.workspace = true
+bgp-rib.workspace = true
+bgp-server.workspace = true
+bytes.workspace = true
+eyre.workspace = true
+futures.workspace = true
+prometheus.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+tokio.workspace = true
+tokio-stream.workspace = true
+tracing.workspace = true
+
+[dev-dependencies]
+reqwest.workspace = true

crates/web/src/api.rs

@@ -0,0 +1,396 @@
+//! JSON handlers and their view types.
+
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use axum::Json;
+use axum::extract::{Query, State};
+use axum::http::StatusCode;
+use axum::response::{IntoResponse, Response};
+use bgp_packet::message::{Message, OriginPathAttribute, PathAttribute, UpdateMessage};
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_primitives::parser::ParserContext;
+use bgp_rib::{LocalRibError, LocalRoute, LocalRouteAttrs, SnapshotEntry};
+use bgp_server::advertise::build_announce;
+use bgp_server::session::SessionHandle;
+use bytes::BytesMut;
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+use crate::AppState;
+use crate::community;
+
+const MAX_OWNER_LEN: usize = 64;
+const MAX_PREPEND: u8 = 16;
+
+// ---------------------------------------------------------------------------
+// Errors
+// ---------------------------------------------------------------------------
+
+#[derive(Debug)]
+pub struct ApiError {
+    status: StatusCode,
+    message: String,
+}
+
+impl ApiError {
+    fn bad_request(message: impl Into<String>) -> Self {
+        Self { status: StatusCode::BAD_REQUEST, message: message.into() }
+    }
+}
+
+impl IntoResponse for ApiError {
+    fn into_response(self) -> Response {
+        (self.status, Json(json!({ "error": self.message }))).into_response()
+    }
+}
+
+impl From<LocalRibError> for ApiError {
+    fn from(e: LocalRibError) -> Self {
+        let status = match e {
+            LocalRibError::NotAllowed(_) => StatusCode::FORBIDDEN,
+            LocalRibError::OwnedByOther { .. } => StatusCode::CONFLICT,
+            LocalRibError::OwnerLimit { .. } => StatusCode::UNPROCESSABLE_ENTITY,
+            LocalRibError::Closed => StatusCode::SERVICE_UNAVAILABLE,
+        };
+        Self { status, message: e.to_string() }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Views
+// ---------------------------------------------------------------------------
+
+fn unix_secs(t: SystemTime) -> u64 {
+    t.duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0)
+}
+
+#[derive(Debug, Serialize)]
+pub struct StatusView {
+    pub router_id: String,
+    pub local_asn: u32,
+    pub version: String,
+    pub uptime_secs: u64,
+    pub sessions: usize,
+    pub local_routes: usize,
+    pub received_routes: usize,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct PeerView {
+    pub peer_addr: String,
+    pub peer_asn: u32,
+    pub local_addr: String,
+    pub established_at_unix: u64,
+    pub uptime_secs: u64,
+    pub four_octet_asn: bool,
+}
+
+impl From<&SessionHandle> for PeerView {
+    fn from(s: &SessionHandle) -> Self {
+        Self {
+            peer_addr: s.peer.addr.to_string(),
+            peer_asn: s.peer.asn,
+            local_addr: s.local_addr.to_string(),
+            established_at_unix: unix_secs(s.established_at),
+            uptime_secs: s.established_at.elapsed().map(|d| d.as_secs()).unwrap_or(0),
+            four_octet_asn: s.four_octet_asn,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct LocalRouteView {
+    pub prefix: String,
+    pub owner: String,
+    pub origin: String,
+    pub prepend: u8,
+    pub med: Option<u32>,
+    pub communities: Vec<String>,
+    pub large_communities: Vec<String>,
+    pub announced_at_unix: u64,
+}
+
+impl From<&LocalRoute> for LocalRouteView {
+    fn from(r: &LocalRoute) -> Self {
+        Self {
+            prefix: r.prefix.to_string(),
+            owner: r.owner.clone(),
+            origin: origin_name(r.attrs.origin).to_string(),
+            prepend: r.attrs.prepend,
+            med: r.attrs.med,
+            communities: r.attrs.communities.iter().copied().map(community::format_standard).collect(),
+            large_communities: r.attrs.large_communities.iter().map(community::format_large).collect(),
+            announced_at_unix: unix_secs(r.announced_at),
+        }
+    }
+}
+
+/// Human-oriented digest of a path attribute list.
+#[derive(Debug, Clone, Default, Serialize)]
+pub struct AttrSummary {
+    pub origin: Option<String>,
+    /// AS_PATH flattened in order; AS_SETs are included in place.
+    pub as_path: Vec<u32>,
+    pub next_hop: Option<String>,
+    pub med: Option<u32>,
+    pub local_pref: Option<u32>,
+    pub communities: Vec<String>,
+    pub large_communities: Vec<String>,
+}
+
+pub fn summarize(attrs: &[PathAttribute]) -> AttrSummary {
+    let mut s = AttrSummary::default();
+    for attr in attrs {
+        match attr {
+            PathAttribute::Origin(o) => s.origin = Some(origin_name(*o).to_string()),
+            PathAttribute::AsPath(p) => {
+                s.as_path = p.segments.iter().flat_map(|seg| seg.path.iter().copied()).collect()
+            }
+            PathAttribute::NextHop(nh) => s.next_hop = Some(nh.0.to_string()),
+            PathAttribute::MpReachNlri(mp) => s.next_hop = Some(format!("{:?}", mp.next_hop)),
+            PathAttribute::MultiExitDisc(m) => s.med = Some(m.0),
+            PathAttribute::LocalPref(l) => s.local_pref = Some(l.0),
+            PathAttribute::Communities(c) => {
+                s.communities = c.0.iter().copied().map(community::format_standard).collect()
+            }
+            PathAttribute::LargeCommunities(l) => {
+                s.large_communities = l.communities.iter().map(community::format_large).collect()
+            }
+            _ => {}
+        }
+    }
+    s
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ReceivedRouteView {
+    pub peer_addr: String,
+    pub peer_asn: u32,
+    pub prefix: String,
+    pub received_at_unix: u64,
+    pub summary: AttrSummary,
+    /// The full decoded attribute list, for the curious.
+    pub attributes: Vec<PathAttribute>,
+}
+
+impl From<&SnapshotEntry> for ReceivedRouteView {
+    fn from(e: &SnapshotEntry) -> Self {
+        Self {
+            peer_addr: e.peer.addr.to_string(),
+            peer_asn: e.peer.asn,
+            prefix: e.prefix.to_string(),
+            received_at_unix: unix_secs(e.entry.received_at),
+            summary: summarize(&e.entry.path_attributes),
+            attributes: e.entry.path_attributes.clone(),
+        }
+    }
+}
+
+/// The exact UPDATE a route produces on one session.
+#[derive(Debug, Clone, Serialize)]
+pub struct WireView {
+    pub peer_addr: String,
+    pub peer_asn: u32,
+    pub next_hop: String,
+    /// Complete BGP message including the 19-byte header, as lowercase hex.
+    pub hex: String,
+    pub length: usize,
+    pub summary: AttrSummary,
+    pub update: UpdateMessage,
+}
+
+fn origin_name(o: OriginPathAttribute) -> &'static str {
+    match o {
+        OriginPathAttribute::IGP => "IGP",
+        OriginPathAttribute::EGP => "EGP",
+        OriginPathAttribute::INCOMPLETE => "INCOMPLETE",
+    }
+}
+
+fn parse_origin(s: &str) -> Result<OriginPathAttribute, ApiError> {
+    match s.trim().to_ascii_uppercase().as_str() {
+        "IGP" => Ok(OriginPathAttribute::IGP),
+        "EGP" => Ok(OriginPathAttribute::EGP),
+        "INCOMPLETE" => Ok(OriginPathAttribute::INCOMPLETE),
+        other => Err(ApiError::bad_request(format!("origin {other:?} must be IGP, EGP or INCOMPLETE"))),
+    }
+}
+
+fn parse_prefix(s: &str) -> Result<IpPrefix, ApiError> {
+    IpPrefix::try_from(s.trim()).map_err(|e| ApiError::bad_request(format!("invalid prefix {s:?}: {e}")))
+}
+
+fn validate_owner(owner: &str) -> Result<String, ApiError> {
+    let owner = owner.trim();
+    if owner.is_empty() {
+        return Err(ApiError::bad_request("owner must not be empty"));
+    }
+    if owner.chars().count() > MAX_OWNER_LEN {
+        return Err(ApiError::bad_request(format!("owner must be at most {MAX_OWNER_LEN} characters")));
+    }
+    Ok(owner.to_string())
+}
+
+// ---------------------------------------------------------------------------
+// Requests
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct AnnounceRequest {
+    pub prefix: String,
+    pub owner: String,
+    #[serde(default)]
+    pub origin: Option<String>,
+    #[serde(default)]
+    pub prepend: u8,
+    #[serde(default)]
+    pub med: Option<u32>,
+    #[serde(default)]
+    pub communities: Vec<String>,
+    #[serde(default)]
+    pub large_communities: Vec<String>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct WithdrawRequest {
+    pub prefix: String,
+    /// Omit to withdraw regardless of owner.
+    #[serde(default)]
+    pub owner: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct WireQuery {
+    pub prefix: String,
+}
+
+// ---------------------------------------------------------------------------
+// Handlers
+// ---------------------------------------------------------------------------
+
+pub async fn status(State(state): State<AppState>) -> Json<StatusView> {
+    Json(StatusView {
+        router_id: state.info.router_id.to_string(),
+        local_asn: state.info.local_asn,
+        version: state.info.version.clone(),
+        uptime_secs: state.info.started.elapsed().as_secs(),
+        sessions: state.registry.list().len(),
+        local_routes: state.local_rib.snapshot().await.len(),
+        received_routes: state.adj_rib_in.snapshot().await.len(),
+    })
+}
+
+pub async fn peers(State(state): State<AppState>) -> Json<Vec<PeerView>> {
+    let mut peers: Vec<PeerView> = state.registry.list().iter().map(PeerView::from).collect();
+    peers.sort_by(|a, b| a.peer_addr.cmp(&b.peer_addr));
+    Json(peers)
+}
+
+pub async fn local_routes(State(state): State<AppState>) -> Json<Vec<LocalRouteView>> {
+    let mut routes: Vec<LocalRouteView> =
+        state.local_rib.snapshot().await.iter().map(LocalRouteView::from).collect();
+    routes.sort_by(|a, b| a.prefix.cmp(&b.prefix));
+    Json(routes)
+}
+
+pub async fn announce(
+    State(state): State<AppState>,
+    Json(req): Json<AnnounceRequest>,
+) -> Result<Json<LocalRouteView>, ApiError> {
+    let prefix = parse_prefix(&req.prefix)?;
+    let owner = validate_owner(&req.owner)?;
+    if req.prepend > MAX_PREPEND {
+        return Err(ApiError::bad_request(format!("prepend must be at most {MAX_PREPEND}")));
+    }
+    let attrs = LocalRouteAttrs {
+        origin: req.origin.as_deref().map(parse_origin).transpose()?.unwrap_or(OriginPathAttribute::IGP),
+        prepend: req.prepend,
+        med: req.med,
+        communities: req
+            .communities
+            .iter()
+            .map(|c| community::parse_standard(c))
+            .collect::<Result<_, _>>()
+            .map_err(ApiError::bad_request)?,
+        large_communities: req
+            .large_communities
+            .iter()
+            .map(|c| community::parse_large(c))
+            .collect::<Result<_, _>>()
+            .map_err(ApiError::bad_request)?,
+    };
+    let route = LocalRoute::new(prefix, owner, attrs);
+    state.local_rib.announce(route.clone()).await?;
+    Ok(Json(LocalRouteView::from(&route)))
+}
+
+pub async fn withdraw(
+    State(state): State<AppState>,
+    Json(req): Json<WithdrawRequest>,
+) -> Result<Json<Option<LocalRouteView>>, ApiError> {
+    let prefix = parse_prefix(&req.prefix)?;
+    let owner = req.owner.as_deref().map(validate_owner).transpose()?;
+    let removed = state.local_rib.withdraw(prefix, owner.as_deref()).await?;
+    Ok(Json(removed.as_ref().map(LocalRouteView::from)))
+}
+
+/// Shows the UPDATE bytes a local route produces on each established session.
+pub async fn wire(
+    State(state): State<AppState>,
+    Query(q): Query<WireQuery>,
+) -> Result<Json<Vec<WireView>>, ApiError> {
+    let prefix = parse_prefix(&q.prefix)?;
+    let route = state
+        .local_rib
+        .snapshot()
+        .await
+        .into_iter()
+        .find(|r| r.prefix == prefix)
+        .ok_or_else(|| ApiError { status: StatusCode::NOT_FOUND, message: format!("{prefix} is not announced") })?;
+
+    let mut views = vec![];
+    for session in state.registry.list() {
+        let Some(update) = build_announce(&route, state.info.local_asn, session.local_addr.ip()) else {
+            continue;
+        };
+        let ctx = ParserContext { four_octet_asn: Some(session.four_octet_asn), address_family: None };
+        let mut buf = BytesMut::new();
+        Message::Update(update.clone())
+            .to_wire(&ctx, &mut buf)
+            .map_err(|e| ApiError { status: StatusCode::INTERNAL_SERVER_ERROR, message: e.to_string() })?;
+        views.push(WireView {
+            peer_addr: session.peer.addr.to_string(),
+            peer_asn: session.peer.asn,
+            next_hop: session.local_addr.ip().to_string(),
+            hex: buf.iter().map(|b| format!("{b:02x}")).collect(),
+            length: buf.len(),
+            summary: summarize(&update.path_attributes),
+            update,
+        });
+    }
+    views.sort_by(|a, b| a.peer_addr.cmp(&b.peer_addr));
+    Ok(Json(views))
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct WellKnownCommunity {
+    pub name: &'static str,
+    pub value: String,
+}
+
+/// Names the API accepts in place of a numeric standard community.
+pub async fn well_known_communities() -> Json<Vec<WellKnownCommunity>> {
+    Json(
+        community::WELL_KNOWN
+            .iter()
+            .map(|(name, value)| WellKnownCommunity { name, value: community::format_standard(*value) })
+            .collect(),
+    )
+}
+
+pub async fn received_routes(State(state): State<AppState>) -> Json<Vec<ReceivedRouteView>> {
+    let mut routes: Vec<ReceivedRouteView> =
+        state.adj_rib_in.snapshot().await.iter().map(ReceivedRouteView::from).collect();
+    routes.sort_by(|a, b| (&a.prefix, &a.peer_addr).cmp(&(&b.prefix, &b.peer_addr)));
+    Json(routes)
+}

crates/web/src/community.rs

@@ -0,0 +1,80 @@
+//! Parsing and formatting of community values as they appear in the API.
+//!
+//! Standard communities are written `ASN:VALUE` (e.g. `65000:100`) or as one
+//! of the well-known names from RFC 1997 / RFC 7999 / RFC 8326. Large
+//! communities are written `GLOBAL:LOCAL1:LOCAL2` (RFC 8092).
+
+use bgp_packet::message::LargeCommunity;
+
+/// Well-known standard communities, by name.
+pub const WELL_KNOWN: &[(&str, (u16, u16))] = &[
+    ("GRACEFUL_SHUTDOWN", (0xFFFF, 0x0000)),
+    ("BLACKHOLE", (0xFFFF, 0x029A)),
+    ("NO_EXPORT", (0xFFFF, 0xFF01)),
+    ("NO_ADVERTISE", (0xFFFF, 0xFF02)),
+    ("NO_EXPORT_SUBCONFED", (0xFFFF, 0xFF03)),
+    ("NOPEER", (0xFFFF, 0xFF04)),
+];
+
+pub fn parse_standard(s: &str) -> Result<(u16, u16), String> {
+    let s = s.trim();
+    if let Some((_, value)) = WELL_KNOWN.iter().find(|(name, _)| name.eq_ignore_ascii_case(s)) {
+        return Ok(*value);
+    }
+    let (high, low) = s
+        .split_once(':')
+        .ok_or_else(|| format!("community {s:?} must be ASN:VALUE or a well-known name"))?;
+    let parse = |part: &str| {
+        part.trim()
+            .parse::<u16>()
+            .map_err(|_| format!("community {s:?}: {part:?} is not a 16-bit number"))
+    };
+    Ok((parse(high)?, parse(low)?))
+}
+
+pub fn format_standard((high, low): (u16, u16)) -> String {
+    format!("{high}:{low}")
+}
+
+pub fn parse_large(s: &str) -> Result<LargeCommunity, String> {
+    let s = s.trim();
+    let parts: Vec<&str> = s.split(':').collect();
+    if parts.len() != 3 {
+        return Err(format!("large community {s:?} must be GLOBAL:LOCAL1:LOCAL2"));
+    }
+    let parse = |part: &str| {
+        part.trim()
+            .parse::<u32>()
+            .map_err(|_| format!("large community {s:?}: {part:?} is not a 32-bit number"))
+    };
+    Ok(LargeCommunity { global_admin: parse(parts[0])?, data_1: parse(parts[1])?, data_2: parse(parts[2])? })
+}
+
+pub fn format_large(c: &LargeCommunity) -> String {
+    format!("{}:{}:{}", c.global_admin, c.data_1, c.data_2)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn standard_roundtrip_and_names() {
+        assert_eq!(parse_standard("65000:100").unwrap(), (65000, 100));
+        assert_eq!(parse_standard(" no_export ").unwrap(), (65535, 65281));
+        assert_eq!(parse_standard("BLACKHOLE").unwrap(), (65535, 666));
+        assert_eq!(format_standard((65000, 100)), "65000:100");
+        assert!(parse_standard("65000").is_err());
+        assert!(parse_standard("70000:1").is_err());
+        assert!(parse_standard("bogus").is_err());
+    }
+
+    #[test]
+    fn large_roundtrip() {
+        let c = parse_large("65000:1:2").unwrap();
+        assert_eq!(c, LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 });
+        assert_eq!(format_large(&c), "65000:1:2");
+        assert!(parse_large("65000:1").is_err());
+        assert!(parse_large("a:b:c").is_err());
+    }
+}

crates/web/src/events.rs

@@ -0,0 +1,90 @@
+//! Server-sent event stream merging session, local-route and received-route
+//! changes into one feed the UI can follow.
+
+use std::convert::Infallible;
+
+use axum::extract::State;
+use axum::response::sse::{Event, KeepAlive, Sse};
+use bgp_rib::{LocalRibEvent, RibEvent};
+use bgp_server::session::SessionEvent;
+use futures::stream::{Stream, StreamExt};
+use serde::Serialize;
+use tokio_stream::wrappers::BroadcastStream;
+use tokio_stream::wrappers::errors::BroadcastStreamRecvError;
+
+use crate::AppState;
+use crate::api::{LocalRouteView, PeerView, ReceivedRouteView, summarize};
+
+#[derive(Debug, Serialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum WireEvent {
+    SessionUp { peer: PeerView },
+    SessionDown { peer_addr: String, peer_asn: u32 },
+    LocalAnnounced { route: LocalRouteView },
+    LocalWithdrawn { prefix: String },
+    ReceivedAnnounced { route: ReceivedRouteView },
+    ReceivedWithdrawn { peer_addr: String, peer_asn: u32, prefix: String },
+    /// The client fell behind and missed `missed` events on `stream`; it
+    /// should refetch the corresponding table.
+    Lagged { stream: &'static str, missed: u64 },
+}
+
+impl WireEvent {
+    fn from_session(ev: Result<SessionEvent, BroadcastStreamRecvError>) -> Self {
+        match ev {
+            Ok(SessionEvent::Up(h)) => Self::SessionUp { peer: PeerView::from(&h) },
+            Ok(SessionEvent::Down(p)) => Self::SessionDown { peer_addr: p.addr.to_string(), peer_asn: p.asn },
+            Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { stream: "sessions", missed: n },
+        }
+    }
+
+    fn from_local(ev: Result<LocalRibEvent, BroadcastStreamRecvError>) -> Self {
+        match ev {
+            Ok(LocalRibEvent::Announced(r)) => Self::LocalAnnounced { route: LocalRouteView::from(&r) },
+            Ok(LocalRibEvent::Withdrawn { prefix }) => Self::LocalWithdrawn { prefix: prefix.to_string() },
+            Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { stream: "local_routes", missed: n },
+        }
+    }
+
+    fn from_received(ev: Result<RibEvent, BroadcastStreamRecvError>) -> Self {
+        match ev {
+            Ok(RibEvent::Announced { peer, prefix, entry }) => Self::ReceivedAnnounced {
+                route: ReceivedRouteView {
+                    peer_addr: peer.addr.to_string(),
+                    peer_asn: peer.asn,
+                    prefix: prefix.to_string(),
+                    received_at_unix: entry
+                        .received_at
+                        .duration_since(std::time::UNIX_EPOCH)
+                        .map(|d| d.as_secs())
+                        .unwrap_or(0),
+                    summary: summarize(&entry.path_attributes),
+                    attributes: entry.path_attributes,
+                },
+            },
+            Ok(RibEvent::Withdrawn { peer, prefix }) => Self::ReceivedWithdrawn {
+                peer_addr: peer.addr.to_string(),
+                peer_asn: peer.asn,
+                prefix: prefix.to_string(),
+            },
+            Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { stream: "received_routes", missed: n },
+        }
+    }
+}
+
+pub async fn stream(
+    State(state): State<AppState>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let sessions = BroadcastStream::new(state.registry.subscribe()).map(WireEvent::from_session);
+    let local = BroadcastStream::new(state.local_rib.subscribe()).map(WireEvent::from_local);
+    let received = BroadcastStream::new(state.adj_rib_in.subscribe()).map(WireEvent::from_received);
+
+    let merged = futures::stream::select(sessions, futures::stream::select(local, received)).map(|ev| {
+        Ok(Event::default()
+            .event("bgp")
+            .json_data(&ev)
+            .unwrap_or_else(|e| Event::default().event("error").data(e.to_string())))
+    });
+
+    Sse::new(merged).keep_alive(KeepAlive::default())
+}

crates/web/src/lib.rs

@@ -0,0 +1,76 @@
+//! HTTP surface of the daemon: a JSON API for originating routes and
+//! inspecting state, a server-sent event stream, and Prometheus metrics.
+//!
+//! Everything here is a thin layer over handles owned by the BGP server
+//! ([`SessionRegistry`], [`LocalRib`], [`AdjRibIn`]); no routing state lives
+//! in this crate.
+
+mod api;
+mod community;
+mod events;
+mod metrics;
+
+use std::net::{Ipv4Addr, SocketAddr};
+use std::sync::Arc;
+use std::time::Instant;
+
+use axum::Router;
+use axum::routing::{delete, get, post};
+use bgp_rib::{AdjRibIn, LocalRib};
+use bgp_server::session::SessionRegistry;
+use eyre::{Result, WrapErr};
+use tokio::net::TcpListener;
+use tracing::info;
+
+pub use api::{
+    AnnounceRequest, AttrSummary, LocalRouteView, PeerView, ReceivedRouteView, StatusView,
+    WellKnownCommunity, WireView, WithdrawRequest,
+};
+
+/// Static facts about the router shown in status output.
+#[derive(Debug, Clone)]
+pub struct RouterInfo {
+    pub router_id: Ipv4Addr,
+    pub local_asn: u32,
+    pub version: String,
+    pub started: Instant,
+}
+
+#[derive(Clone)]
+pub struct AppState {
+    pub info: Arc<RouterInfo>,
+    pub registry: SessionRegistry,
+    pub local_rib: LocalRib,
+    pub adj_rib_in: AdjRibIn,
+}
+
+/// Builds the router. Exposed so tests can drive it without a socket.
+pub fn router(state: AppState) -> Router {
+    Router::new()
+        .route("/api/status", get(api::status))
+        .route("/api/peers", get(api::peers))
+        .route("/api/routes/local", get(api::local_routes).post(api::announce))
+        .route("/api/routes/local", delete(api::withdraw))
+        .route("/api/routes/local/withdraw", post(api::withdraw))
+        .route("/api/routes/local/wire", get(api::wire))
+        .route("/api/routes/received", get(api::received_routes))
+        .route("/api/communities/well-known", get(api::well_known_communities))
+        .route("/api/events", get(events::stream))
+        .route("/metricz", get(metrics::metricz))
+        .with_state(state)
+}
+
+/// Binds `listen` and serves the API until the task is dropped.
+pub async fn serve(state: AppState, listen: SocketAddr) -> Result<()> {
+    let listener = TcpListener::bind(listen)
+        .await
+        .wrap_err_with(|| format!("Failed to bind HTTP listener on {listen}"))?;
+    serve_on(state, listener).await
+}
+
+/// Serves the API on an already-bound listener (lets tests use port 0).
+pub async fn serve_on(state: AppState, listener: TcpListener) -> Result<()> {
+    info!(addr = %listener.local_addr()?, "HTTP API listening");
+    axum::serve(listener, router(state)).await?;
+    Ok(())
+}

crates/web/src/metrics.rs

@@ -0,0 +1,21 @@
+//! Prometheus scrape endpoint. Counters are updated by the server crate as
+//! messages flow; table-size gauges are refreshed at scrape time.
+
+use axum::extract::State;
+use axum::http::{StatusCode, header};
+use axum::response::IntoResponse;
+use bgp_server::metrics::{ADJ_RIB_IN_ROUTES, LOCAL_ROUTES};
+use prometheus::{Encoder, TextEncoder};
+
+use crate::AppState;
+
+pub async fn metricz(State(state): State<AppState>) -> impl IntoResponse {
+    LOCAL_ROUTES.set(state.local_rib.snapshot().await.len() as i64);
+    ADJ_RIB_IN_ROUTES.set(state.adj_rib_in.snapshot().await.len() as i64);
+
+    let mut buf = Vec::new();
+    match TextEncoder::new().encode(&prometheus::gather(), &mut buf) {
+        Ok(()) => (StatusCode::OK, [(header::CONTENT_TYPE, "text/plain; version=0.0.4")], buf).into_response(),
+        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    }
+}

testdata/server.textpb

@@ -13,3 +13,7 @@ local_route_policy {
     allowed_prefixes: "2001:db8:100::/48"
     max_routes_per_owner: 4
 }
+http_endpoint {
+    address: "127.0.0.1"
+    port: 8080
+}