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, 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
    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
        cfg_sources,
100
        config,
101
        client_config,
102
    ))?;
103

            
104
    Ok(())
105
}
106

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

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

            
132
    let client_builder = TorClient::with_runtime(runtime.clone())
133
        .config(client_config)
134
        .bootstrap_behavior(OnDemand);
135
    let client = client_builder.create_unbootstrapped_async().await?;
136

            
137
    #[allow(unused_mut)]
138
    let mut reconfigurable_modules: Vec<Arc<dyn reload_cfg::ReconfigurableModule>> = vec![
139
        Arc::new(client.clone()),
140
        Arc::new(reload_cfg::Application::new(arti_config.clone())),
141
    ];
142

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

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

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

            
185
    let mut proxy: Vec<PinnedFuture<Result<()>>> = Vec::new();
186
    let mut ports = Vec::new();
187
    if !socks_listen.is_empty() {
188
        let runtime = runtime.clone();
189
        let client = client.isolated_client();
190
        let socks_listen = socks_listen.clone();
191
        #[cfg(feature = "http-connect")]
192
        let listener_type = "SOCKS+HTTP";
193
        #[cfg(not(feature = "http-connect"))]
194
        let listener_type = "SOCKS";
195

            
196
        let stream_proxy = proxy::bind_proxy(runtime, client, socks_listen, rpc_mgr)
197
            .await
198
            .with_context(|| format!("Unable to launch {listener_type} proxy"))?;
199
        let port_info = stream_proxy.port_info()?;
200

            
201
        ports.extend(port_info);
202

            
203
        let failure_message = format!("{listener_type} proxy died unexpectedly");
204
        let proxy_future = stream_proxy
205
            .run_proxy()
206
            .map(|future_result| future_result.context(failure_message));
207
        proxy.push(Box::pin(proxy_future));
208
    }
209

            
210
    #[cfg(feature = "dns-proxy")]
211
    if !dns_listen.is_empty() {
212
        let runtime = runtime.clone();
213
        let client = client.isolated_client();
214
        let dns_proxy = dns::bind_dns_resolver(runtime, client, dns_listen)
215
            .await
216
            .context("Unable to launch DNS proxy")?;
217
        ports.extend(dns_proxy.port_info().context("Unable to find DNS ports")?);
218
        let proxy_future = dns_proxy
219
            .run_dns_proxy()
220
            .map(|future_result| future_result.context("DNS proxy died unexpectedly"));
221
        proxy.push(Box::pin(proxy_future));
222
    }
223

            
224
    #[cfg(not(feature = "dns-proxy"))]
225
    if !dns_listen.is_empty() {
226
        warn!(
227
            "Tried to specify a DNS proxy address, but Arti was built without dns-proxy support."
228
        );
229
        return Ok(());
230
    }
231

            
232
    if proxy.is_empty() {
233
        if !launched_onion_svc {
234
            // TODO: rename "socks_listen" to "proxy_listen", preserving compat, once http-connect is stable.
235
            warn!(
236
                "No proxy address set; \
237
                specify -p PORT (to override `socks_listen`) \
238
                or -d PORT (to override `dns_listen`). \
239
                Alternatively, use the `socks_listen` or `dns_listen` configuration options."
240
            );
241
            return Ok(());
242
        } else {
243
            // Push a dummy future to appease future::select_all,
244
            // which expects a non-empty list
245
            proxy.push(Box::pin(futures::future::pending()));
246
        }
247
    }
248

            
249
    cfg_if::cfg_if! {
250
        if #[cfg(feature="rpc")] {
251
            if let Some(rpc_state_sender) = &mut rpc_state_sender {
252
                rpc_state_sender.set_stream_listeners(&ports[..]);
253
            }
254
        }
255
    }
256

            
257
    {
258
        let port_info = port_info::PortInfo { ports };
259
        let port_info_file = arti_config
260
            .storage()
261
            .port_info_file
262
            .path(&path_resolver)
263
            .context("Can't find path for port_info_file")?;
264
        if port_info_file.to_str() != Some("") {
265
            port_info.write_to_file(&fs_mistrust, &port_info_file)?;
266
        }
267
    }
268

            
269
    let proxy = futures::future::select_all(proxy).map(|(finished, _index, _others)| finished);
270
    futures::select!(
271
        r = exit::wait_for_ctrl_c().fuse()
272
            => r.context("waiting for termination signal"),
273
        r = proxy.fuse()
274
            => r,
275
        r = async {
276
            client.bootstrap().await?;
277
            if !socks_listen.is_empty() {
278
                info!("Sufficiently bootstrapped; proxy now functional.");
279
            } else {
280
                info!("Sufficiently bootstrapped.");
281
            }
282
            futures::future::pending::<Result<()>>().await
283
        }.fuse()
284
            => r.context("bootstrap"),
285
    )?;
286

            
287
    // The modules can be dropped now, because we are exiting.
288
    drop(reconfigurable_modules);
289

            
290
    Ok(())
291
}