diff --git a/crates/yaak-proxy/src/lib.rs b/crates/yaak-proxy/src/lib.rs index 00dc784fb..b0406f7de 100644 --- a/crates/yaak-proxy/src/lib.rs +++ b/crates/yaak-proxy/src/lib.rs @@ -4,7 +4,9 @@ mod connection; mod request; mod tunnel; +use std::collections::hash_map::RandomState; use std::collections::{HashMap, HashSet}; +use std::hash::{BuildHasher, Hasher}; use std::net::SocketAddr; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::mpsc as std_mpsc; @@ -12,6 +14,7 @@ use std::sync::{Arc, Mutex}; use std::time::SystemTime; use bytes::Bytes; +use http::HeaderValue; use hyper_util::client::legacy::Client; use hyper_util::client::legacy::connect::HttpConnector; use hyper_util::rt::TokioExecutor; @@ -120,7 +123,9 @@ pub enum ProxyEvent { elapsed_ms: u64, error: Option, }, - /// No response: connecting to or requesting from upstream failed. The client got a 502. + /// No response: connecting to or requesting from upstream failed, and the client got a 502, + /// or the engine refused the request (no host to send it to, or it looped back to the engine) + /// with a 4xx or 5xx of its own. Error { id: u64, error: String, @@ -174,6 +179,9 @@ pub(crate) struct Engine { upstream_tls_h1: Arc, passthrough: Mutex>, handshake_failures: Mutex>, + /// `Via` value marking requests whose URL the engine rebuilt from `Host`, unique to this run, + /// so one that comes back to the engine is refused instead of forwarded again. + via: HeaderValue, } impl Engine { @@ -222,6 +230,11 @@ fn upstream_tls_config( Ok(Arc::new(config)) } +fn via_marker() -> HeaderValue { + let run = RandomState::new().build_hasher().finish(); + HeaderValue::try_from(format!("1.1 yaak-proxy-{run:016x}")).unwrap() +} + /// Starts the engine on its own thread and Tokio runtime, returning once the port is bound. pub fn start_proxy(config: ProxyConfig) -> Result { let mut roots = rustls::RootCertStore::empty(); @@ -243,6 +256,7 @@ pub fn start_proxy(config: ProxyConfig) -> Result { upstream_tls_h1: upstream_tls_config(&roots, &[b"http/1.1"])?, passthrough: Mutex::new(HashSet::new()), handshake_failures: Mutex::new(HashMap::new()), + via: via_marker(), }); let port = config.port; diff --git a/crates/yaak-proxy/src/request.rs b/crates/yaak-proxy/src/request.rs index 01990cff0..a67b4d6d1 100644 --- a/crates/yaak-proxy/src/request.rs +++ b/crates/yaak-proxy/src/request.rs @@ -7,7 +7,8 @@ use std::time::{Duration, Instant, SystemTime}; use bytes::Bytes; use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; -use hyper::header::{HOST, UPGRADE}; +use hyper::header::{CONTENT_TYPE, HOST, UPGRADE, VIA}; +use hyper::http::uri::{Authority, PathAndQuery, Scheme}; use hyper::upgrade::OnUpgrade; use hyper::{Method, Request, Response, StatusCode, Uri}; use hyper_util::client::legacy::connect::HttpInfo; @@ -113,6 +114,21 @@ impl Flow { Response::from_parts(parts, body.boxed()) } + /// Answers the client with `status` and `error` without contacting any server. + fn reject(self, status: StatusCode, error: &str) -> Response { + debug!("Flow {} rejected: {error}", self.id); + self.engine.emit(ProxyEvent::Error { + id: self.id, + error: error.to_string(), + elapsed_ms: self.start.elapsed().as_millis() as u64, + }); + Response::builder() + .status(status) + .header(CONTENT_TYPE, "text/plain; charset=utf-8") + .body(text_body(error.to_string())) + .unwrap() + } + fn fail(self, error: String) -> Response { debug!("Flow {} failed: {error}", self.id); let body = format!("Yaak Proxy could not reach the server: {error}"); @@ -130,14 +146,24 @@ fn bad_gateway(body: String) -> Response { } pub(crate) async fn handle_request( - req: Request, + mut req: Request, engine: Arc, ) -> Result, Infallible> { if req.method() == Method::CONNECT { return Ok(handle_connect(req, engine)); } + // Some clients send a bare path (`POST /api`) as they would to the server itself. + let from_host = req.uri().authority().is_none(); + let rejected = if from_host { url_from_host(&mut req, &engine).err() } else { None }; let flow = Flow::begin(engine.clone(), &req, req.uri().to_string()); + if let Some((status, error)) = rejected { + return Ok(flow.reject(status, error)); + } + if from_host { + req.headers_mut().append(VIA, engine.via.clone()); + } + let (parts, body) = req.into_parts(); let outgoing = Request::from_parts(parts, flow.capture_request(body)); Ok(match engine.http_client.request(outgoing).await { @@ -149,6 +175,41 @@ pub(crate) async fn handle_request( }) } +/// Gives a bare-path request the `http://` URL its `Host` names. Fails when there is no usable +/// `Host`, or when the request carries this engine's `Via` mark: then it is one the engine sent +/// that was routed back to it, and forwarding it again would loop forever. +fn url_from_host( + req: &mut Request, + engine: &Engine, +) -> Result<(), (StatusCode, &'static str)> { + let Some(host) = req.headers().get(HOST) else { + return Err((StatusCode::BAD_REQUEST, "no host in request")); + }; + let Some(authority) = host.to_str().ok().and_then(|h| h.parse::().ok()) else { + return Err((StatusCode::BAD_REQUEST, "invalid host in request")); + }; + let mut parts = std::mem::take(req.uri_mut()).into_parts(); + parts.scheme = Some(Scheme::HTTP); + parts.authority = Some(authority); + if parts.path_and_query.is_none() { + parts.path_and_query = Some(PathAndQuery::from_static("/")); + } + *req.uri_mut() = + Uri::from_parts(parts).map_err(|_| (StatusCode::BAD_REQUEST, "invalid host in request"))?; + + let via = engine.via.to_str().unwrap_or_default(); + let looped = req + .headers() + .get_all(VIA) + .iter() + .filter_map(|v| v.to_str().ok()) + .any(|v| v.split(',').any(|hop| hop.trim() == via)); + if looped { + return Err((StatusCode::LOOP_DETECTED, "request looped back to Yaak Proxy")); + } + Ok(()) +} + /// Where an intercepted CONNECT tunnel leads. struct Target { /// Host for the leaf certificate and upstream SNI. @@ -358,10 +419,19 @@ async fn forward_https( Ok(uri) => uri, Err(e) => return Ok(flow.fail(e.to_string())), }; - if !parts.headers.contains_key(HOST) - && let Ok(host) = target.authority.parse() - { - parts.headers.insert(HOST, host); + match &upstream { + // The URL carries `:authority`. Google's servers reset any stream that also has `Host`, + // even a matching one. + Upstream::H2(_) => { + parts.headers.remove(HOST); + } + Upstream::H1(_) => { + if !parts.headers.contains_key(HOST) + && let Ok(host) = target.authority.parse() + { + parts.headers.insert(HOST, host); + } + } } let outgoing = Request::from_parts(parts, flow.capture_request(body)); diff --git a/crates/yaak-proxy/tests/proxy.rs b/crates/yaak-proxy/tests/proxy.rs index 31eed234d..ec6e76b2b 100644 --- a/crates/yaak-proxy/tests/proxy.rs +++ b/crates/yaak-proxy/tests/proxy.rs @@ -9,7 +9,7 @@ use futures_util::{SinkExt, StreamExt}; use http_body_util::combinators::BoxBody; use http_body_util::{BodyExt, Full, StreamBody}; use hyper::body::{Frame, Incoming}; -use hyper::header::{CONNECTION, CONTENT_LENGTH, UPGRADE}; +use hyper::header::{CONNECTION, CONTENT_LENGTH, CONTENT_TYPE, HOST, UPGRADE}; use hyper::service::service_fn; use hyper::{Request, Response, StatusCode}; use hyper_util::rt::{TokioExecutor, TokioIo}; @@ -87,6 +87,11 @@ async fn route(req: Request) -> Result, Infallible> Ok(match req.uri().path() { "/hello" => Response::new(full("hello")), "/echo" => Response::new(full(req.into_body().collect().await.unwrap().to_bytes())), + "/authority" => Response::new(full(format!( + "authority={} host={}", + req.uri().authority().map_or("none".into(), |a| a.to_string()), + req.headers().get(HOST).map_or("none", |h| h.to_str().unwrap()), + ))), "/big" => Response::new(full(vec![b'x'; BIG_BODY])), "/late-end" => late_end(5), // The stream ends with half the promised bytes, so the upstream resets it mid-body. @@ -200,6 +205,16 @@ impl Harness { self.client_trusting(self.ca.pem()) } + /// Sends `req` to the proxy exactly as built, bare path and all, as a client that doesn't know + /// it is talking to a proxy would. + async fn send_as_written(&self, req: Request) -> Response { + let tcp = TcpStream::connect(("127.0.0.1", self.proxy.port)).await.unwrap(); + let (mut sender, conn) = + hyper::client::conn::http1::handshake(TokioIo::new(tcp)).await.unwrap(); + tokio::spawn(conn); + sender.send_request(req).await.unwrap() + } + /// A raw connection tunneled through the proxy with CONNECT to `localhost:port`. async fn connect_tunnel(&self, port: u16) -> TcpStream { let mut tcp = TcpStream::connect(("127.0.0.1", self.proxy.port)).await.unwrap(); @@ -381,6 +396,51 @@ async fn https_post_with_body() { assert_eq!(body(&events, false), (b"name=yaak".to_vec(), 9)); } +/// Chrome's request headers for `optimizationguide-pa.googleapis.com`, plus the other headers +/// HTTP/2 forbids. Google resets an HTTP/2 stream that has `Host` alongside `:authority`, and +/// hyper's HTTP/2 server resets one with any of the rest. +#[tokio::test(flavor = "multi_thread")] +async fn https_h2_upstream_gets_authority_without_host() { + let h = Harness::start(1024); + let upstream = spawn_upstream(Some(h.pki.server.clone())).await; + let url = format!("https://localhost:{}/authority", upstream.port()); + + let resp = h + .client() + .post(&url) + .header("connection", "keep-alive, x-hop") + .header("x-hop", "1") + .header("keep-alive", "timeout=5") + .header("proxy-connection", "keep-alive") + .header("te", "gzip") + .header("content-type", "application/x-protobuf") + .header("sec-fetch-site", "none") + .header("sec-fetch-mode", "no-cors") + .header("sec-fetch-dest", "empty") + .header( + "user-agent", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like \ + Gecko) Chrome/156.0.0.0 Safari/537.36", + ) + .header("accept-encoding", "gzip, deflate, br, zstd") + .body(vec![0u8; 1034]) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), 200); + let expected = format!("authority=localhost:{} host=none", upstream.port()); + assert_eq!(resp.text().await.unwrap(), expected); + + let events = h.flow_events(); + let version = find!(events, ProxyEvent::ResponseStart { http_version, .. } => http_version); + assert_eq!(version, "HTTP/2"); + let captured_host = events.iter().any(|e| { + matches!(e, ProxyEvent::RequestHeader { name, value, .. } + if name == "host" && *value == format!("localhost:{}", upstream.port())) + }); + assert!(captured_host, "the captured request keeps Host"); +} + #[tokio::test(flavor = "multi_thread")] async fn https_not_found() { let h = Harness::start(1024); @@ -648,3 +708,74 @@ async fn unreachable_upstream_is_an_error() { assert!(error.to_lowercase().contains("connect"), "{error}"); assert!(!events.iter().any(|e| matches!(e, ProxyEvent::ResponseStart { .. }))); } + +/// Telegram's HTTP transport request, `POST /api` with only `Host` to say where it goes, from the +/// failing flows. `host` replaces Telegram's `149.154.175.58:80`. +fn telegram_request(host: Option<&str>, path: &str, payload: &[u8]) -> Request { + let mut builder = Request::post(path); + if let Some(host) = host { + builder = builder.header("host", host); + } + builder + .header("content-length", payload.len()) + .header("content-type", "application/x-www-form-urlencoded") + .header("connection", "Keep-Alive") + .header("accept-encoding", "gzip, deflate") + .header("accept-language", "en-US,*") + .header("user-agent", "Mozilla/5.0") + .body(full(payload.to_vec())) + .unwrap() +} + +#[tokio::test(flavor = "multi_thread")] +async fn plain_http_bare_path_is_sent_to_host() { + let h = Harness::start(1024); + let upstream = spawn_upstream(None).await; + let payload = [0xefu8; 44]; + + let resp = h.send_as_written(telegram_request(Some(&upstream.to_string()), "/echo", &payload)); + let resp = resp.await; + assert_eq!(resp.status(), 200); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), &payload[..]); + + let events = h.flow_events(); + let url = find!(events, ProxyEvent::RequestStart { url, .. } => url.clone()); + assert_eq!(url, format!("http://{upstream}/echo")); + let (status, remote_addr) = find!(events, + ProxyEvent::ResponseStart { status, remote_addr, .. } => (*status, *remote_addr)); + assert_eq!((status, remote_addr), (200, Some(upstream))); + assert_eq!(body(&events, true), (payload.to_vec(), 44)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn plain_http_bare_path_without_host_is_a_bad_request() { + let h = Harness::start(1024); + + let resp = h.send_as_written(telegram_request(None, "/api", &[0xefu8; 44])).await; + assert_eq!(resp.status(), 400); + assert_eq!(resp.headers()[CONTENT_TYPE], "text/plain; charset=utf-8"); + assert_eq!(resp.into_body().collect().await.unwrap().to_bytes(), "no host in request"); + + let events = h.flow_events(); + assert_eq!(find!(events, ProxyEvent::RequestStart { url, .. } => url.clone()), "/api"); + let error = find!(events, ProxyEvent::Error { error, .. } => error.clone()); + assert_eq!(error, "no host in request"); + assert!(!events.iter().any(|e| matches!(e, ProxyEvent::ResponseStart { .. }))); +} + +/// A request whose connection upstream is routed back into the proxy, as happened to Telegram's +/// while the system proxy was on, arrives again with a bare path. Sending that on would loop +/// forever. +#[tokio::test(flavor = "multi_thread")] +async fn plain_http_request_looped_back_to_the_proxy_stops() { + let h = Harness::start(1024); + let url = format!("http://127.0.0.1:{}/api", h.proxy.port); + + let resp = h.client().post(&url).body(vec![0xefu8; 44]).send().await.unwrap(); + assert_eq!(resp.status(), 508); + assert_eq!(resp.text().await.unwrap(), "request looped back to Yaak Proxy"); + + let events = h.flow_events(); + let error = find!(events, ProxyEvent::Error { error, .. } => error.clone()); + assert_eq!(error, "request looped back to Yaak Proxy"); +}