iroh_http_core/ffi/
dispatcher.rs

1//! FFI shaped serve + respond — the JS-callback bridge over the
2//! pure-Rust [`crate::http::server::serve_with_events`] entry.
3//!
4//! Per epic #182 this module is the **only** place that allocates `u64`
5//! handles, fires the [`RequestPayload`] callback, and rendezvous on the
6//! response head from JS. Everything below is one specific implementation
7//! of the generic `tower::Service<Request<Body>>` accepted by the pure-Rust
8//! serve entry — handed in via [`ffi_serve_with_callback`].
9//!
10//! Architecture-test guarantee: this file may import from `crate::http`
11// Legitimate FFI wiring — uses the disallowed types intentionally.
12#![allow(clippy::disallowed_types)]
13//! (the `ffi → http` direction is allowed); `crate::http::*` must NOT
14//! import from here.
15
16use std::convert::Infallible;
17use std::future::Future;
18use std::pin::Pin;
19use std::sync::Arc;
20use std::task::{Context, Poll};
21
22use bytes::Bytes;
23use http::StatusCode;
24use tower::Service;
25
26use crate::ffi::handles::ResponseHeadEntry;
27use crate::ffi::pumps::pump_hyper_body_to_channel;
28use crate::http::server::{
29    options::DEFAULT_MAX_REQUEST_BODY_BYTES, serve_with_events, ConnectionEventFn, RemoteNodeId,
30    ServeHandle, ServeOptions,
31};
32use crate::{Body, CoreError, IrohEndpoint, RequestPayload};
33
34// ── Inline error responses ─────────────────────────────────────────────────
35
36fn internal_error(detail: &'static [u8]) -> hyper::Response<Body> {
37    hyper::Response::builder()
38        .status(StatusCode::INTERNAL_SERVER_ERROR)
39        .body(Body::full(Bytes::from_static(detail)))
40        .expect("static error response args are valid")
41}
42
43fn service_unavailable(detail: &'static [u8]) -> hyper::Response<Body> {
44    hyper::Response::builder()
45        .status(StatusCode::SERVICE_UNAVAILABLE)
46        .body(Body::full(Bytes::from_static(detail)))
47        .expect("static error response args are valid")
48}
49
50// ── respond() ─────────────────────────────────────────────────────────────
51
52/// Send the response head for a request handle previously delivered to JS
53/// via the [`RequestPayload::req_handle`] callback.
54///
55/// Partner of the `head_rx` rendezvous in [`FfiDispatcher::dispatch`]. The
56/// JS handler calls this exactly once per request to provide the status +
57/// headers; subsequent body bytes flow through the response body channel
58/// referenced by [`RequestPayload::res_body_handle`].
59pub fn respond(
60    handles: &crate::ffi::handles::HandleStore,
61    req_handle: u64,
62    status: u16,
63    headers: Vec<(String, String)>,
64) -> Result<(), CoreError> {
65    StatusCode::from_u16(status)
66        .map_err(|_| CoreError::invalid_input(format!("invalid HTTP status code: {status}")))?;
67    for (name, value) in &headers {
68        http::HeaderName::from_bytes(name.as_bytes()).map_err(|_| {
69            CoreError::invalid_input(format!("invalid response header name {:?}", name))
70        })?;
71        http::HeaderValue::from_str(value).map_err(|_| {
72            CoreError::invalid_input(format!("invalid response header value for {:?}", name))
73        })?;
74    }
75
76    let sender = handles
77        .take_req_sender(req_handle)
78        .ok_or_else(|| CoreError::invalid_handle(req_handle))?;
79    sender
80        .send(ResponseHeadEntry { status, headers })
81        .map_err(|_| CoreError::internal("serve task dropped before respond"))
82}
83
84// ── ReqHeadGuard ──────────────────────────────────────────────────────────
85//
86// RAII guard that removes the per-request slab entry from the handle
87// store on every dispatch exit path (success, error, panic). Lifted to
88// module scope from inline-in-`dispatch` per Slice C.4 of #182 (#178).
89
90struct ReqHeadGuard {
91    endpoint: IrohEndpoint,
92    req_handle: u64,
93}
94
95impl Drop for ReqHeadGuard {
96    fn drop(&mut self) {
97        self.endpoint.handles().take_req_sender(self.req_handle);
98    }
99}
100
101// ── FfiDispatcher + IrohHttpService ───────────────────────────────────────
102
103/// FFI-shaped tower service: allocates request/response handles, fires the
104/// JS `on_request` callback, and rendezvous on the response head sent via
105/// [`respond`]. Shared across every accepted connection and request via
106/// `Arc`.
107pub(crate) struct FfiDispatcher {
108    on_request: Arc<dyn Fn(RequestPayload) + Send + Sync>,
109    endpoint: IrohEndpoint,
110    own_node_id: Arc<String>,
111    max_header_size: Option<usize>,
112    max_request_body_wire_bytes: Option<usize>,
113}
114
115#[derive(Clone)]
116pub(crate) struct IrohHttpService {
117    dispatcher: Arc<FfiDispatcher>,
118}
119
120impl IrohHttpService {
121    /// Downgrade to a Weak handle for the self-request slot (#282): the
122    /// endpoint stores only a Weak so it cannot form a strong reference
123    /// cycle (endpoint → service → dispatcher → endpoint).
124    pub(crate) fn downgrade(&self) -> std::sync::Weak<FfiDispatcher> {
125        std::sync::Arc::downgrade(&self.dispatcher)
126    }
127
128    /// Reconstruct a live service from a stored Weak, or None if the
129    /// serve task has been torn down (the "no active server" path).
130    pub(crate) fn upgrade(weak: &std::sync::Weak<FfiDispatcher>) -> Option<Self> {
131        weak.upgrade()
132            .map(|dispatcher| IrohHttpService { dispatcher })
133    }
134}
135
136/// ADR-014 D2 / #175: service is concrete — not generic over `B`.
137/// Body normalisation happens upstream at the hyper → tower seam in
138/// `crate::http::server::pipeline::serve_bistream`.
139impl Service<hyper::Request<Body>> for IrohHttpService {
140    type Response = hyper::Response<Body>;
141    type Error = Infallible;
142    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
143
144    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
145        Poll::Ready(Ok(()))
146    }
147
148    fn call(&mut self, req: hyper::Request<Body>) -> Self::Future {
149        let dispatcher = self.dispatcher.clone();
150        // #177: the authenticated peer id arrives as a request extension
151        // inserted by the per-connection AddExtensionLayer in
152        // serve_with_events. Missing extension is a server-side
153        // bug (the layer is unconditional) — fall back to empty string.
154        let remote_node_id = req
155            .extensions()
156            .get::<RemoteNodeId>()
157            .map(|r| r.0.clone())
158            .unwrap_or_else(|| Arc::new(String::new()));
159        Box::pin(async move { Ok(dispatcher.dispatch(req, remote_node_id).await) })
160    }
161}
162
163impl FfiDispatcher {
164    async fn dispatch(
165        self: Arc<Self>,
166        req: hyper::Request<Body>,
167        remote_node_id: Arc<String>,
168    ) -> hyper::Response<Body> {
169        let handles = self.endpoint.handles();
170        let own_node_id = &*self.own_node_id;
171        let max_header_size = self.max_header_size;
172        let max_request_body_wire_bytes = self.max_request_body_wire_bytes;
173
174        let method = req.method().to_string();
175        let path_and_query = req
176            .uri()
177            .path_and_query()
178            .map(|p| p.as_str())
179            .unwrap_or("/")
180            .to_string();
181
182        // Log the path only, never the query string: query parameters commonly
183        // carry tokens or sensitive identifiers that must not leak into logs.
184        let path_only = req.uri().path();
185        tracing::debug!(
186            method = %method,
187            path = %path_only,
188            peer = %remote_node_id,
189            "iroh-http: incoming request",
190        );
191
192        // Strip any client-supplied peer-id to prevent spoofing,
193        // then inject the authenticated identity from the QUIC connection.
194        //
195        // ISS-011: Use raw byte length for header-size accounting to prevent
196        // bypass via non-UTF8 values.  Reject non-UTF8 header values with 400
197        // instead of silently converting them to empty strings.
198
199        // First pass: measure header bytes using raw values (before lossy conversion).
200        if let Some(limit) = max_header_size {
201            let header_bytes: usize = req
202                .headers()
203                .iter()
204                .filter(|(k, _)| !k.as_str().eq_ignore_ascii_case("peer-id"))
205                .map(|(k, v)| {
206                    k.as_str()
207                        .len()
208                        .saturating_add(v.as_bytes().len())
209                        .saturating_add(4)
210                }) // ": " + "\r\n"
211                .fold(0usize, |acc, x| acc.saturating_add(x))
212                .saturating_add("peer-id".len())
213                .saturating_add(remote_node_id.len())
214                .saturating_add(4)
215                .saturating_add(req.uri().to_string().len())
216                .saturating_add(method.len())
217                .saturating_add(12); // "HTTP/1.1 \r\n\r\n" overhead
218            if header_bytes > limit {
219                let resp = hyper::Response::builder()
220                    .status(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE)
221                    .body(Body::empty())
222                    .expect("static response args are valid");
223                return resp;
224            }
225        }
226
227        // Build header list — reject non-UTF8 values instead of silently dropping.
228        let mut req_headers: Vec<(String, String)> = Vec::new();
229        for (k, v) in req.headers().iter() {
230            if k.as_str().eq_ignore_ascii_case("peer-id") {
231                continue;
232            }
233            match v.to_str() {
234                Ok(s) => req_headers.push((k.as_str().to_string(), s.to_string())),
235                Err(_) => {
236                    let resp = hyper::Response::builder()
237                        .status(StatusCode::BAD_REQUEST)
238                        .body(Body::full(Bytes::from_static(b"non-UTF8 header value")))
239                        .expect("static response args are valid");
240                    return resp;
241                }
242            }
243        }
244        req_headers.push(("peer-id".to_string(), (*remote_node_id).clone()));
245
246        if let (Some(limit), Some(content_length)) = (
247            max_request_body_wire_bytes,
248            req.headers().get(hyper::header::CONTENT_LENGTH),
249        ) {
250            match content_length
251                .to_str()
252                .ok()
253                .and_then(|v| v.parse::<u64>().ok())
254            {
255                Some(len) if len > limit as u64 => {
256                    drop(req.into_body());
257                    return hyper::Response::builder()
258                        .status(StatusCode::PAYLOAD_TOO_LARGE)
259                        .body(Body::empty())
260                        .expect("static response args are valid");
261                }
262                _ => {}
263            }
264        }
265
266        let url = format!("httpi://{own_node_id}{path_and_query}");
267
268        // ── Allocate channels ────────────────────────────────────────────────
269
270        let mut guard = handles.insert_guard();
271        let (req_body_writer, req_body_reader) = handles.make_body_channel();
272        let req_body_handle = match guard.insert_reader(req_body_reader) {
273            Ok(h) => h,
274            Err(_) => return service_unavailable(b"server handle table full"),
275        };
276
277        let (res_body_writer, res_body_reader) = handles.make_body_channel();
278        let res_body_handle = match guard.insert_writer(res_body_writer) {
279            Ok(h) => h,
280            Err(_) => return service_unavailable(b"server handle table full"),
281        };
282
283        let (head_tx, head_rx) = tokio::sync::oneshot::channel::<ResponseHeadEntry>();
284        let req_handle = match guard.allocate_req_handle(head_tx) {
285            Ok(h) => h,
286            Err(_) => return service_unavailable(b"server handle table full"),
287        };
288
289        guard.commit();
290
291        let _req_head_guard = ReqHeadGuard {
292            endpoint: self.endpoint.clone(),
293            req_handle,
294        };
295
296        // ── Pump request body ────────────────────────────────────────────────
297
298        let body = req.into_body();
299        tokio::spawn(pump_hyper_body_to_channel(body, req_body_writer));
300
301        // ── Fire on_request callback ─────────────────────────────────────────
302
303        (self.on_request)(RequestPayload {
304            req_handle,
305            req_body_handle,
306            res_body_handle,
307            method,
308            url,
309            headers: req_headers,
310            remote_node_id: Arc::unwrap_or_clone(remote_node_id),
311            is_bidi: false,
312        });
313
314        // ── Await response head from JS ──────────────────────────────────────
315        let response_head = match head_rx.await {
316            Ok(h) => h,
317            Err(_) => return internal_error(b"JS handler dropped without responding"),
318        };
319
320        // ── Regular HTTP response ─────────────────────────────────────────────
321
322        let mut resp_builder = hyper::Response::builder().status(response_head.status);
323        for (k, v) in &response_head.headers {
324            resp_builder = resp_builder.header(k.as_str(), v.as_str());
325        }
326
327        match resp_builder.body(Body::new(res_body_reader)) {
328            Ok(r) => r,
329            Err(_) => internal_error(b"failed to build response head from JS"),
330        }
331    }
332}
333
334// ── ffi_serve_with_callback ───────────────────────────────────────────────
335
336/// FFI-shaped serve entry. Constructs an [`IrohHttpService`] around the
337/// supplied callback and delegates to the pure-Rust
338/// [`crate::http::server::serve_with_events`].
339///
340/// `on_connection_event` is called on 0→1 (first connection from a peer)
341/// and 1→0 (last connection from a peer closed) count transitions.
342///
343/// # Security
344///
345/// Calling this opens a **public endpoint** on the Iroh overlay network.
346/// Any peer that knows or discovers your node's public key can connect
347/// and send requests. Iroh QUIC authenticates the peer's *identity*
348/// cryptographically, but does not enforce *authorization*. Always
349/// inspect [`RequestPayload::remote_node_id`] and reject untrusted peers.
350pub fn ffi_serve_with_callback<F>(
351    endpoint: IrohEndpoint,
352    options: ServeOptions,
353    on_request: F,
354    on_connection_event: Option<ConnectionEventFn>,
355) -> ServeHandle
356where
357    F: Fn(RequestPayload) + Send + Sync + 'static,
358{
359    let max_header_size = endpoint.max_header_size();
360    let own_node_id = Arc::new(endpoint.node_id().to_string());
361    let on_request = Arc::new(on_request) as Arc<dyn Fn(RequestPayload) + Send + Sync>;
362
363    let dispatcher = Arc::new(FfiDispatcher {
364        on_request,
365        endpoint: endpoint.clone(),
366        own_node_id,
367        max_header_size: if max_header_size == 0 {
368            None
369        } else {
370            Some(max_header_size)
371        },
372        max_request_body_wire_bytes: options
373            .max_request_body_wire_bytes
374            .or(Some(DEFAULT_MAX_REQUEST_BODY_BYTES)),
375    });
376    let svc = IrohHttpService { dispatcher };
377
378    let handle = serve_with_events(endpoint.clone(), options, svc.clone(), on_connection_event);
379
380    // Register the service for in-process self-requests (ADR-015) against the
381    // exact token the common server path already installed. If stop_serve won
382    // the race after registration, the stopped token rejects this late service
383    // install instead of reviving a half-live local router.
384    endpoint.set_local_service_for(&handle, &svc);
385    handle
386}
387
388/// Back-compat 3-arg FFI serve entry: equivalent to
389/// [`ffi_serve_with_callback`] with `on_connection_event = None`. The Node /
390/// Deno / Tauri adapters call this directly.
391pub fn ffi_serve<F>(endpoint: IrohEndpoint, options: ServeOptions, on_request: F) -> ServeHandle
392where
393    F: Fn(RequestPayload) + Send + Sync + 'static,
394{
395    ffi_serve_with_callback(endpoint, options, on_request, None)
396}
397
398#[cfg(test)]
399mod tests {
400    use super::*;
401    use crate::ffi::handles::{HandleStore, StoreConfig};
402    use crate::{NetworkingOptions, NodeOptions};
403    use http::HeaderValue;
404    use std::sync::Mutex;
405
406    // ── respond(): validation + head delivery ────────────────────────────
407
408    #[tokio::test]
409    async fn respond_rejects_invalid_status() {
410        let store = HandleStore::new(StoreConfig::default());
411        // 9999 is outside the valid 100..=999 status range.
412        let err = respond(&store, 0, 9999, vec![]).unwrap_err();
413        assert_eq!(err.code, crate::ErrorCode::InvalidInput);
414    }
415
416    #[tokio::test]
417    async fn respond_rejects_invalid_header_name() {
418        let store = HandleStore::new(StoreConfig::default());
419        let err = respond(&store, 0, 200, vec![("bad name".into(), "v".into())]).unwrap_err();
420        assert_eq!(err.code, crate::ErrorCode::InvalidInput);
421    }
422
423    #[tokio::test]
424    async fn respond_rejects_invalid_header_value() {
425        let store = HandleStore::new(StoreConfig::default());
426        // A newline in a header value is rejected (header injection guard).
427        let err = respond(&store, 0, 200, vec![("x".into(), "bad\nvalue".into())]).unwrap_err();
428        assert_eq!(err.code, crate::ErrorCode::InvalidInput);
429    }
430
431    #[tokio::test]
432    async fn respond_unknown_handle_errors() {
433        let store = HandleStore::new(StoreConfig::default());
434        // Status and headers are valid, but no such request handle exists.
435        let err = respond(&store, 4242, 200, vec![("x".into(), "y".into())]).unwrap_err();
436        assert_eq!(err.code, crate::ErrorCode::InvalidInput);
437        assert!(err.message.contains("4242"));
438    }
439
440    #[tokio::test]
441    async fn respond_delivers_head_to_receiver() {
442        let store = HandleStore::new(StoreConfig::default());
443        let (tx, rx) = tokio::sync::oneshot::channel::<ResponseHeadEntry>();
444        let handle = store.allocate_req_handle(tx).unwrap();
445
446        respond(&store, handle, 201, vec![("x-test".into(), "yes".into())]).unwrap();
447
448        let head = rx.await.unwrap();
449        assert_eq!(head.status, 201);
450        assert_eq!(
451            head.headers,
452            vec![("x-test".to_string(), "yes".to_string())]
453        );
454        // The sender is consumed: a second respond on the same handle fails.
455        let err = respond(&store, handle, 200, vec![]).unwrap_err();
456        assert_eq!(err.code, crate::ErrorCode::InvalidInput);
457    }
458
459    // ── dispatch(): malformed-request rejection (ISS-011 / ISS-003) ───────
460
461    async fn local_endpoint() -> IrohEndpoint {
462        IrohEndpoint::bind(NodeOptions {
463            networking: NetworkingOptions {
464                disabled: true,
465                bind_addrs: vec!["127.0.0.1:0".into()],
466                ..Default::default()
467            },
468            ..Default::default()
469        })
470        .await
471        .unwrap()
472    }
473
474    fn make_dispatcher(
475        endpoint: IrohEndpoint,
476        max_header_size: Option<usize>,
477        sink: Arc<Mutex<Vec<RequestPayload>>>,
478    ) -> Arc<FfiDispatcher> {
479        let own_node_id = Arc::new(endpoint.node_id().to_string());
480        Arc::new(FfiDispatcher {
481            on_request: Arc::new(move |p| sink.lock().unwrap().push(p)),
482            endpoint,
483            own_node_id,
484            max_header_size,
485            max_request_body_wire_bytes: None,
486        })
487    }
488
489    /// ISS-011: a non-UTF8 header value must be rejected with 400, not
490    /// silently coerced, and the JS callback must never fire.
491    #[tokio::test]
492    async fn dispatch_rejects_non_utf8_header_value() {
493        let endpoint = local_endpoint().await;
494        let sink = Arc::new(Mutex::new(Vec::new()));
495        let dispatcher = make_dispatcher(endpoint, None, sink.clone());
496
497        let mut req = hyper::Request::builder()
498            .method("GET")
499            .uri("/")
500            .body(Body::empty())
501            .unwrap();
502        req.headers_mut()
503            .insert("x-bad", HeaderValue::from_bytes(&[0xff, 0xfe]).unwrap());
504
505        let resp = dispatcher.dispatch(req, Arc::new(String::new())).await;
506        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
507        assert!(
508            sink.lock().unwrap().is_empty(),
509            "callback must not fire on a rejected request",
510        );
511    }
512
513    /// ISS-003: request headers exceeding `max_header_size` are rejected with
514    /// 431 before any handle is allocated or the callback fires.
515    #[tokio::test]
516    async fn dispatch_rejects_oversized_headers() {
517        let endpoint = local_endpoint().await;
518        let sink = Arc::new(Mutex::new(Vec::new()));
519        let dispatcher = make_dispatcher(endpoint, Some(50), sink.clone());
520
521        let req = hyper::Request::builder()
522            .method("GET")
523            .uri("/")
524            .header("x-big", "a".repeat(200))
525            .body(Body::empty())
526            .unwrap();
527
528        let resp = dispatcher.dispatch(req, Arc::new(String::new())).await;
529        assert_eq!(resp.status(), StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE);
530        assert!(
531            sink.lock().unwrap().is_empty(),
532            "callback must not fire on a rejected request",
533        );
534    }
535}
iroh_http_core/ffi/dispatcher.rs

iroh_http_core/ffi/
dispatcher.rs
