agentd/runtime/
http_node.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//! The **`http` workflow node**: make an outbound REST call from a workflow —
3//! `GET`/`POST`/`PUT`/`PATCH`/`DELETE` with `headers`, `query`, and a
4//! `json`/`body` payload — and observe `{status, ok, headers, body, json}`. This
5//! is also how a workflow **emits a webhook** (a `POST` to a URL). It runs on an
6//! executor thread over the one SSRF-guarded HTTP client, so every outbound dial
7//! in the daemon passes the same guard; the URL and body are already
8//! template-rendered (`render_spec`) against the run's data.
9//!
10//! Security: the SSRF classifier guards the resolved host — private/loopback/
11//! link-local targets are refused unless the node sets `allow_private: true`
12//! (for a declared internal API). The guard resolves once and the dial takes the
13//! addresses it vetted, so a name that answers the check public and the connect
14//! private cannot rebind its way in. `https://` verifies the server certificate.
15
16use std::time::{Duration, Instant};
17
18use serde_json::{Map, Value, json};
19
20use crate::engine::run::StepStatus;
21
22const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
23
24impl crate::runtime::reactor::Runtime {
25    /// Execute an `http` step (fields already rendered by `render_spec`).
26    pub(crate) fn step_http(&mut self, run_id: &str, step_id: &str, spec: &Map<String, Value>) {
27        let url = spec
28            .get("url")
29            .and_then(Value::as_str)
30            .unwrap_or_default()
31            .to_string();
32        if url.is_empty() {
33            self.finish_step_pub(
34                run_id,
35                step_id,
36                StepStatus::Failed,
37                None,
38                Some("http: url is required".into()),
39                0,
40            );
41            return;
42        }
43        let method = spec
44            .get("method")
45            .and_then(Value::as_str)
46            .unwrap_or("GET")
47            .to_ascii_uppercase();
48        // The `http` step is a covered egress surface. In `closed` mode the URL
49        // must match a `kind: http` service-catalog entry; a literal URL is
50        // judged at config load, but a templated one only resolves here, so the
51        // check has to run again at dial time. A matching entry's `methods:`
52        // list is a ceiling in either mode — declaring an endpoint for reads
53        // must not silently authorize writes to it.
54        {
55            use crate::config::v2 as cfgv2;
56            if let Err(e) = cfgv2::egress_allows(
57                &self.settings.services,
58                self.settings.security.egress,
59                cfgv2::ServiceKind::Http,
60                &url,
61            ) {
62                self.finish_step_pub(run_id, step_id, StepStatus::Failed, None, Some(e), 0);
63                return;
64            }
65            if let Some((name, entry)) =
66                cfgv2::service_match(&self.settings.services, cfgv2::ServiceKind::Http, &url)
67                && let Some(methods) = &entry.methods
68                && !methods.iter().any(|m| m == &method)
69            {
70                self.finish_step_pub(
71                    run_id,
72                    step_id,
73                    StepStatus::Failed,
74                    None,
75                    Some(format!(
76                        "http: {method} is outside services.{name}.methods ({methods:?}) — the catalog's method ceiling"
77                    )),
78                    0,
79                );
80                return;
81            }
82        }
83        let mut headers: Vec<(String, String)> = spec
84            .get("headers")
85            .and_then(Value::as_object)
86            .map(|m| {
87                m.iter()
88                    .map(|(k, v)| (k.clone(), header_value(v)))
89                    .collect()
90            })
91            .unwrap_or_default();
92        // Resolve `{{secret:NAME}}` / `{{secret-file:PATH}}` references through the
93        // redacting resolver — so an `Authorization: Bearer {{secret:API_TOKEN}}`
94        // header (or the `sign` secret below) carries a real credential without it
95        // ever passing through the workflow templater, a log line, or step output.
96        let envs = self.env.clone();
97        let resolve_secret = move |s: &str| -> Result<String, String> {
98            if s.contains("{{secret") {
99                crate::sec::secret::resolve(s, &|k| {
100                    envs.iter().find(|(n, _)| n == k).map(|(_, v)| v.clone())
101                })
102            } else {
103                Ok(s.to_string())
104            }
105        };
106        for (_, v) in headers.iter_mut() {
107            match resolve_secret(v) {
108                Ok(r) => *v = r,
109                Err(e) => {
110                    self.finish_step_pub(
111                        run_id,
112                        step_id,
113                        StepStatus::Failed,
114                        None,
115                        Some(format!("http: header secret: {e}")),
116                        0,
117                    );
118                    return;
119                }
120            }
121        }
122        let query = spec
123            .get("query")
124            .and_then(Value::as_object)
125            .map(|m| {
126                m.iter()
127                    .map(|(k, v)| format!("{}={}", pct(k), pct(&header_value(v))))
128                    .collect::<Vec<_>>()
129                    .join("&")
130            })
131            .unwrap_or_default();
132        // `idempotency: {header|query, value?}` — the retry-safety declaration.
133        // The default value is derived from run+step identity, so every attempt
134        // of THIS step presents the same key and a deduping API treats a retry
135        // as the retry it is; `value:` substitutes an application key (already
136        // rendered, like every field) for APIs where the operation's real
137        // identity is a business fact such as an order id.
138        let mut query = query;
139        if let Some(idem) = spec.get("idempotency") {
140            let value = idem
141                .get("value")
142                .and_then(Value::as_str)
143                .map(str::to_string)
144                .unwrap_or_else(|| crate::engine::run::idempotency_key(run_id, step_id));
145            if let Some(h) = idem.get("header").and_then(Value::as_str) {
146                headers.push((h.to_string(), value));
147            } else if let Some(q) = idem.get("query").and_then(Value::as_str) {
148                let pair = format!("{}={}", pct(q), pct(&value));
149                if query.is_empty() {
150                    query = pair;
151                } else {
152                    query.push('&');
153                    query.push_str(&pair);
154                }
155            }
156        }
157        // Body: `json` (serialized + Content-Type) takes precedence over `body`.
158        let body: Vec<u8> = if let Some(j) = spec.get("json").filter(|v| !v.is_null()) {
159            if !headers
160                .iter()
161                .any(|(k, _)| k.eq_ignore_ascii_case("content-type"))
162            {
163                headers.push(("Content-Type".into(), "application/json".into()));
164            }
165            serde_json::to_vec(j).unwrap_or_default()
166        } else {
167            spec.get("body")
168                .and_then(Value::as_str)
169                .map(|s| s.as_bytes().to_vec())
170                .unwrap_or_default()
171        };
172        // `sign`: HMAC-sign the body so a receiver can verify the payload — this
173        // makes the node a best-practice **webhook emitter**, symmetric with the
174        // inbound `webhook` node's `hmac` verify. `{secret, header?, prefix?}`.
175        if let Some(sig) = spec.get("sign").and_then(Value::as_object)
176            && let Some(secret_ref) = sig.get("secret").and_then(Value::as_str)
177        {
178            let secret = match resolve_secret(secret_ref) {
179                Ok(s) => s,
180                Err(e) => {
181                    self.finish_step_pub(
182                        run_id,
183                        step_id,
184                        StepStatus::Failed,
185                        None,
186                        Some(format!("http: sign secret: {e}")),
187                        0,
188                    );
189                    return;
190                }
191            };
192            let header = sig
193                .get("header")
194                .and_then(Value::as_str)
195                .unwrap_or("X-Signature")
196                .to_string();
197            let prefix = sig
198                .get("prefix")
199                .and_then(Value::as_str)
200                .unwrap_or("sha256=");
201            let mac = crate::sha::hmac_sha256(secret.as_bytes(), &body);
202            let value = format!("{prefix}{}", crate::sha::to_hex(&mac));
203            headers.retain(|(k, _)| !k.eq_ignore_ascii_case(&header));
204            headers.push((header, value));
205        }
206        let timeout = spec
207            .get("timeout")
208            .and_then(crate::engine::model::duration_ms_opt)
209            .map(Duration::from_millis)
210            .unwrap_or(DEFAULT_TIMEOUT);
211        let allow_private = spec
212            .get("allow_private")
213            .and_then(Value::as_bool)
214            .unwrap_or(false);
215        // `expect`: acceptable status codes; default = 2xx is `ok`, else error.
216        let expect: Vec<u64> = spec
217            .get("expect")
218            .and_then(Value::as_array)
219            .map(|a| a.iter().filter_map(Value::as_u64).collect())
220            .unwrap_or_default();
221
222        self.log.info(
223            "http.request",
224            json!({"run": run_id, "step": step_id, "method": method, "url": url}),
225        );
226        let tx = self.events_tx.clone();
227        let (r, s) = (run_id.to_string(), step_id.to_string());
228        self.executing
229            .insert(format!("{run_id}/{step_id}"), Instant::now());
230        std::thread::Builder::new()
231            .name("step:http".into())
232            .spawn(move || {
233                let (output, is_error, error) = match do_http(
234                    &url,
235                    &method,
236                    &query,
237                    &headers,
238                    &body,
239                    timeout,
240                    allow_private,
241                ) {
242                    Ok(v) => {
243                        let status = v["status"].as_u64().unwrap_or(0);
244                        let ok = if expect.is_empty() {
245                            (200..400).contains(&status)
246                        } else {
247                            expect.contains(&status)
248                        };
249                        if ok {
250                            (v, false, None)
251                        } else {
252                            (v.clone(), true, Some(format!("http status {status}")))
253                        }
254                    }
255                    Err(e) => (Value::Null, true, Some(format!("http: {e}"))),
256                };
257                let _ = tx.send(super::events::Event::StepDone {
258                    run: r,
259                    step: s,
260                    output,
261                    is_error,
262                    error,
263                    tokens: 0,
264                });
265            })
266            .ok();
267    }
268}
269
270/// A header value as a string (JSON scalars stringify; non-scalars serialize).
271fn header_value(v: &Value) -> String {
272    match v {
273        Value::String(s) => s.clone(),
274        Value::Null => String::new(),
275        other => other.to_string(),
276    }
277}
278
279/// The blocking request (executor thread): parse + SSRF-guard + connect (+TLS) +
280/// round-trip, returning `{status, ok, headers, body, json}`.
281/// GET a URL and return its body as text.
282///
283/// A thin wrapper over the same guarded path the `http` node uses, for callers
284/// that want a document rather than a step result — loading a workflow
285/// definition from a definitions service, say. Sharing the path matters: the
286/// SSRF guard, the single resolve and the vetted dial are the parts that must
287/// not be reimplemented slightly differently somewhere else.
288pub(crate) fn fetch_text(
289    url: &str,
290    headers: &[(String, String)],
291    timeout: Duration,
292    allow_private: bool,
293) -> Result<String, String> {
294    let v = do_http(url, "GET", "", headers, &[], timeout, allow_private)?;
295    let status = v.get("status").and_then(Value::as_u64).unwrap_or(0);
296    if !(200..300).contains(&status) {
297        return Err(format!("HTTP {status}"));
298    }
299    match v.get("body") {
300        Some(Value::String(s)) => Ok(s.clone()),
301        Some(other) => Ok(other.to_string()),
302        None => Err("empty body".into()),
303    }
304}
305
306/// One outbound request, SSRF-guarded. `pub(crate)` because the `http` node is
307/// not the only outbound dial: a stream `forward:` pushes through the same
308/// guard rather than opening a second, differently-checked path out.
309pub(crate) fn do_http(
310    url: &str,
311    method: &str,
312    query: &str,
313    headers: &[(String, String)],
314    body: &[u8],
315    timeout: Duration,
316    allow_private: bool,
317) -> Result<Value, String> {
318    let u = crate::net::http::Url::parse(url)?;
319    let path = if query.is_empty() {
320        u.path.clone()
321    } else if u.path.contains('?') {
322        format!("{}&{}", u.path, query)
323    } else {
324        format!("{}?{}", u.path, query)
325    };
326    let hdr_refs: Vec<(&str, &str)> = headers
327        .iter()
328        .map(|(k, v)| (k.as_str(), v.as_str()))
329        .collect();
330    // The guard and the dial are one step: a standalone `guard_host` followed by
331    // `connect_tcp` resolves the name twice, and a URL a model or a workflow
332    // author supplied is exactly where a hostile nameserver would answer the
333    // check public and the connect `169.254.169.254`. `connect_vetted` resolves
334    // once, classifies, and dials an address it vetted; TLS/SNI and the `Host`
335    // header stay on the hostname below. The refusal is a PermissionDenied
336    // carrying the same `SsrfError` text, so the `http: {e}` step error a
337    // workflow surfaces reads as it always did.
338    let tcp = crate::net::ssrf::connect_vetted(&u.host, u.port, timeout, allow_private)
339        .map_err(|e| e.to_string())?;
340    let resp = if u.is_tls() {
341        #[cfg(feature = "tls")]
342        {
343            let mut s = crate::net::tls::connect(tcp, &u.host, None).map_err(|e| e.to_string())?;
344            crate::net::http::send(&mut s, &u.host_header(), method, &path, &hdr_refs, body)
345                .map_err(|e| e.to_string())?
346        }
347        #[cfg(not(feature = "tls"))]
348        {
349            return Err("https requires the 'tls' build feature".into());
350        }
351    } else {
352        let mut s = tcp;
353        crate::net::http::send(&mut s, &u.host_header(), method, &path, &hdr_refs, body)
354            .map_err(|e| e.to_string())?
355    };
356    let body_str = resp.body_str().to_string();
357    let headers_obj: Map<String, Value> = resp
358        .headers
359        .iter()
360        .map(|(k, v)| (k.clone(), json!(v)))
361        .collect();
362    Ok(json!({
363        "status": resp.status,
364        "ok": resp.is_success(),
365        "headers": headers_obj,
366        "body": body_str,
367        "json": serde_json::from_str::<Value>(&body_str).ok(),
368    }))
369}
370
371/// Percent-encode one query component. Only the RFC 3986 unreserved characters —
372/// ASCII letters, digits and `-` `_` `.` `~` — pass through untouched;
373/// every other byte becomes `%XX`, so `&`, `=` and `?` in a value cannot
374/// break out and forge extra query parameters.
375fn pct(s: &str) -> String {
376    let mut out = String::with_capacity(s.len());
377    for b in s.bytes() {
378        match b {
379            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
380                out.push(b as char)
381            }
382            _ => out.push_str(&format!("%{b:02X}")),
383        }
384    }
385    out
386}
agentd/runtime/http_node.rs

agentd/runtime/
http_node.rs
