github-jwinterm/grin-wallet
1
0
Fork 0
mirror of https://github.com/jwinterm/grin-wallet.git synced 2026-10-17 15:21:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Integrated Tor arti Client (#750)

Browse Source 
* tor: integrated client for listener, update tokio to last version

* tor: send transaction with integrated client

* build: code format

* tor: optimize bootstrap percent check

* tor: handle config build error

* tor: do not show slatepack address if tor not launched

* tor: handle keystore errors

* tor: keep service launched

* tor: bootstrap timeout, clean cache before client launch

* tor: post request timeout

* fix: post timeout

* build: update deps after merge

* tor: store arti runtime

* tor: do not send over tor if `manual` arg presents or `skip_send_attempt` set at config

* fix: arg name

* tor: handle arti runtime creation error

* tor: handle connection error

* tor: handle bootstrap error

* tor: handle start service name error

* tor: handle service config creation error

* tor: service proxy errors

* tor: handle service key creation emptiness

* build: format code

* build: update arti to 0.43, hyper to 0.10.1

* api: start foreign listener before tor connection

* tor: handle json parse errors

* tor: prevent to create several client runtimes, ability to recreate

* tor: return bootstrap error

* tor: handle host parse error

* tor: restart proxy on error

* tor: do not use integrated client when config value use_integrated is missing, add logging

* build: update lock file
This commit is contained in:
ardocrat
2026-06-21 14:20:34 +02:00
committed by GitHub
parent ca5686aa04
commit 411bcff003
17 changed files with 5070 additions and 828 deletions
Download Patch File Download Diff File Expand all files Collapse all files
impls/Cargo.toml
+18 -5
View File
@@ -22,20 +22,33 @@ ring = "0.16"
uuid = { version = "0.8", features = ["serde", "v4"] }
chrono = { version = "0.4.11", features = ["serde"] }
lazy_static = "1"
tokio = { version = "0.2", features = ["full"] }
reqwest = { version = "0.10", features = ["rustls-tls", "socks"] }
tokio = { version = "1.52.3", features = ["full"] }
reqwest = { version = "0.13.3", features = ["socks"] }
#Socks/Tor/Bridge/Proxy
byteorder = "1"
ed25519-dalek = "1.0.0-pre.4"
x25519-dalek = "0.6"
data-encoding = "2"
arti-ed25519-dalek = { version = "2.1.1", package = "ed25519-dalek" }
regex = "1.3"
timer = "0.2"
sysinfo = "0.29"
base64 = "0.12.0"
url = "2.1"
arti-client = { version = "0.43.0", features = ["static", "pt-client", "onion-service-service", "onion-service-client"] }
tor-rtcompat = { version = "0.43.0", features = ["static"] }
fs-mistrust = "0.14.2"
tor-hsservice = "0.43.0"
tor-hsrproxy = "0.43.0"
tor-keymgr = "0.43.0"
tor-llcrypto = "0.43.0"
tor-hscrypto = "0.43.0"
sha2 = "0.10.8"
curve25519-dalek = "4.1.3"
hyper = { version = "1.10.1", features = ["http1", "client"] }
hyper-util = { version = "0.1.20", features = ["tokio"] }
http-body-util = "0.1.3"
bytes = "1.11.1"
grin_wallet_util = { path = "../util", version = "5.4.0-alpha.1" }
grin_wallet_config = { path = "../config", version = "5.4.0-alpha.1" }
grin_wallet_libwallet = { path = "../libwallet", version = "5.4.0-alpha.1" }
impls/src/adapters/mod.rs
+2 -2
View File
@@ -13,12 +13,12 @@
// limitations under the License.
mod file;
pub mod http;
mod slatepack;
pub mod tor;
pub use self::file::PathToSlate;
pub use self::http::HttpSlateSender;
pub use self::slatepack::PathToSlatepack;
pub use self::tor::TorSlateSender;
use crate::config::WalletConfig;
use crate::libwallet::{Error, Slate};
impls/src/adapters/http.rs → impls/src/adapters/tor.rs
+115 -136
View File
@@ -12,128 +12,111 @@
// See the License for the specific language governing permissions and
// limitations under the License.
/// HTTP Wallet 'plugin' implementation
use crate::client_utils::{Client, ClientError};
use crate::libwallet::slate_versions::{SlateVersion, VersionedSlate};
use crate::libwallet::{Error, Slate};
use crate::tor::bridge::TorBridge;
use crate::tor::proxy::TorProxy;
use crate::SlateSender;
use grin_wallet_config::types::{TorBridgeConfig, TorProxyConfig};
use grin_wallet_config::TorConfig;
use serde::Serialize;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::convert::TryFrom;
use std::net::SocketAddr;
use std::path::MAIN_SEPARATOR;
use std::path::PathBuf;
use std::sync::Arc;
use crate::tor::config as tor_config;
use crate::tor::process as tor_process;
const TOR_CONFIG_PATH: &str = "tor/sender";
use crate::client_utils::{Client, ClientError};
use crate::libwallet::slate_versions::{SlateVersion, VersionedSlate};
use crate::libwallet::{Error, Slate};
use crate::tor::arti::{start_tor_client, tor_post};
use crate::tor::bridge::TorBridge;
use crate::tor::process::TorProcess;
use crate::tor::proxy::TorProxy;
use crate::tor::{config as tor_config, Tor};
use crate::SlateSender;
#[derive(Clone)]
pub struct HttpSlateSender {
pub struct TorSlateSender {
base_url: String,
use_socks: bool,
socks_proxy_addr: Option<SocketAddr>,
tor_config_dir: String,
process: Option<Arc<tor_process::TorProcess>>,
bridge: TorBridgeConfig,
proxy: TorProxyConfig,
config: TorConfig,
tor: Arc<Tor>,
}
impl HttpSlateSender {
impl TorSlateSender {
/// Create, return Err if scheme is not "http"
fn new(base_url: &str) -> Result<HttpSlateSender, SchemeNotHttp> {
pub fn new(base_url: &str, config: TorConfig) -> Result<TorSlateSender, Error> {
if !base_url.starts_with("http") && !base_url.starts_with("https") {
Err(SchemeNotHttp)
Err(Error::GenericError("Scheme must be http".to_string()))
} else {
Ok(HttpSlateSender {
let tor_dir = {
let mut path = PathBuf::from(&config.send_config_dir);
path.push("tor");
path.push("sender");
path
};
let tor = if config.use_integrated.unwrap_or(false) {
start_tor_client(tor_dir.to_str().unwrap(), config.clone())?
} else {
Self::launch_tor_process(&config, &tor_dir)?
};
Ok(TorSlateSender {
base_url: base_url.to_owned(),
use_socks: false,
socks_proxy_addr: None,
tor_config_dir: String::from(""),
process: None,
bridge: TorBridgeConfig::default(),
proxy: TorProxyConfig::default(),
config,
tor: Arc::new(tor),
})
}
}
/// Switch to using socks proxy
pub fn with_socks_proxy(
base_url: &str,
proxy_addr: &str,
tor_config_dir: &str,
tor_bridge: TorBridgeConfig,
tor_proxy: TorProxyConfig,
) -> Result<HttpSlateSender, SchemeNotHttp> {
let mut ret = Self::new(base_url)?;
ret.use_socks = true;
//TODO: Unwrap
ret.socks_proxy_addr = Some(SocketAddr::V4(proxy_addr.parse().unwrap()));
ret.tor_config_dir = tor_config_dir.into();
ret.bridge = tor_bridge;
ret.proxy = tor_proxy;
Ok(ret)
}
/// Launch external Tor process.
fn launch_tor_process(config: &TorConfig, tor_dir: &PathBuf) -> Result<Tor, Error> {
let mut tor = TorProcess::new();
let socks_proxy_addr = SocketAddr::V4(
config
.socks_proxy_addr
.parse()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?,
);
info!("Starting TOR Process for send at {:?}", socks_proxy_addr);
/// launch TOR process
pub fn launch_tor(&mut self) -> Result<(), Error> {
// set up tor send process if needed
let mut tor = tor_process::TorProcess::new();
if self.use_socks && self.process.is_none() {
let tor_dir = format!(
"{}{}{}",
&self.tor_config_dir, MAIN_SEPARATOR, TOR_CONFIG_PATH
);
info!(
"Starting TOR Process for send at {:?}",
self.socks_proxy_addr
);
let mut hm_tor_bridge: HashMap<String, String> = HashMap::new();
if self.bridge.bridge_line.is_some() {
let bridge_struct = TorBridge::try_from(self.bridge.clone())
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
hm_tor_bridge = bridge_struct
.to_hashmap()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
}
let mut hm_tor_proxy: HashMap<String, String> = HashMap::new();
if self.proxy.transport.is_some() || self.proxy.allowed_port.is_some() {
let proxy = TorProxy::try_from(self.proxy.clone())
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
hm_tor_proxy = proxy
.to_hashmap()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
}
tor_config::output_tor_sender_config(
&tor_dir,
&self.socks_proxy_addr.unwrap().to_string(),
hm_tor_bridge,
hm_tor_proxy,
)
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
// Start TOR process
tor.torrc_path(&format!("{}/torrc", &tor_dir))
.working_dir(&tor_dir)
.timeout(20)
.completion_percent(100)
.launch()
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
self.process = Some(Arc::new(tor));
let mut hm_tor_bridge: HashMap<String, String> = HashMap::new();
if config.bridge.bridge_line.is_some() {
let bridge_struct = TorBridge::try_from(config.bridge.clone())
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
hm_tor_bridge = bridge_struct
.to_hashmap()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
}
Ok(())
let mut hm_tor_proxy: HashMap<String, String> = HashMap::new();
if config.proxy.transport.is_some() || config.proxy.allowed_port.is_some() {
let proxy = TorProxy::try_from(config.proxy.clone())
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
hm_tor_proxy = proxy
.to_hashmap()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
}
tor_config::output_tor_sender_config(
tor_dir.to_str().unwrap(),
socks_proxy_addr.to_string().as_str(),
hm_tor_bridge,
hm_tor_proxy,
)
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
// Start TOR process
let mut path = tor_dir.clone();
path.push("torrc");
tor.torrc_path(path.to_str().unwrap())
.working_dir(tor_dir.to_str().unwrap())
.timeout(20)
.completion_percent(100)
.launch()
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
Ok(Tor {
process: Some(tor),
service: None,
client: None,
})
}
/// Check version of the listening wallet
pub fn check_other_version(&mut self, url: &str) -> Result<SlateVersion, Error> {
self.launch_tor()?;
let req = json!({
"jsonrpc": "2.0",
"method": "check_version",
@@ -141,7 +124,7 @@ impl HttpSlateSender {
"params": []
});
let res: String = self.post(url, None, req).map_err(|e| {
let res: String = self.post(url, req).map_err(|e| {
let mut report = format!("Performing version check (is recipient listening?): {}", e);
let err_string = format!("{}", e);
if err_string.contains("404") {
@@ -154,7 +137,8 @@ impl HttpSlateSender {
Error::ClientCallback(report)
})?;
let res: Value = serde_json::from_str(&res).unwrap();
let res: Value = serde_json::from_str(&res)
.map_err(|_| Error::ClientCallback(format!("Can not parse response {:?}", res)))?;
trace!("Response: {}", res);
if res["error"] != json!(null) {
let report = format!(
@@ -168,9 +152,11 @@ impl HttpSlateSender {
let resp_value = res["result"]["Ok"].clone();
trace!("resp_value: {}", resp_value.clone());
let foreign_api_version: u16 =
serde_json::from_value(resp_value["foreign_api_version"].clone()).unwrap();
serde_json::from_value(resp_value["foreign_api_version"].clone())
.map_err(|_| Error::ClientCallback(format!("Can not parse response {:?}", res)))?;
let supported_slate_versions: Vec<String> =
serde_json::from_value(resp_value["supported_slate_versions"].clone()).unwrap();
serde_json::from_value(resp_value["supported_slate_versions"].clone())
.map_err(|_| Error::ClientCallback(format!("Can not parse response {:?}", res)))?;
// trivial tests for now, but will be expanded later
if foreign_api_version < 2 {
@@ -188,32 +174,33 @@ impl HttpSlateSender {
Err(Error::ClientCallback(report))
}
fn post<IN>(
&self,
url: &str,
api_secret: Option<String>,
input: IN,
) -> Result<String, ClientError>
fn post<IN>(&self, url: &str, input: IN) -> Result<String, ClientError>
where
IN: Serialize,
{
let client = if !self.use_socks {
Client::new()
let res = if self.tor.process.is_some() {
let socks_proxy_addr =
SocketAddr::V4(self.config.socks_proxy_addr.parse().map_err(|_| {
ClientError::Internal("Socks proxy address is not set".to_string())
})?);
let client = Client::with_proxy(socks_proxy_addr, "socks5h://")
.map_err(|_| ClientError::Internal("Unable to create http client".into()))?;
let req = client.create_post_request(url, None, &input)?;
let res = client.send_request(req)?;
res
} else {
Client::with_proxy(
self.socks_proxy_addr
.ok_or_else(|| ClientError::Internal("No socks proxy address set".into()))?,
"socks5h://",
)
}
.map_err(|_| ClientError::Internal("Unable to create http client".into()))?;
let req = client.create_post_request(url, api_secret, &input)?;
let res = client.send_request(req)?;
if let Some(client) = &self.tor.client {
tor_post(client.clone(), &input, url)
.map_err(|e| ClientError::RequestError(format!("{:?}", e)))?
} else {
return Err(ClientError::Internal("Tor is not configured".to_string()));
}
};
Ok(res)
}
}
impl SlateSender for HttpSlateSender {
impl SlateSender for TorSlateSender {
fn send_tx(&mut self, slate: &Slate, finalize: bool) -> Result<Slate, Error> {
let trailing = match self.base_url.ends_with('/') {
true => "",
@@ -221,8 +208,6 @@ impl SlateSender for HttpSlateSender {
};
let url_str = format!("{}{}v2/foreign", self.base_url, trailing);
self.launch_tor()?;
let slate_send = match self.check_other_version(&url_str)? {
SlateVersion::V4 => VersionedSlate::into_version(slate.clone(), SlateVersion::V4)?,
};
@@ -250,7 +235,7 @@ impl SlateSender for HttpSlateSender {
trace!("Sending receive_tx request: {}", req);
let res: String = self.post(&url_str, None, req).map_err(|e| {
let res: String = self.post(&url_str, req).map_err(|e| {
let report = format!(
"Sending transaction slate to other wallet (is recipient listening?): {}",
e
@@ -258,7 +243,8 @@ impl SlateSender for HttpSlateSender {
Error::ClientCallback(report)
})?;
let res: Value = serde_json::from_str(&res).unwrap();
let res: Value = serde_json::from_str(&res)
.map_err(|_| Error::ClientCallback(format!("Can not parse response {:?}", res)))?;
trace!("Response: {}", res);
if res["error"] != json!(null) {
let report = format!(
@@ -272,22 +258,15 @@ impl SlateSender for HttpSlateSender {
let slate_value = res["result"]["Ok"].clone();
trace!("slate_value: {}", slate_value);
let slate = Slate::deserialize_upgrade(&serde_json::to_string(&slate_value).unwrap())
.map_err(|e| {
if let Ok(slate_value) = serde_json::to_string(&slate_value) {
let slate = Slate::deserialize_upgrade(&slate_value).map_err(|e| {
error!("Error deserializing response slate: {}", e);
Error::SlateDeser
})?;
Ok(slate)
}
}
#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug)]
pub struct SchemeNotHttp;
impl Into<Error> for SchemeNotHttp {
fn into(self) -> Error {
let err_str = "url scheme must be http".to_string();
Error::GenericError(err_str)
Ok(slate)
} else {
error!("Error deserializing response slate: {}", slate_value);
Err(Error::SlateDeser)
}
}
}
impls/src/client_utils/client.rs
+7 -23
View File
@@ -21,23 +21,14 @@ use reqwest::{ClientBuilder, Method, Proxy, RequestBuilder};
use serde::{Deserialize, Serialize};
use serde_json;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::sync::Arc;
use std::time::Duration;
use tokio::runtime::{Builder, Handle, Runtime};
// Global Tokio runtime.
// Needs a `Mutex` because `Runtime::block_on` requires mutable access.
// Tokio v0.3 requires immutable self, but we are waiting on upstream
// updates before we can upgrade.
// See: https://github.com/seanmonstar/reqwest/pull/1076
lazy_static! {
pub static ref RUNTIME: Arc<Mutex<Runtime>> = Arc::new(Mutex::new(
Builder::new()
.threaded_scheduler()
.enable_all()
.build()
.unwrap()
));
pub static ref RUNTIME: Arc<Runtime> =
Arc::new(Builder::new_multi_thread().enable_all().build().unwrap());
}
#[derive(Clone, Eq, thiserror::Error, PartialEq, Debug)]
@@ -273,18 +264,11 @@ impl Client {
if Handle::try_current().is_ok() {
let rt = RUNTIME.clone();
let client = self.clone();
std::thread::spawn(move || {
rt.lock()
.unwrap()
.block_on(async { client.send_request_async(req).await })
})
.join()
.unwrap()
} else {
RUNTIME
.lock()
std::thread::spawn(move || rt.block_on(async { client.send_request_async(req).await }))
.join()
.unwrap()
.block_on(self.send_request_async(req))
} else {
RUNTIME.block_on(self.send_request_async(req))
}
}
}
impls/src/lib.rs
+2 -2
View File
@@ -42,8 +42,8 @@ pub mod test_framework;
pub mod tor;
pub use crate::adapters::{
HttpSlateSender, PathToSlate, PathToSlatepack, SlateGetter, SlatePutter, SlateReceiver,
SlateSender,
PathToSlate, PathToSlatepack, SlateGetter, SlatePutter, SlateReceiver, SlateSender,
TorSlateSender,
};
pub use crate::error::Error;
pub use crate::lifecycle::DefaultLCProvider;
impls/src/node_clients/http.rs
+1 -1
View File
@@ -272,7 +272,7 @@ impl NodeClient for HTTPNodeClient {
let rt = RUNTIME.clone();
let res: Result<Vec<_>, _> =
std::thread::spawn(move || rt.lock().unwrap().block_on(async move { task.await }))
std::thread::spawn(move || rt.block_on(async move { task.await }))
.join()
.unwrap();
impls/src/tor/arti.rs
+402
View File
@@ -0,0 +1,402 @@
// Copyright 2026 The Grin Developers
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::tor::{ArtiRuntimeWrapper, Tor};
use arti_client::config::pt::TransportConfigBuilder;
use arti_client::config::{BridgeConfigBuilder, TorClientConfigBuilder};
use arti_client::{TorClient, TorClientConfig};
use arti_ed25519_dalek::hazmat::ExpandedSecretKey;
use bytes::Bytes;
use curve25519_dalek::digest::Digest;
use fs_mistrust::Mistrust;
use grin_util::secp::SecretKey;
use grin_wallet_config::TorConfig;
use grin_wallet_libwallet::Error;
use grin_wallet_util::OnionV3Address;
use http_body_util::{BodyExt, Full};
use hyper::{Request, Uri};
use hyper_util::rt::TokioIo;
use lazy_static::lazy_static;
use serde::Serialize;
use sha2::Sha512;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::{Arc, LazyLock, Mutex};
use std::thread;
use std::time::Duration;
use tor_hscrypto::pk::{HsIdKey, HsIdKeypair};
use tor_hsrproxy::config::{
Encapsulation, ProxyAction, ProxyConfigBuilder, ProxyPattern, ProxyRule, TargetAddr,
};
use tor_hsrproxy::OnionServiceReverseProxy;
use tor_hsservice::config::OnionServiceConfigBuilder;
use tor_hsservice::{HsIdKeypairSpecifier, HsIdPublicKeySpecifier, HsNickname};
use tor_keymgr::config::CfgPath;
use tor_keymgr::{ArtiNativeKeystore, KeyMgrBuilder, KeystoreSelector};
use tor_llcrypto::pk::ed25519::ExpandedKeypair;
use tor_rtcompat::tokio::TokioNativeTlsRuntime;
use tor_rtcompat::{SleepProviderExt, ToplevelBlockOn};
// Arti Tokio runtime.
lazy_static! {
pub static ref ARTI_RUNTIME: LazyLock<Mutex<Option<ArtiRuntimeWrapper>>> =
LazyLock::new(|| Mutex::new(ArtiRuntimeWrapper::create().ok()));
}
/// Get Tor client runtime.
fn runtime() -> Result<TokioNativeTlsRuntime, Error> {
let mut runtime = ARTI_RUNTIME.lock().unwrap();
let r = match runtime.as_ref() {
None => runtime.insert(ArtiRuntimeWrapper::create()?),
Some(r) => r
};
Ok(r.runtime.clone())
}
/// Start Tor service from provided key.
pub fn start_tor_service(
key: SecretKey,
tor_dir: &str,
addr: &str,
config: TorConfig,
) -> Result<Tor, Error> {
info!("Starting integrated Tor listener.");
let use_proxy = config.proxy.transport.is_some() && config.proxy.address.is_some();
if use_proxy {
info!("Proxy configuration will be ignored.");
}
let state_path = Path::new(&tor_dir).join("state");
let cache_path = Path::new(&tor_dir).join("cache");
let (client, config) = init_client(&state_path, &cache_path, config)?;
// Add service key to keystore.
let onion_address =
OnionV3Address::from_private(&key.0).map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
let hs = HsNickname::new(onion_address.to_string())
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
let keystore_path = Path::new(&state_path).join("keystore");
let _ = add_service_key(config.fs_mistrust(), &key, &hs, keystore_path)?;
// Launch Onion service.
let service_config = OnionServiceConfigBuilder::default()
.nickname(hs.clone())
.build()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
let running_onion_service = match client.launch_onion_service(service_config) {
Ok(res) => {
if let Some((service, mut request)) = res {
let addr: SocketAddr = addr
.parse()
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
let c = client.clone();
thread::spawn(move || {
c.clone().runtime().block_on(async move {
// Launch service proxy.
async fn run_proxy<S> (c: Arc<TorClient<TokioNativeTlsRuntime>>,
addr: SocketAddr,
request: &mut S,
hs: HsNickname)
where
S: futures::Stream<Item = tor_hsservice::RendRequest> + Unpin + Send + 'static,
{
match run_service_proxy(c.clone(), addr, request, hs.clone()).await {
Ok(_) => {
info!("Tor proxy stopped");
},
Err(e) => {
error!("Tor proxy error: {:?}, restarting", e);
tokio::time::sleep(Duration::from_millis(1000)).await;
Box::pin(run_proxy(c, addr, request, hs)).await;
}
}
}
run_proxy(c.clone(), addr, &mut request, hs.clone()).await;
})
});
service
} else {
return Err(Error::TorProcess("Can not launch onion service".to_owned()));
}
}
Err(e) => return Err(Error::TorProcess(format!("{:?}", e))),
};
Ok(Tor {
process: None,
service: Some(running_onion_service),
client: Some(client),
})
}
/// Start Tor client to send requests.
pub fn start_tor_client(tor_dir: &str, config: TorConfig) -> Result<Tor, Error> {
let state_path = Path::new(tor_dir).join("state");
let cache_path = Path::new(tor_dir).join("cache");
let (client, _) = init_client(&state_path, &cache_path, config)?;
Ok(Tor {
process: None,
service: None,
client: Some(client),
})
}
/// Tor request timeout in milliseconds.
const REQUEST_TIMEOUT_MS: u64 = 60000;
/// Make POST request with provided client.
pub fn tor_post<IN>(
client: Arc<TorClient<TokioNativeTlsRuntime>>,
input: &IN,
url: &str,
) -> Result<String, Error>
where
IN: Serialize,
{
let json = serde_json::to_string(input)
.map_err(|_| Error::GenericError("Could not serialize data to JSON".to_owned()))?;
let url = url.to_string();
let url: Uri = url
.parse()
.map_err(|_| Error::GenericError(format!("Bad URL: {}", url)))?;
let host = match url.host() {
None => return Err(Error::GenericError(format!("URL {} has bad host", url))),
Some(h) => h
}.to_string();
let res: Result<String, Error> = thread::spawn(move || {
let c = client.clone();
client.runtime().block_on(async move {
let res = c
.runtime()
.timeout(Duration::from_millis(REQUEST_TIMEOUT_MS), async {
let stream = c
.connect((host, url.port_u16().unwrap_or(80)))
.await
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
let (mut request_sender, connection) =
hyper::client::conn::http1::handshake(TokioIo::new(stream))
.await
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
// Spawn a task to poll the connection and drive the HTTP state.
tokio::spawn(async move {
if let Err(e) = connection.await {
error!("Tor connection error: {}", e);
}
});
let resp = request_sender
.send_request(
Request::builder()
.uri(url)
.method("POST")
.body::<Full<Bytes>>(Full::from(json))
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?,
)
.await
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
let body_resp = resp
.into_body()
.collect()
.await
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
let body = body_resp.to_bytes().into();
let body_text = String::from_utf8(body)
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
Ok(body_text)
})
.await;
match res {
Err(e) => Err(Error::TorProcess(format!("{:?}", e))),
Ok(body) => Ok(body),
}
})
})
.join()
.unwrap_or_else(|e| return Err(Error::TorProcess(format!("{:?}", e))))?;
res
}
/// Client Bootstrap timeout in milliseconds.
const BOOTSTRAP_TIMEOUT_MS: u64 = 60000;
/// Create Tor client.
fn init_client(
state_path: &PathBuf,
cache_path: &PathBuf,
config: TorConfig,
) -> Result<(Arc<TorClient<TokioNativeTlsRuntime>>, TorClientConfig), Error> {
let mut builder = TorClientConfigBuilder::from_directories(&state_path, cache_path);
builder.address_filter().allow_onion_addrs(true);
// Configure bridge.
if let Some(bridge_line) = config.bridge.bridge_line {
let bridge: BridgeConfigBuilder = bridge_line
.parse()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
builder.bridges().bridges().push(bridge.clone());
match bridge.get_transport() {
None => {
return Err(Error::TorConfig(format!(
"No transport found at {}",
bridge_line
)))
}
Some(t) => {
// Now configure bridge transport. (Requires the "pt-client" feature)
let bin_path = config.bridge.bridge_bin_path.unwrap_or(t.to_owned());
let mut transport = TransportConfigBuilder::default();
transport
.protocols(vec![t
.parse()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?])
.path(CfgPath::new(bin_path))
.run_on_startup(true);
builder.bridges().transports().push(transport);
}
}
}
let config = builder
.build()
.map_err(|e| Error::TorConfig(format!("{:?}", e)))?;
// Launch client.
let r = runtime()?;
let client = TorClient::with_runtime(r)
.config(config.clone())
.create_unbootstrapped()
.map_err(|e| Error::TorProcess(format!("{:?}", e)))?;
let c = client.clone();
let res = client.runtime().block_on(async move {
let bootstrap = async || {
return match c.bootstrap().await {
Ok(_) => {
let mut percent = 0.0;
let mut prev_percent = 0.0;
while percent < 1.0 {
percent = c.bootstrap_status().as_frac();
if percent != prev_percent {
info!("Starting Tor {}%", percent * 100.0);
}
prev_percent = percent;
tokio::time::sleep(Duration::from_millis(1000)).await;
}
Ok(())
}
Err(e) => Err(e),
};
};
match c
.runtime()
.timeout(Duration::from_millis(BOOTSTRAP_TIMEOUT_MS), bootstrap())
.await
{
Ok(r) => {
match r {
Err(e) => Err(Error::TorProcess(format!("{:?}", e))),
Ok(_) => Ok(())
}
},
Err(e) => Err(Error::TorProcess(format!("{:?}", e))),
}
});
match res {
Ok(_) => {
info!("Tor client bootstrapped successfully");
Ok((client, config))
}
Err(e) => Err(e),
}
}
/// Launch Onion service proxy.
async fn run_service_proxy<S>(
client: Arc<TorClient<TokioNativeTlsRuntime>>,
addr: SocketAddr,
request: &mut S,
nickname: HsNickname,
) -> Result<(), Error>
where
S: futures::Stream<Item = tor_hsservice::RendRequest> + Unpin + Send + 'static,
{
let runtime = client.runtime().clone();
// Setup proxy to forward request from Tor address to local address.
let proxy_rule = ProxyRule::new(
ProxyPattern::one_port(80).map_err(|e| Error::TorConfig(format!("{}", e)))?,
ProxyAction::Forward(Encapsulation::Simple, TargetAddr::Inet(addr)),
);
let mut proxy_cfg_builder = ProxyConfigBuilder::default();
proxy_cfg_builder.set_proxy_ports(vec![proxy_rule]);
let proxy_cfg = proxy_cfg_builder
.build()
.map_err(|e| Error::TorConfig(format!("{}", e)))?;
let proxy = OnionServiceReverseProxy::new(proxy_cfg);
// Start proxy for launched service.
proxy
.handle_requests(runtime, nickname, request)
.await
.map_err(|e| Error::TorProcess(format!("{:?}", e)))
}
/// Save Onion service key to keystore.
fn add_service_key(
mistrust: &Mistrust,
key: &SecretKey,
hs_nickname: &HsNickname,
path: PathBuf,
) -> Result<(), Error> {
let arti_store = ArtiNativeKeystore::from_path_and_mistrust(path, mistrust)
.map_err(|e| Error::TorProcess(format!("{}", e)))?;
let key_manager = KeyMgrBuilder::default()
.primary_store(Box::new(arti_store))
.build()
.map_err(|e| Error::TorProcess(format!("{}", e)))?;
let expanded_sk =
ExpandedSecretKey::from_bytes(Sha512::default().chain_update(key).finalize().as_ref());
let mut sk_bytes = [0_u8; 64];
sk_bytes[0..32].copy_from_slice(&expanded_sk.scalar.to_bytes());
sk_bytes[32..64].copy_from_slice(&expanded_sk.hash_prefix);
match ExpandedKeypair::from_secret_key_bytes(sk_bytes) {
None => {
return Err(Error::TorProcess(
"Hidden service key can not be created".into(),
))
}
Some(expanded_kp) => {
key_manager
.insert(
HsIdKey::from(expanded_kp.public().clone()),
&HsIdPublicKeySpecifier::new(hs_nickname.clone()),
KeystoreSelector::Primary,
true,
)
.map_err(|e| Error::TorProcess(format!("{}", e)))?;
key_manager
.insert(
HsIdKeypair::from(expanded_kp),
&HsIdKeypairSpecifier::new(hs_nickname.clone()),
KeystoreSelector::Primary,
true,
)
.map_err(|e| Error::TorProcess(format!("{}", e)))?;
}
}
Ok(())
}
impls/src/tor/mod.rs
+26
View File
@@ -12,7 +12,33 @@
// See the License for the specific language governing permissions and
// limitations under the License.
pub mod arti;
pub mod bridge;
pub mod config;
pub mod process;
pub mod proxy;
/// Running Tor instance control.
pub struct Tor {
/// External to process control.
pub process: Option<process::TorProcess>,
/// Integrated service.
pub service: Option<std::sync::Arc<tor_hsservice::RunningOnionService>>,
/// Integrated client.
pub client:
Option<std::sync::Arc<arti_client::TorClient<tor_rtcompat::tokio::TokioNativeTlsRuntime>>>,
}
/// Arti client runtime wrapper.
#[derive(Clone)]
pub struct ArtiRuntimeWrapper {
pub runtime: tor_rtcompat::tokio::TokioNativeTlsRuntime,
}
impl ArtiRuntimeWrapper {
fn create() -> Result<ArtiRuntimeWrapper, std::io::Error> {
Ok(Self {
runtime: tor_rtcompat::tokio::TokioNativeTlsRuntime::create()?,
})
}
}