1
//! The `proxy` subcommand.
2

            
3
use std::sync::Arc;
4

            
5
use anyhow::{Context, Result};
6
use cfg_if::cfg_if;
7
use clap::ArgMatches;
8
#[allow(unused)]
9
use tor_config_path::CfgPathResolver;
10
use tracing::{info, instrument, warn};
11

            
12
use arti_client::TorClientConfig;
13
use tor_config::{ConfigurationSources, Listen};
14
use tor_rtcompat::ToplevelRuntime;
15

            
16
#[cfg(feature = "dns-proxy")]
17
use crate::dns;
18
use crate::{
19
    ArtiConfig, TorClient, exit, process,
20
    proxy::{self, ListenProtocols, port_info},
21
    reload_cfg,
22
};
23

            
24
#[cfg(feature = "rpc")]
25
use crate::rpc;
26

            
27
#[cfg(feature = "onion-service-service")]
28
use crate::onion_proxy;
29

            
30
/// Shorthand for a boxed and pinned Future.
31
type PinnedFuture<T> = std::pin::Pin<Box<dyn futures::Future<Output = T>>>;
32

            
33
/// Run the `proxy` subcommand.
34
#[instrument(skip_all, level = "trace")]
35
pub(crate) fn run<R: ToplevelRuntime>(
36
    runtime: R,
37
    proxy_matches: &ArgMatches,
38
    cfg_sources: ConfigurationSources,
39
    config: ArtiConfig,
40
    client_config: TorClientConfig,
41
) -> Result<()> {
42
    // Override configured listen addresses from the command line.
43
    // This implies listening on localhost ports.
44

            
45
    // TODO: Parse a string rather than calling new_localhost.
46
    let socks_listen = match proxy_matches.get_one::<u16>("socks-port") {
47
        Some(p) => Listen::new_localhost(*p),
48
        None => config.proxy().socks_listen.clone(),
49
    };
50

            
51
    // TODO: Parse a string rather than calling new_localhost.
52
    let dns_listen = match proxy_matches.get_one::<u16>("dns-port") {
53
        Some(p) => Listen::new_localhost(*p),
54
        None => config.proxy().dns_listen.clone(),
55
    };
56

            
57
    if !socks_listen.is_empty() {
58
        info!(
59
            "Starting Arti {} in proxy mode on {} ...",
60
            env!("CARGO_PKG_VERSION"),
61
            socks_listen
62
        );
63
    }
64

            
65
    if let Some(listen) = {
66
        // https://github.com/metrics-rs/metrics/issues/567
67
        config
68
            .metrics
69
            .prometheus
70
            .listen
71
            .single_address_legacy()
72
            .context("can only listen on a single address for Prometheus metrics")?
73
    } {
74
        cfg_if! {
75
            if #[cfg(feature = "metrics")] {
76
                metrics_exporter_prometheus::PrometheusBuilder::new()
77
                    .with_http_listener(listen)
78
                    .install()
79
                    .with_context(|| format!(
80
                        "set up Prometheus metrics exporter on {listen}"
81
                    ))?;
82
                info!("Arti Prometheus metrics export scraper endpoint http://{listen}");
83
            } else {
84
                return Err(anyhow::anyhow!(
85
        "`metrics.prometheus.listen` config set but `metrics` cargo feature compiled out in `arti` crate"
86
                ));
87
            }
88
        }
89
    }
90

            
91
    #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
92
    process::use_max_file_limit(&config);
93

            
94
    let rt_copy = runtime.clone();
95
    rt_copy.block_on(run_proxy(
96
        runtime,
97
        socks_listen,
98
        dns_listen,
99
        config.proxy().protocols(),
100
        cfg_sources,
101
        config,
102
        client_config,
103
    ))?;
104

            
105
    Ok(())
106
}
107

            
108
/// Run the main loop of the proxy.
109
///
110
/// # Panics
111
///
112
/// Currently, might panic if things go badly enough wrong
113
#[cfg_attr(feature = "experimental-api", visibility::make(pub))]
114
#[cfg_attr(docsrs, doc(cfg(feature = "experimental-api")))]
115
#[instrument(skip_all, level = "trace")]
116
async fn run_proxy<R: ToplevelRuntime>(
117
    runtime: R,
118
    socks_listen: Listen,
119
    dns_listen: Listen,
120
    protocols: ListenProtocols,
121
    config_sources: ConfigurationSources,
122
    arti_config: ArtiConfig,
123
    client_config: TorClientConfig,
124
) -> Result<()> {
125
    // Using OnDemand arranges that, while we are bootstrapping, incoming connections wait
126
    // for bootstrap to complete, rather than getting errors.
127
    use arti_client::BootstrapBehavior;
128
    use futures::FutureExt;
129

            
130
    // TODO: We may instead want to provide a way to get these items out of TorClient.
131
    let fs_mistrust = client_config.fs_mistrust().clone();
132
    let path_resolver: CfgPathResolver = AsRef::<CfgPathResolver>::as_ref(&client_config).clone();
133

            
134
    let defer_bootstrap = arti_config.application().defer_bootstrap;
135

            
136
    let bootstrap_behavior = match defer_bootstrap {
137
        true => BootstrapBehavior::Manual,
138
        false => BootstrapBehavior::OnDemand,
139
    };
140

            
141
    let (cfg_mgr, cfg_watcher_task) =
142
        reload_cfg::CfgMgr::new(runtime.clone(), config_sources, &arti_config, vec![])?;
143

            
144
    let client_builder = TorClient::with_runtime(runtime.clone())
145
        .config(client_config)
146
        .bootstrap_behavior(bootstrap_behavior);
147
    let client = client_builder.create_unbootstrapped_async().await?;
148

            
149
    let launchable_client = Arc::new(reload_cfg::LaunchableTorClient::new(
150
        Arc::clone(&client),
151
        arti_config.application(),
152
    ));
153

            
154
    #[allow(unused_mut)]
155
    let mut reconfigurable_modules: Vec<Arc<dyn reload_cfg::ReconfigurableModule>> = vec![
156
        Arc::clone(&launchable_client) as _,
157
        Arc::new(reload_cfg::Application::new(arti_config.clone())),
158
    ];
159

            
160
    cfg_if::cfg_if! {
161
        if #[cfg(feature = "onion-service-service")] {
162
            let have_onion_svc = if defer_bootstrap {
163
                let onion_services = onion_proxy::ProxySet::new_deferred(Arc::clone(&client));
164
                reconfigurable_modules.push(Arc::new(onion_services));
165
                arti_config.onion_services.values().any(|c| *c.svc_cfg.enabled())
166
            } else {
167
                let onion_services =
168
                    onion_proxy::ProxySet::launch_new(Arc::clone(&client), arti_config.onion_services.clone())?;
169
                let have_onion_svc = !onion_services.is_empty();
170
                reconfigurable_modules.push(Arc::new(onion_services));
171
                have_onion_svc
172
            };
173
        } else {
174
            let have_onion_svc = false;
175
        }
176
    };
177

            
178
    // The add_module function will use references here
179
    // to prevent the task spawned by watch_for_config_changes from
180
    // keeping these modules alive after this function exits.
181
    //
182
    // NOTE: reconfigurable_modules stores the only strong references to these modules,
183
    // so we must keep that variable alive until the end of the function
184
    reconfigurable_modules
185
        .iter()
186
        .try_for_each(|m| cfg_watcher_task.add_module(m))?;
187

            
188
    cfg_watcher_task.launch()?;
189

            
190
    cfg_if::cfg_if! {
191
        if #[cfg(feature = "rpc")] {
192
            let rpc_data = rpc::launch_rpc_mgr(
193
                &runtime,
194
                &arti_config.rpc,
195
                &path_resolver,
196
                &fs_mistrust,
197
                client.clone(),
198
                launchable_client.clone(),
199
                cfg_mgr.clone(),
200
            )
201
            .await?;
202
            let (rpc_mgr, mut rpc_state_sender) = rpc_data
203
                .map(|d| (d.rpc_mgr, d.rpc_state_sender))
204
                .unzip();
205
        } else {
206
            let rpc_mgr = None;
207
        }
208
    }
209

            
210
    // The options that we'll use for our listening proxy sockets.
211
    let mut listen_options = tor_rtcompat::TcpListenOptions::builder();
212
    listen_options
213
        .common()
214
        .send_buffer_size(Some(arti_config.proxy().socket_send_buf_size.as_usize()))
215
        .recv_buffer_size(Some(arti_config.proxy().socket_recv_buf_size.as_usize()));
216
    let listen_options = listen_options.build()?;
217

            
218
    let mut proxy: Vec<PinnedFuture<Result<()>>> = Vec::new();
219
    let mut ports = Vec::new();
220
    if !socks_listen.is_empty() {
221
        let runtime = runtime.clone();
222
        let client = client.isolated_client();
223
        let socks_listen = socks_listen.clone();
224
        let listener_type = protocols.to_string();
225

            
226
        let stream_proxy = proxy::bind_proxy(
227
            runtime,
228
            client,
229
            socks_listen,
230
            listen_options,
231
            protocols,
232
            rpc_mgr,
233
        )
234
        .await
235
        .with_context(|| format!("Unable to launch {listener_type} proxy"))?;
236
        let port_info = stream_proxy.port_info()?;
237

            
238
        ports.extend(port_info);
239

            
240
        let failure_message = format!("{listener_type} proxy died unexpectedly");
241
        let proxy_future = stream_proxy
242
            .run_proxy()
243
            .map(|future_result| future_result.context(failure_message));
244
        proxy.push(Box::pin(proxy_future));
245
    }
246

            
247
    #[cfg(feature = "dns-proxy")]
248
    if !dns_listen.is_empty() {
249
        let runtime = runtime.clone();
250
        let client = client.isolated_client();
251
        let dns_proxy = dns::bind_dns_resolver(runtime, client, dns_listen)
252
            .await
253
            .context("Unable to launch DNS proxy")?;
254
        ports.extend(dns_proxy.port_info().context("Unable to find DNS ports")?);
255
        let proxy_future = dns_proxy
256
            .run_dns_proxy()
257
            .map(|future_result| future_result.context("DNS proxy died unexpectedly"));
258
        proxy.push(Box::pin(proxy_future));
259
    }
260

            
261
    #[cfg(not(feature = "dns-proxy"))]
262
    if !dns_listen.is_empty() {
263
        warn!(
264
            "Tried to specify a DNS proxy address, but Arti was built without dns-proxy support."
265
        );
266
        return Ok(());
267
    }
268

            
269
    if proxy.is_empty() {
270
        if !have_onion_svc {
271
            // TODO: rename "socks_listen" to "proxy_listen", preserving compat, once http-connect is stable.
272
            warn!(
273
                "No proxy address set; \
274
                specify -p PORT (to override `socks_listen`) \
275
                or -d PORT (to override `dns_listen`). \
276
                Alternatively, use the `socks_listen` or `dns_listen` configuration options."
277
            );
278
            return Ok(());
279
        } else {
280
            // Push a dummy future to appease future::select_all,
281
            // which expects a non-empty list
282
            proxy.push(Box::pin(futures::future::pending()));
283
        }
284
    }
285

            
286
    cfg_if::cfg_if! {
287
        if #[cfg(feature="rpc")] {
288
            if let Some(rpc_state_sender) = &mut rpc_state_sender {
289
                rpc_state_sender.set_stream_listeners(&ports[..]);
290
            }
291
        }
292
    }
293

            
294
    {
295
        let port_info = port_info::PortInfo { ports };
296
        let port_info_file = arti_config
297
            .storage()
298
            .port_info_file
299
            .path(&path_resolver)
300
            .context("Can't find path for port_info_file")?;
301
        if port_info_file.to_str() != Some("") {
302
            port_info.write_to_file(&fs_mistrust, &port_info_file)?;
303
        }
304
    }
305

            
306
    let proxy = futures::future::select_all(proxy).map(|(finished, _index, _others)| finished);
307
    futures::select!(
308
        r = exit::wait_for_ctrl_c().fuse()
309
            => r.context("waiting for termination signal"),
310
        r = proxy.fuse()
311
            => r,
312
        r = async {
313
            if defer_bootstrap {
314
                info!("Bootstrapping deferred.");
315
            } else {
316
                client.bootstrap().await?;
317
                if !socks_listen.is_empty() {
318
                    info!("Sufficiently bootstrapped; proxy now functional.");
319
                } else {
320
                    info!("Sufficiently bootstrapped.");
321
                }
322
            }
323
            futures::future::pending::<Result<()>>().await
324
        }.fuse()
325
            => r.context("bootstrap"),
326
    )?;
327

            
328
    // The modules and CfgMgr can be dropped now, because we are exiting.
329
    // (We drop them explicitly to make sure that they were not dropped
330
    // accidentally before.)
331
    drop(reconfigurable_modules);
332
    drop(cfg_mgr);
333

            
334
    Ok(())
335
}