1
//! Implement a simple proxy that relays connections over Tor.
2
//!
3
//! A proxy is launched with [`bind_proxy()`], which opens listener ports.
4
//! `StreamProxy::run_proxy` then listens for new
5
//! connections, handles an appropriate handshake,
6
//! and then relays traffic as appropriate.
7

            
8
semipublic_mod! {
9
    #[cfg(feature="http-connect")]
10
    mod http_connect;
11
    mod socks;
12
    pub(crate) mod port_info;
13
}
14

            
15
use derive_more::Display;
16
use extend::ext;
17
use futures::io::{AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, Error as IoError};
18
use futures::stream::StreamExt;
19
use std::net::IpAddr;
20
use std::sync::Arc;
21
use tor_basic_utils::error_sources::ErrorSources;
22
use tor_log_ratelim::log_ratelim;
23
use tor_rtcompat::{NetStreamProvider, SpawnExt, TcpListenOptions};
24
use tracing::{debug, error, info, instrument, warn};
25

            
26
#[allow(unused)]
27
use arti_client::HasKind;
28
use arti_client::TorClient;
29
#[cfg(feature = "rpc")]
30
use arti_rpcserver::RpcMgr;
31
use tor_config::Listen;
32
use tor_error::{debug_report, warn_report};
33
use tor_rtcompat::{NetStreamListener, Runtime};
34
use tor_socksproto::SocksAuth;
35

            
36
use anyhow::{Context, Result, anyhow};
37

            
38
/// Placeholder type when RPC is disabled at compile time.
39
#[cfg(not(feature = "rpc"))]
40
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
41
pub(crate) enum RpcMgr {}
42

            
43
/// A set of proxy protocols to support on a listener.
44
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
45
#[derive(Copy, Clone, Debug)]
46
#[non_exhaustive]
47
pub(crate) enum ListenProtocols {
48
    /// Only the socks protocol.
49
    SocksOnly,
50
    /// Socks _and_ HTTP CONNECT.
51
    SocksAndHttpConnect,
52
}
53

            
54
impl ListenProtocols {
55
    /// Return true if http connect is included in this set of protocols.
56
    fn http_connect_supported(self) -> bool {
57
        matches!(self, Self::SocksAndHttpConnect)
58
    }
59
}
60

            
61
impl std::fmt::Display for ListenProtocols {
62
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63
        match self {
64
            ListenProtocols::SocksOnly => write!(f, "SOCKS"),
65
            ListenProtocols::SocksAndHttpConnect => write!(f, "SOCKS+HTTP"),
66
        }
67
    }
68
}
69

            
70
/// A Key used to isolate connections.
71
///
72
/// Composed of an usize (representing which listener socket accepted
73
/// the connection, the source IpAddr of the client, and the
74
/// authentication string provided by the client).
75
#[derive(Debug, Clone, PartialEq, Eq)]
76
struct StreamIsolationKey(ListenerIsolation, ProvidedIsolation);
77

            
78
/// Isolation information provided through the proxy connection
79
#[derive(Debug, Clone, PartialEq, Eq)]
80
enum ProvidedIsolation {
81
    /// The socks isolation itself.
82
    LegacySocks(SocksAuth),
83
    /// A bytestring provided as isolation with the extended Socks5 username/password protocol.
84
    ExtendedSocks {
85
        /// Which format was negotiated?
86
        ///
87
        /// (At present, different format codes can't share a circuit.)
88
        format_code: u8,
89
        /// What's the isolation string?
90
        isolation: Box<[u8]>,
91
    },
92
    #[cfg(feature = "http-connect")]
93
    /// An HTTP token, taken from headers.
94
    Http(http_connect::Isolation),
95
}
96

            
97
impl arti_client::isolation::IsolationHelper for StreamIsolationKey {
98
    fn compatible_same_type(&self, other: &Self) -> bool {
99
        self == other
100
    }
101

            
102
    fn join_same_type(&self, other: &Self) -> Option<Self> {
103
        if self == other {
104
            Some(self.clone())
105
        } else {
106
            None
107
        }
108
    }
109

            
110
    fn enables_long_lived_circuits(&self) -> bool {
111
        use ProvidedIsolation as PI;
112
        use SocksAuth as SA;
113
        match &self.1 {
114
            PI::LegacySocks(SA::Socks4(auth)) => !auth.is_empty(),
115
            PI::LegacySocks(SA::Username(uname, pass)) => !(uname.is_empty() && pass.is_empty()),
116
            PI::LegacySocks(_) => false,
117
            PI::ExtendedSocks { isolation, .. } => !isolation.is_empty(),
118
            #[cfg(feature = "http-connect")]
119
            PI::Http(isolation) => !isolation.is_empty(),
120
        }
121
    }
122
}
123

            
124
/// Size of read buffer to apply to application data streams
125
/// and Tor data streams when copying.
126
//
127
// This particular value is chosen more or less arbitrarily.
128
// Larger values let us do fewer reads from the application,
129
// but consume more memory.
130
//
131
// (The default value for BufReader is 8k as of this writing.)
132
const APP_STREAM_BUF_LEN: usize = 4096;
133

            
134
const _: () = {
135
    assert!(APP_STREAM_BUF_LEN >= tor_socksproto::SOCKS_BUF_LEN);
136
};
137

            
138
/// NOTE: The following documentation belongs in a spec.
139
/// But for now, it's our best attempt to document the design and protocol
140
/// implemented here
141
/// for integrating proxies with our RPC system. --nickm
142
///
143
/// Roughly speaking:
144
///
145
/// ## Key concepts
146
///
147
/// A data stream is "RPC-visible" if, when it is created via a proxy connection,
148
/// the RPC system is told about it.
149
///
150
/// Every RPC-visible stream is associated with a given RPC object when it is created.
151
/// (Since the RPC object is being specified in the proxy protocol,
152
/// it must be one with an externally visible Object ID.
153
/// Such Object IDs are cryptographically unguessable and unforgeable,
154
/// and are qualified with a unique identifier for their associated RPC session.)
155
/// Call this RPC Object the "target" object for now.
156
/// This target RPC object must implement
157
/// the [`ConnectWithPrefs`](arti_client::rpc::ConnectWithPrefs) special method.
158
///
159
/// Right now, there are two general kinds of objects that implement this method:
160
/// client-like objects, and one-shot clients.
161
///
162
/// A client-like object is either a `TorClient` or an RPC `Session`.
163
/// It knows about and it is capable of opening multiple data streams.
164
/// Using it as the target object for a proxy connection tells Arti
165
/// that the resulting data stream (if any)
166
/// should be built by it, and associated with its RPC session.
167
///
168
/// An application gets a TorClient by asking the session for one,
169
/// or for asking a TorClient to give you a new variant clone of itself.
170
///
171
/// A one-shot client is an `arti_rpcserver::stream::OneshotClient`.
172
/// It is created from a client-like object, but can only be used for a single data stream.
173
/// When created, it it not yet connected or trying to connect to anywhere:
174
/// the act of using it as the target Object for a proxy connection causes
175
/// it to begin connecting.
176
///
177
/// An application gets a `OneShotClient` by calling `arti:new_oneshot_client`
178
/// on any client-like object.
179
///
180
/// ## The Proxy protocol
181
///
182
/// See the specification for
183
/// [SOCKS extended authentication](https://spec.torproject.org/socks-extensions.html#extended-auth)
184
/// for full details on integrating RPC with SOCKS.
185
/// For HTTP integration, see
186
/// [the relevant section of prop365](https://spec.torproject.org/proposals/365-http-connect-ext.html#x-tor-rpc-target-arti-rpc-support).
187
///
188
/// ### Further restrictions on Object IDs and isolation
189
///
190
/// In some cases,
191
/// the RPC Object ID may denote an object
192
/// that already includes information about its intended stream isolation.
193
/// In such cases, the stream isolation MUST be blank.
194
/// Implementations MUST reject non-blank stream isolation in such cases.
195
///
196
/// In some cases, the RPC object ID may denote an object
197
/// that already includes information
198
/// about its intended destination address and port.
199
/// In such cases, the destination address MUST be `0.0.0.0` or `::`
200
/// (encoded either as an IPv4 address, an IPv6 address, or a hostname)
201
/// and the destination port MUST be 0.
202
/// Implementations MUST reject other addresses in such cases.
203
///
204
/// ### Another proposed change
205
///
206
/// We could add a new method to clients, with a name like
207
/// "open_stream" or "connect_stream".
208
/// This method would include all target and isolation information in its parameters.
209
/// It would actually create a DataStream immediately, tell it to begin connecting,
210
/// and return an externally visible object ID.
211
/// The RPC protocol could be used to watch the DataStream object,
212
/// to see when it was connected.
213
///
214
/// The resulting DataStream object could also be used as the target of a proxy connection.
215
/// We would require in such a case that no isolation be provided in the proxy handshake,
216
/// and that the target address was (e.g.) INADDR_ANY.
217
///
218
/// ## Intended use cases (examples)
219
///
220
/// (These examples assume that the application
221
/// already knows the proxy port it should use.
222
/// I'm leaving out the isolation strings as orthogonal.)
223
///
224
/// These are **NOT** the only possible use cases;
225
/// they're just the two that help understand this system best (I hope).
226
///
227
/// ### Case 1: Using a client-like object directly.
228
///
229
/// Here the application has authenticated to RPC
230
/// and gotten the session ID `SESSION-1`.
231
/// (In reality, this would be a longer ID, and full of crypto).
232
///
233
/// The application wants to open a new stream to www.example.com.
234
/// They don't particularly care about isolation,
235
/// but they do want their stream to use their RPC session.
236
/// They don't want an Object ID for the stream.
237
///
238
/// To do this, they make a SOCKS connection to arti,
239
/// with target address www.example.com.
240
/// They set the username to `<torS0X>0SESSION-1`,
241
/// and the password to the empty string.
242
///
243
/// (Alternatively, it could use HTTP CONNECT, setting
244
/// Tor-Rpc-Target to SESSION-1.)
245
///
246
/// Arti looks up the Session object via the `SESSION-1` object ID
247
/// and tells it (via the ConnectWithPrefs special method)
248
/// to connect to www.example.com.
249
/// The session creates a new DataStream using its internal TorClient,
250
/// but does not register the stream with an RPC Object ID.
251
/// Arti proxies the application's connection through this DataStream.
252
///
253
///
254
/// ### Case 2: Creating an identifiable stream.
255
///
256
/// Here the application wants to be able to refer to its DataStream
257
/// after the stream is created.
258
/// As before, we assume that it's on an RPC session
259
/// where the Session ID is `SESSION-1`.
260
///
261
/// The application sends an RPC request of the form:
262
/// `{"id": 123, "obj": "SESSION-1", "method": "arti:new_oneshot_client", "params": {}}`
263
///
264
/// It receives a reply like:
265
/// `{"id": 123, "result": {"id": "STREAM-1"} }`
266
///
267
/// (In reality, `STREAM-1` would also be longer and full of crypto.)
268
///
269
/// Now the application has an object called `STREAM-1` that is not yet a connected
270
/// stream, but which may become one.
271
///
272
/// This time, it wants to set its isolation string to "xyzzy".
273
///
274
/// The application opens a socks connection as before.
275
/// For the username it sends `<torS0X>0STREAM-1`,
276
/// and for the password it sends `xyzzy`.
277
///
278
/// (Alternatively, it could use HTTP CONNECT, setting Tor-Isolation to xyzzy,
279
/// and Tor-Rpc-Target to STREAM-1.)
280
///
281
/// Now Arti looks up the `RpcDataStream` object via `STREAM-1`,
282
/// and tells it (via the ConnectWithPrefs special method)
283
/// to connect to www.example.com.
284
/// This causes the `RpcDataStream` internally to create a new `DataStream`,
285
/// and to store that `DataStream` in itself.
286
/// The `RpcDataStream` with Object ID `STREAM-1`
287
/// is now an alias for the newly created `DataStream`.
288
/// Arti proxies the application's connection through that `DataStream`.
289
///
290
#[cfg(feature = "rpc")]
291
#[allow(dead_code)]
292
mod socks_and_rpc {}
293

            
294
/// Information used to implement a proxy listener.
295
struct ProxyContext<R: Runtime> {
296
    /// A TorClient to use (by default) to anonymize requests.
297
    tor_client: Arc<TorClient<R>>,
298
    /// If present, an RpcMgr to use when for attaching requests to RPC
299
    /// sessions.
300
    #[cfg(feature = "rpc")]
301
    rpc_mgr: Option<Arc<arti_rpcserver::RpcMgr>>,
302
    /// The protocols that we support.
303
    protocols: ListenProtocols,
304
}
305

            
306
/// Type alias for the isolation information associated with a given proxy
307
/// connection _before_ any negotiation occurs.
308
///
309
/// Currently this is an index for which listener accepted the connection, plus
310
/// the address of the client that connected to the proxy port.
311
type ListenerIsolation = (usize, IpAddr);
312

            
313
/// write_all the data to the writer & flush the writer if write_all is successful.
314
async fn write_all_and_flush<W>(writer: &mut W, buf: &[u8]) -> Result<()>
315
where
316
    W: AsyncWrite + Unpin,
317
{
318
    writer
319
        .write_all(buf)
320
        .await
321
        .context("Error while writing proxy reply")?;
322
    writer
323
        .flush()
324
        .await
325
        .context("Error while flushing proxy stream")
326
}
327

            
328
/// write_all the data to the writer & close the writer if write_all is successful.
329
async fn write_all_and_close<W>(writer: &mut W, buf: &[u8]) -> Result<()>
330
where
331
    W: AsyncWrite + Unpin,
332
{
333
    writer
334
        .write_all(buf)
335
        .await
336
        .context("Error while writing proxy reply")?;
337
    writer
338
        .close()
339
        .await
340
        .context("Error while closing proxy stream")
341
}
342

            
343
/// Return true if a given IoError, when received from accept, is a fatal
344
/// error.
345
fn accept_err_is_fatal(err: &IoError) -> bool {
346
    #![allow(clippy::match_like_matches_macro)]
347

            
348
    /// Re-declaration of WSAEMFILE with the right type to match
349
    /// `raw_os_error()`.
350
    #[cfg(windows)]
351
    const WSAEMFILE: i32 = winapi::shared::winerror::WSAEMFILE as i32;
352

            
353
    // Currently, EMFILE and ENFILE aren't distinguished by ErrorKind;
354
    // we need to use OS-specific errors. :P
355
    match err.raw_os_error() {
356
        #[cfg(unix)]
357
        Some(libc::EMFILE) | Some(libc::ENFILE) => false,
358
        #[cfg(windows)]
359
        Some(WSAEMFILE) => false,
360
        _ => true,
361
    }
362
}
363

            
364
/// A stream proxy listening on one or more local ports, ready to relay traffic.
365
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
366
#[must_use]
367
pub(crate) struct StreamProxy<R: Runtime> {
368
    /// A tor client to use when relaying traffic.
369
    tor_client: Arc<TorClient<R>>,
370
    /// The listeners that we've actually bound to.
371
    listeners: Vec<<R as NetStreamProvider>::Listener>,
372
    /// The protocols we respond to.
373
    protocols: ListenProtocols,
374
    /// An RPC manager to use when incoming requests are tied to streams.
375
    rpc_mgr: Option<Arc<RpcMgr>>,
376
}
377

            
378
/// Launch a proxy to listen on a given set of ports.
379
///
380
/// Requires a `runtime` to use for launching tasks and handling
381
/// timeouts, and a `tor_client` to use in connecting over the Tor
382
/// network.
383
///
384
/// Returns the proxy, and a list of the ports that we have
385
/// bound to.
386
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
387
#[instrument(skip_all, level = "trace")]
388
pub(crate) async fn bind_proxy<R: Runtime>(
389
    runtime: R,
390
    tor_client: Arc<TorClient<R>>,
391
    listen: Listen,
392
    listen_options: TcpListenOptions,
393
    protocols: ListenProtocols,
394
    rpc_mgr: Option<Arc<RpcMgr>>,
395
) -> Result<StreamProxy<R>> {
396
    if !listen.is_loopback_only() {
397
        warn!(
398
            "Configured to listen for proxy connections on non-local addresses. \
399
            This is usually insecure! We recommend listening on localhost only."
400
        );
401
    }
402

            
403
    let mut listeners = Vec::new();
404

            
405
    // Try to bind to the listener ports.
406
    match listen.ip_addrs() {
407
        Ok(addrgroups) => {
408
            for addrgroup in addrgroups {
409
                for addr in addrgroup {
410
                    match runtime.listen(&addr, &listen_options).await {
411
                        Ok(listener) => {
412
                            let bound_addr = listener.local_addr()?;
413
                            info!("Listening on {:?}", bound_addr);
414
                            listeners.push(listener);
415
                        }
416
                        #[cfg(unix)]
417
                        Err(ref e) if e.raw_os_error() == Some(libc::EAFNOSUPPORT) => {
418
                            warn_report!(e, "Address family not supported {}", addr);
419
                        }
420
                        Err(ref e) => {
421
                            return Err(anyhow!("Can't listen on {}: {e}", addr));
422
                        }
423
                    }
424
                }
425
                // TODO: We are supposed to fail if every address in the group failed!
426
            }
427
        }
428
        Err(e) => warn_report!(e, "Invalid listen spec"),
429
    }
430

            
431
    // We weren't able to bind any ports: There's nothing to do.
432
    if listeners.is_empty() {
433
        error!("Couldn't open any listeners.");
434
        return Err(anyhow!("Couldn't open listeners"));
435
    }
436

            
437
    Ok(StreamProxy {
438
        tor_client,
439
        listeners,
440
        protocols,
441
        rpc_mgr,
442
    })
443
}
444

            
445
impl<R: Runtime> StreamProxy<R> {
446
    /// Run indefinitely, processing incoming connections and relaying traffic.
447
    pub(crate) async fn run_proxy(self) -> Result<()> {
448
        let StreamProxy {
449
            tor_client,
450
            listeners,
451
            protocols,
452
            rpc_mgr,
453
        } = self;
454
        run_proxy_with_listeners(tor_client, listeners, protocols, rpc_mgr).await
455
    }
456

            
457
    /// Return a list of the ports that we've bound to.
458
    pub(crate) fn port_info(&self) -> Result<Vec<port_info::Port>> {
459
        let mut ports = Vec::new();
460
        for listener in &self.listeners {
461
            let address = listener.local_addr()?;
462
            ports.push(port_info::Port {
463
                protocol: port_info::SupportedProtocol::Socks,
464
                address: address.into(),
465
            });
466
            #[cfg(feature = "http-connect")]
467
            if self.protocols.http_connect_supported() {
468
                ports.push(port_info::Port {
469
                    protocol: port_info::SupportedProtocol::Http,
470
                    address: address.into(),
471
                });
472
            }
473
        }
474

            
475
        Ok(ports)
476
    }
477
}
478

            
479
/// Launch a proxy from a given set of already bound listeners.
480
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
481
#[instrument(skip_all, level = "trace")]
482
pub(crate) async fn run_proxy_with_listeners<R: Runtime>(
483
    tor_client: Arc<TorClient<R>>,
484
    listeners: Vec<<R as tor_rtcompat::NetStreamProvider>::Listener>,
485
    protocols: ListenProtocols,
486
    rpc_mgr: Option<Arc<RpcMgr>>,
487
) -> Result<()> {
488
    // Create a stream of (incoming socket, listener_id) pairs, selected
489
    // across all the listeners.
490
    let mut incoming = futures::stream::select_all(
491
        listeners
492
            .into_iter()
493
            .map(NetStreamListener::incoming)
494
            .enumerate()
495
            .map(|(listener_id, incoming_conns)| {
496
                incoming_conns.map(move |socket| (socket, listener_id))
497
            }),
498
    );
499

            
500
    // Loop over all incoming connections.  For each one, call
501
    // handle_proxy_conn() in a new task.
502
    while let Some((stream, sock_id)) = incoming.next().await {
503
        let (stream, addr) = match stream {
504
            Ok((s, a)) => (s, a),
505
            Err(err) => {
506
                if accept_err_is_fatal(&err) {
507
                    return Err(err).context("Failed to receive incoming stream on proxy port");
508
                } else {
509
                    warn_report!(err, "Incoming stream failed");
510
                    continue;
511
                }
512
            }
513
        };
514
        let proxy_context = ProxyContext {
515
            tor_client: tor_client.clone(),
516
            #[cfg(feature = "rpc")]
517
            rpc_mgr: rpc_mgr.clone(),
518
            protocols,
519
        };
520
        tor_client.runtime().spawn(async move {
521
            let res = handle_proxy_conn(proxy_context, stream, (sock_id, addr.ip())).await;
522
            if let Err(e) = res {
523
                report_proxy_error(e);
524
            }
525
        })?;
526
    }
527

            
528
    Ok(())
529
}
530

            
531
/// A (possibly) supported proxy protocol.
532
enum ProxyProtocols {
533
    /// Some HTTP/1 command or other.
534
    ///
535
    /// (We only support CONNECT and OPTIONS, but we reject other commands in [`http_connect`].)
536
    Http1,
537
    /// SOCKS4 or SOCKS5.
538
    Socks,
539
}
540

            
541
/// Look at the first byte of a proxy connection, and guess what protocol
542
/// what protocol it is trying to speak.
543
fn classify_protocol_from_first_byte(byte: u8) -> Option<ProxyProtocols> {
544
    match byte {
545
        b'a'..=b'z' | b'A'..=b'Z' => Some(ProxyProtocols::Http1),
546
        4 | 5 => Some(ProxyProtocols::Socks),
547
        _ => None,
548
    }
549
}
550

            
551
/// Handle a single connection `stream` from an application.
552
///
553
/// Depending on what protocol the application is speaking
554
/// (and what protocols we support!), negotiate an appropriate set of options,
555
/// and relay traffic to and from the application.
556
async fn handle_proxy_conn<R, S>(
557
    context: ProxyContext<R>,
558
    stream: S,
559
    isolation_info: ListenerIsolation,
560
) -> Result<()>
561
where
562
    R: Runtime,
563
    S: AsyncRead + AsyncWrite + Send + Sync + Unpin + 'static,
564
{
565
    let mut stream = BufReader::with_capacity(APP_STREAM_BUF_LEN, stream);
566
    use futures::AsyncBufReadExt as _;
567

            
568
    let buf: &[u8] = stream.fill_buf().await?;
569
    if buf.is_empty() {
570
        // connection closed
571
        return Ok(());
572
    }
573
    match classify_protocol_from_first_byte(buf[0]) {
574
        Some(ProxyProtocols::Http1) => {
575
            #[cfg(feature = "http-connect")]
576
            if context.protocols.http_connect_supported() {
577
                return http_connect::handle_http_conn(context, stream, isolation_info).await;
578
            }
579

            
580
            write_all_and_close(&mut stream, socks::WRONG_PROTOCOL_PAYLOAD).await?;
581
            Ok(())
582
        }
583
        Some(ProxyProtocols::Socks) => {
584
            socks::handle_socks_conn(context, stream, isolation_info).await
585
        }
586
        None => {
587
            // We have no idea what protocol the client expects,
588
            // so we have no idea how to tell it so.
589
            warn!(
590
                "Unrecognized protocol on proxy listener (first byte {:x})",
591
                buf[0]
592
            );
593
            Ok(())
594
        }
595
    }
596
}
597

            
598
/// If any source of the provided `error` is a [`tor_proto::Error`], return a reference to that
599
/// [`tor_proto::Error`].
600
fn extract_proto_err<'a>(
601
    error: &'a (dyn std::error::Error + 'static),
602
) -> Option<&'a tor_proto::Error> {
603
    for error in ErrorSources::new(error) {
604
        if let Some(downcast) = error.downcast_ref::<tor_proto::Error>() {
605
            return Some(downcast);
606
        }
607
    }
608

            
609
    None
610
}
611

            
612
/// A wrapper that makes anyhow::Error Sized and Cloneable
613
/// by wrapping it in an Arc.
614
// TODO Add better handling for anyhow errors in tor-log-ratelim
615
// create tor-log-ratelim/src/anyhow.rs and put them there
616
#[derive(Debug, Clone, Display)]
617
#[display("{}", _0)]
618
pub(crate) struct RateLimitError(Arc<anyhow::Error>);
619

            
620
impl RateLimitError {
621
    /// Creates a new `RateLimitError` from an `anyhow::Error`.
622
    pub(crate) fn new(e: anyhow::Error) -> Self {
623
        Self(Arc::new(e))
624
    }
625
}
626

            
627
impl std::error::Error for RateLimitError {}
628

            
629
/// Wraps a `Result` containing an `anyhow::Error` into a `Result`
630
/// containing a `Sized` `RateLimitError` to allow usage with `log_ratelim!`.
631
#[ext]
632
impl<T> Result<T, anyhow::Error> {
633
    /// Wraps an `anyhow::Error` into a `Sized` `RateLimitError`
634
    fn wrap_for_ratelimit(self) -> Result<T, RateLimitError> {
635
        self.map_err(RateLimitError::new)
636
    }
637
}
638

            
639
/// Report an error that occurred within a single proxy task.
640
fn report_proxy_error(e: anyhow::Error) {
641
    use tor_proto::Error as PE;
642
    // TODO: In the long run it might be a good idea to use an ErrorKind here if we can get one.
643
    // This is a bit of a kludge based on the fact that we're using anyhow.
644
    //
645
    // TODO: It might be handy to have a way to collapse CircuitClosed into EOF earlier.
646
    // But that loses information, so it should be optional.
647
    //
648
    // TODO: Maybe we should look at io::ErrorKind as well, if it's there.  That's another reason
649
    // to discard or restrict our anyhow usage.
650
    match extract_proto_err(e.as_ref()) {
651
        Some(e @ PE::CircuitClosed) => debug_report!(e, "Connection exited"),
652
        // We can't use `debug_report!` here because `NotConnected`s error kind is `BadApiUsage`,
653
        // which upgrades this to a warning.
654
        // https://gitlab.torproject.org/tpo/core/arti/-/issues/2439
655
        Some(e @ PE::NotConnected) => debug!(error = (e as &dyn std::error::Error), "Connection exited"),
656
        _ => {
657
            // TODO: See https://gitlab.torproject.org/tpo/core/arti/-/work_items/2632
658
            // We use the root cause of the error as the activity key for rate-limiting
659
            // because it provides only the original OS error without any wrappers on the text.
660
            // This avoids a manual if-else chain while maintaining separate buckets
661
            // for different errors.
662
            let bucket = e.root_cause().to_string();
663
            let r: Result<(), RateLimitError> = Err(e).wrap_for_ratelimit();
664

            
665
            log_ratelim!(
666
                "Connection exited (cause: {})", bucket;
667
                r;
668
            );
669
        }
670
    }
671
}