//! Per-connection logic: parse the first request, decide allowed/denied,//! either pump bytes through to the upstream or close (with an explanatory//! 511 for policy denials).//!//! Both CONNECT and HTTP forward go through here. After the policy decision,//! the TCP connection is pinned to the approved destination — directly, or//! via a CONNECT tunnel through the upstream proxy — and everything becomes//! opaque byte pumping. We don't parse anything else (chunked encoding,//! keep-alive HTTP requests after the first, etc.); because the whole//! connection can only reach the one approved host, later requests the//! client sends on it cannot escape the policy decision. Per-TCP-connection//! event granularity, by design. use crate::pinned_host::{PinnedHost, PinnedHostError};use crate::proxy::{    DenyReason, ProxyEvent, RequestMethod, RequestOutcome, RuntimeState, UpstreamProxy,};use anyhow::{Context, Result, anyhow, bail};use base64::Engine as _;use std::io::{Read, Write};use std::net::{IpAddr, Shutdown, SocketAddr, TcpStream, ToSocketAddrs as _};#[cfg(unix)]use std::os::unix::net::UnixStream;use std::sync::Arc;use std::thread;use std::time::{Duration, Instant};use url::Url; /// Buffer size for each direction of bidir copy. 64 KiB is the sweet spot/// for most networks — large enough to keep the pipe full, small enough/// not to balloon memory under many concurrent connections.const PUMP_BUFFER_SIZE: usize = 64 * 1024; /// Cap on request/response header bytes. The proxy runs inside the editor/// process and its sole client is model-driven code — exactly the party the/// sandbox distrusts — so an unbounded header read would let a malicious/// command balloon the editor's memory. 64 KiB is far beyond what real HTTP/// clients send.const MAX_HEADER_BYTES: usize = 64 * 1024; /// How long to wait for the client's request headers. Pooled connections/// that never send a request get closed; well-behaved clients retry on a/// fresh connection. Cleared before the pump phase so long-lived idle/// tunnels (long polls, slow downloads) are unaffected.const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(60); /// Timeout for outbound TCP connects (direct or to the upstream proxy),/// so a black-holed destination doesn't pin a connection thread for the/// OS default (~75s or more).const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); /// How long to wait for the upstream proxy's CONNECT response.const UPSTREAM_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); /// Top-level entry from the listener thread. Owns the client connection/// and the runtime state; emits events; never returns errors that escape/// the connection (those are logged at the listener level).pub(crate) enum ClientStream {    Tcp(TcpStream),    #[cfg(unix)]    Unix(UnixStream),} impl ClientStream {    fn try_clone(&self) -> std::io::Result<Self> {        match self {            Self::Tcp(stream) => stream.try_clone().map(Self::Tcp),            #[cfg(unix)]            Self::Unix(stream) => stream.try_clone().map(Self::Unix),        }    }     fn set_read_timeout(&self, duration: Option<Duration>) -> std::io::Result<()> {        match self {            Self::Tcp(stream) => stream.set_read_timeout(duration),            #[cfg(unix)]            Self::Unix(stream) => stream.set_read_timeout(duration),        }    }} impl Read for ClientStream {    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {        match self {            Self::Tcp(stream) => stream.read(buf),            #[cfg(unix)]            Self::Unix(stream) => stream.read(buf),        }    }} impl Write for ClientStream {    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {        match self {            Self::Tcp(stream) => stream.write(buf),            #[cfg(unix)]            Self::Unix(stream) => stream.write(buf),        }    }     fn flush(&mut self) -> std::io::Result<()> {        match self {            Self::Tcp(stream) => stream.flush(),            #[cfg(unix)]            Self::Unix(stream) => stream.flush(),        }    }} trait StreamIo: Read + Write + Send + 'static {    fn shutdown(&self, how: Shutdown) -> std::io::Result<()>;} impl StreamIo for ClientStream {    fn shutdown(&self, how: Shutdown) -> std::io::Result<()> {        match self {            Self::Tcp(stream) => stream.shutdown(how),            #[cfg(unix)]            Self::Unix(stream) => stream.shutdown(how),        }    }} impl StreamIo for TcpStream {    fn shutdown(&self, how: Shutdown) -> std::io::Result<()> {        TcpStream::shutdown(self, how)    }} pub(crate) fn handle(client: ClientStream, state: Arc<RuntimeState>) -> Result<()> {    if let ClientStream::Tcp(stream) = &client        && let Err(error) = stream.set_nodelay(true)    {        log::debug!("[http_proxy] failed to set TCP_NODELAY on client socket: {error}");    }     let mut client = client;    if let Err(error) = client.set_read_timeout(Some(HEADER_READ_TIMEOUT)) {        log::debug!("[http_proxy] failed to set client header read timeout: {error}");    }     // Read until end-of-headers (`\r\n\r\n`). The first request determines    // the destination for this entire TCP connection.    let (header_buf, header_end) = read_request_headers(&mut client)?;     // The header timeout must not apply to the pump phase. (Socket options    // are shared with the `try_clone` handles used by the pump.)    if let Err(error) = client.set_read_timeout(None) {        log::debug!("[http_proxy] failed to clear client read timeout: {error}");    }     let request = ParsedRequest::parse(&header_buf[..header_end])?;     // Bytes after the request headers — the start of an HTTP request body    // for forwarding cases, or (for unusually eager clients) bytes pipelined    // ahead of the CONNECT response. Replayed to the upstream either way.    let leftover_body = header_buf[header_end..].to_vec();     match request {        ParsedRequest::Connect { host, port } => {            handle_connect(client, host, port, leftover_body, state)        }        ParsedRequest::Http {            method,            host,            port,            request_bytes,        } => handle_http_forward(            client,            method,            host,            port,            request_bytes,            leftover_body,            state,        ),    }} /// Reads request headers (until `\r\n\r\n`) from the client. Returns the/// full buffer (which may include some bytes after the headers) and the/// offset where the headers ended. Capped at [`MAX_HEADER_BYTES`].fn read_request_headers(client: &mut ClientStream) -> Result<(Vec<u8>, usize)> {    let mut buf = Vec::with_capacity(4096);    let mut tmp = [0u8; 4096];    let mut searched = 0usize;    loop {        let n = client.read(&mut tmp)?;        if n == 0 {            bail!("client closed before sending complete request headers");        }        buf.extend_from_slice(&tmp[..n]);        // Only scan the new bytes (plus 3 bytes of overlap for a delimiter        // straddling the read boundary), keeping the search linear overall.        let scan_start = searched.saturating_sub(3);        if let Some(end) = find_double_crlf(&buf[scan_start..]) {            return Ok((buf, scan_start + end));        }        searched = buf.len();        if buf.len() > MAX_HEADER_BYTES {            bail!("request headers exceed {MAX_HEADER_BYTES} bytes");        }    }} fn find_double_crlf(buf: &[u8]) -> Option<usize> {    buf.windows(4).position(|w| w == b"\r\n\r\n").map(|i| i + 4)} /// Parsed first request from the client.enum ParsedRequest {    Connect {        host: String,        port: u16,    },    Http {        method: String,        host: String,        port: u16,        /// Request bytes (line + headers) to forward to the origin.        /// Absolute-form requests are rewritten to origin-form (see        /// [`build_origin_form_request`]); origin-form requests pass        /// through verbatim.        request_bytes: Vec<u8>,    },} impl ParsedRequest {    fn parse(headers: &[u8]) -> Result<Self> {        let mut header_storage = [httparse::EMPTY_HEADER; 64];        let mut req = httparse::Request::new(&mut header_storage);        let status = req.parse(headers).context("malformed HTTP request")?;        if !status.is_complete() {            bail!("incomplete HTTP request after \\r\\n\\r\\n boundary");        }         let method = req            .method            .ok_or_else(|| anyhow!("missing HTTP method"))?            .to_string();        let target = req.path.ok_or_else(|| anyhow!("missing request target"))?;         if method.eq_ignore_ascii_case("CONNECT") {            let (host, port) = parse_authority_form(target)?;            Ok(ParsedRequest::Connect { host, port })        } else if target.starts_with("http://") || target.starts_with("https://") {            let (host, port, url) = parse_absolute_form_target(target)?;            let request_bytes = build_origin_form_request(&method, &url, req.version, req.headers);            Ok(ParsedRequest::Http {                method,                host,                port,                request_bytes,            })        } else {            // Origin-form request: destination comes from the Host: header,            // and the bytes are already in the form an origin server expects,            // so forward them verbatim.            let host_hdr = req                .headers                .iter()                .find(|h| h.name.eq_ignore_ascii_case("host"))                .ok_or_else(|| anyhow!("origin-form request missing Host: header"))?;            let value =                std::str::from_utf8(host_hdr.value).context("Host: header is not valid UTF-8")?;            let (host, port) = parse_host_header(value)?;            Ok(ParsedRequest::Http {                method,                host,                port,                request_bytes: headers.to_vec(),            })        }    }} /// Parse `host:port` (CONNECT authority-form). Port is required, per RFC 9112/// §3.2.3.////// The port is split off manually rather than via `Url::parse`, because `Url`/// elides scheme-default ports — `host:80` parsed with an `http://` prefix/// becomes indistinguishable from a missing port, and `CONNECT host:80` is/// legitimate (e.g. `ws://` through a proxy). `Url` still canonicalizes the/// host part (bracketed IPv6, IDN-to-punycode, lowercasing).fn parse_authority_form(input: &str) -> Result<(String, u16)> {    let (host_part, port_part) = input        .rsplit_once(':')        .ok_or_else(|| anyhow!("CONNECT target '{input}' must include a port"))?;    let port: u16 = port_part        .parse()        .with_context(|| format!("CONNECT target '{input}' has an invalid port"))?;    let parsed = Url::parse(&format!("http://{host_part}"))        .with_context(|| format!("parsing CONNECT target '{input}'"))?;    let host = parsed        .host_str()        .ok_or_else(|| anyhow!("CONNECT target '{input}' has no host"))?        .to_string();    Ok((host, port))} /// Parse an absolute-form HTTP request target like `http://foo.com/path`./// Returns the parsed URL too, for the origin-form rewrite.fn parse_absolute_form_target(target: &str) -> Result<(String, u16, Url)> {    let parsed =        Url::parse(target).with_context(|| format!("parsing absolute-form target '{target}'"))?;    let host = parsed        .host_str()        .ok_or_else(|| anyhow!("absolute-form target '{target}' has no host"))?        .to_string();    // `port_or_known_default` covers `http://foo.com` (→ 80) without forcing    // the agent to spell it.    let port = parsed        .port_or_known_default()        .ok_or_else(|| anyhow!("absolute-form target '{target}' has no port"))?;    Ok((host, port, parsed))} /// Rewrite an absolute-form proxy request into origin-form for the origin/// server.////// RFC 9112 §3.2.2 requires origin servers to accept absolute-form, but many/// real servers and frameworks don't handle it; every production proxy/// rewrites, and so do we. The `Host` header is regenerated from the URL/// (per the same section, a proxy must use the URI host and ignore a/// mismatched `Host`), and `Proxy-*` headers — which are addressed to us and/// may carry credentials (`Proxy-Authorization`) — are stripped rather than/// leaked to the origin.fn build_origin_form_request(    method: &str,    url: &Url,    version: Option<u8>,    headers: &[httparse::Header],) -> Vec<u8> {    let mut target = url.path().to_string();    if let Some(query) = url.query() {        target.push('?');        target.push_str(query);    }    // `host_str` keeps brackets on IPv6 literals, which is what the Host    // header wants. `url.port()` is None for scheme-default ports, matching    // the convention of omitting them.    let host_value = match (url.host_str(), url.port()) {        (Some(host), Some(port)) => format!("{host}:{port}"),        (Some(host), None) => host.to_string(),        // Unreachable in practice: callers parsed the host already.        (None, _) => String::new(),    };    let minor_version = version.unwrap_or(1);     let mut out = Vec::with_capacity(256);    out.extend_from_slice(format!("{method} {target} HTTP/1.{minor_version}\r\n").as_bytes());    out.extend_from_slice(format!("Host: {host_value}\r\n").as_bytes());    for header in headers {        let name = header.name;        if name.eq_ignore_ascii_case("host")            || name                .get(.."proxy-".len())                .is_some_and(|prefix| prefix.eq_ignore_ascii_case("proxy-"))        {            continue;        }        out.extend_from_slice(name.as_bytes());        out.extend_from_slice(b": ");        out.extend_from_slice(header.value);        out.extend_from_slice(b"\r\n");    }    out.extend_from_slice(b"\r\n");    out} /// Parse a `Host:` header value into `(host, port)`. Default port is 80/// since this is only called for HTTP forward, never CONNECT.fn parse_host_header(value: &str) -> Result<(String, u16)> {    let value = value.trim();    if value.is_empty() {        bail!("empty Host header");    }    let parsed = Url::parse(&format!("http://{value}"))        .with_context(|| format!("parsing Host header '{value}'"))?;    let host = parsed        .host_str()        .ok_or_else(|| anyhow!("Host header '{value}' has no host"))?        .to_string();    let port = parsed.port().unwrap_or(80);    Ok((host, port))} /// Normalize a hostname for allowlist matching. Strips brackets from IPv6/// literals and a single trailing dot. Lowercasing happens inside the/// allowlist matcher.fn normalize_host(host: &str) -> String {    let stripped = host        .strip_prefix('[')        .and_then(|s| s.strip_suffix(']'))        .unwrap_or(host);    stripped.trim_end_matches('.').to_string()} /// Whether a hostname is an IP literal, per our policy.fn is_ip_literal(host: &str) -> bool {    let stripped = host        .strip_prefix('[')        .and_then(|s| s.strip_suffix(']'))        .unwrap_or(host);    stripped.parse::<IpAddr>().is_ok()} /// The policy check shared by CONNECT and HTTP forward: IP-literal targets/// and hosts outside the allowlist are denied. Both checks are skipped when/// the allowlist allows any host — that grant means unrestricted egress,/// including IP literals (matching the pre-allowlist `allow_network`/// behavior).fn policy_denial(host: &str, port: u16, state: &RuntimeState) -> Option<DenyReason> {    if state.allowlist.allows_any() {        return None;    }    if is_ip_literal(host) {        return Some(DenyReason::IpLiteralRejected {            target: format!("{host}:{port}"),        });    }    if !state.allowlist.allows(host) {        return Some(DenyReason::HostNotInAllowlist {            host: host.to_string(),        });    }    None} /// How an approved request will reach its destination.enum Route {    /// Connect directly to one of the addresses pinned when the host was    /// resolved and vetted.    Direct(PinnedHost),    /// Tunnel through the upstream proxy with a CONNECT handshake.    ViaUpstream(UpstreamProxy),} enum RouteFailure {    /// Policy denial — respond with 511.    Denied(DenyReason),    /// Network-level failure — close silently, per "no proxy here".    Error(anyhow::Error),} /// Decide how to reach `host:port`, resolving and vetting addresses for/// direct connections.////// Resolution happens here, in the unsandboxed editor process, so this is/// also where we keep allowlisted hostnames from smuggling the sandboxed/// command onto the local machine or local network: a hostname whose DNS/// points into loopback / private / link-local space (DNS rebinding) is/// denied, and the connection later uses the vetted addresses rather than/// re-resolving. The filter is skipped when the allowlist allows any host,/// since that grant means unrestricted egress. Upstream-proxied destinations/// aren't resolved locally at all — the upstream does its own resolution/// inside the user's trusted network.fn plan_route(host: &str, port: u16, state: &RuntimeState) -> Result<Route, RouteFailure> {    if let Some(upstream) = &state.upstream        && !upstream.bypasses(host, port)    {        return Ok(Route::ViaUpstream(upstream.clone()));    }     match PinnedHost::resolve_for_allowlist(host, port, &state.allowlist) {        Ok(pinned) => Ok(Route::Direct(pinned)),        Err(PinnedHostError::AllAddressesForbidden { host }) => {            Err(RouteFailure::Denied(DenyReason::ResolvedToForbiddenIp {                host,            }))        }        Err(error) => Err(RouteFailure::Error(anyhow!(error))),    }} fn handle_connect(    mut client: ClientStream,    host: String,    port: u16,    leftover_body: Vec<u8>,    state: Arc<RuntimeState>,) -> Result<()> {    let normalized = normalize_host(&host);     if let Some(reason) = policy_denial(&normalized, port, &state) {        return deny_request(            &mut client,            &state,            normalized,            port,            RequestMethod::Connect,            reason,        );    }     let route = match plan_route(&normalized, port, &state) {        Ok(route) => route,        Err(RouteFailure::Denied(reason)) => {            return deny_request(                &mut client,

Lines 1–500 of 1115. Load more