iroh-http-core 0.4.0

Iroh QUIC endpoint, HTTP/1.1 over hyper, fetch/serve with FFI-friendly types
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275

//! FFI-shaped `fetch` — flat-string wrapper around the pure-Rust
//! [`crate::http::client::fetch_request`].
//!
//! Slice D (#186): translates the JS adapter calling convention into a
//! typed [`hyper::Request<Body>`] / [`iroh::EndpointAddr`] pair, hands
//! them to [`crate::http::client::fetch_request`], and re-packages the
//! response as a [`FfiResponse`] with a slotmap body handle. Maps
//! [`crate::http::client::FetchError`] variants onto [`crate::CoreError`]
//! codes for the FFI boundary.
// Legitimate FFI wiring — uses the disallowed types intentionally.
#![allow(clippy::disallowed_types)]

use std::time::Duration;

use http::{HeaderName, HeaderValue, Method};

use crate::{
    ffi::{handles::BodyReader, pumps::pump_hyper_body_to_channel_limited},
    http::client::{fetch_request, FetchError},
    parse_node_addr, Body, CoreError, FfiResponse, IrohEndpoint,
};

/// FFI-shaped fetch — re-exported as `iroh_http_core::fetch` for FFI
/// binary compatibility (Slice D acceptance #5). Composition mirrors the
/// pre-Slice-D function exactly; the moving parts that became pure Rust
/// live in [`crate::http::client::fetch_request`].
#[allow(clippy::too_many_arguments)]
pub async fn fetch(
    endpoint: &IrohEndpoint,
    remote_node_id: &str,
    url: &str,
    method: &str,
    headers: &[(String, String)],
    req_body_reader: Option<BodyReader>,
    fetch_token: Option<u64>,
    direct_addrs: Option<&[std::net::SocketAddr]>,
    timeout: Option<Duration>,
    decompress: bool,
    max_response_body_bytes: Option<usize>,
) -> Result<FfiResponse, CoreError> {
    // Reject standard web schemes.
    {
        let lower = url.to_ascii_lowercase();
        if lower.starts_with("https://") || lower.starts_with("http://") {
            let scheme_end = lower
                .find("://")
                .map(|i| i.saturating_add(3))
                .unwrap_or(lower.len());
            return Err(CoreError::invalid_input(format!(
                "iroh-http URLs must use the \"httpi://\" scheme, not \"{}\". \
                 Example: httpi://nodeId/path",
                &url[..scheme_end]
            )));
        }
    }

    // Validate method and headers at the FFI boundary.
    let http_method = Method::from_bytes(method.as_bytes())
        .map_err(|_| CoreError::invalid_input(format!("invalid HTTP method {:?}", method)))?;
    for (name, value) in headers {
        HeaderName::from_bytes(name.as_bytes())
            .map_err(|_| CoreError::invalid_input(format!("invalid header name {:?}", name)))?;
        HeaderValue::from_str(value).map_err(|_| {
            CoreError::invalid_input(format!("invalid header value for {:?}", name))
        })?;
    }

    // Resolve the EndpointAddr once: bare node id, ticket, or JSON, plus
    // any caller-supplied direct addresses.
    let parsed = parse_node_addr(remote_node_id)?;
    let mut addr = iroh::EndpointAddr::new(parsed.node_id);
    for a in &parsed.direct_addrs {
        addr = addr.with_ip_addr(*a);
    }
    if let Some(addrs) = direct_addrs {
        for a in addrs {
            addr = addr.with_ip_addr(*a);
        }
    }
    let remote_str = crate::base32_encode(parsed.node_id.as_bytes());
    let path = extract_path(url);

    // Build the typed Request<Body>.
    let mut req_builder = hyper::Request::builder()
        .method(http_method)
        .uri(&path)
        .header(hyper::header::HOST, &remote_str);

    // When compression is enabled, advertise zstd-only Accept-Encoding —
    // but only if the caller has not already set Accept-Encoding. A caller
    // passing `Accept-Encoding: identity` is opting out of compression and
    // must not be overridden.
    {
        let has_accept_encoding = headers
            .iter()
            .any(|(k, _)| k.eq_ignore_ascii_case("accept-encoding"));
        if !has_accept_encoding {
            req_builder = req_builder.header("accept-encoding", "zstd");
        }
    }
    for (k, v) in headers {
        req_builder = req_builder.header(k.as_str(), v.as_str());
    }

    let req_body: Body = if let Some(reader) = req_body_reader {
        Body::new(reader)
    } else {
        Body::empty()
    };
    let req = req_builder
        .body(req_body)
        .map_err(|e| CoreError::internal(format!("build request: {e}")))?;

    let cancel_notify = fetch_token.and_then(|t| endpoint.handles().get_fetch_cancel_notify(t));

    // Wire FFI-supplied knobs into the shared stack config.
    // - `timeout` bounds time-to-response-head inside `fetch_request`.
    // - `decompress` toggles the tower-http Decompression layer.
    // Per-frame body-read timeout and the response-body byte limit are
    // enforced below by `pump_hyper_body_to_channel_limited`; cancellation
    // is handled by the `tokio::select!` on the cancel-token notifier.
    let cfg = crate::http::server::stack::StackConfig {
        timeout,
        decompression: decompress,
        ..crate::http::server::stack::StackConfig::default()
    };

    let fetch_fut = fetch_request(endpoint, &addr, req, &cfg);
    let resp = match cancel_notify {
        Some(notify) => tokio::select! {
            _ = notify.notified() => {
                if let Some(t) = fetch_token {
                    endpoint.handles().remove_fetch_token(t);
                }
                return Err(CoreError::cancelled());
            }
            r = fetch_fut => r,
        },
        None => fetch_fut.await,
    };

    // Always clean up the cancellation token, even on error.
    if let Some(t) = fetch_token {
        endpoint.handles().remove_fetch_token(t);
    }

    let resp = resp.map_err(fetch_error_to_core)?;

    package_response(endpoint, resp, &remote_str, &path, max_response_body_bytes).await
}

/// Translate the typed [`FetchError`] surface into the flat
/// [`CoreError`] the FFI boundary expects.
fn fetch_error_to_core(e: FetchError) -> CoreError {
    match e {
        FetchError::ConnectionFailed { detail, .. } => CoreError::connection_failed(detail),
        FetchError::HeaderTooLarge { detail } => CoreError::header_too_large(detail),
        FetchError::BodyTooLarge => CoreError::body_too_large("response body too large"),
        FetchError::Timeout => CoreError::timeout("request timed out"),
        FetchError::Cancelled => CoreError::cancelled(),
        FetchError::Internal(msg) => CoreError::internal(msg),
    }
}

/// Extract the path portion from an `httpi://nodeId/path?query#frag` URL.
///
/// Lives next to [`fetch`] because that is the sole caller — it constructs
/// the request-target line for the outgoing HTTP/1.1 request.
pub(crate) fn extract_path(url: &str) -> String {
    if let Some(rest) = url.strip_prefix("httpi://") {
        if let Some(slash) = rest.find('/') {
            return rest[slash..].to_string();
        }
        return "/".to_string();
    }
    if url.starts_with('/') {
        return url.to_string();
    }
    format!("/{url}")
}

/// Validate response head, allocate body channel, and assemble [`FfiResponse`].
///
/// Lives in `ffi::fetch` because every step here exists to satisfy the
/// FFI contract: header-byte budget enforcement, RFC-9110 null-body
/// handling, slotmap body handle allocation, `Vec<(String, String)>`
/// header conversion. The pure-Rust caller would just consume
/// `resp.into_body()` directly.
async fn package_response(
    endpoint: &IrohEndpoint,
    resp: hyper::Response<Body>,
    remote_str: &str,
    path: &str,
    max_response_body_bytes: Option<usize>,
) -> Result<FfiResponse, CoreError> {
    let max_header_size = endpoint.max_header_size();
    // Per-call limit takes precedence; fall back to the endpoint-wide default.
    let max_response_body_bytes =
        max_response_body_bytes.unwrap_or_else(|| endpoint.max_response_body_bytes());
    let handles = endpoint.handles();

    let status = resp.status().as_u16();
    // ISS-011: measure header bytes using raw values before string conversion;
    // reject non-UTF8 response header values deterministically.
    let header_bytes: usize = resp
        .headers()
        .iter()
        .map(|(k, v)| {
            k.as_str()
                .len()
                .saturating_add(v.as_bytes().len())
                .saturating_add(4) // "name: value\r\n"
        })
        .fold(16usize, |acc, x| acc.saturating_add(x)); // approximate status line
    if header_bytes > max_header_size {
        return Err(CoreError::header_too_large(format!(
            "response header size {header_bytes} exceeds limit {max_header_size}"
        )));
    }

    let mut resp_headers: Vec<(String, String)> = Vec::new();
    for (k, v) in resp.headers().iter() {
        match v.to_str() {
            Ok(s) => resp_headers.push((k.as_str().to_string(), s.to_string())),
            Err(_) => {
                return Err(CoreError::invalid_input(format!(
                    "non-UTF8 response header value for '{}'",
                    k.as_str()
                )));
            }
        }
    }

    let response_url = format!("httpi://{remote_str}{path}");

    // RFC 9110 §6.3: responses with status 204, 205, or 304 MUST NOT carry a
    // message body. Skip channel allocation entirely and return the slotmap
    // null sentinel (0) for body_handle so the JS layer can use
    // `bodyHandle === 0n` as a clean structural check without re-encoding
    // HTTP semantics in every adapter.
    if matches!(status, 204 | 205 | 304) {
        // Dropping the body signals to hyper that we are done reading.
        // For a spec-compliant server the body is already empty; this is a
        // defensive drain for misbehaving peers.
        drop(resp.into_body());
        return Ok(FfiResponse {
            status,
            headers: resp_headers,
            body_handle: 0,
            url: response_url,
        });
    }

    // Allocate channels for streaming the response body to JS.
    let mut guard = handles.insert_guard();
    let (res_writer, res_reader) = handles.make_body_channel();
    let body = resp.into_body();
    let frame_timeout = res_writer.drain_timeout;
    tokio::spawn(pump_hyper_body_to_channel_limited(
        body,
        res_writer,
        Some(max_response_body_bytes),
        frame_timeout,
        None,
    ));

    let body_handle = guard.insert_reader(res_reader)?;
    guard.commit();
    Ok(FfiResponse {
        status,
        headers: resp_headers,
        body_handle,
        url: response_url,
    })
}