1
#![cfg_attr(docsrs, feature(doc_cfg))]
2
#![doc = include_str!("../README.md")]
3
// @@ begin lint list maintained by maint/add_warning @@
4
#![allow(renamed_and_removed_lints)] // @@REMOVE_WHEN(ci_arti_stable)
5
#![allow(unknown_lints)] // @@REMOVE_WHEN(ci_arti_nightly)
6
#![warn(missing_docs)]
7
#![warn(noop_method_call)]
8
#![warn(unreachable_pub)]
9
#![warn(clippy::all)]
10
#![deny(clippy::await_holding_lock)]
11
#![deny(clippy::cargo_common_metadata)]
12
#![deny(clippy::cast_lossless)]
13
#![deny(clippy::checked_conversions)]
14
#![allow(clippy::cognitive_complexity)] // See arti#2556
15
#![deny(clippy::debug_assert_with_mut_call)]
16
#![deny(clippy::exhaustive_enums)]
17
#![deny(clippy::exhaustive_structs)]
18
#![deny(clippy::expl_impl_clone_on_copy)]
19
#![deny(clippy::fallible_impl_from)]
20
#![deny(clippy::implicit_clone)]
21
#![deny(clippy::large_stack_arrays)]
22
#![warn(clippy::manual_ok_or)]
23
#![deny(clippy::missing_docs_in_private_items)]
24
#![warn(clippy::needless_borrow)]
25
#![warn(clippy::needless_pass_by_value)]
26
#![warn(clippy::option_option)]
27
#![warn(clippy::or_fun_call)]
28
#![deny(clippy::print_stderr)]
29
#![deny(clippy::print_stdout)]
30
#![warn(clippy::rc_buffer)]
31
#![deny(clippy::ref_option_ref)]
32
#![warn(clippy::semicolon_if_nothing_returned)]
33
#![warn(clippy::trait_duplication_in_bounds)]
34
#![deny(clippy::unchecked_time_subtraction)]
35
#![deny(clippy::unnecessary_wraps)]
36
#![warn(clippy::unseparated_literal_suffix)]
37
#![deny(clippy::unwrap_used)]
38
#![deny(clippy::mod_module_files)]
39
#![allow(clippy::let_unit_value)] // This can reasonably be done for explicitness
40
#![allow(clippy::uninlined_format_args)]
41
#![allow(clippy::significant_drop_in_scrutinee)] // arti/-/merge_requests/588/#note_2812945
42
#![allow(clippy::result_large_err)] // temporary workaround for arti#587
43
#![allow(clippy::needless_raw_string_hashes)] // complained-about code is fine, often best
44
#![allow(clippy::needless_lifetimes)] // See arti#1765
45
#![allow(mismatched_lifetime_syntaxes)] // temporary workaround for arti#2060
46
#![allow(clippy::collapsible_if)] // See arti#2342
47
#![deny(clippy::unused_async)]
48
#![deny(clippy::string_slice)] // See arti#2571
49
#![allow(recursion_depth_exceeding_limit)] // arti#2715, rust/issues/159228
50
//! <!-- @@ end lint list maintained by maint/add_warning @@ -->
51

            
52
// TODO probably remove this at some point - see tpo/core/arti#1060
53
#![cfg_attr(
54
    not(all(feature = "full", feature = "experimental")),
55
    allow(unused_imports)
56
)]
57

            
58
mod body;
59
mod err;
60
pub mod request;
61
mod response;
62
mod util;
63

            
64
use tor_circmgr::{CircMgr, DirInfo};
65
use tor_error::bad_api_usage;
66
use tor_rtcompat::{Runtime, SleepProvider, SleepProviderExt};
67

            
68
// Zlib is required; the others are optional.
69
#[cfg(feature = "xz")]
70
use async_compression::futures::bufread::XzDecoder;
71
use async_compression::futures::bufread::ZlibDecoder;
72
#[cfg(feature = "zstd")]
73
use async_compression::futures::bufread::ZstdDecoder;
74

            
75
use futures::FutureExt;
76
use futures::io::{
77
    AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
78
};
79
use std::sync::Arc;
80
use std::time::Duration;
81
use tracing::{info, instrument};
82

            
83
pub use err::{Error, RequestError, RequestFailedError};
84
pub use response::{DirResponse, SourceInfo};
85

            
86
/// Type for results returned in this crate.
87
pub type Result<T> = std::result::Result<T, Error>;
88

            
89
/// Type for internal results  containing a RequestError.
90
pub type RequestResult<T> = std::result::Result<T, RequestError>;
91

            
92
/// Flag to declare whether a request is always anonymized or not.
93
///
94
/// This is used by tor-dirclient to control whether *other* deanonymizing metadata
95
/// might be added to the request (eg in request headers):
96
/// Some requests (like those to download onion service descriptors) are always
97
/// anonymized, and should never be sent in a way that leaks information about
98
/// our settings or configuration.
99
///
100
/// It is up to the *caller* of `tor-dirclient` to ensure that
101
///
102
///   - every request whose anonymization status is `AnonymizedRequest::Direct`
103
///     is sent only over non-anonymous connections.
104
///
105
///     (Sending an `AnonymizedRequest::Direct` request over an anonymized connection
106
///     would weaken the connection's anonymity, and can therefore weaken the anonymity
107
///     of user traffic sharing the same circuit.)
108
///
109
///   - every request whose anonymization status is `AnonymizedRequest::Anonymized`
110
///     is sent over only anonymous connections (ie, multi-hop circuits).
111
///
112
///     (Sending an `AnonymizedRequest::Anonymized` request over a direct connection
113
///     would directly reveal user behaviour data to the directory server.)
114
///
115
/// TODO the calling code cannot easily be sure to get this right this because
116
/// the anonymization status is a run-time property and the choice of connection kind
117
/// is statically defined in the calling code.  (Perhaps this could be checked in tests?)
118
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
119
#[non_exhaustive]
120
pub enum AnonymizedRequest {
121
    /// This request's content or semantics reveals or is correlated with sensitive information.
122
    ///
123
    /// For example, requests for hidden service descriptors reveal which hidden services
124
    /// the client is connecting to.
125
    ///
126
    /// The request must be sent over an anonymous circuit by the caller
127
    /// and no additional deanonymizing information should be added to it by `tor-dirclient`.
128
    /// (For example, no client-version-specific information should be
129
    /// sent in HTTP headers when the request is made.)
130
    Anonymized,
131

            
132
    /// Making this request does not reveal anything sensitive, nor any user behaviour.
133
    ///
134
    /// The request body is uncorrelated with such things as the websites the user might visit,
135
    /// the onion services the user is visiting or running, etc.
136
    ///
137
    /// For example, requests for all router microdescriptors are made by all clients,
138
    /// so which microdescriptor(s) are requested reveals nothing to any attacker.
139
    ///
140
    /// tor-dirclient is allowed to add include information about our capabilities
141
    /// when sending this request.
142
    /// The request must *not* be sent over an anonymous circuit by the caller
143
    /// (at least, not one used for anything else).
144
    Direct,
145
}
146

            
147
/// Fetch the resource described by `req` over the Tor network.
148
///
149
/// Circuits are built or found using `circ_mgr`, using paths
150
/// constructed using `dirinfo`.
151
///
152
/// For more fine-grained control over the circuit and stream used,
153
/// construct them yourself, and then call [`send_request`] instead.
154
///
155
/// # TODO
156
///
157
/// This is the only function in this crate that knows about CircMgr and
158
/// DirInfo.  Perhaps this function should move up a level into DirMgr?
159
#[instrument(level = "trace", skip_all)]
160
pub async fn get_resource<CR, R, SP>(
161
    req: &CR,
162
    dirinfo: DirInfo<'_>,
163
    runtime: &SP,
164
    circ_mgr: Arc<CircMgr<R>>,
165
) -> Result<DirResponse>
166
where
167
    CR: request::Requestable + ?Sized,
168
    R: Runtime,
169
    SP: SleepProvider,
170
{
171
    let tunnel = circ_mgr.get_or_launch_dir(dirinfo).await?;
172

            
173
    if req.anonymized() == AnonymizedRequest::Anonymized {
174
        return Err(bad_api_usage!("Tried to use get_resource for an anonymized request").into());
175
    }
176

            
177
    // TODO(nickm) This should be an option, and is too long.
178
    let begin_timeout = Duration::from_secs(5);
179
    let source = match SourceInfo::from_tunnel(&tunnel) {
180
        Ok(source) => source,
181
        Err(e) => {
182
            return Err(Error::RequestFailed(RequestFailedError {
183
                source: None,
184
                error: e.into(),
185
            }));
186
        }
187
    };
188

            
189
    let wrap_err = |error| {
190
        Error::RequestFailed(RequestFailedError {
191
            source: source.clone(),
192
            error,
193
        })
194
    };
195

            
196
    req.check_circuit_skip_optimization(&tunnel)
197
        .await
198
        .map_err(wrap_err)?;
199

            
200
    // Launch the stream.
201
    let mut stream = runtime
202
        .timeout(begin_timeout, tunnel.begin_dir_stream())
203
        .await
204
        .map_err(RequestError::from)
205
        .map_err(wrap_err)?
206
        .map_err(RequestError::from)
207
        .map_err(wrap_err)?; // TODO(nickm) handle fatalities here too
208

            
209
    // TODO: Perhaps we want separate timeouts for each phase of this.
210
    // For now, we just use higher-level timeouts in `dirmgr`.
211
    let r = send_request(runtime, req, &mut stream, source.clone()).await;
212

            
213
    if should_retire_circ(&r) {
214
        retire_circ(&circ_mgr, &tunnel.unique_id(), "Partial response");
215
    }
216

            
217
    r
218
}
219

            
220
/// Return true if `result` holds an error indicating that we should retire the
221
/// circuit used for the corresponding request.
222
fn should_retire_circ(result: &Result<DirResponse>) -> bool {
223
    match result {
224
        Err(e) => e.should_retire_circ(),
225
        Ok(dr) => dr.error().map(RequestError::should_retire_circ) == Some(true),
226
    }
227
}
228

            
229
/// Fetch a Tor directory object from a provided stream.
230
#[deprecated(since = "0.8.1", note = "Use send_request instead.")]
231
pub async fn download<R, S, SP>(
232
    runtime: &SP,
233
    req: &R,
234
    stream: &mut S,
235
    source: Option<SourceInfo>,
236
) -> Result<DirResponse>
237
where
238
    R: request::Requestable + ?Sized,
239
    S: AsyncRead + AsyncWrite + Send + Unpin,
240
    SP: SleepProvider,
241
{
242
    send_request(runtime, req, stream, source).await
243
}
244

            
245
/// Fetch or upload a Tor directory object using the provided stream.
246
///
247
/// To do this, we send a simple HTTP/1.0 request for the described
248
/// object in `req` over `stream`, and then wait for a response.  In
249
/// log messages, we describe the origin of the data as coming from
250
/// `source`.
251
///
252
/// # Notes
253
///
254
/// It's kind of bogus to have a 'source' field here at all; we may
255
/// eventually want to remove it.
256
///
257
/// This function doesn't close the stream; you may want to do that
258
/// yourself.
259
///
260
/// The only error variant returned is [`Error::RequestFailed`].
261
// TODO: should the error return type change to `RequestFailedError`?
262
// If so, that would simplify some code in_dirmgr::bridgedesc.
263
105
pub async fn send_request<R, S, SP>(
264
105
    runtime: &SP,
265
105
    req: &R,
266
105
    stream: &mut S,
267
105
    source: Option<SourceInfo>,
268
105
) -> Result<DirResponse>
269
105
where
270
105
    R: request::Requestable + ?Sized,
271
105
    S: AsyncRead + AsyncWrite + Send + Unpin,
272
105
    SP: SleepProvider,
273
105
{
274
105
    let wrap_err = |error| {
275
20
        Error::RequestFailed(RequestFailedError {
276
20
            source: source.clone(),
277
20
            error,
278
20
        })
279
20
    };
280

            
281
105
    let partial_ok = req.partial_response_body_ok();
282
105
    let maxlen = req.max_response_len();
283
105
    let anonymized = req.anonymized();
284
105
    let req = req.make_request().map_err(wrap_err)?;
285
105
    let method = req.method().clone();
286
105
    let encoded = util::encode_request(&req);
287

            
288
    // Write the request.
289
193
    for chunk in encoded.iter() {
290
193
        stream
291
193
            .write_all(chunk)
292
193
            .await
293
193
            .map_err(RequestError::from)
294
193
            .map_err(wrap_err)?;
295
    }
296
105
    stream
297
105
        .flush()
298
105
        .await
299
105
        .map_err(RequestError::from)
300
105
        .map_err(wrap_err)?;
301

            
302
105
    let mut buffered = BufReader::new(stream);
303

            
304
    // Handle the response
305
105
    let header = read_headers(runtime, &mut buffered)
306
105
        .await
307
105
        .map_err(wrap_err)?;
308
86
    if header.status != Some(200) {
309
17
        return Ok(DirResponse::new(
310
17
            method,
311
17
            header.status.unwrap_or(0),
312
17
            header.status_message,
313
17
            None,
314
17
            vec![],
315
17
            source,
316
17
        ));
317
69
    }
318

            
319
69
    let mut decoder =
320
69
        get_decoder(buffered, header.encoding.as_deref(), anonymized).map_err(wrap_err)?;
321

            
322
69
    let mut result = Vec::new();
323
69
    let ok = read_and_decompress(runtime, &mut decoder, maxlen, &mut result).await;
324

            
325
69
    let ok = match (partial_ok, ok, result.len()) {
326
1
        (true, Err(e), n) if n > 0 => {
327
            // Note that we _don't_ return here: we want the partial response.
328
1
            Err(e)
329
        }
330
1
        (_, Err(e), _) => {
331
1
            return Err(wrap_err(e));
332
        }
333
67
        (_, Ok(()), _) => Ok(()),
334
    };
335

            
336
68
    Ok(DirResponse::new(
337
68
        method,
338
68
        200,
339
68
        None,
340
68
        ok.err(),
341
68
        result,
342
68
        source,
343
68
    ))
344
105
}
345

            
346
/// Maximum length for the HTTP headers in a single request or response.
347
///
348
/// Chosen more or less arbitrarily.
349
const MAX_HEADERS_LEN: usize = 16384;
350

            
351
/// Read and parse HTTP/1 headers from `stream`.
352
///
353
/// Wraps around [`read_headers_internal()`] with a timeout.
354
109
async fn read_headers<S, SP>(runtime: &SP, stream: &mut S) -> RequestResult<HeaderStatus>
355
109
where
356
109
    S: AsyncBufRead + Unpin,
357
109
    SP: SleepProvider,
358
109
{
359
    // Just like with the other timeouts in this code, this should probably be
360
    // configurable and may even be too much.
361
109
    let read_timeout = Duration::from_secs(10);
362
109
    runtime
363
109
        .timeout(read_timeout, read_headers_internal(stream))
364
109
        .await
365
109
        .map_err(|_| RequestError::DirTimeout)
366
109
        .flatten()
367
109
}
368

            
369
/// Internal version of [`read_headers()`] without a timeout.
370
109
async fn read_headers_internal<S>(stream: &mut S) -> RequestResult<HeaderStatus>
371
109
where
372
109
    S: AsyncBufRead + Unpin,
373
109
{
374
109
    let mut buf = Vec::with_capacity(1024);
375

            
376
    // Note: Using `take` here will ensure that we will get an EOF if the header length
377
    // would exceed the maximum.  No extra bytes will actually be extracted.
378
109
    let mut limited_stream = stream.take(MAX_HEADERS_LEN as u64);
379

            
380
    loop {
381
        // TODO: it's inefficient to do this a line at a time; it would
382
        // probably be better to read until the CRLF CRLF ending of the
383
        // response.  But this should be fast enough.
384
208
        let n = limited_stream.read_until(b'\n', &mut buf).await?;
385

            
386
        // TODO(nickm): Better maximum and/or let this expand.
387
192
        let mut headers = [httparse::EMPTY_HEADER; 32];
388
192
        let mut response = httparse::Response::new(&mut headers);
389

            
390
192
        match response.parse(&buf[..])? {
391
            httparse::Status::Partial => {
392
                // We didn't get a whole response; we may need to try again.
393

            
394
103
                if n == 0 {
395
                    // We hit an EOF; no more progress can be made.
396
3
                    return Err(RequestError::TruncatedHeaders);
397
100
                }
398

            
399
100
                if buf.len() >= MAX_HEADERS_LEN {
400
1
                    return Err(RequestError::HeadersTooLong(buf.len()));
401
99
                }
402
            }
403
88
            httparse::Status::Complete(n_parsed) => {
404
88
                if response.code != Some(200) {
405
18
                    return Ok(HeaderStatus {
406
18
                        status: response.code,
407
18
                        status_message: response.reason.map(str::to_owned),
408
18
                        encoding: None,
409
18
                    });
410
70
                }
411
70
                let encoding = if let Some(enc) = response
412
70
                    .headers
413
70
                    .iter()
414
70
                    .find(|h| h.name == "Content-Encoding")
415
                {
416
5
                    Some(String::from_utf8(enc.value.to_vec())?)
417
                } else {
418
65
                    None
419
                };
420
                /*
421
                if let Some(clen) = response.headers.iter().find(|h| h.name == "Content-Length") {
422
                    let clen = std::str::from_utf8(clen.value)?;
423
                    length = Some(clen.parse()?);
424
                }
425
                 */
426
70
                assert!(n_parsed == buf.len());
427
70
                return Ok(HeaderStatus {
428
70
                    status: Some(200),
429
70
                    status_message: None,
430
70
                    encoding,
431
70
                });
432
            }
433
        }
434
99
        if n == 0 {
435
            return Err(RequestError::TruncatedHeaders);
436
99
        }
437
    }
438
109
}
439

            
440
/// Return value from read_headers
441
#[derive(Debug, Clone)]
442
struct HeaderStatus {
443
    /// HTTP status code.
444
    status: Option<u16>,
445
    /// HTTP status message associated with the status code.
446
    status_message: Option<String>,
447
    /// The Content-Encoding header, if any.
448
    encoding: Option<String>,
449
}
450

            
451
/// Helper: download directory information from `stream` and
452
/// decompress it into a result buffer.  Assumes that `buf` is empty.
453
///
454
/// If we get more than maxlen bytes after decompression, give an error.
455
///
456
/// Returns the status of our download attempt, stores any data that
457
/// we were able to download into `result`.  Existing contents of
458
/// `result` are overwritten.
459
77
async fn read_and_decompress<S, SP>(
460
77
    runtime: &SP,
461
77
    mut stream: S,
462
77
    maxlen: usize,
463
77
    result: &mut Vec<u8>,
464
77
) -> RequestResult<()>
465
77
where
466
77
    S: AsyncRead + Unpin,
467
77
    SP: SleepProvider,
468
77
{
469
77
    let buffer_window_size = 1024;
470
77
    let mut written_total: usize = 0;
471
    // TODO(nickm): This should be an option, and is maybe too long.
472
    // Though for some users it may be too short?
473
77
    let read_timeout = Duration::from_secs(10);
474
77
    let timer = runtime.sleep(read_timeout).fuse();
475
77
    futures::pin_mut!(timer);
476

            
477
    loop {
478
        // allocate buffer for next read
479
416
        result.resize(written_total + buffer_window_size, 0);
480
416
        let buf: &mut [u8] = &mut result[written_total..written_total + buffer_window_size];
481

            
482
416
        let status = futures::select! {
483
416
            status = stream.read(buf).fuse() => status,
484
            _ = timer => {
485
                result.resize(written_total, 0); // truncate as needed
486
                return Err(RequestError::DirTimeout);
487
            }
488
        };
489
416
        let written_in_this_loop = match status {
490
412
            Ok(n) => n,
491
4
            Err(other) => {
492
4
                result.resize(written_total, 0); // truncate as needed
493
4
                return Err(other.into());
494
            }
495
        };
496

            
497
412
        written_total += written_in_this_loop;
498

            
499
        // exit conditions below
500

            
501
412
        if written_in_this_loop == 0 {
502
            /*
503
            in case we read less than `buffer_window_size` in last `read`
504
            we need to shrink result because otherwise we'll return those
505
            un-read 0s
506
            */
507
72
            if written_total < result.len() {
508
72
                result.resize(written_total, 0);
509
72
            }
510
72
            return Ok(());
511
340
        }
512

            
513
        // TODO: It would be good to detect compression bombs, but
514
        // that would require access to the internal stream, which
515
        // would in turn require some tricky programming.  For now, we
516
        // use the maximum length here to prevent an attacker from
517
        // filling our RAM.
518
340
        if written_total > maxlen {
519
1
            result.resize(maxlen, 0);
520
1
            return Err(RequestError::ResponseTooLong(written_total));
521
339
        }
522
    }
523
77
}
524

            
525
/// Retire a directory circuit because of an error we've encountered on it.
526
fn retire_circ<R>(circ_mgr: &Arc<CircMgr<R>>, id: &tor_proto::circuit::UniqId, error: &str)
527
where
528
    R: Runtime,
529
{
530
    info!(
531
        "{}: Retiring circuit because of directory failure: {}",
532
        &id, &error
533
    );
534
    circ_mgr.retire_circ(id);
535
}
536

            
537
/// Memory limit of the LZMA dictionary we are willing to allocate.
538
///
539
/// Without it, it may lead to an allocation of 4GiB, which is a bit overkill.
540
///
541
/// Right now, we use 16 MiB, as the spec limits the compression quality to
542
/// "6", meaning 9 MiB of decompression memory, which rounds up to 16 MiB in
543
/// terms of base 2.
544
///
545
/// From the xz(1) manual page:
546
/// ```text
547
/// Preset   DictSize   CompCPU   CompMem   DecMem
548
///  -0e     256 KiB       8        4 MiB    1 MiB
549
///  -1e       1 MiB       8       13 MiB    2 MiB
550
///  -2e       2 MiB       8       25 MiB    3 MiB
551
///  -3e       4 MiB       7       48 MiB    5 MiB
552
///  -4e       4 MiB       8       48 MiB    5 MiB
553
///  -5e       8 MiB       7       94 MiB    9 MiB
554
///  -6e       8 MiB       8       94 MiB    9 MiB
555
///  -7e      16 MiB       8      186 MiB   17 MiB
556
///  -8e      32 MiB       8      370 MiB   33 MiB
557
///  -9e      64 MiB       8      674 MiB   65 MiB
558
/// ```
559
const LZMA_DICT_MEM_LIMIT: u64 = 1 << 24; // 16 MiB
560

            
561
/// Helper: Return a boxed decoder object that wraps the stream  $s.
562
macro_rules! decoder {
563
    (XzDecoder, $s:expr) => {{
564
        let mut decoder = XzDecoder::with_mem_limit($s, LZMA_DICT_MEM_LIMIT);
565
        decoder.multiple_members(true);
566
        Ok(Box::new(decoder))
567
    }};
568
    ($dec:ident, $s:expr) => {{
569
        let mut decoder = $dec::new($s);
570
        decoder.multiple_members(true);
571
        Ok(Box::new(decoder))
572
    }};
573
}
574

            
575
/// Wrap `stream` in an appropriate type to undo the content encoding
576
/// as described in `encoding`.
577
78
fn get_decoder<'a, S: AsyncBufRead + Unpin + Send + 'a>(
578
78
    stream: S,
579
78
    encoding: Option<&str>,
580
78
    anonymized: AnonymizedRequest,
581
78
) -> RequestResult<Box<dyn AsyncRead + Unpin + Send + 'a>> {
582
    use AnonymizedRequest::Direct;
583
78
    match (encoding, anonymized) {
584
75
        (None | Some("identity"), _) => Ok(Box::new(stream)),
585
8
        (Some("deflate"), _) => decoder!(ZlibDecoder, stream),
586
        // We only admit to supporting these on a direct connection; otherwise,
587
        // a hostile directory could send them back even though we hadn't
588
        // requested them.
589
        #[cfg(feature = "xz")]
590
4
        (Some("x-tor-lzma"), Direct) => decoder!(XzDecoder, stream),
591
        #[cfg(feature = "zstd")]
592
2
        (Some("x-zstd"), Direct) => decoder!(ZstdDecoder, stream),
593
1
        (Some(other), _) => Err(RequestError::ContentEncoding(other.into())),
594
    }
595
78
}
596

            
597
#[cfg(test)]
598
mod test {
599
    // @@ begin test lint list maintained by maint/add_warning @@
600
    #![allow(clippy::bool_assert_comparison)]
601
    #![allow(clippy::clone_on_copy)]
602
    #![allow(clippy::dbg_macro)]
603
    #![allow(clippy::mixed_attributes_style)]
604
    #![allow(clippy::print_stderr)]
605
    #![allow(clippy::print_stdout)]
606
    #![allow(clippy::single_char_pattern)]
607
    #![allow(clippy::unwrap_used)]
608
    #![allow(clippy::unchecked_time_subtraction)]
609
    #![allow(clippy::useless_vec)]
610
    #![allow(clippy::needless_pass_by_value)]
611
    #![allow(clippy::string_slice)] // See arti#2571
612
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
613
    use super::*;
614
    use tor_rtmock::io::stream_pair;
615

            
616
    use tor_rtmock::simple_time::SimpleMockTimeProvider;
617
    use web_time_compat::{SystemTime, SystemTimeExt};
618

            
619
    use futures_await_test::async_test;
620

            
621
    // Basic decompression wrapper.
622
    async fn decomp_basic(
623
        encoding: Option<&str>,
624
        data: &[u8],
625
        maxlen: usize,
626
    ) -> (RequestResult<()>, Vec<u8>) {
627
        // We don't need to do anything fancy here, since we aren't simulating
628
        // a timeout.
629
        #[allow(deprecated)] // TODO #1885
630
        let mock_time = SimpleMockTimeProvider::from_wallclock(SystemTime::get());
631

            
632
        let mut output = Vec::new();
633
        let mut stream = match get_decoder(data, encoding, AnonymizedRequest::Direct) {
634
            Ok(s) => s,
635
            Err(e) => return (Err(e), output),
636
        };
637

            
638
        let r = read_and_decompress(&mock_time, &mut stream, maxlen, &mut output).await;
639

            
640
        (r, output)
641
    }
642

            
643
    #[async_test]
644
    async fn decompress_identity() -> RequestResult<()> {
645
        let mut text = Vec::new();
646
        for _ in 0..1000 {
647
            text.extend(b"This is a string with a nontrivial length that we'll use to make sure that the loop is executed more than once.");
648
        }
649

            
650
        let limit = 10 << 20;
651
        let (s, r) = decomp_basic(None, &text[..], limit).await;
652
        s?;
653
        assert_eq!(r, text);
654

            
655
        let (s, r) = decomp_basic(Some("identity"), &text[..], limit).await;
656
        s?;
657
        assert_eq!(r, text);
658

            
659
        // Try truncated result
660
        let limit = 100;
661
        let (s, r) = decomp_basic(Some("identity"), &text[..], limit).await;
662
        assert!(s.is_err());
663
        assert_eq!(r, &text[..100]);
664

            
665
        Ok(())
666
    }
667

            
668
    #[async_test]
669
    async fn decomp_zlib() -> RequestResult<()> {
670
        let compressed =
671
            hex::decode("789cf3cf4b5548cb2cce500829cf8730825253200ca79c52881c00e5970c88").unwrap();
672

            
673
        let limit = 10 << 20;
674
        let (s, r) = decomp_basic(Some("deflate"), &compressed, limit).await;
675
        s?;
676
        assert_eq!(r, b"One fish Two fish Red fish Blue fish");
677

            
678
        Ok(())
679
    }
680

            
681
    #[cfg(feature = "zstd")]
682
    #[async_test]
683
    async fn decomp_zstd() -> RequestResult<()> {
684
        let compressed = hex::decode("28b52ffd24250d0100c84f6e6520666973682054776f526564426c756520666973680a0200600c0e2509478352cb").unwrap();
685
        let limit = 10 << 20;
686
        let (s, r) = decomp_basic(Some("x-zstd"), &compressed, limit).await;
687
        s?;
688
        assert_eq!(r, b"One fish Two fish Red fish Blue fish\n");
689

            
690
        Ok(())
691
    }
692

            
693
    #[cfg(feature = "xz")]
694
    #[async_test]
695
    async fn decomp_xz() -> RequestResult<()> {
696
        // echo "One fish Two fish Red fish Blue fish" | xz -6 | xxd -p
697
        let compressed = hex::decode("fd377a585a000004e6d6b44604c02525210116000000000000000000bfdbca33e00024001d5d00279b88a202ca8612cfb3c19c87c34248a570451e4851d3323d34ab8000000000000901af64854c91f600014125281fe0821fb6f37d010000000004595a").unwrap();
698
        let limit = LZMA_DICT_MEM_LIMIT as usize;
699
        let (s, r) = decomp_basic(Some("x-tor-lzma"), &compressed, limit).await;
700
        s?;
701
        assert_eq!(r, b"One fish Two fish Red fish Blue fish\n");
702

            
703
        Ok(())
704
    }
705

            
706
    #[cfg(feature = "xz")]
707
    #[async_test]
708
    async fn decomp_xz_bad() -> RequestResult<()> {
709
        // Not so good at tiny files...
710
        let compressed = hex::decode("fd377a585a000004e6d6b446020021011c00000010cf58cce00024001d5d00279b88a202ca8612cfb3c19c87c34248a570451e4851d3323d34ab8000000000000901af64854c91f600013925d6ec06651fb6f37d010000000004595a").unwrap();
711
        let limit = 10 << 20;
712
        let (s, _) = decomp_basic(Some("x-tor-lzma"), &compressed, limit).await;
713
        assert!(matches!(s, Err(RequestError::IoError(_))));
714

            
715
        Ok(())
716
    }
717

            
718
    #[async_test]
719
    async fn decomp_unknown() {
720
        let compressed = hex::decode("28b52ffd24250d0100c84f6e6520666973682054776f526564426c756520666973680a0200600c0e2509478352cb").unwrap();
721
        let limit = 10 << 20;
722
        let (s, _r) = decomp_basic(Some("x-proprietary-rle"), &compressed, limit).await;
723

            
724
        assert!(matches!(s, Err(RequestError::ContentEncoding(_))));
725
    }
726

            
727
    #[async_test]
728
    async fn decomp_bad_data() {
729
        let compressed = b"This is not good zlib data";
730
        let limit = 10 << 20;
731
        let (s, _r) = decomp_basic(Some("deflate"), compressed, limit).await;
732

            
733
        // This should possibly be a different type in the future.
734
        assert!(matches!(s, Err(RequestError::IoError(_))));
735
    }
736

            
737
    #[test]
738
    fn headers_ok() -> RequestResult<()> {
739
        let text = b"HTTP/1.0 200 OK\r\nDate: ignored\r\nContent-Encoding: Waffles\r\n\r\n";
740

            
741
        let mut s = &text[..];
742
        let h = tor_rtcompat::test_with_one_runtime!(async |rt| read_headers(&rt, &mut s).await)?;
743

            
744
        assert_eq!(h.status, Some(200));
745
        assert_eq!(h.encoding.as_deref(), Some("Waffles"));
746

            
747
        // now try truncated
748
        let mut s = &text[..15];
749
        let h = tor_rtcompat::test_with_one_runtime!(async |rt| read_headers(&rt, &mut s).await);
750
        assert!(matches!(h, Err(RequestError::TruncatedHeaders)));
751

            
752
        // now try with no encoding.
753
        let text = b"HTTP/1.0 404 Not found\r\n\r\n";
754
        let mut s = &text[..];
755
        let h = tor_rtcompat::test_with_one_runtime!(async |rt| read_headers(&rt, &mut s).await)?;
756

            
757
        assert_eq!(h.status, Some(404));
758
        assert!(h.encoding.is_none());
759

            
760
        Ok(())
761
    }
762

            
763
    #[test]
764
    fn headers_bogus() {
765
        let text = b"HTTP/999.0 WHAT EVEN\r\n\r\n";
766
        let mut s = &text[..];
767
        let h = tor_rtcompat::test_with_one_runtime!(async |rt| read_headers(&rt, &mut s).await);
768

            
769
        assert!(h.is_err());
770
        assert!(matches!(h, Err(RequestError::HttparseError(_))));
771
    }
772

            
773
    /// Run a trivial download example with a response provided as a binary
774
    /// string.
775
    ///
776
    /// Return the directory response (if any) and the request as encoded (if
777
    /// any.)
778
    fn run_download_test<Req: request::Requestable>(
779
        req: Req,
780
        response: &[u8],
781
    ) -> (Result<DirResponse>, RequestResult<Vec<u8>>) {
782
        let (mut s1, s2) = stream_pair();
783
        let (mut s2_r, mut s2_w) = s2.split();
784

            
785
        tor_rtcompat::test_with_one_runtime!(|rt| async move {
786
            let rt2 = rt.clone();
787
            let (v1, v2, v3): (
788
                Result<DirResponse>,
789
                RequestResult<Vec<u8>>,
790
                RequestResult<()>,
791
            ) = futures::join!(
792
                async {
793
                    // Run the download function.
794
                    let r = send_request(&rt, &req, &mut s1, None).await;
795
                    s1.close().await.map_err(|error| {
796
                        Error::RequestFailed(RequestFailedError {
797
                            source: None,
798
                            error: error.into(),
799
                        })
800
                    })?;
801
                    r
802
                },
803
                async {
804
                    // Take the request from the client, and return it in "v2"
805
                    let mut v = Vec::new();
806
                    s2_r.read_to_end(&mut v).await?;
807
                    Ok(v)
808
                },
809
                async {
810
                    // Send back a response.
811
                    s2_w.write_all(response).await?;
812
                    // We wait a moment to give the other side time to notice it
813
                    // has data.
814
                    //
815
                    // (Tentative diagnosis: The `async-compress` crate seems to
816
                    // be behave differently depending on whether the "close"
817
                    // comes right after the incomplete data or whether it comes
818
                    // after a delay.  If there's a delay, it notices the
819
                    // truncated data and tells us about it. But when there's
820
                    // _no_delay, it treats the data as an error and doesn't
821
                    // tell our code.)
822

            
823
                    // TODO: sleeping in tests is not great.
824
                    rt2.sleep(Duration::from_millis(50)).await;
825
                    s2_w.close().await?;
826
                    Ok(())
827
                }
828
            );
829

            
830
            assert!(v3.is_ok());
831

            
832
            (v1, v2)
833
        })
834
    }
835

            
836
    #[test]
837
    fn test_send_request() -> RequestResult<()> {
838
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
839

            
840
        let (response, request) = run_download_test(
841
            req,
842
            b"HTTP/1.0 200 OK\r\n\r\nThis is where the descs would go.",
843
        );
844

            
845
        let request = request?;
846
        assert!(request[..].starts_with(
847
            b"GET /tor/micro/d/CQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQk HTTP/1.0\r\n"
848
        ));
849

            
850
        let response = response.unwrap();
851
        assert_eq!(response.status_code(), 200);
852
        assert!(!response.is_partial());
853
        assert!(response.error().is_none());
854
        assert!(response.source().is_none());
855
        let out_ref = response.output_unchecked();
856
        assert_eq!(out_ref, b"This is where the descs would go.");
857
        let out = response.into_output_unchecked();
858
        assert_eq!(&out, b"This is where the descs would go.");
859

            
860
        Ok(())
861
    }
862

            
863
    #[test]
864
    fn test_download_truncated() {
865
        // Request only one md, so "partial ok" will not be set.
866
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
867
        let mut response_text: Vec<u8> =
868
            (*b"HTTP/1.0 200 OK\r\nContent-Encoding: deflate\r\n\r\n").into();
869
        // "One fish two fish" as above twice, but truncated the second time
870
        response_text.extend(
871
            hex::decode("789cf3cf4b5548cb2cce500829cf8730825253200ca79c52881c00e5970c88").unwrap(),
872
        );
873
        response_text.extend(
874
            hex::decode("789cf3cf4b5548cb2cce500829cf8730825253200ca79c52881c00e5").unwrap(),
875
        );
876
        let (response, request) = run_download_test(req, &response_text);
877
        assert!(request.is_ok());
878
        assert!(response.is_err()); // The whole download should fail, since partial_ok wasn't set.
879

            
880
        // request two microdescs, so "partial_ok" will be set.
881
        let req: request::MicrodescRequest = vec![[9; 32]; 2].into_iter().collect();
882

            
883
        let (response, request) = run_download_test(req, &response_text);
884
        assert!(request.is_ok());
885

            
886
        let response = response.unwrap();
887
        assert_eq!(response.status_code(), 200);
888
        assert!(response.error().is_some());
889
        assert!(response.is_partial());
890
        assert!(response.output_unchecked().len() < 37 * 2);
891
        assert!(response.output_unchecked().starts_with(b"One fish"));
892
    }
893

            
894
    #[test]
895
    fn test_404() {
896
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
897
        let response_text = b"HTTP/1.0 418 I'm a teapot\r\n\r\n";
898
        let (response, _request) = run_download_test(req, response_text);
899

            
900
        assert_eq!(response.unwrap().status_code(), 418);
901
    }
902

            
903
    #[test]
904
    fn test_headers_truncated() {
905
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
906
        let response_text = b"HTTP/1.0 404 truncation happens here\r\n";
907
        let (response, _request) = run_download_test(req, response_text);
908

            
909
        assert!(matches!(
910
            response,
911
            Err(Error::RequestFailed(RequestFailedError {
912
                error: RequestError::TruncatedHeaders,
913
                ..
914
            }))
915
        ));
916

            
917
        // Try a completely empty response.
918
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
919
        let response_text = b"";
920
        let (response, _request) = run_download_test(req, response_text);
921

            
922
        assert!(matches!(
923
            response,
924
            Err(Error::RequestFailed(RequestFailedError {
925
                error: RequestError::TruncatedHeaders,
926
                ..
927
            }))
928
        ));
929
    }
930

            
931
    #[test]
932
    fn test_headers_too_long() {
933
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
934
        let mut response_text: Vec<u8> = (*b"HTTP/1.0 418 I'm a teapot\r\nX-Too-Many-As: ").into();
935
        response_text.resize(16384, b'A');
936
        let (response, _request) = run_download_test(req, &response_text);
937

            
938
        assert!(response.as_ref().unwrap_err().should_retire_circ());
939
        assert!(matches!(
940
            response,
941
            Err(Error::RequestFailed(RequestFailedError {
942
                error: RequestError::HeadersTooLong(_),
943
                ..
944
            }))
945
        ));
946
    }
947

            
948
    #[test]
949
    fn test_bad_utf8() {
950
        let req: request::MicrodescRequest = vec![[9; 32]].into_iter().collect();
951
        let faulty_utf8 = vec![0, 159, 146, 150];
952

            
953
        let mut response_text: Vec<u8> = b"HTTP/1.0 200 OK\r\n\r\n".into();
954
        response_text.extend(faulty_utf8);
955

            
956
        let (response, _request) = run_download_test(req, &response_text);
957

            
958
        assert!(matches!(
959
            response.unwrap().into_output_string().unwrap_err(),
960
            RequestFailedError {
961
                error: RequestError::Utf8Encoding(_),
962
                ..
963
            }
964
        ));
965
    }
966
}