use crate::address::IntoTorAddr;
use crate::config::{ClientAddrConfig, StreamTimeoutConfig, TorClientConfig};
use tor_circmgr::{DirInfo, IsolationToken, StreamIsolationBuilder, TargetPort};
use tor_config::MutCfg;
use tor_dirmgr::DirEvent;
use tor_persist::{FsStateMgr, StateMgr};
use tor_proto::circuit::ClientCirc;
use tor_proto::stream::{DataStream, IpVersionPreference, StreamParameters};
use tor_rtcompat::{PreferredRuntime, Runtime, SleepProviderExt};
use futures::lock::Mutex as AsyncMutex;
use futures::stream::StreamExt;
use futures::task::SpawnExt;
use std::convert::TryInto;
use std::net::IpAddr;
use std::result::Result as StdResult;
use std::sync::{Arc, Mutex, Weak};
use std::time::Duration;
use crate::err::ErrorDetail;
use crate::{status, util, TorClientBuilder};
use tracing::{debug, error, info, warn};
#[derive(Clone)]
pub struct TorClient<R: Runtime> {
runtime: R,
client_isolation: IsolationToken,
connect_prefs: StreamPrefs,
circmgr: Arc<tor_circmgr::CircMgr<R>>,
dirmgr: Arc<tor_dirmgr::DirMgr<R>>,
statemgr: FsStateMgr,
addrcfg: Arc<MutCfg<ClientAddrConfig>>,
timeoutcfg: Arc<MutCfg<StreamTimeoutConfig>>,
reconfigure_lock: Arc<Mutex<()>>,
status_receiver: status::BootstrapEvents,
bootstrap_in_progress: Arc<AsyncMutex<()>>,
should_bootstrap: BootstrapBehavior,
}
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum BootstrapBehavior {
OnDemand,
Manual,
}
impl Default for BootstrapBehavior {
fn default() -> Self {
BootstrapBehavior::OnDemand
}
}
#[derive(Debug, Clone, Default)]
pub struct StreamPrefs {
ip_ver_pref: IpVersionPreference,
isolation: StreamIsolationPreference,
optimistic_stream: bool,
}
#[derive(Debug, Clone)]
enum StreamIsolationPreference {
None,
Explicit(IsolationToken),
EveryStream,
}
impl Default for StreamIsolationPreference {
fn default() -> Self {
StreamIsolationPreference::None
}
}
impl StreamPrefs {
pub fn new() -> Self {
Self::default()
}
pub fn ipv6_preferred(&mut self) -> &mut Self {
self.ip_ver_pref = IpVersionPreference::Ipv6Preferred;
self
}
pub fn ipv6_only(&mut self) -> &mut Self {
self.ip_ver_pref = IpVersionPreference::Ipv6Only;
self
}
pub fn ipv4_preferred(&mut self) -> &mut Self {
self.ip_ver_pref = IpVersionPreference::Ipv4Preferred;
self
}
pub fn ipv4_only(&mut self) -> &mut Self {
self.ip_ver_pref = IpVersionPreference::Ipv4Only;
self
}
pub fn optimistic(&mut self) -> &mut Self {
self.optimistic_stream = true;
self
}
fn wrap_target_port(&self, port: u16) -> TargetPort {
match self.ip_ver_pref {
IpVersionPreference::Ipv6Only => TargetPort::ipv6(port),
_ => TargetPort::ipv4(port),
}
}
fn stream_parameters(&self) -> StreamParameters {
let mut params = StreamParameters::default();
params
.ip_version(self.ip_ver_pref)
.optimistic(self.optimistic_stream);
params
}
pub fn set_isolation_group(&mut self, isolation_group: IsolationToken) -> &mut Self {
self.isolation = StreamIsolationPreference::Explicit(isolation_group);
self
}
pub fn new_isolation_group(&mut self) -> &mut Self {
self.isolation = StreamIsolationPreference::Explicit(IsolationToken::new());
self
}
pub fn isolate_every_stream(&mut self) -> &mut Self {
self.isolation = StreamIsolationPreference::EveryStream;
self
}
fn isolation_group(&self) -> Option<IsolationToken> {
use StreamIsolationPreference as SIP;
match self.isolation {
SIP::None => None,
SIP::Explicit(ig) => Some(ig),
SIP::EveryStream => Some(IsolationToken::new()),
}
}
}
impl TorClient<PreferredRuntime> {
pub async fn create_bootstrapped(config: TorClientConfig) -> crate::Result<Self> {
let runtime = PreferredRuntime::current()
.expect("TorClient could not get an asynchronous runtime; are you running in the right context?");
Self::with_runtime(runtime)
.config(config)
.create_bootstrapped()
.await
}
pub fn builder() -> TorClientBuilder<PreferredRuntime> {
let runtime = PreferredRuntime::current()
.expect("TorClient could not get an asynchronous runtime; are you running in the right context?");
TorClientBuilder::new(runtime)
}
}
impl<R: Runtime> TorClient<R> {
pub fn with_runtime(runtime: R) -> TorClientBuilder<R> {
TorClientBuilder::new(runtime)
}
pub(crate) fn create_inner(
runtime: R,
config: TorClientConfig,
autobootstrap: BootstrapBehavior,
) -> StdResult<Self, ErrorDetail> {
let circ_cfg = config.get_circmgr_config()?;
let dir_cfg = config.get_dirmgr_config()?;
let statemgr = FsStateMgr::from_path(config.storage.expand_state_dir()?)?;
let addr_cfg = config.address_filter.clone();
let timeout_cfg = config.stream_timeouts;
let (status_sender, status_receiver) = postage::watch::channel();
let status_receiver = status::BootstrapEvents {
inner: status_receiver,
};
let chanmgr = Arc::new(tor_chanmgr::ChanMgr::new(runtime.clone()));
let circmgr =
tor_circmgr::CircMgr::new(circ_cfg, statemgr.clone(), &runtime, Arc::clone(&chanmgr))
.map_err(ErrorDetail::CircMgrSetup)?;
let dirmgr = tor_dirmgr::DirMgr::create_unbootstrapped(
dir_cfg,
runtime.clone(),
Arc::clone(&circmgr),
)?;
let conn_status = chanmgr.bootstrap_events();
let dir_status = dirmgr.bootstrap_events();
runtime
.spawn(status::report_status(
status_sender,
conn_status,
dir_status,
))
.map_err(|e| ErrorDetail::from_spawn("top-level status reporter", e))?;
runtime
.spawn(continually_expire_channels(
runtime.clone(),
Arc::downgrade(&chanmgr),
))
.map_err(|e| ErrorDetail::from_spawn("channel expiration task", e))?;
runtime
.spawn(keep_circmgr_params_updated(
dirmgr.events(),
Arc::downgrade(&circmgr),
Arc::downgrade(&dirmgr),
))
.map_err(|e| ErrorDetail::from_spawn("circmgr parameter updater", e))?;
runtime
.spawn(update_persistent_state(
runtime.clone(),
Arc::downgrade(&circmgr),
statemgr.clone(),
))
.map_err(|e| ErrorDetail::from_spawn("persistent state updater", e))?;
runtime
.spawn(continually_launch_timeout_testing_circuits(
runtime.clone(),
Arc::downgrade(&circmgr),
Arc::downgrade(&dirmgr),
))
.map_err(|e| ErrorDetail::from_spawn("timeout-probe circuit launcher", e))?;
runtime
.spawn(continually_preemptively_build_circuits(
runtime.clone(),
Arc::downgrade(&circmgr),
Arc::downgrade(&dirmgr),
))
.map_err(|e| ErrorDetail::from_spawn("preemptive circuit launcher", e))?;
let client_isolation = IsolationToken::new();
Ok(TorClient {
runtime,
client_isolation,
connect_prefs: Default::default(),
circmgr,
dirmgr,
statemgr,
addrcfg: Arc::new(addr_cfg.into()),
timeoutcfg: Arc::new(timeout_cfg.into()),
reconfigure_lock: Arc::new(Mutex::new(())),
status_receiver,
bootstrap_in_progress: Arc::new(AsyncMutex::new(())),
should_bootstrap: autobootstrap,
})
}
pub async fn bootstrap(&self) -> crate::Result<()> {
self.bootstrap_inner().await.map_err(ErrorDetail::into)
}
async fn bootstrap_inner(&self) -> StdResult<(), ErrorDetail> {
let _bootstrap_lock = self.bootstrap_in_progress.lock().await;
if self.statemgr.try_lock()?.held() {
debug!("It appears we have the lock on our state files.");
} else {
info!(
"Another process has the lock on our state files. We'll proceed in read-only mode."
);
}
let unlock_guard = util::StateMgrUnlockGuard::new(&self.statemgr);
self.dirmgr.bootstrap().await?;
self.circmgr
.update_network_parameters(self.dirmgr.netdir()?.params());
unlock_guard.disarm();
Ok(())
}
async fn wait_for_bootstrap(&self) -> StdResult<(), ErrorDetail> {
match self.should_bootstrap {
BootstrapBehavior::OnDemand => {
self.bootstrap_inner().await?;
}
BootstrapBehavior::Manual => {
self.bootstrap_in_progress.lock().await;
}
}
Ok(())
}
pub fn reconfigure(
&self,
new_config: &TorClientConfig,
how: tor_config::Reconfigure,
) -> crate::Result<()> {
let _guard = self.reconfigure_lock.lock().expect("Poisoned lock");
match how {
tor_config::Reconfigure::AllOrNothing => {
self.reconfigure(new_config, tor_config::Reconfigure::CheckAllOrNothing)?;
}
tor_config::Reconfigure::CheckAllOrNothing => {}
tor_config::Reconfigure::WarnOnFailures => {}
_ => {}
}
let circ_cfg = new_config.get_circmgr_config().map_err(wrap_err)?;
let dir_cfg = new_config.get_dirmgr_config().map_err(wrap_err)?;
let state_cfg = new_config.storage.expand_state_dir().map_err(wrap_err)?;
let addr_cfg = &new_config.address_filter;
let timeout_cfg = &new_config.stream_timeouts;
if state_cfg != self.statemgr.path() {
how.cannot_change("storage.state_dir").map_err(wrap_err)?;
}
self.circmgr.reconfigure(&circ_cfg, how).map_err(wrap_err)?;
self.dirmgr.reconfigure(&dir_cfg, how).map_err(wrap_err)?;
if how == tor_config::Reconfigure::CheckAllOrNothing {
return Ok(());
}
self.addrcfg.replace(addr_cfg.clone());
self.timeoutcfg.replace(timeout_cfg.clone());
Ok(())
}
#[must_use]
pub fn isolated_client(&self) -> TorClient<R> {
let mut result = self.clone();
result.client_isolation = IsolationToken::new();
result
}
pub async fn connect<A: IntoTorAddr>(&self, target: A) -> crate::Result<DataStream> {
self.connect_with_prefs(target, &self.connect_prefs).await
}
pub async fn connect_with_prefs<A: IntoTorAddr>(
&self,
target: A,
prefs: &StreamPrefs,
) -> crate::Result<DataStream> {
let addr = target.into_tor_addr().map_err(wrap_err)?;
addr.enforce_config(&self.addrcfg.get())?;
let (addr, port) = addr.into_string_and_port();
let exit_ports = [prefs.wrap_target_port(port)];
let circ = self
.get_or_launch_exit_circ(&exit_ports, prefs)
.await
.map_err(wrap_err)?;
info!("Got a circuit for {}:{}", addr, port);
let stream_future = circ.begin_stream(&addr, port, Some(prefs.stream_parameters()));
let stream = self
.runtime
.timeout(self.timeoutcfg.get().connect_timeout, stream_future)
.await
.map_err(|_| ErrorDetail::ExitTimeout)?
.map_err(wrap_err)?;
Ok(stream)
}
fn set_stream_prefs(&mut self, connect_prefs: StreamPrefs) {
self.connect_prefs = connect_prefs;
}
#[must_use]
pub fn clone_with_prefs(&self, connect_prefs: StreamPrefs) -> Self {
let mut result = self.clone();
result.set_stream_prefs(connect_prefs);
result
}
pub async fn resolve(&self, hostname: &str) -> crate::Result<Vec<IpAddr>> {
self.resolve_with_prefs(hostname, &self.connect_prefs).await
}
pub async fn resolve_with_prefs(
&self,
hostname: &str,
prefs: &StreamPrefs,
) -> crate::Result<Vec<IpAddr>> {
let addr = (hostname, 0).into_tor_addr().map_err(wrap_err)?;
addr.enforce_config(&self.addrcfg.get()).map_err(wrap_err)?;
let circ = self.get_or_launch_exit_circ(&[], prefs).await?;
let resolve_future = circ.resolve(hostname);
let addrs = self
.runtime
.timeout(self.timeoutcfg.get().resolve_timeout, resolve_future)
.await
.map_err(|_| ErrorDetail::ExitTimeout)?
.map_err(wrap_err)?;
Ok(addrs)
}
pub async fn resolve_ptr(&self, addr: IpAddr) -> crate::Result<Vec<String>> {
self.resolve_ptr_with_prefs(addr, &self.connect_prefs).await
}
pub async fn resolve_ptr_with_prefs(
&self,
addr: IpAddr,
prefs: &StreamPrefs,
) -> crate::Result<Vec<String>> {
let circ = self.get_or_launch_exit_circ(&[], prefs).await?;
let resolve_ptr_future = circ.resolve_ptr(addr);
let hostnames = self
.runtime
.timeout(
self.timeoutcfg.get().resolve_ptr_timeout,
resolve_ptr_future,
)
.await
.map_err(|_| ErrorDetail::ExitTimeout)?
.map_err(wrap_err)?;
Ok(hostnames)
}
#[cfg(feature = "experimental-api")]
pub fn dirmgr(&self) -> Arc<tor_dirmgr::DirMgr<R>> {
Arc::clone(&self.dirmgr)
}
#[cfg(feature = "experimental-api")]
pub fn circmgr(&self) -> Arc<tor_circmgr::CircMgr<R>> {
Arc::clone(&self.circmgr)
}
pub fn runtime(&self) -> &R {
&self.runtime
}
async fn get_or_launch_exit_circ(
&self,
exit_ports: &[TargetPort],
prefs: &StreamPrefs,
) -> StdResult<ClientCirc, ErrorDetail> {
self.wait_for_bootstrap().await?;
let dir = self
.dirmgr
.opt_netdir()
.ok_or(ErrorDetail::BootstrapRequired {
action: "launch a circuit",
})?;
let isolation = {
let mut b = StreamIsolationBuilder::new();
b.owner_token(self.client_isolation);
if let Some(tok) = prefs.isolation_group() {
b.stream_token(tok);
}
b.build().expect("Failed to construct StreamIsolation")
};
let circ = self
.circmgr
.get_or_launch_exit(dir.as_ref().into(), exit_ports, isolation)
.await
.map_err(|cause| ErrorDetail::ObtainExitCircuit {
cause,
exit_ports: exit_ports.into(),
})?;
drop(dir);
Ok(circ)
}
pub fn bootstrap_status(&self) -> status::BootstrapStatus {
self.status_receiver.inner.borrow().clone()
}
pub fn bootstrap_events(&self) -> status::BootstrapEvents {
self.status_receiver.clone()
}
}
pub(crate) fn wrap_err<T>(err: T) -> crate::Error
where
ErrorDetail: From<T>,
{
ErrorDetail::from(err).into()
}
async fn keep_circmgr_params_updated<R: Runtime>(
mut events: impl futures::Stream<Item = DirEvent> + Unpin,
circmgr: Weak<tor_circmgr::CircMgr<R>>,
dirmgr: Weak<tor_dirmgr::DirMgr<R>>,
) {
use DirEvent::*;
while let Some(event) = events.next().await {
match event {
NewConsensus => {
if let (Some(cm), Some(dm)) = (Weak::upgrade(&circmgr), Weak::upgrade(&dirmgr)) {
let netdir = dm
.netdir()
.expect("got new consensus event, without a netdir?");
cm.update_network_parameters(netdir.params());
cm.update_network(&netdir);
} else {
debug!("Circmgr or dirmgr has disappeared; task exiting.");
break;
}
}
NewDescriptors => {
if let (Some(cm), Some(dm)) = (Weak::upgrade(&circmgr), Weak::upgrade(&dirmgr)) {
let netdir = dm
.netdir()
.expect("got new descriptors event, without a netdir?");
cm.update_network(&netdir);
} else {
debug!("Circmgr or dirmgr has disappeared; task exiting.");
break;
}
}
_ => {
}
}
}
}
async fn update_persistent_state<R: Runtime>(
runtime: R,
circmgr: Weak<tor_circmgr::CircMgr<R>>,
statemgr: FsStateMgr,
) {
loop {
if let Some(circmgr) = Weak::upgrade(&circmgr) {
use tor_persist::LockStatus::*;
match statemgr.try_lock() {
Err(e) => {
error!("Problem with state lock file: {}", e);
break;
}
Ok(NewlyAcquired) => {
info!("We now own the lock on our state files.");
if let Err(e) = circmgr.upgrade_to_owned_persistent_state() {
error!("Unable to upgrade to owned state files: {}", e);
break;
}
}
Ok(AlreadyHeld) => {
if let Err(e) = circmgr.store_persistent_state() {
error!("Unable to flush circmgr state: {}", e);
break;
}
}
Ok(NoLock) => {
if let Err(e) = circmgr.reload_persistent_state() {
error!("Unable to reload circmgr state: {}", e);
break;
}
}
}
} else {
debug!("Circmgr has disappeared; task exiting.");
return;
}
runtime.sleep(Duration::from_secs(60)).await;
}
error!("State update task is exiting prematurely.");
}
async fn continually_launch_timeout_testing_circuits<R: Runtime>(
rt: R,
circmgr: Weak<tor_circmgr::CircMgr<R>>,
dirmgr: Weak<tor_dirmgr::DirMgr<R>>,
) {
while let (Some(cm), Some(dm)) = (Weak::upgrade(&circmgr), Weak::upgrade(&dirmgr)) {
if let Some(netdir) = dm.opt_netdir() {
if let Err(e) = cm.launch_timeout_testing_circuit_if_appropriate(&netdir) {
warn!("Problem launching a timeout testing circuit: {}", e);
}
let delay = netdir
.params()
.cbt_testing_delay
.try_into()
.expect("Out-of-bounds value from BoundedInt32");
drop((cm, dm));
rt.sleep(delay).await;
} else {
rt.sleep(Duration::from_secs(10)).await;
}
}
}
async fn continually_preemptively_build_circuits<R: Runtime>(
rt: R,
circmgr: Weak<tor_circmgr::CircMgr<R>>,
dirmgr: Weak<tor_dirmgr::DirMgr<R>>,
) {
while let (Some(cm), Some(dm)) = (Weak::upgrade(&circmgr), Weak::upgrade(&dirmgr)) {
if let Some(netdir) = dm.opt_netdir() {
cm.launch_circuits_preemptively(DirInfo::Directory(&netdir))
.await;
rt.sleep(Duration::from_secs(10)).await;
} else {
rt.sleep(Duration::from_secs(10)).await;
}
}
}
async fn continually_expire_channels<R: Runtime>(rt: R, chanmgr: Weak<tor_chanmgr::ChanMgr<R>>) {
loop {
let delay = if let Some(cm) = Weak::upgrade(&chanmgr) {
cm.expire_channels()
} else {
return;
};
rt.sleep(Duration::from_secs(delay.as_secs())).await;
}
}
#[cfg(test)]
mod test {
#![allow(clippy::unwrap_used)]
use super::*;
use crate::config::TorClientConfigBuilder;
use crate::{ErrorKind, HasKind};
#[test]
fn create_unbootstrapped() {
tor_rtcompat::test_with_one_runtime!(|rt| async {
let state_dir = tempfile::tempdir().unwrap();
let cache_dir = tempfile::tempdir().unwrap();
let cfg = TorClientConfigBuilder::from_directories(state_dir, cache_dir)
.build()
.unwrap();
let _ = TorClient::with_runtime(rt)
.config(cfg)
.bootstrap_behavior(BootstrapBehavior::Manual)
.create_unbootstrapped()
.unwrap();
});
}
#[test]
fn unbootstrapped_client_unusable() {
tor_rtcompat::test_with_one_runtime!(|rt| async {
let state_dir = tempfile::tempdir().unwrap();
let cache_dir = tempfile::tempdir().unwrap();
let cfg = TorClientConfigBuilder::from_directories(state_dir, cache_dir)
.build()
.unwrap();
let client = TorClient::with_runtime(rt)
.config(cfg)
.bootstrap_behavior(BootstrapBehavior::Manual)
.create_unbootstrapped()
.unwrap();
let result = client.connect("example.com:80").await;
assert!(result.is_err());
assert_eq!(result.err().unwrap().kind(), ErrorKind::BootstrapRequired);
});
}
}