github-jwinterm/mwixnet
1
0
Fork 0
mirror of https://github.com/jwinterm/mwixnet.git synced 2026-10-19 07:08:33 +02:00
Code Issues Packages Projects Releases Wiki Activity

replace TorProcess with rust-native arti client and upgrade to tokio1

Browse Source
This commit is contained in:
scilio
2024-04-16 11:16:53 -04:00
parent df2cb31258
commit 75f2bdf2ba
21 changed files with 5663 additions and 2432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bin/mwixnet.rs
+305 -305
View File
@@ -1,372 +1,372 @@
use mwixnet::config::{self, ServerConfig};
use mwixnet::node::HttpGrinNode;
use mwixnet::servers;
use mwixnet::store::SwapStore;
use mwixnet::tor;
use mwixnet::wallet::HttpWallet;
#[macro_use]
extern crate clap;
use clap::App;
use grin_core::global;
use grin_core::global::ChainTypes;
use grin_onion::crypto;
use grin_onion::crypto::dalek::DalekPublicKey;
use grin_util::{StopState, ZeroingString};
use mwixnet::client::{MixClient, MixClientImpl};
use mwixnet::node::GrinNode;
use mwixnet::store::StoreError;
use rand::{thread_rng, Rng};
use rpassword;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::thread::{sleep, spawn};
use std::time::Duration;
#[macro_use]
extern crate clap;
use clap::App;
use grin_core::global;
use grin_core::global::ChainTypes;
use grin_util::{StopState, ZeroingString};
use rand::{Rng, thread_rng};
use rpassword;
use tor_rtcompat::PreferredRuntime;
use grin_onion::crypto;
use grin_onion::crypto::dalek::DalekPublicKey;
use mwixnet::config::{self, ServerConfig};
use mwixnet::mix_client::{MixClient, MixClientImpl};
use mwixnet::node::GrinNode;
use mwixnet::node::HttpGrinNode;
use mwixnet::servers;
use mwixnet::store::StoreError;
use mwixnet::store::SwapStore;
use mwixnet::tor;
use mwixnet::wallet::HttpWallet;
const DEFAULT_INTERVAL: u32 = 12 * 60 * 60;
fn main() -> Result<(), Box<dyn std::error::Error>> {
real_main()?;
std::process::exit(0);
real_main()?;
std::process::exit(0);
}
fn real_main() -> Result<(), Box<dyn std::error::Error>> {
let yml = load_yaml!("mwixnet.yml");
let args = App::from_yaml(yml).get_matches();
let chain_type = if args.is_present("testnet") {
ChainTypes::Testnet
} else {
ChainTypes::Mainnet
};
global::set_local_chain_type(chain_type);
let yml = load_yaml!("mwixnet.yml");
let args = App::from_yaml(yml).get_matches();
let chain_type = if args.is_present("testnet") {
ChainTypes::Testnet
} else {
ChainTypes::Mainnet
};
global::set_local_chain_type(chain_type);
let config_path = match args.value_of("config_file") {
Some(path) => PathBuf::from(path),
None => {
let mut grin_path = config::get_grin_path(&chain_type);
grin_path.push("mwixnet-config.toml");
grin_path
}
};
let config_path = match args.value_of("config_file") {
Some(path) => PathBuf::from(path),
None => {
let mut grin_path = config::get_grin_path(&chain_type);
grin_path.push("mwixnet-config.toml");
grin_path
}
};
let round_time = args
.value_of("round_time")
.map(|t| t.parse::<u32>().unwrap());
let bind_addr = args.value_of("bind_addr");
let socks_addr = args.value_of("socks_addr");
let grin_node_url = args.value_of("grin_node_url");
let grin_node_secret_path = args.value_of("grin_node_secret_path");
let wallet_owner_url = args.value_of("wallet_owner_url");
let wallet_owner_secret_path = args.value_of("wallet_owner_secret_path");
let prev_server = args
.value_of("prev_server")
.map(|p| DalekPublicKey::from_hex(&p).unwrap());
let next_server = args
.value_of("next_server")
.map(|p| DalekPublicKey::from_hex(&p).unwrap());
let round_time = args
.value_of("round_time")
.map(|t| t.parse::<u32>().unwrap());
let bind_addr = args.value_of("bind_addr");
let grin_node_url = args.value_of("grin_node_url");
let grin_node_secret_path = args.value_of("grin_node_secret_path");
let wallet_owner_url = args.value_of("wallet_owner_url");
let wallet_owner_secret_path = args.value_of("wallet_owner_secret_path");
let prev_server = args
.value_of("prev_server")
.map(|p| DalekPublicKey::from_hex(&p).unwrap());
let next_server = args
.value_of("next_server")
.map(|p| DalekPublicKey::from_hex(&p).unwrap());
// Write a new config file if init-config command is supplied
if let ("init-config", Some(_)) = args.subcommand() {
if config_path.exists() {
panic!(
"Config file already exists at {}",
config_path.to_string_lossy()
);
}
// Write a new config file if init-config command is supplied
if let ("init-config", Some(_)) = args.subcommand() {
if config_path.exists() {
panic!(
"Config file already exists at {}",
config_path.to_string_lossy()
);
}
let server_config = ServerConfig {
key: crypto::secp::random_secret(),
interval_s: round_time.unwrap_or(DEFAULT_INTERVAL),
addr: bind_addr.unwrap_or("127.0.0.1:3000").parse()?,
socks_proxy_addr: socks_addr.unwrap_or("127.0.0.1:3001").parse()?,
grin_node_url: match grin_node_url {
Some(u) => u.parse()?,
None => config::grin_node_url(&chain_type),
},
grin_node_secret_path: match grin_node_secret_path {
Some(p) => Some(p.to_owned()),
None => config::node_secret_path(&chain_type)
.to_str()
.map(|p| p.to_owned()),
},
wallet_owner_url: match wallet_owner_url {
Some(u) => u.parse()?,
None => config::wallet_owner_url(&chain_type),
},
wallet_owner_secret_path: match wallet_owner_secret_path {
Some(p) => Some(p.to_owned()),
None => config::wallet_owner_secret_path(&chain_type)
.to_str()
.map(|p| p.to_owned()),
},
prev_server,
next_server,
};
let server_config = ServerConfig {
key: crypto::secp::random_secret(),
interval_s: round_time.unwrap_or(DEFAULT_INTERVAL),
addr: bind_addr.unwrap_or("127.0.0.1:3000").parse()?,
grin_node_url: match grin_node_url {
Some(u) => u.parse()?,
None => config::grin_node_url(&chain_type),
},
grin_node_secret_path: match grin_node_secret_path {
Some(p) => Some(p.to_owned()),
None => config::node_secret_path(&chain_type)
.to_str()
.map(|p| p.to_owned()),
},
wallet_owner_url: match wallet_owner_url {
Some(u) => u.parse()?,
None => config::wallet_owner_url(&chain_type),
},
wallet_owner_secret_path: match wallet_owner_secret_path {
Some(p) => Some(p.to_owned()),
None => config::wallet_owner_secret_path(&chain_type)
.to_str()
.map(|p| p.to_owned()),
},
prev_server,
next_server,
};
let password = prompt_password_confirm();
config::write_config(&config_path, &server_config, &password)?;
println!(
"Config file written to {:?}. Please back this file up in a safe place.",
config_path
);
return Ok(());
}
let password = prompt_password_confirm();
config::write_config(&config_path, &server_config, &password)?;
println!(
"Config file written to {:?}. Please back this file up in a safe place.",
config_path
);
return Ok(());
}
let password = prompt_password();
let mut server_config = config::load_config(&config_path, &password)?;
let password = prompt_password();
let mut server_config = config::load_config(&config_path, &password)?;
// Override grin_node_url, if supplied
if let Some(grin_node_url) = grin_node_url {
server_config.grin_node_url = grin_node_url.parse()?;
}
// Override grin_node_url, if supplied
if let Some(grin_node_url) = grin_node_url {
server_config.grin_node_url = grin_node_url.parse()?;
}
// Override grin_node_secret_path, if supplied
if let Some(grin_node_secret_path) = grin_node_secret_path {
server_config.grin_node_secret_path = Some(grin_node_secret_path.to_owned());
}
// Override grin_node_secret_path, if supplied
if let Some(grin_node_secret_path) = grin_node_secret_path {
server_config.grin_node_secret_path = Some(grin_node_secret_path.to_owned());
}
// Override wallet_owner_url, if supplied
if let Some(wallet_owner_url) = wallet_owner_url {
server_config.wallet_owner_url = wallet_owner_url.parse()?;
}
// Override wallet_owner_url, if supplied
if let Some(wallet_owner_url) = wallet_owner_url {
server_config.wallet_owner_url = wallet_owner_url.parse()?;
}
// Override wallet_owner_secret_path, if supplied
if let Some(wallet_owner_secret_path) = wallet_owner_secret_path {
server_config.wallet_owner_secret_path = Some(wallet_owner_secret_path.to_owned());
}
// Override wallet_owner_secret_path, if supplied
if let Some(wallet_owner_secret_path) = wallet_owner_secret_path {
server_config.wallet_owner_secret_path = Some(wallet_owner_secret_path.to_owned());
}
// Override bind_addr, if supplied
if let Some(bind_addr) = bind_addr {
server_config.addr = bind_addr.parse()?;
}
// Override bind_addr, if supplied
if let Some(bind_addr) = bind_addr {
server_config.addr = bind_addr.parse()?;
}
// Override socks_addr, if supplied
if let Some(socks_addr) = socks_addr {
server_config.socks_proxy_addr = socks_addr.parse()?;
}
// Override prev_server, if supplied
if let Some(prev_server) = prev_server {
server_config.prev_server = Some(prev_server);
}
// Override prev_server, if supplied
if let Some(prev_server) = prev_server {
server_config.prev_server = Some(prev_server);
}
// Override next_server, if supplied
if let Some(next_server) = next_server {
server_config.next_server = Some(next_server);
}
// Override next_server, if supplied
if let Some(next_server) = next_server {
server_config.next_server = Some(next_server);
}
// Create GrinNode
let node = HttpGrinNode::new(
&server_config.grin_node_url,
&server_config.node_api_secret(),
);
// Create GrinNode
let node = HttpGrinNode::new(
&server_config.grin_node_url,
&server_config.node_api_secret(),
);
// Node API health check
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
if let Err(e) = rt.block_on(node.async_get_chain_tip()) {
eprintln!("Node communication failure. Is node listening?");
return Err(e.into());
};
// Node API health check
let mut rt = tokio::runtime::Builder::new()
.threaded_scheduler()
.enable_all()
.build()?;
if let Err(e) = rt.block_on(node.async_get_chain_tip()) {
eprintln!("Node communication failure. Is node listening?");
return Err(e.into());
};
// Open wallet
let wallet_pass = prompt_wallet_password(&args.value_of("wallet_pass"));
let wallet = rt.block_on(HttpWallet::async_open_wallet(
&server_config.wallet_owner_url,
&server_config.wallet_owner_api_secret(),
&wallet_pass,
));
let wallet = match wallet {
Ok(w) => w,
Err(e) => {
eprintln!("Wallet communication failure. Is wallet listening?");
return Err(e.into());
}
};
// Open wallet
let wallet_pass = prompt_wallet_password(&args.value_of("wallet_pass"));
let wallet = rt.block_on(HttpWallet::async_open_wallet(
&server_config.wallet_owner_url,
&server_config.wallet_owner_api_secret(),
&wallet_pass,
));
let wallet = match wallet {
Ok(w) => w,
Err(e) => {
eprintln!("Wallet communication failure. Is wallet listening?");
return Err(e.into());
}
};
let tor_instance = rt.block_on(tor::async_init_tor(
PreferredRuntime::current().unwrap(),
&config::get_grin_path(&chain_type).to_str().unwrap(),
&server_config,
))?;
let tor_instance = Arc::new(grin_util::Mutex::new(tor_instance));
let tor_clone = tor_instance.clone();
let mut tor_process = tor::init_tor_listener(
&config::get_grin_path(&chain_type).to_str().unwrap(),
&server_config,
)?;
let stop_state = Arc::new(StopState::new());
let stop_state_clone = stop_state.clone();
let stop_state = Arc::new(StopState::new());
let stop_state_clone = stop_state.clone();
rt.spawn(async move {
futures::executor::block_on(build_signals_fut());
tor_clone.lock().stop();
stop_state_clone.stop();
});
rt.spawn(async move {
futures::executor::block_on(build_signals_fut());
let _ = tor_process.kill();
stop_state_clone.stop();
});
let next_mixer: Option<Arc<dyn MixClient>> = server_config.next_server.clone().map(|pk| {
let client: Arc<dyn MixClient> = Arc::new(MixClientImpl::new(
server_config.clone(),
tor_instance.clone(),
pk.clone(),
));
client
});
let next_mixer: Option<Arc<dyn MixClient>> = server_config.next_server.clone().map(|pk| {
let client: Arc<dyn MixClient> =
Arc::new(MixClientImpl::new(server_config.clone(), pk.clone()));
client
});
if server_config.prev_server.is_some() {
// Start the JSON-RPC HTTP 'mix' server
println!(
"Starting MIX server with public key {:?}",
server_config.server_pubkey().to_hex()
);
if server_config.prev_server.is_some() {
// Start the JSON-RPC HTTP 'mix' server
println!(
"Starting MIX server with public key {:?}",
server_config.server_pubkey().to_hex()
);
let (_, http_server) = servers::mix_rpc::listen(
rt.handle(),
server_config,
next_mixer,
Arc::new(wallet),
Arc::new(node),
)?;
let (_, http_server) = servers::mix_rpc::listen(
rt.handle(),
server_config,
next_mixer,
Arc::new(wallet),
Arc::new(node),
)?;
let close_handle = http_server.close_handle();
let round_handle = spawn(move || loop {
if stop_state.is_stopped() {
close_handle.close();
break;
}
let close_handle = http_server.close_handle();
let round_handle = spawn(move || loop {
if stop_state.is_stopped() {
close_handle.close();
break;
}
sleep(Duration::from_millis(100));
});
sleep(Duration::from_millis(100));
});
http_server.wait();
round_handle.join().unwrap();
} else {
println!(
"Starting SWAP server with public key {:?}",
server_config.server_pubkey().to_hex()
);
http_server.wait();
round_handle.join().unwrap();
} else {
println!(
"Starting SWAP server with public key {:?}",
server_config.server_pubkey().to_hex()
);
// Open SwapStore
let store = SwapStore::new(
config::get_grin_path(&chain_type)
.join("db")
.to_str()
.ok_or(StoreError::OpenError(grin_store::lmdb::Error::FileErr(
"db_root path error".to_string(),
)))?,
)?;
// Open SwapStore
let store = SwapStore::new(
config::get_grin_path(&chain_type)
.join("db")
.to_str()
.ok_or(StoreError::OpenError(grin_store::lmdb::Error::FileErr(
"db_root path error".to_string(),
)))?,
)?;
// Start the mwixnet JSON-RPC HTTP 'swap' server
let (swap_server, http_server) = servers::swap_rpc::listen(
rt.handle(),
&server_config,
next_mixer,
Arc::new(wallet),
Arc::new(node),
store,
)?;
// Start the mwixnet JSON-RPC HTTP 'swap' server
let (swap_server, http_server) = servers::swap_rpc::listen(
rt.handle(),
&server_config,
next_mixer,
Arc::new(wallet),
Arc::new(node),
store,
)?;
let close_handle = http_server.close_handle();
let round_handle = spawn(move || {
let mut rng = thread_rng();
let mut secs = 0u32;
let mut reorg_secs = 0u32;
let mut reorg_window = rng.gen_range(900u32, 3600u32);
let prev_tx = Arc::new(Mutex::new(None));
let server = swap_server.clone();
let close_handle = http_server.close_handle();
let round_handle = spawn(move || {
let mut rng = thread_rng();
let mut secs = 0u32;
let mut reorg_secs = 0u32;
let mut reorg_window = rng.gen_range(900u32, 3600u32);
let prev_tx = Arc::new(Mutex::new(None));
let server = swap_server.clone();
loop {
if stop_state.is_stopped() {
close_handle.close();
break;
}
loop {
if stop_state.is_stopped() {
close_handle.close();
break;
}
sleep(Duration::from_secs(1));
secs = (secs + 1) % server_config.interval_s;
reorg_secs = (reorg_secs + 1) % reorg_window;
sleep(Duration::from_secs(1));
secs = (secs + 1) % server_config.interval_s;
reorg_secs = (reorg_secs + 1) % reorg_window;
if secs == 0 {
let prev_tx_clone = prev_tx.clone();
let server_clone = server.clone();
rt.spawn(async move {
let result = server_clone.lock().await.execute_round().await;
let mut prev_tx_lock = prev_tx_clone.lock().unwrap();
*prev_tx_lock = match result {
Ok(Some(tx)) => Some(tx),
_ => None,
};
});
reorg_secs = 0;
reorg_window = rng.gen_range(900u32, 3600u32);
} else if reorg_secs == 0 {
let prev_tx_clone = prev_tx.clone();
let server_clone = server.clone();
rt.spawn(async move {
let tx_option = {
let prev_tx_lock = prev_tx_clone.lock().unwrap();
prev_tx_lock.clone()
}; // Lock is dropped here
if secs == 0 {
let prev_tx_clone = prev_tx.clone();
let server_clone = server.clone();
rt.spawn(async move {
let result = server_clone.lock().await.execute_round().await;
let mut prev_tx_lock = prev_tx_clone.lock().unwrap();
*prev_tx_lock = match result {
Ok(Some(tx)) => Some(tx),
_ => None,
};
});
reorg_secs = 0;
reorg_window = rng.gen_range(900u32, 3600u32);
} else if reorg_secs == 0 {
let prev_tx_clone = prev_tx.clone();
let server_clone = server.clone();
rt.spawn(async move {
let tx_option = {
let prev_tx_lock = prev_tx_clone.lock().unwrap();
prev_tx_lock.clone()
}; // Lock is dropped here
if let Some(tx) = tx_option {
let result = server_clone.lock().await.check_reorg(&tx).await;
let mut prev_tx_lock = prev_tx_clone.lock().unwrap();
*prev_tx_lock = match result {
Ok(Some(tx)) => Some(tx),
_ => None,
};
}
});
reorg_window = rng.gen_range(900u32, 3600u32);
}
}
});
if let Some(tx) = tx_option {
let result = server_clone.lock().await.check_reorg(tx).await;
let mut prev_tx_lock = prev_tx_clone.lock().unwrap();
*prev_tx_lock = match result {
Ok(Some(tx)) => Some(tx),
_ => None,
};
}
});
reorg_window = rng.gen_range(900u32, 3600u32);
}
}
});
http_server.wait();
round_handle.join().unwrap();
}
http_server.wait();
round_handle.join().unwrap();
}
Ok(())
Ok(())
}
#[cfg(unix)]
async fn build_signals_fut() {
use tokio::signal::unix::{signal, SignalKind};
use tokio::signal::unix::{signal, SignalKind};
// Listen for SIGINT, SIGQUIT, and SIGTERM
let mut terminate_signal =
signal(SignalKind::terminate()).expect("failed to create terminate signal");
let mut quit_signal = signal(SignalKind::quit()).expect("failed to create quit signal");
let mut interrupt_signal =
signal(SignalKind::interrupt()).expect("failed to create interrupt signal");
// Listen for SIGINT, SIGQUIT, and SIGTERM
let mut terminate_signal =
signal(SignalKind::terminate()).expect("failed to create terminate signal");
let mut quit_signal = signal(SignalKind::quit()).expect("failed to create quit signal");
let mut interrupt_signal =
signal(SignalKind::interrupt()).expect("failed to create interrupt signal");
futures::future::select_all(vec![
Box::pin(terminate_signal.recv()),
Box::pin(quit_signal.recv()),
Box::pin(interrupt_signal.recv()),
])
.await;
futures::future::select_all(vec![
Box::pin(terminate_signal.recv()),
Box::pin(quit_signal.recv()),
Box::pin(interrupt_signal.recv()),
])
.await;
}
#[cfg(not(unix))]
async fn build_signals_fut() {
tokio::signal::ctrl_c()
.await
.expect("failed to install CTRL+C signal handler");
tokio::signal::ctrl_c()
.await
.expect("failed to install CTRL+C signal handler");
}
fn prompt_password() -> ZeroingString {
ZeroingString::from(rpassword::prompt_password_stdout("Server password: ").unwrap())
ZeroingString::from(rpassword::prompt_password_stdout("Server password: ").unwrap())
}
fn prompt_password_confirm() -> ZeroingString {
let mut first = "first".to_string();
let mut second = "second".to_string();
while first != second {
first = rpassword::prompt_password_stdout("Server password: ").unwrap();
second = rpassword::prompt_password_stdout("Confirm server password: ").unwrap();
}
ZeroingString::from(first)
let mut first = "first".to_string();
let mut second = "second".to_string();
while first != second {
first = rpassword::prompt_password_stdout("Server password: ").unwrap();
second = rpassword::prompt_password_stdout("Confirm server password: ").unwrap();
}
ZeroingString::from(first)
}
fn prompt_wallet_password(wallet_pass: &Option<&str>) -> ZeroingString {
match *wallet_pass {
Some(wallet_pass) => ZeroingString::from(wallet_pass),
None => {
ZeroingString::from(rpassword::prompt_password_stdout("Wallet password: ").unwrap())
}
}
match *wallet_pass {
Some(wallet_pass) => ZeroingString::from(wallet_pass),
None => {
ZeroingString::from(rpassword::prompt_password_stdout("Wallet password: ").unwrap())
}
}
}
src/bin/mwixnet.yml
-4
View File
@@ -38,10 +38,6 @@ args:
help: Address to bind the rpc server to (e.g. 127.0.0.1:3000)
long: bind_addr
takes_value: true
- socks_addr:
help: Address to bind the SOCKS5 tor proxy to (e.g. 127.0.0.1:3001)
long: socks_addr
takes_value: true
- prev_server:
help: Hex public key of the previous swap/mix server
long: prev_server
src/config.rs
+17 -22
View File
@@ -1,20 +1,21 @@
use grin_onion::crypto::dalek::DalekPublicKey;
use grin_onion::crypto::secp::SecretKey;
use core::num::NonZeroU32;
use grin_core::global::ChainTypes;
use grin_util::{file, ToHex, ZeroingString};
use grin_wallet_util::OnionV3Address;
use rand::{thread_rng, Rng};
use ring::{aead, pbkdf2};
use serde_derive::{Deserialize, Serialize};
use std::fs::File;
use std::io::prelude::*;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::result::Result;
use grin_core::global::ChainTypes;
use grin_util::{file, ToHex, ZeroingString};
use grin_wallet_util::OnionV3Address;
use rand::{Rng, thread_rng};
use ring::{aead, pbkdf2};
use serde_derive::{Deserialize, Serialize};
use thiserror::Error;
use grin_onion::crypto::dalek::DalekPublicKey;
use grin_onion::crypto::secp::SecretKey;
const GRIN_HOME: &str = ".grin";
const NODE_API_SECRET_FILE_NAME: &str = ".api_secret";
const WALLET_OWNER_API_SECRET_FILE_NAME: &str = ".owner_api_secret";
@@ -28,8 +29,6 @@ pub struct ServerConfig {
pub interval_s: u32,
/// socket address the server listener should bind to
pub addr: SocketAddr,
/// socket address the tor sender should bind to
pub socks_proxy_addr: SocketAddr,
/// foreign api address of the grin node
pub grin_node_url: SocketAddr,
/// path to file containing api secret for the grin node
@@ -181,7 +180,6 @@ struct RawConfig {
nonce: String,
interval_s: u32,
addr: SocketAddr,
socks_proxy_addr: SocketAddr,
grin_node_url: SocketAddr,
grin_node_secret_path: Option<String>,
wallet_owner_url: SocketAddr,
@@ -206,7 +204,6 @@ pub fn write_config(
nonce: encrypted.nonce,
interval_s: server_config.interval_s,
addr: server_config.addr,
socks_proxy_addr: server_config.socks_proxy_addr,
grin_node_url: server_config.grin_node_url,
grin_node_secret_path: server_config.grin_node_secret_path.clone(),
wallet_owner_url: server_config.wallet_owner_url,
@@ -243,7 +240,6 @@ pub fn load_config(
key: secret_key,
interval_s: raw_config.interval_s,
addr: raw_config.addr,
socks_proxy_addr: raw_config.socks_proxy_addr,
grin_node_url: raw_config.grin_node_url,
grin_node_secret_path: raw_config.grin_node_secret_path,
wallet_owner_url: raw_config.wallet_owner_url,
@@ -254,10 +250,7 @@ pub fn load_config(
}
pub fn get_grin_path(chain_type: &ChainTypes) -> PathBuf {
let mut grin_path = match dirs::home_dir() {
Some(p) => p,
None => PathBuf::new(),
};
let mut grin_path = dirs::home_dir().unwrap_or_else(|| PathBuf::new());
grin_path.push(GRIN_HOME);
grin_path.push(chain_type.shortname());
grin_path
@@ -289,10 +282,12 @@ pub fn wallet_owner_url(_chain_type: &ChainTypes) -> SocketAddr {
#[cfg(test)]
pub mod test_util {
use crate::config::ServerConfig;
use std::net::TcpListener;
use grin_onion::crypto::dalek::DalekPublicKey;
use secp256k1zkp::SecretKey;
use std::net::TcpListener;
use crate::config::ServerConfig;
pub fn local_config(
server_key: &SecretKey,
@@ -303,7 +298,6 @@ pub mod test_util {
key: server_key.clone(),
interval_s: 1,
addr: TcpListener::bind("127.0.0.1:0")?.local_addr()?,
socks_proxy_addr: TcpListener::bind("127.0.0.1:0")?.local_addr()?,
grin_node_url: "127.0.0.1:3413".parse()?,
grin_node_secret_path: None,
wallet_owner_url: "127.0.0.1:3420".parse()?,
@@ -317,9 +311,10 @@ pub mod test_util {
#[cfg(test)]
mod tests {
use super::*;
use grin_onion::crypto::secp;
use super::*;
#[test]
fn server_key_encrypt() {
let password = ZeroingString::from("password");
src/http.rs
+127
View File
@@ -0,0 +1,127 @@
use std::time::Duration;
use grin_api::json_rpc;
use grin_util::to_base64;
use grin_wallet_api::{EncryptedRequest, EncryptedResponse, JsonId};
use hyper::body::Body as HyperBody;
use hyper::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, USER_AGENT};
use hyper::Request;
use serde_json::json;
use thiserror::Error;
use secp256k1zkp::SecretKey;
/// Error types for HTTP client connections
#[derive(Error, Debug)]
pub enum HttpError {
#[error("Error decrypting response")]
DecryptResponseError(),
#[error("Hyper HTTP error: {0:?}")]
HyperHttpError(hyper::http::Error),
#[error("Hyper request failed with error: {0:?}")]
RequestFailed(hyper::Error),
#[error("Error with response body: {0:?}")]
ResponseBodyError(hyper::Error),
#[error("Error deserializing JSON response: {0:?}")]
ResponseJsonError(serde_json::Error),
#[error("Error decoding JSON-RPC response: {0:?}")]
ResponseParseError(json_rpc::Error),
#[error("Wrong response code: {0}")]
ResponseStatusError(hyper::StatusCode),
}
pub async fn async_send_enc_request<D: serde::de::DeserializeOwned>(
url: &String,
api_secret: &Option<String>,
method: &str,
params: &serde_json::Value,
shared_key: &SecretKey,
) -> Result<D, HttpError> {
let req = json!({
"method": method,
"params": params,
"id": JsonId::IntId(1),
"jsonrpc": "2.0",
});
let enc_req = EncryptedRequest::from_json(&JsonId::IntId(1), &req, &shared_key).unwrap();
let req = build_request(&url, &api_secret, serde_json::to_string(&enc_req).unwrap())?;
let response_str = send_request_async(req).await?;
let enc_res: EncryptedResponse =
serde_json::from_str(&response_str).map_err(HttpError::ResponseJsonError)?;
let decrypted = enc_res
.decrypt(&shared_key)
.map_err(|_| HttpError::DecryptResponseError())?;
let response: json_rpc::Response =
serde_json::from_value(decrypted).map_err(HttpError::ResponseJsonError)?;
let parsed = response
.clone()
.into_result()
.map_err(HttpError::ResponseParseError)?;
Ok(parsed)
}
pub async fn async_send_json_request<D: serde::de::DeserializeOwned>(
url: &String,
api_secret: &Option<String>,
method: &str,
params: &serde_json::Value,
) -> Result<D, HttpError> {
let req_body = json!({
"method": method,
"params": params,
"id": 1,
"jsonrpc": "2.0",
});
let req = build_request(&url, &api_secret, serde_json::to_string(&req_body).unwrap())?;
let data = send_request_async(req).await?;
let ser: json_rpc::Response =
serde_json::from_str(&data).map_err(HttpError::ResponseJsonError)?;
let parsed = ser
.clone()
.into_result()
.map_err(HttpError::ResponseParseError)?;
Ok(parsed)
}
pub fn build_request(
url: &String,
api_secret: &Option<String>,
req_body: String,
) -> Result<Request<HyperBody>, HttpError> {
let mut req_builder = hyper::Request::builder();
if let Some(api_secret) = api_secret {
let basic_auth = format!("Basic {}", to_base64(&format!("grin:{}", api_secret)));
req_builder = req_builder.header(AUTHORIZATION, basic_auth);
}
req_builder
.method(hyper::Method::POST)
.uri(url)
.header(USER_AGENT, "grin-client")
.header(ACCEPT, "application/json")
.header(CONTENT_TYPE, "application/json")
.body(HyperBody::from(req_body))
.map_err(HttpError::HyperHttpError)
}
async fn send_request_async(req: Request<HyperBody>) -> Result<String, HttpError> {
let client = hyper::Client::builder()
.pool_idle_timeout(Duration::from_secs(30))
.build_http();
let resp = client
.request(req)
.await
.map_err(HttpError::RequestFailed)?;
if !resp.status().is_success() {
return Err(HttpError::ResponseStatusError(resp.status()));
}
let raw = hyper::body::to_bytes(resp)
.await
.map_err(HttpError::ResponseBodyError)?;
Ok(String::from_utf8_lossy(&raw).to_string())
}
src/lib.rs
+3 -2
View File
@@ -1,8 +1,9 @@
#[macro_use]
extern crate log;
pub mod client;
pub mod config;
pub mod http;
pub mod mix_client;
pub mod node;
pub mod servers;
pub mod store;
@@ -10,8 +11,8 @@ pub mod tor;
pub mod tx;
pub mod wallet;
pub use client::MixClient;
pub use config::ServerConfig;
pub use mix_client::MixClient;
pub use node::{GrinNode, HttpGrinNode, NodeError};
pub use servers::mix::{MixError, MixServer};
pub use servers::mix_rpc::listen as mix_listen;
src/client.rs → src/mix_client.rs
+59 -81
View File
@@ -1,27 +1,30 @@
use crate::config::ServerConfig;
use crate::servers::mix_rpc::{MixReq, MixResp};
use crate::tor;
use grin_onion::crypto::dalek::{self, DalekPublicKey};
use grin_onion::onion::Onion;
use std::sync::Arc;
use async_trait::async_trait;
use grin_api::json_rpc::{build_request, Response};
use grin_core::ser;
use grin_core::ser::ProtocolVersion;
use grin_wallet_util::OnionV3Address;
use hyper::client::HttpConnector;
use hyper::header::{ACCEPT, CONTENT_TYPE, USER_AGENT};
use hyper_socks2::SocksConnector;
use serde_json;
use serde_json::json;
use thiserror::Error;
use tor_rtcompat::Runtime;
use grin_onion::crypto::dalek::{self, DalekPublicKey};
use grin_onion::onion::Onion;
use crate::config::ServerConfig;
use crate::servers::mix_rpc::{MixReq, MixResp};
use crate::tor::TorService;
use crate::{http, tor};
/// Error types for interacting with nodes
#[derive(Error, Debug)]
pub enum ClientError {
pub enum MixClientError {
#[error("Tor Error: {0:?}")]
Tor(tor::TorError),
#[error("API Error: {0:?}")]
API(grin_api::Error),
#[error("Communication Error: {0:?}")]
CommError(http::HttpError),
#[error("Dalek Error: {0:?}")]
Dalek(dalek::DalekError),
#[error("Error decoding JSON response: {0:?}")]
@@ -36,18 +39,23 @@ pub enum ClientError {
#[async_trait]
pub trait MixClient: Send + Sync {
/// Swaps the outputs provided and returns the final swapped outputs and kernels.
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError>;
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, MixClientError>;
}
pub struct MixClientImpl {
pub struct MixClientImpl<R: Runtime> {
config: ServerConfig,
tor: Arc<grin_util::Mutex<TorService<R>>>,
addr: OnionV3Address,
}
impl MixClientImpl {
pub fn new(config: ServerConfig, next_pubkey: DalekPublicKey) -> Self {
impl<R: Runtime> MixClientImpl<R> {
pub fn new(
config: ServerConfig,
tor: Arc<grin_util::Mutex<TorService<R>>>,
next_pubkey: DalekPublicKey,
) -> Self {
let addr = OnionV3Address::from_bytes(next_pubkey.as_ref().to_bytes());
MixClientImpl { config, addr }
MixClientImpl { config, tor, addr }
}
async fn async_send_json_request<D: serde::de::DeserializeOwned>(
@@ -55,61 +63,29 @@ impl MixClientImpl {
addr: &OnionV3Address,
method: &str,
params: &serde_json::Value,
) -> Result<D, ClientError> {
let proxy = {
let proxy_uri = format!(
"socks5://{}:{}",
self.config.socks_proxy_addr.ip(),
self.config.socks_proxy_addr.port()
)
.parse()
.unwrap();
let mut connector = HttpConnector::new();
connector.enforce_http(false);
let proxy_connector = SocksConnector {
proxy_addr: proxy_uri,
auth: None,
connector,
};
proxy_connector
};
) -> Result<D, MixClientError> {
let url = format!("{}/v1", addr.to_http_str());
let request_str = serde_json::to_string(&build_request(method, params)).unwrap();
let hyper_request =
http::build_request(&url, &None, request_str).map_err(MixClientError::CommError)?;
let body =
hyper::body::Body::from(serde_json::to_string(&build_request(method, params)).unwrap());
let req = hyper::Request::builder()
.method(hyper::Method::POST)
.uri(url)
.header(USER_AGENT, "grin-client")
.header(ACCEPT, "application/json")
.header(CONTENT_TYPE, "application/json")
.body(body)
.map_err(|e| {
ClientError::API(grin_api::Error::RequestError(format!(
"Cannot make request: {}",
e
)))
})?;
let client = hyper::Client::builder().build::<_, hyper::Body>(proxy);
let res = client.request(req).await.unwrap();
let hyper_client = self.tor.lock().new_hyper_client();
let res = hyper_client.request(hyper_request).await.unwrap();
let body_bytes = hyper::body::to_bytes(res.into_body()).await.unwrap();
let res = String::from_utf8(body_bytes.to_vec()).unwrap();
let response: Response =
serde_json::from_str(&res).map_err(ClientError::DecodeResponseError)?;
serde_json::from_str(&res).map_err(MixClientError::DecodeResponseError)?;
if let Some(ref e) = response.error {
return Err(ClientError::ResponseError(e.clone()));
return Err(MixClientError::ResponseError(e.clone()));
}
let result = match response.result.clone() {
Some(r) => serde_json::from_value(r).map_err(ClientError::DecodeResponseError),
Some(r) => serde_json::from_value(r).map_err(MixClientError::DecodeResponseError),
None => serde_json::from_value(serde_json::Value::Null)
.map_err(ClientError::DecodeResponseError),
.map_err(MixClientError::DecodeResponseError),
}?;
Ok(result)
@@ -117,34 +93,29 @@ impl MixClientImpl {
}
#[async_trait]
impl MixClient for MixClientImpl {
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError> {
impl<R: Runtime> MixClient for MixClientImpl<R> {
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, MixClientError> {
let serialized = ser::ser_vec(&onions, ProtocolVersion::local()).unwrap();
let sig =
dalek::sign(&self.config.key, serialized.as_slice()).map_err(ClientError::Dalek)?;
// println!(
// "Created sig ({:?}) with public key ({}) for server ({})",
// &sig,
// DalekPublicKey::from_secret(&self.config.key).to_hex(),
// self.config.next_server.as_ref().unwrap().to_hex()
// );
dalek::sign(&self.config.key, serialized.as_slice()).map_err(MixClientError::Dalek)?;
let mix = MixReq::new(onions.clone(), sig);
let params = serde_json::json!([mix]);
self.async_send_json_request::<MixResp>(&self.addr, "mix", &params)
self.async_send_json_request::<MixResp>(&self.addr, "mix", &json!([mix]))
.await
}
}
#[cfg(test)]
pub mod mock {
use super::{ClientError, MixClient};
use std::collections::HashMap;
use async_trait::async_trait;
use grin_onion::onion::Onion;
use crate::servers::mix_rpc::MixResp;
use async_trait::async_trait;
use std::collections::HashMap;
use super::{MixClient, MixClientError};
pub struct MockMixClient {
results: HashMap<Vec<Onion>, MixResp>,
@@ -164,27 +135,33 @@ pub mod mock {
#[async_trait]
impl MixClient for MockMixClient {
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError> {
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, MixClientError> {
self.results
.get(onions)
.map(|r| Ok(r.clone()))
.unwrap_or(Err(ClientError::Custom("No response set for input".into())))
.unwrap_or(Err(MixClientError::Custom(
"No response set for input".into(),
)))
}
}
}
#[cfg(test)]
pub mod test_util {
use super::{ClientError, MixClient};
use crate::servers::mix::MixServer;
use crate::servers::mix_rpc::MixResp;
use std::sync::Arc;
use async_trait::async_trait;
use grin_core::ser;
use grin_core::ser::ProtocolVersion;
use grin_onion::crypto::dalek::{self, DalekPublicKey};
use grin_onion::crypto::secp::SecretKey;
use grin_onion::onion::Onion;
use std::sync::Arc;
use crate::servers::mix::MixServer;
use crate::servers::mix_rpc::MixResp;
use super::{MixClient, MixClientError};
/// Implementation of the 'MixClient' trait that calls a mix server implementation directly.
/// No JSON-RPC serialization or socket communication occurs.
@@ -196,9 +173,10 @@ pub mod test_util {
#[async_trait]
impl MixClient for DirectMixClient {
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, ClientError> {
async fn mix_outputs(&self, onions: &Vec<Onion>) -> Result<MixResp, MixClientError> {
let serialized = ser::ser_vec(&onions, ProtocolVersion::local()).unwrap();
let sig = dalek::sign(&self.key, serialized.as_slice()).map_err(ClientError::Dalek)?;
let sig =
dalek::sign(&self.key, serialized.as_slice()).map_err(MixClientError::Dalek)?;
sig.verify(
&DalekPublicKey::from_secret(&self.key),
src/node.rs
+48 -46
View File
@@ -1,19 +1,20 @@
use grin_onion::crypto::secp::Commitment;
use grin_api::json_rpc::{build_request, Request, Response};
use grin_api::{client, LocatedTxKernel};
use grin_api::{OutputPrintable, OutputType, Tip};
use grin_core::consensus::COINBASE_MATURITY;
use grin_core::core::{Committed, Input, OutputFeatures, Transaction};
use grin_util::ToHex;
use async_trait::async_trait;
use serde_json::json;
use std::net::SocketAddr;
use std::sync::Arc;
use async_trait::async_trait;
use grin_api::LocatedTxKernel;
use grin_api::{OutputPrintable, OutputType, Tip};
use grin_core::consensus::COINBASE_MATURITY;
use grin_core::core::hash::Hash;
use grin_core::core::{Committed, Input, OutputFeatures, Transaction};
use grin_util::ToHex;
use serde_json::json;
use thiserror::Error;
use grin_onion::crypto::secp::Commitment;
use crate::http;
#[async_trait]
pub trait GrinNode: Send + Sync {
/// Retrieves the unspent output with a matching commitment
@@ -22,8 +23,8 @@ pub trait GrinNode: Send + Sync {
output_commit: &Commitment,
) -> Result<Option<OutputPrintable>, NodeError>;
/// Gets the height and hash of the chain tip
async fn async_get_chain_tip(&self) -> Result<(u64, Hash), NodeError>;
/// Gets the height and hash of the chain tip
async fn async_get_chain_tip(&self) -> Result<(u64, Hash), NodeError>;
/// Posts a transaction to the grin node
async fn async_post_tx(&self, tx: &Transaction) -> Result<(), NodeError>;
@@ -47,6 +48,8 @@ pub enum NodeError {
DecodeResponseError(serde_json::Error),
#[error("JSON-RPC API communication error: {0:?}")]
ApiCommError(grin_api::Error),
#[error("Client error: {0:?}")]
NodeCommError(http::HttpError),
#[error("Error decoding JSON-RPC response: {0:?}")]
ResponseParseError(grin_api::json_rpc::Error),
}
@@ -109,21 +112,24 @@ pub async fn async_build_input(
Ok(None)
}
pub async fn async_is_tx_valid(node: &Arc<dyn GrinNode>, tx: &Transaction) -> Result<bool, NodeError> {
let next_block_height = node.async_get_chain_tip().await?.0 + 1;
for input_commit in &tx.inputs_committed() {
if !async_is_spendable(&node, &input_commit, next_block_height).await? {
return Ok(false);
}
}
pub async fn async_is_tx_valid(
node: &Arc<dyn GrinNode>,
tx: &Transaction,
) -> Result<bool, NodeError> {
let next_block_height = node.async_get_chain_tip().await?.0 + 1;
for input_commit in &tx.inputs_committed() {
if !async_is_spendable(&node, &input_commit, next_block_height).await? {
return Ok(false);
}
}
for output_commit in &tx.outputs_committed() {
if async_is_unspent(&node, &output_commit).await? {
return Ok(false);
}
}
for output_commit in &tx.outputs_committed() {
if async_is_unspent(&node, &output_commit).await? {
return Ok(false);
}
}
Ok(true)
Ok(true)
}
/// HTTP (JSON-RPC) implementation of the 'GrinNode' trait
@@ -149,18 +155,9 @@ impl HttpGrinNode {
params: &serde_json::Value,
) -> Result<D, NodeError> {
let url = format!("http://{}{}", self.node_url, ENDPOINT);
let req = build_request(method, params);
let res = client::post_async::<Request, Response>(
url.as_str(),
&req,
self.node_api_secret.clone(),
)
.await
.map_err(NodeError::ApiCommError)?;
let parsed = res
.clone()
.into_result()
.map_err(NodeError::ResponseParseError)?;
let parsed = http::async_send_json_request(&url, &self.node_api_secret, &method, &params)
.await
.map_err(NodeError::NodeCommError)?;
Ok(parsed)
}
}
@@ -194,7 +191,7 @@ impl GrinNode for HttpGrinNode {
Ok(Some(outputs[0].clone()))
}
async fn async_get_chain_tip(&self) -> Result<(u64, Hash), NodeError> {
async fn async_get_chain_tip(&self) -> Result<(u64, Hash), NodeError> {
let params = json!([]);
let tip_json = self
.async_send_request::<serde_json::Value>("get_tip", &params)
@@ -202,7 +199,10 @@ impl GrinNode for HttpGrinNode {
let tip =
serde_json::from_value::<Tip>(tip_json).map_err(NodeError::DecodeResponseError)?;
Ok((tip.height, Hash::from_hex(tip.last_block_pushed.as_str()).unwrap()))
Ok((
tip.height,
Hash::from_hex(tip.last_block_pushed.as_str()).unwrap(),
))
}
async fn async_post_tx(&self, tx: &Transaction) -> Result<(), NodeError> {
@@ -236,15 +236,17 @@ impl GrinNode for HttpGrinNode {
#[cfg(test)]
pub mod mock {
use super::{GrinNode, NodeError};
use std::collections::HashMap;
use std::sync::RwLock;
use async_trait::async_trait;
use grin_api::{LocatedTxKernel, OutputPrintable, OutputType};
use grin_core::core::hash::Hash;
use grin_core::core::Transaction;
use grin_onion::crypto::secp::Commitment;
use std::collections::HashMap;
use std::sync::RwLock;
use grin_core::core::hash::Hash;
use super::{GrinNode, NodeError};
/// Implementation of 'GrinNode' trait that mocks a grin node instance.
/// Use only for testing purposes.
@@ -318,8 +320,8 @@ pub mod mock {
Ok(None)
}
async fn async_get_chain_tip(&self) -> Result<(u64, Hash), NodeError> {
Ok((100, Hash::default()))
async fn async_get_chain_tip(&self) -> Result<(u64, Hash), NodeError> {
Ok((100, Hash::default()))
}
async fn async_post_tx(&self, tx: &Transaction) -> Result<(), NodeError> {
src/servers/mix.rs
+28 -23
View File
@@ -1,24 +1,26 @@
use crate::client::MixClient;
use crate::config::ServerConfig;
use crate::node::{self, GrinNode};
use crate::tx::{self, TxComponents};
use crate::wallet::Wallet;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use crate::servers::mix_rpc::MixResp;
use async_trait::async_trait;
use futures::stream::{self, StreamExt};
use grin_core::core::{Output, OutputFeatures, TransactionBody};
use grin_core::global::DEFAULT_ACCEPT_FEE_BASE;
use grin_core::ser;
use grin_core::ser::ProtocolVersion;
use itertools::Itertools;
use thiserror::Error;
use grin_onion::crypto::dalek::{self, DalekSignature};
use grin_onion::onion::{Onion, OnionError, PeeledOnion};
use itertools::Itertools;
use secp256k1zkp::key::ZERO_KEY;
use secp256k1zkp::Secp256k1;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use thiserror::Error;
use crate::config::ServerConfig;
use crate::mix_client::MixClient;
use crate::node::{self, GrinNode};
use crate::servers::mix_rpc::MixResp;
use crate::tx::{self, TxComponents};
use crate::wallet::Wallet;
/// Mixer error types
#[derive(Error, Debug)]
@@ -46,7 +48,7 @@ pub enum MixError {
#[error("Wallet error: {0:?}")]
WalletError(crate::wallet::WalletError),
#[error("Client comm error: {0:?}")]
Client(crate::client::ClientError),
Client(crate::mix_client::MixClientError),
}
/// An internal MWixnet server - a "Mixer"
@@ -298,16 +300,17 @@ impl MixServer for MixServerImpl {
#[cfg(test)]
mod test_util {
use crate::client::test_util::DirectMixClient;
use crate::client::MixClient;
use crate::config;
use crate::node::mock::MockGrinNode;
use crate::servers::mix::MixServerImpl;
use crate::wallet::mock::MockWallet;
use std::sync::Arc;
use grin_onion::crypto::dalek::DalekPublicKey;
use secp256k1zkp::SecretKey;
use std::sync::Arc;
use crate::config;
use crate::mix_client::MixClient;
use crate::mix_client::test_util::DirectMixClient;
use crate::node::mock::MockGrinNode;
use crate::servers::mix::MixServerImpl;
use crate::wallet::mock::MockWallet;
pub fn new_mixer(
server_key: &SecretKey,
@@ -340,18 +343,20 @@ mod test_util {
#[cfg(test)]
mod tests {
use crate::client::MixClient;
use crate::node::mock::MockGrinNode;
use std::collections::HashSet;
use std::sync::Arc;
use ::function_name::named;
use grin_onion::{create_onion, Hop, new_hop};
use grin_onion::crypto::dalek::DalekPublicKey;
use grin_onion::crypto::secp::{self, Commitment};
use grin_onion::test_util as onion_test_util;
use grin_onion::{create_onion, new_hop, Hop};
use secp256k1zkp::pedersen::RangeProof;
use secp256k1zkp::SecretKey;
use std::collections::HashSet;
use std::sync::Arc;
use crate::mix_client::MixClient;
use crate::node::mock::MockGrinNode;
macro_rules! init_test {
() => {{
src/servers/mix_rpc.rs
+17 -16
View File
@@ -1,19 +1,20 @@
use crate::client::MixClient;
use crate::config::ServerConfig;
use crate::node::GrinNode;
use crate::servers::mix::{MixError, MixServer, MixServerImpl};
use crate::wallet::Wallet;
use std::sync::Arc;
use crate::tx::TxComponents;
use futures::FutureExt;
use jsonrpc_derive::rpc;
use jsonrpc_http_server::{DomainsValidation, ServerBuilder};
use jsonrpc_http_server::jsonrpc_core::{self, BoxFuture, IoHandler};
use serde::{Deserialize, Serialize};
use grin_onion::crypto::dalek::{self, DalekSignature};
use grin_onion::onion::Onion;
use jsonrpc_core::BoxFuture;
use jsonrpc_derive::rpc;
use jsonrpc_http_server::jsonrpc_core::{self as jsonrpc, IoHandler};
use jsonrpc_http_server::{DomainsValidation, ServerBuilder};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use crate::config::ServerConfig;
use crate::mix_client::MixClient;
use crate::node::GrinNode;
use crate::servers::mix::{MixError, MixServer, MixServerImpl};
use crate::tx::TxComponents;
use crate::wallet::Wallet;
#[derive(Serialize, Deserialize)]
pub struct MixReq {
@@ -37,7 +38,7 @@ impl MixReq {
#[rpc(server)]
pub trait MixAPI {
#[rpc(name = "mix")]
fn mix(&self, mix: MixReq) -> BoxFuture<jsonrpc::Result<MixResp>>;
fn mix(&self, mix: MixReq) -> BoxFuture<jsonrpc_core::Result<MixResp>>;
}
#[derive(Clone)]
@@ -67,14 +68,14 @@ impl RPCMixServer {
}
}
impl From<MixError> for jsonrpc::Error {
impl From<MixError> for jsonrpc_core::Error {
fn from(e: MixError) -> Self {
jsonrpc::Error::invalid_params(e.to_string())
jsonrpc_core::Error::invalid_params(e.to_string())
}
}
impl MixAPI for RPCMixServer {
fn mix(&self, mix: MixReq) -> BoxFuture<jsonrpc::Result<MixResp>> {
fn mix(&self, mix: MixReq) -> BoxFuture<jsonrpc_core::Result<MixResp>> {
let server = self.server.clone();
async move {
let response = server
src/servers/swap.rs
+835 -807
View File
File diff suppressed because it is too large Load Diff
src/servers/swap_rpc.rs
+29 -31
View File
@@ -1,20 +1,21 @@
use crate::client::MixClient;
use std::sync::Arc;
use futures::FutureExt;
use jsonrpc_core::{BoxFuture, Value};
use jsonrpc_derive::rpc;
use jsonrpc_http_server::{DomainsValidation, ServerBuilder};
use serde::{Deserialize, Serialize};
use grin_onion::crypto::comsig::{self, ComSignature};
use grin_onion::onion::Onion;
use crate::config::ServerConfig;
use crate::mix_client::MixClient;
use crate::node::GrinNode;
use crate::servers::swap::{SwapError, SwapServer, SwapServerImpl};
use crate::store::SwapStore;
use crate::wallet::Wallet;
use futures::FutureExt;
use grin_onion::crypto::comsig::{self, ComSignature};
use grin_onion::onion::Onion;
use jsonrpc_core::Value;
use jsonrpc_derive::rpc;
use jsonrpc_http_server::jsonrpc_core::*;
use jsonrpc_http_server::{DomainsValidation, ServerBuilder};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
#[derive(Serialize, Deserialize)]
pub struct SwapReq {
onion: Onion,
@@ -37,7 +38,7 @@ struct RPCSwapServer {
impl RPCSwapServer {
/// Spin up an instance of the JSON-RPC HTTP server.
fn start_http(&self, runtime_handle: tokio::runtime::Handle) -> jsonrpc_http_server::Server {
let mut io = IoHandler::new();
let mut io = jsonrpc_core::IoHandler::new();
io.extend_with(RPCSwapServer::to_delegate(self.clone()));
ServerBuilder::new(io)
@@ -55,15 +56,15 @@ impl RPCSwapServer {
}
}
impl From<SwapError> for Error {
impl From<SwapError> for jsonrpc_core::Error {
fn from(e: SwapError) -> Self {
match e {
SwapError::UnknownError(_) => Error {
SwapError::UnknownError(_) => jsonrpc_core::Error {
message: e.to_string(),
code: ErrorCode::InternalError,
code: jsonrpc_core::ErrorCode::InternalError,
data: None,
},
_ => Error::invalid_params(e.to_string()),
_ => jsonrpc_core::Error::invalid_params(e.to_string()),
}
}
}
@@ -115,20 +116,21 @@ pub fn listen(
#[cfg(test)]
mod tests {
use crate::config::ServerConfig;
use crate::servers::swap::mock::MockSwapServer;
use crate::servers::swap::{SwapError, SwapServer};
use crate::servers::swap_rpc::{RPCSwapServer, SwapReq};
use grin_onion::create_onion;
use grin_onion::crypto::comsig::ComSignature;
use grin_onion::crypto::secp;
use std::net::TcpListener;
use std::sync::Arc;
use hyper::{Body, Client, Request, Response};
use tokio::sync::Mutex;
use grin_onion::create_onion;
use grin_onion::crypto::comsig::ComSignature;
use grin_onion::crypto::secp;
use crate::config::ServerConfig;
use crate::servers::swap::{SwapError, SwapServer};
use crate::servers::swap::mock::MockSwapServer;
use crate::servers::swap_rpc::{RPCSwapServer, SwapReq};
async fn body_to_string(req: Response<Body>) -> String {
let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
String::from_utf8(body_bytes.to_vec()).unwrap()
@@ -144,7 +146,6 @@ mod tests {
key: secp::random_secret(),
interval_s: 1,
addr: TcpListener::bind("127.0.0.1:0")?.local_addr()?,
socks_proxy_addr: TcpListener::bind("127.0.0.1:0")?.local_addr()?,
grin_node_url: "127.0.0.1:3413".parse()?,
grin_node_secret_path: None,
wallet_owner_url: "127.0.0.1:3420".parse()?,
@@ -186,8 +187,7 @@ mod tests {
/// Demonstrates a successful swap response
#[test]
fn swap_success() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut rt = tokio::runtime::Builder::new()
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
let commitment = secp::commit(1234, &secp::random_secret())?;
@@ -214,8 +214,7 @@ mod tests {
#[test]
fn swap_bad_request() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut rt = tokio::runtime::Builder::new()
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
let server: Arc<Mutex<dyn SwapServer>> = Arc::new(Mutex::new(MockSwapServer::new()));
@@ -235,8 +234,7 @@ mod tests {
/// Returns "Commitment not found" when there's no matching output in the UTXO set.
#[test]
fn swap_utxo_missing() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut rt = tokio::runtime::Builder::new()
.threaded_scheduler()
let rt = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
src/store.rs
+318 -309
View File
@@ -1,16 +1,16 @@
use grin_core::core::hash::Hash;
use grin_onion::crypto::secp::{self, Commitment, RangeProof, SecretKey};
use grin_onion::onion::Onion;
use grin_onion::util::{read_optional, write_optional};
use grin_core::core::{Input, Transaction};
use grin_core::core::hash::Hash;
use grin_core::ser::{
self, DeserializationMode, ProtocolVersion, Readable, Reader, Writeable, Writer,
self, DeserializationMode, ProtocolVersion, Readable, Reader, Writeable, Writer,
};
use grin_store::{self as store, Store};
use grin_util::ToHex;
use thiserror::Error;
use grin_onion::crypto::secp::{self, Commitment, RangeProof, SecretKey};
use grin_onion::onion::Onion;
use grin_onion::util::{read_optional, write_optional};
const DB_NAME: &str = "swap";
const STORE_SUBPATH: &str = "swaps";
@@ -23,380 +23,389 @@ const TX_PREFIX: u8 = b'T';
/// Swap statuses
#[derive(Clone, Debug, PartialEq)]
pub enum SwapStatus {
Unprocessed,
InProcess {
kernel_commit: Commitment,
},
Completed {
kernel_commit: Commitment,
block_hash: Hash,
},
Failed,
Unprocessed,
InProcess {
kernel_commit: Commitment,
},
Completed {
kernel_commit: Commitment,
block_hash: Hash,
},
Failed,
}
impl Writeable for SwapStatus {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
match self {
SwapStatus::Unprocessed => {
writer.write_u8(0)?;
}
SwapStatus::InProcess { kernel_commit } => {
writer.write_u8(1)?;
kernel_commit.write(writer)?;
}
SwapStatus::Completed {
kernel_commit,
block_hash,
} => {
writer.write_u8(2)?;
kernel_commit.write(writer)?;
block_hash.write(writer)?;
}
SwapStatus::Failed => {
writer.write_u8(3)?;
}
};
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
match self {
SwapStatus::Unprocessed => {
writer.write_u8(0)?;
}
SwapStatus::InProcess { kernel_commit } => {
writer.write_u8(1)?;
kernel_commit.write(writer)?;
}
SwapStatus::Completed {
kernel_commit,
block_hash,
} => {
writer.write_u8(2)?;
kernel_commit.write(writer)?;
block_hash.write(writer)?;
}
SwapStatus::Failed => {
writer.write_u8(3)?;
}
};
Ok(())
}
Ok(())
}
}
impl Readable for SwapStatus {
fn read<R: Reader>(reader: &mut R) -> Result<SwapStatus, ser::Error> {
let status = match reader.read_u8()? {
0 => SwapStatus::Unprocessed,
1 => {
let kernel_commit = Commitment::read(reader)?;
SwapStatus::InProcess { kernel_commit }
}
2 => {
let kernel_commit = Commitment::read(reader)?;
let block_hash = Hash::read(reader)?;
SwapStatus::Completed {
kernel_commit,
block_hash,
}
}
3 => SwapStatus::Failed,
_ => {
return Err(ser::Error::CorruptedData);
}
};
Ok(status)
}
fn read<R: Reader>(reader: &mut R) -> Result<SwapStatus, ser::Error> {
let status = match reader.read_u8()? {
0 => SwapStatus::Unprocessed,
1 => {
let kernel_commit = Commitment::read(reader)?;
SwapStatus::InProcess { kernel_commit }
}
2 => {
let kernel_commit = Commitment::read(reader)?;
let block_hash = Hash::read(reader)?;
SwapStatus::Completed {
kernel_commit,
block_hash,
}
}
3 => SwapStatus::Failed,
_ => {
return Err(ser::Error::CorruptedData);
}
};
Ok(status)
}
}
/// Data needed to swap a single output.
#[derive(Clone, Debug, PartialEq)]
pub struct SwapData {
/// The total excess for the output commitment
pub excess: SecretKey,
/// The derived output commitment after applying excess and fee
pub output_commit: Commitment,
/// The rangeproof, included only for the final hop (node N)
pub rangeproof: Option<RangeProof>,
/// Transaction input being spent
pub input: Input,
/// Transaction fee
pub fee: u64,
/// The remaining onion after peeling off our layer
pub onion: Onion,
/// The status of the swap
pub status: SwapStatus,
/// The total excess for the output commitment
pub excess: SecretKey,
/// The derived output commitment after applying excess and fee
pub output_commit: Commitment,
/// The rangeproof, included only for the final hop (node N)
pub rangeproof: Option<RangeProof>,
/// Transaction input being spent
pub input: Input,
/// Transaction fee
pub fee: u64,
/// The remaining onion after peeling off our layer
pub onion: Onion,
/// The status of the swap
pub status: SwapStatus,
}
impl Writeable for SwapData {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u8(CURRENT_SWAP_VERSION)?;
writer.write_fixed_bytes(&self.excess)?;
writer.write_fixed_bytes(&self.output_commit)?;
write_optional(writer, &self.rangeproof)?;
self.input.write(writer)?;
writer.write_u64(self.fee.into())?;
self.onion.write(writer)?;
self.status.write(writer)?;
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u8(CURRENT_SWAP_VERSION)?;
writer.write_fixed_bytes(&self.excess)?;
writer.write_fixed_bytes(&self.output_commit)?;
write_optional(writer, &self.rangeproof)?;
self.input.write(writer)?;
writer.write_u64(self.fee.into())?;
self.onion.write(writer)?;
self.status.write(writer)?;
Ok(())
}
Ok(())
}
}
impl Readable for SwapData {
fn read<R: Reader>(reader: &mut R) -> Result<SwapData, ser::Error> {
let version = reader.read_u8()?;
if version != CURRENT_SWAP_VERSION {
return Err(ser::Error::UnsupportedProtocolVersion);
}
fn read<R: Reader>(reader: &mut R) -> Result<SwapData, ser::Error> {
let version = reader.read_u8()?;
if version != CURRENT_SWAP_VERSION {
return Err(ser::Error::UnsupportedProtocolVersion);
}
let excess = secp::read_secret_key(reader)?;
let output_commit = Commitment::read(reader)?;
let rangeproof = read_optional(reader)?;
let input = Input::read(reader)?;
let fee = reader.read_u64()?;
let onion = Onion::read(reader)?;
let status = SwapStatus::read(reader)?;
Ok(SwapData {
excess,
output_commit,
rangeproof,
input,
fee,
onion,
status,
})
}
let excess = secp::read_secret_key(reader)?;
let output_commit = Commitment::read(reader)?;
let rangeproof = read_optional(reader)?;
let input = Input::read(reader)?;
let fee = reader.read_u64()?;
let onion = Onion::read(reader)?;
let status = SwapStatus::read(reader)?;
Ok(SwapData {
excess,
output_commit,
rangeproof,
input,
fee,
onion,
status,
})
}
}
/// A transaction created as part of a swap round.
#[derive(Clone, Debug, PartialEq)]
pub struct SwapTx {
pub tx: Transaction,
pub chain_tip: (u64, Hash),
// TODO: Include status
pub tx: Transaction,
pub chain_tip: (u64, Hash),
// TODO: Include status
}
impl Writeable for SwapTx {
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u8(CURRENT_TX_VERSION)?;
self.tx.write(writer)?;
writer.write_u64(self.chain_tip.0)?;
self.chain_tip.1.write(writer)?;
Ok(())
}
fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
writer.write_u8(CURRENT_TX_VERSION)?;
self.tx.write(writer)?;
writer.write_u64(self.chain_tip.0)?;
self.chain_tip.1.write(writer)?;
Ok(())
}
}
impl Readable for SwapTx {
fn read<R: Reader>(reader: &mut R) -> Result<SwapTx, ser::Error> {
let version = reader.read_u8()?;
if version != CURRENT_TX_VERSION {
return Err(ser::Error::UnsupportedProtocolVersion);
}
fn read<R: Reader>(reader: &mut R) -> Result<SwapTx, ser::Error> {
let version = reader.read_u8()?;
if version != CURRENT_TX_VERSION {
return Err(ser::Error::UnsupportedProtocolVersion);
}
let tx = Transaction::read(reader)?;
let height = reader.read_u64()?;
let block_hash = Hash::read(reader)?;
Ok(SwapTx {
tx,
chain_tip: (height, block_hash),
})
}
let tx = Transaction::read(reader)?;
let height = reader.read_u64()?;
let block_hash = Hash::read(reader)?;
Ok(SwapTx {
tx,
chain_tip: (height, block_hash),
})
}
}
/// Storage facility for swap data.
pub struct SwapStore {
db: Store,
db: Store,
}
/// Store error types
#[derive(Clone, Error, Debug, PartialEq)]
pub enum StoreError {
#[error("Swap entry already exists for '{0:?}'")]
AlreadyExists(Commitment),
#[error("Error occurred while attempting to open db: {0}")]
OpenError(store::lmdb::Error),
#[error("Serialization error occurred: {0}")]
SerializationError(ser::Error),
#[error("Error occurred while attempting to read from db: {0}")]
ReadError(store::lmdb::Error),
#[error("Error occurred while attempting to write to db: {0}")]
WriteError(store::lmdb::Error),
#[error("Swap entry already exists for '{0:?}'")]
AlreadyExists(Commitment),
#[error("Entry does not exist for '{0:?}'")]
NotFound(Commitment),
#[error("Error occurred while attempting to open db: {0}")]
OpenError(store::lmdb::Error),
#[error("Serialization error occurred: {0}")]
SerializationError(ser::Error),
#[error("Error occurred while attempting to read from db: {0}")]
ReadError(store::lmdb::Error),
#[error("Error occurred while attempting to write to db: {0}")]
WriteError(store::lmdb::Error),
}
impl From<ser::Error> for StoreError {
fn from(e: ser::Error) -> StoreError {
StoreError::SerializationError(e)
}
fn from(e: ser::Error) -> StoreError {
StoreError::SerializationError(e)
}
}
impl SwapStore {
/// Create new chain store
pub fn new(db_root: &str) -> Result<SwapStore, StoreError> {
let db = Store::new(db_root, Some(DB_NAME), Some(STORE_SUBPATH), None)
.map_err(StoreError::OpenError)?;
Ok(SwapStore { db })
}
/// Create new chain store
pub fn new(db_root: &str) -> Result<SwapStore, StoreError> {
let db = Store::new(db_root, Some(DB_NAME), Some(STORE_SUBPATH), None)
.map_err(StoreError::OpenError)?;
Ok(SwapStore { db })
}
/// Writes a single key-value pair to the database
fn write<K: AsRef<[u8]>>(
&self,
prefix: u8,
k: K,
value: &Vec<u8>,
overwrite: bool,
) -> Result<bool, store::lmdb::Error> {
let batch = self.db.batch()?;
let key = store::to_key(prefix, k);
if !overwrite && batch.exists(&key[..])? {
Ok(false)
} else {
batch.put(&key[..], &value[..])?;
batch.commit()?;
Ok(true)
}
}
/// Writes a single key-value pair to the database
fn write<K: AsRef<[u8]>>(
&self,
prefix: u8,
k: K,
value: &Vec<u8>,
overwrite: bool,
) -> Result<bool, store::lmdb::Error> {
let batch = self.db.batch()?;
let key = store::to_key(prefix, k);
if !overwrite && batch.exists(&key[..])? {
Ok(false)
} else {
batch.put(&key[..], &value[..])?;
batch.commit()?;
Ok(true)
}
}
/// Reads a single value by key
fn read<K: AsRef<[u8]> + Copy, V: Readable>(&self, prefix: u8, k: K) -> Result<V, StoreError> {
store::option_to_not_found(self.db.get_ser(&store::to_key(prefix, k), None), || {
format!("{}:{}", prefix, k.to_hex())
})
.map_err(StoreError::ReadError)
}
/// Reads a single value by key
fn read<K: AsRef<[u8]> + Copy, V: Readable>(&self, prefix: u8, k: K) -> Result<V, StoreError> {
store::option_to_not_found(self.db.get_ser(&store::to_key(prefix, k), None), || {
format!("{}:{}", prefix, k.to_hex())
})
.map_err(StoreError::ReadError)
}
/// Saves a swap to the database
pub fn save_swap(&self, s: &SwapData, overwrite: bool) -> Result<(), StoreError> {
let data = ser::ser_vec(&s, ProtocolVersion::local())?;
let saved = self
.write(SWAP_PREFIX, &s.input.commit, &data, overwrite)
.map_err(StoreError::WriteError)?;
if !saved {
Err(StoreError::AlreadyExists(s.input.commit.clone()))
} else {
Ok(())
}
}
/// Saves a swap to the database
pub fn save_swap(&self, s: &SwapData, overwrite: bool) -> Result<(), StoreError> {
let data = ser::ser_vec(&s, ProtocolVersion::local())?;
let saved = self
.write(SWAP_PREFIX, &s.input.commit, &data, overwrite)
.map_err(StoreError::WriteError)?;
if !saved {
Err(StoreError::AlreadyExists(s.input.commit.clone()))
} else {
Ok(())
}
}
/// Iterator over all swaps.
pub fn swaps_iter(&self) -> Result<impl Iterator<Item=SwapData>, StoreError> {
let key = store::to_key(SWAP_PREFIX, "");
let protocol_version = self.db.protocol_version();
self.db
.iter(&key[..], move |_, mut v| {
ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
.map_err(From::from)
})
.map_err(|e| StoreError::ReadError(e))
}
/// Iterator over all swaps.
pub fn swaps_iter(&self) -> Result<impl Iterator<Item = SwapData>, StoreError> {
let key = store::to_key(SWAP_PREFIX, "");
let protocol_version = self.db.protocol_version();
self.db
.iter(&key[..], move |_, mut v| {
ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
.map_err(From::from)
})
.map_err(|e| StoreError::ReadError(e))
}
/// Checks if a matching swap exists in the database
#[allow(dead_code)]
pub fn swap_exists(&self, input_commit: &Commitment) -> Result<bool, StoreError> {
let key = store::to_key(SWAP_PREFIX, input_commit);
self.db
.batch()
.map_err(StoreError::ReadError)?
.exists(&key[..])
.map_err(StoreError::ReadError)
}
/// Checks if a matching swap exists in the database
#[allow(dead_code)]
pub fn swap_exists(&self, input_commit: &Commitment) -> Result<bool, StoreError> {
let key = store::to_key(SWAP_PREFIX, input_commit);
self.db
.batch()
.map_err(StoreError::ReadError)?
.exists(&key[..])
.map_err(StoreError::ReadError)
}
/// Reads a swap from the database
pub fn get_swap(&self, input_commit: &Commitment) -> Result<SwapData, StoreError> {
self.read(SWAP_PREFIX, input_commit)
}
/// Reads a swap from the database
pub fn get_swap(&self, input_commit: &Commitment) -> Result<SwapData, StoreError> {
self.read(SWAP_PREFIX, input_commit)
}
/// Saves a swap transaction to the database
pub fn save_swap_tx(&self, s: &SwapTx) -> Result<(), StoreError> {
let data = ser::ser_vec(&s, ProtocolVersion::local())?;
self
.write(TX_PREFIX, &s.tx.kernels().first().unwrap().excess, &data, true)
.map_err(StoreError::WriteError)?;
/// Saves a swap transaction to the database
pub fn save_swap_tx(&self, s: &SwapTx) -> Result<(), StoreError> {
let data = ser::ser_vec(&s, ProtocolVersion::local())?;
self.write(
TX_PREFIX,
&s.tx.kernels().first().unwrap().excess,
&data,
true,
)
.map_err(StoreError::WriteError)?;
Ok(())
}
Ok(())
}
/// Reads a swap tx from the database
pub fn get_swap_tx(&self, kernel_excess: &Commitment) -> Result<SwapTx, StoreError> {
self.read(TX_PREFIX, kernel_excess)
}
/// Reads a swap tx from the database
pub fn get_swap_tx(&self, kernel_excess: &Commitment) -> Result<SwapTx, StoreError> {
self.read(TX_PREFIX, kernel_excess)
}
}
#[cfg(test)]
mod tests {
use crate::store::{StoreError, SwapData, SwapStatus, SwapStore};
use grin_core::core::{Input, OutputFeatures};
use grin_core::global::{self, ChainTypes};
use grin_onion::crypto::secp;
use grin_onion::test_util as onion_test_util;
use rand::RngCore;
use std::cmp::Ordering;
use std::cmp::Ordering;
fn new_store(test_name: &str) -> SwapStore {
global::set_local_chain_type(ChainTypes::AutomatedTesting);
let db_root = format!("./target/tmp/.{}", test_name);
let _ = std::fs::remove_dir_all(db_root.as_str());
SwapStore::new(db_root.as_str()).unwrap()
}
use grin_core::core::{Input, OutputFeatures};
use grin_core::global::{self, ChainTypes};
use rand::RngCore;
fn rand_swap_with_status(status: SwapStatus) -> SwapData {
SwapData {
excess: secp::random_secret(),
output_commit: onion_test_util::rand_commit(),
rangeproof: Some(onion_test_util::rand_proof()),
input: Input::new(OutputFeatures::Plain, onion_test_util::rand_commit()),
fee: rand::thread_rng().next_u64(),
onion: onion_test_util::rand_onion(),
status,
}
}
use grin_onion::crypto::secp;
use grin_onion::test_util as onion_test_util;
fn rand_swap() -> SwapData {
let s = rand::thread_rng().next_u64() % 3;
let status = if s == 0 {
SwapStatus::Unprocessed
} else if s == 1 {
SwapStatus::InProcess {
kernel_commit: onion_test_util::rand_commit(),
}
} else {
SwapStatus::Completed {
kernel_commit: onion_test_util::rand_commit(),
block_hash: onion_test_util::rand_hash(),
}
};
rand_swap_with_status(status)
}
use crate::store::{StoreError, SwapData, SwapStatus, SwapStore};
#[test]
fn swap_iter() -> Result<(), Box<dyn std::error::Error>> {
let store = new_store("swap_iter");
let mut swaps: Vec<SwapData> = Vec::new();
for _ in 0..5 {
let swap = rand_swap();
store.save_swap(&swap, false)?;
swaps.push(swap);
}
fn new_store(test_name: &str) -> SwapStore {
global::set_local_chain_type(ChainTypes::AutomatedTesting);
let db_root = format!("./target/tmp/.{}", test_name);
let _ = std::fs::remove_dir_all(db_root.as_str());
SwapStore::new(db_root.as_str()).unwrap()
}
swaps.sort_by(|a, b| {
if a.input.commit < b.input.commit {
Ordering::Less
} else if a.input.commit == b.input.commit {
Ordering::Equal
} else {
Ordering::Greater
}
});
fn rand_swap_with_status(status: SwapStatus) -> SwapData {
SwapData {
excess: secp::random_secret(),
output_commit: onion_test_util::rand_commit(),
rangeproof: Some(onion_test_util::rand_proof()),
input: Input::new(OutputFeatures::Plain, onion_test_util::rand_commit()),
fee: rand::thread_rng().next_u64(),
onion: onion_test_util::rand_onion(),
status,
}
}
let mut i: usize = 0;
for swap in store.swaps_iter()? {
assert_eq!(swap, *swaps.get(i).unwrap());
i += 1;
}
fn rand_swap() -> SwapData {
let s = rand::thread_rng().next_u64() % 3;
let status = if s == 0 {
SwapStatus::Unprocessed
} else if s == 1 {
SwapStatus::InProcess {
kernel_commit: onion_test_util::rand_commit(),
}
} else {
SwapStatus::Completed {
kernel_commit: onion_test_util::rand_commit(),
block_hash: onion_test_util::rand_hash(),
}
};
rand_swap_with_status(status)
}
Ok(())
}
#[test]
fn swap_iter() -> Result<(), Box<dyn std::error::Error>> {
let store = new_store("swap_iter");
let mut swaps: Vec<SwapData> = Vec::new();
for _ in 0..5 {
let swap = rand_swap();
store.save_swap(&swap, false)?;
swaps.push(swap);
}
#[test]
fn save_swap() -> Result<(), Box<dyn std::error::Error>> {
let store = new_store("save_swap");
swaps.sort_by(|a, b| {
if a.input.commit < b.input.commit {
Ordering::Less
} else if a.input.commit == b.input.commit {
Ordering::Equal
} else {
Ordering::Greater
}
});
let mut swap = rand_swap_with_status(SwapStatus::Unprocessed);
assert!(!store.swap_exists(&swap.input.commit)?);
let mut i: usize = 0;
for swap in store.swaps_iter()? {
assert_eq!(swap, *swaps.get(i).unwrap());
i += 1;
}
store.save_swap(&swap, false)?;
assert_eq!(swap, store.get_swap(&swap.input.commit)?);
assert!(store.swap_exists(&swap.input.commit)?);
Ok(())
}
swap.status = SwapStatus::InProcess {
kernel_commit: onion_test_util::rand_commit(),
};
let result = store.save_swap(&swap, false);
assert_eq!(
Err(StoreError::AlreadyExists(swap.input.commit.clone())),
result
);
#[test]
fn save_swap() -> Result<(), Box<dyn std::error::Error>> {
let store = new_store("save_swap");
store.save_swap(&swap, true)?;
assert_eq!(swap, store.get_swap(&swap.input.commit)?);
let mut swap = rand_swap_with_status(SwapStatus::Unprocessed);
assert!(!store.swap_exists(&swap.input.commit)?);
Ok(())
}
store.save_swap(&swap, false)?;
assert_eq!(swap, store.get_swap(&swap.input.commit)?);
assert!(store.swap_exists(&swap.input.commit)?);
swap.status = SwapStatus::InProcess {
kernel_commit: onion_test_util::rand_commit(),
};
let result = store.save_swap(&swap, false);
assert_eq!(
Err(StoreError::AlreadyExists(swap.input.commit.clone())),
result
);
store.save_swap(&swap, true)?;
assert_eq!(swap, store.get_swap(&swap.input.commit)?);
Ok(())
}
}
src/tor.rs
+172 -79
View File
@@ -1,9 +1,32 @@
use crate::config::ServerConfig;
use std::sync::Arc;
use grin_wallet_impls::tor::config as tor_config;
use grin_wallet_impls::tor::process::TorProcess;
use std::collections::HashMap;
use arti_client::config::TorClientConfigBuilder;
use arti_client::{TorClient, TorClientConfig};
use arti_hyper::ArtiHttpConnector;
use curve25519_dalek::digest::Digest;
use ed25519_dalek::hazmat::ExpandedSecretKey;
use futures::task::SpawnExt;
use sha2::Sha512;
use thiserror::Error;
use tls_api::{TlsConnector as TlsConnectorTrait, TlsConnectorBuilder};
use tls_api_native_tls::TlsConnector;
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, RunningOnionService,
};
use tor_keymgr::key_specifier_derive::internal;
use tor_keymgr::{ArtiNativeKeystore, KeyMgrBuilder, KeystoreSelector};
use tor_llcrypto::pk::ed25519::ExpandedKeypair;
use tor_rtcompat::Runtime;
use secp256k1zkp::SecretKey;
use crate::config::ServerConfig;
/// Tor error types
#[derive(Error, Debug)]
@@ -14,93 +37,163 @@ pub enum TorError {
ProcessError(grin_wallet_impls::tor::process::Error),
}
pub fn init_tor_listener(
data_dir: &str,
server_config: &ServerConfig,
) -> Result<TorProcess, TorError> {
warn!("Initializing tor listener");
pub struct TorService<R: Runtime> {
tor_client: Option<TorClient<R>>,
hidden_services: Vec<Arc<RunningOnionService>>,
}
let tor_dir = format!("{}/tor/listener", &data_dir);
trace!(
"Dir: {}, Proxy: {}",
&tor_dir,
server_config.socks_proxy_addr.to_string()
);
impl<R: Runtime> TorService<R> {
/// Builds a hyper::Client with an ArtiHttpConnector over the TorClient.
/// The returned Client makes HTTP requests through the TorClient directly, eliminating the need for a socks proxy.
pub fn new_hyper_client(
&self,
) -> hyper::Client<ArtiHttpConnector<R, TlsConnector>, hyper::Body> {
let tls_connector = TlsConnector::builder().unwrap().build().unwrap();
let tor_connector = ArtiHttpConnector::new(self.tor_client.clone().unwrap(), tls_connector);
// create data directory if it doesn't exist
std::fs::create_dir_all(&format!("{}/data", tor_dir)).unwrap();
let service_dir = tor_config::output_onion_service_config(tor_dir.as_str(), &server_config.key)
.map_err(|e| TorError::ConfigError(e))?;
let service_dirs = vec![service_dir.to_string()];
tor_config::output_torrc(
tor_dir.as_str(),
server_config.addr.to_string().as_str(),
&server_config.socks_proxy_addr.to_string(),
&service_dirs,
HashMap::new(),
HashMap::new(),
)
.map_err(|e| TorError::ConfigError(e))?;
// Start TOR process
let mut process = TorProcess::new();
process
.torrc_path("./torrc")
.working_dir(tor_dir.as_str())
.timeout(30)
.completion_percent(100);
let mut attempts = 0;
let max_attempts = 3;
let mut result;
loop {
attempts += 1;
info!("Launching TorProcess... Attempt {}", attempts);
result = process.launch();
if result.is_ok() || attempts >= max_attempts {
break;
}
hyper::Client::builder().build::<_, hyper::Body>(tor_connector)
}
result.map_err(TorError::ProcessError)?;
pub fn stop(&mut self) {
self.tor_client = None;
self.hidden_services.clear();
}
}
pub async fn async_init_tor<R>(
runtime: R,
data_dir: &str,
server_config: &ServerConfig,
) -> Result<TorService<R>, TorError>
where
R: Runtime,
{
warn!("Initializing TOR");
let state_dir = format!("{}/tor/state", &data_dir);
let cache_dir = format!("{}/tor/cache", &data_dir);
let hs_nickname = HsNickname::new("listener".to_string()).unwrap();
let mut client_config_builder =
TorClientConfigBuilder::from_directories(state_dir.clone(), cache_dir.clone());
client_config_builder
.address_filter()
.allow_onion_addrs(true);
let client_config = client_config_builder.build().unwrap();
add_key_to_store(&client_config, &state_dir, &server_config.key, &hs_nickname)?;
let tor_client = TorClient::with_runtime(runtime)
.config(client_config)
.create_bootstrapped()
.await
.unwrap();
let service =
async_launch_hidden_service(hs_nickname.clone(), &tor_client, &server_config).await?;
let tor_instance = TorService {
tor_client: Some(tor_client),
hidden_services: vec![service],
};
Ok(tor_instance)
}
async fn async_launch_hidden_service<R>(
hs_nickname: HsNickname,
tor_client: &TorClient<R>,
server_config: &ServerConfig,
) -> Result<Arc<RunningOnionService>, TorError>
where
R: Runtime,
{
let svc_cfg = OnionServiceConfigBuilder::default()
.nickname(hs_nickname.clone())
.build()
.unwrap();
let (service, request_stream) = tor_client.launch_onion_service(svc_cfg).unwrap();
let proxy_rule = ProxyRule::new(
ProxyPattern::one_port(80).unwrap(),
ProxyAction::Forward(Encapsulation::Simple, TargetAddr::Inet(server_config.addr)),
);
let mut proxy_cfg_builder = ProxyConfigBuilder::default();
proxy_cfg_builder.set_proxy_ports(vec![proxy_rule]);
let proxy = OnionServiceReverseProxy::new(proxy_cfg_builder.build().unwrap());
{
let proxy = proxy.clone();
let runtime_clone = tor_client.runtime().clone();
tor_client
.runtime()
.spawn(async move {
match proxy
.handle_requests(runtime_clone, hs_nickname.clone(), request_stream)
.await
{
Ok(()) => {
debug!("Onion service {} exited cleanly.", hs_nickname);
}
Err(e) => {
warn!("Onion service {} exited with an error: {}", hs_nickname, e);
}
}
})
.unwrap();
}
warn!(
"Server listening at http://{}.onion",
server_config.onion_address().to_ov3_str()
);
Ok(process)
Ok(service)
}
pub fn init_tor_sender(
data_dir: &str,
server_config: &ServerConfig,
) -> Result<TorProcess, TorError> {
warn!(
"Starting TOR Process for send at {:?}",
server_config.socks_proxy_addr
// TODO: Add proper error handling
fn add_key_to_store(
tor_config: &TorClientConfig,
state_dir: &String,
secret_key: &SecretKey,
hs_nickname: &HsNickname,
) -> Result<(), TorError> {
let key_store_dir = format!("{}/keystore", &state_dir);
let arti_store =
ArtiNativeKeystore::from_path_and_mistrust(&key_store_dir, &tor_config.fs_mistrust())
.unwrap();
info!("Using keystore from {key_store_dir:?}");
let key_manager = KeyMgrBuilder::default()
.default_store(Box::new(arti_store))
.build()
.map_err(|_| internal!("failed to build keymgr"))
.unwrap();
let expanded_sk = ExpandedSecretKey::from_bytes(
Sha512::default()
.chain_update(secret_key)
.finalize()
.as_ref(),
);
let tor_dir = format!("{}/tor/sender", data_dir);
tor_config::output_tor_sender_config(
tor_dir.as_str(),
&server_config.socks_proxy_addr.to_string(),
HashMap::new(),
HashMap::new(),
)
.map_err(|e| TorError::ConfigError(e))?;
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);
let expanded_kp = ExpandedKeypair::from_secret_key_bytes(sk_bytes).unwrap();
// Start TOR process
let mut tor_process = TorProcess::new();
tor_process
.torrc_path("./torrc")
.working_dir(tor_dir.as_str())
.timeout(40)
.completion_percent(100)
.launch()
.map_err(TorError::ProcessError)?;
Ok(tor_process)
key_manager
.insert(
HsIdKey::from(expanded_kp.public().clone()),
&HsIdPublicKeySpecifier::new(hs_nickname.clone()),
KeystoreSelector::Default,
)
.unwrap();
key_manager
.insert(
HsIdKeypair::from(expanded_kp),
&HsIdKeypairSpecifier::new(hs_nickname.clone()),
KeystoreSelector::Default,
)
.unwrap();
Ok(())
}
src/wallet.rs
+38 -95
View File
@@ -1,18 +1,20 @@
use std::net::SocketAddr;
use async_trait::async_trait;
use grin_api::client;
use grin_api::json_rpc::{build_request, Request, Response, RpcError};
use grin_core::core::Output;
use grin_core::libtx::secp_ser;
use grin_keychain::BlindingFactor;
use grin_onion::crypto::secp;
use grin_util::{ToHex, ZeroingString};
use grin_wallet_api::{EncryptedRequest, EncryptedResponse, JsonId, Token};
use secp256k1zkp::{PublicKey, Secp256k1, SecretKey};
use grin_wallet_api::Token;
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::net::SocketAddr;
use thiserror::Error;
use grin_onion::crypto::secp;
use secp256k1zkp::{PublicKey, Secp256k1, SecretKey};
use crate::http;
#[async_trait]
pub trait Wallet: Send + Sync {
/// Builds an output for the wallet with the provided amount.
@@ -25,18 +27,8 @@ pub trait Wallet: Send + Sync {
/// Error types for interacting with wallets
#[derive(Error, Debug)]
pub enum WalletError {
#[error("Error encrypting request: {0:?}")]
EncryptRequestError(grin_wallet_libwallet::Error),
#[error("Error decrypting response: {0:?}")]
DecryptResponseError(grin_wallet_libwallet::Error),
#[error("Error decoding JSON response: {0:?}")]
DecodeResponseError(serde_json::Error),
#[error("JSON-RPC API communication error: {0:?}")]
ApiCommError(grin_api::Error),
#[error("Error decoding JSON-RPC response: {0:?}")]
ResponseParseError(grin_api::json_rpc::Error),
#[error("Unsucessful response returned: {0:?}")]
ResponseRpcError(Option<RpcError>),
#[error("Error communication with wallet: {0:?}")]
WalletCommError(http::HttpError),
}
/// HTTP (JSONRPC) implementation of the 'Wallet' trait.
@@ -73,14 +65,16 @@ impl HttpWallet {
"name": null,
"password": wallet_pass.to_string()
});
let token: Token = HttpWallet::async_send_enc_request(
&wallet_owner_url,
let url = format!("http://{}{}", wallet_owner_url, ENDPOINT);
let token: Token = http::async_send_enc_request(
&url,
&wallet_owner_secret,
"open_wallet",
&open_wallet_params,
&shared_key,
)
.await?;
.await
.map_err(WalletError::WalletCommError)?;
info!("Connected to wallet");
Ok(HttpWallet {
@@ -98,17 +92,20 @@ impl HttpWallet {
let secp = Secp256k1::new();
let ephemeral_sk = secp::random_secret();
let ephemeral_pk = PublicKey::from_secret_key(&secp, &ephemeral_sk).unwrap();
let ephemeral_pk_bytes = ephemeral_pk.serialize_vec(&secp, true);
let init_params = json!({
"ecdh_pubkey": ephemeral_pk.serialize_vec(&secp, true).to_hex()
"ecdh_pubkey": ephemeral_pk_bytes.to_hex()
});
let response_pk: ECDHPubkey = HttpWallet::async_send_json_request(
&wallet_owner_url,
let url = format!("http://{}{}", wallet_owner_url, ENDPOINT);
let response_pk: ECDHPubkey = http::async_send_json_request(
&url,
&wallet_owner_secret,
"init_secure_api",
&init_params,
)
.await?;
.await
.map_err(WalletError::WalletCommError)?;
let shared_key = {
let mut shared_pubkey = response_pk.ecdh_pubkey.clone();
@@ -126,75 +123,16 @@ impl HttpWallet {
method: &str,
params: &serde_json::Value,
) -> Result<D, WalletError> {
HttpWallet::async_send_enc_request(
&self.wallet_owner_url,
let url = format!("http://{}{}", self.wallet_owner_url, ENDPOINT);
http::async_send_enc_request(
&url,
&self.wallet_owner_secret,
method,
params,
&self.shared_key,
)
.await
}
async fn async_send_enc_request<D: serde::de::DeserializeOwned>(
wallet_owner_url: &SocketAddr,
wallet_owner_secret: &Option<String>,
method: &str,
params: &serde_json::Value,
shared_key: &SecretKey,
) -> Result<D, WalletError> {
let url = format!("http://{}{}", wallet_owner_url, ENDPOINT);
let req = json!({
"method": method,
"params": params,
"id": JsonId::IntId(1),
"jsonrpc": "2.0",
});
let enc_req = EncryptedRequest::from_json(&JsonId::IntId(1), &req, &shared_key)
.map_err(WalletError::EncryptRequestError)?;
let res = client::post_async::<EncryptedRequest, EncryptedResponse>(
url.as_str(),
&enc_req,
wallet_owner_secret.clone(),
)
.await
.map_err(WalletError::ApiCommError)?;
let decrypted = res
.decrypt(&shared_key)
.map_err(WalletError::DecryptResponseError)?;
let response: Response =
serde_json::from_value(decrypted).map_err(WalletError::DecodeResponseError)?;
let result = response
.result
.ok_or(WalletError::ResponseRpcError(response.error.clone()))?;
let ok = result
.get("Ok")
.ok_or(WalletError::ResponseRpcError(response.error.clone()))?;
let parsed =
serde_json::from_value(ok.clone()).map_err(WalletError::DecodeResponseError)?;
Ok(parsed)
}
async fn async_send_json_request<D: serde::de::DeserializeOwned>(
wallet_owner_url: &SocketAddr,
wallet_owner_secret: &Option<String>,
method: &str,
params: &serde_json::Value,
) -> Result<D, WalletError> {
let url = format!("http://{}{}", wallet_owner_url, ENDPOINT);
let req = build_request(method, params);
let res = client::post_async::<Request, Response>(
url.as_str(),
&req,
wallet_owner_secret.clone(),
)
.await
.map_err(WalletError::ApiCommError)?;
let parsed = res
.clone()
.into_result()
.map_err(WalletError::ResponseParseError)?;
Ok(parsed)
.map_err(WalletError::WalletCommError)
}
pub fn get_token(&self) -> Token {
@@ -219,35 +157,40 @@ impl Wallet for HttpWallet {
&self,
amount: u64,
) -> Result<(BlindingFactor, Output), WalletError> {
let req_json = json!({
let params = json!({
"token": self.token,
"features": "Plain",
"amount": amount
});
let output: OutputWithBlind = HttpWallet::async_send_enc_request(
&self.wallet_owner_url,
let url = format!("http://{}{}", self.wallet_owner_url, ENDPOINT);
let output: OutputWithBlind = http::async_send_enc_request(
&url,
&self.wallet_owner_secret,
"build_output",
&req_json,
&params,
&self.shared_key,
)
.await?;
.await
.map_err(WalletError::WalletCommError)?;
Ok((output.blind, output.output))
}
}
#[cfg(test)]
pub mod mock {
use super::{Wallet, WalletError};
use std::borrow::BorrowMut;
use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use grin_core::core::{Output, OutputFeatures};
use grin_keychain::BlindingFactor;
use grin_onion::crypto::secp;
use secp256k1zkp::pedersen::Commitment;
use secp256k1zkp::Secp256k1;
use std::sync::{Arc, Mutex};
use super::{Wallet, WalletError};
/// Mock implementation of the 'Wallet' trait for unit-tests.
#[derive(Clone)]