toolkit_contract/runtime/
http.rs

1//! HTTP helpers shared by the generated REST client codegen.
2//!
3//! These are intentionally low-level and provider-agnostic so that the macro
4//! output stays small and the helpers can be unit-tested in isolation.
5
6use bytes::Bytes;
7use futures_core::Stream;
8use http_body::Body;
9use http_body_util::BodyStream;
10use percent_encoding::{AsciiSet, CONTROLS, utf8_percent_encode};
11use toolkit_canonical_errors::Problem;
12use toolkit_http::RequestBuilder;
13
14use crate::ir::binding::{HttpFieldBinding, HttpMethod, HttpMethodBindingIr};
15use crate::runtime::config::InternalTokenProvider;
16use crate::runtime::transport_error::TransportError;
17
18// RFC 3986 path-segment encode set: encode everything except unreserved
19// characters (`A-Z a-z 0-9 - . _ ~`).
20const PATH_SEGMENT: &AsciiSet = &CONTROLS
21    .add(b' ')
22    .add(b'"')
23    .add(b'#')
24    .add(b'%')
25    .add(b'&')
26    .add(b'\'')
27    .add(b'(')
28    .add(b')')
29    .add(b'*')
30    .add(b'+')
31    .add(b',')
32    .add(b'/')
33    .add(b':')
34    .add(b';')
35    .add(b'<')
36    .add(b'=')
37    .add(b'>')
38    .add(b'?')
39    .add(b'@')
40    .add(b'[')
41    .add(b'\\')
42    .add(b']')
43    .add(b'^')
44    .add(b'`')
45    .add(b'{')
46    .add(b'|')
47    .add(b'}');
48
49/// Adapt any `http_body::Body` into a `Stream<Item = Result<Bytes, E>>` of
50/// data frames, dropping trailers.
51///
52/// `toolkit_http::HttpResponse::into_body()` returns a `ResponseBody` that
53/// implements [`http_body::Body`] but the SSE parser
54/// ([`crate::runtime::sse::parse_sse_stream_with_id`]) expects a flat
55/// `Stream` of byte chunks. SSE has no use for trailers, so non-data frames
56/// are simply skipped.
57pub fn body_to_byte_stream<B>(body: B) -> impl Stream<Item = Result<Bytes, B::Error>> + Send
58where
59    B: Body<Data = Bytes> + Send + 'static,
60    B::Error: Send + 'static,
61{
62    use futures_util::StreamExt as _;
63    BodyStream::new(body).filter_map(|frame_res| async move {
64        match frame_res {
65            Ok(frame) => frame.into_data().ok().map(Ok),
66            Err(e) => Some(Err(e)),
67        }
68    })
69}
70
71/// Maximum error-body prefix buffered when classifying a non-success streaming
72/// open. The body only feeds a diagnostic — an RFC 9457 [`Problem`], or a
73/// truncated `HttpStatus.body` — so a short prefix is all it is ever used for,
74/// mirroring toolkit-http's own `ERROR_BODY_PREVIEW_LIMIT`.
75pub(crate) const ERROR_BODY_PREVIEW_LIMIT: usize = 8 * 1024;
76
77/// Read at most [`ERROR_BODY_PREVIEW_LIMIT`] bytes of `body`'s data frames,
78/// abandoning the rest of the body unread once the cap is reached.
79///
80/// The streaming open's error path needs only a short prefix to build its
81/// diagnostic. `HttpResponse::bytes()` would instead buffer the whole body up
82/// to the client's `max_body_size` (megabytes by default), so a peer could make
83/// a failed open allocate far more than the message ever uses. Capping the read
84/// itself — not merely truncating the resulting string — is what bounds that
85/// allocation. A transport error encountered before the cap surfaces as `Err`.
86pub(crate) async fn read_error_body_prefix<B>(body: B) -> Result<Bytes, B::Error>
87where
88    B: Body<Data = Bytes>,
89{
90    use http_body_util::BodyExt as _;
91
92    let mut body = std::pin::pin!(body);
93    let mut buf: Vec<u8> = Vec::new();
94    while buf.len() < ERROR_BODY_PREVIEW_LIMIT {
95        let Some(frame) = body.frame().await else {
96            break;
97        };
98        if let Ok(data) = frame?.into_data() {
99            let take = (ERROR_BODY_PREVIEW_LIMIT - buf.len()).min(data.len());
100            buf.extend_from_slice(&data[..take]);
101        }
102    }
103    Ok(Bytes::from(buf))
104}
105
106/// Build a fully-qualified URL by substituting path parameters and appending a
107/// pre-encoded query string, returning [`TransportError`] on failure.
108///
109/// `fields` is expected to be a JSON object whose keys correspond to the
110/// `field` names referenced by `method_binding.field_bindings`. Missing path
111/// parameters yield [`TransportError::UrlBuild`].
112///
113/// `query` is the already-encoded query string (no leading `?`), produced by
114/// [`crate::query::to_query_string`]. It arrives pre-encoded rather than as
115/// structured data on purpose: the server decodes it with the same
116/// `serde_html_form` codec, and routing it through this function's own
117/// serializer is what previously let the two ends disagree on `Vec` and nested
118/// shapes.
119///
120/// # Errors
121/// Returns [`TransportError::UrlBuild`] when a required path parameter is missing,
122/// null, or empty, or when a referenced field is not convertible to a string.
123pub fn build_request_url(
124    base_url: &str,
125    base_path: &str,
126    method_binding: &HttpMethodBindingIr,
127    fields: &serde_json::Value,
128    query: Option<&str>,
129) -> Result<String, TransportError> {
130    let mut path = method_binding.path_template.clone();
131
132    for binding in &method_binding.field_bindings {
133        match binding {
134            HttpFieldBinding::Path { field, param } => {
135                let value = field_as_string(fields, field)?.ok_or_else(|| {
136                    TransportError::UrlBuild(format!(
137                        "required path parameter '{field}' is missing or null"
138                    ))
139                })?;
140                if value.is_empty() {
141                    return Err(TransportError::UrlBuild(format!(
142                        "required path parameter '{field}' is empty"
143                    )));
144                }
145                let encoded = utf8_percent_encode(&value, PATH_SEGMENT).to_string();
146                path = path.replace(&format!("{{{param}}}"), &encoded);
147            }
148            // Query values are encoded by the caller; the binding entry stays in
149            // the IR for validation and spec generation.
150            HttpFieldBinding::Query { .. } | HttpFieldBinding::Body => {}
151        }
152    }
153
154    let base = base_url.trim_end_matches('/');
155    let base_p = base_path.trim_end_matches('/');
156    let mut url = format!("{base}{base_p}{path}");
157
158    if let Some(q) = query.filter(|q| !q.is_empty()) {
159        url.push('?');
160        url.push_str(q);
161    }
162
163    Ok(url)
164}
165
166/// Attach the platform-plane credential from `provider` (if any) to a REST
167/// [`RequestBuilder`] as the sensitive `X-ToolKit-Internal-Token` header.
168///
169/// The single audited REST emit point, shared by the unary-attempt closure and
170/// the SSE reconnect factory (so both re-resolve per attempt and pick up
171/// rotation). REST sibling of [`crate::grpc::attach_internal_token`]; both
172/// delegate the attach policy to [`InternalTokenProvider::resolve_for_attach`].
173pub fn attach_internal_token(
174    builder: RequestBuilder,
175    provider: Option<&InternalTokenProvider>,
176    rpc: &str,
177) -> RequestBuilder {
178    match InternalTokenProvider::resolve_for_attach(provider, rpc) {
179        Some(token) => builder.internal_token_auth(&token),
180        None => builder,
181    }
182}
183
184/// Map an HTTP method enum to [`http::Method`].
185#[must_use]
186pub fn to_http_method(method: HttpMethod) -> http::Method {
187    match method {
188        HttpMethod::Get => http::Method::GET,
189        HttpMethod::Post => http::Method::POST,
190        HttpMethod::Put => http::Method::PUT,
191        HttpMethod::Patch => http::Method::PATCH,
192        HttpMethod::Delete => http::Method::DELETE,
193    }
194}
195
196/// Map a non-success HTTP response into a [`TransportError`].
197///
198/// Tries to parse the body as an RFC 9457 [`Problem`] envelope first,
199/// falling back to [`TransportError::HttpStatus`] with a truncated body
200/// excerpt for peers that don't speak the canonical-errors envelope.
201///
202/// `retry_after` is the parsed `Retry-After` header (see [`parse_retry_after`]);
203/// it is attached to both the [`TransportError::Problem`] and the
204/// [`TransportError::HttpStatus`] fallback so the retry loop honors a
205/// server-advised backoff regardless of whether the peer speaks canonical
206/// errors.
207#[must_use]
208pub fn map_http_error(
209    status: u16,
210    body: String,
211    retry_after: Option<std::time::Duration>,
212) -> TransportError {
213    if let Ok(mut problem) = serde_json::from_str::<Problem>(&body) {
214        // RFC 9457 §3.1 makes `status` advisory; if the peer omitted it,
215        // the real response status is right here.
216        problem.status.get_or_insert(status);
217        return TransportError::Problem {
218            problem: Box::new(problem),
219            retry_after,
220        };
221    }
222    TransportError::HttpStatus {
223        status,
224        body: truncate(body, 256),
225        retry_after,
226    }
227}
228
229/// Parse a `Retry-After` response header as a delta-seconds [`Duration`].
230///
231/// Only the numeric delta-seconds form is supported (the common case for
232/// `429`/`503`); the HTTP-date form is ignored (returns `None`), as is a
233/// missing or malformed header.
234#[must_use]
235pub fn parse_retry_after(headers: &http::HeaderMap) -> Option<std::time::Duration> {
236    let raw = headers.get(http::header::RETRY_AFTER)?;
237    let secs: u64 = raw.to_str().ok()?.trim().parse().ok()?;
238    Some(std::time::Duration::from_secs(secs))
239}
240
241fn field_as_string(
242    fields: &serde_json::Value,
243    field_name: &str,
244) -> Result<Option<String>, TransportError> {
245    let Some(value) = fields.get(field_name) else {
246        return Ok(None);
247    };
248    match value {
249        serde_json::Value::String(s) => Ok(Some(s.clone())),
250        serde_json::Value::Number(n) => Ok(Some(n.to_string())),
251        serde_json::Value::Bool(b) => Ok(Some(b.to_string())),
252        serde_json::Value::Null => Ok(None),
253        _ => Err(TransportError::UrlBuild(format!(
254            "field '{field_name}' has non-scalar type and cannot be embedded into the URL"
255        ))),
256    }
257}
258
259fn truncate(mut s: String, max: usize) -> String {
260    if s.len() > max {
261        // `s` is a peer-controlled response body (arbitrary UTF-8); truncating
262        // at a raw byte offset panics if `max` lands inside a multi-byte
263        // character. Floor to the nearest char boundary at or below `max`.
264        let cut = (0..=max)
265            .rev()
266            .find(|&i| s.is_char_boundary(i))
267            .unwrap_or(0);
268        s.truncate(cut);
269        s.push('\u{2026}');
270    }
271    s
272}
273
274#[cfg(test)]
275#[cfg_attr(coverage_nightly, coverage(off))]
276#[allow(clippy::unwrap_used)]
277mod tests {
278    use super::*;
279    use crate::ir::binding::{HttpFieldBinding, HttpMethodBindingIr, StreamFraming};
280
281    /// Build an `http_body::Body` whose data arrives across several frames, so
282    /// the prefix reader is exercised on the cross-frame accumulation path (not
283    /// just a single buffered chunk).
284    fn framed_body(chunks: Vec<Vec<u8>>) -> impl http_body::Body<Data = Bytes, Error = String> {
285        use http_body::Frame;
286        let frames = chunks
287            .into_iter()
288            .map(|c| Ok::<_, String>(Frame::data(Bytes::from(c))));
289        http_body_util::StreamBody::new(futures_util::stream::iter(frames))
290    }
291
292    #[tokio::test]
293    async fn error_prefix_returns_a_short_body_intact() {
294        let body = framed_body(vec![b"service ".to_vec(), b"unavailable".to_vec()]);
295        let bytes = read_error_body_prefix(body).await.unwrap();
296        assert_eq!(&bytes[..], b"service unavailable");
297    }
298
299    #[tokio::test]
300    async fn error_prefix_caps_an_oversized_body_at_the_limit() {
301        // Two frames that each fit under the cap but together exceed it: the
302        // reader must stop at exactly ERROR_BODY_PREVIEW_LIMIT and abandon the
303        // rest rather than buffering the whole body.
304        let big = vec![b'x'; ERROR_BODY_PREVIEW_LIMIT];
305        let body = framed_body(vec![big.clone(), big]);
306        let bytes = read_error_body_prefix(body).await.unwrap();
307        assert_eq!(bytes.len(), ERROR_BODY_PREVIEW_LIMIT);
308    }
309
310    fn binding(template: &str, fields: Vec<HttpFieldBinding>) -> HttpMethodBindingIr {
311        HttpMethodBindingIr {
312            method_name: "x".to_owned(),
313            http_method: HttpMethod::Get,
314            path_template: template.to_owned(),
315            field_bindings: fields,
316            retryable: false,
317            streaming: false,
318            stream_framing: StreamFraming::default(),
319            optional: false,
320        }
321    }
322
323    #[test]
324    fn substitutes_path_param() {
325        let b = binding(
326            "/items/{id}",
327            vec![HttpFieldBinding::Path {
328                field: "id".into(),
329                param: "id".into(),
330            }],
331        );
332        let url = build_request_url(
333            "https://x.example",
334            "/api",
335            &b,
336            &serde_json::json!({ "id": "42" }),
337            None,
338        )
339        .unwrap();
340        assert_eq!(url, "https://x.example/api/items/42");
341    }
342
343    #[test]
344    fn appends_the_encoded_query_string() {
345        // The query arrives already encoded (by `crate::query::to_query_string`,
346        // the same codec the server decodes with); this function only joins it
347        // onto the URL.
348        let b = binding(
349            "/list",
350            vec![HttpFieldBinding::Query {
351                field: "filter".into(),
352                param: "filter".into(),
353            }],
354        );
355        let url = build_request_url(
356            "https://x.example",
357            "/api",
358            &b,
359            &serde_json::json!({}),
360            Some("status=paid&currency=USD"),
361        )
362        .unwrap();
363        assert_eq!(url, "https://x.example/api/list?status=paid&currency=USD");
364    }
365
366    #[test]
367    fn omits_the_separator_for_an_empty_query() {
368        let b = binding("/list", vec![]);
369        for query in [None, Some("")] {
370            let url = build_request_url(
371                "https://x.example",
372                "/api",
373                &b,
374                &serde_json::json!({}),
375                query,
376            )
377            .unwrap();
378            assert_eq!(url, "https://x.example/api/list", "query = {query:?}");
379        }
380    }
381
382    #[test]
383    fn maps_problem_envelope() {
384        // Canonical RFC 9457 Problem (per docs/arch/errors/DESIGN.md §3.3).
385        let body = serde_json::json!({
386            "type": "gts://gts.cf.core.errors.err.v1~cf.core.err.internal.v1~",
387            "title": "Internal",
388            "status": 500,
389            "detail": "broke",
390            "context": {}
391        })
392        .to_string();
393        let err = map_http_error(500, body, None);
394        match err {
395            TransportError::Problem { problem: p, .. } => {
396                assert_eq!(p.status, Some(500));
397                assert_eq!(p.detail, "broke");
398                assert!(p.problem_type.contains("internal"));
399            }
400            other => panic!("unexpected {other:?}"),
401        }
402    }
403
404    #[test]
405    fn maps_a_minimal_spec_compliant_problem_envelope() {
406        // RFC 9457 §3.1 makes `detail` and `context` optional. A genuinely
407        // foreign peer's Problem can omit both and still be fully
408        // spec-compliant - it must still map to `TransportError::Problem`
409        // (preserving the real `type`/`title`), not silently degrade to the
410        // generic `HttpStatus` fallback meant for peers that don't speak the
411        // canonical-errors envelope at all.
412        let body = serde_json::json!({
413            "type": "https://example.com/probs/out-of-credit",
414            "title": "You do not have enough credit.",
415            "status": 409
416        })
417        .to_string();
418        let err = map_http_error(409, body, None);
419        match err {
420            TransportError::Problem { problem: p, .. } => {
421                assert_eq!(p.problem_type, "https://example.com/probs/out-of-credit");
422                assert_eq!(p.title, "You do not have enough credit.");
423                assert_eq!(p.status, Some(409));
424                assert_eq!(p.detail, "");
425                assert_eq!(p.context, serde_json::json!({}));
426            }
427            other => panic!("unexpected {other:?}"),
428        }
429    }
430
431    #[test]
432    fn problem_envelope_carries_retry_after() {
433        // A canonical `Problem` peer that also sends `Retry-After` must have the
434        // advised delay threaded onto the `Problem` variant so the retry loop
435        // honors it (M-16) — not just the non-canonical `HttpStatus` fallback.
436        let body = serde_json::json!({
437            "type": "gts://gts.cf.core.errors.err.v1~cf.core.err.service_unavailable.v1~",
438            "title": "Service unavailable",
439            "status": 503,
440            "detail": "draining",
441            "context": {}
442        })
443        .to_string();
444        let err = map_http_error(503, body, Some(std::time::Duration::from_secs(2)));
445        assert_eq!(err.retry_after(), Some(std::time::Duration::from_secs(2)));
446        assert!(matches!(err, TransportError::Problem { .. }));
447    }
448
449    #[test]
450    fn truncate_does_not_panic_on_multibyte_char_at_boundary() {
451        // 255 ASCII bytes + a 3-byte UTF-8 char (é is 2 bytes; use a 3-byte
452        // char to straddle byte 256 exactly) — a raw `s.truncate(256)` would
453        // panic because byte 256 falls inside the multi-byte character.
454        let body = format!("{}€", "a".repeat(255)); // '€' is 3 bytes (U+20AC)
455        assert_eq!(body.len(), 258);
456        let out = truncate(body, 256);
457        // Truncated to the last valid boundary at or below 256 (255, since the
458        // '€' starts at byte 255), with the ellipsis marker appended.
459        assert_eq!(out, format!("{}\u{2026}", "a".repeat(255)));
460    }
461
462    #[test]
463    fn falls_back_to_http_status_for_non_problem_body() {
464        let err = map_http_error(503, "service unavailable".into(), None);
465        match err {
466            TransportError::HttpStatus { status, body, .. } => {
467                assert_eq!(status, 503);
468                assert!(body.contains("service unavailable"));
469            }
470            other => panic!("unexpected {other:?}"),
471        }
472    }
473
474    #[test]
475    fn parses_retry_after_delta_seconds() {
476        let mut headers = http::HeaderMap::new();
477        headers.insert(http::header::RETRY_AFTER, "2".parse().unwrap());
478        assert_eq!(
479            parse_retry_after(&headers),
480            Some(std::time::Duration::from_secs(2))
481        );
482
483        // HTTP-date form is ignored (unsupported), as is a missing header.
484        let mut date = http::HeaderMap::new();
485        date.insert(
486            http::header::RETRY_AFTER,
487            "Wed, 21 Oct 2026 07:28:00 GMT".parse().unwrap(),
488        );
489        assert_eq!(parse_retry_after(&date), None);
490        assert_eq!(parse_retry_after(&http::HeaderMap::new()), None);
491    }
492
493    #[test]
494    fn missing_path_param_returns_url_build_error() {
495        let b = binding(
496            "/items/{id}",
497            vec![HttpFieldBinding::Path {
498                field: "id".into(),
499                param: "id".into(),
500            }],
501        );
502        let err = build_request_url(
503            "https://x.example",
504            "/api",
505            &b,
506            &serde_json::json!({}),
507            None,
508        )
509        .unwrap_err();
510        assert!(matches!(err, TransportError::UrlBuild(_)));
511    }
512}
toolkit_contract/runtime/http.rs

toolkit_contract/runtime/
http.rs
