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

            
7
use super::{ListenerIsolation, ProxyContext};
8
use anyhow::{Context as _, anyhow};
9
use arti_client::{StreamPrefs, TorAddr};
10
use extend::ext;
11
use futures::{AsyncRead, AsyncWrite, io::BufReader};
12
use http::{Method, StatusCode, response::Builder as ResponseBuilder};
13
use hyper::{Response, server::conn::http1::Builder as ServerBuilder, service::service_fn};
14
use safelog::{Sensitive as Sv, sensitive as sv};
15
use std::sync::Arc;
16
use tor_error::{ErrorKind, ErrorReport as _, HasKind, into_internal, warn_report};
17
use tor_rtcompat::Runtime;
18
use tor_rtcompat::SpawnExt as _;
19
use tracing::{instrument, warn};
20

            
21
use hyper_futures_io::FuturesIoCompat;
22

            
23
#[cfg(feature = "rpc")]
24
use {crate::rpc::conntarget::ConnTarget, tor_rpcbase as rpc};
25

            
26
cfg_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.
37
type 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.)
44
type Body = String;
45

            
46
/// A value used to isolate streams received via HTTP CONNECT.
47
#[derive(Clone, Debug, Eq, PartialEq)]
48
pub(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

            
57
impl 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.
72
mod 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
8
    pub(super) fn uniq_utf8(
111
8
        map: &http::HeaderMap,
112
8
        name: impl http::header::AsHeaderName,
113
8
    ) -> Result<Option<&str>, super::HttpConnectError> {
114
8
        let mut iter = map.get_all(name).iter();
115
8
        let val = match iter.next() {
116
8
            Some(v) => v,
117
            None => return Ok(None),
118
        };
119
8
        match iter.next() {
120
            Some(_) => Err(super::HttpConnectError::DuplicateHeader),
121
8
            None => val
122
8
                .to_str()
123
8
                .map(Some)
124
8
                .map_err(|_| super::HttpConnectError::HeaderNotUtf8),
125
        }
126
8
    }
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")]
136
8
pub(super) async fn handle_http_conn<R, S>(
137
8
    context: super::ProxyContext<R>,
138
8
    stream: BufReader<S>,
139
8
    isolation_info: ListenerIsolation,
140
8
) -> crate::Result<()>
141
8
where
142
8
    R: Runtime,
143
8
    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
144
8
{
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
8
            service_fn(|request| handle_http_request::<R, S>(request, &context, isolation_info)),
157
        )
158
        .with_upgrades()
159
        .await?;
160

            
161
    Ok(())
162
8
}
163

            
164
/// Handle a single HTTP request.
165
///
166
/// This function is invoked by hyper.
167
8
async fn handle_http_request<R, S>(
168
8
    request: Request,
169
8
    context: &ProxyContext<R>,
170
8
    listener_isolation: ListenerIsolation,
171
8
) -> Result<Response<Body>, anyhow::Error>
172
8
where
173
8
    R: Runtime,
174
8
    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
175
8
{
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
8
    if request.method() != Method::CONNECT {
184
8
        match hdr::uniq_utf8(request.headers(), hdr::HOST) {
185
            Err(e) => return Err(e).context("Host header invalid. Rejecting request."),
186
8
            Ok(Some(host)) if !host_is_localhost(host) => {
187
4
                return Err(anyhow!(
188
4
                    "Host header {host:?} was not localhost. Rejecting request."
189
4
                ));
190
            }
191
4
            Ok(_) => {}
192
        }
193
    }
194

            
195
4
    match *request.method() {
196
4
        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
8
}
208

            
209
/// Return an appropriate reply to the given OPTIONS request.
210
4
fn handle_options_request(request: &Request) -> Result<Response<Body>, anyhow::Error> {
211
    use hyper::body::Body as _;
212

            
213
4
    let target = request.uri().to_string();
214
4
    match target.as_str() {
215
4
        "*" => {}
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
4
    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
4
    }
233
4
    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
4
    }
238

            
239
4
    Ok(ResponseBuilder::new()
240
4
        .header("Allow", "OPTIONS, CONNECT")
241
4
        .status(StatusCode::OK)
242
4
        .ok(&Method::OPTIONS)?)
243
4
}
244

            
245
/// Return an appropriate reply to the given CONNECT request.
246
async fn handle_connect_request<R, S>(
247
    request: Request,
248
    context: &ProxyContext<R>,
249
    listener_isolation: ListenerIsolation,
250
) -> anyhow::Result<Response<Body>>
251
where
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.)
265
async fn handle_connect_request_impl<R, S>(
266
    request: Request,
267
    context: &ProxyContext<R>,
268
    listener_isolation: ListenerIsolation,
269
) -> Result<Response<Body>, HttpConnectError>
270
where
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`.
316
fn 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.
344
fn 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)]
368
pub(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

            
375
impl 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")]
423
fn 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"))]
446
fn 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]
459
impl ResponseBuilder {
460
4
    fn ok(self, method: &Method) -> Result<Response<Body>, HttpConnectError> {
461
4
        let bld = add_common_headers(self, method);
462
4
        Ok(bld
463
4
            .body("".into())
464
4
            .map_err(into_internal!("Formatting HTTP response"))?)
465
4
    }
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.
480
4
fn capabilities() -> &'static str {
481
    use std::sync::LazyLock;
482
2
    static CAPS: LazyLock<String> = LazyLock::new(|| {
483
2
        let mut caps = hdr::ALL_REQUEST_HEADERS.to_vec();
484
2
        caps.sort();
485
2
        caps.join(" ")
486
2
    });
487

            
488
4
    CAPS.as_str()
489
4
}
490

            
491
/// Add all common headers to the builder `bld`, and return a new builder.
492
4
fn add_common_headers(mut bld: ResponseBuilder, method: &Method) -> ResponseBuilder {
493
4
    bld = bld.header(hdr::TOR_CAPABILITIES, capabilities());
494
4
    if let (Some(software), Some(version)) = (
495
4
        option_env!("CARGO_PKG_NAME"),
496
4
        option_env!("CARGO_PKG_VERSION"),
497
    ) {
498
4
        if method == Method::CONNECT {
499
            bld = bld.header(
500
                hdr::VIA,
501
                format!("tor/1.0 tor-network ({software} {version})"),
502
            );
503
4
        } else {
504
4
            bld = bld.header(hdr::SERVER, format!("tor/1.0 ({software} {version})"));
505
4
        }
506
    }
507
4
    bld
508
4
}
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)]
513
enum 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

            
554
impl 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

            
571
impl 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
632
fn 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`].
657
fn deconstruct_upgrade<S>(upgraded: hyper::upgrade::Upgraded) -> Result<BufReader<S>, anyhow::Error>
658
where
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`.
678
async fn transfer<S>(request: Request, tor_stream: arti_client::DataStream) -> anyhow::Result<()>
679
where
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.
702
44
fn host_is_localhost(host: &str) -> bool {
703
44
    if let Ok(addr) = host.parse::<std::net::SocketAddr>() {
704
12
        addr.ip().is_loopback()
705
32
    } else if let Ok(ip) = host.parse::<std::net::IpAddr>() {
706
6
        ip.is_loopback()
707
26
    } else if let Some((addr, port)) = host.split_once(':') {
708
14
        port.parse::<std::num::NonZeroU16>().is_ok() && addr.eq_ignore_ascii_case("localhost")
709
    } else {
710
12
        host.eq_ignore_ascii_case("localhost")
711
    }
712
44
}
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.
718
mod 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
16
        fn poll_read(
739
16
            self: Pin<&mut Self>,
740
16
            cx: &mut Context<'_>,
741
16
            mut buf: ReadBufCursor<'_>,
742
16
        ) -> Poll<Result<(), io::Error>> {
743
16
            let mut this = self.project();
744

            
745
16
            let available: &[u8] = ready!(this.0.as_mut().poll_fill_buf(cx))?;
746
8
            let n_available = available.len();
747

            
748
8
            if !available.is_empty() {
749
8
                buf.put_slice(available);
750
8
                this.0.consume(n_available);
751
8
            }
752

            
753
            // This means either "data arrived" or "EOF" depending on whether we added new bytes.
754
8
            Poll::Ready(Ok(()))
755
16
        }
756
    }
757

            
758
    impl<T> hyper::rt::Write for FuturesIoCompat<T>
759
    where
760
        T: futures::io::AsyncWrite,
761
    {
762
4
        fn poll_write(
763
4
            self: Pin<&mut Self>,
764
4
            cx: &mut Context<'_>,
765
4
            buf: &[u8],
766
4
        ) -> Poll<Result<usize, std::io::Error>> {
767
4
            self.project().0.poll_write(cx, buf)
768
4
        }
769

            
770
8
        fn poll_flush(
771
8
            self: Pin<&mut Self>,
772
8
            cx: &mut Context<'_>,
773
8
        ) -> Poll<Result<(), std::io::Error>> {
774
8
            self.project().0.poll_flush(cx)
775
8
        }
776

            
777
4
        fn poll_shutdown(
778
4
            self: Pin<&mut Self>,
779
4
            cx: &mut Context<'_>,
780
4
        ) -> Poll<Result<(), std::io::Error>> {
781
4
            self.project().0.poll_close(cx)
782
4
        }
783
    }
784
}
785

            
786
#[cfg(test)]
787
mod 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
}