tor_ptmgr/
managed.rs

1//! Infrastructure required to support managed PTs.
2
3use crate::config::{ManagedTransportOptions, TransportOptions};
4use crate::err;
5use crate::err::PtError;
6use crate::ipc::{
7    PluggableClientTransport, PluggableTransport, PtClientParameters, PtCommonParameters,
8    sealed::PluggableTransportPrivate,
9};
10use crate::{PtClientMethod, PtSharedState};
11use futures::channel::mpsc::UnboundedReceiver;
12use futures::stream::FuturesUnordered;
13use futures::{FutureExt, StreamExt, select};
14use oneshot_fused_workaround as oneshot;
15use std::collections::{HashMap, HashSet};
16use std::future::Future;
17use std::path::{Path, PathBuf};
18use std::pin::Pin;
19use std::sync::{Arc, RwLock};
20use tor_chanmgr::ProxyProtocol;
21use tor_config_path::CfgPathResolver;
22use tor_error::internal;
23use tor_linkspec::PtTransportName;
24use tor_rtcompat::Runtime;
25use tracing::{debug, warn};
26
27/// A message to the `PtReactor`.
28pub(crate) enum PtReactorMessage {
29    /// Notify the reactor that the currently configured set of PTs has changed.
30    Reconfigured,
31    /// Ask the reactor to spawn a pluggable transport binary.
32    Spawn {
33        /// Spawn a binary to provide this PT.
34        pt: PtTransportName,
35        /// Notify the result via this channel.
36        result: oneshot::Sender<err::Result<PtClientMethod>>,
37    },
38}
39
40/// The result of a spawn attempt: the list of transports the spawned binary covers, and the result.
41type SpawnResult = (Vec<PtTransportName>, err::Result<PluggableClientTransport>);
42
43/// Background reactor to handle managing pluggable transport binaries.
44pub(crate) struct PtReactor<R> {
45    /// Runtime.
46    rt: R,
47    /// Currently running pluggable transport binaries.
48    running: Vec<PluggableClientTransport>,
49    /// A map of asked-for transports.
50    ///
51    /// If a transport name has an entry, we will append any additional requests for that entry.
52    /// If no entry is present, we will start a request.
53    requests: HashMap<PtTransportName, Vec<oneshot::Sender<err::Result<PtClientMethod>>>>,
54    /// FuturesUnordered that spawned tasks get pushed on to.
55    ///
56    /// WARNING: This MUST always contain one "will never resolve" future!
57    spawning: FuturesUnordered<Pin<Box<dyn Future<Output = SpawnResult> + Send>>>,
58    /// State for the corresponding PtMgr.
59    state: Arc<RwLock<PtSharedState>>,
60    /// PtMgr channel.
61    /// (Unbounded so that we can reconfigure without blocking: we're unlikely to have the reactor
62    /// get behind.)
63    rx: UnboundedReceiver<PtReactorMessage>,
64    /// State directory.
65    state_dir: PathBuf,
66    /// Path resolver for configuration files.
67    path_resolver: Arc<CfgPathResolver>,
68}
69
70impl<R: Runtime> PtReactor<R> {
71    /// Make a new reactor.
72    pub(crate) fn new(
73        rt: R,
74        state: Arc<RwLock<PtSharedState>>,
75        rx: UnboundedReceiver<PtReactorMessage>,
76        state_dir: PathBuf,
77        path_resolver: Arc<CfgPathResolver>,
78    ) -> Self {
79        let spawning = FuturesUnordered::new();
80        spawning.push(Box::pin(futures::future::pending::<SpawnResult>())
81            as Pin<Box<dyn Future<Output = _> + Send>>);
82        Self {
83            rt,
84            running: vec![],
85            requests: Default::default(),
86            spawning,
87            state,
88            rx,
89            state_dir,
90            path_resolver,
91        }
92    }
93
94    /// Called when a spawn request completes.
95    #[allow(clippy::needless_pass_by_value)]
96    fn handle_spawned(
97        &mut self,
98        covers: Vec<PtTransportName>,
99        result: err::Result<PluggableClientTransport>,
100    ) {
101        match result {
102            Err(e) => {
103                warn!("Spawning PT for {:?} failed: {}", covers, e);
104                // Go and tell all the transports about the bad news.
105                let senders = covers
106                    .iter()
107                    .flat_map(|x| self.requests.remove(x))
108                    .flatten();
109                for sender in senders {
110                    // We don't really care if the sender went away.
111                    let _ = sender.send(Err(e.clone()));
112                }
113            }
114            Ok(pt) => {
115                let mut state = self.state.write().expect("ptmgr state poisoned");
116                for (transport, method) in pt.transport_methods() {
117                    state
118                        .managed_cmethods
119                        .insert(transport.clone(), method.clone());
120                    for sender in self.requests.remove(transport).into_iter().flatten() {
121                        let _ = sender.send(Ok(method.clone()));
122                    }
123                }
124
125                let requested: HashSet<_> = covers.iter().collect();
126                let found: HashSet<_> = pt.transport_methods().keys().collect();
127                if requested != found {
128                    warn!(
129                        "Bug: PT {} succeeded, but did not give the same transports we asked for. ({:?} vs {:?})",
130                        pt.identifier(),
131                        found,
132                        requested
133                    );
134                }
135                self.running.push(pt);
136            }
137        }
138    }
139
140    /// Called to remove a pluggable transport from the shared state.
141    fn remove_pt(&self, pt: PluggableClientTransport) {
142        let mut state = self.state.write().expect("ptmgr state poisoned");
143        for transport in pt.transport_methods().keys() {
144            state.managed_cmethods.remove(transport);
145        }
146        // to satisfy clippy, and make it clear that this is a desired side-effect: doing this
147        // shuts down the PT (asynchronously).
148        drop(pt);
149    }
150
151    /// Run one step of the reactor. Returns true if the reactor should terminate.
152    pub(crate) async fn run_one_step(&mut self) -> err::Result<bool> {
153        use futures::future::Either;
154
155        let mut all_next_messages = self
156            .running
157            .iter_mut()
158            .map(|pt| pt.next_message())
159            .collect::<Vec<_>>();
160
161        // We can't construct a select_all if all_next_messages is empty.
162        let mut next_message = if all_next_messages.is_empty() {
163            Either::Left(futures::future::pending())
164        } else {
165            Either::Right(futures::future::select_all(all_next_messages.iter_mut()).fuse())
166        };
167
168        select! {
169            (result, idx, _) = next_message => {
170                drop(all_next_messages); // no idea why NLL doesn't just infer this but sure
171
172                match result {
173                    Ok(m) => {
174                        // FIXME(eta): We should forward the Status messages onto API consumers.
175                        debug!("PT {} message: {:?}", self.running[idx].identifier(), m);
176                    },
177                    Err(e) => {
178                        warn!("PT {} quit: {:?}", self.running[idx].identifier(), e);
179                        let pt = self.running.remove(idx);
180                        self.remove_pt(pt);
181                    }
182                }
183            },
184            spawn_result = self.spawning.next() => {
185                drop(all_next_messages);
186                // See the Warning in this field's documentation.
187                let (covers, result) = spawn_result.expect("self.spawning should never dry up");
188                self.handle_spawned(covers, result);
189            }
190            internal = self.rx.next() => {
191                drop(all_next_messages);
192
193                match internal {
194                    Some(PtReactorMessage::Reconfigured) => {},
195                    Some(PtReactorMessage::Spawn { pt, result }) => {
196                        // Make sure we don't already have a running request.
197                        if let Some(requests) = self.requests.get_mut(&pt) {
198                            requests.push(result);
199                            return Ok(false);
200                        }
201                        // Make sure we don't already have a binary for this PT.
202                        for rpt in self.running.iter() {
203                            if let Some(cmethod) = rpt.transport_methods().get(&pt) {
204                                let _ = result.send(Ok(cmethod.clone()));
205                                return Ok(false);
206                            }
207                        }
208                        // We don't, so time to spawn one.
209                        let (config, outbound_proxy) = {
210                            let state = self.state.read().expect("ptmgr state poisoned");
211                            (state.configured.get(&pt).cloned(), state.outbound_proxy.clone())
212                        };
213
214                        let Some(config) = config else {
215                            let _ = result.send(Err(PtError::UnconfiguredTransportDueToConcurrentReconfiguration));
216                            return Ok(false);
217                        };
218
219                        let TransportOptions::Managed(config) = config else {
220                            let _ = result.send(Err(internal!("Tried to spawn an unmanaged transport").into()));
221                            return Ok(false);
222                        };
223
224                        // Keep track of the request, and also fill holes in other protocols so
225                        // we don't try and run another spawn request for those.
226                        self.requests.entry(pt).or_default().push(result);
227                        for proto in config.protocols.iter() {
228                            self.requests.entry(proto.clone()).or_default();
229                        }
230
231                        // Add the spawn future to our pile of them.
232                        let spawn_fut = Box::pin(
233                            spawn_from_config(
234                                self.rt.clone(),
235                                self.state_dir.clone(),
236                                config.clone(),
237                                Arc::clone(&self.path_resolver),
238                                outbound_proxy,
239                            )
240                            .map(|result| (config.protocols, result))
241                        );
242                        self.spawning.push(spawn_fut);
243                    },
244                    None => return Ok(true)
245                }
246            }
247        }
248        Ok(false)
249    }
250}
251
252/// Spawn a managed `PluggableTransport` using a `ManagedTransportOptions`.
253async fn spawn_from_config<R: Runtime>(
254    rt: R,
255    state_dir: PathBuf,
256    cfg: ManagedTransportOptions,
257    path_resolver: Arc<CfgPathResolver>,
258    outbound_proxy: Option<ProxyProtocol>,
259) -> Result<PluggableClientTransport, PtError> {
260    // FIXME(eta): I really think this expansion should happen at builder validation time...
261
262    let cfg_path = cfg.path;
263
264    let binary_path = cfg_path
265        .path(&path_resolver)
266        .map_err(|e| PtError::PathExpansionFailed {
267            path: cfg_path.clone(),
268            error: e,
269        })?;
270
271    let filename = pt_identifier_as_path(&binary_path)?;
272
273    // HACK(eta): Currently the state directory is named after the PT binary name. Maybe we should
274    //            invent a better way of doing this?
275    let new_state_dir = state_dir.join(filename);
276    std::fs::create_dir_all(&new_state_dir).map_err(|e| PtError::StatedirCreateFailed {
277        path: new_state_dir.clone(),
278        error: Arc::new(e),
279    })?;
280
281    // FIXME(eta): make the rest of these parameters configurable
282    let pt_common_params = PtCommonParameters::builder()
283        .state_location(new_state_dir)
284        .build()
285        .expect("PtCommonParameters constructed incorrectly");
286
287    // The PT spec defines `TOR_PT_PROXY` as a URI, so we only render the
288    // structured `ProxyProtocol` to a string at this boundary.
289    let pt_client_params = PtClientParameters::builder()
290        .transports(cfg.protocols)
291        .proxy_uri(outbound_proxy.as_ref().map(ToString::to_string))
292        .build()
293        .expect("PtClientParameters constructed incorrectly");
294
295    let mut pt = PluggableClientTransport::new(
296        binary_path,
297        cfg.arguments,
298        pt_common_params,
299        pt_client_params,
300    );
301    pt.launch(rt).await?;
302    Ok(pt)
303}
304
305/// Given a path to a binary for a pluggable transport, return an identifier for
306/// that binary in a format that can be used as a path component.
307fn pt_identifier_as_path(binary_path: impl AsRef<Path>) -> Result<PathBuf, PtError> {
308    // Extract the final component.
309    let mut filename =
310        PathBuf::from(
311            binary_path
312                .as_ref()
313                .file_name()
314                .ok_or_else(|| PtError::NotAFile {
315                    path: binary_path.as_ref().to_path_buf(),
316                })?,
317        );
318
319    // Strip an "exe" off the end, if appropriate.
320    if let Some(ext) = filename.extension() {
321        if ext.eq_ignore_ascii_case(std::env::consts::EXE_EXTENSION) {
322            filename.set_extension("");
323        }
324    }
325
326    Ok(filename)
327}
328
329/// Given a path to a binary for a pluggable transport, return an identifier for
330/// that binary in human-readable form.
331pub(crate) fn pt_identifier(binary_path: impl AsRef<Path>) -> Result<String, PtError> {
332    Ok(pt_identifier_as_path(binary_path)?
333        .to_string_lossy()
334        .to_string())
335}
tor_ptmgr/managed.rs

tor_ptmgr/
managed.rs
