use crate::{v2_epoch, State}; use anyhow::{ensure, Result}; use base64::Engine; use bytes::Bytes; use http_body_util::{BodyExt, Full}; use hyper::{body::Incoming, Request, Response, StatusCode}; use hyper_util::rt::TokioIo; use std::convert::Infallible; use std::sync::{Arc, atomic::Ordering}; use std::time::Duration; type Body = Full; const MAX_REQUEST_BODY: usize = 100 * 1024; const MAX_COMMAND_BODY: usize = 180 * 1024; #[derive(serde::Deserialize)] #[serde(deny_unknown_fields)] struct SendRequest { url: String, method: String, #[serde(default)] headers: Vec<(String, String)>, #[serde(default)] body_b64: String, } fn request_parts(query: SendRequest, own_name: &str) -> Result<(url::Url, hyper::Method, http::HeaderMap, Bytes)> { let target = url::Url::parse(&query.url)?; validate_target(&target, own_name)?; ensure!(matches!(query.method.as_str(), "GET" | "HEAD" | "POST" | "PUT" | "PATCH" | "DELETE" | "OPTIONS"), "unsupported method"); let method = hyper::Method::from_bytes(query.method.as_bytes())?; ensure!(query.body_b64.len() <= 4 * MAX_REQUEST_BODY.div_ceil(3), "request body too large"); let body = base64::engine::general_purpose::STANDARD.decode(&query.body_b64)?; ensure!(body.len() <= MAX_REQUEST_BODY, "request body too large"); ensure!(!matches!(method, hyper::Method::GET | hyper::Method::HEAD) || body.is_empty(), "GET and HEAD bodies unsupported"); ensure!(query.headers.len() <= 64 && query.headers.iter().map(|(k,v)| k.len()+v.len()).sum::() <= 16384, "too many request headers"); let mut headers = http::HeaderMap::new(); for (name, value) in query.headers { let name = http::header::HeaderName::from_bytes(name.as_bytes())?; ensure!(!tlproxy_common::HOP_BY_HOP.contains(&name.as_str()) && !matches!(name.as_str(), "host" | "content-length" | "expect" | "accept-encoding") && !name.as_str().starts_with("proxy-") && !name.as_str().starts_with("x-attested-relay-"), "reserved request header"); ensure!(!headers.contains_key(&name), "duplicate request header"); headers.insert(name, http::header::HeaderValue::from_str(&value)?); } Ok((target, method, headers, Bytes::from(body))) } async fn send_command(state: Arc, req: Request) -> Result> { if req.method() != hyper::Method::POST { return Ok(error(StatusCode::METHOD_NOT_ALLOWED, "POST required\n")); } if req.uri().query().is_some() || req.headers().get("content-type").is_none_or(|v| v != "application/json") { return Ok(error(StatusCode::BAD_REQUEST, "JSON body required\n")); } let Ok(permit) = state.requests.try_acquire() else { return Ok(error(StatusCode::SERVICE_UNAVAILABLE, "relay is at capacity\n")); }; // Admission and deadline precede body buffering; disconnected clients cannot // cancel an already forwarded operation before its audit record is sealed. let parsed = tokio::time::timeout(Duration::from_secs(30), async { let mut body = req.into_body(); let mut raw = zeroize::Zeroizing::new(Vec::new()); while let Some(frame) = body.frame().await { let frame = frame?; ensure!(frame.trailers_ref().is_none(), "command trailers unsupported"); if let Some(data) = frame.data_ref() { ensure!(data.len() <= MAX_COMMAND_BODY.saturating_sub(raw.len()), "command too large"); raw.extend_from_slice(data); } } request_parts(serde_json::from_slice(&raw)?, &state.config.tls_dns_name) }).await; let (target, method, headers, body) = match parsed { Ok(Ok(parts)) => parts, _ => return Ok(error(StatusCode::BAD_REQUEST, "invalid or incomplete request command\n")), }; forward(&state, target, method, headers, body, true, permit).await } #[derive(serde::Serialize)] struct Exchange { url: String, method: String, status: Option, headers: Vec<(String, String)>, trailers: Vec<(String, String)>, #[serde(serialize_with = "base64_body")] response_body_b64: Vec, complete: bool, } fn base64_body(bytes: &[u8], serializer: S) -> Result { serializer.serialize_str(&base64::engine::general_purpose::STANDARD.encode(bytes)) } fn canonical_record(record: &serde_json::Value) -> Result> { // Value's ordered maps use CBOR canonical key ordering recursively. Audit // records contain integers, strings, arrays, booleans and null, never floats. Ok(serde_cbor::to_vec(&serde_cbor::value::to_value(record)?)?) } fn response(status: StatusCode, content_type: &str, body: impl Into) -> Response { Response::builder().status(status).header("content-type", content_type) .header("cache-control", "no-store").body(Full::new(body.into())).unwrap() } fn error(status: StatusCode, message: &'static str) -> Response { response(status, "text/plain; charset=utf-8", message) } pub async fn handle(state: Arc, req: Request) -> Result, Infallible> { // Detached admitted work retains its request permit and epoch lease until // capture/sealing/persistence finishes, even if the sender disconnects. // dispatch uses try_acquire before outbound traffic, bounding live workers. let result = tokio::spawn(dispatch(state, req)).await; Ok(match result { Ok(Ok(response)) => response, _ => error(StatusCode::BAD_GATEWAY, "relay operation failed\n"), }) } async fn dispatch(state: Arc, req: Request) -> Result> { if matches!(req.uri().path(), "/v1/pastes" | "/v1/pastes/read" | "/v1/commands/read_pastebin_tag" | "/v1/commands/write_pastebin") { return crate::v2_paste::handle(state, req).await; } if req.uri().path() == "/v1/commands/send_request" { return send_command(state, req).await; } if req.method() != hyper::Method::GET { return Ok(error(StatusCode::METHOD_NOT_ALLOWED, "GET required\n")); } if req.headers().contains_key("transfer-encoding") || req.headers().get("content-length").is_some_and(|v| v != "0") { return Ok(error(StatusCode::BAD_REQUEST, "GET body is not supported\n")); } match req.uri().path() { "/v1/attestation" => { let values: Vec<_> = url::form_urlencoded::parse(req.uri().query().unwrap_or("").as_bytes()).collect(); ensure!(values.len() == 1 && values[0].0 == "nonce", "nonce required"); let nonce = base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(values[0].1.as_bytes())?; let attestation = state.attestation(&nonce).await?; return Ok(response(StatusCode::OK, "application/json", serde_json::to_vec(&attestation)?)); } "/v1/status" => return Ok(response(StatusCode::OK, "application/json", serde_json::to_vec(&state.policy().await?)?)), "/" | "/index.html" => return Ok(response(StatusCode::OK, "text/plain; charset=utf-8", "Attested relay v2. Use the attested-relay Python client with an independently pinned PCR0. Outer HTTPS GETs carry an inner TLS stream. Plain web_fetch cannot perform TLS or verify Nitro signatures by itself. Public puzzles unlock recorded traffic after calibrated sequential RandomX work.\n")), _ => (), } let Some(destination) = req.uri().path_and_query().map(|v| v.as_str()).and_then(|path| path.strip_prefix("/f/https/")) else { return Ok(error(StatusCode::NOT_FOUND, "use /f/https//\n")); }; let target = url::Url::parse(&format!("https://{destination}"))?; validate_target(&target, &state.config.tls_dns_name)?; let Ok(permit) = state.requests.try_acquire() else { return Ok(error(StatusCode::SERVICE_UNAVAILABLE, "relay is at capacity\n")); }; forward(&state, target, hyper::Method::GET, http::HeaderMap::new(), Bytes::new(), false, permit).await } async fn forward(state: &State, target: url::Url, method: hyper::Method, request_headers: http::HeaderMap, request_body: Bytes, command: bool, _permit: tokio::sync::SemaphorePermit<'_>) -> Result> { let now = state.attester.trusted_time_ms_uncached()?; let Some(epoch) = state.active.read().await.as_ref().filter(|epoch| epoch.accepts(now)).cloned() else { return Ok(error(StatusCode::SERVICE_UNAVAILABLE, "waiting for a published epoch\n")); }; if !state.network.is_ready() { return Ok(error(StatusCode::SERVICE_UNAVAILABLE, "waiting for Mullvad tunnel\n")); } // Reserve before outbound traffic; an exhausted counter must not cause an // upstream request whose encrypted record cannot be assigned a nonce. let sequence = next_sequence(&epoch.sequence)?; // Keep admission charged until encrypted record persistence finishes. let duration = Duration::from_secs(state.config.request_timeout_seconds); let mut history = Vec::new(); let forwarded = tokio::time::timeout(duration, fetch(state, target.clone(), &method, &request_headers, &request_body, !command, &mut history)).await; let (status, headers, body, outcome) = match forwarded { Ok(Ok(result)) => (result.0, result.1, result.2, "complete"), Ok(Err(_)) => (StatusCode::BAD_GATEWAY, http::HeaderMap::new(), Bytes::from_static(b"upstream failed\n"), "upstream_error"), Err(_) => (StatusCode::GATEWAY_TIMEOUT, http::HeaderMap::new(), Bytes::from_static(b"upstream deadline exceeded\n"), "timeout"), }; let record = serde_json::json!({ "version": 3, "software_version": crate::VERSION, "epoch": epoch.manifest.puzzle.epoch, "sequence": sequence, "started_at_ms": now, "outcome": outcome, "request": {"method":method.as_str(), "url":target.as_str(), "headers":request_headers.iter().map(|(k,v)|(k.as_str(),base64::engine::general_purpose::STANDARD.encode(v.as_bytes()))).collect::>(), "body_b64":base64::engine::general_purpose::STANDARD.encode(&request_body)}, "exchanges": history, }); let plaintext = zeroize::Zeroizing::new(canonical_record(&record)?); let sealed = relay_timelock::encrypt_record(&epoch.manifest, &epoch.key, sequence, &plaintext)?; let sealed_bytes = serde_json::to_vec(&sealed)?; let name = tokio::time::timeout(Duration::from_secs(30), v2_epoch::publish(state, "record.json", &sealed_bytes)).await??; if command { return Ok(response(StatusCode::OK, "application/json", serde_json::to_vec(&serde_json::json!({ "status":status.as_u16(), "headers":headers.iter().map(|(k,v)|(k.as_str(),base64::engine::general_purpose::STANDARD.encode(v.as_bytes()))).collect::>(), "body_b64":base64::engine::general_purpose::STANDARD.encode(&body), "record":name,"puzzle":epoch.puzzle_artifact,"evidence":epoch.evidence_artifact }))?)); } let mut result = Response::builder().status(status); for (name, value) in &headers { if !tlproxy_common::HOP_BY_HOP.contains(&name.as_str()) && ![ "set-cookie", "content-length", "cache-control", "location", "alt-svc", ].contains(&name.as_str()) && !name.as_str().starts_with("x-attested-relay-") { result = result.header(name, value); } } Ok(result.header("cache-control", "no-store").header("x-attested-relay-record", name) .header("x-attested-relay-puzzle", &epoch.puzzle_artifact) .header("x-attested-relay-evidence", &epoch.evidence_artifact) .body(Full::new(body))?) } pub(crate) fn next_sequence(sequence: &std::sync::atomic::AtomicU64) -> Result { sequence.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1)) .map_err(|_| anyhow::anyhow!("record sequence exhausted")) } fn validate_target(target: &url::Url, own_name: &str) -> Result<()> { ensure!(target.scheme() == "https" && target.username().is_empty() && target.password().is_none() && target.fragment().is_none(), "HTTPS target required"); ensure!(target.port_or_known_default() == Some(443), "only HTTPS port 443 supported"); let host = target.host_str().ok_or_else(|| anyhow::anyhow!("missing host"))?; ensure!(crate::transport::valid_host(host) && !host.eq_ignore_ascii_case(own_name), "invalid target host"); ensure!(!host.eq_ignore_ascii_case("localhost") && !host.ends_with(".localhost") && !host.ends_with(".internal"), "local target refused"); if let Ok(ip) = host.parse::() { ensure!(crate::net::public_destination(ip), "private target refused"); } Ok(()) } async fn connect_upstream(state: &State, host: &str) -> Result { // Deterministic local TLS fixture, only in explicitly unmeasured --dev mode. // Still verifies a real certificate chain and the original target DNS name. if state.dev { if let Ok(address) = std::env::var("RELAY_DEV_UPSTREAM_ADDRESS") { let address: std::net::SocketAddr = address.parse()?; ensure!(address.ip().is_loopback(), "development upstream must be loopback"); let path = std::env::var("RELAY_DEV_UPSTREAM_CA_DER")?; let certificate = tokio::fs::read(path).await?; ensure!(certificate.len() <= 16384, "development CA exceeds limit"); let mut roots = rustls::RootCertStore::empty(); roots.add(rustls_pki_types::CertificateDer::from(certificate))?; let config = rustls::ClientConfig::builder().with_root_certificates(roots).with_no_client_auth(); let raw = state.transport.connect_ip(address.ip(), address.port()).await?; let name = rustls_pki_types::ServerName::try_from(host.to_owned())?; let tls = tokio_rustls::TlsConnector::from(Arc::new(config)).connect(name, raw).await?; return Ok(Box::new(tls)); } } Ok(Box::new(state.dialer.connect_tls(host, 443, crate::relay::Purpose::Upstream).await?.stream)) } async fn fetch(state: &State, mut target: url::Url, method: &hyper::Method, headers: &http::HeaderMap, body: &Bytes, follow_redirects: bool, history: &mut Vec) -> Result<(StatusCode, http::HeaderMap, Bytes)> { let mut total = 0usize; for redirect in 0..=5 { validate_target(&target, &state.config.tls_dns_name)?; let host = target.host_str().unwrap(); let upstream = connect_upstream(state, host).await?; let (mut sender, connection) = hyper::client::conn::http1::handshake(TokioIo::new(upstream)).await?; tokio::spawn(async move { let _ = connection.await; }); let path = match target.query() { Some(q) => format!("{}?{q}", target.path()), None => target.path().to_string() }; history.push(Exchange {url:target.to_string(),method:method.to_string(),status:None,headers:Vec::new(),trailers:Vec::new(),response_body_b64:Vec::new(),complete:false}); let mut message = Request::builder().method(method.clone()).uri(path) .header("host", host).header("accept-encoding", "identity") .header("user-agent", "attested-relay/2").body(Full::new(body.clone()))?; for (name, value) in headers { message.headers_mut().insert(name.clone(), value.clone()); } let received = sender.send_request(message).await?; let (parts, mut incoming) = received.into_parts(); ensure!(parts.headers.iter().map(|(k,v)|k.as_str().len()+v.len()).sum::() <= 65536, "upstream headers too large"); let event = history.last_mut().unwrap(); event.status = Some(parts.status.as_u16()); event.headers = parts.headers.iter().map(|(k,v)|(k.to_string(),base64::engine::general_purpose::STANDARD.encode(v.as_bytes()))).collect(); while let Some(frame) = incoming.frame().await { let frame = frame?; if let Some(data) = frame.data_ref() { ensure!(data.len() <= state.config.max_response_bytes.saturating_sub(total), "upstream response limit exceeded"); total += data.len(); event.response_body_b64.extend_from_slice(data); } if let Some(trailers) = frame.trailers_ref() { ensure!(trailers.iter().map(|(k,v)|k.as_str().len()+v.len()).sum::() <= 65536, "upstream trailers too large"); event.trailers = trailers.iter().map(|(k,v)|(k.to_string(),base64::engine::general_purpose::STANDARD.encode(v.as_bytes()))).collect(); } } event.complete = true; if follow_redirects && matches!(parts.status.as_u16(), 301|302|303|307|308) { ensure!(redirect < 5, "redirect limit exceeded"); let location = parts.headers.get("location").ok_or_else(|| anyhow::anyhow!("redirect without location"))?.to_str()?; target = target.join(location)?; continue; } return Ok((parts.status, parts.headers, Bytes::copy_from_slice(&event.response_body_b64))); } anyhow::bail!("redirect limit exceeded") } #[cfg(test)] mod tests { use super::*; #[test] fn request_commands_reject_routing_and_framing_overrides() { for name in ["Host", "Content-Length", "Transfer-Encoding", "Connection", "Expect", "Proxy-Authorization", "Upgrade"] { let query = SendRequest { url:"https://example.com/".into(),method:"POST".into(),headers:vec![(name.into(),"value".into())],body_b64:String::new() }; assert!(request_parts(query,"relay.example").is_err(), "{name}"); } for url in ["http://example.com/", "https://127.0.0.1/", "https://example.com:8443/", "https://user:secret@example.com/"] { let query = SendRequest { url:url.into(),method:"POST".into(),headers:vec![],body_b64:String::new() }; assert!(request_parts(query,"relay.example").is_err()); } for method in ["CONNECT", "TRACE", "post"] { let query = SendRequest { url:"https://example.com/".into(),method:method.into(),headers:vec![],body_b64:String::new() }; assert!(request_parts(query,"relay.example").is_err()); } } #[test] fn record_encoding_uses_canonical_cbor_key_order() { let encoded = canonical_record(&serde_json::json!({"aa": 1, "b": 2})).unwrap(); // Independently written CBOR vector: shorter encoded key precedes aa, // even though the JSON object's lexical ordering puts aa first. assert_eq!(hex::encode(encoded), "a261620262616101"); } #[test] fn sequence_exhaustion_never_wraps_or_reuses_a_nonce() { let sequence = std::sync::atomic::AtomicU64::new(u64::MAX-1); assert_eq!(next_sequence(&sequence).unwrap(), u64::MAX-1); assert!(next_sequence(&sequence).is_err()); assert!(next_sequence(&sequence).is_err()); assert_eq!(sequence.load(Ordering::Relaxed), u64::MAX); } #[test] fn rejects_private_redirects_and_credentials() { for address in ["http://example.com/", "https://127.0.0.1/", "https://169.254.169.254/", "https://10.0.0.1/", "https://u:p@example.com/", "https://example.com:444/", "https://relay.example/", "https://localhost/"] { assert!(validate_target(&url::Url::parse(address).unwrap(), "relay.example").is_err(), "{address}"); } assert!(validate_target(&url::Url::parse("https://example.com/path?q=x").unwrap(), "relay.example").is_ok()); } }