arti/rpc/
session.rs

1//! Declare the RPC session object as exposed from the RPC server run by the `arti` crate.
2
3use arti_client::TorClient;
4use arti_rpcserver::RpcAuthentication;
5use derive_deftly::Deftly;
6use futures::stream::StreamExt as _;
7use std::sync::Arc;
8use tor_async_utils::{DropNotifyEofSignallable, DropNotifyWatchSender};
9use tor_rpc_connect::SuperuserPermission;
10use tor_rpcbase::{self as rpc};
11use tor_rtcompat::Runtime;
12
13use crate::{
14    proxy::port_info,
15    reload_cfg::{CfgMgr, LaunchableTorClient},
16    rpc::{listener::RpcConnInfo, superuser::RpcSuperuser},
17};
18
19use super::proxyinfo::{self, ProxyInfo};
20
21/// A top-level RPC session object.
22///
23/// This is the first object that an RPC user receives upon authenticating;
24/// It is returned by `auth:authenticate`.
25///
26/// Other objects (`TorClient`,`RpcDataStream`, etc)
27/// are available using methods on this object.
28/// (See the list of available methods.)
29///
30/// This type wraps and delegates to [`arti_rpcserver::RpcSession`],
31/// but exposes additional functionality not available at the
32/// level of [`arti_rpcserver`], including information about configured proxies.
33///
34/// This ObjectID for this object can be used as the target of a SOCKS stream.
35#[derive(Deftly)]
36#[derive_deftly(rpc::Object)]
37#[deftly(rpc(
38    delegate_with = "|this: &Self| Some(this.session.clone())",
39    delegate_type = "arti_rpcserver::RpcSession"
40))]
41#[deftly(rpc(expose_outside_of_session))]
42pub(super) struct ArtiRpcSession {
43    /// State about the `arti` server, as seen by the Rpc system.
44    pub(super) arti_state: Arc<RpcVisibleArtiState>,
45    /// The underlying RpcSession object that we delegate to.
46    session: Arc<arti_rpcserver::RpcSession>,
47}
48
49/// Information about the current global top-level Arti state,
50/// as exposed to an Rpc Session.
51//
52// TODO: This type is dangerously close to being a collection of globals.
53// We should refactor it aggressively when we refactor the `arti` crate.
54//
55// TODO: Right now this is constructed in the same form that it's used in
56// ArtiRpcSession.  Later on, we could split it into one type that
57// the rest of this crate constructs, and another type that the
58// ArtiRpcSession actually uses. We should do that if the needs seem to diverge.
59pub(crate) struct RpcVisibleArtiState {
60    /// A `ProxyInfo` that we hand out when asked to list our proxy ports.
61    ///
62    /// Right now it only lists Socks; in the future it may list more.
63    proxy_info: postage::watch::Receiver<ProxyInfoState>,
64}
65
66/// Handle to set RPC state across RPC sessions.  (See `RpcVisibleArtiState`.)
67#[derive(Debug)]
68pub(crate) struct RpcStateSender {
69    /// Sender for setting our list of proxy ports.
70    proxy_info_sender: DropNotifyWatchSender<ProxyInfoState>,
71}
72
73impl ArtiRpcSession {
74    /// Construct a new `ArtiRpcSession`.
75    ///
76    /// Privileges on the session (if any) are derived from `auth`, which describes
77    /// how the user authenticated.
78    ///
79    /// The session receives a new isolated TorClient, based on `client_root`.
80    pub(super) fn new<R: Runtime>(
81        auth: &RpcAuthentication,
82        client_root: &Arc<TorClient<R>>,
83        launchable_client: &Arc<LaunchableTorClient<R>>,
84        arti_state: &Arc<RpcVisibleArtiState>,
85        cfg_mgr: &Arc<CfgMgr<R>>,
86        listener_info: &RpcConnInfo,
87    ) -> Arc<Self> {
88        let _ = auth; // This is currently unused; any authentication gives the same result.
89        let client = client_root.isolated_client();
90        let session = arti_rpcserver::RpcSession::new_with_client(client);
91        if listener_info.allow_superuser == SuperuserPermission::Allowed {
92            session.provide_superuser_permission(Arc::new(RpcSuperuser::new(
93                client_root.clone(),
94                launchable_client.clone(),
95                cfg_mgr.clone(),
96            )) as _);
97        }
98        Arc::new(ArtiRpcSession {
99            session,
100            arti_state: arti_state.clone(),
101        })
102    }
103}
104
105/// Possible state for a watched proxy_info.
106#[derive(Debug, Clone)]
107enum ProxyInfoState {
108    /// We haven't set it yet.
109    Unset,
110    /// We've set it to a given value.
111    Set(Arc<ProxyInfo>),
112    /// The sender has been dropped.
113    Eof,
114}
115
116impl DropNotifyEofSignallable for ProxyInfoState {
117    fn eof() -> Self {
118        Self::Eof
119    }
120}
121
122impl RpcVisibleArtiState {
123    /// Construct a new `RpcVisibleArtiState`.
124    pub(crate) fn new() -> (Arc<Self>, RpcStateSender) {
125        let (proxy_info_sender, proxy_info) = postage::watch::channel_with(ProxyInfoState::Unset);
126        let proxy_info_sender = DropNotifyWatchSender::new(proxy_info_sender);
127        (
128            Arc::new(Self { proxy_info }),
129            RpcStateSender { proxy_info_sender },
130        )
131    }
132
133    /// Return the latest proxy info, waiting until it is set.
134    ///
135    /// Return an error if the sender has been closed.
136    pub(super) async fn get_proxy_info(&self) -> Result<Arc<ProxyInfo>, ()> {
137        let mut proxy_info = self.proxy_info.clone();
138        while let Some(v) = proxy_info.next().await {
139            match v {
140                ProxyInfoState::Unset => {
141                    // Not yet set, try again.
142                }
143                ProxyInfoState::Set(proxyinfo) => return Ok(Arc::clone(&proxyinfo)),
144                ProxyInfoState::Eof => return Err(()),
145            }
146        }
147        Err(())
148    }
149}
150
151impl RpcStateSender {
152    /// Set the list of stream listener addresses on this state.
153    ///
154    /// This method may only be called once per state.
155    pub(crate) fn set_stream_listeners(&mut self, ports: &[port_info::Port]) {
156        let info = ProxyInfo {
157            proxies: ports
158                .iter()
159                .filter_map(|port| {
160                    Some(proxyinfo::Proxy {
161                        listener: proxyinfo::ProxyListener::try_from_portinfo(port)?,
162                    })
163                })
164                .collect(),
165        };
166        *self.proxy_info_sender.borrow_mut() = ProxyInfoState::Set(Arc::new(info));
167    }
168}
169
170#[cfg(test)]
171mod test {
172    // @@ begin test lint list maintained by maint/add_warning @@
173    #![allow(clippy::bool_assert_comparison)]
174    #![allow(clippy::clone_on_copy)]
175    #![allow(clippy::dbg_macro)]
176    #![allow(clippy::mixed_attributes_style)]
177    #![allow(clippy::print_stderr)]
178    #![allow(clippy::print_stdout)]
179    #![allow(clippy::single_char_pattern)]
180    #![allow(clippy::unwrap_used)]
181    #![allow(clippy::unchecked_time_subtraction)]
182    #![allow(clippy::useless_vec)]
183    #![allow(clippy::needless_pass_by_value)]
184    #![allow(clippy::string_slice)] // See arti#2571
185    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
186
187    use tor_rtcompat::SpawnExt as _;
188    use tor_rtmock::MockRuntime;
189
190    use super::*;
191
192    #[test]
193    fn set_proxy_info() {
194        MockRuntime::test_with_various(|rt| async move {
195            let (state, mut sender) = RpcVisibleArtiState::new();
196            let _task = rt.clone().spawn_with_handle(async move {
197                sender.set_stream_listeners(&[port_info::Port {
198                    protocol: port_info::SupportedProtocol::Socks,
199                    address: "8.8.8.8:40".parse().unwrap(),
200                }]);
201                sender // keep sender alive
202            });
203
204            let value = state.get_proxy_info().await;
205
206            // At this point, we've returned once, so this will test that we get a fresh answer even
207            // if we already set the inner value.
208            let value_again = state.get_proxy_info().await;
209            assert_eq!(value.unwrap(), value_again.unwrap());
210        });
211    }
212}
arti/rpc/session.rs

arti/rpc/
session.rs
