diff --git a/Cargo.lock b/Cargo.lock index fe900ce5d4..3c8a132e47 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -577,6 +577,7 @@ name = "busbar-core-connector" version = "1.6.0" dependencies = [ "axum", + "base64 0.22.1", "busbar-contract", "busbar-kernel", "busbar-kernel-egress", @@ -600,6 +601,7 @@ dependencies = [ "tokio", "tokio-rustls", "tokio-util", + "url", "webpki-roots", "zeroize", ] diff --git a/crates/busbar-contract/src/lib.rs b/crates/busbar-contract/src/lib.rs index d6cef7e076..553d9d38f7 100644 --- a/crates/busbar-contract/src/lib.rs +++ b/crates/busbar-contract/src/lib.rs @@ -99,6 +99,7 @@ pub mod plugin; pub mod plugin_rows; #[allow(missing_docs, missing_debug_implementations)] pub mod protocol; +pub mod proxy_route; pub mod redacted; // The money-path durable record SHAPES (DECISIONS #83: contract = shapes, ledger = semantics; // DECISIONS #84: nothing on the plugin path may link a crate holding semantics). The module is a diff --git a/crates/busbar-contract/src/proxy_route.rs b/crates/busbar-contract/src/proxy_route.rs new file mode 100644 index 0000000000..5c104a6e87 --- /dev/null +++ b/crates/busbar-contract/src/proxy_route.rs @@ -0,0 +1,216 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! THE PROXY ROUTE CHOICE: which proxy, if any, a dial goes through (status row 130 E5a). A shape, +//! shared: the connector reads the process environment into it once (`busbar_core_connector::proxy`, +//! the one parser) and every dial it makes asks it; the composition root hands the same value to +//! the kernel's remaining egress engine, which reads no environment of its own. +//! +//! The rules are reqwest 0.12's implicit proxy environment, as 1.5.5 shipped them: a secured target +//! uses the secured slot and every other target the plain slot (each set from its own variable, +//! with the catch-all variable as the fallback for both), and the exemption list excludes a host +//! from both before either is read, with curl's rule set (host, domain suffix, IP literal, CIDR). + +use std::net::IpAddr; +use std::sync::Arc; + +/// ONE PROXY ENDPOINT: where to connect, and the `Proxy-Authorization` value its userinfo became. +#[derive(Clone, PartialEq, Eq)] +pub struct ProxyHop { + host: String, + port: u16, + auth: Option, +} + +impl std::fmt::Debug for ProxyHop { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // The credential is never rendered. + f.debug_struct("ProxyHop") + .field("host", &self.host) + .field("port", &self.port) + .field("auth", &self.auth.as_ref().map(|_| "")) + .finish() + } +} + +impl ProxyHop { + /// The endpoint `host:port`, presenting `auth` (the whole `Proxy-Authorization` value) when set. + #[must_use] + pub fn new(host: String, port: u16, auth: Option) -> Self { + Self { host, port, auth } + } + + /// The proxy's host. + #[must_use] + pub fn host(&self) -> &str { + &self.host + } + + /// The proxy's port. + #[must_use] + pub fn port(&self) -> u16 { + self.port + } + + /// The `Proxy-Authorization` value, where the proxy's userinfo set one. + #[must_use] + pub fn proxy_authorization(&self) -> Option<&str> { + self.auth.as_deref() + } + + /// The proxy's own `host:port`, as a dial reaches it (a v6 host bracketed). + #[must_use] + pub fn endpoint(&self) -> String { + if self.host.contains(':') && !self.host.starts_with('[') { + format!("[{}]:{}", self.host, self.port) + } else { + format!("{}:{}", self.host, self.port) + } + } +} + +/// THE CHOICE: the secured slot, the plain slot, and the hosts exempt from both. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProxyChoice { + secured: Option>, + plain: Option>, + exempt: NoProxy, +} + +impl ProxyChoice { + /// The choice of a `secured` slot, a `plain` slot, and the exemption list `exempt` as its + /// variable spells it (comma-separated entries). + #[must_use] + pub fn new(secured: Option>, plain: Option>, exempt: &str) -> Self { + Self { + secured, + plain, + exempt: NoProxy::parse(exempt), + } + } + + /// The per-dial choice — `None` → direct (the host is exempt, or no proxy is set for the + /// target's slot), `Some` → tunnel through that proxy. + #[must_use] + pub fn select(&self, secured_target: bool, target_host: &str) -> Option> { + if self.exempt.matches(target_host) { + return None; + } + if secured_target { + self.secured.clone() + } else { + self.plain.clone() + } + } + + /// Whether the exemption list excludes `target_host` from both slots. + #[must_use] + pub fn exempts(&self, target_host: &str) -> bool { + self.exempt.matches(target_host) + } +} + +/// Parsed `NO_PROXY` entries, reqwest's full rule set (hyper-util `matcher::NoProxy`, which +/// documents itself as curl's rules): `*` matches everything; an IP literal matches exactly; +/// a CIDR block (`10.0.0.0/8`, `fd00::/8`) matches by containment; anything else is a domain +/// matched exactly or as a `.`-boundary suffix. Like reqwest, an IP-literal target host is +/// checked ONLY against the IP/CIDR entries and a domain host ONLY against the domain +/// entries. +#[derive(Debug, Clone, PartialEq, Eq)] +struct NoProxy { + entries: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum NoProxyEntry { + MatchAll, + /// Lowercased, leading dot stripped (`.google.com` ≡ `google.com`, per curl/reqwest). + Domain(String), + Ip(IpAddr), + /// Network address + prefix length, validated ≤32 (v4) / ≤128 (v6) at parse. + Cidr(IpAddr, u8), +} + +impl NoProxy { + /// Comma-separated entries, whitespace trimmed, empties dropped. Entry classification + /// mirrors reqwest: try CIDR, then IP literal, then fall back to domain — a malformed + /// CIDR (`10.0.0.0/40`) therefore degrades to a domain entry that never matches, exactly + /// as `ipnet` parse failure does under reqwest, rather than aborting boot. + fn parse(raw: &str) -> Self { + let mut entries = Vec::new(); + for part in raw.split(',') { + let part = part.trim(); + if part.is_empty() { + continue; + } + if part == "*" { + entries.push(NoProxyEntry::MatchAll); + continue; + } + if let Some((addr, len)) = part.split_once('/') { + if let (Ok(addr), Ok(len)) = (addr.parse::(), len.parse::()) { + let max = if addr.is_ipv4() { 32 } else { 128 }; + if len <= max { + entries.push(NoProxyEntry::Cidr(addr, len)); + continue; + } + } + } else if let Ok(addr) = part.parse::() { + entries.push(NoProxyEntry::Ip(addr)); + continue; + } + entries.push(NoProxyEntry::Domain( + part.trim_start_matches('.').to_ascii_lowercase(), + )); + } + NoProxy { entries } + } + + fn matches(&self, target_host: &str) -> bool { + // A URI's host may carry an IPv6 literal in brackets; strip them before the + // IpAddr parse, as hyper-util's NoProxy::contains does. + let bare = target_host.trim_start_matches('[').trim_end_matches(']'); + let ip: Option = bare.parse().ok(); + let host = target_host.to_ascii_lowercase(); + self.entries.iter().any(|entry| match entry { + NoProxyEntry::MatchAll => true, + NoProxyEntry::Ip(a) => ip == Some(*a), + NoProxyEntry::Cidr(net, len) => ip.is_some_and(|h| cidr_contains(net, *len, &h)), + NoProxyEntry::Domain(d) => { + ip.is_none() + && (host == *d + || (host.len() > d.len() + && host.ends_with(d.as_str()) + && host.as_bytes()[host.len() - d.len() - 1] == b'.')) + } + }) + } +} + +/// CIDR containment by prefix-length bit compare over octets — the ~10 lines that make an +/// `ipnet` dependency unnecessary. Families never cross: a v4 block does not match a v6 +/// address and vice versa, INCLUDING v6-mapped forms (`10.0.0.0/8` does NOT match +/// `::ffff:10.0.0.1` — it is a v6 address), matching `ipnet`/reqwest, which do not +/// special-case mapped addresses either. +fn cidr_contains(net: &IpAddr, prefix: u8, addr: &IpAddr) -> bool { + match (net, addr) { + (IpAddr::V4(n), IpAddr::V4(a)) => octets_prefix_eq(&n.octets(), &a.octets(), prefix), + (IpAddr::V6(n), IpAddr::V6(a)) => octets_prefix_eq(&n.octets(), &a.octets(), prefix), + _ => false, + } +} + +/// The first `prefix` bits of `net` and `addr` are equal. `prefix` is pre-validated against +/// the address width; `/0` matches everything (whole bytes to compare: none, remainder: 0). +fn octets_prefix_eq(net: &[u8], addr: &[u8], prefix: u8) -> bool { + let whole = usize::from(prefix / 8); + let rem = prefix % 8; + if net[..whole] != addr[..whole] { + return false; + } + if rem == 0 { + return true; + } + let mask = 0xffu8 << (8 - rem); + (net[whole] & mask) == (addr[whole] & mask) +} diff --git a/crates/busbar-core-connector/Cargo.toml b/crates/busbar-core-connector/Cargo.toml index a0930e86fa..b632a02ebc 100644 --- a/crates/busbar-core-connector/Cargo.toml +++ b/crates/busbar-core-connector/Cargo.toml @@ -71,6 +71,10 @@ crc32fast = { workspace = true } # PROGRAM is the child's pipes, spawned and killed by the connector (its reactor carries them). tokio = { workspace = true, features = ["net", "time", "rt", "process"] } libc = { workspace = true } +# THE PROXY ENVIRONMENT (`proxy`, row 130 E5a): a proxy variable read as the kernel engine read it +# (`url`), and the `Proxy-Authorization: Basic` credential its userinfo becomes (`base64`). +url = { workspace = true } +base64 = { workspace = true } [features] diff --git a/crates/busbar-core-connector/src/compose.rs b/crates/busbar-core-connector/src/compose.rs index b20df03adb..aeefd57091 100644 --- a/crates/busbar-core-connector/src/compose.rs +++ b/crates/busbar-core-connector/src/compose.rs @@ -223,6 +223,18 @@ pub struct Connection { security_error: Option, /// Why the connection failed, once it has (its stage and the underlying error's own text). cause: Option, + /// THE PROXY TUNNEL a dial through a proxy asks for before anything else rides the socket + /// ([`crate::proxy`]); `None` once the proxy answered, and on every direct connection. + tunnel: Option, +} + +/// A CONNECT exchange in flight: the request (until it is queued), the proxy's answer so far, and +/// the authority asked for (for the refusal's words). +struct Tunnel { + request: Vec, + asked: bool, + head: Vec, + authority: String, } /// THE REFUSAL OF A FAR END WHOSE KEY IS NOT THE PINNED ONE (the transport pin, ARCHITECT 2026-10-03): the handshake's chain @@ -418,6 +430,13 @@ impl Planned { self.located.secure } + /// Whether a framer rides the dial (an application protocol over the stream), not a carrier + /// carried as itself: the dials the proxy environment applies to ([`crate::proxy`]). + #[must_use] + pub fn framed(&self) -> bool { + self.door.is_some() + } + /// The name the entry located for connection security, where it names one. #[must_use] pub fn name(&self) -> Option<&str> { @@ -453,6 +472,33 @@ impl Planned { /// [`Failure::Refused`] when security is asked for and none is set; [`Failure::Failed`] off a /// worker or when the socket cannot be made. pub fn dial_at(self, addr: std::net::SocketAddr) -> Result { + self.dial_via(addr, None) + } + + /// Dial the PROXY at `proxy` and, once it connects, ask it for a tunnel to this plan's + /// authority with `connect` (the CONNECT request, [`crate::proxy::connect_request`]); the + /// secure layer (this plan's name and the target's certificate) and the framing run over the + /// tunnel once the proxy answers 2xx ([`crate::proxy`]). The connection comes back at once, its + /// open in flight. + /// + /// # Errors + /// + /// As [`Self::dial_at`]. + pub fn dial_through( + self, + proxy: std::net::SocketAddr, + connect: Vec, + ) -> Result { + self.dial_via(proxy, Some(connect)) + } + + /// Dial `addr`; with `connect`, `addr` is a proxy and `connect` the tunnel request. + fn dial_via( + self, + addr: std::net::SocketAddr, + connect: Option>, + ) -> Result { + let authority_for_tunnel = self.located.authority.clone(); let Self { door, via, @@ -510,14 +556,21 @@ impl Planned { ) .map_err(dial_failure)?; let offered_name = tls.as_ref().and(located.name.clone()); - Connection::open_dialled( + let mut conn = Connection::open_dialled( door, carried, dial, (tls, presented), offered_name, &located.offer, - ) + )?; + conn.tunnel = connect.map(|request| Tunnel { + request, + asked: false, + head: Vec::new(), + authority: authority_for_tunnel, + }); + Ok(conn) } } @@ -617,6 +670,7 @@ impl Connection { presented, security_error: None, cause: None, + tunnel: None, }) } } @@ -709,6 +763,7 @@ impl Connection { let mut conn = Self { framer_error: None, cause: None, + tunnel: None, security_error: None, key_pin: None, presented: Arc::new(std::sync::atomic::AtomicBool::new(false)), @@ -858,6 +913,7 @@ impl Connection { presented: Arc::new(std::sync::atomic::AtomicBool::new(false)), security_error: None, cause: None, + tunnel: None, }; if conn.tls.is_none() { conn.begin()?; @@ -1310,16 +1366,26 @@ impl Connection { /// moved; `Ok(false)` = nothing can until a waker fires. fn drive(&mut self, cx: &mut Context<'_>) -> Result { let mut moved = false; - if matches!(self.phase, Phase::Connecting) { + if self.tunnel.as_ref().is_some_and(|t| t.asked) { + // THE PROXY'S ANSWER: what the tunnel request owes is written, and the proxy's head is + // read raw (nothing of it reaches the secure layer or the framer). + moved |= self.flush(cx)?; + moved |= self.tunnel_read(cx)?; + } else if matches!(self.phase, Phase::Connecting) { // The carrier's `flush` on a fresh dial: its connect settled (or the far end refused). match self.carried.poll_connected(cx) { Poll::Ready(Ok(())) => { moved = true; - if self.tls.is_some() { - self.phase = Phase::Handshaking; - self.tls_out()?; + if let Some(t) = self.tunnel.as_mut() { + // Connected to the PROXY: ask it for the tunnel first, under the CONNECT + // handshake's own bound. + t.asked = true; + let request = std::mem::take(&mut t.request); + self.queue(&request, false); + self.open_deadline = Some(Instant::now() + crate::proxy::CONNECT_HANDSHAKE); + moved |= self.flush(cx)?; } else { - self.begin()?; + self.connected()?; } } Poll::Ready(Err(e)) => return Err(Failure::Refused(e)), @@ -1337,6 +1403,49 @@ impl Connection { Ok(moved) } + /// The socket reaches the far end: connection security's handshake starts, or the framing + /// begins. + fn connected(&mut self) -> Result<(), Failure> { + if self.tls.is_some() { + self.phase = Phase::Handshaking; + self.tls_out() + } else { + self.begin() + } + } + + /// Read the proxy's answer to the tunnel request: once its head is whole and 2xx the socket is + /// the target's ([`Self::connected`]), and any bytes after the head are the target's own. + fn tunnel_read(&mut self, cx: &mut Context<'_>) -> Result { + let mut buf = [0_u8; 256]; + let n = match self.carried.poll_read(cx, &mut buf) { + Poll::Pending => return Ok(false), + Poll::Ready(Err(e)) => return Err(failed(e)), + Poll::Ready(Ok((0, _))) => { + return Err(Failure::Failed( + "proxy closed the connection during CONNECT".into(), + )) + } + Poll::Ready(Ok((n, _))) => n, + }; + let Some(t) = self.tunnel.as_mut() else { + return Ok(false); + }; + t.head.extend_from_slice(&buf[..n]); + let Some(end) = + crate::proxy::connect_answer(&t.head, &t.authority).map_err(Failure::Refused)? + else { + return Ok(true); + }; + let rest = t.head.split_off(end); + self.tunnel = None; + self.connected()?; + if !rest.is_empty() { + self.feed(&rest, false, false)?; + } + Ok(true) + } + /// The bytes of the pieces read and not yet taken. fn unread_bytes(&self) -> usize { self.inbox.iter().map(|g| g.bytes.len()).sum() diff --git a/crates/busbar-core-connector/src/lib.rs b/crates/busbar-core-connector/src/lib.rs index 34d46a8334..ea86771791 100644 --- a/crates/busbar-core-connector/src/lib.rs +++ b/crates/busbar-core-connector/src/lib.rs @@ -34,7 +34,9 @@ //! [`registry`] is the view of which entry serves which scheme; [`wire`] presents an entry over //! carried connections to the kernel's transport seam. //! * [`process`] builds the process's one connector for the root: the deployment's one destination -//! guard behind the one dial judge, and the default outbound trust. +//! guard behind the one dial judge, the default outbound trust, and the proxy environment. +//! * [`proxy`] is the proxy environment every framed dial honours (`HTTPS_PROXY`, `HTTP_PROXY`, +//! `ALL_PROXY`, `NO_PROXY`), the CONNECT tunnel 1.5.5's clients ran. //! //! No `unsafe` is written here outside the tests' framer entry, which writes a host sink through its //! raw pointers as a plugin does. @@ -59,6 +61,7 @@ pub mod listen; pub mod pool; pub mod process; mod program; +pub mod proxy; pub mod registry; pub mod socket; /// The TLS test kit (far ends, the recording TLS fixture server, a private test CA) for this crate's @@ -327,6 +330,25 @@ struct Judging { /// Writes the caller made before the dial, in order. early: Vec<(Vec, bool, bool)>, answer: Answer, + /// A dial through a proxy ([`proxy`]): the proxy, and whether the target's own judgement has + /// answered (the pended answer is then the proxy's). + proxied: Option, +} + +/// A dial through a proxy, judged in two steps: the target (its name arm, and its own resolution +/// when it resolves here; a name that does not resolve here goes to the proxy, which resolves it), +/// then the proxy's own address (a private proxy admitted, cloud metadata never). +struct Proxied { + spec: Arc, + judged_class: u32, + target_judged: bool, +} + +/// Whether a target judgement's refusal is only that the name does not resolve HERE: behind a +/// proxy, the proxy resolves it (the engine's tunnelled arm; reach behind the proxy is the proxy's +/// boundary). +fn unresolved_here(verdict: Verdict) -> bool { + verdict == DEST_UNRESOLVABLE || verdict == DEST_NO_ADDRESSES } /// What an exchange lent a pooled line keeps to redial it ONCE on a fresh connection when that line @@ -502,6 +524,8 @@ pub struct Connector { authority_held: Mutex>, /// THE HOST'S I/O every carrier is served from ([`hostio`]): the one holder of OS handles. io: Arc, + /// THE PROXY ENVIRONMENT every framed dial honours ([`proxy`]); `None` = every dial direct. + proxy: Option>, } impl std::fmt::Debug for Connector { @@ -527,6 +551,7 @@ impl Default for Connector { auths: Mutex::new(HashMap::new()), authority_held: Mutex::new(HashSet::new()), io: hostio::process(), + proxy: None, } } } @@ -652,6 +677,58 @@ impl Connector { framer::locate(door.as_ref(), target).ok() } + /// The same connector dialling every framed connection through the proxy `config` selects for + /// it ([`proxy`]: the process's `HTTPS_PROXY` / `HTTP_PROXY` / `ALL_PROXY` / `NO_PROXY`, as + /// 1.5.5's clients honoured them); `None` = every dial direct. + #[must_use] + pub fn proxying(mut self, config: Option>) -> Self { + self.proxy = config; + self + } + + /// The proxy choice this connector's dials honour (the process's environment, read once), for + /// the composition root to hand the kernel's remaining egress engine. + #[must_use] + pub fn proxy(&self) -> Option> { + self.proxy.clone() + } + + /// The proxy a dial of `planned` goes through: a framed dial (never a carrier carried as + /// itself) not pinned `within` an address set, to a host the scheme's slot has a proxy for and + /// `NO_PROXY` does not exclude. + fn tunnel_for(&self, planned: &Planned, within: &[IpAddr]) -> Option> { + if !planned.framed() || !within.is_empty() { + return None; + } + self.proxy + .as_ref()? + .select(planned.secure(), proxy::host_of(planned.authority())) + } + + /// Judge `dest` under `class` (with the need's private reach when `reach`), the pended answer + /// written to `answer` and its waker woken. + fn judge_into( + &self, + dest: &str, + class: u32, + reach: bool, + answer: &Answer, + ) -> Option> { + let answering = Arc::clone(answer); + let done: Judged = Box::new(move |v| { + let mut a = answering.lock().expect("judgement"); + a.0 = Some(v); + if let Some(w) = a.1.take() { + w.wake(); + } + }); + if reach { + self.judge.judge_dial_reaching(dest, class, done) + } else { + self.judge.judge_dial(dest, class, done) + } + } + /// The same connector keeping dialled connections under `posture` (the deployment's /// `limits.pool_max_idle_per_host` / `limits.pool_idle_timeout_secs`): an open to a place a /// finished exchange's connection went back to rides that connection ([`pool`]). @@ -1116,10 +1193,57 @@ impl Connector { Some(got) => got, } }; - let addr = got.map_err(pended_verdict)?; let secure = j.planned.as_ref().is_some_and(Planned::secure); - let authority = j.planned.as_ref().map_or("", Planned::authority); - if !class_admits(j.egress_class, secure, &j.within, authority, addr) { + let target = j + .planned + .as_ref() + .map_or_else(String::new, |p| p.authority().to_owned()); + if let Some(p) = j.proxied.as_mut() { + let proxy_addr = if p.target_judged { + got.map_err(pended_verdict)? + } else { + match got { + Err(v) if !unresolved_here(v) => return Err(pended_verdict(v)), + Ok(at) if !class_admits(j.egress_class, secure, &[], &target, at) => { + j.answer.lock().expect("judgement").0 = Some(Err(DEST_PLAINTEXT)); + return Err(ConnError::Refused); + } + _ => {} + } + p.target_judged = true; + { + let mut a = j.answer.lock().expect("judgement"); + a.0 = None; + if let Some(w) = waker { + a.1 = Some(w.clone()); + } + } + match self.judge_into(&p.spec.endpoint(), p.judged_class, true, &j.answer) { + None => return Ok(false), + Some(Err(v)) => { + j.answer.lock().expect("judgement").0 = Some(Err(v)); + return Err(pended_verdict(v)); + } + Some(Ok(at)) => at, + } + }; + let spec = Arc::clone(&p.spec); + let planned = j.planned.take().ok_or(ConnError::Refused)?; + let early = std::mem::take(&mut j.early); + *judging = None; + let connect = proxy::connect_request(planned.authority(), &spec); + let conn = planned + .dial_through(proxy_addr, connect) + .map_err(|f| map(&f))?; + let line = self.ride(held, conn); + for (bytes, end, text) in early { + line.emit(EXCHANGE_STREAM, &bytes, end, text) + .map_err(|f| map(&f))?; + } + return Ok(true); + } + let addr = got.map_err(pended_verdict)?; + if !class_admits(j.egress_class, secure, &j.within, &target, addr) { // The refusal stays the connection's answer. j.answer.lock().expect("judgement").0 = Some(Err(DEST_PLAINTEXT)); return Err(ConnError::Refused); @@ -1164,23 +1288,67 @@ impl Connector { reach: bool, ) -> Result<(Option, Option), ConnError> { let answer: Answer = Arc::new(Mutex::new((None, None))); - let answering = Arc::clone(&answer); - let done: Judged = Box::new(move |v| { - let mut a = answering.lock().expect("judgement"); - a.0 = Some(v); - if let Some(w) = a.1.take() { - w.wake(); - } - }); // A destination the need holds a private reach to is judged with it (the one guard, as an // allowlist entry naming it; the class unchanged). - let judged = if reach { - self.judge - .judge_dial_reaching(planned.authority(), judged_class, done) - } else { - self.judge - .judge_dial(planned.authority(), judged_class, done) - }; + let judged = self.judge_into(planned.authority(), judged_class, reach, &answer); + // THROUGH A PROXY ([`proxy`]): the target judged first, then the proxy's own address. + if let Some(spec) = self.tunnel_for(&planned, within) { + let mut proxied = Proxied { + spec, + judged_class, + target_judged: false, + }; + match judged { + None => { + return Ok(( + None, + Some(Judging { + planned: Some(planned), + egress_class, + within: Vec::new(), + early: Vec::new(), + answer, + proxied: Some(proxied), + }), + )) + } + Some(Err(v)) if !unresolved_here(v) => return Err(ConnError::Refused), + Some(Ok(at)) + if !class_admits( + egress_class, + planned.secure(), + &[], + planned.authority(), + at, + ) => + { + return Err(ConnError::Refused) + } + Some(_) => {} + } + proxied.target_judged = true; + let answer: Answer = Arc::new(Mutex::new((None, None))); + let proxy_authority = proxied.spec.endpoint(); + return match self.judge_into(&proxy_authority, judged_class, true, &answer) { + None => Ok(( + None, + Some(Judging { + planned: Some(planned), + egress_class, + within: Vec::new(), + early: Vec::new(), + answer, + proxied: Some(proxied), + }), + )), + Some(Err(_)) => Err(ConnError::Refused), + Some(Ok(at)) => { + let connect = proxy::connect_request(planned.authority(), &proxied.spec); + let conn = planned.dial_through(at, connect).map_err(|f| map(&f))?; + Ok((Some(conn), None)) + } + }; + } Ok(match judged { // Decided at once: a refusal answers the open, as 1.5.5 answered it. Some(Err(_)) => return Err(ConnError::Refused), @@ -1204,6 +1372,7 @@ impl Connector { within: within.to_vec(), early: Vec::new(), answer, + proxied: None, }), ), }) diff --git a/crates/busbar-core-connector/src/process.rs b/crates/busbar-core-connector/src/process.rs index a9d00aa41e..b8c93aaac9 100644 --- a/crates/busbar-core-connector/src/process.rs +++ b/crates/busbar-core-connector/src/process.rs @@ -27,7 +27,8 @@ use crate::registry::{Entry, Transports}; use crate::{Connector, DialJudge, Judged, Verdict, WakeTicket}; /// THE PROCESS'S CONNECTOR over the framer entries `entries` yields, judging every dial through -/// `dest` ([`judge`], the node's `own_ports` refused to a loopback-allowed need), securing a +/// `dest` ([`judge`], the node's `own_ports` refused to a loopback-allowed need), dialling a framed +/// connection through the proxy the process's environment names ([`crate::proxy`]), securing a /// connection with the default outbound trust where its target asks for it, waking a plugin's /// ticket through `wake`, and keeping a dialled connection whose exchange finished whole under the /// deployment's `pool` posture ([`crate::pool`]). The trust is built before `entries` is asked for, @@ -35,7 +36,8 @@ use crate::{Connector, DialJudge, Judged, Verdict, WakeTicket}; /// /// # Errors /// -/// The default outbound trust will not build, `entries` refuses, two entries claim one scheme, or +/// The default outbound trust will not build, a proxy variable is set and does not parse, +/// `entries` refuses, two entries claim one scheme, or /// an entry states no claim. pub fn build( entries: impl FnOnce() -> Result, String>, @@ -46,9 +48,14 @@ pub fn build( ) -> Result, String> { let tls = crate::tls::client::build_client_config(&EgressTrust::default()) .map_err(|e| format!("its connection security: {e}"))?; + // THE PROXY ENVIRONMENT, read once (row 130 E5a): a set variable that does not parse refuses + // the boot, as the kernel engine's install did. + let proxy = crate::proxy::from_env().map_err(|e| format!("its proxy environment: {e}"))?; let view = Transports::new(entries()?).map_err(|e| e.to_string())?; Ok(Arc::new( - Connector::serving(view, judge(dest, own_ports), Some(Arc::new(tls)), wake).pooling(pool), + Connector::serving(view, judge(dest, own_ports), Some(Arc::new(tls)), wake) + .pooling(pool) + .proxying(proxy), )) } diff --git a/crates/busbar-core-connector/src/proxy.rs b/crates/busbar-core-connector/src/proxy.rs new file mode 100644 index 0000000000..ae8a8b9c98 --- /dev/null +++ b/crates/busbar-core-connector/src/proxy.rs @@ -0,0 +1,281 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! THE PROXY ENVIRONMENT, AS EVERY CONNECTOR DIAL HONOURS IT (status row 130 E5a; ARCHITECT ruling +//! 2026-10-12: the engine's proxy behaviour moves into the connector UNCHANGED, so every dial +//! through the connector honours `HTTPS_PROXY` / `HTTP_PROXY` / `ALL_PROXY` / `NO_PROXY` exactly +//! as 1.5.5's reqwest clients did). +//! +//! This is the ONE place the proxy environment is read (ARCHITECT ruling 2026-10-12, a move, not a +//! copy): the parser is the kernel engine's former one (`busbar-kernel` `egress/engine/mod.rs`, +//! `mod tunnel`), moved verbatim into here, and it fills the shared choice +//! (`busbar_contract::proxy_route`), which the composition root hands the kernel's remaining +//! engine; the kernel reads no environment of its own. The rules: scheme-scoped slots (https targets use +//! `HTTPS_PROXY`, everything else `HTTP_PROXY`, each falling back to `ALL_PROXY`, uppercase read +//! before lowercase), `NO_PROXY` with reqwest's full rule set (host, domain suffix, IP literal, +//! CIDR), `Proxy-Authorization: Basic` from the proxy URL's userinfo, an `https://` proxy refused +//! loudly, a garbage value a boot refusal that never echoes the proxy password. What changes is +//! where the bytes go: the connection ([`crate::compose`]) dials the proxy, writes +//! [`connect_request`] before anything else, reads the proxy's answer ([`connect_answer`]) and only +//! then runs the secure layer (the target's name, the target's certificate) and the framer over the +//! tunnel, so the proxy sees `CONNECT host:port` and never a decrypted byte. Plain-http targets are +//! tunnelled too (CONNECT for everything, as the engine and `curl -p` do; the request bytes are +//! identical either way). +//! +//! A FRAMED dial is tunnelled (an application protocol over the stream, as 1.5.5's HTTP clients +//! were); a carrier carried as itself (a database's or a directory's raw stream) and a program +//! never are, as 1.5.5's drivers never were. + +use std::sync::Arc; + +pub use busbar_contract::proxy_route::{ProxyChoice, ProxyHop}; + +/// The proxy env value as a REFUSAL may render it: userinfo replaced by `***`, everything +/// else verbatim. +/// +/// `HTTPS_PROXY`/`HTTP_PROXY`/`ALL_PROXY` carry `user:password@` (RFC 3986 §3.2.1) in every +/// corporate deployment that authenticates its proxy, and [`parse_proxy`]'s refusals are not +/// swallowed: `resolve_config` hands them to [`from_env`], whose `Err` refuses the boot. Interpolating the raw value put the +/// operator's proxy password on stderr, in the crash report and in whatever CI log collected +/// them. The host, port and scheme are what the operator needs to fix the variable, so they +/// survive — only the credential goes. +/// +/// TEXTUAL, not a re-parse, deliberately: the first caller is the arm where `url::Url::parse` +/// has ALREADY FAILED, so a masker that had to parse could only fail the same way and return +/// the raw value. Userinfo is everything before the LAST `@` of the authority — last, because +/// a password may itself contain an unencoded `@` (`svcacct:hunter2-Pr0xyP@ss@proxy.corp`), +/// and the authority ends at the first `/`, `?` or `#` because userinfo may not contain one. +/// An `@` that appears only AFTER the authority cannot be userinfo, but neither can this +/// function prove the value is well-formed enough for that reasoning to hold, so that residue +/// case redacts wholesale rather than print a maybe-credential. +fn redact_userinfo(v: &str) -> String { + let (scheme, rest) = match v.split_once("://") { + Some((s, r)) => (Some(s), r), + None => (None, v), + }; + let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len()); + let (authority, tail) = rest.split_at(authority_end); + let shown = match authority.rsplit_once('@') { + Some((_userinfo, host_port)) => format!("***@{host_port}{tail}"), + None if rest.contains('@') => "***".to_string(), + None => rest.to_string(), + }; + match scheme { + Some(s) => format!("{s}://{shown}"), + None => shown, + } +} + +/// Parse a proxy env value (`http://[user:pass@]host[:port]`; a bare `host:port` is accepted +/// too). `https://` proxies are refused loudly: TLS-to-proxy is a different transport this +/// tunnel does not speak, and silently downgrading it to a plain stream would be a lie. +/// +/// Every refusal renders the value through [`redact_userinfo`] — see there for why the raw +/// value must never reach the message, and why the host still does. +/// +/// # Errors +/// +/// The value does not read as an `http://` proxy URL with a host. +pub fn parse_proxy(v: &str) -> Result { + let shown = redact_userinfo(v); + let url = if v.contains("://") { + v.to_string() + } else { + format!("http://{v}") + }; + let parsed = url::Url::parse(&url) + .map_err(|e| format!("proxy env value {shown:?} is not a valid URL: {e}"))?; + if parsed.scheme() != "http" { + return Err(format!( + "proxy env value {shown:?} uses scheme {:?}: only plain http:// CONNECT proxies \ + are supported (an https:// proxy would need TLS-to-proxy, which this tunnel does \ + not speak)", + parsed.scheme() + )); + } + let host = parsed + .host_str() + .ok_or_else(|| format!("proxy env value {shown:?} has no host"))? + .to_string(); + let port = parsed.port().unwrap_or(80); + let auth = (!parsed.username().is_empty()).then(|| { + use base64::Engine; + let creds = format!( + "{}:{}", + parsed.username(), + parsed.password().unwrap_or_default() + ); + format!( + "Basic {}", + base64::engine::general_purpose::STANDARD.encode(creds) + ) + }); + Ok(ProxyHop::new(host, port, auth)) +} + +/// The raw proxy env, read once — a plain value bag so precedence resolution +/// ([`resolve_config`]) is a pure function tests exercise WITHOUT mutating process env +/// (parallel tests race on env mutation). +#[derive(Default)] +pub struct ProxyEnvValues { + /// `HTTPS_PROXY` / `https_proxy`. + pub https: Option, + /// `HTTP_PROXY` / `http_proxy`. + pub http: Option, + /// `ALL_PROXY` / `all_proxy`. + pub all: Option, + /// `NO_PROXY` / `no_proxy`. + pub no: Option, +} + +impl ProxyEnvValues { + /// Uppercase read before lowercase, empty values treated as unset — hyper-util + /// `matcher::Builder::from_env`'s `get_first_env(&["HTTPS_PROXY", "https_proxy"])` + /// ordering (empty parses to nothing there; skipping it here is the same outcome). + pub fn from_process_env() -> Self { + fn first(keys: [&str; 2]) -> Option { + keys.iter() + .find_map(|k| std::env::var(k).ok().filter(|v| !v.is_empty())) + } + ProxyEnvValues { + https: first(["HTTPS_PROXY", "https_proxy"]), + // Read unconditionally — hyper-util's matcher additionally skips uppercase + // `HTTP_PROXY` under CGI (`REQUEST_METHOD` set; the httpoxy guard). busbar is a + // server binary, never a CGI child, so the guard's precondition cannot hold and + // the unconditional read is behavior-identical where this process runs. + http: first(["HTTP_PROXY", "http_proxy"]), + all: first(["ALL_PROXY", "all_proxy"]), + no: first(["NO_PROXY", "no_proxy"]), + } + } +} + +/// Resolve the value bag into the per-scheme slots, reqwest's precedence: scheme-specific +/// var first, `ALL_PROXY` as the fallback for BOTH slots (hyper-util `Builder::build`: +/// `http: http.or(all)`, `https: https.or(all)`). No proxy var at all → `None`, the direct +/// arm. Any PRESENT value that does not parse is a hard error — reqwest silently ignored +/// garbage and fell through, but silently egressing direct past a configured proxy is the +/// dangerous direction, so busbar refuses to boot instead (documented deviation). +/// +/// # Errors +/// +/// A set value that does not parse ([`parse_proxy`]). +pub fn resolve_config(env: &ProxyEnvValues) -> Result>, String> { + let all = env + .all + .as_deref() + .map(parse_proxy) + .transpose()? + .map(Arc::new); + let scheme_slot = |v: &Option| -> Result>, String> { + Ok(match v.as_deref() { + Some(v) => Some(Arc::new(parse_proxy(v)?)), + None => all.clone(), + }) + }; + let https = scheme_slot(&env.https)?; + let http = scheme_slot(&env.http)?; + if https.is_none() && http.is_none() { + return Ok(None); + } + Ok(Some(Arc::new(ProxyChoice::new( + https, + http, + env.no.as_deref().unwrap_or(""), + )))) +} + +/// THE PROCESS'S PROXY CHOICE, read once from its environment: `None` when no proxy variable is +/// set (the direct arm every known deployment takes). +/// +/// # Errors +/// +/// A proxy variable that is set and does not parse: the boot refuses (fail-loud beats silently +/// egressing direct past a configured proxy), the password never in the text. +pub fn from_env() -> Result>, String> { + resolve_config(&ProxyEnvValues::from_process_env()) +} + +/// Hard ceiling on the proxy's CONNECT response head; a real response is one status line + a few +/// headers. Anything larger is a misbehaving proxy and fails the connect. +pub const CONNECT_HEAD_CAP: usize = 8 * 1024; + +/// Wall-clock bound on the CONNECT handshake — the same 10s the socket connect itself carries. +pub const CONNECT_HANDSHAKE: std::time::Duration = std::time::Duration::from_secs(10); + +/// `CONNECT host:port HTTP/1.1` (+ Host, + Proxy-Authorization when configured), the bytes the +/// engine's `connect_handshake` wrote, for `target` (`host:port`, a v6 host bracketed). +#[must_use] +pub fn connect_request(target: &str, proxy: &ProxyHop) -> Vec { + let mut req = format!("CONNECT {target} HTTP/1.1\r\nHost: {target}\r\n"); + if let Some(auth) = proxy.proxy_authorization() { + req.push_str("Proxy-Authorization: "); + req.push_str(auth); + req.push_str("\r\n"); + } + req.push_str("\r\n"); + req.into_bytes() +} + +/// What the proxy has answered so far to a CONNECT for `addr` (`host:port`): `Ok(None)` = its head is not +/// complete yet; `Ok(Some(n))` = the tunnel is up and the head is the first `n` bytes (anything +/// after them is the target's); `Err` = the proxy refused (any status but 2xx: RFC 9110, 2xx means +/// the tunnel is established) or its head ran past [`CONNECT_HEAD_CAP`]. +/// +/// # Errors +/// +/// The refusal, in the engine's words. +pub fn connect_answer(head: &[u8], addr: &str) -> Result, String> { + let Some(at) = head.windows(4).position(|w| w == b"\r\n\r\n") else { + if head.len() > CONNECT_HEAD_CAP { + return Err("proxy CONNECT response head exceeded 8KiB".into()); + } + return Ok(None); + }; + // Status line: `HTTP/1.x NNN ...` — accept any 2xx (some proxies say 200 Connection + // established, some plain 200 OK). + let line = head.split(|&b| b == b'\r').next().unwrap_or_default(); + let status_2xx = line + .split(|&b| b == b' ') + .nth(1) + .is_some_and(|code| code.first() == Some(&b'2') && code.len() == 3); + if !status_2xx { + return Err(format!( + "proxy refused CONNECT {addr}: {}", + String::from_utf8_lossy(line) + )); + } + Ok(Some(at + 4)) +} + +/// The host of `target` (`host:port`, a v6 host bracketed), as `NO_PROXY` matches it. +#[must_use] +pub fn host_of(target: &str) -> &str { + if target.starts_with('[') { + return target.find(']').map_or(target, |end| &target[..=end]); + } + target.rsplit_once(':').map_or(target, |(host, _)| host) +} + +/// A whole choice whose BOTH slots point at one proxy at `host:port`: what a test wires directly +/// so parallel tests never race an environment variable. +#[cfg(test)] +pub(crate) fn test_config( + host: &str, + port: u16, + auth: Option, + no_proxy: &str, +) -> Arc { + let hop = Arc::new(ProxyHop::new(host.to_string(), port, auth)); + Arc::new(ProxyChoice::new(Some(hop.clone()), Some(hop), no_proxy)) +} + +/// The exemption list alone, as tests read it. +#[cfg(test)] +pub(crate) fn no_proxy_matches(entries: &str, target_host: &str) -> bool { + ProxyChoice::new(None, None, entries).exempts(target_host) +} + +#[cfg(test)] +#[path = "tests/proxy_tests.rs"] +mod tests; diff --git a/crates/busbar-core-connector/src/tests/proxy_tests.rs b/crates/busbar-core-connector/src/tests/proxy_tests.rs new file mode 100644 index 0000000000..136ae3835a --- /dev/null +++ b/crates/busbar-core-connector/src/tests/proxy_tests.rs @@ -0,0 +1,496 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! THE PROXY ENVIRONMENT, AS THE CONNECTOR HONOURS IT (status row 130 E5a), held to 1.5.5's +//! semantics: reqwest 0.12's implicit proxy environment, as the kernel engine's owned tunnel +//! re-provided it (`busbar-kernel` `egress/engine/tests/engine_tests.rs`, whose rows these port): +//! per-scheme slot resolution (`HTTPS_PROXY` / `HTTP_PROXY`, `ALL_PROXY` as the fallback for both), +//! scheme-scoped selection, the full `NO_PROXY` rule set (host, domain suffix, IP literal, CIDR), +//! an `https://` proxy and a garbage value refused without echoing the proxy password; and, end to +//! end through [`Connector::open`], a framed dial that writes `CONNECT host:port` (+ Basic +//! credential) to the proxy before anything else and then carries the exchange over the tunnel. Every +//! proxy row drives the parse / resolve / select functions or a connector wired with a config +//! directly: NEVER by mutating process environment, which races under the parallel test runner. + +use std::sync::Arc; + +use busbar_contract::conn::{ConnError, Conns, NeedId, OpenDesc, Ticket}; + +use super::*; +use crate::registry::{Entry, Transports}; +use crate::support::{worker, TestDoor}; +use crate::Connector; + +const OWNER: busbar_contract::conn::InstanceId = busbar_contract::conn::InstanceId(1); + +/// The selected proxy's `host:port` for a target, or `None` for the direct arm. +fn selected(config: &Arc, https: bool, host: &str) -> Option { + config.select(https, host).map(|p| p.endpoint()) +} + +/// The pure value bag, as the environment would fill it. +fn env( + https: Option<&str>, + http: Option<&str>, + all: Option<&str>, + no: Option<&str>, +) -> ProxyEnvValues { + ProxyEnvValues { + https: https.map(String::from), + http: http.map(String::from), + all: all.map(String::from), + no: no.map(String::from), + } +} + +#[test] +fn proxy_url_parse_extracts_auth_and_refuses_https_proxies() { + let err = parse_proxy("https://proxy.corp:3128").unwrap_err(); + assert!( + err.contains("https"), + "an https:// proxy must be refused loudly, naming why: {err}" + ); + assert!(parse_proxy("http://").is_err(), "garbage must not parse"); + assert!(parse_proxy("proxy.corp:3128").is_ok()); + assert!(parse_proxy("http://u:p@proxy.corp").is_ok()); + // The Basic credential rides the CONNECT exactly as the engine wrote it. + let spec = parse_proxy("http://u:p@proxy.corp").unwrap(); + assert_eq!( + String::from_utf8(connect_request("upstream.test:443", &spec)).unwrap(), + "CONNECT upstream.test:443 HTTP/1.1\r\nHost: upstream.test:443\r\n\ + Proxy-Authorization: Basic dTpw\r\n\r\n" + ); + assert_eq!( + spec.endpoint(), + "proxy.corp:80", + "a proxy URL's port defaults to 80" + ); +} + +/// (HTTPS_PROXY, HTTP_PROXY, ALL_PROXY) → (https slot, http slot), reqwest's precedence. +#[test] +fn resolve_config_matches_reqwest_precedence() { + type Row = ( + Option<&'static str>, + Option<&'static str>, + Option<&'static str>, + Option<&'static str>, + Option<&'static str>, + ); + let rows: &[Row] = &[ + (None, None, None, None, None), + (Some("hs.corp:1"), None, None, Some("hs.corp:1"), None), + (None, Some("h.corp:2"), None, None, Some("h.corp:2")), + ( + None, + None, + Some("all.corp:3"), + Some("all.corp:3"), + Some("all.corp:3"), + ), + ( + Some("hs.corp:1"), + None, + Some("all.corp:3"), + Some("hs.corp:1"), + Some("all.corp:3"), + ), + ( + None, + Some("h.corp:2"), + Some("all.corp:3"), + Some("all.corp:3"), + Some("h.corp:2"), + ), + ( + Some("hs.corp:1"), + Some("h.corp:2"), + Some("all.corp:3"), + Some("hs.corp:1"), + Some("h.corp:2"), + ), + ]; + for (https, http, all, want_https, want_http) in rows { + let config = resolve_config(&env(*https, *http, *all, None)).unwrap(); + match (want_https, want_http) { + (None, None) => assert!( + config.is_none(), + "no proxy vars must resolve to no config: {https:?}/{http:?}/{all:?}" + ), + _ => { + let config = config.expect("some slot is set, a config must exist"); + assert_eq!( + selected(&config, true, "api.example.com").as_deref(), + *want_https, + "https slot for {https:?}/{http:?}/{all:?}" + ); + assert_eq!( + selected(&config, false, "api.example.com").as_deref(), + *want_http, + "http slot for {https:?}/{http:?}/{all:?}" + ); + } + } + } +} + +#[test] +fn resolve_config_fail_louds_on_garbage_in_any_slot() { + for (https, http, all) in [ + (Some("https://tls-proxy.corp"), None, None), + (None, Some("http://"), None), + (None, None, Some("https://tls-proxy.corp")), + ] { + assert!( + resolve_config(&env(https, http, all, None)).is_err(), + "garbage in {https:?}/{http:?}/{all:?} must refuse to resolve" + ); + } +} + +#[test] +fn no_proxy_suffix_matching_is_conventional() { + let config = test_config("proxy.corp", 3128, None, "example.com, internal"); + let via = |host: &str| selected(&config, true, host).is_some(); + assert!(!via("example.com"), "exact NO_PROXY entry bypasses"); + assert!(!via("api.example.com"), "domain suffix bypasses"); + assert!( + via("notexample.com"), + "a non-boundary suffix must NOT bypass (substring is not a domain match)" + ); + assert!(!via("INTERNAL"), "matching is case-insensitive"); + assert!(via("api.upstream.test"), "unlisted hosts tunnel"); + let all = test_config("proxy.corp", 3128, None, "*"); + assert!( + selected(&all, true, "api.upstream.test").is_none(), + "`*` disables tunneling" + ); + assert!(no_proxy_matches(".example.com", "example.com")); + assert!(no_proxy_matches(".example.com", "api.example.com")); +} + +#[test] +fn no_proxy_ip_and_cidr_matrix() { + let m = no_proxy_matches; + assert!(m("10.1.2.3", "10.1.2.3")); + assert!(!m("10.1.2.3", "10.1.2.30")); + assert!(!m("10.1.2.3", "110.1.2.3")); + assert!(m("::1", "::1")); + assert!(m("::1", "[::1]"), "bracketed v6 literal hosts match too"); + assert!(m("10.0.0.0/8", "10.255.255.255")); + assert!(m("10.0.0.0/8", "10.0.0.1")); + assert!(!m("10.0.0.0/8", "11.0.0.0")); + assert!(!m("10.0.0.0/8", "9.255.255.255")); + assert!(m("0.0.0.0/0", "203.0.113.7"), "/0 matches every v4 address"); + assert!(m("192.168.1.42/32", "192.168.1.42"), "/32 is exact"); + assert!(!m("192.168.1.42/32", "192.168.1.43")); + assert!(m("192.168.1.0/25", "192.168.1.127")); + assert!(!m("192.168.1.0/25", "192.168.1.128")); + assert!(m("fd00::/8", "fd12:3456::1")); + assert!(!m("fd00::/8", "fe80::1")); + assert!(m("::/0", "2001:db8::1"), "/0 matches every v6 address"); + assert!(m("2001:db8::1/128", "2001:db8::1"), "/128 is exact"); + assert!(!m("2001:db8::1/128", "2001:db8::2")); + assert!(!m("10.0.0.0/8", "::ffff:10.0.0.1")); + assert!(!m("::/0", "10.0.0.1"), "a v6 catch-all does not catch v4"); + assert!( + !m("0.0.0.0/0", "2001:db8::1"), + "a v4 catch-all does not catch v6" + ); + assert!(!m("0.0.1", "10.0.0.1")); + assert!(!m("10.0.0.0/40", "10.0.0.1")); + assert!(m("10.0.0.0/40, example.com", "example.com")); +} + +#[test] +fn selection_is_scheme_scoped_and_no_proxy_excludes_both() { + let config = resolve_config(&env( + Some("hs.corp:1"), + Some("h.corp:2"), + None, + Some("internal.corp, 10.0.0.0/8"), + )) + .unwrap() + .unwrap(); + assert_eq!( + selected(&config, true, "api.upstream.test").as_deref(), + Some("hs.corp:1") + ); + assert_eq!( + selected(&config, false, "api.upstream.test").as_deref(), + Some("h.corp:2") + ); + for https in [true, false] { + assert!(selected(&config, https, "db.internal.corp").is_none()); + assert!(selected(&config, https, "10.20.30.40").is_none()); + } +} + +#[test] +fn proxy_env_refusals_never_echo_the_proxy_password() { + const PASSWORD: &str = "hunter2-Pr0xyP@ss"; + let values = [ + "https://svcacct:hunter2-Pr0xyP@ss@proxy.corp:3128", + "http://svcacct:hunter2-Pr0xyP@ss@proxy.corp:notaport", + ]; + for v in values { + let err = parse_proxy(v).expect_err("both values must be refused"); + assert!(!err.contains(PASSWORD) && !err.contains("hunter2"), "{err}"); + assert!(err.contains("proxy.corp"), "the host survives: {err}"); + } + for slot in 0..3 { + let bad = Some(values[0]); + let err = resolve_config(&env( + bad.filter(|_| slot == 0), + bad.filter(|_| slot == 1), + bad.filter(|_| slot == 2), + None, + )) + .expect_err("a garbage proxy value must refuse to resolve"); + assert!( + !err.contains(PASSWORD) && !err.contains("hunter2"), + "slot {slot}: {err}" + ); + } +} + +/// The proxy's answer is read whole before the tunnel is up: an incomplete head waits, a 2xx +/// head is the tunnel (and only its own bytes are the head's), any other status refuses in the +/// engine's words, and an endless head refuses at 8 KiB. +#[test] +fn the_proxys_answer_is_judged_on_its_status_line() { + assert_eq!(connect_answer(b"HTTP/1.1 200 Conn", "t:1"), Ok(None)); + let head = b"HTTP/1.1 200 Connection established\r\n\r\n"; + assert_eq!(connect_answer(head, "t:1"), Ok(Some(head.len()))); + let mut with_target_bytes = head.to_vec(); + with_target_bytes.extend_from_slice(b"\x16\x03"); + assert_eq!( + connect_answer(&with_target_bytes, "t:1"), + Ok(Some(head.len())) + ); + assert_eq!( + connect_answer(b"HTTP/1.0 204 OK\r\n\r\n", "t:1"), + Ok(Some(19)) + ); + assert_eq!( + connect_answer(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n", "t:1"), + Err("proxy refused CONNECT t:1: HTTP/1.1 407 Proxy Authentication Required".into()) + ); + assert!(connect_answer(&[b'x'; CONNECT_HEAD_CAP + 1], "t:1").is_err()); + assert_eq!(host_of("upstream.test:8080"), "upstream.test"); + assert_eq!(host_of("[::1]:8080"), "[::1]"); +} + +// ── end to end, through the connector ── + +/// A framer over the test carrier, the dial judged by a judge that pins IP literals and finds no +/// name (so a named target is one only the proxy resolves), proxying under `config`. +fn proxying(config: Option>) -> Connector { + let view = Transports::new(vec![ + Entry { + door: Arc::new( + TestDoor::identity("bytes").with_role(busbar_contract::abi::transport::ROLE_FRAMER), + ), + alpn: Vec::new(), + }, + Entry { + door: Arc::new(TestDoor::identity("carrier")), + alpn: Vec::new(), + }, + ]) + .unwrap(); + let judge = |dest: &str, _: u32, _: crate::Judged| { + Some( + crate::socket::address_of(dest) + .ok_or(busbar_contract::abi::host::service::DEST_UNRESOLVABLE), + ) + }; + Connector::serving(view, Arc::new(judge), None, Arc::new(|_: Ticket| {})).proxying(config) +} + +/// A scripted proxy on loopback: it records the CONNECT head, answers `answer`, and (on a 2xx) +/// acts as the target on the same socket, answering `ping` with `pong`. +async fn scripted_proxy(answer: &'static [u8]) -> (u16, Arc>) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let seen = Arc::new(std::sync::Mutex::new(String::new())); + let seen2 = Arc::clone(&seen); + tokio::spawn(async move { + let Ok((mut sock, _)) = listener.accept().await else { + return; + }; + let mut head = Vec::new(); + let mut buf = [0_u8; 512]; + while !head.windows(4).any(|w| w == b"\r\n\r\n") { + let n = sock.read(&mut buf).await.unwrap(); + if n == 0 { + return; + } + head.extend_from_slice(&buf[..n]); + } + *seen2.lock().unwrap() = String::from_utf8_lossy(&head).into_owned(); + sock.write_all(answer).await.unwrap(); + let mut ping = [0_u8; 4]; + if sock.read_exact(&mut ping).await.is_ok() && &ping == b"ping" { + let _ = sock.write_all(b"pong").await; + } + }); + (port, seen) +} + +/// Read until `want` arrived on `id`, or the connection answered an error. +async fn read_until( + c: &Connector, + id: busbar_contract::conn::ConnId, + want: &[u8], +) -> Result, ConnError> { + let mut got = Vec::new(); + let mut buf = [0_u8; 64]; + for _ in 0..5000 { + match c.read(OWNER, id, 7, &mut buf) { + Err(ConnError::Pending) => { + tokio::time::sleep(std::time::Duration::from_millis(1)).await + } + Err(e) => return Err(e), + Ok(p) => { + got.extend_from_slice(&buf[..p.len]); + if got.windows(want.len()).any(|w| w == want) { + return Ok(got); + } + } + } + } + panic!("no answer in five seconds: {got:?}"); +} + +/// THE PROOF: a framed dial to a name only the proxy resolves goes to the proxy, asks it for the +/// tunnel first (`CONNECT host:port`, `Host`, the Basic credential, exactly the engine's bytes), +/// and the exchange then rides the tunnel: the target's answer comes back over the same socket. +#[test] +fn a_framed_dial_rides_a_connect_tunnel_through_the_proxy() { + worker().block_on(async { + let (port, seen) = scripted_proxy(b"HTTP/1.1 200 Connection established\r\n\r\n").await; + let c = proxying(Some(test_config( + "127.0.0.1", + port, + Some("Basic dTpw".to_string()), + "", + ))); + c.declare_over(OWNER, NeedId(0), "bytes").expect("declares"); + let id = c + .open( + OWNER, + NeedId(0), + &OpenDesc { + target: "upstream.test:8080", + ..OpenDesc::default() + }, + ) + .expect("a name the proxy resolves opens"); + assert_eq!(c.write(OWNER, id, b"ping", false, false), Ok(4)); + let got = read_until(&c, id, b"pong") + .await + .expect("the target answered"); + assert!(got.ends_with(b"pong"), "{got:?}"); + assert_eq!( + seen.lock().unwrap().as_str(), + "CONNECT upstream.test:8080 HTTP/1.1\r\nHost: upstream.test:8080\r\n\ + Proxy-Authorization: Basic dTpw\r\n\r\n", + "the proxy saw the CONNECT for the real target, and nothing before it" + ); + c.close(OWNER, id).unwrap(); + }); +} + +/// RED: a proxy that answers anything but 2xx refuses the exchange; nothing reaches a target. +#[test] +fn a_proxy_refusing_the_tunnel_refuses_the_exchange() { + worker().block_on(async { + let (port, seen) = + scripted_proxy(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n").await; + let c = proxying(Some(test_config("127.0.0.1", port, None, ""))); + c.declare_over(OWNER, NeedId(0), "bytes").expect("declares"); + let id = c + .open( + OWNER, + NeedId(0), + &OpenDesc { + target: "upstream.test:8080", + ..OpenDesc::default() + }, + ) + .expect("opens"); + assert_eq!( + read_until(&c, id, b"pong").await, + Err(ConnError::Refused), + "the proxy's refusal is the exchange's" + ); + assert!(seen + .lock() + .unwrap() + .starts_with("CONNECT upstream.test:8080 HTTP/1.1\r\n")); + }); +} + +/// RED: a host `NO_PROXY` names is dialled direct, and a carrier carried as itself (a driver's raw +/// stream) is never tunnelled; the proxy is never reached. With no proxy configured, a name only a +/// proxy could resolve is refused (the direct arm judges it). +#[test] +fn no_proxy_and_a_raw_stream_dial_direct() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + worker().block_on(async { + let (port, seen) = scripted_proxy(b"HTTP/1.1 200 OK\r\n\r\n").await; + let config = Some(test_config("127.0.0.1", port, None, "127.0.0.1")); + for (need, scheme) in [(NeedId(0), "bytes"), (NeedId(1), "carrier")] { + let far = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let target = far.local_addr().unwrap().to_string(); + tokio::spawn(async move { + let (mut s, _) = far.accept().await.unwrap(); + let mut ping = [0_u8; 4]; + s.read_exact(&mut ping).await.unwrap(); + s.write_all(b"pong").await.unwrap(); + }); + // The raw-stream need's proxy config names no exclusion: only its being a carrier + // keeps it direct. + let c = if scheme == "carrier" { + proxying(Some(test_config("127.0.0.1", port, None, ""))) + } else { + proxying(config.clone()) + }; + c.declare_over(OWNER, need, scheme).expect("declares"); + let id = c + .open( + OWNER, + need, + &OpenDesc { + target: &target, + ..OpenDesc::default() + }, + ) + .expect("opens"); + assert_eq!(c.write(OWNER, id, b"ping", false, false), Ok(4)); + read_until(&c, id, b"pong") + .await + .expect("the far end answered direct"); + c.close(OWNER, id).unwrap(); + } + assert!(seen.lock().unwrap().is_empty(), "the proxy was never asked"); + let direct = proxying(None); + direct + .declare_over(OWNER, NeedId(0), "bytes") + .expect("declares"); + assert_eq!( + direct.open( + OWNER, + NeedId(0), + &OpenDesc { + target: "upstream.test:8080", + ..OpenDesc::default() + }, + ), + Err(ConnError::Refused), + "with no proxy, a name nothing here resolves is refused" + ); + }); +} diff --git a/crates/busbar-kernel/src/egress/engine/mod.rs b/crates/busbar-kernel/src/egress/engine/mod.rs index 9d1041172e..9e093f65e8 100644 --- a/crates/busbar-kernel/src/egress/engine/mod.rs +++ b/crates/busbar-kernel/src/egress/engine/mod.rs @@ -33,20 +33,14 @@ //! * TCP: keepalive 60s + nodelay, same values, same rationale (delayed-ACK tail spike). //! * redirects: hyper never follows redirects — the SSRF guard reqwest needed a policy for is //! now structural. -//! * proxy env: reqwest honored `HTTP_PROXY`/`HTTPS_PROXY`/`ALL_PROXY`/`NO_PROXY` implicitly -//! (undocumented; hyper-util's env matcher under the hood). Re-provided SCHEME-SCOPED as an -//! owned CONNECT tunnel (owner-ruled): https:// targets use `HTTPS_PROXY`, http:// targets -//! use `HTTP_PROXY`, each falling back to `ALL_PROXY`, uppercase read before lowercase — -//! reqwest's exact precedence. When the target's scheme has a proxy and the host is not -//! `NO_PROXY`-excluded, the connect goes TCP-to-proxy + `CONNECT host:port` + (for https -//! targets) TLS over the tunnel — the same layering reqwest used — with -//! `Proxy-Authorization: Basic` from the proxy URL's userinfo. `NO_PROXY` matches -//! hosts/domain-suffixes AND IP literals AND CIDR blocks, reqwest's full rule set. One -//! documented deviation: plain-http targets are ALSO tunneled via CONNECT (reqwest forwarded -//! them absolute-form); CONNECT-for-everything is what curl -p does and keeps request bytes -//! identical either way. Deployments with no proxy env — every known one — take the direct -//! arm, a `None` check per CONNECT (not per request; the pool reuses tunneled sockets like -//! any other). +//! * proxy env: re-provided as an owned CONNECT tunnel. The environment itself is read in ONE +//! place, the connector (`busbar_core_connector::proxy`, status row 130 E5a), into the +//! shared choice the composition root installs here ([`install_proxy`]); this engine reads +//! no environment. When the choice names a proxy for the target, the connect goes to the +//! proxy + `CONNECT host:port` + (for a secured target) TLS over the tunnel, with the +//! choice's `Proxy-Authorization`. Deployments with no proxy — every known one — take the +//! direct arm, a `None` check per CONNECT (not per request; the pool reuses tunneled sockets +//! like any other). //! * decompression: none before (no gzip/brotli features), none now. use std::net::IpAddr; @@ -144,8 +138,8 @@ pub struct EngineSpec { /// CONNECT-tunnel posture (see [`EngineSpec::proxy`]). pub enum ProxyPosture { - /// Honor the proxy env resolved at boot (`install_proxy_tunnel_if_configured`) — the pooled - /// posture's reqwest-parity behavior. + /// Honor the proxy choice installed at boot ([`install_proxy`], the connector's parse of the + /// proxy environment) — the pooled posture's reqwest-parity behavior. BootEnv, /// Never tunnel: the tunnel arm is structurally absent (`TunnelConnector::new(http, None)`). Direct, @@ -615,7 +609,7 @@ mod observe_tests; #[path = "tests/tls_tests.rs"] mod tls_tests; -pub use tunnel::install_proxy_tunnel_if_configured; +pub use tunnel::install_proxy; /// The owned CONNECT tunnel for the proxy-env parity case (owner-ruled: full tunnel, not the /// boot-refusal interim). Constructed ONLY when a proxy env var is present at boot; the direct @@ -624,314 +618,27 @@ mod tunnel { use std::net::IpAddr; use std::sync::{Arc, OnceLock}; - /// One boot-resolved proxy endpoint: where to TCP-connect and what to put in - /// `Proxy-Authorization`. Parsed ONCE at boot; shared by refcount from the per-scheme slots - /// of [`ProxyConfig`]. - #[cfg_attr(test, derive(Debug))] // tests unwrap_err() around it; production never prints it - pub struct ProxySpec { - /// Proxy endpoint as the plain `host:port` CONNECT dial string. - host: String, - port: u16, - /// `Proxy-Authorization: Basic ` prebuilt from the proxy URL's userinfo. - auth: Option, - } - - /// The boot-resolved proxy DECISION, reqwest-scoped: https:// targets use the `https` slot - /// (`HTTPS_PROXY`, `ALL_PROXY` fallback), everything else uses the `http` slot (`HTTP_PROXY`, - /// `ALL_PROXY` fallback), and `NO_PROXY` excludes hosts from both. This mirrors hyper-util's - /// `client::proxy::matcher::Builder::build` (`http: http.or(all)`, `https: https.or(all)`) - /// and `Matcher::intercept` (NO_PROXY first, then the slot picked by `dst.scheme_str()`) — - /// the machinery reqwest 0.12 delegates its implicit env behavior to. - #[cfg_attr(test, derive(Debug))] - pub struct ProxyConfig { - https: Option>, - http: Option>, - no_proxy: NoProxy, - } - - impl ProxyConfig { - /// The per-CONNECT decision: `None` → direct (NO_PROXY-excluded, or no proxy configured - /// for the target's scheme), `Some` → tunnel through that proxy. - fn select(&self, dst_is_https: bool, target_host: &str) -> Option> { - if self.no_proxy.matches(target_host) { - return None; - } - if dst_is_https { - self.https.clone() - } else { - self.http.clone() - } - } - } - - /// Parsed `NO_PROXY` entries, reqwest's full rule set (hyper-util `matcher::NoProxy`, which - /// documents itself as curl's rules): `*` matches everything; an IP literal matches exactly; - /// a CIDR block (`10.0.0.0/8`, `fd00::/8`) matches by containment; anything else is a domain - /// matched exactly or as a `.`-boundary suffix. Like reqwest, an IP-literal target host is - /// checked ONLY against the IP/CIDR entries and a domain host ONLY against the domain - /// entries. - #[cfg_attr(test, derive(Debug))] - struct NoProxy { - entries: Vec, - } - - #[cfg_attr(test, derive(Debug))] - enum NoProxyEntry { - MatchAll, - /// Lowercased, leading dot stripped (`.google.com` ≡ `google.com`, per curl/reqwest). - Domain(String), - Ip(IpAddr), - /// Network address + prefix length, validated ≤32 (v4) / ≤128 (v6) at parse. - Cidr(IpAddr, u8), - } - - impl NoProxy { - /// Comma-separated entries, whitespace trimmed, empties dropped. Entry classification - /// mirrors reqwest: try CIDR, then IP literal, then fall back to domain — a malformed - /// CIDR (`10.0.0.0/40`) therefore degrades to a domain entry that never matches, exactly - /// as `ipnet` parse failure does under reqwest, rather than aborting boot. - fn parse(raw: &str) -> Self { - let mut entries = Vec::new(); - for part in raw.split(',') { - let part = part.trim(); - if part.is_empty() { - continue; - } - if part == "*" { - entries.push(NoProxyEntry::MatchAll); - continue; - } - if let Some((addr, len)) = part.split_once('/') { - if let (Ok(addr), Ok(len)) = (addr.parse::(), len.parse::()) { - let max = if addr.is_ipv4() { 32 } else { 128 }; - if len <= max { - entries.push(NoProxyEntry::Cidr(addr, len)); - continue; - } - } - } else if let Ok(addr) = part.parse::() { - entries.push(NoProxyEntry::Ip(addr)); - continue; - } - entries.push(NoProxyEntry::Domain( - part.trim_start_matches('.').to_ascii_lowercase(), - )); - } - NoProxy { entries } - } + /// THE PROXY CHOICE IS NOT READ HERE (status row 130 E5a; ARCHITECT ruling 2026-10-12): the + /// proxy environment is parsed in ONE place, the connector (`busbar_core_connector::proxy`), into + /// the shared [`ProxyChoice`]; the composition root hands this engine that value at boot + /// ([`install_proxy`]), and the engine reads no environment of its own. + use busbar_contract::proxy_route::{ProxyChoice, ProxyHop}; - fn matches(&self, target_host: &str) -> bool { - // `http::Uri::host()` may carry an IPv6 literal in brackets; strip them before the - // IpAddr parse, as hyper-util's NoProxy::contains does. - let bare = target_host.trim_start_matches('[').trim_end_matches(']'); - let ip: Option = bare.parse().ok(); - let host = target_host.to_ascii_lowercase(); - self.entries.iter().any(|entry| match entry { - NoProxyEntry::MatchAll => true, - NoProxyEntry::Ip(a) => ip == Some(*a), - NoProxyEntry::Cidr(net, len) => ip.is_some_and(|h| cidr_contains(net, *len, &h)), - NoProxyEntry::Domain(d) => { - ip.is_none() - && (host == *d - || (host.len() > d.len() - && host.ends_with(d.as_str()) - && host.as_bytes()[host.len() - d.len() - 1] == b'.')) - } - }) - } - } + /// The boot-installed proxy choice, `None` in every deployment without a proxy environment. + /// `OnceLock` so config applies rebuild clients against the SAME boot choice — the process + /// environment, immutable for the process lifetime, exactly reqwest's read-once behavior. + static INSTALLED: OnceLock>> = OnceLock::new(); - /// CIDR containment by prefix-length bit compare over octets — the ~10 lines that make an - /// `ipnet` dependency unnecessary. Families never cross: a v4 block does not match a v6 - /// address and vice versa, INCLUDING v6-mapped forms (`10.0.0.0/8` does NOT match - /// `::ffff:10.0.0.1` — it is a v6 address), matching `ipnet`/reqwest, which do not - /// special-case mapped addresses either. - fn cidr_contains(net: &IpAddr, prefix: u8, addr: &IpAddr) -> bool { - match (net, addr) { - (IpAddr::V4(n), IpAddr::V4(a)) => octets_prefix_eq(&n.octets(), &a.octets(), prefix), - (IpAddr::V6(n), IpAddr::V6(a)) => octets_prefix_eq(&n.octets(), &a.octets(), prefix), - _ => false, - } + /// Install the process's proxy choice — the connector's, handed over by the composition root + /// at boot, before any client is built. First install wins. + pub fn install_proxy(choice: Option>) { + let _ = INSTALLED.set(choice); } - /// The first `prefix` bits of `net` and `addr` are equal. `prefix` is pre-validated against - /// the address width; `/0` matches everything (whole bytes to compare: none, remainder: 0). - fn octets_prefix_eq(net: &[u8], addr: &[u8], prefix: u8) -> bool { - let whole = usize::from(prefix / 8); - let rem = prefix % 8; - if net[..whole] != addr[..whole] { - return false; - } - if rem == 0 { - return true; - } - let mask = 0xffu8 << (8 - rem); - (net[whole] & mask) == (addr[whole] & mask) - } - - /// The proxy env value as a REFUSAL may render it: userinfo replaced by `***`, everything - /// else verbatim. - /// - /// `HTTPS_PROXY`/`HTTP_PROXY`/`ALL_PROXY` carry `user:password@` (RFC 3986 §3.2.1) in every - /// corporate deployment that authenticates its proxy, and [`parse_proxy`]'s refusals are not - /// swallowed: `resolve_config` hands them to `install_proxy_tunnel_if_configured`, whose `Err` - /// the llm engine's runtime build turns into a boot panic. Interpolating the raw value put the - /// operator's proxy password on stderr, in the crash report and in whatever CI log collected - /// them. The host, port and scheme are what the operator needs to fix the variable, so they - /// survive — only the credential goes. - /// - /// TEXTUAL, not a re-parse, deliberately: the first caller is the arm where `url::Url::parse` - /// has ALREADY FAILED, so a masker that had to parse could only fail the same way and return - /// the raw value. Userinfo is everything before the LAST `@` of the authority — last, because - /// a password may itself contain an unencoded `@` (`svcacct:hunter2-Pr0xyP@ss@proxy.corp`), - /// and the authority ends at the first `/`, `?` or `#` because userinfo may not contain one. - /// An `@` that appears only AFTER the authority cannot be userinfo, but neither can this - /// function prove the value is well-formed enough for that reasoning to hold, so that residue - /// case redacts wholesale rather than print a maybe-credential. - fn redact_userinfo(v: &str) -> String { - let (scheme, rest) = match v.split_once("://") { - Some((s, r)) => (Some(s), r), - None => (None, v), - }; - let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len()); - let (authority, tail) = rest.split_at(authority_end); - let shown = match authority.rsplit_once('@') { - Some((_userinfo, host_port)) => format!("***@{host_port}{tail}"), - None if rest.contains('@') => "***".to_string(), - None => rest.to_string(), - }; - match scheme { - Some(s) => format!("{s}://{shown}"), - None => shown, - } - } - - /// Parse a proxy env value (`http://[user:pass@]host[:port]`; a bare `host:port` is accepted - /// too). `https://` proxies are refused loudly: TLS-to-proxy is a different transport this - /// tunnel does not speak, and silently downgrading it to TCP would be a lie. - /// - /// Every refusal renders the value through [`redact_userinfo`] — see there for why the raw - /// value must never reach the message, and why the host still does. - pub(super) fn parse_proxy(v: &str) -> Result { - let shown = redact_userinfo(v); - let url = if v.contains("://") { - v.to_string() - } else { - format!("http://{v}") - }; - let parsed = url::Url::parse(&url) - .map_err(|e| format!("proxy env value {shown:?} is not a valid URL: {e}"))?; - if parsed.scheme() != "http" { - return Err(format!( - "proxy env value {shown:?} uses scheme {:?}: only plain http:// CONNECT proxies \ - are supported (an https:// proxy would need TLS-to-proxy, which this tunnel does \ - not speak)", - parsed.scheme() - )); - } - let host = parsed - .host_str() - .ok_or_else(|| format!("proxy env value {shown:?} has no host"))? - .to_string(); - let port = parsed.port().unwrap_or(80); - let auth = (!parsed.username().is_empty()).then(|| { - use base64::Engine; - let creds = format!( - "{}:{}", - parsed.username(), - parsed.password().unwrap_or_default() - ); - format!( - "Basic {}", - base64::engine::general_purpose::STANDARD.encode(creds) - ) - }); - Ok(ProxySpec { host, port, auth }) - } - - /// The raw proxy env, read once — a plain value bag so precedence resolution - /// ([`resolve_config`]) is a pure function tests exercise WITHOUT mutating process env - /// (parallel tests race on env mutation). - #[derive(Default)] - pub(super) struct ProxyEnvValues { - pub(super) https: Option, - pub(super) http: Option, - pub(super) all: Option, - pub(super) no: Option, - } - - impl ProxyEnvValues { - /// Uppercase read before lowercase, empty values treated as unset — hyper-util - /// `matcher::Builder::from_env`'s `get_first_env(&["HTTPS_PROXY", "https_proxy"])` - /// ordering (empty parses to nothing there; skipping it here is the same outcome). - pub(super) fn from_process_env() -> Self { - fn first(keys: [&str; 2]) -> Option { - keys.iter() - .find_map(|k| std::env::var(k).ok().filter(|v| !v.is_empty())) - } - ProxyEnvValues { - https: first(["HTTPS_PROXY", "https_proxy"]), - // Read unconditionally — hyper-util's matcher additionally skips uppercase - // `HTTP_PROXY` under CGI (`REQUEST_METHOD` set; the httpoxy guard). busbar is a - // server binary, never a CGI child, so the guard's precondition cannot hold and - // the unconditional read is behavior-identical where this process runs. - http: first(["HTTP_PROXY", "http_proxy"]), - all: first(["ALL_PROXY", "all_proxy"]), - no: first(["NO_PROXY", "no_proxy"]), - } - } - } - - /// Resolve the value bag into the per-scheme slots, reqwest's precedence: scheme-specific - /// var first, `ALL_PROXY` as the fallback for BOTH slots (hyper-util `Builder::build`: - /// `http: http.or(all)`, `https: https.or(all)`). No proxy var at all → `None`, the direct - /// arm. Any PRESENT value that does not parse is a hard error — reqwest silently ignored - /// garbage and fell through, but silently egressing direct past a configured proxy is the - /// dangerous direction, so busbar refuses to boot instead (documented deviation). - pub(super) fn resolve_config(env: &ProxyEnvValues) -> Result>, String> { - let all = env - .all - .as_deref() - .map(parse_proxy) - .transpose()? - .map(Arc::new); - let scheme_slot = |v: &Option| -> Result>, String> { - Ok(match v.as_deref() { - Some(v) => Some(Arc::new(parse_proxy(v)?)), - None => all.clone(), - }) - }; - let https = scheme_slot(&env.https)?; - let http = scheme_slot(&env.http)?; - if https.is_none() && http.is_none() { - return Ok(None); - } - Ok(Some(Arc::new(ProxyConfig { - https, - http, - no_proxy: NoProxy::parse(env.no.as_deref().unwrap_or("")), - }))) - } - - /// The boot-installed proxy decision, `None` in every deployment without a proxy env. - /// `OnceLock` so config applies rebuild clients against the SAME boot decision — proxy env - /// is process environment, immutable for the process lifetime, exactly reqwest's read-once - /// behavior. - static INSTALLED: OnceLock>> = OnceLock::new(); - - /// Resolve the proxy env at boot: absent → direct (the common arm), present-and-valid → - /// install the scheme-scoped config for every client build, present-and-garbage → refuse to - /// start (fail-loud beats silently egressing direct past a configured proxy). - pub fn install_proxy_tunnel_if_configured() -> Result<(), String> { - let config = resolve_config(&ProxyEnvValues::from_process_env())?; - let _ = INSTALLED.set(config); - Ok(()) - } - - /// What the client builder wires in: the boot decision, or `None` for a build that runs - /// before/without `install_proxy_tunnel_if_configured` (tests, tools) — direct, like reqwest - /// built without proxy env. - pub fn installed_proxy() -> Option> { + /// What the client builder wires in: the boot choice, or `None` for a build that runs + /// before/without [`install_proxy`] (tests, tools) — direct, like reqwest built without proxy + /// env. + pub fn installed_proxy() -> Option> { INSTALLED.get().cloned().flatten() } @@ -942,7 +649,7 @@ mod tunnel { #[derive(Clone)] pub struct TunnelConnector { inner: hyper_util::client::legacy::connect::HttpConnector, - config: Option>, + config: Option>, /// Per-shard connect pacing — see [`ConnectGate`]. Shared by clones of this connector /// (hyper clones the service per connect), NOT across shards: each worker paces its own /// establishment so no cross-worker lock ever appears on the connect path. @@ -1030,7 +737,7 @@ mod tunnel { impl TunnelConnector { pub fn new( inner: hyper_util::client::legacy::connect::HttpConnector, - config: Option>, + config: Option>, dial_bound: usize, posture: DialPosture, ) -> Self { @@ -1180,14 +887,16 @@ mod tunnel { let target_port = dst .port_u16() .unwrap_or(if dst_is_https { 443 } else { 80 }); - let proxy_uri: http::Uri = match format!("http://{}:{}", proxy.host, proxy.port).parse() - { - Ok(u) => u, - Err(e) => return Box::pin(async move { Err(Box::new(e) as BoxError) }), - }; + let proxy_uri: http::Uri = + match format!("http://{}:{}", proxy.host(), proxy.port()).parse() { + Ok(u) => u, + Err(e) => return Box::pin(async move { Err(Box::new(e) as BoxError) }), + }; // Tunneled arm: the storm lands on the PROXY socket, so gate on the proxy // authority; the permit spans dial + CONNECT handshake (both are establishment). - let gate = self.gate.slot(&format!("{}:{}", proxy.host, proxy.port)); + let gate = self + .gate + .slot(&format!("{}:{}", proxy.host(), proxy.port())); let dial = tower::Service::call(&mut self.inner, proxy_uri); let posture = self.posture.clone(); Box::pin(async move { @@ -1224,11 +933,11 @@ mod tunnel { stream: &mut tokio::net::TcpStream, host: &str, port: u16, - proxy: &ProxySpec, + proxy: &ProxyHop, ) -> Result<(), BoxError> { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let mut req = format!("CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n"); - if let Some(auth) = &proxy.auth { + if let Some(auth) = proxy.proxy_authorization() { req.push_str("Proxy-Authorization: "); req.push_str(auth); req.push_str("\r\n"); @@ -1272,7 +981,7 @@ mod tunnel { Ok(()) } - /// A full config whose BOTH slots point at one scripted proxy — what the end-to-end test + /// A whole choice whose BOTH slots point at one scripted proxy — what the end-to-end test /// wires directly so parallel tests never race an env var or the OnceLock. #[cfg(test)] pub(super) fn test_config( @@ -1280,22 +989,11 @@ mod tunnel { port: u16, auth: Option, no_proxy: &str, - ) -> Arc { - let spec = Arc::new(ProxySpec { - host: host.to_string(), - port, - auth, - }); - Arc::new(ProxyConfig { - https: Some(spec.clone()), - http: Some(spec), - no_proxy: NoProxy::parse(no_proxy), - }) + ) -> Arc { + let hop = Arc::new(ProxyHop::new(host.to_string(), port, auth)); + Arc::new(ProxyChoice::new(Some(hop.clone()), Some(hop), no_proxy)) } - #[cfg(test)] - pub(super) use parse_proxy as parse_proxy_for_tests; - #[cfg(test)] pub(super) use connects_per_shard as connects_per_shard_for_tests; @@ -1318,26 +1016,4 @@ mod tunnel { self.slot(authority) } } - #[cfg(test)] - pub(super) use resolve_config as resolve_config_for_tests; - #[cfg(test)] - pub(super) use ProxyEnvValues as ProxyEnvValuesForTests; - - /// The per-CONNECT decision as tests see it: the selected proxy's `host:port`, or `None` for - /// the direct arm. - #[cfg(test)] - pub(super) fn select_for_tests( - config: &Arc, - dst_is_https: bool, - target_host: &str, - ) -> Option { - config - .select(dst_is_https, target_host) - .map(|p| format!("{}:{}", p.host, p.port)) - } - - #[cfg(test)] - pub(super) fn no_proxy_matches_for_tests(entries: &str, target_host: &str) -> bool { - NoProxy::parse(entries).matches(target_host) - } } diff --git a/crates/busbar-kernel/src/egress/engine/tests/engine_tests.rs b/crates/busbar-kernel/src/egress/engine/tests/engine_tests.rs index 9b3bd47f52..976fd7a295 100644 --- a/crates/busbar-kernel/src/egress/engine/tests/engine_tests.rs +++ b/crates/busbar-kernel/src/egress/engine/tests/engine_tests.rs @@ -2,12 +2,10 @@ // Copyright (C) 2026 Busbar Inc and contributors //! The owned egress client's building blocks: request assembly is byte-exact from precomputed -//! parts, the h1-only/h2c precedence mirrors the old reqwest builder's apply-order, and the -//! proxy-env machinery matches reqwest 0.12's scoping — per-scheme slot resolution -//! (HTTPS_PROXY/HTTP_PROXY with ALL_PROXY fallback), scheme-scoped selection, and the full -//! NO_PROXY rule set (host, domain suffix, IP literal, CIDR). Every proxy test drives the parse/ -//! resolve/select functions directly — NEVER by mutating process env, which races under the -//! parallel test runner. +//! parts, the h1-only/h2c precedence mirrors the old reqwest builder's apply-order, and the CONNECT +//! tunnel runs the exchange a proxy choice names. The proxy environment's parse and the choice's +//! rules are the connector's (`busbar_core_connector::proxy`, status row 130 E5a) and its suite +//! holds them; this suite wires a choice directly, never by mutating process env. use super::*; @@ -36,237 +34,6 @@ fn builder_smoke_all_shapes() { } } -#[test] -fn proxy_url_parse_extracts_auth_and_refuses_https_proxies() { - // No env mutation (racy under parallel tests): the parser is exercised directly. - let err = tunnel::parse_proxy_for_tests("https://proxy.corp:3128").unwrap_err(); - assert!( - err.contains("https"), - "an https:// proxy must be refused loudly, naming why: {err}" - ); - let garbage = tunnel::parse_proxy_for_tests("http://").unwrap_err(); - assert!(!garbage.is_empty(), "garbage must not parse"); - // Bare host:port and userinfo both parse; the auth header is prebuilt Basic. - assert!(tunnel::parse_proxy_for_tests("proxy.corp:3128").is_ok()); - assert!(tunnel::parse_proxy_for_tests("http://u:p@proxy.corp").is_ok()); -} - -/// Precedence resolution, table-driven over the pure value bag (reqwest via hyper-util -/// `matcher::Builder::build`: scheme-specific var first, ALL_PROXY fallback for BOTH slots). -/// Each row: (HTTPS_PROXY, HTTP_PROXY, ALL_PROXY) → (https slot host:port, http slot host:port). -#[test] -fn resolve_config_matches_reqwest_precedence() { - /// (HTTPS_PROXY, HTTP_PROXY, ALL_PROXY, expected https slot, expected http slot). - type Row = ( - Option<&'static str>, - Option<&'static str>, - Option<&'static str>, - Option<&'static str>, - Option<&'static str>, - ); - let rows: &[Row] = &[ - // No proxy env at all → no config (the direct arm every known deployment takes). - (None, None, None, None, None), - // HTTPS_PROXY alone scopes to https targets ONLY — http targets go direct, never - // through the https proxy (gap (b): we used to send them through it). - (Some("hs.corp:1"), None, None, Some("hs.corp:1"), None), - // HTTP_PROXY alone scopes to http targets ONLY (gap (a): we used to ignore it and go - // direct — the dangerous direction). - (None, Some("h.corp:2"), None, None, Some("h.corp:2")), - // ALL_PROXY alone fills BOTH slots. - ( - None, - None, - Some("all.corp:3"), - Some("all.corp:3"), - Some("all.corp:3"), - ), - // Scheme-specific var beats ALL_PROXY for its scheme; ALL_PROXY still fills the other. - ( - Some("hs.corp:1"), - None, - Some("all.corp:3"), - Some("hs.corp:1"), - Some("all.corp:3"), - ), - ( - None, - Some("h.corp:2"), - Some("all.corp:3"), - Some("all.corp:3"), - Some("h.corp:2"), - ), - // All three set: each scheme var wins its own slot. - ( - Some("hs.corp:1"), - Some("h.corp:2"), - Some("all.corp:3"), - Some("hs.corp:1"), - Some("h.corp:2"), - ), - ]; - for (https, http, all, want_https, want_http) in rows { - let env = tunnel::ProxyEnvValuesForTests { - https: https.map(String::from), - http: http.map(String::from), - all: all.map(String::from), - no: None, - }; - let config = tunnel::resolve_config_for_tests(&env).unwrap(); - match (want_https, want_http) { - (None, None) => assert!( - config.is_none(), - "no proxy vars must resolve to no config: {https:?}/{http:?}/{all:?}" - ), - _ => { - let config = config.expect("some slot is set, a config must exist"); - let got_https = tunnel::select_for_tests(&config, true, "api.example.com"); - let got_http = tunnel::select_for_tests(&config, false, "api.example.com"); - assert_eq!( - got_https.as_deref(), - *want_https, - "https slot for {https:?}/{http:?}/{all:?}" - ); - assert_eq!( - got_http.as_deref(), - *want_http, - "http slot for {https:?}/{http:?}/{all:?}" - ); - } - } - } -} - -/// A garbage value in ANY set proxy var fails resolution loudly — busbar refuses to boot rather -/// than silently egressing direct past a configured proxy (documented deviation from reqwest's -/// silent fall-through). -#[test] -fn resolve_config_fail_louds_on_garbage_in_any_slot() { - for (https, http, all) in [ - (Some("https://tls-proxy.corp"), None, None), - (None, Some("http://"), None), - (None, None, Some("https://tls-proxy.corp")), - ] { - let env = tunnel::ProxyEnvValuesForTests { - https: https.map(String::from), - http: http.map(String::from), - all: all.map(String::from), - no: None, - }; - assert!( - tunnel::resolve_config_for_tests(&env).is_err(), - "garbage in {https:?}/{http:?}/{all:?} must refuse to resolve" - ); - } -} - -#[test] -fn no_proxy_suffix_matching_is_conventional() { - let config = tunnel::test_config("proxy.corp", 3128, None, "example.com, internal"); - let via = |host: &str| tunnel::select_for_tests(&config, true, host).is_some(); - assert!(!via("example.com"), "exact NO_PROXY entry bypasses"); - assert!(!via("api.example.com"), "domain suffix bypasses"); - assert!( - via("notexample.com"), - "a non-boundary suffix must NOT bypass (substring is not a domain match)" - ); - assert!(!via("INTERNAL"), "matching is case-insensitive"); - assert!(via("api.upstream.test"), "unlisted hosts tunnel"); - let all = tunnel::test_config("proxy.corp", 3128, None, "*"); - assert!( - tunnel::select_for_tests(&all, true, "api.upstream.test").is_none(), - "`*` disables tunneling" - ); - // Leading dots are equivalent to none (curl/reqwest): `.example.com` ≡ `example.com`. - assert!(tunnel::no_proxy_matches_for_tests( - ".example.com", - "example.com" - )); - assert!(tunnel::no_proxy_matches_for_tests( - ".example.com", - "api.example.com" - )); -} - -/// NO_PROXY IP-literal and CIDR matching (gap (c)) — the matrix over families, boundary prefix -/// lengths (/0, /8, /32, /128), containment edges, and the documented non-special-casing of -/// v6-mapped addresses. -#[test] -fn no_proxy_ip_and_cidr_matrix() { - let m = tunnel::no_proxy_matches_for_tests; - // IP literals: exact match only, and never as a domain suffix. - assert!(m("10.1.2.3", "10.1.2.3")); - assert!(!m("10.1.2.3", "10.1.2.30")); - assert!(!m("10.1.2.3", "110.1.2.3")); - assert!(m("::1", "::1")); - assert!(m("::1", "[::1]"), "bracketed v6 literal hosts match too"); - // v4 CIDR containment, boundary prefixes. - assert!(m("10.0.0.0/8", "10.255.255.255")); - assert!(m("10.0.0.0/8", "10.0.0.1")); - assert!(!m("10.0.0.0/8", "11.0.0.0")); - assert!(!m("10.0.0.0/8", "9.255.255.255")); - assert!(m("0.0.0.0/0", "203.0.113.7"), "/0 matches every v4 address"); - assert!(m("192.168.1.42/32", "192.168.1.42"), "/32 is exact"); - assert!(!m("192.168.1.42/32", "192.168.1.43")); - // Non-octet-aligned prefix: the partial-byte mask must apply. - assert!(m("192.168.1.0/25", "192.168.1.127")); - assert!(!m("192.168.1.0/25", "192.168.1.128")); - // v6 CIDR containment, boundary prefixes. - assert!(m("fd00::/8", "fd12:3456::1")); - assert!(!m("fd00::/8", "fe80::1")); - assert!(m("::/0", "2001:db8::1"), "/0 matches every v6 address"); - assert!(m("2001:db8::1/128", "2001:db8::1"), "/128 is exact"); - assert!(!m("2001:db8::1/128", "2001:db8::2")); - // Families never cross — INCLUDING v6-mapped forms: `::ffff:10.0.0.1` is a v6 address, so a - // v4 block does not match it (same as ipnet/reqwest, which do not special-case mapped - // addresses). - assert!(!m("10.0.0.0/8", "::ffff:10.0.0.1")); - assert!(!m("::/0", "10.0.0.1"), "a v6 catch-all does not catch v4"); - assert!( - !m("0.0.0.0/0", "2001:db8::1"), - "a v4 catch-all does not catch v6" - ); - // An IP-literal host never matches domain entries (reqwest checks IP hosts only against - // IP/CIDR entries). - assert!(!m("0.0.1", "10.0.0.1")); - // A malformed CIDR degrades to a never-matching domain entry, exactly as ipnet parse - // failure does under reqwest — it must not poison the other entries. - assert!(!m("10.0.0.0/40", "10.0.0.1")); - assert!(m("10.0.0.0/40, example.com", "example.com")); -} - -/// Scheme-scoped selection with NO_PROXY layered on top: exclusion applies to both slots. -#[test] -fn selection_is_scheme_scoped_and_no_proxy_excludes_both() { - let env = tunnel::ProxyEnvValuesForTests { - https: Some("hs.corp:1".to_string()), - http: Some("h.corp:2".to_string()), - all: None, - no: Some("internal.corp, 10.0.0.0/8".to_string()), - }; - let config = tunnel::resolve_config_for_tests(&env).unwrap().unwrap(); - assert_eq!( - tunnel::select_for_tests(&config, true, "api.upstream.test").as_deref(), - Some("hs.corp:1") - ); - assert_eq!( - tunnel::select_for_tests(&config, false, "api.upstream.test").as_deref(), - Some("h.corp:2") - ); - for is_https in [true, false] { - assert!( - tunnel::select_for_tests(&config, is_https, "db.internal.corp").is_none(), - "NO_PROXY domain excludes the {} slot", - if is_https { "https" } else { "http" } - ); - assert!( - tunnel::select_for_tests(&config, is_https, "10.20.30.40").is_none(), - "NO_PROXY CIDR excludes the {} slot", - if is_https { "https" } else { "http" } - ); - } -} - /// The full CONNECT exchange against a scripted proxy: the proxy must see the exact /// `CONNECT host:port` + `Proxy-Authorization` bytes, and after its 200 the SAME socket must /// carry the real HTTP exchange end-to-end through the pooled client. @@ -352,43 +119,6 @@ async fn connect_tunnel_end_to_end_through_scripted_proxy() { ); } -/// The boot install agrees with the pure resolve over the SAME environment, in whichever arm this -/// run happens to be in. Env-var permutations are covered by the direct resolve/select tests above — -/// mutating process env in a parallel test binary is a race, so none of these tests do it. -/// -/// Both arms assert. The clean arm is the one every boot takes: install is Ok and nothing is -/// installed, because a deployment with no proxy configured must not acquire one. A run under a -/// developer's or a CI runner's proxy variables takes the other arm, and there install must answer -/// exactly what resolving those same values answers — which is what makes a run in a dirty -/// environment evidence rather than a test that quietly did nothing. -#[test] -fn install_agrees_with_the_resolve_over_the_same_environment() { - let env = tunnel::ProxyEnvValuesForTests::from_process_env(); - let clean = env.https.is_none() && env.http.is_none() && env.all.is_none(); - let resolved = tunnel::resolve_config_for_tests(&env); - let installed = install_proxy_tunnel_if_configured(); - assert_eq!( - installed.is_ok(), - resolved.is_ok(), - "the boot install and the pure resolve disagreed over the same environment" - ); - if clean { - assert!( - installed.is_ok(), - "a deployment with no proxy variables must boot" - ); - assert!( - tunnel::installed_proxy().is_none(), - "a deployment with no proxy variables must not acquire one" - ); - } else { - assert!( - resolved.is_err() || tunnel::installed_proxy().is_some() == resolved.unwrap().is_some(), - "the environment names a proxy and the installed decision does not match the resolve" - ); - } -} - /// The connect gate bounds ESTABLISHMENT concurrency per authority and shares nothing across /// authorities — the overload-cliff fix's core invariant. Driven directly (the semaphore IS the /// mechanism); the storm-scale proof lives on the bench rig's overload rungs. @@ -585,73 +315,3 @@ fn request_adds_no_accept_a_plane_hop_did_not_name() { assert!(req.headers().get(http::header::ACCEPT).is_none()); assert!(req.headers().is_empty()); } - -/// AN OPERATOR'S PROXY PASSWORD MUST NOT REACH THE REFUSAL TEXT. -/// -/// `HTTPS_PROXY`/`HTTP_PROXY`/`ALL_PROXY` may carry userinfo (RFC 3986 §3.2.1) and a corporate -/// deployment routinely puts a service account there — -/// `https://svcacct:hunter2-Pr0xyP@ss@proxy.corp:3128`. Every refusal arm of `parse_proxy` -/// interpolated the RAW env value with `{v:?}`, userinfo included, and that string is not -/// swallowed: `resolve_config` hands it to `install_proxy_tunnel_if_configured`, whose `Err` the -/// llm engine's runtime build turns into a `panic!` with no `catch_unwind` under it. So the -/// password reached stderr, the crash report, and whatever CI log collected them — from a value the -/// operator set once in a systemd unit and never looked at again. -/// -/// The password here is a planted marker and its ABSENCE is what is asserted; the test never prints -/// a real credential to fail informatively. `hunter2-Pr0xyP@ss` is unique in this file, so a -/// regression that reinstates `{v:?}` fails on the very first assertion. -/// -/// `parse_proxy`'s third arm (`has no host`) is not driven here because it is unreachable: the -/// scheme arm above it has already established `http`, and `url::Url` refuses an empty host for a -/// special scheme, so `host_str()` cannot be `None` by the time control gets there. It is redacted -/// all the same — a defensive arm that leaks when it is finally reached is still a leak. -#[test] -fn proxy_env_refusals_never_echo_the_proxy_password() { - const PASSWORD: &str = "hunter2-Pr0xyP@ss"; - // Two values, both carrying the same credential, one per REACHABLE refusal arm: - // * an `https://` proxy — parses fine, refused by the scheme arm (the brief's repro); - // * a bad port — refused by the `url::Url::parse` arm, where a URL-reparsing masker is a - // provable no-op (the parse it would have to do is the parse that just failed), which is - // why the redaction this path uses is textual. - let values = [ - "https://svcacct:hunter2-Pr0xyP@ss@proxy.corp:3128", - "http://svcacct:hunter2-Pr0xyP@ss@proxy.corp:notaport", - ]; - for v in values { - let err = tunnel::parse_proxy_for_tests(v) - .expect_err("both values must be refused, or this test proves nothing"); - assert!( - !err.contains(PASSWORD), - "the proxy password must never be interpolated into a refusal, got: {err}" - ); - assert!( - !err.contains("hunter2"), - "not even a fragment of it, got: {err}" - ); - // What the operator IS owed still arrives: the host they typed, so they can tell WHICH - // variable is wrong, and why it was refused. - assert!( - err.contains("proxy.corp"), - "the host is the diagnosis and must survive redaction, got: {err}" - ); - } - - // AND THROUGH THE BOOT PATH THAT ACTUALLY RUNS, so the assertion is anchored to the reachable - // call and not only to the parser underneath it: this is the exact function - // `install_proxy_tunnel_if_configured` calls, with the env read injected instead of mutated. - for slot in 0..3 { - let bad = Some(values[0].to_string()); - let env = tunnel::ProxyEnvValuesForTests { - https: (slot == 0).then(|| bad.clone()).flatten(), - http: (slot == 1).then(|| bad.clone()).flatten(), - all: (slot == 2).then(|| bad.clone()).flatten(), - no: None, - }; - let err = tunnel::resolve_config_for_tests(&env) - .expect_err("a garbage proxy value must refuse to resolve"); - assert!( - !err.contains(PASSWORD) && !err.contains("hunter2"), - "the boot refusal must not carry the proxy password either (slot {slot}), got: {err}" - ); - } -} diff --git a/crates/busbar-kernel/src/proxy/mod.rs b/crates/busbar-kernel/src/proxy/mod.rs index 977c3c0861..590afc3806 100644 --- a/crates/busbar-kernel/src/proxy/mod.rs +++ b/crates/busbar-kernel/src/proxy/mod.rs @@ -51,7 +51,7 @@ pub use proxy_vocab::{ // (`EgressConnector`) have no remaining in-core reader after the move, and an externally-visible // re-export cannot rot into an unused-import warning while the path contract stands. pub use busbar_kernel::egress::engine::{ - egress_request, install_proxy_tunnel_if_configured, EngineClient as EgressClient, + egress_request, install_proxy, EngineClient as EgressClient, EngineConnector as EgressConnector, EngineError as EgressError, EngineSpec as EgressClientSpec, }; diff --git a/crates/busbar-llm/src/engine/build_runtime.rs b/crates/busbar-llm/src/engine/build_runtime.rs index 9c3012c08c..a08f7b73e8 100644 --- a/crates/busbar-llm/src/engine/build_runtime.rs +++ b/crates/busbar-llm/src/engine/build_runtime.rs @@ -318,8 +318,8 @@ pub(crate) fn build_runtime( let client = if let (true, Some(pr)) = (reuse_prior_client, prior_rt) { pr.client.clone() } else { - crate::engine::install_proxy_tunnel_if_configured() - .unwrap_or_else(|e| panic!("upstream proxy tunnel: {e}")); + // The proxy choice is the connector's parse of the process environment, installed into the + // engine by the composition root at boot (status row 130 E5a); nothing is read here. let shard_count = busbar_kernel::topology::UpstreamClients::shard_count(); let idle_per_host_per_shard = input .client_settings diff --git a/crates/busbar-llm/src/engine/mod.rs b/crates/busbar-llm/src/engine/mod.rs index a7b5f10407..01195ccbf0 100644 --- a/crates/busbar-llm/src/engine/mod.rs +++ b/crates/busbar-llm/src/engine/mod.rs @@ -118,8 +118,7 @@ pub(crate) use busbar_kernel::proxy::{ // historical `EgressClientSpec`/`EgressError` short names are preserved via the same aliases core used. #[cfg_attr(not(test), allow(unused_imports))] pub(crate) use busbar_kernel::egress::engine::{ - egress_request, install_proxy_tunnel_if_configured, EngineError as EgressError, - EngineSpec as EgressClientSpec, + egress_request, EngineError as EgressError, EngineSpec as EgressClientSpec, }; // The NEUTRAL content-type / disposition / error-KIND vocabulary named at its TRUE substrate home diff --git a/crates/busbar/src/root/connector.rs b/crates/busbar/src/root/connector.rs index 8afd76e8c4..c7655a6268 100644 --- a/crates/busbar/src/root/connector.rs +++ b/crates/busbar/src/root/connector.rs @@ -231,6 +231,9 @@ pub fn boot( eprintln!("busbar: a second connector was built; the process has one"); std::process::exit(2); }); + // THE ONE PROXY CHOICE (status row 130 E5a): the connector read the process environment; the + // kernel's remaining egress engine takes the same value and reads none of its own. + busbar_kernel::egress::engine::install_proxy(one.proxy()); one } diff --git a/qa/kind-isolation.toml b/qa/kind-isolation.toml index a5898f5ce4..6a3aed8ab6 100644 --- a/qa/kind-isolation.toml +++ b/qa/kind-isolation.toml @@ -1734,8 +1734,8 @@ cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain [[law0]] crate = "busbar-core-connector" axis = "transport" -count = "68" -cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain is THE DESIGN §1 / DoD item 2: core names no instance" +count = "69" +cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain is THE DESIGN §1 / DoD item 2: core names no instance; REBALANCED 68 -> 69 by row 130 E5a (#804, ruled MOVE, net flat): the proxy tunnel moved out of busbar-kernel's egress engine into this crate's proxy.rs, and its `\"http\"` proxy-scheme check moved with it verbatim, 1 hit (busbar-kernel × transport gives the same 1 up, 284 -> 278)" [[law0]] crate = "busbar-core-oauth2" @@ -1788,8 +1788,8 @@ cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain [[law0]] crate = "busbar-kernel" axis = "plane" -count = "1122" -cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain is THE DESIGN §1 / DoD item 2: core names no instance; LOWERED 1193 -> 1175, predev 9c0a394771's measure: the drain landed without giving its row back (Law 9: the ceiling ratchets down); LOWERED 1175 -> 1173, the merge-base d7929b661d's measure, which GetBusbar/busbar#570 holds (its scrape double spells no plane alias) (Law 9: the ceiling ratchets down to today's number); LOWERED 1173 -> 1172, the gate's measure on this branch merged with predev 1617e9f019: Q128 A4's plane_driver/gated.rs drain; REBALANCED 1172 -> 1168 (Q128, PR #686 architect ruling, BUSBAR-1.6.0.md l.475-483): the cross-plane cells moved verbatim into the composition root's crates/busbar/tests/, which takes the same 4 hits (busbar × plane 822 -> 826); LOWERED 1168 -> 1141 by row 157: the hits drained with the deleted HOT trust and guard slots (plane_host/trust.rs, plane_host/guard.rs); LOWERED 1141 -> 1132 by row 156: the hits drained with the deleted kernel call log (calllog.rs, plane/calllog.rs); LOWERED 1132 -> 1128 by this PR: the deletion of the kernel's dead ask-state seal (audit K2-H3) removes four hits, its two `crate::mcp::roots` doc links and its two `\\x6dcp` seal domains; LOWERED 1128 -> 1122 by status row 130 E2: net_guard.rs and its two suites deleted (six plane hits, the gate's measure on the E2 candidate)" +count = "1118" +cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain is THE DESIGN §1 / DoD item 2: core names no instance; LOWERED 1193 -> 1175, predev 9c0a394771's measure: the drain landed without giving its row back (Law 9: the ceiling ratchets down); LOWERED 1175 -> 1173, the merge-base d7929b661d's measure, which GetBusbar/busbar#570 holds (its scrape double spells no plane alias) (Law 9: the ceiling ratchets down to today's number); LOWERED 1173 -> 1172, the gate's measure on this branch merged with predev 1617e9f019: Q128 A4's plane_driver/gated.rs drain; REBALANCED 1172 -> 1168 (Q128, PR #686 architect ruling, BUSBAR-1.6.0.md l.475-483): the cross-plane cells moved verbatim into the composition root's crates/busbar/tests/, which takes the same 4 hits (busbar × plane 822 -> 826); LOWERED 1168 -> 1141 by row 157: the hits drained with the deleted HOT trust and guard slots (plane_host/trust.rs, plane_host/guard.rs); LOWERED 1141 -> 1132 by row 156: the hits drained with the deleted kernel call log (calllog.rs, plane/calllog.rs); LOWERED 1132 -> 1128 by this PR: the deletion of the kernel's dead ask-state seal (audit K2-H3) removes four hits, its two `crate::mcp::roots` doc links and its two `\\x6dcp` seal domains; LOWERED 1128 -> 1122 by status row 130 E2: net_guard.rs and its two suites deleted (six plane hits, the gate's measure on the E2 candidate); LOWERED 1122 -> 1118 by row 130 E5a (#804): the hits drained with the proxy tunnel the egress engine gave up (two `llm engine` prose mentions, the `decision:` doc key and the `installed decision` assert message in engine/mod.rs and engine_tests.rs); none moved — busbar-contract's proxy_route.rs names the per-dial choice without the word" [[law0]] crate = "busbar-kernel" @@ -1806,8 +1806,8 @@ cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain [[law0]] crate = "busbar-kernel" axis = "transport" -count = "266" -cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain is THE DESIGN §1 / DoD item 2: core names no instance; LOWERED 288 -> 286, predev 9c0a394771's measure: the drain landed without giving its row back (Law 9: the ceiling ratchets down); LOWERED 286 -> 285 by GetBusbar/busbar#570: the kernel's auth proof no longer cites the root's stdio-served proof; LOWERED 285 -> 284 by row 157: the hit drained with the same deletion; REBALANCED 284 -> 267 (-17), the verbatim listener serve loop moved out of busbar-kernel into the root (#710), paired with busbar x transport 208 -> 225 (+17) so the law0 total does not rise (ARCHITECT RULING 2026-10-10, https://github.com/GetBusbar/busbar/pull/710#issuecomment-6096420101; BUSBAR-1.6.0.md l.475-483 INBOUND-LISTEN: the root's stream hand-up); LOWERED 267 -> 266 by GetBusbar/busbar#532 (Q1): the kernel's cold auth path is deleted (auth/token.rs's `\"http\"` scheme check), a different hit from row 157's" +count = "260" +cite = "measured on predev f830faa691 (ARCHITECT 2026-10-07, Law 9); the drain is THE DESIGN §1 / DoD item 2: core names no instance; LOWERED 288 -> 286, predev 9c0a394771's measure: the drain landed without giving its row back (Law 9: the ceiling ratchets down); LOWERED 286 -> 285 by GetBusbar/busbar#570: the kernel's auth proof no longer cites the root's stdio-served proof; LOWERED 285 -> 284 by row 157: the hit drained with the same deletion; REBALANCED 284 -> 267 (-17), the verbatim listener serve loop moved out of busbar-kernel into the root (#710), paired with busbar x transport 208 -> 225 (+17) so the law0 total does not rise (ARCHITECT RULING 2026-10-10, https://github.com/GetBusbar/busbar/pull/710#issuecomment-6096420101; BUSBAR-1.6.0.md l.475-483 INBOUND-LISTEN: the root's stream hand-up); LOWERED 267 -> 266 by GetBusbar/busbar#532 (Q1): the kernel's cold auth path is deleted (auth/token.rs's `\"http\"` scheme check), a different hit from row 157's; LOWERED 266 -> 260 by row 130 E5a (#804): the proxy tunnel left the egress engine (engine/mod.rs, engine_tests.rs) for busbar-core-connector's proxy.rs — 5 hits drained with it (three `TCP` prose mentions, two `\"http\"` test literals) and 1 MOVED verbatim (the `\"http\"` proxy-scheme check; busbar-core-connector × transport 68 -> 69, net flat)" [[law0]] crate = "busbar-kernel-breaker"