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
#[allow(clippy::cognitive_complexity)]
36
pub(crate) fn run<R: ToplevelRuntime>(
37
    runtime: R,
38
    proxy_matches: &ArgMatches,
39
    cfg_sources: ConfigurationSources,
40
    config: ArtiConfig,
41
    client_config: TorClientConfig,
42
) -> Result<()> {
43
    // Override configured listen addresses from the command line.
44
    // This implies listening on localhost ports.
45

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

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

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

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

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

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

            
106
    Ok(())
107
}
108

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

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

            
135
    let client_builder = TorClient::with_runtime(runtime.clone())
136
        .config(client_config)
137
        .bootstrap_behavior(OnDemand);
138
    let client = client_builder.create_unbootstrapped_async().await?;
139

            
140
    #[allow(unused_mut)]
141
    let mut reconfigurable_modules: Vec<Arc<dyn reload_cfg::ReconfigurableModule>> = vec![
142
        Arc::clone(&client) as _,
143
        Arc::new(reload_cfg::Application::new(arti_config.clone())),
144
    ];
145

            
146
    cfg_if::cfg_if! {
147
        if #[cfg(feature = "onion-service-service")] {
148
            let onion_services =
149
                onion_proxy::ProxySet::launch_new(Arc::clone(&client), arti_config.onion_services.clone())?;
150
            let launched_onion_svc = !onion_services.is_empty();
151
            reconfigurable_modules.push(Arc::new(onion_services));
152
        } else {
153
            let launched_onion_svc = false;
154
        }
155
    };
156

            
157
    // We weak references here to prevent the thread spawned by watch_for_config_changes from
158
    // keeping these modules alive after this function exits.
159
    //
160
    // NOTE: reconfigurable_modules stores the only strong references to these modules,
161
    // so we must keep the variable alive until the end of the function
162
    let weak_modules = reconfigurable_modules.iter().map(Arc::downgrade).collect();
163
    reload_cfg::watch_for_config_changes(
164
        client.runtime(),
165
        config_sources,
166
        &arti_config,
167
        weak_modules,
168
    )?;
169

            
170
    cfg_if::cfg_if! {
171
        if #[cfg(feature = "rpc")] {
172
            let rpc_data = rpc::launch_rpc_mgr(
173
                &runtime,
174
                &arti_config.rpc,
175
                &path_resolver,
176
                &fs_mistrust,
177
                client.clone(),
178
            )
179
            .await?;
180
            let (rpc_mgr, mut rpc_state_sender) = rpc_data
181
                .map(|d| (d.rpc_mgr, d.rpc_state_sender))
182
                .unzip();
183
        } else {
184
            let rpc_mgr = None;
185
        }
186
    }
187

            
188
    // The options that we'll use for our listening proxy sockets.
189
    let mut listen_options = tor_rtcompat::TcpListenOptions::builder();
190
    listen_options
191
        .common()
192
        .send_buffer_size(Some(arti_config.proxy().socket_send_buf_size.as_usize()))
193
        .recv_buffer_size(Some(arti_config.proxy().socket_recv_buf_size.as_usize()));
194
    let listen_options = listen_options.build()?;
195

            
196
    let mut proxy: Vec<PinnedFuture<Result<()>>> = Vec::new();
197
    let mut ports = Vec::new();
198
    if !socks_listen.is_empty() {
199
        let runtime = runtime.clone();
200
        let client = client.isolated_client();
201
        let socks_listen = socks_listen.clone();
202
        let listener_type = protocols.to_string();
203

            
204
        let stream_proxy = proxy::bind_proxy(
205
            runtime,
206
            client,
207
            socks_listen,
208
            listen_options,
209
            protocols,
210
            rpc_mgr,
211
        )
212
        .await
213
        .with_context(|| format!("Unable to launch {listener_type} proxy"))?;
214
        let port_info = stream_proxy.port_info()?;
215

            
216
        ports.extend(port_info);
217

            
218
        let failure_message = format!("{listener_type} proxy died unexpectedly");
219
        let proxy_future = stream_proxy
220
            .run_proxy()
221
            .map(|future_result| future_result.context(failure_message));
222
        proxy.push(Box::pin(proxy_future));
223
    }
224

            
225
    #[cfg(feature = "dns-proxy")]
226
    if !dns_listen.is_empty() {
227
        let runtime = runtime.clone();
228
        let client = client.isolated_client();
229
        let dns_proxy = dns::bind_dns_resolver(runtime, client, dns_listen)
230
            .await
231
            .context("Unable to launch DNS proxy")?;
232
        ports.extend(dns_proxy.port_info().context("Unable to find DNS ports")?);
233
        let proxy_future = dns_proxy
234
            .run_dns_proxy()
235
            .map(|future_result| future_result.context("DNS proxy died unexpectedly"));
236
        proxy.push(Box::pin(proxy_future));
237
    }
238

            
239
    #[cfg(not(feature = "dns-proxy"))]
240
    if !dns_listen.is_empty() {
241
        warn!(
242
            "Tried to specify a DNS proxy address, but Arti was built without dns-proxy support."
243
        );
244
        return Ok(());
245
    }
246

            
247
    if proxy.is_empty() {
248
        if !launched_onion_svc {
249
            // TODO: rename "socks_listen" to "proxy_listen", preserving compat, once http-connect is stable.
250
            warn!(
251
                "No proxy address set; \
252
                specify -p PORT (to override `socks_listen`) \
253
                or -d PORT (to override `dns_listen`). \
254
                Alternatively, use the `socks_listen` or `dns_listen` configuration options."
255
            );
256
            return Ok(());
257
        } else {
258
            // Push a dummy future to appease future::select_all,
259
            // which expects a non-empty list
260
            proxy.push(Box::pin(futures::future::pending()));
261
        }
262
    }
263

            
264
    cfg_if::cfg_if! {
265
        if #[cfg(feature="rpc")] {
266
            if let Some(rpc_state_sender) = &mut rpc_state_sender {
267
                rpc_state_sender.set_stream_listeners(&ports[..]);
268
            }
269
        }
270
    }
271

            
272
    {
273
        let port_info = port_info::PortInfo { ports };
274
        let port_info_file = arti_config
275
            .storage()
276
            .port_info_file
277
            .path(&path_resolver)
278
            .context("Can't find path for port_info_file")?;
279
        if port_info_file.to_str() != Some("") {
280
            port_info.write_to_file(&fs_mistrust, &port_info_file)?;
281
        }
282
    }
283

            
284
    let proxy = futures::future::select_all(proxy).map(|(finished, _index, _others)| finished);
285
    futures::select!(
286
        r = exit::wait_for_ctrl_c().fuse()
287
            => r.context("waiting for termination signal"),
288
        r = proxy.fuse()
289
            => r,
290
        r = async {
291
            client.bootstrap().await?;
292
            if !socks_listen.is_empty() {
293
                info!("Sufficiently bootstrapped; proxy now functional.");
294
            } else {
295
                info!("Sufficiently bootstrapped.");
296
            }
297
            futures::future::pending::<Result<()>>().await
298
        }.fuse()
299
            => r.context("bootstrap"),
300
    )?;
301

            
302
    // The modules can be dropped now, because we are exiting.
303
    drop(reconfigurable_modules);
304

            
305
    Ok(())
306
}