rings_node/onion/proxy/http/
mod.rs

1//! Native HTTP CONNECT ingress for onion proxy clients.
2
3use std::net::SocketAddr;
4use std::sync::Arc;
5use std::time::Duration;
6
7use tokio::io::AsyncReadExt;
8use tokio::io::AsyncWriteExt;
9use tokio::net::TcpListener;
10use tokio::net::TcpStream;
11use tokio::sync::Semaphore;
12use tokio::time::timeout;
13
14use super::OnionProxyConfig;
15use super::OnionProxyTarget;
16use crate::error::Error;
17use crate::error::Result;
18use crate::onion::tcp::NativeOnionCircuitHandle;
19use crate::onion::OnionServiceName;
20use crate::processor::Processor;
21
22const MAX_CONNECT_HEADER_BYTES: usize = 8192;
23/// Default deadline for receiving a complete CONNECT header.
24pub const DEFAULT_CONNECT_HEADER_TIMEOUT_SECS: u64 = 10;
25/// Default concurrent connection bound for the native CONNECT ingress.
26pub const DEFAULT_MAX_CONNECT_CONNECTIONS: usize = 1024;
27
28/// Return the default CONNECT header deadline in seconds.
29pub const fn default_connect_header_timeout_secs() -> u64 {
30    DEFAULT_CONNECT_HEADER_TIMEOUT_SECS
31}
32
33/// Return the default native CONNECT ingress concurrency bound.
34pub const fn default_max_connect_connections() -> usize {
35    DEFAULT_MAX_CONNECT_CONNECTIONS
36}
37
38/// Runtime options for the native onion HTTP CONNECT proxy.
39#[derive(Clone, Debug, Eq, PartialEq)]
40pub struct OnionHttpProxyOptions {
41    /// Local bind address.
42    pub listen_addr: SocketAddr,
43    /// TCP onion-exit service used for local CONNECT requests.
44    pub service: OnionServiceName,
45    /// Desired hop count including the exit. `0` uses node default.
46    pub hop_count: usize,
47    /// Whether route selection may use fewer hops when too few relays are live.
48    pub allow_short_paths: bool,
49    /// Maximum concurrent local CONNECT requests accepted by this ingress.
50    pub max_connections: usize,
51    /// Deadline for receiving a complete CONNECT header.
52    pub header_timeout: Duration,
53}
54
55impl OnionHttpProxyOptions {
56    /// Build options with production defaults for resource bounds.
57    pub fn new(
58        listen_addr: SocketAddr,
59        service: OnionServiceName,
60        hop_count: usize,
61        allow_short_paths: bool,
62    ) -> Self {
63        Self {
64            listen_addr,
65            service,
66            hop_count,
67            allow_short_paths,
68            max_connections: DEFAULT_MAX_CONNECT_CONNECTIONS,
69            header_timeout: Duration::from_secs(DEFAULT_CONNECT_HEADER_TIMEOUT_SECS),
70        }
71    }
72
73    fn validate(&self) -> Result<()> {
74        if self.max_connections == 0 {
75            return Err(Error::InvalidConfig(
76                "onion_http_proxy_max_connections must be greater than zero".to_string(),
77            ));
78        }
79        if self.header_timeout.is_zero() {
80            return Err(Error::InvalidConfig(
81                "onion_http_proxy_header_timeout_secs must be greater than zero".to_string(),
82            ));
83        }
84        self.proxy_config().map(|_| ())
85    }
86
87    fn proxy_config(&self) -> Result<OnionProxyConfig> {
88        OnionProxyConfig::tcp_connect_service(
89            self.service.clone(),
90            self.hop_count,
91            self.allow_short_paths,
92        )
93    }
94}
95
96/// Run a native HTTP CONNECT proxy for onion TCP exits.
97pub async fn run_onion_http_proxy(
98    options: OnionHttpProxyOptions,
99    processor: Arc<Processor>,
100    onion: NativeOnionCircuitHandle,
101) -> Result<()> {
102    options.validate()?;
103    let listener = TcpListener::bind(options.listen_addr)
104        .await
105        .map_err(|error| Error::OnionProxyIoError(format!("bind HTTP proxy listener: {error}")))?;
106    let listen_addr = listener.local_addr().map_err(|error| {
107        Error::OnionProxyIoError(format!("read HTTP proxy listener address: {error}"))
108    })?;
109    println!("Onion HTTP CONNECT proxy endpoint: http://{listen_addr}");
110    let permits = Arc::new(Semaphore::new(options.max_connections));
111
112    loop {
113        let permit = permits
114            .clone()
115            .acquire_owned()
116            .await
117            .map_err(|_| Error::Lock)?;
118        let (stream, peer_addr) = listener.accept().await.map_err(|error| {
119            Error::OnionProxyIoError(format!("accept HTTP proxy connection: {error}"))
120        })?;
121        let processor = processor.clone();
122        let onion = onion.clone();
123        let options = options.clone();
124        tokio::spawn(async move {
125            let _permit = permit;
126            if let Err(error) = handle_connect(stream, processor, onion, options).await {
127                tracing::warn!("onion HTTP proxy request from {peer_addr} failed: {error:?}");
128            }
129        });
130    }
131}
132
133async fn handle_connect(
134    mut stream: TcpStream,
135    processor: Arc<Processor>,
136    onion: NativeOnionCircuitHandle,
137    options: OnionHttpProxyOptions,
138) -> Result<()> {
139    let target = match read_connect_target(&mut stream, options.header_timeout).await {
140        Ok(target) => target,
141        Err(error) => {
142            let _ = write_proxy_response(&mut stream, "400 Bad Request").await;
143            return Err(error);
144        }
145    };
146    let proxy_route = processor
147        .build_onion_proxy_route(options.proxy_config()?, target)
148        .await?;
149    let opened = onion
150        .open_tcp_stream(proxy_route.route, proxy_route.target)
151        .await?;
152    write_proxy_response(&mut stream, "200 Connection Established").await?;
153    opened.relay(stream);
154    Ok(())
155}
156
157async fn read_connect_target(
158    stream: &mut TcpStream,
159    header_timeout: Duration,
160) -> Result<OnionProxyTarget> {
161    let header = timeout(header_timeout, read_http_header(stream))
162        .await
163        .map_err(|_| {
164            Error::HttpRequestError(format!(
165                "HTTP CONNECT header timed out after {} ms",
166                header_timeout.as_millis()
167            ))
168        })??;
169    let header = std::str::from_utf8(&header)
170        .map_err(|_| Error::HttpRequestError("HTTP CONNECT header is not UTF-8".to_string()))?;
171    let request_line = header
172        .lines()
173        .next()
174        .ok_or_else(|| Error::HttpRequestError("missing HTTP request line".to_string()))?;
175    parse_connect_request_line(request_line)
176}
177
178async fn read_http_header(stream: &mut TcpStream) -> Result<Vec<u8>> {
179    let mut header = Vec::new();
180    let mut byte = [0_u8; 1];
181    while header.len() < MAX_CONNECT_HEADER_BYTES {
182        let n = stream.read(byte.as_mut_slice()).await.map_err(|error| {
183            Error::HttpRequestError(format!("read HTTP CONNECT header: {error}"))
184        })?;
185        if n == 0 {
186            return Err(Error::HttpRequestError(
187                "connection closed before HTTP CONNECT header completed".to_string(),
188            ));
189        }
190        header.push(byte[0]);
191        if header.ends_with(b"\r\n\r\n") {
192            return Ok(header);
193        }
194    }
195    Err(Error::HttpRequestError(format!(
196        "HTTP CONNECT header exceeded {MAX_CONNECT_HEADER_BYTES} bytes"
197    )))
198}
199
200fn parse_connect_request_line(request_line: &str) -> Result<OnionProxyTarget> {
201    let mut parts = request_line.split_whitespace();
202    let method = parts
203        .next()
204        .ok_or_else(|| Error::HttpRequestError("missing HTTP method".to_string()))?;
205    let authority = parts
206        .next()
207        .ok_or_else(|| Error::HttpRequestError("missing HTTP CONNECT target".to_string()))?;
208    let version = parts
209        .next()
210        .ok_or_else(|| Error::HttpRequestError("missing HTTP version".to_string()))?;
211
212    if parts.next().is_some() {
213        return Err(Error::HttpRequestError(format!(
214            "invalid HTTP CONNECT request line {request_line:?}"
215        )));
216    }
217    if method != "CONNECT" {
218        return Err(Error::HttpRequestError(format!(
219            "unsupported onion proxy method {method:?}; expected CONNECT"
220        )));
221    }
222    if !version.starts_with("HTTP/") {
223        return Err(Error::HttpRequestError(format!(
224            "invalid HTTP version {version:?}"
225        )));
226    }
227
228    OnionProxyTarget::parse_authority(authority)
229}
230
231async fn write_proxy_response(stream: &mut TcpStream, status: &str) -> Result<()> {
232    let response = format!("HTTP/1.1 {status}\r\n\r\n");
233    stream
234        .write_all(response.as_bytes())
235        .await
236        .map_err(|error| Error::HttpRequestError(format!("write HTTP proxy response: {error}")))
237}
238
239#[cfg(test)]
240mod tests;
241
242#[cfg(test)]
243mod test_http_unit;
rings_node/onion/proxy/http/mod.rs

rings_node/onion/proxy/http/
mod.rs
