everruns_core/
egress.rs

1//! Host-owned outbound network boundary.
2//!
3//! `EgressService` is the runtime contract for traffic that leaves an
4//! Everruns deployment. Provider drivers, capabilities, toolkit integrations,
5//! and system services should depend on this trait instead of constructing
6//! transport clients directly.
7
8use crate::network_access::NetworkAccessList;
9use crate::system_allowlist::SystemAllowlist;
10use async_trait::async_trait;
11use futures::{Stream, StreamExt};
12use serde::{Deserialize, Serialize};
13use std::collections::BTreeMap;
14use std::pin::Pin;
15use std::sync::Arc;
16use std::time::Duration;
17use thiserror::Error;
18
19const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
20const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
21
22pub type EgressResult<T> = std::result::Result<T, EgressError>;
23pub type EgressByteStream = Pin<Box<dyn Stream<Item = EgressResult<Vec<u8>>> + Send>>;
24
25#[derive(Debug, Error)]
26pub enum EgressError {
27    #[error("Invalid egress request: {0}")]
28    InvalidRequest(String),
29
30    #[error("Outbound request blocked by network access policy: {url}")]
31    NetworkAccessDenied { url: String },
32
33    #[error("Outbound request signing is not configured")]
34    SigningUnavailable,
35
36    #[error("Outbound transport error: {0}")]
37    Transport(String),
38}
39
40impl EgressError {
41    fn invalid(message: impl Into<String>) -> Self {
42        Self::InvalidRequest(message.into())
43    }
44}
45
46/// Logical owner of an outbound request.
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48#[serde(rename_all = "snake_case")]
49pub enum EgressRequestKind {
50    Provider,
51    Capability,
52    Integration,
53    SystemEmail,
54    UtilityLlm,
55    Mcp,
56    Other(String),
57}
58
59/// Request signing behavior requested by the caller.
60#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
61#[serde(rename_all = "snake_case")]
62pub enum EgressSigning {
63    Disabled,
64    PlatformDefault,
65    Required,
66}
67
68/// Provider-neutral HTTP request carried over the egress boundary.
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
70pub struct EgressRequest {
71    pub method: String,
72    pub url: String,
73    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
74    pub headers: BTreeMap<String, String>,
75    #[serde(default, skip_serializing_if = "Vec::is_empty")]
76    pub body: Vec<u8>,
77    pub kind: EgressRequestKind,
78    #[serde(default = "default_signing")]
79    pub signing: EgressSigning,
80    #[serde(skip_serializing_if = "Option::is_none")]
81    pub network_access: Option<NetworkAccessList>,
82    #[serde(skip_serializing_if = "Option::is_none")]
83    pub timeout_ms: Option<u64>,
84    /// Whether `DirectEgressService` must perform DNS-pinned SSRF validation
85    /// after all egress policies pass and before connecting (TM-TOOL-018).
86    #[serde(default)]
87    pub dns_pinning_required: bool,
88    /// Pre-resolved socket addresses for DNS pinning (TM-TOOL-018).
89    ///
90    /// When set, `DirectEgressService` builds a per-request client pinned to
91    /// these addresses via `resolve_to_addrs`, closing the TOCTOU window
92    /// between URL validation and the actual TCP connect.  Not serialized —
93    /// runtime-only hint owned by the egress boundary.
94    #[serde(skip)]
95    pub pinned_addrs: Option<(String, Vec<std::net::SocketAddr>)>,
96}
97
98impl EgressRequest {
99    pub fn new(method: impl Into<String>, url: impl Into<String>, kind: EgressRequestKind) -> Self {
100        Self {
101            method: method.into(),
102            url: url.into(),
103            headers: BTreeMap::new(),
104            body: Vec::new(),
105            kind,
106            signing: EgressSigning::Disabled,
107            network_access: None,
108            timeout_ms: None,
109            dns_pinning_required: false,
110            pinned_addrs: None,
111        }
112    }
113
114    /// Require the egress boundary to perform DNS-pinned SSRF validation after
115    /// all egress policy checks pass and before opening a connection.
116    pub fn require_dns_pinning(mut self) -> Self {
117        self.dns_pinning_required = true;
118        self
119    }
120
121    /// Pin the outbound connection to pre-resolved addresses (TM-TOOL-018).
122    ///
123    /// No-op when `addrs` is empty (e.g. IP-literal URLs where the static
124    /// check already validated the address).
125    ///
126    /// Kept public for callers that resolve-then-check themselves and hand the
127    /// pinned addresses in (e.g. the web_fetch fetchkit transport and the MCP
128    /// client). New direct call sites should prefer `require_dns_pinning()` so
129    /// DNS resolution stays inside the egress boundary, after policy checks.
130    pub fn pinned_addrs(
131        mut self,
132        host: impl Into<String>,
133        addrs: Vec<std::net::SocketAddr>,
134    ) -> Self {
135        if !addrs.is_empty() {
136            self.pinned_addrs = Some((host.into(), addrs));
137        }
138        self
139    }
140
141    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
142        self.headers.insert(name.into(), value.into());
143        self
144    }
145
146    pub fn body(mut self, body: impl Into<Vec<u8>>) -> Self {
147        self.body = body.into();
148        self
149    }
150
151    pub fn signing(mut self, signing: EgressSigning) -> Self {
152        self.signing = signing;
153        self
154    }
155
156    pub fn network_access(mut self, network_access: Option<NetworkAccessList>) -> Self {
157        self.network_access = network_access;
158        self
159    }
160
161    pub fn timeout_ms(mut self, timeout_ms: u64) -> Self {
162        self.timeout_ms = Some(timeout_ms);
163        self
164    }
165}
166
167/// Provider-neutral HTTP response returned by the egress boundary.
168#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
169pub struct EgressResponse {
170    pub status: u16,
171    pub headers: BTreeMap<String, String>,
172    pub body: Vec<u8>,
173}
174
175pub struct EgressStreamResponse {
176    pub status: u16,
177    pub headers: BTreeMap<String, String>,
178    pub body: EgressByteStream,
179}
180
181#[async_trait]
182pub trait EgressService: Send + Sync {
183    async fn send(&self, request: EgressRequest) -> EgressResult<EgressResponse>;
184
185    async fn send_stream(&self, request: EgressRequest) -> EgressResult<EgressStreamResponse>;
186
187    fn name(&self) -> &'static str {
188        "EgressService"
189    }
190}
191
192#[derive(Debug, Clone, Default)]
193pub struct DisabledEgressService;
194
195#[async_trait]
196impl EgressService for DisabledEgressService {
197    async fn send(&self, _request: EgressRequest) -> EgressResult<EgressResponse> {
198        Err(EgressError::Transport(
199            "outbound egress service is disabled".to_string(),
200        ))
201    }
202
203    async fn send_stream(&self, _request: EgressRequest) -> EgressResult<EgressStreamResponse> {
204        Err(EgressError::Transport(
205            "outbound egress service is disabled".to_string(),
206        ))
207    }
208
209    fn name(&self) -> &'static str {
210        "DisabledEgressService"
211    }
212}
213
214#[derive(Clone)]
215pub struct DirectEgressService {
216    client: reqwest::Client,
217    /// Optional host-wide allowlist applied to every outbound request,
218    /// independently of the per-request `network_access`. `None` means no
219    /// global enforcement. See `crate::system_allowlist`.
220    system_allowlist: Option<Arc<SystemAllowlist>>,
221}
222
223impl std::fmt::Debug for DirectEgressService {
224    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
225        f.debug_struct("DirectEgressService")
226            .finish_non_exhaustive()
227    }
228}
229
230impl Default for DirectEgressService {
231    fn default() -> Self {
232        Self::new()
233    }
234}
235
236impl DirectEgressService {
237    pub fn new() -> Self {
238        Self {
239            client: reqwest::Client::builder()
240                .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
241                .timeout(DEFAULT_REQUEST_TIMEOUT)
242                .redirect(reqwest::redirect::Policy::none())
243                .build()
244                .expect("build direct egress HTTP client"),
245            system_allowlist: None,
246        }
247    }
248
249    pub fn with_client(client: reqwest::Client) -> Self {
250        Self {
251            client,
252            system_allowlist: None,
253        }
254    }
255
256    /// Construct the default direct transport for tenant/agent runtime egress.
257    ///
258    /// This honors `EVERRUNS_SYSTEM_ALLOWLIST_ENABLED` and is the right default
259    /// for capabilities, MCP, integrations, and runtime HTTP surfaces.
260    pub fn for_runtime_traffic_from_env() -> Self {
261        Self::from_env()
262    }
263
264    /// Construct with the global system allowlist resolved from the environment.
265    ///
266    /// Enforcement is active only when `EVERRUNS_SYSTEM_ALLOWLIST_ENABLED` is
267    /// set; otherwise this behaves exactly like [`DirectEgressService::new`].
268    /// Prefer [`DirectEgressService::for_runtime_traffic_from_env`] for new
269    /// runtime/agent call sites. Host-owned services should use direct provider
270    /// clients instead of `EgressService`.
271    pub fn from_env() -> Self {
272        Self::new().with_system_allowlist(SystemAllowlist::from_env())
273    }
274
275    /// Attach (or clear) the host-wide system allowlist.
276    pub fn with_system_allowlist(mut self, system_allowlist: Option<Arc<SystemAllowlist>>) -> Self {
277        self.system_allowlist = system_allowlist;
278        self
279    }
280
281    fn validate_request(&self, request: &EgressRequest) -> EgressResult<()> {
282        if request.method.trim().is_empty() {
283            return Err(EgressError::invalid("method is required"));
284        }
285        let parsed = reqwest::Url::parse(&request.url)
286            .map_err(|error| EgressError::invalid(format!("invalid URL: {error}")))?;
287        match parsed.scheme() {
288            "http" | "https" => {}
289            scheme => {
290                return Err(EgressError::invalid(format!(
291                    "URL must use http or https, got '{scheme}'"
292                )));
293            }
294        }
295        if let Some(acl) = &request.network_access
296            && !acl.is_url_allowed(&request.url)
297        {
298            return Err(EgressError::NetworkAccessDenied {
299                url: request.url.clone(),
300            });
301        }
302        // Host-wide allowlist: when enabled, every request routed through this
303        // runtime egress boundary must match one of the curated public
304        // resources. Host-owned services should not use `EgressService`.
305        if let Some(allowlist) = &self.system_allowlist
306            && !allowlist.is_url_allowed(&request.url)
307        {
308            return Err(EgressError::NetworkAccessDenied {
309                url: request.url.clone(),
310            });
311        }
312        Ok(())
313    }
314
315    async fn prepare_request(&self, mut request: EgressRequest) -> EgressResult<EgressRequest> {
316        self.validate_request(&request)?;
317        if request.signing == EgressSigning::Required {
318            return Err(EgressError::SigningUnavailable);
319        }
320        if request.dns_pinning_required {
321            let (validated_url, resolved_addrs) =
322                crate::url_validation::validate_url_dns_pinned(&request.url)
323                    .await
324                    .map_err(|error| EgressError::NetworkAccessDenied {
325                        url: format!("{} ({error})", request.url),
326                    })?;
327            let pin_host = validated_url.host_str().unwrap_or("").to_string();
328            request = request.pinned_addrs(pin_host, resolved_addrs);
329        }
330        Ok(request)
331    }
332
333    fn build_request(&self, request: EgressRequest) -> EgressResult<reqwest::RequestBuilder> {
334        let EgressRequest {
335            method,
336            url,
337            headers,
338            body,
339            timeout_ms,
340            pinned_addrs,
341            ..
342        } = request;
343
344        let method = reqwest::Method::from_bytes(method.as_bytes())
345            .map_err(|error| EgressError::invalid(format!("invalid HTTP method: {error}")))?;
346
347        // When pre-resolved addresses are provided, build a per-request client
348        // pinned to those IPs.  This closes the TOCTOU window between
349        // validate_url_dns_pinned and the actual TCP connect (TM-TOOL-018).
350        let mut builder = if let Some((ref host, ref addrs)) = pinned_addrs {
351            reqwest::Client::builder()
352                .connect_timeout(DEFAULT_CONNECT_TIMEOUT)
353                .redirect(reqwest::redirect::Policy::none())
354                .resolve_to_addrs(host, addrs)
355                .build()
356                .map_err(|e| EgressError::invalid(format!("pinned client build failed: {e}")))?
357                .request(method, &url)
358        } else {
359            self.client.request(method, &url)
360        };
361
362        for (name, value) in headers {
363            builder = builder.header(name, value);
364        }
365        if let Some(timeout_ms) = timeout_ms {
366            builder = builder.timeout(Duration::from_millis(timeout_ms));
367        }
368        if !body.is_empty() {
369            builder = builder.body(body);
370        }
371        Ok(builder)
372    }
373}
374
375#[async_trait]
376impl EgressService for DirectEgressService {
377    async fn send(&self, request: EgressRequest) -> EgressResult<EgressResponse> {
378        let request = self.prepare_request(request).await?;
379        let response = self
380            .build_request(request)?
381            .send()
382            .await
383            .map_err(|error| EgressError::Transport(error.to_string()))?;
384        let status = response.status().as_u16();
385        let headers = response
386            .headers()
387            .iter()
388            .filter_map(|(name, value)| {
389                value
390                    .to_str()
391                    .ok()
392                    .map(|value| (name.as_str().to_string(), value.to_string()))
393            })
394            .collect();
395        let body = response
396            .bytes()
397            .await
398            .map_err(|error| EgressError::Transport(error.to_string()))?
399            .to_vec();
400
401        Ok(EgressResponse {
402            status,
403            headers,
404            body,
405        })
406    }
407
408    async fn send_stream(&self, request: EgressRequest) -> EgressResult<EgressStreamResponse> {
409        let request = self.prepare_request(request).await?;
410        let response = self
411            .build_request(request)?
412            .send()
413            .await
414            .map_err(|error| EgressError::Transport(error.to_string()))?;
415        let status = response.status().as_u16();
416        let headers = response
417            .headers()
418            .iter()
419            .filter_map(|(name, value)| {
420                value
421                    .to_str()
422                    .ok()
423                    .map(|value| (name.as_str().to_string(), value.to_string()))
424            })
425            .collect();
426        let body = response.bytes_stream().map(|chunk| {
427            chunk
428                .map(|bytes| bytes.to_vec())
429                .map_err(|error| EgressError::Transport(error.to_string()))
430        });
431
432        Ok(EgressStreamResponse {
433            status,
434            headers,
435            body: Box::pin(body),
436        })
437    }
438
439    fn name(&self) -> &'static str {
440        "DirectEgressService"
441    }
442}
443
444fn default_signing() -> EgressSigning {
445    EgressSigning::Disabled
446}
447
448#[cfg(test)]
449mod tests {
450    use super::*;
451    use futures::StreamExt;
452    use serde_json::json;
453    use wiremock::matchers::{body_json, header, method, path};
454    use wiremock::{Mock, MockServer, ResponseTemplate};
455
456    #[tokio::test]
457    async fn direct_service_sends_json_request() {
458        let server = MockServer::start().await;
459        Mock::given(method("POST"))
460            .and(path("/v1/test"))
461            .and(header("Authorization", "Bearer test"))
462            .and(body_json(json!({"ok": true})))
463            .respond_with(ResponseTemplate::new(201).set_body_json(json!({
464                "id": "response_123"
465            })))
466            .expect(1)
467            .mount(&server)
468            .await;
469
470        let response = DirectEgressService::new()
471            .send(
472                EgressRequest::new(
473                    "POST",
474                    format!("{}/v1/test", server.uri()),
475                    EgressRequestKind::Capability,
476                )
477                .header("Authorization", "Bearer test")
478                .header("Content-Type", "application/json")
479                .body(serde_json::to_vec(&json!({"ok": true})).unwrap()),
480            )
481            .await
482            .unwrap();
483
484        assert_eq!(response.status, 201);
485        assert_eq!(
486            serde_json::from_slice::<serde_json::Value>(&response.body).unwrap()["id"],
487            "response_123"
488        );
489    }
490
491    #[tokio::test]
492    async fn direct_service_enforces_network_access() {
493        let error = DirectEgressService::new()
494            .send(
495                EgressRequest::new(
496                    "GET",
497                    "https://blocked.example.com/path",
498                    EgressRequestKind::Capability,
499                )
500                .network_access(Some(NetworkAccessList::allow_only(["allowed.example.com"]))),
501            )
502            .await
503            .unwrap_err();
504
505        assert!(matches!(error, EgressError::NetworkAccessDenied { .. }));
506    }
507
508    #[tokio::test]
509    async fn system_allowlist_blocks_unlisted_hosts() {
510        use crate::system_allowlist::SystemAllowlist;
511        let server = MockServer::start().await;
512        Mock::given(method("GET"))
513            .and(path("/blocked"))
514            .respond_with(ResponseTemplate::new(200))
515            .expect(0)
516            .mount(&server)
517            .await;
518        let allowlist = SystemAllowlist::from_toml(
519            r#"
520            [groups.test]
521            allowed = ["allowed.example.com"]
522            "#,
523        )
524        .unwrap();
525
526        let service = DirectEgressService::new().with_system_allowlist(Some(Arc::new(allowlist)));
527        let error = service
528            .send(EgressRequest::new(
529                "GET",
530                format!("{}/blocked", server.uri()),
531                EgressRequestKind::Capability,
532            ))
533            .await
534            .unwrap_err();
535
536        assert!(matches!(error, EgressError::NetworkAccessDenied { .. }));
537    }
538
539    #[tokio::test]
540    async fn system_allowlist_cannot_be_overridden_by_request_acl() {
541        use crate::system_allowlist::SystemAllowlist;
542        // The system allowlist is a hard ceiling: even a request whose own
543        // network_access explicitly permits the host must still be denied when
544        // the system allowlist does not list it. Sessions can narrow, never widen.
545        let allowlist = SystemAllowlist::from_toml(
546            r#"
547            [groups.test]
548            allowed = ["allowed.example.com"]
549            "#,
550        )
551        .unwrap();
552
553        let service = DirectEgressService::new().with_system_allowlist(Some(Arc::new(allowlist)));
554        let error = service
555            .send(
556                EgressRequest::new(
557                    "GET",
558                    "https://blocked.example.com/path",
559                    EgressRequestKind::Capability,
560                )
561                // A maximally-permissive per-request ACL that allows the host.
562                .network_access(Some(NetworkAccessList::allow_only(["blocked.example.com"]))),
563            )
564            .await
565            .unwrap_err();
566
567        assert!(matches!(error, EgressError::NetworkAccessDenied { .. }));
568    }
569
570    #[tokio::test]
571    async fn system_allowlist_permits_listed_hosts() {
572        use crate::system_allowlist::SystemAllowlist;
573        let server = MockServer::start().await;
574        let host = reqwest::Url::parse(&server.uri())
575            .unwrap()
576            .host_str()
577            .unwrap()
578            .to_string();
579        Mock::given(method("GET"))
580            .and(path("/ok"))
581            .respond_with(ResponseTemplate::new(200))
582            .expect(1)
583            .mount(&server)
584            .await;
585
586        let allowlist =
587            SystemAllowlist::from_toml(&format!("[groups.test]\nallowed = [\"{host}\"]\n"))
588                .unwrap();
589        let service = DirectEgressService::new().with_system_allowlist(Some(Arc::new(allowlist)));
590        let response = service
591            .send(EgressRequest::new(
592                "GET",
593                format!("{}/ok", server.uri()),
594                EgressRequestKind::Capability,
595            ))
596            .await
597            .unwrap();
598
599        assert_eq!(response.status, 200);
600    }
601
602    #[tokio::test]
603    async fn dns_pinning_blocks_loopback_after_request_policy_passes() {
604        let error = DirectEgressService::new()
605            .send(
606                EgressRequest::new(
607                    "GET",
608                    "http://127.0.0.1/latest/meta-data",
609                    EgressRequestKind::Capability,
610                )
611                .network_access(Some(NetworkAccessList::allow_only(["127.0.0.1"])))
612                .require_dns_pinning(),
613            )
614            .await
615            .unwrap_err();
616
617        assert!(matches!(error, EgressError::NetworkAccessDenied { .. }));
618        assert!(error.to_string().contains("private/internal address"));
619    }
620
621    #[tokio::test]
622    async fn dns_pinning_does_not_run_before_system_allowlist_denial() {
623        use crate::system_allowlist::SystemAllowlist;
624        let allowlist = SystemAllowlist::from_toml(
625            r#"
626            [groups.test]
627            allowed = ["allowed.example.com"]
628            "#,
629        )
630        .unwrap();
631
632        let blocked_url = "https://blocked.invalid/path";
633        let error = DirectEgressService::new()
634            .with_system_allowlist(Some(Arc::new(allowlist)))
635            .send(
636                EgressRequest::new("GET", blocked_url, EgressRequestKind::Capability)
637                    .network_access(Some(NetworkAccessList::allow_only(["blocked.invalid"])))
638                    .require_dns_pinning(),
639            )
640            .await
641            .unwrap_err();
642
643        assert!(
644            matches!(error, EgressError::NetworkAccessDenied { ref url } if url == blocked_url)
645        );
646    }
647
648    #[tokio::test]
649    async fn required_signing_fails_when_no_signer_is_configured() {
650        let error = DirectEgressService::new()
651            .send(
652                EgressRequest::new("GET", "https://example.com", EgressRequestKind::Capability)
653                    .signing(EgressSigning::Required),
654            )
655            .await
656            .unwrap_err();
657
658        assert!(matches!(error, EgressError::SigningUnavailable));
659    }
660
661    #[tokio::test]
662    async fn direct_service_does_not_follow_redirects() {
663        let redirect_server = MockServer::start().await;
664        Mock::given(method("GET"))
665            .and(path("/secret"))
666            .respond_with(ResponseTemplate::new(200).set_body_string("secret"))
667            .expect(0)
668            .mount(&redirect_server)
669            .await;
670
671        let server = MockServer::start().await;
672        Mock::given(method("GET"))
673            .and(path("/start"))
674            .respond_with(
675                ResponseTemplate::new(302)
676                    .insert_header("Location", format!("{}/secret", redirect_server.uri())),
677            )
678            .expect(1)
679            .mount(&server)
680            .await;
681
682        let response = DirectEgressService::new()
683            .send(EgressRequest::new(
684                "GET",
685                format!("{}/start", server.uri()),
686                EgressRequestKind::Capability,
687            ))
688            .await
689            .unwrap();
690
691        assert_eq!(response.status, 302);
692    }
693
694    #[tokio::test]
695    async fn direct_service_streams_response_body() {
696        let server = MockServer::start().await;
697        Mock::given(method("GET"))
698            .and(path("/stream"))
699            .respond_with(ResponseTemplate::new(200).set_body_string("data: one\n\ndata: two\n\n"))
700            .expect(1)
701            .mount(&server)
702            .await;
703
704        let mut response = DirectEgressService::new()
705            .send_stream(EgressRequest::new(
706                "GET",
707                format!("{}/stream", server.uri()),
708                EgressRequestKind::Capability,
709            ))
710            .await
711            .unwrap();
712
713        assert_eq!(response.status, 200);
714        let mut body = Vec::new();
715        while let Some(chunk) = response.body.next().await {
716            body.extend(chunk.unwrap());
717        }
718        assert_eq!(
719            String::from_utf8(body).unwrap(),
720            "data: one\n\ndata: two\n\n"
721        );
722    }
723}
everruns_core/egress.rs

everruns_core/
egress.rs
