arti/proxy/
http_connect.rs

1//! Implement an HTTP1 CONNECT proxy using `hyper`.
2//!
3//! Note that Tor defines several extensions to HTTP CONNECT;
4//! See [the spec](spec.torproject.org/http-connect.html)
5//! for more information.
6
7use super::{ListenerIsolation, ProxyContext};
8use anyhow::{Context as _, anyhow};
9use arti_client::{StreamPrefs, TorAddr};
10use extend::ext;
11use futures::{AsyncRead, AsyncWrite, io::BufReader};
12use http::{Method, StatusCode, response::Builder as ResponseBuilder};
13use hyper::{Response, server::conn::http1::Builder as ServerBuilder, service::service_fn};
14use safelog::{Sensitive as Sv, sensitive as sv};
15use std::sync::Arc;
16use tor_error::{ErrorKind, ErrorReport as _, HasKind, into_internal, warn_report};
17use tor_rtcompat::Runtime;
18use tor_rtcompat::SpawnExt as _;
19use tracing::{instrument, warn};
20
21use hyper_futures_io::FuturesIoCompat;
22
23#[cfg(feature = "rpc")]
24use {crate::rpc::conntarget::ConnTarget, tor_rpcbase as rpc};
25
26cfg_if::cfg_if! {
27    if #[cfg(feature="rpc")] {
28        /// Error type returned from a failed connect_with_prefs.
29        type ClientError = Box<dyn arti_client::rpc::ClientConnectionError>;
30    } else {
31        /// Error type returned from a failed connect_with_prefs.
32        type ClientError = arti_client::Error;
33    }
34}
35
36/// Request type that we receive from Hyper.
37type Request = hyper::Request<hyper::body::Incoming>;
38
39/// We use "String" as our body type, since we only return a body on error,
40/// in which case it already starts life as a formatted string.
41///
42/// (We could use () or `Empty` for our (200 OK) replies,
43/// but empty strings are cheap enough that it isn't worth it.)
44type Body = String;
45
46/// A value used to isolate streams received via HTTP CONNECT.
47#[derive(Clone, Debug, Eq, PartialEq)]
48pub(super) struct Isolation {
49    /// The value of the Proxy-Authorization header.
50    proxy_auth: Option<ProxyAuthorization>,
51    /// The legacy X-Tor-Isolation token.
52    x_tor_isolation: Option<String>,
53    /// The up-to-date Tor-Isolation token.
54    tor_isolation: Option<String>,
55}
56
57impl Isolation {
58    /// Return true if no isolation field in this object is set.
59    pub(super) fn is_empty(&self) -> bool {
60        let Isolation {
61            proxy_auth,
62            x_tor_isolation,
63            tor_isolation,
64        } = self;
65        proxy_auth.as_ref().is_none_or(ProxyAuthorization::is_empty)
66            && x_tor_isolation.as_ref().is_none_or(String::is_empty)
67            && tor_isolation.as_ref().is_none_or(String::is_empty)
68    }
69}
70
71/// Constants and code for the HTTP headers we use.
72mod hdr {
73    pub(super) use http::header::{CONTENT_TYPE, HOST, PROXY_AUTHORIZATION, SERVER, VIA};
74
75    /// Client-to-proxy: Which IP family should we use?
76    pub(super) const TOR_FAMILY_PREFERENCE: &str = "Tor-Family-Preference";
77
78    /// Client-To-Proxy: The ID of an RPC object to receive our request.
79    pub(super) const TOR_RPC_TARGET: &str = "Tor-RPC-Target";
80
81    /// Client-To-Proxy: An isolation token to use with our stream.
82    /// (Legacy name.)
83    pub(super) const X_TOR_STREAM_ISOLATION: &str = "X-Tor-Stream-Isolation";
84
85    /// Client-To-Proxy: An isolation token to use with our stream.
86    pub(super) const TOR_STREAM_ISOLATION: &str = "Tor-Stream-Isolation";
87
88    /// Proxy-to-client: A list of the capabilities that this proxy provides.
89    pub(super) const TOR_CAPABILITIES: &str = "Tor-Capabilities";
90
91    /// Proxy-to-client: A machine-readable list of failure reasons.
92    pub(super) const TOR_REQUEST_FAILED: &str = "Tor-Request-Failed";
93
94    /// A list of all the headers that we support from client-to-proxy.
95    ///
96    /// Does not include headers that we check for HTTP conformance,
97    /// but not for any other purpose.
98    pub(super) const ALL_REQUEST_HEADERS: &[&str] = &[
99        TOR_FAMILY_PREFERENCE,
100        TOR_RPC_TARGET,
101        X_TOR_STREAM_ISOLATION,
102        TOR_STREAM_ISOLATION,
103        // Can't use 'PROXY_AUTHORIZATION', since it isn't a str, and its as_str() isn't const.
104        "Proxy-Authorization",
105    ];
106
107    /// Return the unique string-valued value of the header `name`;
108    /// or None if the header doesn't exist,
109    /// or an error if the header is duplicated or not UTF-8.
110    pub(super) fn uniq_utf8(
111        map: &http::HeaderMap,
112        name: impl http::header::AsHeaderName,
113    ) -> Result<Option<&str>, super::HttpConnectError> {
114        let mut iter = map.get_all(name).iter();
115        let val = match iter.next() {
116            Some(v) => v,
117            None => return Ok(None),
118        };
119        match iter.next() {
120            Some(_) => Err(super::HttpConnectError::DuplicateHeader),
121            None => val
122                .to_str()
123                .map(Some)
124                .map_err(|_| super::HttpConnectError::HeaderNotUtf8),
125        }
126    }
127}
128
129/// Given a just-received TCP connection `S` on a HTTP proxy port, handle the
130/// HTTP handshake and relay the connection over the Tor network.
131///
132/// Uses `isolation_info` to decide which circuits this connection
133/// may use.  Requires that `isolation_info` is a pair listing the listener
134/// id and the source address for the HTTP request.
135#[instrument(skip_all, level = "trace")]
136pub(super) async fn handle_http_conn<R, S>(
137    context: super::ProxyContext<R>,
138    stream: BufReader<S>,
139    isolation_info: ListenerIsolation,
140) -> crate::Result<()>
141where
142    R: Runtime,
143    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
144{
145    // NOTES:
146    // * We _could_ use a timeout, but we trust that the client is not trying to DOS us.
147    ServerBuilder::new()
148        .half_close(false)
149        .keep_alive(true)
150        .max_headers(256)
151        .max_buf_size(16 * 1024)
152        .title_case_headers(true)
153        .auto_date_header(false) // We omit the date header out of general principle.
154        .serve_connection(
155            FuturesIoCompat(stream),
156            service_fn(|request| handle_http_request::<R, S>(request, &context, isolation_info)),
157        )
158        .with_upgrades()
159        .await?;
160
161    Ok(())
162}
163
164/// Handle a single HTTP request.
165///
166/// This function is invoked by hyper.
167async fn handle_http_request<R, S>(
168    request: Request,
169    context: &ProxyContext<R>,
170    listener_isolation: ListenerIsolation,
171) -> Result<Response<Body>, anyhow::Error>
172where
173    R: Runtime,
174    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
175{
176    // Avoid cross-site attacks based on DNS forgery by validating that the Host
177    // header is in fact localhost.  In these cases, we don't want to reply at all,
178    // _even with an error message_, since our headers could be used to tell a hostile
179    // webpage information about the local arti process.
180    //
181    // We don't do this for CONNECT requests, since those are forbidden by
182    // XHR and JS fetch(), and since Host _will_ be non-localhost for those.
183    if request.method() != Method::CONNECT {
184        match hdr::uniq_utf8(request.headers(), hdr::HOST) {
185            Err(e) => return Err(e).context("Host header invalid. Rejecting request."),
186            Ok(Some(host)) if !host_is_localhost(host) => {
187                return Err(anyhow!(
188                    "Host header {host:?} was not localhost. Rejecting request."
189                ));
190            }
191            Ok(_) => {}
192        }
193    }
194
195    match *request.method() {
196        Method::OPTIONS => handle_options_request(&request),
197        Method::CONNECT => {
198            handle_connect_request::<R, S>(request, context, listener_isolation).await
199        }
200        _ => Ok(ResponseBuilder::new()
201            .status(StatusCode::NOT_IMPLEMENTED)
202            .err(
203                request.method(),
204                format!("{} is not supported", request.method()),
205            )?),
206    }
207}
208
209/// Return an appropriate reply to the given OPTIONS request.
210fn handle_options_request(request: &Request) -> Result<Response<Body>, anyhow::Error> {
211    use hyper::body::Body as _;
212
213    let target = request.uri().to_string();
214    match target.as_str() {
215        "*" => {}
216        s if TorAddr::from(s).is_ok() => {}
217        _ => {
218            return Ok(ResponseBuilder::new()
219                .status(StatusCode::BAD_REQUEST)
220                .err(&Method::OPTIONS, "Target was not a valid address")?);
221        }
222    }
223    if request.headers().contains_key(hdr::CONTENT_TYPE) {
224        // RFC 9110 says that if a client wants to include a body with its OPTIONS request (!),
225        // it must include a Content-Type header.  Therefore, we reject such requests.
226        return Ok(ResponseBuilder::new()
227            .status(StatusCode::BAD_REQUEST)
228            .err(&Method::OPTIONS, "Unexpected Content-Type on OPTIONS")?);
229
230        // TODO: It would be cool to detect nonempty bodies in other ways, though in practice
231        // it should never come up.
232    }
233    if !request.body().is_end_stream() {
234        return Ok(ResponseBuilder::new()
235            .status(StatusCode::BAD_REQUEST)
236            .err(&Method::OPTIONS, "Unexpected body on OPTIONS request")?);
237    }
238
239    Ok(ResponseBuilder::new()
240        .header("Allow", "OPTIONS, CONNECT")
241        .status(StatusCode::OK)
242        .ok(&Method::OPTIONS)?)
243}
244
245/// Return an appropriate reply to the given CONNECT request.
246async fn handle_connect_request<R, S>(
247    request: Request,
248    context: &ProxyContext<R>,
249    listener_isolation: ListenerIsolation,
250) -> anyhow::Result<Response<Body>>
251where
252    R: Runtime,
253    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
254{
255    match handle_connect_request_impl::<R, S>(request, context, listener_isolation).await {
256        Ok(response) => Ok(response),
257        Err(e) => Ok(e.try_into_response()?),
258    }
259}
260
261/// Helper for handle_connect_request:
262/// return an error type that can be converted into an HTTP message.
263///
264/// (This is a separate function to make error handling simpler.)
265async fn handle_connect_request_impl<R, S>(
266    request: Request,
267    context: &ProxyContext<R>,
268    listener_isolation: ListenerIsolation,
269) -> Result<Response<Body>, HttpConnectError>
270where
271    R: Runtime,
272    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
273{
274    let target = request.uri().to_string();
275    let tor_addr =
276        TorAddr::from(&target).map_err(|e| HttpConnectError::InvalidStreamTarget(sv(target), e))?;
277
278    let mut stream_prefs = StreamPrefs::default();
279    set_family_preference(&mut stream_prefs, &tor_addr, request.headers())?;
280
281    set_isolation(&mut stream_prefs, request.headers(), listener_isolation)?;
282
283    let client = find_conn_target(
284        context,
285        hdr::uniq_utf8(request.headers(), hdr::TOR_RPC_TARGET)?,
286    )?;
287
288    // If we reach this point, the request looks okay, so we'll try to connect.
289    let tor_stream = client
290        .connect_with_prefs(&tor_addr, &stream_prefs)
291        .await
292        .map_err(|e| HttpConnectError::ConnectFailed(sv(tor_addr), e))?;
293
294    // We have connected.  We need to launch a separate task to actually be the proxy, though,
295    // since IIUC hyper::upgrade::on won't return an answer
296    // until after the response is given to the client.
297    context
298        .tor_client
299        .runtime()
300        .spawn(async move {
301            match transfer::<S>(request, tor_stream).await {
302                Ok(()) => {}
303                Err(e) => {
304                    warn_report!(e, "Error while launching transfer");
305                }
306            }
307        })
308        .map_err(into_internal!("Unable to spawn transfer task"))?;
309
310    ResponseBuilder::new()
311        .status(StatusCode::OK)
312        .ok(&Method::CONNECT)
313}
314
315/// Set the IP family preference in `prefs`.
316fn set_family_preference(
317    prefs: &mut StreamPrefs,
318    addr: &TorAddr,
319    headers: &http::HeaderMap,
320) -> Result<(), HttpConnectError> {
321    if let Some(val) = hdr::uniq_utf8(headers, hdr::TOR_FAMILY_PREFERENCE)? {
322        match val.trim() {
323            "ipv4-preferred" => prefs.ipv4_preferred(),
324            "ipv6-preferred" => prefs.ipv6_preferred(),
325            "ipv4-only" => prefs.ipv4_only(),
326            "ipv6-only" => prefs.ipv6_only(),
327            _ => return Err(HttpConnectError::InvalidFamilyPreference),
328        };
329    } else if let Some(ip) = addr.as_ip_address() {
330        // TODO: Perhaps we should check unconditionally whether the IP address is consistent with header,
331        // if one was given?  On the other hand, if the application tells us to make an IPV6-only
332        // connection to an IPv4 address, it probably deserves what it gets.
333        if ip.is_ipv4() {
334            prefs.ipv4_only();
335        } else {
336            prefs.ipv6_only();
337        }
338    }
339
340    Ok(())
341}
342
343/// Configure the stream isolation from the provided headers.
344fn set_isolation(
345    prefs: &mut StreamPrefs,
346    headers: &http::HeaderMap,
347    listener_isolation: ListenerIsolation,
348) -> Result<(), HttpConnectError> {
349    let proxy_auth =
350        hdr::uniq_utf8(headers, hdr::PROXY_AUTHORIZATION)?.map(ProxyAuthorization::from_header);
351    let x_tor_isolation = hdr::uniq_utf8(headers, hdr::X_TOR_STREAM_ISOLATION)?.map(str::to_owned);
352    let tor_isolation = hdr::uniq_utf8(headers, hdr::TOR_STREAM_ISOLATION)?.map(str::to_owned);
353
354    let isolation = super::ProvidedIsolation::Http(Isolation {
355        proxy_auth,
356        x_tor_isolation,
357        tor_isolation,
358    });
359
360    let isolation = super::StreamIsolationKey(listener_isolation, isolation);
361    prefs.set_isolation(isolation);
362
363    Ok(())
364}
365
366/// An isolation value based on the Proxy-Authorization header.
367#[derive(Debug, Clone, Eq, PartialEq)]
368pub(super) enum ProxyAuthorization {
369    /// The entire contents of the Proxy-Authorization header.
370    Legacy(String),
371    /// The decoded value of the basic authorization, with the user set to "tor-iso".
372    Modern(Vec<u8>),
373}
374
375impl ProxyAuthorization {
376    /// Return a ProxyAuthorization based on the value of the Proxy-Authorization header.
377    ///
378    /// Give a warning if the header is in the legacy (obsolete) format.
379    fn from_header(value: &str) -> Self {
380        if let Some(result) = Self::modern_from_header(value) {
381            result
382        } else {
383            warn!(
384                "{} header in obsolete format. If you want isolation, use {}, \
385                 or {} with Basic authentication and username 'tor-iso'",
386                hdr::PROXY_AUTHORIZATION,
387                hdr::X_TOR_STREAM_ISOLATION,
388                hdr::PROXY_AUTHORIZATION
389            );
390            Self::Legacy(value.to_owned())
391        }
392    }
393
394    /// Helper: Try to return a Modern authorization value, if this is one.
395    fn modern_from_header(value: &str) -> Option<Self> {
396        use base64ct::Encoding as _;
397        let value = value.trim_ascii();
398        let (kind, value) = value.split_once(' ')?;
399        if kind != "Basic" {
400            return None;
401        }
402        let value = value.trim_ascii();
403        // TODO: Is this the right format, or should we allow missing padding?
404        let decoded = base64ct::Base64::decode_vec(value).ok()?;
405        if decoded.starts_with(b"tor-iso:") {
406            Some(ProxyAuthorization::Modern(decoded))
407        } else {
408            None
409        }
410    }
411
412    /// Return true if this ProxyAuthorization has no authorization information.
413    fn is_empty(&self) -> bool {
414        match self {
415            ProxyAuthorization::Legacy(s) => s.is_empty(),
416            ProxyAuthorization::Modern(v) => v.is_empty(),
417        }
418    }
419}
420
421/// Look up the connection target given the value of an Tor-RPC-Target header.
422#[cfg(feature = "rpc")]
423fn find_conn_target<R: Runtime>(
424    context: &ProxyContext<R>,
425    rpc_target: Option<&str>,
426) -> Result<ConnTarget<R>, HttpConnectError> {
427    let Some(target_id) = rpc_target else {
428        return Ok(ConnTarget::Client(Arc::clone(&context.tor_client)));
429    };
430
431    let Some(rpc_mgr) = &context.rpc_mgr else {
432        return Err(HttpConnectError::NoRpcSupport);
433    };
434
435    let (context, object) = rpc_mgr
436        .lookup_object(&rpc::ObjectId::from(target_id))
437        .map_err(|_| HttpConnectError::RpcObjectNotFound)?;
438
439    Ok(ConnTarget::Rpc { object, context })
440}
441
442/// Look up the connection target given the value of an Tor-RPC-Target header
443//
444// (This is the implementation when we have no RPC support.)
445#[cfg(not(feature = "rpc"))]
446fn find_conn_target<R: Runtime>(
447    context: &ProxyContext<R>,
448    rpc_target: Option<&str>,
449) -> Result<Arc<arti_client::TorClient<R>>, HttpConnectError> {
450    if rpc_target.is_some() {
451        Err(HttpConnectError::NoRpcSupport)
452    } else {
453        Ok(context.tor_client.clone())
454    }
455}
456
457/// Extension trait on ResponseBuilder
458#[ext]
459impl ResponseBuilder {
460    fn ok(self, method: &Method) -> Result<Response<Body>, HttpConnectError> {
461        let bld = add_common_headers(self, method);
462        Ok(bld
463            .body("".into())
464            .map_err(into_internal!("Formatting HTTP response"))?)
465    }
466
467    fn err(
468        self,
469        method: &Method,
470        message: impl Into<String>,
471    ) -> Result<Response<Body>, HttpConnectError> {
472        let bld = add_common_headers(self, method).header(hdr::CONTENT_TYPE, "text/plain");
473        Ok(bld
474            .body(message.into())
475            .map_err(into_internal!("Formatting HTTP response"))?)
476    }
477}
478
479/// Return a string representing our capabilities.
480fn capabilities() -> &'static str {
481    use std::sync::LazyLock;
482    static CAPS: LazyLock<String> = LazyLock::new(|| {
483        let mut caps = hdr::ALL_REQUEST_HEADERS.to_vec();
484        caps.sort();
485        caps.join(" ")
486    });
487
488    CAPS.as_str()
489}
490
491/// Add all common headers to the builder `bld`, and return a new builder.
492fn add_common_headers(mut bld: ResponseBuilder, method: &Method) -> ResponseBuilder {
493    bld = bld.header(hdr::TOR_CAPABILITIES, capabilities());
494    if let (Some(software), Some(version)) = (
495        option_env!("CARGO_PKG_NAME"),
496        option_env!("CARGO_PKG_VERSION"),
497    ) {
498        if method == Method::CONNECT {
499            bld = bld.header(
500                hdr::VIA,
501                format!("tor/1.0 tor-network ({software} {version})"),
502            );
503        } else {
504            bld = bld.header(hdr::SERVER, format!("tor/1.0 ({software} {version})"));
505        }
506    }
507    bld
508}
509
510/// An error that occurs during an HTTP CONNECT attempt, which can (usually)
511/// be reported to the client.
512#[derive(Clone, Debug, thiserror::Error)]
513enum HttpConnectError {
514    /// Tried to connect to an invalid stream target.
515    #[error("Invalid target address {0:?}")]
516    InvalidStreamTarget(Sv<String>, #[source] arti_client::TorAddrError),
517
518    /// We found a duplicate HTTP header that we do not allow.
519    ///
520    /// (We only enforce this for the headers that we look at ourselves.)
521    #[error("Duplicate HTTP header found.")]
522    DuplicateHeader,
523
524    /// We tried to found an HTTP header whose value wasn't encode as UTF-8.
525    ///
526    /// (We only enforce this for the headers that we look at ourselves.)
527    #[error("HTTP header value was not in UTF-8")]
528    HeaderNotUtf8,
529
530    /// The Tor-Family-Preference header wasn't as expected.
531    #[error("Unrecognized value for {}", hdr::TOR_FAMILY_PREFERENCE)]
532    InvalidFamilyPreference,
533
534    /// The user asked to use an RPC object, but we don't support RPC.
535    #[error(
536        "Found {} header, but we are running without RPC support",
537        hdr::TOR_RPC_TARGET
538    )]
539    NoRpcSupport,
540
541    /// The user asked to use an RPC object, but we didn't find the one they wanted.
542    #[error("RPC target object not found")]
543    RpcObjectNotFound,
544
545    /// arti_client was unable to connect to a stream target.
546    #[error("Unable to connect to {0}")]
547    ConnectFailed(Sv<TorAddr>, #[source] ClientError),
548
549    /// We encountered an internal error.
550    #[error("Internal error while handling request")]
551    Internal(#[from] tor_error::Bug),
552}
553
554impl HasKind for HttpConnectError {
555    fn kind(&self) -> ErrorKind {
556        use ErrorKind as EK;
557        use HttpConnectError as HCE;
558        match self {
559            HCE::InvalidStreamTarget(_, _)
560            | HCE::DuplicateHeader
561            | HCE::HeaderNotUtf8
562            | HCE::InvalidFamilyPreference
563            | HCE::RpcObjectNotFound => EK::LocalProtocolViolation,
564            HCE::NoRpcSupport => EK::FeatureDisabled,
565            HCE::ConnectFailed(_, e) => e.kind(),
566            HCE::Internal(e) => e.kind(),
567        }
568    }
569}
570
571impl HttpConnectError {
572    /// Return an appropriate HTTP status code for this error.
573    fn status_code(&self) -> StatusCode {
574        use HttpConnectError as HCE; // Not a Joyce reference
575        use StatusCode as SC;
576        if let Some(end_reason) = self.remote_end_reason() {
577            return end_reason_to_http_status(end_reason);
578        }
579        match self {
580            HCE::InvalidStreamTarget(_, _)
581            | HCE::DuplicateHeader
582            | HCE::HeaderNotUtf8
583            | HCE::InvalidFamilyPreference
584            | HCE::RpcObjectNotFound
585            | HCE::NoRpcSupport => SC::BAD_REQUEST,
586            HCE::ConnectFailed(_, e) => e.kind().http_status_code(),
587            HCE::Internal(e) => e.kind().http_status_code(),
588        }
589    }
590
591    /// If possible, return a response that we should give to this error.
592    fn try_into_response(self) -> Result<Response<Body>, HttpConnectError> {
593        let error_kind = self.kind();
594        let end_reason = self.remote_end_reason();
595        let status_code = self.status_code();
596        let mut request_failed = format!("arti/{error_kind:?}");
597        if let Some(end_reason) = end_reason {
598            request_failed.push_str(&format!(" end/{end_reason}"));
599        }
600
601        ResponseBuilder::new()
602            .status(status_code)
603            .header(hdr::TOR_REQUEST_FAILED, request_failed)
604            .err(&Method::CONNECT, self.report().to_string())
605    }
606
607    /// Return the end reason for this error, if this error does in fact represent an END message
608    /// from the remote side of a stream.
609    //
610    // TODO: This function is a bit fragile; it forces us to use APIs from tor-proto and
611    // tor-cell that are not re-exported from arti-client.  It also relies on the fact that
612    // there is a single error type way down in `tor-proto` representing a received END message.
613    fn remote_end_reason(&self) -> Option<tor_cell::relaycell::msg::EndReason> {
614        use tor_proto::Error::EndReceived;
615        if let Some(EndReceived(reason)) = super::extract_proto_err(self) {
616            Some(*reason)
617        } else {
618            None
619        }
620    }
621}
622
623/// Return the appropriate HTTP status code for a remote END reason.
624///
625/// Return `None` if the END reason is unrecognized and we should use the `ErrorKind`
626///
627/// (We  _could_ use the ErrorKind unconditionally,
628/// but the mapping from END reason to ErrorKind is [given in the spec][spec],
629/// so we try to obey it.)
630///
631/// [spec]: https://spec.torproject.org/http-connect.html#error-codes
632fn end_reason_to_http_status(end_reason: tor_cell::relaycell::msg::EndReason) -> StatusCode {
633    use StatusCode as S;
634    use tor_cell::relaycell::msg::EndReason as R;
635    match end_reason {
636        //
637        R::CONNECTREFUSED => S::FORBIDDEN, // 403
638        // 500: Internal server error.
639        R::MISC | R::NOTDIRECTORY => S::INTERNAL_SERVER_ERROR,
640
641        // 502: Bad Gateway.
642        R::DESTROY | R::DONE | R::HIBERNATING | R::INTERNAL | R::RESOURCELIMIT | R::TORPROTOCOL => {
643            S::BAD_GATEWAY
644        }
645        // 503: Service unavailable
646        R::CONNRESET | R::EXITPOLICY | R::NOROUTE | R::RESOLVEFAILED => S::SERVICE_UNAVAILABLE,
647
648        // 504: Gateway timeout.
649        R::TIMEOUT => S::GATEWAY_TIMEOUT,
650
651        // This is possible if the other side sent an unrecognized error code.
652        _ => S::INTERNAL_SERVER_ERROR, // 500
653    }
654}
655
656/// Recover the original stream from a [`hyper::upgrade::Upgraded`].
657fn deconstruct_upgrade<S>(upgraded: hyper::upgrade::Upgraded) -> Result<BufReader<S>, anyhow::Error>
658where
659    S: AsyncRead + AsyncWrite + Unpin + 'static,
660{
661    let parts: hyper::upgrade::Parts<FuturesIoCompat<BufReader<S>>> = upgraded
662        .downcast()
663        .map_err(|_| anyhow!("downcast failed!"))?;
664    let hyper::upgrade::Parts { io, read_buf, .. } = parts;
665    if !read_buf.is_empty() {
666        // TODO Figure out whether this can happen, due to possible race conditions if the client
667        // gets the OK before we check this?.
668        return Err(anyhow!(
669            "Extraneous data on hyper buffer after upgrade to proxy mode"
670        ));
671    }
672    let io: BufReader<S> = io.0;
673    Ok(io)
674}
675
676/// Recover the application stream from `request`, and launch tasks to transfer data between the application and
677/// the `tor_stream`.
678async fn transfer<S>(request: Request, tor_stream: arti_client::DataStream) -> anyhow::Result<()>
679where
680    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
681{
682    let upgraded = hyper::upgrade::on(request)
683        .await
684        .context("Unable to upgrade connection")?;
685    let app_stream: BufReader<S> = deconstruct_upgrade(upgraded)?;
686    let tor_stream = BufReader::with_capacity(super::APP_STREAM_BUF_LEN, tor_stream);
687
688    // Finally. relay traffic between
689    // the application stream and the tor stream, forever.
690    let _ = futures_copy::copy_buf_bidirectional(
691        app_stream,
692        tor_stream,
693        futures_copy::eof::Close,
694        futures_copy::eof::Close,
695    )
696    .await?;
697
698    Ok(())
699}
700
701/// Return true if `host` is a possible value for a Host header addressing localhost.
702fn host_is_localhost(host: &str) -> bool {
703    if let Ok(addr) = host.parse::<std::net::SocketAddr>() {
704        addr.ip().is_loopback()
705    } else if let Ok(ip) = host.parse::<std::net::IpAddr>() {
706        ip.is_loopback()
707    } else if let Some((addr, port)) = host.split_once(':') {
708        port.parse::<std::num::NonZeroU16>().is_ok() && addr.eq_ignore_ascii_case("localhost")
709    } else {
710        host.eq_ignore_ascii_case("localhost")
711    }
712}
713
714/// Helper module: Make `futures` types usable by `hyper`.
715//
716// TODO: We may want to expose this as a separate crate, or move it into tor-async-utils,
717// if we turn out to need it elsewhere.
718mod hyper_futures_io {
719    use pin_project::pin_project;
720    use std::{
721        io,
722        pin::Pin,
723        task::{Context, Poll, ready},
724    };
725
726    use hyper::rt::ReadBufCursor;
727
728    /// A wrapper around an AsyncBufRead + AsyncWrite to implement traits required by hyper.
729    #[derive(Debug)]
730    #[pin_project]
731    pub(super) struct FuturesIoCompat<T>(#[pin] pub(super) T);
732
733    impl<T> hyper::rt::Read for FuturesIoCompat<T>
734    where
735        // We require AsyncBufRead here it is a good match for ReadBufCursor::put_slice.
736        T: futures::io::AsyncBufRead,
737    {
738        fn poll_read(
739            self: Pin<&mut Self>,
740            cx: &mut Context<'_>,
741            mut buf: ReadBufCursor<'_>,
742        ) -> Poll<Result<(), io::Error>> {
743            let mut this = self.project();
744
745            let available: &[u8] = ready!(this.0.as_mut().poll_fill_buf(cx))?;
746            let n_available = available.len();
747
748            if !available.is_empty() {
749                buf.put_slice(available);
750                this.0.consume(n_available);
751            }
752
753            // This means either "data arrived" or "EOF" depending on whether we added new bytes.
754            Poll::Ready(Ok(()))
755        }
756    }
757
758    impl<T> hyper::rt::Write for FuturesIoCompat<T>
759    where
760        T: futures::io::AsyncWrite,
761    {
762        fn poll_write(
763            self: Pin<&mut Self>,
764            cx: &mut Context<'_>,
765            buf: &[u8],
766        ) -> Poll<Result<usize, std::io::Error>> {
767            self.project().0.poll_write(cx, buf)
768        }
769
770        fn poll_flush(
771            self: Pin<&mut Self>,
772            cx: &mut Context<'_>,
773        ) -> Poll<Result<(), std::io::Error>> {
774            self.project().0.poll_flush(cx)
775        }
776
777        fn poll_shutdown(
778            self: Pin<&mut Self>,
779            cx: &mut Context<'_>,
780        ) -> Poll<Result<(), std::io::Error>> {
781            self.project().0.poll_close(cx)
782        }
783    }
784}
785
786#[cfg(test)]
787mod test {
788    // @@ begin test lint list maintained by maint/add_warning @@
789    #![allow(clippy::bool_assert_comparison)]
790    #![allow(clippy::clone_on_copy)]
791    #![allow(clippy::dbg_macro)]
792    #![allow(clippy::mixed_attributes_style)]
793    #![allow(clippy::print_stderr)]
794    #![allow(clippy::print_stdout)]
795    #![allow(clippy::single_char_pattern)]
796    #![allow(clippy::unwrap_used)]
797    #![allow(clippy::unchecked_time_subtraction)]
798    #![allow(clippy::useless_vec)]
799    #![allow(clippy::needless_pass_by_value)]
800    #![allow(clippy::string_slice)] // See arti#2571
801    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
802
803    use arti_client::{BootstrapBehavior, TorClient, config::TorClientConfigBuilder};
804    use futures::{AsyncReadExt as _, AsyncWriteExt as _};
805    use tor_rtmock::{MockRuntime, io::stream_pair};
806
807    use super::*;
808
809    // Make sure that HeaderMap is case-insensitive as the documentation implies.
810    #[test]
811    fn headermap_casei() {
812        use http::header::{HeaderMap, HeaderValue};
813        let mut hm = HeaderMap::new();
814        hm.append(
815            "my-head-is-a-house-for",
816            HeaderValue::from_str("a-secret").unwrap(),
817        );
818        assert_eq!(
819            hm.get("My-Head-Is-A-House-For").unwrap().as_bytes(),
820            b"a-secret"
821        );
822        assert_eq!(
823            hm.get("MY-HEAD-IS-A-HOUSE-FOR").unwrap().as_bytes(),
824            b"a-secret"
825        );
826    }
827
828    #[test]
829    fn host_header_localhost() {
830        assert_eq!(host_is_localhost("localhost"), true);
831        assert_eq!(host_is_localhost("localhost:9999"), true);
832        assert_eq!(host_is_localhost("localHOSt:9999"), true);
833        assert_eq!(host_is_localhost("127.0.0.1:9999"), true);
834        assert_eq!(host_is_localhost("[::1]:9999"), true);
835        assert_eq!(host_is_localhost("127.1.2.3:1234"), true);
836        assert_eq!(host_is_localhost("127.0.0.1"), true);
837        assert_eq!(host_is_localhost("::1"), true);
838
839        assert_eq!(host_is_localhost("[::1]"), false); // not in the right format!
840        assert_eq!(host_is_localhost("www.torproject.org"), false);
841        assert_eq!(host_is_localhost("www.torproject.org:1234"), false);
842        assert_eq!(host_is_localhost("localhost:0"), false);
843        assert_eq!(host_is_localhost("localhost:999999"), false);
844        assert_eq!(host_is_localhost("plocalhost:1234"), false);
845        assert_eq!(host_is_localhost("[::0]:1234"), false);
846        assert_eq!(host_is_localhost("192.0.2.55:1234"), false);
847        assert_eq!(host_is_localhost("3fff::1"), false);
848        assert_eq!(host_is_localhost("[3fff::1]:1234"), false);
849    }
850
851    fn interactive_test_setup(
852        rt: &MockRuntime,
853    ) -> anyhow::Result<(
854        tor_rtmock::io::LocalStream,
855        impl Future<Output = anyhow::Result<()>>,
856        tempfile::TempDir,
857    )> {
858        let (s1, s2) = stream_pair();
859        let s1: BufReader<_> = BufReader::new(s1);
860
861        let iso: ListenerIsolation = (7, "127.0.0.1".parse().unwrap());
862        let dir = tempfile::TempDir::new().unwrap();
863        let cfg = TorClientConfigBuilder::from_directories(
864            dir.as_ref().join("state"),
865            dir.as_ref().join("cache"),
866        )
867        .build()
868        .unwrap();
869        let tor_client = TorClient::with_runtime(rt.clone())
870            .config(cfg)
871            .bootstrap_behavior(BootstrapBehavior::Manual)
872            .create_unbootstrapped()?;
873        let context: ProxyContext<_> = ProxyContext {
874            tor_client,
875            #[cfg(feature = "rpc")]
876            rpc_mgr: None,
877            protocols: crate::proxy::ListenProtocols::SocksAndHttpConnect,
878        };
879        let handle = rt.spawn_join("HTTP Handler", handle_http_conn(context, s1, iso));
880        Ok((s2, handle, dir))
881    }
882
883    #[test]
884    fn successful_options_test() -> anyhow::Result<()> {
885        // Try an OPTIONS request and make sure we get a plausible-looking answer.
886        //
887        // (This test is mostly here to make sure that invalid_host_test() isn't failing because
888        // of anything besides the Host header.)
889        MockRuntime::try_test_with_various(async |rt| -> anyhow::Result<()> {
890            let (mut s, join, _dir) = interactive_test_setup(&rt)?;
891
892            s.write_all(b"OPTIONS * HTTP/1.0\r\nHost: localhost\r\n\r\n")
893                .await?;
894            let mut buf = Vec::new();
895            let _n_read = s.read_to_end(&mut buf).await?;
896            let () = join.await?;
897
898            let reply = std::str::from_utf8(&buf)?;
899            assert!(dbg!(reply).starts_with("HTTP/1.0 200 OK\r\n"));
900
901            Ok(())
902        })
903    }
904
905    #[test]
906    fn invalid_host_test() -> anyhow::Result<()> {
907        // Try a hostname that looks like a CSRF attempt and make sure that we discard it without
908        // any reply.
909        MockRuntime::try_test_with_various(async |rt| -> anyhow::Result<()> {
910            let (mut s, join, _dir) = interactive_test_setup(&rt)?;
911
912            s.write_all(b"OPTIONS * HTTP/1.0\r\nHost: csrf.example.com\r\n\r\n")
913                .await?;
914            let mut buf = Vec::new();
915            let n_read = s.read_to_end(&mut buf).await?;
916            let http_outcome = join.await;
917
918            assert_eq!(n_read, 0);
919            assert!(buf.is_empty());
920            assert!(http_outcome.is_err());
921
922            let error_msg = http_outcome.unwrap_err().source().unwrap().to_string();
923            assert_eq!(
924                error_msg,
925                r#"Host header "csrf.example.com" was not localhost. Rejecting request."#
926            );
927
928            Ok(())
929        })
930    }
931}
arti/proxy/http_connect.rs

arti/proxy/
http_connect.rs
