Optional Tor Send/Listen Functionality (#226)

* udpate for beta release

* initial tor explorations

* rustfmt

* basic tor tx send working

* rustfmt

* add tor proxy info to config file

* rustfmt

* add utilities to output tor hidden service configuration files

* output tor config as part of listener startup

* rustfmt

* fully automate config and startup of tor process

* rustfmt

* remove unnecessary process kill commands from listener

* rustfmt

* assume defaults for tor sending config if section doesn't exist in grin-wallet.toml

* rustfmt

* ignore tor dev test

* update default paths output by config, compilation + confirmed working on windows

* rustfmt

* fix on osx/unix

* add timeout to tor connector, remove unwrap in client

* allow specifiying tor address without 'http://[].onion' on the command line

* fix api test

* rustfmt

* update address derivation path as per spec

* rustfmt

* move tor init to separate function

* rustfmt

* re-ignore tor dev test

* listen on tor by default if tor available

* rustfmt

* test fix

* remove explicit send via tor flag, and assume tor if address fits

* rustfmt

impls/src/client_utils/client.rs

@@ -0,0 +1,396 @@
+// Copyright 2018 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.
+
+//! High level JSON/HTTP client API
+
+use crate::client_utils::Socksv5Connector;
+use crate::util::to_base64;
+use failure::{Backtrace, Context, Fail, ResultExt};
+use futures::future::result;
+use futures::future::{err, ok, Either};
+use futures::stream::Stream;
+use http::uri::{InvalidUri, Uri};
+use hyper::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, USER_AGENT};
+use hyper::rt::Future;
+use hyper::{self, Body, Request};
+use hyper_rustls;
+use hyper_timeout::TimeoutConnector;
+use serde::{Deserialize, Serialize};
+use serde_json;
+use std::fmt::{self, Display};
+use std::net::SocketAddr;
+use std::time::Duration;
+use tokio::runtime::Runtime;
+
+/// Errors that can be returned by an ApiEndpoint implementation.
+#[derive(Debug)]
+pub struct Error {
+	inner: Context<ErrorKind>,
+}
+
+#[derive(Clone, Eq, PartialEq, Debug, Fail)]
+pub enum ErrorKind {
+	#[fail(display = "Internal error: {}", _0)]
+	Internal(String),
+	#[fail(display = "Bad arguments: {}", _0)]
+	Argument(String),
+	#[fail(display = "Not found.")]
+	_NotFound,
+	#[fail(display = "Request error: {}", _0)]
+	RequestError(String),
+	#[fail(display = "ResponseError error: {}", _0)]
+	ResponseError(String),
+}
+
+impl Fail for Error {
+	fn cause(&self) -> Option<&dyn Fail> {
+		self.inner.cause()
+	}
+
+	fn backtrace(&self) -> Option<&Backtrace> {
+		self.inner.backtrace()
+	}
+}
+
+impl Display for Error {
+	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		Display::fmt(&self.inner, f)
+	}
+}
+
+impl Error {
+	pub fn _kind(&self) -> &ErrorKind {
+		self.inner.get_context()
+	}
+}
+
+impl From<ErrorKind> for Error {
+	fn from(kind: ErrorKind) -> Error {
+		Error {
+			inner: Context::new(kind),
+		}
+	}
+}
+
+impl From<Context<ErrorKind>> for Error {
+	fn from(inner: Context<ErrorKind>) -> Error {
+		Error { inner: inner }
+	}
+}
+
+pub type ClientResponseFuture<T> = Box<dyn Future<Item = T, Error = Error> + Send>;
+
+pub struct Client {
+	/// Whether to use socks proxy
+	pub use_socks: bool,
+	/// Proxy url/port
+	pub socks_proxy_addr: Option<SocketAddr>,
+}
+
+impl Client {
+	/// New client
+	pub fn new() -> Self {
+		Client {
+			use_socks: false,
+			socks_proxy_addr: None,
+		}
+	}
+
+	/// Helper function to easily issue a HTTP GET request against a given URL that
+	/// returns a JSON object. Handles request building, JSON deserialization and
+	/// response code checking.
+	pub fn get<'a, T>(&self, url: &'a str, api_secret: Option<String>) -> Result<T, Error>
+	where
+		for<'de> T: Deserialize<'de>,
+	{
+		self.handle_request(self.build_request(url, "GET", api_secret, None)?)
+	}
+
+	/// Helper function to easily issue an async HTTP GET request against a given
+	/// URL that returns a future. Handles request building, JSON deserialization
+	/// and response code checking.
+	pub fn get_async<'a, T>(
+		&self,
+		url: &'a str,
+		api_secret: Option<String>,
+	) -> ClientResponseFuture<T>
+	where
+		for<'de> T: Deserialize<'de> + Send + 'static,
+	{
+		match self.build_request(url, "GET", api_secret, None) {
+			Ok(req) => Box::new(self.handle_request_async(req)),
+			Err(e) => Box::new(err(e)),
+		}
+	}
+
+	/// Helper function to easily issue a HTTP GET request
+	/// on a given URL that returns nothing. Handles request
+	/// building and response code checking.
+	pub fn _get_no_ret(&self, url: &str, api_secret: Option<String>) -> Result<(), Error> {
+		let req = self.build_request(url, "GET", api_secret, None)?;
+		self.send_request(req)?;
+		Ok(())
+	}
+
+	/// Helper function to easily issue a HTTP POST request with the provided JSON
+	/// object as body on a given URL that returns a JSON object. Handles request
+	/// building, JSON serialization and deserialization, and response code
+	/// checking.
+	pub fn _post<IN, OUT>(
+		&self,
+		url: &str,
+		api_secret: Option<String>,
+		input: &IN,
+	) -> Result<OUT, Error>
+	where
+		IN: Serialize,
+		for<'de> OUT: Deserialize<'de>,
+	{
+		let req = self.create_post_request(url, api_secret, input)?;
+		self.handle_request(req)
+	}
+
+	/// Helper function to easily issue an async HTTP POST request with the
+	/// provided JSON object as body on a given URL that returns a future. Handles
+	/// request building, JSON serialization and deserialization, and response code
+	/// checking.
+	pub fn _post_async<IN, OUT>(
+		&self,
+		url: &str,
+		input: &IN,
+		api_secret: Option<String>,
+	) -> ClientResponseFuture<OUT>
+	where
+		IN: Serialize,
+		OUT: Send + 'static,
+		for<'de> OUT: Deserialize<'de>,
+	{
+		match self.create_post_request(url, api_secret, input) {
+			Ok(req) => Box::new(self.handle_request_async(req)),
+			Err(e) => Box::new(err(e)),
+		}
+	}
+
+	/// Helper function to easily issue a HTTP POST request with the provided JSON
+	/// object as body on a given URL that returns nothing. Handles request
+	/// building, JSON serialization, and response code
+	/// checking.
+	pub fn post_no_ret<IN>(
+		&self,
+		url: &str,
+		api_secret: Option<String>,
+		input: &IN,
+	) -> Result<(), Error>
+	where
+		IN: Serialize,
+	{
+		let req = self.create_post_request(url, api_secret, input)?;
+		self.send_request(req)?;
+		Ok(())
+	}
+
+	/// Helper function to easily issue an async HTTP POST request with the
+	/// provided JSON object as body on a given URL that returns a future. Handles
+	/// request building, JSON serialization and deserialization, and response code
+	/// checking.
+	pub fn _post_no_ret_async<IN>(
+		&self,
+		url: &str,
+		api_secret: Option<String>,
+		input: &IN,
+	) -> ClientResponseFuture<()>
+	where
+		IN: Serialize,
+	{
+		match self.create_post_request(url, api_secret, input) {
+			Ok(req) => Box::new(self.send_request_async(req).and_then(|_| ok(()))),
+			Err(e) => Box::new(err(e)),
+		}
+	}
+
+	fn build_request(
+		&self,
+		url: &str,
+		method: &str,
+		api_secret: Option<String>,
+		body: Option<String>,
+	) -> Result<Request<Body>, Error> {
+		let uri = url.parse::<Uri>().map_err::<Error, _>(|e: InvalidUri| {
+			e.context(ErrorKind::Argument(format!("Invalid url {}", url)))
+				.into()
+		})?;
+		let mut builder = Request::builder();
+		if let Some(api_secret) = api_secret {
+			let basic_auth = format!("Basic {}", to_base64(&format!("grin:{}", api_secret)));
+			builder.header(AUTHORIZATION, basic_auth);
+		}
+
+		builder
+			.method(method)
+			.uri(uri)
+			.header(USER_AGENT, "grin-client")
+			.header(ACCEPT, "application/json")
+			.header(CONTENT_TYPE, "application/json")
+			.body(match body {
+				None => Body::empty(),
+				Some(json) => json.into(),
+			})
+			.map_err(|e| {
+				ErrorKind::RequestError(format!("Bad request {} {}: {}", method, url, e)).into()
+			})
+	}
+
+	pub fn create_post_request<IN>(
+		&self,
+		url: &str,
+		api_secret: Option<String>,
+		input: &IN,
+	) -> Result<Request<Body>, Error>
+	where
+		IN: Serialize,
+	{
+		let json = serde_json::to_string(input).context(ErrorKind::Internal(
+			"Could not serialize data to JSON".to_owned(),
+		))?;
+		self.build_request(url, "POST", api_secret, Some(json))
+	}
+
+	fn handle_request<T>(&self, req: Request<Body>) -> Result<T, Error>
+	where
+		for<'de> T: Deserialize<'de>,
+	{
+		let data = self.send_request(req)?;
+		serde_json::from_str(&data).map_err(|e| {
+			e.context(ErrorKind::ResponseError("Cannot parse response".to_owned()))
+				.into()
+		})
+	}
+
+	fn handle_request_async<T>(&self, req: Request<Body>) -> ClientResponseFuture<T>
+	where
+		for<'de> T: Deserialize<'de> + Send + 'static,
+	{
+		Box::new(self.send_request_async(req).and_then(|data| {
+			serde_json::from_str(&data).map_err(|e| {
+				e.context(ErrorKind::ResponseError("Cannot parse response".to_owned()))
+					.into()
+			})
+		}))
+	}
+
+	fn send_request_async(
+		&self,
+		req: Request<Body>,
+	) -> Box<dyn Future<Item = String, Error = Error> + Send> {
+		//TODO: redundant code, enjoy figuring out type params for dynamic dispatch of client
+		match self.use_socks {
+			false => {
+				let https = hyper_rustls::HttpsConnector::new(1);
+				let mut connector = TimeoutConnector::new(https);
+				connector.set_connect_timeout(Some(Duration::from_secs(20)));
+				connector.set_read_timeout(Some(Duration::from_secs(20)));
+				connector.set_write_timeout(Some(Duration::from_secs(20)));
+				let client = hyper::Client::builder().build::<_, hyper::Body>(connector);
+				Box::new(
+					client
+						.request(req)
+						.map_err(|e| {
+							ErrorKind::RequestError(format!("Cannot make request: {}", e)).into()
+						})
+						.and_then(|resp| {
+							if !resp.status().is_success() {
+								Either::A(err(ErrorKind::RequestError(format!(
+									"Wrong response code: {} with data {:?}",
+									resp.status(),
+									resp.body()
+								))
+								.into()))
+							} else {
+								Either::B(
+									resp.into_body()
+										.map_err(|e| {
+											ErrorKind::RequestError(format!(
+												"Cannot read response body: {}",
+												e
+											))
+											.into()
+										})
+										.concat2()
+										.and_then(|ch| {
+											ok(String::from_utf8_lossy(&ch.to_vec()).to_string())
+										}),
+								)
+							}
+						}),
+				)
+			}
+			true => {
+				let addr = match self.socks_proxy_addr {
+					Some(a) => a,
+					None => {
+						return Box::new(result(Err(ErrorKind::RequestError(format!(
+							"Can't parse Socks proxy address"
+						))
+						.into())))
+					}
+				};
+				let socks_connector = Socksv5Connector::new(addr);
+				let mut connector = TimeoutConnector::new(socks_connector);
+				connector.set_connect_timeout(Some(Duration::from_secs(20)));
+				connector.set_read_timeout(Some(Duration::from_secs(20)));
+				connector.set_write_timeout(Some(Duration::from_secs(20)));
+				let client = hyper::Client::builder().build::<_, hyper::Body>(connector);
+				Box::new(
+					client
+						.request(req)
+						.map_err(|e| {
+							ErrorKind::RequestError(format!("Cannot make request: {}", e)).into()
+						})
+						.and_then(|resp| {
+							if !resp.status().is_success() {
+								Either::A(err(ErrorKind::RequestError(format!(
+									"Wrong response code: {} with data {:?}",
+									resp.status(),
+									resp.body()
+								))
+								.into()))
+							} else {
+								Either::B(
+									resp.into_body()
+										.map_err(|e| {
+											ErrorKind::RequestError(format!(
+												"Cannot read response body: {}",
+												e
+											))
+											.into()
+										})
+										.concat2()
+										.and_then(|ch| {
+											ok(String::from_utf8_lossy(&ch.to_vec()).to_string())
+										}),
+								)
+							}
+						}),
+				)
+			}
+		}
+	}
+
+	pub fn send_request(&self, req: Request<Body>) -> Result<String, Error> {
+		let task = self.send_request_async(req);
+		let mut rt =
+			Runtime::new().context(ErrorKind::Internal("can't create Tokio runtime".to_owned()))?;
+		Ok(rt.block_on(task)?)
+	}
+}

impls/src/client_utils/mod.rs

@@ -0,0 +1,19 @@
+// Copyright 2019 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.
+
+mod client;
+mod socksv5;
+
+pub use self::socksv5::Socksv5Connector;
+pub use client::{Client, Error as ClientError};

impls/src/client_utils/socksv5.rs

@@ -0,0 +1,254 @@
+// MIT License
+//
+// Copyright (c) 2017 Vesa Vilhonen
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+// Copyright 2019 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 byteorder::{BigEndian, WriteBytesExt};
+use futures::future::ok;
+use futures::{Future, IntoFuture};
+//use hyper_tls::MaybeHttpsStream;
+//use native_tls::TlsConnector;
+use std::io::{self, Error, ErrorKind, Write};
+use std::net::SocketAddr;
+use tokio_io::io::{read_exact, write_all};
+use tokio_tcp::TcpStream;
+//use tokio_tls::TlsConnectorExt;
+use hyper::client::connect::{Connect, Connected, Destination};
+
+pub struct Socksv5Connector {
+	proxy_addr: SocketAddr,
+	creds: Option<(Vec<u8>, Vec<u8>)>,
+}
+
+impl Socksv5Connector {
+	pub fn new(proxy_addr: SocketAddr) -> Socksv5Connector {
+		Socksv5Connector {
+			proxy_addr,
+			creds: None,
+		}
+	}
+
+	pub fn _new_with_creds<T: Into<Vec<u8>>>(
+		proxy_addr: SocketAddr,
+		creds: (T, T),
+	) -> io::Result<Socksv5Connector> {
+		let username = creds.0.into();
+		let password = creds.1.into();
+		if username.len() > 255 || password.len() > 255 {
+			Err(Error::new(ErrorKind::Other, "invalid credentials"))
+		} else {
+			Ok(Socksv5Connector {
+				proxy_addr,
+				creds: Some((username, password)),
+			})
+		}
+	}
+}
+
+impl Connect for Socksv5Connector {
+	type Transport = TcpStream;
+	type Error = Error;
+	type Future = Box<dyn Future<Item = (Self::Transport, Connected), Error = Self::Error> + Send>;
+
+	fn connect(&self, dst: Destination) -> Self::Future {
+		let creds = self.creds.clone();
+		Box::new(
+			TcpStream::connect(&self.proxy_addr)
+				.and_then(move |socket| do_handshake(socket, dst, creds)),
+		)
+	}
+}
+
+type HandshakeFutureConnected<T> = Box<dyn Future<Item = (T, Connected), Error = Error> + Send>;
+type HandshakeFuture<T> = Box<dyn Future<Item = T, Error = Error> + Send>;
+
+fn auth_negotiation(
+	socket: TcpStream,
+	creds: Option<(Vec<u8>, Vec<u8>)>,
+) -> HandshakeFuture<TcpStream> {
+	let (username, password) = creds.unwrap();
+	let mut creds_msg: Vec<u8> = Vec::with_capacity(username.len() + password.len() + 3);
+	creds_msg.push(1);
+	creds_msg.push(username.len() as u8);
+	creds_msg.extend_from_slice(&username);
+	creds_msg.push(password.len() as u8);
+	creds_msg.extend_from_slice(&password);
+	Box::new(
+		write_all(socket, creds_msg)
+			.and_then(|(socket, _)| read_exact(socket, [0; 2]))
+			.and_then(|(socket, resp)| {
+				if resp[0] == 1 && resp[1] == 0 {
+					Ok(socket)
+				} else {
+					Err(Error::new(ErrorKind::InvalidData, "unauthorized"))
+				}
+			}),
+	)
+}
+
+fn answer_hello(
+	socket: TcpStream,
+	response: [u8; 2],
+	creds: Option<(Vec<u8>, Vec<u8>)>,
+) -> HandshakeFuture<TcpStream> {
+	if response[0] == 5 && response[1] == 0 {
+		Box::new(ok(socket))
+	} else if response[0] == 5 && response[1] == 2 && creds.is_some() {
+		Box::new(auth_negotiation(socket, creds).and_then(|socket| ok(socket)))
+	} else {
+		Box::new(
+			Err(Error::new(
+				ErrorKind::InvalidData,
+				"wrong response from socks server",
+			))
+			.into_future(),
+		)
+	}
+}
+
+fn write_addr(socket: TcpStream, req: Destination) -> HandshakeFuture<TcpStream> {
+	let host = req.host();
+	if host.len() > u8::max_value() as usize {
+		return Box::new(Err(Error::new(ErrorKind::InvalidInput, "Host too long")).into_future());
+	}
+
+	let port = match req.port() {
+		Some(port) => port,
+		_ if req.scheme() == "https" => 443,
+		_ if req.scheme() == "http" => 80,
+		_ => {
+			return Box::new(
+				Err(Error::new(
+					ErrorKind::InvalidInput,
+					"Supports only http/https",
+				))
+				.into_future(),
+			)
+		}
+	};
+
+	let mut packet = Vec::new();
+	packet.write_all(&vec![5, 1, 0]).unwrap();
+
+	packet.write_u8(3).unwrap();
+	packet.write_u8(host.as_bytes().len() as u8).unwrap();
+	packet.write_all(host.as_bytes()).unwrap();
+	packet.write_u16::<BigEndian>(port).unwrap();
+
+	Box::new(write_all(socket, packet).map(|(socket, _)| socket))
+}
+
+fn read_response(socket: TcpStream, response: [u8; 3]) -> HandshakeFuture<TcpStream> {
+	if response[0] != 5 {
+		return Box::new(Err(Error::new(ErrorKind::Other, "invalid version")).into_future());
+	}
+	match response[1] {
+		0 => {}
+		1 => {
+			return Box::new(
+				Err(Error::new(ErrorKind::Other, "general SOCKS server failure")).into_future(),
+			)
+		}
+		2 => {
+			return Box::new(
+				Err(Error::new(
+					ErrorKind::Other,
+					"connection not allowed by ruleset",
+				))
+				.into_future(),
+			)
+		}
+		3 => {
+			return Box::new(Err(Error::new(ErrorKind::Other, "network unreachable")).into_future())
+		}
+		4 => return Box::new(Err(Error::new(ErrorKind::Other, "host unreachable")).into_future()),
+		5 => {
+			return Box::new(Err(Error::new(ErrorKind::Other, "connection refused")).into_future())
+		}
+		6 => return Box::new(Err(Error::new(ErrorKind::Other, "TTL expired")).into_future()),
+		7 => {
+			return Box::new(
+				Err(Error::new(ErrorKind::Other, "command not supported")).into_future(),
+			)
+		}
+		8 => {
+			return Box::new(
+				Err(Error::new(ErrorKind::Other, "address kind not supported")).into_future(),
+			)
+		}
+		_ => return Box::new(Err(Error::new(ErrorKind::Other, "unknown error")).into_future()),
+	};
+
+	if response[2] != 0 {
+		return Box::new(
+			Err(Error::new(ErrorKind::InvalidData, "invalid reserved byt")).into_future(),
+		);
+	}
+
+	Box::new(
+		read_exact(socket, [0; 1])
+			.and_then(|(socket, response)| match response[0] {
+				1 => read_exact(socket, [0; 6]),
+				_ => unimplemented!(),
+			})
+			.map(|(socket, _)| socket),
+	)
+}
+
+fn do_handshake(
+	socket: TcpStream,
+	req: Destination,
+	creds: Option<(Vec<u8>, Vec<u8>)>,
+) -> HandshakeFutureConnected<TcpStream> {
+	let _is_https = req.scheme() == "https";
+	let _host = req.host();
+	let method: u8 = creds.clone().map(|_| 2).unwrap_or(0);
+	let established = write_all(socket, [5, 1, method])
+		.and_then(|(socket, _)| read_exact(socket, [0; 2]))
+		.and_then(|(socket, response)| answer_hello(socket, response, creds))
+		.and_then(|socket| write_addr(socket, req))
+		.and_then(|socket| read_exact(socket, [0; 3]))
+		.and_then(|(socket, response)| read_response(socket, response));
+	/*if is_https {
+			Box::new(established.and_then(move |socket| {
+					let tls = TlsConnector::builder().unwrap().build().unwrap();
+					tls.connect_async(&host, socket)
+							.map_err(|err| Error::new(ErrorKind::Other, err))
+							.map(|socket| MaybeHttpsStream::Https(socket))
+			}))
+	} else {*/
+	//Box::new(established.map(|socket| TcpStream::Http(socket)))
+	Box::new(established.map(|socket| (socket, Connected::new())))
+	/*}*/
+}