agent_block_core/bus/
http_source.rs

1//! HTTP → bus source adapter: a listener whose every request is an [`Event`].
2//!
3//! The mesh adapter in `host.rs` turns a relay request into an event with an
4//! ack channel and answers with what the Lua handler returned. This is the
5//! same shape over HTTP: one listener, every request (any method, any path)
6//! becomes an event of one `kind` (`"http"` unless configured), and the
7//! handler's return value is the response. Nothing here knows what the
8//! paths mean — routing is the handler's, in Lua, which is what keeps the
9//! adapter one file and the surface it fronts a Lua concern.
10//!
11//! # What crosses
12//!
13//! `payload` is `{ method, path, query?, body }` — `body` parsed as JSON when
14//! it is JSON, the raw text otherwise, `null` when empty. `meta` is the
15//! configured [`HttpSourceConfig::meta`] with `remote` (the peer address)
16//! added. The handler answers `{ status?, body? }` — `status` defaults to
17//! 200 and `body` is sent as JSON — or any other value, which is sent as a
18//! 200 JSON body. A handler error is a 500 `{ error }`, a handler that did
19//! not answer within [`ACK_TIMEOUT`] a 504.
20//!
21//! # What is refused before it crosses
22//!
23//! Binding to loopback is not enough on its own: a page in a browser on the
24//! same machine can resolve its own name to `127.0.0.1` after the origin
25//! check and reach a loopback server that does not look at `Host` (DNS
26//! rebinding; Ollama CVE-2024-28224, and the same class in rmcp's streamable
27//! HTTP server, GHSA-89vp-x53w-74fx). So the `Host` header must name a
28//! loopback name, or the address this listener is bound to, and an `Origin`
29//! header, when present, must too — a 403 otherwise. When a bearer token is
30//! configured every request must carry it, loopback included: a 401
31//! otherwise. Those two checks happen before the request becomes an event,
32//! so a refused request never reaches Lua.
33//!
34//! Remote use is a matter of what the listener is bound to. The default
35//! callers pass is loopback; binding wider is their opt-in, and the token is
36//! what stands between the wider bind and anyone who can reach the port.
37
38use std::net::SocketAddr;
39use std::sync::Arc;
40use std::time::Duration;
41
42use axum::body::Bytes;
43use axum::extract::{ConnectInfo, State};
44use axum::http::{header, HeaderMap, Method, StatusCode, Uri};
45use axum::response::{IntoResponse, Response};
46use axum::Router;
47use serde_json::{json, Value};
48use tokio::sync::{mpsc, oneshot};
49use tokio_util::sync::CancellationToken;
50
51use super::Event;
52
53/// How long a request waits for the handler's answer. Mirrors the mesh
54/// adapter's bound.
55pub const ACK_TIMEOUT: Duration = Duration::from_secs(30);
56
57/// How long shutdown waits for the listener to finish in-flight requests.
58const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
59
60/// The event kind requests are dispatched under when none is configured.
61pub const DEFAULT_KIND: &str = "http";
62
63/// What the listener is.
64#[derive(Debug, Clone)]
65pub struct HttpSourceConfig {
66    /// The address to bind, e.g. `127.0.0.1:7788`. Port `0` picks a free
67    /// one; the bound address is on the [`HttpSourceHandle`].
68    pub bind: String,
69    /// The bearer token every request must carry. `None` means no token,
70    /// which is only defensible on a bind nothing else can reach.
71    pub token: Option<String>,
72    /// The `kind` requests are dispatched under.
73    pub kind: String,
74    /// Carried on every event's `meta` (an object; anything else is treated
75    /// as empty). What a handler needs to know that a request cannot tell
76    /// it — which store to open, which manager it fronts.
77    pub meta: Value,
78}
79
80impl HttpSourceConfig {
81    /// A listener on `bind`, no token, kind [`DEFAULT_KIND`], empty meta.
82    pub fn new(bind: impl Into<String>) -> Self {
83        Self {
84            bind: bind.into(),
85            token: None,
86            kind: DEFAULT_KIND.to_string(),
87            meta: Value::Object(Default::default()),
88        }
89    }
90
91    pub fn token(mut self, token: impl Into<String>) -> Self {
92        self.token = Some(token.into());
93        self
94    }
95
96    pub fn kind(mut self, kind: impl Into<String>) -> Self {
97        self.kind = kind.into();
98        self
99    }
100
101    pub fn meta(mut self, meta: Value) -> Self {
102        self.meta = meta;
103        self
104    }
105}
106
107/// A running listener: its bound address, and the means to stop it.
108#[derive(Debug)]
109pub struct HttpSourceHandle {
110    /// Where it is listening — what the caller prints, and the only way to
111    /// learn the port when `bind` asked for `0`.
112    pub addr: SocketAddr,
113    task: tokio::task::JoinHandle<()>,
114    cancel: CancellationToken,
115}
116
117impl HttpSourceHandle {
118    /// Stop accepting, let in-flight requests finish (bounded), and join.
119    pub async fn shutdown(self) {
120        self.cancel.cancel();
121        match tokio::time::timeout(SHUTDOWN_TIMEOUT, self.task).await {
122            Ok(Ok(())) => tracing::info!("http source: listener shut down"),
123            Ok(Err(e)) => tracing::error!(error = %e, "http source: listener task join error"),
124            Err(_) => tracing::warn!(
125                timeout_secs = SHUTDOWN_TIMEOUT.as_secs(),
126                "http source: listener did not stop in time; abandoned"
127            ),
128        }
129    }
130}
131
132struct Inner {
133    tx: mpsc::Sender<Event>,
134    token: Option<String>,
135    kind: String,
136    meta: Value,
137    allowed_hosts: Vec<String>,
138}
139
140/// Bind and start serving. Requests become events on `tx`; the handle stops
141/// the listener. Fails only when the bind fails.
142pub async fn start(
143    config: HttpSourceConfig,
144    tx: mpsc::Sender<Event>,
145) -> Result<HttpSourceHandle, String> {
146    let listener = tokio::net::TcpListener::bind(&config.bind)
147        .await
148        .map_err(|e| format!("http source: bind {}: {e}", config.bind))?;
149    let addr = listener
150        .local_addr()
151        .map_err(|e| format!("http source: local_addr: {e}"))?;
152
153    let mut allowed_hosts = vec![
154        "localhost".to_string(),
155        "127.0.0.1".to_string(),
156        "::1".to_string(),
157    ];
158    if !addr.ip().is_loopback() {
159        allowed_hosts.push(addr.ip().to_string());
160    }
161
162    let inner = Arc::new(Inner {
163        tx,
164        token: config.token,
165        kind: config.kind,
166        meta: if config.meta.is_object() {
167            config.meta
168        } else {
169            Value::Object(Default::default())
170        },
171        allowed_hosts,
172    });
173    let app = Router::new().fallback(handle).with_state(inner);
174
175    let cancel = CancellationToken::new();
176    let stop = cancel.clone();
177    let task = tokio::spawn(async move {
178        let server = axum::serve(
179            listener,
180            app.into_make_service_with_connect_info::<SocketAddr>(),
181        )
182        .with_graceful_shutdown(async move { stop.cancelled().await });
183        if let Err(e) = server.await {
184            tracing::error!(error = %e, "http source: listener ended with error");
185        }
186    });
187    tracing::info!(%addr, "http source: listening");
188    Ok(HttpSourceHandle { addr, task, cancel })
189}
190
191/// The host part of a `Host` header value, or of an origin's authority:
192/// the port dropped, IPv6 brackets dropped.
193pub fn host_only(value: &str) -> String {
194    let value = value.trim();
195    if let Some(rest) = value.strip_prefix('[') {
196        if let Some(end) = rest.find(']') {
197            return rest[..end].to_string();
198        }
199    }
200    match value.rsplit_once(':') {
201        Some((host, port)) if port.chars().all(|c| c.is_ascii_digit()) => host.to_string(),
202        _ => value.to_string(),
203    }
204}
205
206/// The host part of an `Origin` header value (`scheme://host[:port]`).
207fn origin_host(value: &str) -> Option<String> {
208    let rest = value.trim().split_once("://")?.1;
209    let authority = rest.split('/').next()?;
210    Some(host_only(authority))
211}
212
213fn reply(status: u16, body: Value) -> Response {
214    let status = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
215    (status, axum::Json(body)).into_response()
216}
217
218/// What a handler's return value means as a response: `{ status?, body? }`
219/// is read as such, anything else is a 200 with that value as the body.
220fn answer(value: Value) -> Response {
221    if let Some(obj) = value.as_object() {
222        if obj.contains_key("status") || obj.contains_key("body") {
223            let status = obj
224                .get("status")
225                .and_then(Value::as_u64)
226                .and_then(|s| u16::try_from(s).ok())
227                .unwrap_or(200);
228            let body = obj.get("body").cloned().unwrap_or(Value::Null);
229            return reply(status, body);
230        }
231    }
232    reply(200, value)
233}
234
235async fn handle(
236    State(inner): State<Arc<Inner>>,
237    ConnectInfo(remote): ConnectInfo<SocketAddr>,
238    method: Method,
239    uri: Uri,
240    headers: HeaderMap,
241    body: Bytes,
242) -> Response {
243    let host = headers
244        .get(header::HOST)
245        .and_then(|v| v.to_str().ok())
246        .map(host_only);
247    let host_ok = host
248        .as_ref()
249        .map(|h| inner.allowed_hosts.iter().any(|a| a == h))
250        .unwrap_or(false);
251    if !host_ok {
252        return reply(403, json!({ "error": "host not allowed" }));
253    }
254    if let Some(origin) = headers.get(header::ORIGIN).and_then(|v| v.to_str().ok()) {
255        let origin_ok = origin_host(origin)
256            .map(|h| inner.allowed_hosts.iter().any(|a| a == &h))
257            .unwrap_or(false);
258        if !origin_ok {
259            return reply(403, json!({ "error": "origin not allowed" }));
260        }
261    }
262    if let Some(expected) = &inner.token {
263        let presented = headers
264            .get(header::AUTHORIZATION)
265            .and_then(|v| v.to_str().ok())
266            .and_then(|v| v.strip_prefix("Bearer "))
267            .map(str::trim);
268        if presented != Some(expected.as_str()) {
269            return reply(401, json!({ "error": "token required" }));
270        }
271    }
272
273    let body_value = if body.is_empty() {
274        Value::Null
275    } else {
276        serde_json::from_slice(&body)
277            .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&body).into_owned()))
278    };
279    let payload = json!({
280        "method": method.as_str(),
281        "path": uri.path(),
282        "query": uri.query(),
283        "body": body_value,
284    });
285    let mut meta = inner.meta.clone();
286    if let Some(obj) = meta.as_object_mut() {
287        obj.insert("remote".to_string(), Value::String(remote.to_string()));
288    }
289
290    let id = uuid::Uuid::new_v4().to_string();
291    let (ack_tx, ack_rx) = oneshot::channel();
292    let event = Event {
293        kind: inner.kind.clone(),
294        id: id.clone(),
295        payload,
296        meta,
297        ack_tx: Some(ack_tx),
298    };
299    if let Err(e) = inner.tx.send(event).await {
300        tracing::error!(error = %e, id = %id, "http source: bus channel closed; rejecting request");
301        return reply(503, json!({ "error": "bus channel closed" }));
302    }
303    match tokio::time::timeout(ACK_TIMEOUT, ack_rx).await {
304        Ok(Ok(Ok(value))) => answer(value),
305        Ok(Ok(Err(e))) => {
306            tracing::error!(id = %id, error = %e, "http source: handler returned error");
307            reply(500, json!({ "error": e.to_string() }))
308        }
309        Ok(Err(e)) => {
310            tracing::error!(id = %id, error = %e, "http source: ack receiver dropped");
311            reply(500, json!({ "error": "ack dropped" }))
312        }
313        Err(_) => {
314            tracing::error!(id = %id, timeout_secs = ACK_TIMEOUT.as_secs(), "http source: handler timeout");
315            reply(504, json!({ "error": "handler timeout" }))
316        }
317    }
318}
319
320#[cfg(test)]
321mod tests {
322    use super::*;
323
324    #[test]
325    fn host_only_drops_a_port_and_ipv6_brackets() {
326        assert_eq!(host_only("localhost:7788"), "localhost");
327        assert_eq!(host_only("127.0.0.1"), "127.0.0.1");
328        assert_eq!(host_only("[::1]:7788"), "::1");
329        assert_eq!(host_only("[::1]"), "::1");
330        assert_eq!(host_only("evil.example"), "evil.example");
331    }
332
333    #[test]
334    fn origin_host_reads_the_authority() {
335        assert_eq!(
336            origin_host("http://localhost:3000").as_deref(),
337            Some("localhost")
338        );
339        assert_eq!(
340            origin_host("https://evil.example/x").as_deref(),
341            Some("evil.example")
342        );
343        assert_eq!(origin_host("null"), None);
344    }
345
346    /// A listener on an ephemeral loopback port whose one handler echoes the
347    /// event back as `{ status = 201, body = { path, method, body, remote? } }`.
348    async fn echoing() -> (HttpSourceHandle, tokio::task::JoinHandle<()>) {
349        let (tx, mut rx) = mpsc::channel::<Event>(8);
350        let handle = start(HttpSourceConfig::new("127.0.0.1:0").token("t0k3n"), tx)
351            .await
352            .expect("bind");
353        let pump = tokio::spawn(async move {
354            while let Some(mut ev) = rx.recv().await {
355                let ack = ev.ack_tx.take().expect("ack channel");
356                let has_remote = ev.meta.get("remote").is_some();
357                let _ = ack.send(Ok(json!({
358                    "status": 201,
359                    "body": { "echo": ev.payload, "kind": ev.kind, "remote": has_remote },
360                })));
361            }
362        });
363        (handle, pump)
364    }
365
366    #[tokio::test]
367    async fn a_request_with_the_token_becomes_an_event_and_the_answer_the_response() {
368        let (handle, pump) = echoing().await;
369        let url = format!(
370            "http://127.0.0.1:{}/jobs/x/runs?limit=3",
371            handle.addr.port()
372        );
373        let res = reqwest::Client::new()
374            .post(&url)
375            .bearer_auth("t0k3n")
376            .header(header::CONTENT_TYPE, "application/json")
377            .body(r#"{"by":"test"}"#)
378            .send()
379            .await
380            .expect("request");
381        assert_eq!(res.status().as_u16(), 201);
382        let body: Value = res.json().await.expect("json");
383        assert_eq!(body["kind"], "http");
384        assert_eq!(body["remote"], true);
385        assert_eq!(body["echo"]["method"], "POST");
386        assert_eq!(body["echo"]["path"], "/jobs/x/runs");
387        assert_eq!(body["echo"]["query"], "limit=3");
388        assert_eq!(body["echo"]["body"]["by"], "test");
389        handle.shutdown().await;
390        pump.abort();
391    }
392
393    #[tokio::test]
394    async fn without_the_token_nothing_reaches_the_bus() {
395        let (handle, pump) = echoing().await;
396        let url = format!("http://127.0.0.1:{}/jobs", handle.addr.port());
397        let res = reqwest::get(&url).await.expect("request");
398        assert_eq!(res.status().as_u16(), 401);
399        let res = reqwest::Client::new()
400            .get(&url)
401            .bearer_auth("wrong")
402            .send()
403            .await
404            .expect("request");
405        assert_eq!(res.status().as_u16(), 401);
406        handle.shutdown().await;
407        pump.abort();
408    }
409
410    #[tokio::test]
411    async fn a_foreign_host_or_origin_is_refused_before_the_token_is_read() {
412        let (handle, pump) = echoing().await;
413        let url = format!("http://127.0.0.1:{}/jobs", handle.addr.port());
414        let client = reqwest::Client::new();
415        let res = client
416            .get(&url)
417            .header(header::HOST, "evil.example")
418            .send()
419            .await
420            .expect("request");
421        assert_eq!(res.status().as_u16(), 403);
422        let res = client
423            .get(&url)
424            .bearer_auth("t0k3n")
425            .header(header::ORIGIN, "http://evil.example")
426            .send()
427            .await
428            .expect("request");
429        assert_eq!(res.status().as_u16(), 403);
430        let res = client
431            .get(&url)
432            .bearer_auth("t0k3n")
433            .header(header::ORIGIN, "http://localhost:5173")
434            .send()
435            .await
436            .expect("request");
437        assert_eq!(res.status().as_u16(), 201);
438        handle.shutdown().await;
439        pump.abort();
440    }
441}
agent_block_core/bus/http_source.rs

agent_block_core/bus/
http_source.rs
