iroh_http_core/ffi/
fetch.rs

1//! FFI-shaped `fetch` — flat-string wrapper around the pure-Rust
2//! [`crate::http::client::fetch_request`].
3//!
4//! Slice D (#186): translates the JS adapter calling convention into a
5//! typed [`hyper::Request<Body>`] / [`iroh::EndpointAddr`] pair, hands
6//! them to [`crate::http::client::fetch_request`], and re-packages the
7//! response as a [`FfiResponse`] with a slotmap body handle. Maps
8//! [`crate::http::client::FetchError`] variants onto [`crate::CoreError`]
9//! codes for the FFI boundary.
10// Legitimate FFI wiring — uses the disallowed types intentionally.
11#![allow(clippy::disallowed_types)]
12
13use std::time::Duration;
14
15use http::{HeaderName, HeaderValue, Method};
16
17use crate::{
18    ffi::{handles::BodyReader, pumps::pump_hyper_body_to_channel_limited},
19    http::client::{fetch_request, FetchError},
20    parse_node_addr, Body, CoreError, FfiResponse, IrohEndpoint,
21};
22
23/// FFI-shaped fetch — re-exported as `iroh_http_core::fetch` for FFI
24/// binary compatibility (Slice D acceptance #5). Composition mirrors the
25/// pre-Slice-D function exactly; the moving parts that became pure Rust
26/// live in [`crate::http::client::fetch_request`].
27#[allow(clippy::too_many_arguments)]
28pub async fn fetch(
29    endpoint: &IrohEndpoint,
30    remote_node_id: &str,
31    url: &str,
32    method: &str,
33    headers: &[(String, String)],
34    req_body_reader: Option<BodyReader>,
35    fetch_token: Option<u64>,
36    direct_addrs: Option<&[std::net::SocketAddr]>,
37    timeout: Option<Duration>,
38    decompress: bool,
39    max_response_body_bytes: Option<usize>,
40) -> Result<FfiResponse, CoreError> {
41    // Reject standard web schemes.
42    {
43        let lower = url.to_ascii_lowercase();
44        if lower.starts_with("https://") || lower.starts_with("http://") {
45            let scheme_end = lower
46                .find("://")
47                .map(|i| i.saturating_add(3))
48                .unwrap_or(lower.len());
49            return Err(CoreError::invalid_input(format!(
50                "iroh-http URLs must use the \"httpi://\" scheme, not \"{}\". \
51                 Example: httpi://nodeId/path",
52                &url[..scheme_end]
53            )));
54        }
55    }
56
57    // Validate method and headers at the FFI boundary.
58    let http_method = Method::from_bytes(method.as_bytes())
59        .map_err(|_| CoreError::invalid_input(format!("invalid HTTP method {:?}", method)))?;
60    for (name, value) in headers {
61        HeaderName::from_bytes(name.as_bytes())
62            .map_err(|_| CoreError::invalid_input(format!("invalid header name {:?}", name)))?;
63        HeaderValue::from_str(value).map_err(|_| {
64            CoreError::invalid_input(format!("invalid header value for {:?}", name))
65        })?;
66    }
67
68    // Resolve the EndpointAddr once: bare node id, ticket, or JSON, plus
69    // any caller-supplied direct addresses.
70    let parsed = parse_node_addr(remote_node_id)?;
71    let mut addr = iroh::EndpointAddr::new(parsed.node_id);
72    for relay in parsed.relay_urls {
73        addr = addr.with_relay_url(relay);
74    }
75    for a in &parsed.direct_addrs {
76        addr = addr.with_ip_addr(*a);
77    }
78    if let Some(addrs) = direct_addrs {
79        for a in addrs {
80            addr = addr.with_ip_addr(*a);
81        }
82    }
83    let remote_str = crate::base32_encode(parsed.node_id.as_bytes());
84    let path = extract_path(url);
85
86    // Build the typed Request<Body>.
87    let mut req_builder = hyper::Request::builder()
88        .method(http_method)
89        .uri(&path)
90        .header(hyper::header::HOST, &remote_str);
91
92    // When compression is enabled, advertise zstd-only Accept-Encoding —
93    // but only if the caller has not already set Accept-Encoding. A caller
94    // passing `Accept-Encoding: identity` is opting out of compression and
95    // must not be overridden.
96    {
97        let has_accept_encoding = headers
98            .iter()
99            .any(|(k, _)| k.eq_ignore_ascii_case("accept-encoding"));
100        if !has_accept_encoding {
101            req_builder = req_builder.header("accept-encoding", "zstd");
102        }
103    }
104    for (k, v) in headers {
105        req_builder = req_builder.header(k.as_str(), v.as_str());
106    }
107
108    let req_body: Body = if let Some(reader) = req_body_reader {
109        Body::new(reader)
110    } else {
111        Body::empty()
112    };
113    let req = req_builder
114        .body(req_body)
115        .map_err(|e| CoreError::internal(format!("build request: {e}")))?;
116
117    // Self-request (ADR-015): a node fetching its own node id. iroh's
118    // transport forbids self-dial ("Connecting to ourself is not supported"),
119    // so route the request in-process to this node's own serve service instead
120    // of attempting a QUIC connection.
121    if remote_str == endpoint.node_id() {
122        // `decompress` is intentionally not forwarded: compression negotiation
123        // is a wire-only, per-connection layer, so it is a no-op on loopback.
124        return self_fetch(
125            endpoint,
126            req,
127            &remote_str,
128            &path,
129            fetch_token,
130            timeout,
131            max_response_body_bytes,
132        )
133        .await;
134    }
135
136    // Wire FFI-supplied knobs into the shared stack config.
137    // - `timeout` bounds time-to-response-head inside `fetch_request`.
138    // - `decompress` toggles the tower-http Decompression layer.
139    // Per-frame body-read timeout and the response-body byte limit are
140    // enforced below by `pump_hyper_body_to_channel_limited`; cancellation and
141    // token cleanup are centralized by `run_with_cancel_and_timeout`.
142    let cfg = crate::http::server::stack::StackConfig {
143        timeout,
144        decompression: decompress,
145        ..crate::http::server::stack::StackConfig::default()
146    };
147
148    let fetch_fut = async {
149        fetch_request(endpoint, &addr, req, &cfg)
150            .await
151            .map_err(fetch_error_to_core)
152    };
153    let resp = run_with_cancel_and_timeout(endpoint, fetch_token, None, fetch_fut).await?;
154
155    package_response(endpoint, resp, &remote_str, &path, max_response_body_bytes).await
156}
157
158/// In-process self-request — dispatch a `fetch()` to this node's own id
159/// directly to the locally-registered serve service (ADR-015).
160///
161/// iroh's transport refuses self-dial, so there is no QUIC connection: the
162/// request is handed to the exact same [`crate::ffi::dispatcher::IrohHttpService`]
163/// that remote peers reach, as an in-process `tower::Service` call. The node's
164/// own id is injected as the authenticated [`crate::http::server::RemoteNodeId`]
165/// — truthful, since the peer is us. Cancellation and `timeout` mirror the QUIC
166/// path; the response is packaged identically via [`package_response`].
167///
168/// This deliberately bypasses the wire: no QUIC/TLS handshake, no compression
169/// negotiation, and none of the per-connection server stack (timeouts and body
170/// limits applied at the accept loop). A self-request is therefore not a
171/// network-reachability check. See ADR-015 for the full semantics.
172#[allow(clippy::too_many_arguments)]
173async fn self_fetch(
174    endpoint: &IrohEndpoint,
175    mut req: hyper::Request<Body>,
176    remote_str: &str,
177    path: &str,
178    fetch_token: Option<u64>,
179    timeout: Option<Duration>,
180    max_response_body_bytes: Option<usize>,
181) -> Result<FfiResponse, CoreError> {
182    use tower::ServiceExt;
183
184    let svc = endpoint.local_service().ok_or_else(|| {
185        CoreError::connection_failed(
186            "self-request: this node has no active server to handle a request to its \
187             own node id. Call serve() before fetching httpi://<your-own-node-id>/…",
188        )
189    })?;
190
191    // The QUIC path receives the authenticated peer id from the per-connection
192    // AddExtensionLayer; in-process we inject it directly so the serve handler
193    // still sees a truthful `Peer-Id` (our own id).
194    req.extensions_mut()
195        .insert(crate::http::server::RemoteNodeId(std::sync::Arc::new(
196            remote_str.to_string(),
197        )));
198
199    // `IrohHttpService` is `Infallible`; dispatch and await the response head.
200    let dispatch = async move {
201        match svc.oneshot(req).await {
202            Ok(resp) => resp,
203            Err(never) => match never {},
204        }
205    };
206
207    let dispatch_fut = async { Ok::<_, CoreError>(dispatch.await) };
208    let resp = run_with_cancel_and_timeout(endpoint, fetch_token, timeout, dispatch_fut).await?;
209
210    package_response(endpoint, resp, remote_str, path, max_response_body_bytes).await
211}
212
213/// Drive `fut` to completion while honoring an optional fetch-cancel token and
214/// an optional hard timeout, and ALWAYS remove the fetch-cancel token on every
215/// exit (success, error, timeout, or cancel). Centralizes the token-leak
216/// guarantee shared by the wire `fetch` and `self_fetch` loopback paths.
217async fn run_with_cancel_and_timeout<F, T>(
218    endpoint: &IrohEndpoint,
219    fetch_token: Option<u64>,
220    timeout: Option<Duration>,
221    fut: F,
222) -> Result<T, CoreError>
223where
224    F: std::future::Future<Output = Result<T, CoreError>>,
225{
226    let cancel_notify = fetch_token.and_then(|t| endpoint.handles().get_fetch_cancel_notify(t));
227    let timed = async {
228        match timeout {
229            Some(t) => tokio::time::timeout(t, fut)
230                .await
231                .map_err(|_| CoreError::timeout("request timed out"))?,
232            None => fut.await,
233        }
234    };
235    let result = match cancel_notify {
236        Some(notify) => {
237            tokio::select! {
238                _ = notify.notified() => Err(CoreError::cancelled()),
239                r = timed => r,
240            }
241        }
242        None => timed.await,
243    };
244    if let Some(t) = fetch_token {
245        endpoint.handles().remove_fetch_token(t);
246    }
247    result
248}
249
250/// Translate the typed [`FetchError`] surface into the flat
251/// [`CoreError`] the FFI boundary expects.
252fn fetch_error_to_core(e: FetchError) -> CoreError {
253    match e {
254        FetchError::ConnectionFailed { detail, .. } => CoreError::connection_failed(detail),
255        FetchError::RequestBodyFailed { detail, .. } => {
256            CoreError::internal(format!("request body failed: {detail}"))
257        }
258        FetchError::HeaderTooLarge { detail } => CoreError::header_too_large(detail),
259        FetchError::BodyTooLarge => CoreError::body_too_large("response body too large"),
260        FetchError::Timeout => CoreError::timeout("request timed out"),
261        FetchError::Cancelled => CoreError::cancelled(),
262        FetchError::Internal(msg) => CoreError::internal(msg),
263    }
264}
265
266/// Extract the path portion from an `httpi://nodeId/path?query#frag` URL.
267///
268/// Lives next to [`fetch`] because that is the sole caller — it constructs
269/// the request-target line for the outgoing HTTP/1.1 request.
270pub(crate) fn extract_path(url: &str) -> String {
271    // The fragment is client-local (RFC 3986 §3.5) and must never appear on
272    // the wire. Drop everything from the first '#' before extracting the path,
273    // as defense-in-depth in case a caller bypasses the JS-side stripping.
274    let url = url.split('#').next().unwrap_or(url);
275    if let Some(rest) = url.strip_prefix("httpi://") {
276        if let Some(slash) = rest.find('/') {
277            return rest[slash..].to_string();
278        }
279        return "/".to_string();
280    }
281    if url.starts_with('/') {
282        return url.to_string();
283    }
284    format!("/{url}")
285}
286
287/// Validate response head, allocate body channel, and assemble [`FfiResponse`].
288///
289/// Lives in `ffi::fetch` because every step here exists to satisfy the
290/// FFI contract: header-byte budget enforcement, RFC-9110 null-body
291/// handling, slotmap body handle allocation, `Vec<(String, String)>`
292/// header conversion. The pure-Rust caller would just consume
293/// `resp.into_body()` directly.
294async fn package_response(
295    endpoint: &IrohEndpoint,
296    resp: hyper::Response<Body>,
297    remote_str: &str,
298    path: &str,
299    max_response_body_bytes: Option<usize>,
300) -> Result<FfiResponse, CoreError> {
301    let max_header_size = endpoint.max_header_size();
302    // Per-call limit takes precedence; fall back to the endpoint-wide default.
303    let max_response_body_bytes =
304        max_response_body_bytes.unwrap_or_else(|| endpoint.max_response_body_bytes());
305    let handles = endpoint.handles();
306
307    let status = resp.status().as_u16();
308    // ISS-011: measure header bytes using raw values before string conversion;
309    // reject non-UTF8 response header values deterministically.
310    let header_bytes: usize = resp
311        .headers()
312        .iter()
313        .map(|(k, v)| {
314            k.as_str()
315                .len()
316                .saturating_add(v.as_bytes().len())
317                .saturating_add(4) // "name: value\r\n"
318        })
319        .fold(16usize, |acc, x| acc.saturating_add(x)); // approximate status line
320    if header_bytes > max_header_size {
321        return Err(CoreError::header_too_large(format!(
322            "response header size {header_bytes} exceeds limit {max_header_size}"
323        )));
324    }
325
326    let mut resp_headers: Vec<(String, String)> = Vec::new();
327    for (k, v) in resp.headers().iter() {
328        match v.to_str() {
329            Ok(s) => resp_headers.push((k.as_str().to_string(), s.to_string())),
330            Err(_) => {
331                return Err(CoreError::invalid_input(format!(
332                    "non-UTF8 response header value for '{}'",
333                    k.as_str()
334                )));
335            }
336        }
337    }
338
339    let response_url = format!("httpi://{remote_str}{path}");
340
341    // RFC 9110 §6.3: responses with status 204, 205, or 304 MUST NOT carry a
342    // message body. Skip channel allocation entirely and return the slotmap
343    // null sentinel (0) for body_handle so the JS layer can use
344    // `bodyHandle === 0n` as a clean structural check without re-encoding
345    // HTTP semantics in every adapter.
346    if matches!(status, 204 | 205 | 304) {
347        // Dropping the body signals to hyper that we are done reading.
348        // For a spec-compliant server the body is already empty; this is a
349        // defensive drain for misbehaving peers.
350        drop(resp.into_body());
351        return Ok(FfiResponse {
352            status,
353            headers: resp_headers,
354            body_handle: 0,
355            url: response_url,
356        });
357    }
358
359    // Allocate channels for streaming the response body to JS.
360    let mut guard = handles.insert_guard();
361    let (res_writer, res_reader) = handles.make_body_channel();
362    let body = resp.into_body();
363    let frame_timeout = res_writer.drain_timeout;
364    tokio::spawn(pump_hyper_body_to_channel_limited(
365        body,
366        res_writer,
367        Some(max_response_body_bytes),
368        frame_timeout,
369        None,
370    ));
371
372    let body_handle = guard.insert_reader(res_reader)?;
373    guard.commit();
374    Ok(FfiResponse {
375        status,
376        headers: resp_headers,
377        body_handle,
378        url: response_url,
379    })
380}
381
382#[cfg(test)]
383mod tests {
384    use super::extract_path;
385
386    #[test]
387    fn extracts_path_and_query() {
388        assert_eq!(extract_path("httpi://node/a/b?c=1"), "/a/b?c=1");
389        assert_eq!(extract_path("httpi://node/"), "/");
390        assert_eq!(extract_path("httpi://node"), "/");
391    }
392
393    #[test]
394    fn strips_fragment_from_request_target() {
395        // Fragments are client-local (RFC 3986 §3.5) and must never reach the
396        // peer, even if a caller bypasses the JS-side stripping.
397        assert_eq!(extract_path("httpi://node/a?b=1#secret"), "/a?b=1");
398        assert_eq!(extract_path("httpi://node/a#frag"), "/a");
399        assert_eq!(extract_path("httpi://node/#frag"), "/");
400        assert_eq!(extract_path("httpi://node#frag"), "/");
401        assert_eq!(extract_path("/path?q=1#frag"), "/path?q=1");
402    }
403}
iroh_http_core/ffi/fetch.rs

iroh_http_core/ffi/
fetch.rs
