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
#![deny(clippy::print_stderr)]
28
#![deny(clippy::print_stdout)]
29
#![warn(clippy::rc_buffer)]
30
#![deny(clippy::ref_option_ref)]
31
#![warn(clippy::semicolon_if_nothing_returned)]
32
#![warn(clippy::trait_duplication_in_bounds)]
33
#![deny(clippy::unchecked_time_subtraction)]
34
#![deny(clippy::unnecessary_wraps)]
35
#![warn(clippy::unseparated_literal_suffix)]
36
#![deny(clippy::unwrap_used)]
37
#![deny(clippy::mod_module_files)]
38
#![allow(clippy::let_unit_value)] // This can reasonably be done for explicitness
39
#![allow(clippy::uninlined_format_args)]
40
#![allow(clippy::significant_drop_in_scrutinee)] // arti/-/merge_requests/588/#note_2812945
41
#![allow(clippy::result_large_err)] // temporary workaround for arti#587
42
#![allow(clippy::needless_raw_string_hashes)] // complained-about code is fine, often best
43
#![allow(clippy::needless_lifetimes)] // See arti#1765
44
#![allow(mismatched_lifetime_syntaxes)] // temporary workaround for arti#2060
45
#![allow(clippy::collapsible_if)] // See arti#2342
46
#![deny(clippy::unused_async)]
47
#![deny(clippy::string_slice)] // See arti#2571
48
#![allow(recursion_depth_exceeding_limit)] // arti#2715, rust/issues/159228
49
//! <!-- @@ end lint list maintained by maint/add_warning @@ -->
50

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

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

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

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

            
74
use futures::FutureExt;
75
use futures::io::{
76
    AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
77
};
78
use memchr::memchr;
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(&tunnel).await.map_err(wrap_err)?;
197

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

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

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

            
215
    r
216
}
217

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

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

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

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

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

            
300
194
    let mut buffered = BufReader::new(stream);
301

            
302
    // Handle the response
303
    // TODO: should there be a separate timeout here?
304
194
    let header = read_headers(&mut buffered).await.map_err(wrap_err)?;
305
156
    if header.status != Some(200) {
306
34
        return Ok(DirResponse::new(
307
34
            method,
308
34
            header.status.unwrap_or(0),
309
34
            header.status_message,
310
34
            None,
311
34
            vec![],
312
34
            source,
313
34
        ));
314
122
    }
315

            
316
122
    let mut decoder =
317
122
        get_decoder(buffered, header.encoding.as_deref(), anonymized).map_err(wrap_err)?;
318

            
319
122
    let mut result = Vec::new();
320
122
    let ok = read_and_decompress(runtime, &mut decoder, maxlen, &mut result).await;
321

            
322
122
    let ok = match (partial_ok, ok, result.len()) {
323
2
        (true, Err(e), n) if n > 0 => {
324
            // Note that we _don't_ return here: we want the partial response.
325
2
            Err(e)
326
        }
327
2
        (_, Err(e), _) => {
328
2
            return Err(wrap_err(e));
329
        }
330
118
        (_, Ok(()), _) => Ok(()),
331
    };
332

            
333
120
    Ok(DirResponse::new(
334
120
        method,
335
120
        200,
336
120
        None,
337
120
        ok.err(),
338
120
        result,
339
120
        source,
340
120
    ))
341
194
}
342

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

            
348
/// Read and parse HTTP/1 headers from `stream`.
349
202
async fn read_headers<S>(stream: &mut S) -> RequestResult<HeaderStatus>
350
202
where
351
202
    S: AsyncBufRead + Unpin,
352
202
{
353
202
    let mut buf = Vec::with_capacity(1024);
354

            
355
    loop {
356
        // TODO: it's inefficient to do this a line at a time; it would
357
        // probably be better to read until the CRLF CRLF ending of the
358
        // response.  But this should be fast enough.
359
398
        let n = read_until_limited(stream, b'\n', 2048, &mut buf).await?;
360

            
361
        // TODO(nickm): Better maximum and/or let this expand.
362
366
        let mut headers = [httparse::EMPTY_HEADER; 32];
363
366
        let mut response = httparse::Response::new(&mut headers);
364

            
365
366
        match response.parse(&buf[..])? {
366
            httparse::Status::Partial => {
367
                // We didn't get a whole response; we may need to try again.
368

            
369
204
                if n == 0 {
370
                    // We hit an EOF; no more progress can be made.
371
6
                    return Err(RequestError::TruncatedHeaders);
372
198
                }
373

            
374
198
                if buf.len() >= MAX_HEADERS_LEN {
375
2
                    return Err(RequestError::HeadersTooLong(buf.len()));
376
196
                }
377
            }
378
160
            httparse::Status::Complete(n_parsed) => {
379
160
                if response.code != Some(200) {
380
36
                    return Ok(HeaderStatus {
381
36
                        status: response.code,
382
36
                        status_message: response.reason.map(str::to_owned),
383
36
                        encoding: None,
384
36
                    });
385
124
                }
386
124
                let encoding = if let Some(enc) = response
387
124
                    .headers
388
124
                    .iter()
389
124
                    .find(|h| h.name == "Content-Encoding")
390
                {
391
10
                    Some(String::from_utf8(enc.value.to_vec())?)
392
                } else {
393
114
                    None
394
                };
395
                /*
396
                if let Some(clen) = response.headers.iter().find(|h| h.name == "Content-Length") {
397
                    let clen = std::str::from_utf8(clen.value)?;
398
                    length = Some(clen.parse()?);
399
                }
400
                 */
401
124
                assert!(n_parsed == buf.len());
402
124
                return Ok(HeaderStatus {
403
124
                    status: Some(200),
404
124
                    status_message: None,
405
124
                    encoding,
406
124
                });
407
            }
408
        }
409
196
        if n == 0 {
410
            return Err(RequestError::TruncatedHeaders);
411
196
        }
412
    }
413
202
}
414

            
415
/// Return value from read_headers
416
#[derive(Debug, Clone)]
417
struct HeaderStatus {
418
    /// HTTP status code.
419
    status: Option<u16>,
420
    /// HTTP status message associated with the status code.
421
    status_message: Option<String>,
422
    /// The Content-Encoding header, if any.
423
    encoding: Option<String>,
424
}
425

            
426
/// Helper: download directory information from `stream` and
427
/// decompress it into a result buffer.  Assumes that `buf` is empty.
428
///
429
/// If we get more than maxlen bytes after decompression, give an error.
430
///
431
/// Returns the status of our download attempt, stores any data that
432
/// we were able to download into `result`.  Existing contents of
433
/// `result` are overwritten.
434
136
async fn read_and_decompress<S, SP>(
435
136
    runtime: &SP,
436
136
    mut stream: S,
437
136
    maxlen: usize,
438
136
    result: &mut Vec<u8>,
439
136
) -> RequestResult<()>
440
136
where
441
136
    S: AsyncRead + Unpin,
442
136
    SP: SleepProvider,
443
136
{
444
136
    let buffer_window_size = 1024;
445
136
    let mut written_total: usize = 0;
446
    // TODO(nickm): This should be an option, and is maybe too long.
447
    // Though for some users it may be too short?
448
136
    let read_timeout = Duration::from_secs(10);
449
136
    let timer = runtime.sleep(read_timeout).fuse();
450
136
    futures::pin_mut!(timer);
451

            
452
    loop {
453
        // allocate buffer for next read
454
814
        result.resize(written_total + buffer_window_size, 0);
455
814
        let buf: &mut [u8] = &mut result[written_total..written_total + buffer_window_size];
456

            
457
814
        let status = futures::select! {
458
814
            status = stream.read(buf).fuse() => status,
459
            _ = timer => {
460
                result.resize(written_total, 0); // truncate as needed
461
                return Err(RequestError::DirTimeout);
462
            }
463
        };
464
814
        let written_in_this_loop = match status {
465
808
            Ok(n) => n,
466
6
            Err(other) => {
467
6
                result.resize(written_total, 0); // truncate as needed
468
6
                return Err(other.into());
469
            }
470
        };
471

            
472
808
        written_total += written_in_this_loop;
473

            
474
        // exit conditions below
475

            
476
808
        if written_in_this_loop == 0 {
477
            /*
478
            in case we read less than `buffer_window_size` in last `read`
479
            we need to shrink result because otherwise we'll return those
480
            un-read 0s
481
            */
482
128
            if written_total < result.len() {
483
128
                result.resize(written_total, 0);
484
128
            }
485
128
            return Ok(());
486
680
        }
487

            
488
        // TODO: It would be good to detect compression bombs, but
489
        // that would require access to the internal stream, which
490
        // would in turn require some tricky programming.  For now, we
491
        // use the maximum length here to prevent an attacker from
492
        // filling our RAM.
493
680
        if written_total > maxlen {
494
2
            result.resize(maxlen, 0);
495
2
            return Err(RequestError::ResponseTooLong(written_total));
496
678
        }
497
    }
498
136
}
499

            
500
/// Retire a directory circuit because of an error we've encountered on it.
501
fn retire_circ<R>(circ_mgr: &Arc<CircMgr<R>>, id: &tor_proto::circuit::UniqId, error: &str)
502
where
503
    R: Runtime,
504
{
505
    info!(
506
        "{}: Retiring circuit because of directory failure: {}",
507
        &id, &error
508
    );
509
    circ_mgr.retire_circ(id);
510
}
511

            
512
/// As AsyncBufReadExt::read_until, but stops after reading `max` bytes.
513
///
514
/// Note that this function might not actually read any byte of value
515
/// `byte`, since EOF might occur, or we might fill the buffer.
516
///
517
/// A return value of 0 indicates an end-of-file.
518
404
async fn read_until_limited<S>(
519
404
    stream: &mut S,
520
404
    byte: u8,
521
404
    max: usize,
522
404
    buf: &mut Vec<u8>,
523
404
) -> std::io::Result<usize>
524
404
where
525
404
    S: AsyncBufRead + Unpin,
526
404
{
527
404
    let mut n_added = 0;
528
    loop {
529
562
        let data = stream.fill_buf().await?;
530
530
        if data.is_empty() {
531
            // End-of-file has been reached.
532
12
            return Ok(n_added);
533
518
        }
534
518
        debug_assert!(n_added < max);
535
518
        let remaining_space = max - n_added;
536
518
        let (available, found_byte) = match memchr(byte, data) {
537
346
            Some(idx) => (idx + 1, true),
538
172
            None => (data.len(), false),
539
        };
540
518
        debug_assert!(available >= 1);
541
518
        let n_to_copy = std::cmp::min(remaining_space, available);
542
518
        buf.extend(&data[..n_to_copy]);
543
518
        stream.consume_unpin(n_to_copy);
544
518
        n_added += n_to_copy;
545
518
        if found_byte || n_added == max {
546
360
            return Ok(n_added);
547
158
        }
548
    }
549
404
}
550

            
551
/// Helper: Return a boxed decoder object that wraps the stream  $s.
552
macro_rules! decoder {
553
    ($dec:ident, $s:expr) => {{
554
        let mut decoder = $dec::new($s);
555
        decoder.multiple_members(true);
556
        Ok(Box::new(decoder))
557
    }};
558
}
559

            
560
/// Wrap `stream` in an appropriate type to undo the content encoding
561
/// as described in `encoding`.
562
138
fn get_decoder<'a, S: AsyncBufRead + Unpin + Send + 'a>(
563
138
    stream: S,
564
138
    encoding: Option<&str>,
565
138
    anonymized: AnonymizedRequest,
566
138
) -> RequestResult<Box<dyn AsyncRead + Unpin + Send + 'a>> {
567
    use AnonymizedRequest::Direct;
568
138
    match (encoding, anonymized) {
569
132
        (None | Some("identity"), _) => Ok(Box::new(stream)),
570
14
        (Some("deflate"), _) => decoder!(ZlibDecoder, stream),
571
        // We only admit to supporting these on a direct connection; otherwise,
572
        // a hostile directory could send them back even though we hadn't
573
        // requested them.
574
        #[cfg(feature = "xz")]
575
6
        (Some("x-tor-lzma"), Direct) => decoder!(XzDecoder, stream),
576
        #[cfg(feature = "zstd")]
577
4
        (Some("x-zstd"), Direct) => decoder!(ZstdDecoder, stream),
578
2
        (Some(other), _) => Err(RequestError::ContentEncoding(other.into())),
579
    }
580
138
}
581

            
582
#[cfg(test)]
583
mod test {
584
    // @@ begin test lint list maintained by maint/add_warning @@
585
    #![allow(clippy::bool_assert_comparison)]
586
    #![allow(clippy::clone_on_copy)]
587
    #![allow(clippy::dbg_macro)]
588
    #![allow(clippy::mixed_attributes_style)]
589
    #![allow(clippy::print_stderr)]
590
    #![allow(clippy::print_stdout)]
591
    #![allow(clippy::single_char_pattern)]
592
    #![allow(clippy::unwrap_used)]
593
    #![allow(clippy::unchecked_time_subtraction)]
594
    #![allow(clippy::useless_vec)]
595
    #![allow(clippy::needless_pass_by_value)]
596
    #![allow(clippy::string_slice)] // See arti#2571
597
    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
598
    use super::*;
599
    use tor_rtmock::io::stream_pair;
600

            
601
    use tor_rtmock::simple_time::SimpleMockTimeProvider;
602
    use web_time_compat::{SystemTime, SystemTimeExt};
603

            
604
    use futures_await_test::async_test;
605

            
606
    #[async_test]
607
    async fn test_read_until_limited() -> RequestResult<()> {
608
        let mut out = Vec::new();
609
        let bytes = b"This line eventually ends\nthen comes another\n";
610

            
611
        // Case 1: find a whole line.
612
        let mut s = &bytes[..];
613
        let res = read_until_limited(&mut s, b'\n', 100, &mut out).await;
614
        assert_eq!(res?, 26);
615
        assert_eq!(&out[..], b"This line eventually ends\n");
616

            
617
        // Case 2: reach the limit.
618
        let mut s = &bytes[..];
619
        out.clear();
620
        let res = read_until_limited(&mut s, b'\n', 10, &mut out).await;
621
        assert_eq!(res?, 10);
622
        assert_eq!(&out[..], b"This line ");
623

            
624
        // Case 3: reach EOF.
625
        let mut s = &bytes[..];
626
        out.clear();
627
        let res = read_until_limited(&mut s, b'Z', 100, &mut out).await;
628
        assert_eq!(res?, 45);
629
        assert_eq!(&out[..], &bytes[..]);
630

            
631
        Ok(())
632
    }
633

            
634
    // Basic decompression wrapper.
635
    async fn decomp_basic(
636
        encoding: Option<&str>,
637
        data: &[u8],
638
        maxlen: usize,
639
    ) -> (RequestResult<()>, Vec<u8>) {
640
        // We don't need to do anything fancy here, since we aren't simulating
641
        // a timeout.
642
        #[allow(deprecated)] // TODO #1885
643
        let mock_time = SimpleMockTimeProvider::from_wallclock(SystemTime::get());
644

            
645
        let mut output = Vec::new();
646
        let mut stream = match get_decoder(data, encoding, AnonymizedRequest::Direct) {
647
            Ok(s) => s,
648
            Err(e) => return (Err(e), output),
649
        };
650

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

            
653
        (r, output)
654
    }
655

            
656
    #[async_test]
657
    async fn decompress_identity() -> RequestResult<()> {
658
        let mut text = Vec::new();
659
        for _ in 0..1000 {
660
            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.");
661
        }
662

            
663
        let limit = 10 << 20;
664
        let (s, r) = decomp_basic(None, &text[..], limit).await;
665
        s?;
666
        assert_eq!(r, text);
667

            
668
        let (s, r) = decomp_basic(Some("identity"), &text[..], limit).await;
669
        s?;
670
        assert_eq!(r, text);
671

            
672
        // Try truncated result
673
        let limit = 100;
674
        let (s, r) = decomp_basic(Some("identity"), &text[..], limit).await;
675
        assert!(s.is_err());
676
        assert_eq!(r, &text[..100]);
677

            
678
        Ok(())
679
    }
680

            
681
    #[async_test]
682
    async fn decomp_zlib() -> RequestResult<()> {
683
        let compressed =
684
            hex::decode("789cf3cf4b5548cb2cce500829cf8730825253200ca79c52881c00e5970c88").unwrap();
685

            
686
        let limit = 10 << 20;
687
        let (s, r) = decomp_basic(Some("deflate"), &compressed, limit).await;
688
        s?;
689
        assert_eq!(r, b"One fish Two fish Red fish Blue fish");
690

            
691
        Ok(())
692
    }
693

            
694
    #[cfg(feature = "zstd")]
695
    #[async_test]
696
    async fn decomp_zstd() -> RequestResult<()> {
697
        let compressed = hex::decode("28b52ffd24250d0100c84f6e6520666973682054776f526564426c756520666973680a0200600c0e2509478352cb").unwrap();
698
        let limit = 10 << 20;
699
        let (s, r) = decomp_basic(Some("x-zstd"), &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_xz2() -> RequestResult<()> {
709
        // Not so good at tiny files...
710
        let compressed = hex::decode("fd377a585a000004e6d6b446020021011c00000010cf58cce00024001d5d00279b88a202ca8612cfb3c19c87c34248a570451e4851d3323d34ab8000000000000901af64854c91f600013925d6ec06651fb6f37d010000000004595a").unwrap();
711
        let limit = 10 << 20;
712
        let (s, r) = decomp_basic(Some("x-tor-lzma"), &compressed, limit).await;
713
        s?;
714
        assert_eq!(r, b"One fish Two fish Red fish Blue fish\n");
715

            
716
        Ok(())
717
    }
718

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

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

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

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

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

            
742
        let mut s = &text[..];
743
        let h = read_headers(&mut s).await?;
744

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

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

            
753
        // now try with no encoding.
754
        let text = b"HTTP/1.0 404 Not found\r\n\r\n";
755
        let mut s = &text[..];
756
        let h = read_headers(&mut s).await?;
757

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

            
761
        Ok(())
762
    }
763

            
764
    #[async_test]
765
    async fn headers_bogus() -> Result<()> {
766
        let text = b"HTTP/999.0 WHAT EVEN\r\n\r\n";
767
        let mut s = &text[..];
768
        let h = read_headers(&mut s).await;
769

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

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

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

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

            
832
            assert!(v3.is_ok());
833

            
834
            (v1, v2)
835
        })
836
    }
837

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

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

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

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

            
862
        Ok(())
863
    }
864

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

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

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

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

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

            
902
        assert_eq!(response.unwrap().status_code(), 418);
903
    }
904

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

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

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

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

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

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

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

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

            
958
        let (response, _request) = run_download_test(req, &response_text);
959

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