tenant.openagents/omega

1//! The proxy itself: listener, connection handlers, upstream chaining.
2//!
3//! All synchronous, thread-per-connection. `ProxyHandle::spawn` binds a
4//! `std::net::TcpListener` on `127.0.0.1:0` and returns once the listener
5//! is bound and the listener thread has been spawned. Drop the handle to
6//! shut everything down — the listener thread stops accepting new
7//! connections; in-flight connection threads finish on their own when
8//! either side closes.
9//!
10//! See the crate-level docs for trust assumptions and the "no proxy here"
11//! principle.
12
13mod connection;
14mod upstream;
15
16use crate::allowlist::Allowlist;
17use anyhow::{Context, Result};
18use futures::channel::mpsc;
19use std::net::{Ipv4Addr, TcpListener, TcpStream};
20#[cfg(unix)]
21use std::os::unix::net::{UnixListener, UnixStream};
22use std::path::{Path, PathBuf};
23use std::sync::Arc;
24#[cfg(unix)]
25use std::sync::atomic::AtomicU64;
26use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
27use std::thread;
28
29/// Cap on concurrently handled connections. Each connection costs the
30/// editor process two threads and two pump buffers; the cap keeps a
31/// runaway (or malicious) sandboxed command from exhausting the editor's
32/// thread/fd budget. Well above what parallel package managers open.
33const MAX_CONCURRENT_CONNECTIONS: usize = 256;
34
35pub use upstream::UpstreamProxy;
36
37/// Configuration for spawning a proxy.
38#[derive(Debug, Clone)]
39pub struct ProxyConfig {
40    /// Hosts the proxy will allow to be reached.
41    pub allowlist: Allowlist,
42    /// Optional upstream HTTP proxy to chain through, with `NO_PROXY`-style
43    /// bypasses for hosts that should connect direct.
44    pub upstream: Option<UpstreamProxy>,
45    /// Where the proxy reports per-connection events. Use
46    /// [`mpsc::unbounded`] so connection threads (which are sync) never
47    /// block on send. The receiver is async-friendly so `gpui` / `tokio`
48    /// callers can poll it from their executor of choice.
49    pub events: mpsc::UnboundedSender<ProxyEvent>,
50}
51
52/// A request method seen by the proxy.
53///
54/// Either a CONNECT (HTTPS tunnel) or an HTTP forward request.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub enum RequestMethod {
57    Connect,
58    Http(String),
59}
60
61impl RequestMethod {
62    pub fn as_str(&self) -> &str {
63        match self {
64            RequestMethod::Connect => "CONNECT",
65            RequestMethod::Http(method) => method.as_str(),
66        }
67    }
68}
69
70/// Outcome of a single connection's policy decision.
71#[derive(Debug, Clone)]
72pub enum RequestOutcome {
73    Allowed,
74    Denied { reason: DenyReason },
75}
76
77/// Why an attempted connection was denied.
78#[derive(Debug, Clone, PartialEq, Eq)]
79pub enum DenyReason {
80    /// Hostname (in punycode form on the wire) wasn't in the allowlist.
81    HostNotInAllowlist { host: String },
82    /// CONNECT or HTTP request targeted an IP literal. Denied unless the
83    /// allowlist allows any host.
84    IpLiteralRejected { target: String },
85    /// The hostname resolved only to loopback / private / link-local
86    /// addresses, which the sandbox policy never reaches via the allowlist
87    /// (DNS-rebinding protection). Not applied when the allowlist allows
88    /// any host.
89    ResolvedToForbiddenIp { host: String },
90}
91
92impl DenyReason {
93    pub(crate) fn proxy_status_error(&self) -> &'static str {
94        match self {
95            DenyReason::HostNotInAllowlist { .. } => "destination_ip_prohibited",
96            DenyReason::IpLiteralRejected { .. } => "destination_ip_prohibited",
97            DenyReason::ResolvedToForbiddenIp { .. } => "destination_ip_prohibited",
98        }
99    }
100
101    pub(crate) fn human_explanation(&self) -> String {
102        match self {
103            DenyReason::HostNotInAllowlist { host } => {
104                format!("host '{host}' is not in this conversation's network allowlist")
105            }
106            DenyReason::IpLiteralRejected { target } => format!(
107                "target '{target}' is an IP literal; only hostnames are permitted by sandbox policy"
108            ),
109            DenyReason::ResolvedToForbiddenIp { host } => format!(
110                "host '{host}' resolves only to loopback/private/link-local addresses, \
111                 which sandbox policy blocks"
112            ),
113        }
114    }
115}
116
117/// Events emitted by the proxy as it handles connections.
118#[derive(Debug, Clone)]
119pub enum ProxyEvent {
120    /// Sent once after the listener is bound. Always the first event for
121    /// a given proxy instance.
122    Ready { port: u16 },
123
124    /// Emitted at policy-decision time, before bytes flow to the upstream.
125    RequestAttempt {
126        host: String,
127        port: u16,
128        method: RequestMethod,
129        outcome: RequestOutcome,
130    },
131
132    /// Emitted after an `Allowed` connection finishes. Carries throughput
133    /// totals for diagnostics. Not emitted for denied connections.
134    RequestCompleted {
135        host: String,
136        port: u16,
137        method: RequestMethod,
138        bytes_to_remote: u64,
139        bytes_from_remote: u64,
140        duration_ms: u64,
141    },
142}
143
144/// Handle to a running proxy. Drop to stop the listener; in-flight
145/// connection threads finish on their own as soon as either side closes.
146pub struct ProxyHandle {
147    port: u16,
148    socket_path: Option<PathBuf>,
149    cleanup_directory: Option<PathBuf>,
150    /// Listener thread sees this flip to `true` after `accept` returns and
151    /// then exits.
152    shutdown: Arc<AtomicBool>,
153    /// Joined on drop to make shutdown deterministic in tests; ignored if
154    /// the listener has already exited.
155    listener_thread: Option<thread::JoinHandle<()>>,
156}
157
158impl ProxyHandle {
159    /// Spawns the proxy: binds a listener on `127.0.0.1:0`, spawns the
160    /// listener thread, sends a `Ready` event, and returns. The returned
161    /// port is what callers should use for `HTTPS_PROXY`/`HTTP_PROXY` env
162    /// vars and for the seatbelt rule narrowing `localhost:<port>`.
163    pub fn spawn(config: ProxyConfig) -> Result<ProxyHandle> {
164        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
165            .context("failed to bind proxy listener on 127.0.0.1:0")?;
166        let port = listener
167            .local_addr()
168            .context("failed to read proxy local addr")?
169            .port();
170
171        // Inform the parent the proxy is ready before starting the accept
172        // loop. Send is fire-and-forget on an unbounded channel — never
173        // blocks, never errors meaningfully.
174        let _ = config.events.unbounded_send(ProxyEvent::Ready { port });
175
176        let shutdown = Arc::new(AtomicBool::new(false));
177        let runtime_state = Arc::new(RuntimeState {
178            allowlist: config.allowlist,
179            upstream: config.upstream,
180            events: config.events,
181            active_connections: AtomicUsize::new(0),
182        });
183
184        let listener_thread = thread::Builder::new()
185            .name("http-proxy-listener".to_string())
186            // Listener thread does almost nothing on its stack — accept,
187            // spawn, loop. 128 KiB is plenty.
188            .stack_size(128 * 1024)
189            .spawn({
190                let shutdown = shutdown.clone();
191                move || run_listener(listener, runtime_state, shutdown)
192            })
193            .context("failed to spawn proxy listener thread")?;
194
195        Ok(ProxyHandle {
196            port,
197            socket_path: None,
198            cleanup_directory: None,
199            shutdown,
200            listener_thread: Some(listener_thread),
201        })
202    }
203
204    /// Spawns the proxy on a fresh pathname Unix socket under the system temp
205    /// directory and reserves a loopback port number for the in-sandbox bridge.
206    #[cfg(unix)]
207    pub fn spawn_unix_temp(config: ProxyConfig) -> Result<ProxyHandle> {
208        let directory = unique_temp_socket_directory();
209        let path = directory.join("proxy.sock");
210        Self::spawn_unix_with_cleanup(path, Some(directory), config)
211    }
212
213    /// Spawns the proxy on a pathname Unix socket and reserves a loopback port
214    /// number for the in-sandbox bridge to listen on.
215    #[cfg(unix)]
216    pub fn spawn_unix(path: impl AsRef<Path>, config: ProxyConfig) -> Result<ProxyHandle> {
217        Self::spawn_unix_with_cleanup(path.as_ref().to_path_buf(), None, config)
218    }
219
220    #[cfg(unix)]
221    fn spawn_unix_with_cleanup(
222        path: PathBuf,
223        cleanup_directory: Option<PathBuf>,
224        config: ProxyConfig,
225    ) -> Result<ProxyHandle> {
226        if let Some(parent) = path.parent() {
227            std::fs::create_dir_all(parent).with_context(|| {
228                format!(
229                    "failed to create proxy socket directory {}",
230                    parent.display()
231                )
232            })?;
233        }
234        if path.exists() {
235            std::fs::remove_file(&path).with_context(|| {
236                format!("failed to remove stale proxy socket {}", path.display())
237            })?;
238        }
239
240        let listener = UnixListener::bind(&path)
241            .with_context(|| format!("failed to bind proxy Unix socket {}", path.display()))?;
242        let port = reserve_loopback_port()?;
243
244        let _ = config.events.unbounded_send(ProxyEvent::Ready { port });
245
246        let shutdown = Arc::new(AtomicBool::new(false));
247        let runtime_state = Arc::new(RuntimeState {
248            allowlist: config.allowlist,
249            upstream: config.upstream,
250            events: config.events,
251            active_connections: AtomicUsize::new(0),
252        });
253
254        let listener_thread = thread::Builder::new()
255            .name("http-proxy-unix-listener".to_string())
256            .stack_size(128 * 1024)
257            .spawn({
258                let shutdown = shutdown.clone();
259                move || run_unix_listener(listener, runtime_state, shutdown)
260            })
261            .context("failed to spawn proxy Unix listener thread")?;
262
263        Ok(ProxyHandle {
264            port,
265            socket_path: Some(path),
266            cleanup_directory,
267            shutdown,
268            listener_thread: Some(listener_thread),
269        })
270    }
271
272    /// The loopback TCP port clients should use for proxy environment variables.
273    pub fn port(&self) -> u16 {
274        self.port
275    }
276
277    /// Path of the Unix socket listener, for Linux bridge mode.
278    pub fn socket_path(&self) -> Option<&Path> {
279        self.socket_path.as_deref()
280    }
281}
282
283impl Drop for ProxyHandle {
284    fn drop(&mut self) {
285        self.shutdown.store(true, Ordering::SeqCst);
286        // The listener is blocked in `accept()`. Waking it up cleanly via
287        // a flag alone isn't possible with `std::net::TcpListener` — there's
288        // no way to interrupt the syscall. Connect to ourselves: the
289        // listener wakes up, accepts the connection, sees the shutdown
290        // flag, breaks the loop. The accepted connection's worker thread
291        // will read the empty stream and exit too.
292        if let Some(path) = &self.socket_path {
293            #[cfg(unix)]
294            let _ = UnixStream::connect(path);
295            #[cfg(not(unix))]
296            let _ = path;
297        } else {
298            let _ = TcpStream::connect((Ipv4Addr::LOCALHOST, self.port));
299        }
300
301        if let Some(thread) = self.listener_thread.take() {
302            // Give the listener a chance to clean up. A join error means the
303            // listener thread panicked; there's nothing to recover, but it
304            // shouldn't pass unnoticed.
305            if thread.join().is_err() {
306                log::warn!("[http_proxy] listener thread panicked");
307            }
308        }
309        if let Some(path) = &self.socket_path
310            && let Err(error) = std::fs::remove_file(path)
311            && error.kind() != std::io::ErrorKind::NotFound
312        {
313            log::debug!(
314                "[http_proxy] failed to remove proxy Unix socket {}: {error}",
315                path.display()
316            );
317        }
318        if let Some(directory) = &self.cleanup_directory
319            && let Err(error) = std::fs::remove_dir(directory)
320            && error.kind() != std::io::ErrorKind::NotFound
321        {
322            log::debug!(
323                "[http_proxy] failed to remove proxy socket directory {}: {error}",
324                directory.display()
325            );
326        }
327    }
328}
329
330#[cfg(unix)]
331fn unique_temp_socket_directory() -> PathBuf {
332    static COUNTER: AtomicU64 = AtomicU64::new(0);
333    let counter = COUNTER.fetch_add(1, Ordering::Relaxed);
334    let nanos = std::time::SystemTime::now()
335        .duration_since(std::time::UNIX_EPOCH)
336        .map(|duration| duration.as_nanos())
337        .unwrap_or(0);
338    std::env::temp_dir().join(format!(
339        "zed-proxy-{}-{nanos}-{counter}",
340        std::process::id()
341    ))
342}
343
344/// State shared across all connection threads for a single proxy instance.
345pub(crate) struct RuntimeState {
346    pub(crate) allowlist: Allowlist,
347    pub(crate) upstream: Option<UpstreamProxy>,
348    pub(crate) events: mpsc::UnboundedSender<ProxyEvent>,
349    active_connections: AtomicUsize,
350}
351
352/// Decrements the active-connection count when a connection thread finishes
353/// (normally or by panic).
354struct ConnectionSlot(Arc<RuntimeState>);
355
356impl Drop for ConnectionSlot {
357    fn drop(&mut self) {
358        self.0.active_connections.fetch_sub(1, Ordering::SeqCst);
359    }
360}
361
362#[cfg(unix)]
363fn reserve_loopback_port() -> Result<u16> {
364    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
365        .context("failed to reserve proxy bridge port")?;
366    Ok(listener
367        .local_addr()
368        .context("failed to read reserved proxy bridge port")?
369        .port())
370}
371
372fn run_listener(listener: TcpListener, state: Arc<RuntimeState>, shutdown: Arc<AtomicBool>) {
373    for stream in listener.incoming() {
374        if shutdown.load(Ordering::SeqCst) {
375            log::debug!("[http_proxy] listener stopping (shutdown signaled)");
376            break;
377        }
378        match stream {
379            Ok(stream) => spawn_connection(connection::ClientStream::Tcp(stream), &state),
380            Err(e) => {
381                // EMFILE / per-process fd exhaustion is the realistic
382                // failure here. Log and keep going — accept errors are
383                // usually transient.
384                log::warn!("[http_proxy] accept failed: {e}");
385            }
386        }
387    }
388}
389
390#[cfg(unix)]
391fn run_unix_listener(listener: UnixListener, state: Arc<RuntimeState>, shutdown: Arc<AtomicBool>) {
392    for stream in listener.incoming() {
393        if shutdown.load(Ordering::SeqCst) {
394            log::debug!("[http_proxy] Unix listener stopping (shutdown signaled)");
395            break;
396        }
397        match stream {
398            Ok(stream) => spawn_connection(connection::ClientStream::Unix(stream), &state),
399            Err(error) => log::warn!("[http_proxy] Unix accept failed: {error}"),
400        }
401    }
402}
403
404fn spawn_connection(stream: connection::ClientStream, state: &Arc<RuntimeState>) {
405    let previous = state.active_connections.fetch_add(1, Ordering::SeqCst);
406    if previous >= MAX_CONCURRENT_CONNECTIONS {
407        state.active_connections.fetch_sub(1, Ordering::SeqCst);
408        log::warn!(
409            "[http_proxy] dropping connection: {MAX_CONCURRENT_CONNECTIONS} \
410             connections already active"
411        );
412        drop(stream);
413        return;
414    }
415    let slot = ConnectionSlot(state.clone());
416    let state = state.clone();
417    let result = thread::Builder::new()
418        .name("http-proxy-conn".to_string())
419        .stack_size(128 * 1024)
420        .spawn(move || {
421            let _slot = slot;
422            if let Err(error) = connection::handle(stream, state) {
423                log::debug!("[http_proxy] connection handler error: {error}");
424            }
425        });
426    if let Err(error) = result {
427        log::warn!("[http_proxy] failed to spawn connection thread: {error}");
428    }
429}
430