1
//! Entry point of a Tor relay that is the [`TorRelay`] objects
2

            
3
use std::net::SocketAddr;
4
use std::path::{Path, PathBuf};
5
use std::sync::{Arc, Weak};
6

            
7
use anyhow::Context;
8
use tokio::task::JoinSet;
9
use tracing::debug;
10
#[cfg(unix)]
11
use tracing::warn;
12

            
13
use fs_mistrust::Mistrust;
14
use tor_basic_utils::iter_join;
15
use tor_cell::relaycell::RelayCmd;
16
use tor_chanmgr::{ChanMgr, ChanMgrConfig, Dormancy};
17
use tor_config_path::CfgPathResolver;
18
use tor_dircommon::authority::AuthorityContacts;
19
use tor_dircommon::config::{DirTolerance, DownloadScheduleConfig};
20
use tor_dirmgr::DirMgrConfig;
21
use tor_dirserver::mirror::DirMirror;
22
use tor_keymgr::{ArtiNativeKeystore, KeyMgr, KeyMgrBuilder};
23
use tor_memquota::MemoryQuotaTracker;
24
use tor_netdir::params::NetParameters;
25
use tor_persist::state_dir::StateDirectory;
26
use tor_persist::{FsStateMgr, StateMgr};
27
use tor_proto::relay::{CircuitIncomingStreamReceiver, CreateRequestHandler};
28
use tor_rtcompat::{NetStreamProvider, Runtime};
29

            
30
use crate::client::RelayClient;
31
use crate::config::TorRelayConfig;
32
use crate::stream::RequestFilter;
33
use crate::tasks::channel::build_circ_net_params;
34
use crate::tasks::crypto::InitKeyMaterial;
35

            
36
use futures::channel::mpsc;
37

            
38
// TODO(relay): this is in the client module, but not client-specific
39
use tor_proto::client::stream::DataStream;
40

            
41
/// An initialized but unbootstrapped relay.
42
///
43
/// This intentionally does not have access to the runtime to prevent it from doing network io.
44
///
45
/// The idea is that we can build up the relay's components in an `InertTorRelay` without a runtime,
46
/// and then call `init()` on it and provide a runtime to turn it into a network-capable relay.
47
/// This gives us two advantages:
48
///
49
/// - We can initialize the internal data structures in the `InertTorRelay` (load the keystores,
50
///   configure memquota, etc), which leaves `TorRelay` to just "running" the relay (bootstrapping,
51
///   setting up listening sockets, etc). We don't need to combine the initialization and "running
52
///   the relay" all within the same object.
53
/// - We will likely want to share some of arti's key management subcommands in the future.
54
///   arti-client has an `InertTorClient` which is used so that arti subcommands can access the
55
///   keystore. If we do a similar thing here in arti-relay in the future, it might be nice to have
56
///   an `InertTorRelay` which has these internal data structures, but doesn't need a runtime or
57
///   have any networking capabilities.
58
///
59
/// Time will tell if this ends up being a bad design decision in practice, and we can always change
60
/// it later.
61
pub(crate) struct InertTorRelay {
62
    /// The configuration options for the relay.
63
    config: TorRelayConfig,
64

            
65
    /// The configuration options for the client's directory manager.
66
    dirmgr_config: DirMgrConfig,
67

            
68
    /// Path resolver for expanding variables in [`CfgPath`](tor_config_path::CfgPath)s.
69
    #[expect(unused)] // TODO RELAY remove
70
    path_resolver: CfgPathResolver,
71

            
72
    /// State directory path.
73
    ///
74
    /// The [`StateDirectory`] stored in `state_dir` doesn't seem to have a way of getting the state
75
    /// directory path, so we need to store a copy of the path here.
76
    #[expect(unused)] // TODO RELAY remove
77
    state_path: PathBuf,
78

            
79
    /// Relay's state directory.
80
    #[expect(unused)] // TODO RELAY remove
81
    state_dir: StateDirectory,
82

            
83
    /// Location on disk where we store persistent data.
84
    state_mgr: FsStateMgr,
85

            
86
    /// Key manager.
87
    keymgr: KeyMgr,
88
}
89

            
90
impl InertTorRelay {
91
    /// Create a new Tor relay with the given configuration.
92
    pub(crate) fn new(
93
        config: TorRelayConfig,
94
        path_resolver: CfgPathResolver,
95
    ) -> anyhow::Result<Self> {
96
        let state_path = config.storage.state_dir(&path_resolver)?;
97
        let cache_path = config.storage.cache_dir(&path_resolver)?;
98

            
99
        let state_dir = StateDirectory::new(&state_path, config.storage.permissions())
100
            .context("Failed to create `StateDirectory`")?;
101
        let state_mgr =
102
            FsStateMgr::from_path_and_mistrust(&state_path, config.storage.permissions())
103
                .context("Failed to create `FsStateMgr`")?;
104

            
105
        // Try to take state ownership early, so we'll know if we have it.
106
        // Note that this `try_lock()` may return `Ok` even if we can't acquire the lock.
107
        // (At this point we don't yet care if we have it.)
108
        let _ignore_status = state_mgr
109
            .try_lock()
110
            .context("Failed to try locking the state manager")?;
111

            
112
        let keymgr = Self::create_keymgr(&state_path, config.storage.permissions())
113
            .context("Failed to create key manager")?;
114

            
115
        let dirmgr_config = DirMgrConfig {
116
            cache_dir: cache_path,
117
            cache_trust: config.storage.permissions().clone(),
118
            network: config.tor_network.clone(),
119
            schedule: Default::default(),
120
            tolerance: Default::default(),
121
            override_net_params: Default::default(),
122
            extensions: Default::default(),
123
        };
124

            
125
        Ok(Self {
126
            config,
127
            dirmgr_config,
128
            path_resolver,
129
            state_path,
130
            state_dir,
131
            state_mgr,
132
            keymgr,
133
        })
134
    }
135

            
136
    /// Connect the [`InertTorRelay`] to the Tor network.
137
    pub(crate) async fn init<R: Runtime>(self, runtime: R) -> anyhow::Result<TorRelay<R>> {
138
        // Attempt to generate any missing keys/cert from the KeyMgr.
139
        let init_key_material = crate::tasks::crypto::init_keys(&runtime, &self.keymgr)
140
            .context("Failed to generate keys")?;
141

            
142
        TorRelay::init(runtime, self, init_key_material).await
143
    }
144

            
145
    /// Create the [key manager](KeyMgr).
146
    fn create_keymgr(state_path: &Path, mistrust: &Mistrust) -> anyhow::Result<KeyMgr> {
147
        let key_store_dir = state_path.join("keystore");
148

            
149
        let persistent_store = ArtiNativeKeystore::from_path_and_mistrust(&key_store_dir, mistrust)
150
            .context("Failed to construct the native keystore")?;
151

            
152
        // Should only log fs paths at debug level or lower,
153
        // unless they're part of a diagnostic message.
154
        debug!("Using relay keystore from {key_store_dir:?}");
155

            
156
        let keymgr = KeyMgrBuilder::default()
157
            .primary_store(Box::new(persistent_store))
158
            .build()
159
            .context("Failed to build the 'KeyMgr'")?;
160

            
161
        // TODO: support C-tor keystore
162

            
163
        Ok(keymgr)
164
    }
165
}
166

            
167
/// Represent an active Relay on the Tor network.
168
pub(crate) struct TorRelay<R: Runtime> {
169
    /// Asynchronous runtime object.
170
    runtime: R,
171

            
172
    /// Memory quota tracker.
173
    #[expect(unused)] // TODO RELAY remove
174
    memquota: Arc<MemoryQuotaTracker>,
175

            
176
    /// A "client" used by relays to construct circuits.
177
    client: RelayClient<R>,
178

            
179
    /// The directory authorities that were either configured or the compiled-in defaults.
180
    ///
181
    /// We keep a copy here so we can pass it to the descriptor publisher task. These are not
182
    /// exposed by a [`tor_dirmgr::DirProvider`] hence why we keep that copy from the config.
183
    authorities: AuthorityContacts,
184

            
185
    /// The directory mirror object, used for handling BEGIN_DIR.
186
    dir_mirror: DirMirror,
187

            
188
    /// Channel manager, used by circuits etc.
189
    chanmgr: Arc<ChanMgr<R>>,
190

            
191
    /// Handles CREATE* requests on channels.
192
    ///
193
    /// Given to the [`ChanMgr`],
194
    /// which gives it to each channel.
195
    /// We can access this handler directly to update consensus parameters or keys.
196
    create_request_handler: Arc<CreateRequestHandler>,
197

            
198
    /// The receiver for the [`Stream`](futures::Stream)s of `IncomingStream` of all circuits.
199
    ///
200
    /// Receives one [`Stream`](futures::Stream) (of tor streams) per circuit.
201
    /// Each of these is handled in a new task.
202
    circuit_stream_rx: CircuitIncomingStreamReceiver,
203

            
204
    /// See [`InertTorRelay::keymgr`].
205
    keymgr: KeyMgr,
206

            
207
    /// Listening OR ports.
208
    or_listeners: Vec<<R as NetStreamProvider<SocketAddr>>::Listener>,
209
}
210

            
211
impl<R: Runtime> TorRelay<R> {
212
    /// Create a new Tor relay with the given [`runtime`][tor_rtcompat].
213
    ///
214
    /// We use this to initialize components, open sockets, etc.
215
    /// Doing work with these components should happen in [`TorRelay::run()`].
216
    ///
217
    /// Expected to be called from [`InertTorRelay::init()`].
218
    async fn init(
219
        runtime: R,
220
        inert: InertTorRelay,
221
        init_key_material: InitKeyMaterial,
222
    ) -> anyhow::Result<Self> {
223
        let memquota = MemoryQuotaTracker::new(&runtime, inert.config.system.memory.clone())
224
            .context("Failed to initialize memquota tracker")?;
225

            
226
        // Init the channel manager.
227
        let config = ChanMgrConfig::new(inert.config.channel.clone())
228
            .with_my_addrs(inert.config.relay.advertise.all_addr())
229
            .with_auth_material(Arc::new(init_key_material.chan_auth_keys));
230
        let chanmgr = Arc::new(
231
            ChanMgr::new(
232
                runtime.clone(),
233
                config,
234
                Dormancy::Active,
235
                // TODO: It seems wrong to start with the compiled-in defaults when we might have
236
                // a newer network status on disk that would provide a better initial value,
237
                // but `TorClient` does this too so let's not worry about it.
238
                &NetParameters::default(),
239
                memquota.clone(),
240
            )
241
            .context("Failed to build chan manager")?,
242
        );
243

            
244
        let authorities = inert.dirmgr_config.authorities().clone();
245

            
246
        // Init the relay's client.
247
        let client = RelayClient::new(
248
            runtime.clone(),
249
            Arc::clone(&chanmgr),
250
            &inert.config,
251
            &inert.config,
252
            inert.dirmgr_config,
253
            inert.state_mgr,
254
        )
255
        .context("Failed to construct the relay's client")?;
256

            
257
        // Circuit-related network status parameters.
258
        let circ_net_params = build_circ_net_params(client.dirmgr().params().as_ref().as_ref())
259
            .context("Failed to build circuit parameters for CREATE* request handler")?;
260

            
261
        // TODO(relay): add exit configuration, and update this to reject BEGIN and RESOLVE
262
        // if we are not configured to run as an exit
263
        let allow_incoming = &[RelayCmd::BEGIN, RelayCmd::BEGIN_DIR, RelayCmd::RESOLVE];
264

            
265
        // A handler that will process CREATE* requests on channels.
266
        let (create_request_handler, circuit_stream_rx) = CreateRequestHandler::new(
267
            Arc::downgrade(&chanmgr) as Weak<_>,
268
            circ_net_params,
269
            init_key_material.ntor_keys,
270
            Box::new(|| Box::new(RequestFilter::default()) as Box<_>),
271
            allow_incoming,
272
        );
273
        let create_request_handler = Arc::new(create_request_handler);
274

            
275
        // Configure the channel manager to handle CREATE* requests.
276
        //
277
        // We do this once, and can later update its network parameters and keys using the
278
        // `Arc` handle that we store.
279
        // The `ChanMgr` will hold an `Arc<CreateRequestHandler>` and
280
        // the `CreateRequestHandler` will hold a `Weak<ChanMgr>`.
281
        //
282
        // We could technically do something fancier by creating the `ChanMgr` and handler
283
        // inside an `Arc::new_cyclic()` and pass the handler as part of the `ChanMgrConfig`,
284
        // but the code becomes a mess.
285
        chanmgr
286
            .set_create_request_handler(Arc::clone(&create_request_handler))
287
            .context("Failed to set the CREATE* request handler")?;
288

            
289
        // We don't use any custom options on the listening socket.
290
        let listen_options = Default::default();
291

            
292
        // An iterator of `listen()` futures with some extra error handling.
293
        let or_listeners = inert.config.relay.listen.addrs().map(async |addr| {
294
            match runtime.listen(addr, &listen_options).await {
295
                Ok(x) => Some(Ok(x)),
296
                // If we don't support the address family (typically IPv6), only warn.
297
                #[cfg(unix)]
298
                Err(ref e) if e.raw_os_error() == Some(libc::EAFNOSUPPORT) => {
299
                    let message =
300
                        format!("Could not listen at {addr}: address family not supported");
301
                    if addr.is_ipv6() {
302
                        warn!("{message}");
303
                    } else {
304
                        // If we got `EAFNOSUPPORT` for a non-IPv6 address, then warn louder.
305
                        tor_error::warn_report!(e, "{message}");
306
                    }
307
                    None
308
                }
309
                Err(e) => {
310
                    Some(Err(e).with_context(|| format!("Failed to listen at address {addr}")))
311
                }
312
            }
313
        });
314

            
315
        // We await the futures sequentially rather than with something like `join_all` to make
316
        // errors more reproducible.
317
        let or_listeners = {
318
            let mut awaited_listeners = vec![];
319
            for listener in or_listeners {
320
                match listener.await {
321
                    Some(Ok(x)) => awaited_listeners.push(x),
322
                    Some(Err(e)) => return Err(e),
323
                    None => {}
324
                };
325
            }
326
            awaited_listeners
327
        };
328

            
329
        // Typically we would have returned with an error if we failed to listen on an address,
330
        // but we ignore `EAFNOSUPPORT` errors above, so it's possible that all failed with
331
        // `EAFNOSUPPORT` and we ended up here.
332
        if or_listeners.is_empty() {
333
            return Err(anyhow::anyhow!(
334
                "Could not listen at any OR port addresses: {}",
335
                iter_join(", ", inert.config.relay.listen.addrs()),
336
            ));
337
        }
338

            
339
        // TODO DIRMIRROR: Need a config for the DirMirror and should be same as our
340
        // TorRelay one.
341
        let path: PathBuf = PathBuf::from("/dev/null");
342
        let dir_mirror_authorities: AuthorityContacts = Default::default();
343
        let schedule: DownloadScheduleConfig = Default::default();
344
        let tolerance: DirTolerance = Default::default();
345

            
346
        let dir_mirror = DirMirror::new(path, dir_mirror_authorities, schedule, tolerance);
347

            
348
        Ok(Self {
349
            runtime,
350
            memquota,
351
            client,
352
            authorities,
353
            dir_mirror,
354
            chanmgr,
355
            create_request_handler,
356
            keymgr: inert.keymgr,
357
            or_listeners,
358
            circuit_stream_rx,
359
        })
360
    }
361

            
362
    /// Run the actual relay.
363
    ///
364
    /// This only returns if something has gone wrong.
365
    /// Otherwise it runs forever.
366
    pub(crate) async fn run(self) -> anyhow::Result<void::Void> {
367
        let mut task_handles = JoinSet::new();
368

            
369
        // Channel housekeeping task.
370
        task_handles.spawn({
371
            let mut t = crate::tasks::ChannelHouseKeepingTask::new(&self.chanmgr);
372
            async move {
373
                t.start()
374
                    .await
375
                    .context("Failed to run channel house keeping task")
376
            }
377
        });
378

            
379
        // Update the CREATE* request handler when there are new network parameters.
380
        task_handles.spawn({
381
            let create_request_handler = Arc::clone(&self.create_request_handler);
382
            let dir_provider = Arc::clone(self.client.dirmgr());
383
            async {
384
                crate::tasks::channel::update_create_request_handler_netparams(
385
                    create_request_handler,
386
                    dir_provider as Arc<_>,
387
                )
388
                .await
389
                .context("Failed to run create request handler update task")
390
            }
391
        });
392

            
393
        // Listen for new Tor (OR) connections.
394
        task_handles.spawn({
395
            let runtime = self.runtime.clone();
396
            let chanmgr = Arc::clone(&self.chanmgr);
397
            async {
398
                // TODO: Should we give all tasks a `start` method?
399
                crate::tasks::listeners::or_listener(runtime, chanmgr, self.or_listeners)
400
                    .await
401
                    .context("Failed to run OR listener task")
402
            }
403
        });
404

            
405
        // TODO DIRMIRROR: The buffer size here was picked mostly arbitrarily
406
        // (on my new and not-very-busy relay, I noticed bursts of ~5000 BEGIN_DIR requests per second,
407
        // but I'm not sure how representative this is).
408
        //
409
        // We may be able to make this buffer even smaller, assuming the consumer (i.e. DirMirror)
410
        // reads from it quickly enough (presumably it will read from this in a loop,
411
        // dispatching each request to a new task?)
412
        #[allow(clippy::disallowed_methods)]
413
        let (begin_dir_tx, begin_dir_rx) = mpsc::channel::<tor_proto::Result<DataStream>>(4096);
414

            
415
        // Spawn a directory mirror server task, if we are a dir cache.
416
        task_handles.spawn(async {
417
            // TODO DIRMIRROR: it would be nicer if serve() returned
418
            // Result<Void, _> to statically prove that it indeed never
419
            // returns with a non-error.
420
            // Plus, if we do that, we can simplify this invocation,
421
            // because we won't need the anyhow! error below.
422
            self.dir_mirror.serve(begin_dir_rx).await?;
423
            Err(anyhow::anyhow!("dir mirror exited"))
424
        });
425

            
426
        let runtime = self.runtime.clone();
427
        // Listen for new Tor streams
428
        task_handles.spawn(
429
            // TODO: Should we give all tasks a `start` method?
430
            crate::stream::handle_incoming_streams(runtime, begin_dir_tx, self.circuit_stream_rx),
431
        );
432

            
433
        // Channel used to ask the descriptor publisher to rebuild and re-publish the descriptor.
434
        let (desc_command_tx, desc_command_rx) = crate::tasks::descriptor::new_command_channel();
435
        let (crypto_command_tx, crypto_command_rx) = crate::tasks::crypto::new_command_channel();
436

            
437
        // Start the crypto task.
438
        task_handles.spawn({
439
            let reactor = crate::tasks::crypto::Reactor::new(
440
                self.runtime.clone(),
441
                self.chanmgr.clone(),
442
                self.create_request_handler.clone(),
443
                self.keymgr,
444
                self.client.dirmgr().clone(),
445
                desc_command_tx,
446
                crypto_command_rx,
447
            )?;
448
            async {
449
                reactor
450
                    .run()
451
                    .await
452
                    .context("Failed to run key rotation task")
453
            }
454
        });
455

            
456
        // Build and publish the relay's own descriptor.
457
        task_handles.spawn({
458
            let netdir = Arc::clone(self.client.dirmgr()) as Arc<_>;
459
            let authorities = self.authorities;
460
            async move {
461
                crate::tasks::RelayDescriptorPublisherTask::new(
462
                    &self.runtime,
463
                    netdir,
464
                    authorities,
465
                    crypto_command_tx,
466
                    desc_command_rx,
467
                )
468
                .context("Failed to create descriptor publisher task")?
469
                .start()
470
                .await
471
                .context("Failed to run descriptor publisher task")
472
            }
473
        });
474

            
475
        // Launch client tasks.
476
        //
477
        // We need to hold on to these handles until the relay stops, otherwise dropping these
478
        // handles would stop the background tasks.
479
        //
480
        // These are `tor_rtcompat::scheduler::TaskHandle`s, which don't notify us if they
481
        // stop/crash.
482
        //
483
        // TODO: Whose responsibility is it to ensure that these background tasks don't crash?
484
        // Should we have a way of monitoring these tasks? Or should the circuit manager re-launch
485
        // crashed tasks?
486
        let _client_task_handles = self.client.launch_background_tasks();
487

            
488
        // TODO: More tasks will be spawned here.
489

            
490
        // Now that background tasks are started, bootstrap the client.
491
        self.client
492
            .bootstrap()
493
            .await
494
            .context("Failed to bootstrap the relay's client")?;
495

            
496
        // We block until facism is erradicated or a task ends which means the relay will shutdown
497
        // and facism will have one more chance.
498
        let void = task_handles
499
            .join_next()
500
            .await
501
            .context("Relay task set is empty")?
502
            .context("Relay task join failed")?
503
            .context("Relay task stopped unexpectedly")?;
504

            
505
        // We can never get here since a `Void` cannot be constructed.
506
        void::unreachable(void);
507
    }
508
}