Something went wrong. Try again.
Hardened web server and reverse proxy: TLS/HTTP3, ACME, rate limiting, host policy, GeoIP, A/B splitting, privilege dropping
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046use crate::absplit::AbSplitRouter;use crate::balance::Balancer;use crate::config::{self, Handler as HandlerType, HttpBlock, TimeoutSettings, UpstreamRef};use crate::geoip::GeoIpService;use crate::logging::AccessLogger;use crate::proxy::{ProxyClient, ProxyError};use crate::rewrite::{RewriteAction, RewriteEngine};use crate::security::{host_policy, upstream_filter, uri_validate};use crate::static_file;use bytes::Bytes;use flate2::read::GzDecoder;use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode};use http_body_util::{BodyExt, Either, Full, Limited};use hyper::body::Body;use std::io::Read;use std::net::{IpAddr, SocketAddr};use std::sync::Arc;use std::time::Duration;use tracing::warn;
/// A location with its regex pattern compiled once at construction instead of/// per request.struct CompiledLocation { location: config::Location, regex: Option<regex::Regex>,}
/// Response body: buffered for generated and proxied responses, streamed from/// disk for static files.pub type ResponseBody = Either<Full<Bytes>, static_file::FileBody>;
pub struct HandlerService { http_block: Arc<HttpBlock>, access_logger: Arc<AccessLogger>, /// Present when this listener belongs to an `acme.enabled` block; used to /// answer HTTP-01 challenges (C51) from the shared challenge store. acme_store: Option<crate::acme::ChallengeStore>, /// Shared upstream client: DNS pinning + pooling (Invariants 41/42). proxy: ProxyClient, balancer: Balancer, geoip: Option<GeoIpService>, rewriter: RewriteEngine, ab_router: Option<AbSplitRouter>, locations: Vec<CompiledLocation>, /// "https" when this handler serves a TLS or QUIC listener. scheme: &'static str, /// Precomputed Strict-Transport-Security value (TLS listeners only). hsts: Option<HeaderValue>, max_body: usize, /// Idle limit for CONNECT tunnels (Invariant 13). idle: Duration,}
impl HandlerService { pub fn new( http_block: Arc<HttpBlock>, access_logger: Arc<AccessLogger>, acme_store: Option<crate::acme::ChallengeStore>, is_tls: bool, timeouts: &TimeoutSettings, ) -> Self { let allowed_networks = upstream_filter::parse_networks(http_block.upstream_allowed_networks.as_deref()); let trusted_proxies = upstream_filter::parse_networks(http_block.trusted_proxies.as_deref());
let proxy = ProxyClient::new( allowed_networks, Duration::from_secs(http_block.dns_cache_ttl_seconds.unwrap_or(30)), Duration::from_secs(timeouts.proxy_read), http_block .max_upstream_response_size .unwrap_or(100 * 1024 * 1024), trusted_proxies, http_block.normalize_headers_before_proxy, );
let balancer = Balancer::new(&http_block);
let geoip = http_block .geoip_db .as_deref() .and_then(|path| match GeoIpService::new(path) { Ok(svc) => Some(svc), Err(e) => { warn!("GeoIP database unavailable ({:#}); lookups disabled", e); None } });
let rewriter = RewriteEngine::new(&http_block.rewrites); let ab_router = (!http_block.ab_splits.is_empty()).then(|| AbSplitRouter::new(&http_block.ab_splits));
let locations = http_block .locations .iter() .map(|l| CompiledLocation { regex: (l.match_type == "regex") .then(|| regex::Regex::new(&l.pattern).ok()) .flatten(), location: l.clone(), }) .collect();
let scheme = if is_tls { "https" } else { "http" }; let hsts = is_tls .then_some(http_block.hsts_max_age) .flatten() .and_then(|secs| { HeaderValue::from_str(&format!("max-age={}; includeSubDomains", secs)).ok() }); let max_body = http_block.max_body_size.unwrap_or(10 * 1024 * 1024);
Self { http_block, access_logger, acme_store, proxy, balancer, geoip, rewriter, ab_router, locations, scheme, hsts, max_body, idle: Duration::from_secs(timeouts.idle), } }
/// C51 / invariant 67: serve the HTTP-01 key authorization for a known /// token as `application/octet-stream`, else 404. Returns `None` when the /// path is not an ACME challenge so normal routing proceeds. fn acme_challenge_response(&self, uri_path: &str) -> Option<Response<ResponseBody>> { const PREFIX: &str = "/.well-known/acme-challenge/"; let store = self.acme_store.as_ref()?; let token = uri_path.strip_prefix(PREFIX)?; // A token is a single path segment; reject anything with further slashes. if token.is_empty() || token.contains('/') { return Some(self.error_response(StatusCode::NOT_FOUND)); } match store.get_http01(token) { Some(key_auth) => { let mut resp = Response::new(Either::Left(Full::new(Bytes::from(key_auth)))); resp.headers_mut().insert( http::header::CONTENT_TYPE, HeaderValue::from_static("application/octet-stream"), ); Some(resp) } None => Some(self.error_response(StatusCode::NOT_FOUND)), } }
fn log( &self, remote_addr: SocketAddr, method: &Method, target: &str, resp: &Response<ResponseBody>, user_agent: &str, ) { self.access_logger.log( &remote_addr.ip().to_string(), method.as_str(), target, resp.status().as_u16(), resp.body().size_hint().lower() as usize, user_agent, ); }
/// HTTP/1.x and HTTP/2 entry point (streaming hyper body). pub async fn handle( &self, req: Request<hyper::body::Incoming>, remote_addr: SocketAddr, ) -> Response<ResponseBody> { let method = req.method().clone(); let uri = req.uri().clone(); let uri_path = uri.path().to_string(); let user_agent = req .headers() .get(http::header::USER_AGENT) .and_then(|v| v.to_str().ok()) .unwrap_or("-") .to_string();
// C51: HTTP-01 ACME challenge — answered before any other routing or // host checks so the CA can always reach it over plain HTTP. if let Some(resp) = self.acme_challenge_response(&uri_path) { self.log(remote_addr, &method, &uri_path, &resp, &user_agent); return resp; }
// Invariant 24: CONNECT tunneling — authority-form URI, handled // before URI validation (which expects path-form). if method == Method::CONNECT { let target = uri.to_string(); let resp = self.handle_connect(req, remote_addr).await; self.log(remote_addr, &method, &target, &resp, &user_agent); return resp; }
// Header/framing checks before touching the body so malformed // requests are rejected without reading their payload. if let Err(status) = self.security_checks(&req) { let resp = self.finalize(self.error_response(status)); self.log(remote_addr, &method, &uri_path, &resp, &user_agent); return resp; }
// Invariant 16: collect the body under a hard cap. The Content-Length // check in security_checks cannot bound chunked bodies — `Limited` // enforces the cap regardless of framing. let (mut parts, body) = req.into_parts(); if let Err(status) = reconcile_authority(&mut parts) { let resp = self.finalize(self.error_response(status)); self.log(remote_addr, &method, &uri_path, &resp, &user_agent); return resp; } let body_bytes = match Limited::new(body, self.max_body).collect().await { Ok(collected) => collected.to_bytes(), Err(e) => { let status = if e .downcast_ref::<http_body_util::LengthLimitError>() .is_some() { StatusCode::PAYLOAD_TOO_LARGE } else { StatusCode::BAD_REQUEST }; let resp = self.finalize(self.error_response(status)); self.log(remote_addr, &method, &uri_path, &resp, &user_agent); return resp; } };
let req = Request::from_parts(parts, body_bytes); let resp = self.handle_collected(req, remote_addr).await; self.log(remote_addr, &method, &uri_path, &resp, &user_agent); resp }
/// HTTP/3 entry point. The h3 layer hands us a `Request<()>` (pseudo-headers /// already lifted into the URI) plus the body collected separately. /// Authority is reconciled exactly as in [`Self::handle`], then the shared /// pipeline runs. pub async fn handle_h3( &self, req: Request<()>, body: Bytes, remote_addr: SocketAddr, ) -> Response<ResponseBody> { let method = req.method().clone(); let uri_path = req.uri().path().to_string(); let user_agent = req .headers() .get(http::header::USER_AGENT) .and_then(|v| v.to_str().ok()) .unwrap_or("-") .to_string();
let (mut parts, _) = req.into_parts(); if let Err(status) = reconcile_authority(&mut parts) { let resp = self.finalize(self.error_response(status)); self.log(remote_addr, &method, &uri_path, &resp, &user_agent); return resp; }
let req = Request::from_parts(parts, body); let resp = self.handle_collected(req, remote_addr).await; self.log(remote_addr, &method, &uri_path, &resp, &user_agent); resp }
/// Shared pipeline for a fully-buffered request: security checks, /// decompression guard, host policy, rewrites, GeoIP stamping, routing. /// Public so integration tests can exercise policy without a listener. pub async fn handle_collected( &self, req: Request<Bytes>, remote_addr: SocketAddr, ) -> Response<ResponseBody> { match self.process(req, remote_addr).await { Ok(resp) => self.finalize(resp), Err(status) => self.finalize(self.error_response(status)), } }
async fn process( &self, req: Request<Bytes>, remote_addr: SocketAddr, ) -> Result<Response<ResponseBody>, StatusCode> { self.security_checks(&req)?;
// Uniform body cap for entry points that collect outside `handle` // (HTTP/3, tests). if req.body().len() > self.max_body { return Err(StatusCode::PAYLOAD_TOO_LARGE); }
let method = req.method().clone();
// C46: decompression bomb guard. let mut req = self.maybe_decompress(req)?;
// Invariant 22: host policy. let host = req .headers() .get(http::header::HOST) .and_then(|v| v.to_str().ok()); if !host_policy::validate_host(host, &self.http_block) { return Err(StatusCode::BAD_REQUEST); }
if !Self::check_method_allowed(&method) { return Err(StatusCode::METHOD_NOT_ALLOWED); }
// Rewrites (first matching rule wins). let mut uri_path = req.uri().path().to_string(); match self.rewriter.apply(&uri_path) { Some(RewriteAction::Redirect { target, status }) => { return Self::redirect_response(&target, status); } Some(RewriteAction::Internal(new_path)) => { // The rewritten path is operator + capture-group derived; // re-validate before it reaches routing or upstreams. if uri_validate::validate_uri(&new_path).is_err() { return Err(StatusCode::BAD_REQUEST); } Self::set_path(&mut req, &new_path)?; uri_path = new_path; } None => {} }
// GeoIP enrichment — never trusts client-supplied x-geoip-* headers. self.stamp_geoip(req.headers_mut(), remote_addr.ip());
match self.find_location(&uri_path) { Some(loc) => self.dispatch(loc, &uri_path, req, remote_addr).await, None => { if let Some(root) = self.http_block.root.as_deref() { static_file::serve_file(root, &uri_path) .await .map(|r| r.map(Either::Right)) } else { Err(StatusCode::NOT_FOUND) } } } }
/// Invariants 11, 16, 17, 19, 25, 26, 60 — header/framing checks shared by /// every protocol entry point. Body-generic: inspects only headers, /// version and URI. fn security_checks<B>(&self, req: &Request<B>) -> Result<(), StatusCode> { // Invariant 60: max header count let max_headers = self.http_block.max_header_count.unwrap_or(100); if req.headers().len() > max_headers { return Err(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE); }
// Invariant 11: max header size let max_header_size = self.http_block.max_header_size.unwrap_or(8192); let header_size: usize = req .headers() .iter() .map(|(n, v)| n.as_str().len() + v.as_bytes().len() + 2) .sum(); if header_size > max_header_size { return Err(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE); }
// Invariant 16: reject oversized declared bodies before reading. let max_body = self.max_body; if let Some(cl) = req.headers().get(http::header::CONTENT_LENGTH) && let Ok(cl_val) = cl.to_str() && let Ok(cl_num) = cl_val.parse::<usize>() && cl_num > max_body { return Err(StatusCode::PAYLOAD_TOO_LARGE); }
// Invariant 19: HTTP/1.0 + Transfer-Encoding is invalid per RFC 9112 (C23) if req.version() == http::Version::HTTP_10 && req.headers().get(http::header::TRANSFER_ENCODING).is_some() { return Err(StatusCode::BAD_REQUEST); }
// Invariant 17: HTTP/1.x framing ambiguity — reject CL + TE if self.http_block.reject_ambiguous_framing { let has_cl = req.headers().get(http::header::CONTENT_LENGTH).is_some(); let has_te = req.headers().get(http::header::TRANSFER_ENCODING).is_some(); if has_cl && has_te { return Err(StatusCode::BAD_REQUEST); } let cl_count = req .headers() .get_all(http::header::CONTENT_LENGTH) .iter() .count(); if cl_count > 1 { return Err(StatusCode::BAD_REQUEST); } }
// Invariant 26: reject header values containing CR/LF, including // values that are not valid UTF-8 (checked at the byte level). if self.http_block.reject_headers_with_control_chars { for (_, value) in req.headers().iter() { if value.as_bytes().iter().any(|&b| b == b'\r' || b == b'\n') { return Err(StatusCode::BAD_REQUEST); } } }
// Invariant 25: strict URI parsing if let Err((status, _)) = uri_validate::validate_uri(req.uri().path()) { return Err(status); }
Ok(()) }
/// C46: decompress a gzip request body with the expansion capped *during* /// decompression — the previous implementation inflated the entire body /// into memory before checking the ratio, defeating the bomb guard. fn maybe_decompress(&self, req: Request<Bytes>) -> Result<Request<Bytes>, StatusCode> { let is_gzip = req .headers() .get(http::header::CONTENT_ENCODING) .and_then(|v| v.to_str().ok()) .map(|v| v.to_lowercase().contains("gzip")) .unwrap_or(false); if !is_gzip { return Ok(req); }
let max_ratio = self.http_block.max_decompression_ratio.unwrap_or(10); let max_size = self .http_block .max_decompression_size .unwrap_or(10 * 1024 * 1024);
let (mut parts, body) = req.into_parts(); if body.is_empty() { return Ok(Request::from_parts(parts, body)); }
let compressed_len = body.len(); let mut decoder = GzDecoder::new(&body[..]).take(max_size as u64 + 1); let mut decompressed = Vec::new(); if decoder.read_to_end(&mut decompressed).is_err() { // Not valid gzip — pass the original bytes through unchanged. return Ok(Request::from_parts(parts, body)); }
if decompressed.len() > max_size { return Err(StatusCode::PAYLOAD_TOO_LARGE); } let ratio = decompressed.len().saturating_div(compressed_len.max(1)); if ratio > max_ratio { return Err(StatusCode::PAYLOAD_TOO_LARGE); }
// The body is no longer gzip; drop the stale encoding header so // upstreams don't try to decompress plain bytes. parts.headers.remove(http::header::CONTENT_ENCODING); Ok(Request::from_parts(parts, Bytes::from(decompressed))) }
/// Replace the request path, preserving the query string. fn set_path(req: &mut Request<Bytes>, new_path: &str) -> Result<(), StatusCode> { let mut parts = req.uri().clone().into_parts(); let pq = match req.uri().query() { Some(q) => format!("{}?{}", new_path, q), None => new_path.to_string(), }; parts.path_and_query = Some(pq.parse().map_err(|_| StatusCode::BAD_REQUEST)?); *req.uri_mut() = http::Uri::from_parts(parts).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; Ok(()) }
/// Inject GeoIP headers derived from the connecting address. Any /// client-supplied x-geoip-* headers are stripped first so upstreams can /// trust the values. fn stamp_geoip(&self, headers: &mut HeaderMap, ip: IpAddr) { const GEO_HEADERS: &[&str] = &["x-geoip-country", "x-geoip-city", "x-geoip-continent"]; for h in GEO_HEADERS { headers.remove(*h); } let Some(geoip) = &self.geoip else { return }; let Some(info) = geoip.lookup(ip) else { return }; let mut put = |name: &'static str, value: Option<String>| { if let Some(v) = value && let Ok(hv) = HeaderValue::from_str(&v) { headers.insert(name, hv); } }; put("x-geoip-country", info.iso_code); put("x-geoip-city", info.city); put("x-geoip-continent", info.continent); }
fn check_method_allowed(method: &Method) -> bool { matches!( method, &Method::GET | &Method::HEAD | &Method::POST | &Method::PUT | &Method::DELETE | &Method::PATCH | &Method::OPTIONS ) }
fn find_location(&self, uri_path: &str) -> Option<&CompiledLocation> { let mut best_match: Option<&CompiledLocation> = None; let mut best_prefix_len = 0usize;
for compiled in &self.locations { match compiled.location.match_type.as_str() { "exact" => { if compiled.location.pattern == uri_path { return Some(compiled); } } "prefix" => { if uri_path.starts_with(&compiled.location.pattern) && compiled.location.pattern.len() > best_prefix_len { best_prefix_len = compiled.location.pattern.len(); best_match = Some(compiled); } } "regex" => { if let Some(re) = &compiled.regex && re.is_match(uri_path) { return Some(compiled); } } _ => {} } }
best_match }
async fn dispatch( &self, compiled: &CompiledLocation, uri_path: &str, req: Request<Bytes>, remote_addr: SocketAddr, ) -> Result<Response<ResponseBody>, StatusCode> { let handler = &compiled.location.handler; match handler.handler_type.as_str() { "static" => { let root = handler .root .as_deref() .or(self.http_block.root.as_deref()) .ok_or(StatusCode::INTERNAL_SERVER_ERROR)?; static_file::serve_file(root, uri_path) .await .map(|r| r.map(Either::Right)) } "proxy" | "reverse_proxy" => self .dispatch_proxy(handler, req, remote_addr) .await .map(|r| r.map(Either::Left)), "return" => { let status = handler.status.unwrap_or(200); let body = handler.target.as_deref().unwrap_or(""); let status_code = StatusCode::from_u16(status).unwrap_or(StatusCode::OK); let mut resp = Response::new(Either::Left(Full::new(Bytes::from(body.to_string())))); *resp.status_mut() = status_code; resp.headers_mut().insert( http::header::CONTENT_TYPE, HeaderValue::from_static("text/plain; charset=utf-8"), ); Ok(resp) } "redirect" => { let target = handler .target .as_deref() .ok_or(StatusCode::INTERNAL_SERVER_ERROR)?; Self::redirect_response(target, handler.status.unwrap_or(301)) } "deny" => Err(StatusCode::FORBIDDEN), _ => Err(StatusCode::NOT_IMPLEMENTED), } }
fn redirect_response(target: &str, status: u16) -> Result<Response<ResponseBody>, StatusCode> { let status_code = StatusCode::from_u16(status).unwrap_or(StatusCode::MOVED_PERMANENTLY); let location = HeaderValue::from_str(target).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let mut resp = Response::new(Either::Left(Full::new(Bytes::new()))); *resp.status_mut() = status_code; resp.headers_mut().insert(http::header::LOCATION, location); Ok(resp) }
/// Pick the upstream pool for a proxy location: A/B split group (keyed by /// the split's session cookie, else client IP) when configured, else the location's inline upstream, else the block's /// first shared upstream. fn select_upstream<'a>( &'a self, handler: &'a HandlerType, client: IpAddr, req: &Request<Bytes>, ) -> Result<&'a UpstreamRef, StatusCode> { if let Some(split_name) = handler.ab_split.as_deref() { if let Some(router) = &self.ab_router && let Some(split) = self .http_block .ab_splits .iter() .find(|s| s.name == split_name) { // Contract §2: sticky routing is per session cookie. The // split's `cookie` value keys assignment; without one the // client address does. let session_id = split .cookie .as_deref() .and_then(|name| cookie_value(req.headers(), name)) .unwrap_or_else(|| client.to_string()); if let Some(group_name) = router.resolve(split_name, &session_id) && let Some(group) = split.groups.iter().find(|g| g.name == group_name) && let Some(upstream) = config::resolve_upstream(&self.http_block, &group.upstream) { return Ok(upstream); } } warn!( "ab_split '{}' did not resolve to a configured upstream", split_name ); return Err(StatusCode::BAD_GATEWAY); }
handler .upstream .as_ref() .or_else(|| self.http_block.upstreams.first()) .ok_or(StatusCode::BAD_GATEWAY) }
/// Proxy with load balancing and connect-failure failover: candidates are /// tried in balancer order; a server that refuses the TCP connection is /// skipped, any HTTP-level outcome is final. async fn dispatch_proxy( &self, handler: &HandlerType, req: Request<Bytes>, remote_addr: SocketAddr, ) -> Result<Response<Full<Bytes>>, StatusCode> { let upstream = self.select_upstream(handler, remote_addr.ip(), &req)?; let (parts, body) = req.into_parts(); let mut last_status = StatusCode::BAD_GATEWAY;
for idx in self.balancer.candidates(upstream, remote_addr.ip()) { let Some(server) = upstream.servers.get(idx) else { continue; };
// Invariant 41: enforced at request time, not only at config load. if upstream_filter::is_upstream_allowed( &server.address, self.http_block.upstream_allowed_networks.as_deref(), ) .is_err() { last_status = StatusCode::FORBIDDEN; continue; }
let _inflight = self.balancer.track(upstream, idx); match self .proxy .forward( &server.address, &parts, body.clone(), remote_addr.ip(), self.scheme, ) .await { Ok(resp) => return Ok(resp), Err(ProxyError::Connect) => { warn!( "upstream {} unreachable; trying next candidate", server.address ); last_status = StatusCode::BAD_GATEWAY; } Err(ProxyError::Status(status)) => return Err(status), } }
Err(last_status) }
/// Invariant 24: CONNECT is denied (405) unless `allow_connect` is set, /// and then only for targets on the explicit allowlist. Permitted targets /// are resolved through the pinning resolver (Invariants 41/42), the /// tunnel is established first, and the 200 response hands the connection /// over to a bidirectional copy via hyper's upgrade mechanism. async fn handle_connect( &self, req: Request<hyper::body::Incoming>, _remote_addr: SocketAddr, ) -> Response<ResponseBody> { if !self.http_block.allow_connect { return self .simple_response(StatusCode::METHOD_NOT_ALLOWED, "405 Method Not Allowed\n"); }
let Some(authority) = req.uri().authority().map(|a| a.as_str().to_string()) else { return self.simple_response( StatusCode::BAD_REQUEST, "CONNECT requires an authority-form target\n", ); };
let (host, port) = upstream_filter::split_host_port(&authority); let Some(port) = port else { return self.simple_response( StatusCode::BAD_REQUEST, "CONNECT target must include an explicit port\n", ); }; let host = host.to_string();
if !self.connect_target_permitted(&host, port, &authority) { return self.simple_response(StatusCode::FORBIDDEN, "CONNECT target not permitted\n"); }
// Resolve and vet the target (Invariant 41 applies to tunnels too); // connections go to the vetted addresses, never a re-resolution. let vetted: Vec<SocketAddr> = if let Ok(ip) = host.parse::<IpAddr>() { let allowed = upstream_filter::parse_networks( self.http_block.upstream_allowed_networks.as_deref(), ); if !upstream_filter::ip_allowed(ip, &allowed) { return self .simple_response(StatusCode::FORBIDDEN, "CONNECT target not permitted\n"); } vec![SocketAddr::new(ip, port)] } else { match self.proxy.resolver().resolve_checked(&host).await { Ok(addrs) => addrs .into_iter() .map(|sa| SocketAddr::new(sa.ip(), port)) .collect(), Err(e) => { warn!("CONNECT target '{}' rejected: {}", host, e); return self .simple_response(StatusCode::FORBIDDEN, "CONNECT target not permitted\n"); } } };
let mut upstream = None; for addr in vetted { match tokio::time::timeout( Duration::from_secs(10), tokio::net::TcpStream::connect(addr), ) .await { Ok(Ok(stream)) => { upstream = Some(stream); break; } _ => continue, } } let Some(mut upstream) = upstream else { return self.simple_response(StatusCode::BAD_GATEWAY, "502 Bad Gateway\n"); };
// The upgrade resolves after the 200 below is flushed; from then on // bytes are copied verbatim in both directions until either side // closes or the tunnel sits idle past the idle timeout. let idle = self.idle; tokio::spawn(async move { match hyper::upgrade::on(req).await { Ok(upgraded) => { let mut client_io = hyper_util::rt::TokioIo::new(upgraded); let _ = crate::security::limits::copy_bidirectional_idle( &mut client_io, &mut upstream, idle, ) .await; } Err(e) => warn!("CONNECT upgrade failed: {}", e), } });
self.simple_response(StatusCode::OK, "") }
/// A CONNECT target is permitted when it matches `connect_allowed_targets` /// ("host:port" exact, or bare "host" for any port) or, absent that list, /// when it exactly matches a `connect_upstream` server address. fn connect_target_permitted(&self, host: &str, port: u16, full_authority: &str) -> bool { if let Some(targets) = &self.http_block.connect_allowed_targets { return targets.iter().any(|t| { let (t_host, t_port) = upstream_filter::split_host_port(t); t_host.eq_ignore_ascii_case(host) && t_port.is_none_or(|p| p == port) }); } if let Some(up) = &self.http_block.connect_upstream { return up.servers.iter().any(|s| { if s.address.eq_ignore_ascii_case(full_authority) { return true; } let (s_host, s_port) = upstream_filter::split_host_port(&s.address); s_host.eq_ignore_ascii_case(host) && s_port == Some(port) }); } false }
/// Stamp response-wide headers (currently HSTS for TLS listeners). fn finalize(&self, mut resp: Response<ResponseBody>) -> Response<ResponseBody> { if let Some(hsts) = &self.hsts { resp.headers_mut() .entry(http::header::STRICT_TRANSPORT_SECURITY) .or_insert_with(|| hsts.clone()); } resp }
fn simple_response(&self, status: StatusCode, body: &'static str) -> Response<ResponseBody> { let mut resp = Response::new(Either::Left(Full::new(Bytes::from_static(body.as_bytes())))); *resp.status_mut() = status; if !body.is_empty() { resp.headers_mut().insert( http::header::CONTENT_TYPE, HeaderValue::from_static("text/plain; charset=utf-8"), ); } self.finalize(resp) }
fn error_response(&self, status: StatusCode) -> Response<ResponseBody> { let phrase = status.canonical_reason().unwrap_or("Error"); let body = format!("{} {}\n", status.as_u16(), phrase); let mut resp = Response::new(Either::Left(Full::new(Bytes::from(body)))); *resp.status_mut() = status; resp.headers_mut().insert( http::header::CONTENT_TYPE, HeaderValue::from_static("text/plain; charset=utf-8"), ); resp }}
/// The TLS SNI name of the connection a request arrived on, attached to the/// request's extensions by the TLS and QUIC listeners (Invariant 50).#[derive(Clone, Debug)]pub struct TlsServerName(Arc<str>);
impl TlsServerName { pub fn new(name: &str) -> Self { Self(Arc::from(name)) }}
/// Invariant 50 — authority consistency for every protocol entry point.////// When the request target carries an authority (HTTP/2 and HTTP/3/// `:authority`, HTTP/1.1 absolute-form), an explicit `Host` that names a/// different authority is a conflict (400). Without a `Host` header the/// authority is copied into one, so host policy and proxying see the same/// value for every protocol — HTTP/2 clients normally send no `Host` at all./// On TLS connections the resulting host must also equal the SNI name (400/// otherwise); clients send no SNI for IP-literal hosts, so those pass.fn reconcile_authority(parts: &mut http::request::Parts) -> Result<(), StatusCode> { let headers = &mut parts.headers; if let Some(authority) = parts.uri.authority() { match headers.get(http::header::HOST) { Some(host) if !host .as_bytes() .eq_ignore_ascii_case(authority.as_str().as_bytes()) => { return Err(StatusCode::BAD_REQUEST); } Some(_) => {} None => { let value = HeaderValue::from_str(authority.as_str()) .map_err(|_| StatusCode::BAD_REQUEST)?; headers.insert(http::header::HOST, value); } } } if let Some(TlsServerName(sni)) = parts.extensions.get::<TlsServerName>() { let host = headers .get(http::header::HOST) .and_then(|h| h.to_str().ok()) .map(|h| upstream_filter::split_host_port(h).0); if let Some(host) = host { let canonical = |name: &str| name.trim_end_matches('.').to_ascii_lowercase(); if canonical(host) != canonical(sni) { return Err(StatusCode::BAD_REQUEST); } } } Ok(())}
/// Value of cookie `name` across the request's `Cookie` headers, if present/// and non-empty.fn cookie_value(headers: &HeaderMap, name: &str) -> Option<String> { headers .get_all(http::header::COOKIE) .iter() .filter_map(|h| h.to_str().ok()) .flat_map(|raw| raw.split(';')) .filter_map(|pair| pair.split_once('=')) .find(|(k, v)| k.trim() == name && !v.trim().is_empty()) .map(|(_, v)| v.trim().to_string())}
#[cfg(test)]mod tests { use super::*;
fn headers(pairs: &[(http::header::HeaderName, &str)]) -> HeaderMap { let mut h = HeaderMap::new(); for (k, v) in pairs { h.append(k.clone(), HeaderValue::from_str(v).unwrap()); } h }
/// Request parts for `uri`, `Host` value, and optional SNI name. fn parts(uri: &str, host: Option<&str>, sni: Option<&str>) -> http::request::Parts { let mut req = Request::builder().uri(uri); if let Some(host) = host { req = req.header(http::header::HOST, host); } let (mut parts, ()) = req.body(()).unwrap().into_parts(); if let Some(sni) = sni { parts.extensions.insert(TlsServerName::new(sni)); } parts }
#[test] fn authority_fills_missing_host_and_rejects_conflicts() { let uri = "https://Example.com:8443/x"; let mut p = parts(uri, None, None); assert_eq!(reconcile_authority(&mut p), Ok(())); assert_eq!(p.headers[http::header::HOST], "Example.com:8443");
assert_eq!( reconcile_authority(&mut parts(uri, Some("example.COM:8443"), None)), Ok(()) ); assert_eq!( reconcile_authority(&mut parts(uri, Some("evil.com"), None)), Err(StatusCode::BAD_REQUEST) );
// Origin-form (plain HTTP/1.1): Host is left to host policy. let mut p = parts("/x", None, None); assert_eq!(reconcile_authority(&mut p), Ok(())); assert!(p.headers.get(http::header::HOST).is_none()); }
#[test] fn sni_must_match_effective_host() { // Host (any port, case, trailing dot) matching SNI passes. assert_eq!( reconcile_authority(&mut parts( "/x", Some("Example.com:443"), Some("example.com") )), Ok(()) ); assert_eq!( reconcile_authority(&mut parts("/x", Some("example.com."), Some("example.com"))), Ok(()) ); // HTTP/2: the :authority-derived Host is checked too. assert_eq!( reconcile_authority(&mut parts( "https://example.com/x", None, Some("example.com") )), Ok(()) ); assert_eq!( reconcile_authority(&mut parts("https://other.com/x", None, Some("example.com"))), Err(StatusCode::BAD_REQUEST) ); assert_eq!( reconcile_authority(&mut parts("/x", Some("other.com"), Some("example.com"))), Err(StatusCode::BAD_REQUEST) ); // No SNI (plaintext, or an IP-literal host): nothing to compare. assert_eq!( reconcile_authority(&mut parts("/x", Some("127.0.0.1"), None)), Ok(()) ); }
#[test] fn cookie_value_finds_named_cookie_across_headers() { let h = headers(&[ (http::header::COOKIE, "a=1; flexd_ab=grp7"), (http::header::COOKIE, "b=2"), ]); assert_eq!(cookie_value(&h, "flexd_ab").as_deref(), Some("grp7")); assert_eq!(cookie_value(&h, "b").as_deref(), Some("2")); assert_eq!(cookie_value(&h, "missing"), None); assert_eq!( cookie_value(&headers(&[(http::header::COOKIE, "flexd_ab=")]), "flexd_ab"), None ); }}