Use IPC transport in mullvad-ipc-client

2 files changed, 109 insertions, 208 deletions

diff --git a/mullvad-ipc-client/Cargo.toml b/mullvad-ipc-client/Cargo.toml
index 7a7513eeab..443cffa885 100644
--- a/mullvad-ipc-client/Cargo.toml
+++ b/mullvad-ipc-client/Cargo.toml
@@ -12,6 +12,13 @@ serde = "1.0"
 talpid-ipc = { path = "../talpid-ipc" }
 talpid-types = { path = "../talpid-types" }
 mullvad-paths = { path = "../mullvad-paths" }
+jsonrpc-client-core = { git = "https://github.com/mullvad/jsonrpc-client-rs", branch = "master" }
+jsonrpc-client-ipc = { git = "https://github.com/mullvad/jsonrpc-client-rs", branch = "master" }
+tokio-core = "0.1"
+tokio-timer = "0.1"
+futures = "0.1"
+log = "0.4"
+
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3.5", features = ["accctrl", "aclapi", "securitybaseapi", "winbase", "winerror", "winnt"] }
diff --git a/mullvad-ipc-client/src/lib.rs b/mullvad-ipc-client/src/lib.rs
index 6c1734a4c9..62a829b4d0 100644
--- a/mullvad-ipc-client/src/lib.rs
+++ b/mullvad-ipc-client/src/lib.rs
@@ -1,15 +1,25 @@
 #[macro_use]
+extern crate log;
+
+#[macro_use]
 extern crate error_chain;
+
+extern crate jsonrpc_client_core;
+extern crate jsonrpc_client_ipc;
+
+extern crate futures;
 extern crate mullvad_paths;
 extern crate mullvad_types;
 extern crate serde;
 extern crate talpid_ipc;
 extern crate talpid_types;
+extern crate tokio_core;
+extern crate tokio_timer;
 
-use std::fs::File;
-use std::io::{BufRead, BufReader};
-use std::path::{Path, PathBuf};
+use std::path::Path;
 use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
 
 use mullvad_types::account::{AccountData, AccountToken};
 use mullvad_types::location::GeoIpLocation;
@@ -18,10 +28,14 @@ use mullvad_types::relay_list::RelayList;
 use mullvad_types::states::DaemonState;
 use mullvad_types::version::AppVersionInfo;
 use serde::{Deserialize, Serialize};
-use talpid_ipc::WsIpcClient;
 use talpid_types::net::TunnelOptions;
 
-use platform_specific::ensure_written_by_admin;
+use futures::stream::{self, Stream};
+use futures::sync::oneshot;
+use jsonrpc_client_core::{Client, ClientHandle, Future};
+pub use jsonrpc_client_core::{Error as RpcError, ErrorKind as RpcErrorKind};
+use jsonrpc_client_ipc::IpcTransport;
+use tokio_core::reactor;
 
 error_chain! {
     errors {
@@ -29,32 +43,6 @@ error_chain! {
             description("Failed to authenticate the connection with the daemon")
         }
 
-        EmptyRpcFile(path: PathBuf) {
-            description("RPC connection file is empty")
-            display("RPC connection file \"{}\" is empty", path.display())
-        }
-
-        InsecureRpcFile(path: PathBuf) {
-            description(
-                "RPC connection file is insecure because it might not have been written by an \
-                administrator user"
-            )
-            display(
-                "RPC connection file \"{}\" is insecure because it might not have been written by \
-                an administrator user", path.display()
-            )
-        }
-
-        MissingRpcCredentials(path: PathBuf) {
-            description("no credentials found in RPC connection file")
-            display("no credentials found in RPC connection file {}", path.display())
-        }
-
-        ReadRpcFileError(path: PathBuf) {
-            description("Failed to read RPC connection information")
-            display("Failed to read RPC connection information from {}", path.display())
-        }
-
         RpcCallError(method: String) {
             description("Failed to call RPC method")
             display("Failed to call RPC method \"{}\"", method)
@@ -69,6 +57,14 @@ error_chain! {
             description("Failed to start RPC client")
             display("Failed to start RPC client to {}", address)
         }
+
+        TokioError {
+            description("Failed to setup a standalone event loop")
+        }
+
+        TransportError {
+            description("Failed to setup a transport")
+        }
     }
     links {
         UnknownRpcAddressPath(mullvad_paths::Error, mullvad_paths::ErrorKind);
@@ -77,66 +73,60 @@ error_chain! {
 
 static NO_ARGS: [u8; 0] = [];
 
-pub struct DaemonRpcClient {
-    rpc_client: WsIpcClient,
-}
-
-impl DaemonRpcClient {
-    pub fn new() -> Result<Self> {
-        Self::with_rpc_address_file(mullvad_paths::get_rpc_address_path()?)
-    }
-
-    pub fn with_rpc_address_file<P: AsRef<Path>>(file_path: P) -> Result<Self> {
-        ensure_written_by_admin(&file_path)?;
-
-        let (address, credentials) = Self::read_rpc_file(file_path)?;
-
-        Self::with_address_and_credentials(address, credentials)
-    }
-
-    pub fn with_insecure_rpc_address_file<P: AsRef<Path>>(file_path: P) -> Result<Self> {
-        let (address, credentials) = Self::read_rpc_file(file_path)?;
-
-        Self::with_address_and_credentials(address, credentials)
-    }
-
-    fn with_address_and_credentials(address: String, credentials: String) -> Result<Self> {
-        let rpc_client =
-            WsIpcClient::connect(&address).chain_err(|| ErrorKind::StartRpcClient(address))?;
-        let mut instance = DaemonRpcClient { rpc_client };
+pub fn new_standalone_transport<
+    F: Send + 'static + FnOnce(String, reactor::Handle) -> Result<T>,
+    T: jsonrpc_client_core::Transport,
+>(
+    rpc_path: String,
+    transport_func: F,
+) -> Result<DaemonRpcClient> {
+    let (tx, rx) = oneshot::channel();
+    thread::spawn(move || match spawn_transport(rpc_path, transport_func) {
+        Err(e) => tx
+            .send(Err(e))
+            .expect("Failed to send error back to caller"),
+        Ok((mut core, client, client_handle)) => {
+            tx.send(Ok(client_handle))
+                .expect("Failed to send client handle");
+            if let Err(e) = core.run(client) {
+                error!("JSON-RPC client failed: {}", e.description());
+            }
+        }
+    });
 
-        instance
-            .auth(&credentials)
-            .chain_err(|| ErrorKind::AuthenticationError)?;
+    rx.wait()
+        .chain_err(|| ErrorKind::TransportError)?
+        .map(|client_handle| DaemonRpcClient::new(client_handle))
+}
 
-        Ok(instance)
-    }
+pub fn new_standalone_ipc_client(path: &impl AsRef<Path>) -> Result<DaemonRpcClient> {
+    let path = path.as_ref().to_string_lossy().to_string();
 
-    fn read_rpc_file<P>(file_path: P) -> Result<(String, String)>
-    where
-        P: AsRef<Path>,
-    {
-        let file_path = file_path.as_ref();
-        let rpc_file = File::open(file_path)
-            .chain_err(|| ErrorKind::ReadRpcFileError(file_path.to_owned()))?;
+    new_standalone_transport(path, |path, handle| {
+        IpcTransport::new(&path, &handle).chain_err(|| ErrorKind::TransportError)
+    })
+}
 
-        let reader = BufReader::new(rpc_file);
-        let mut lines = reader.lines();
+fn spawn_transport<
+    F: Send + FnOnce(String, reactor::Handle) -> Result<T>,
+    T: jsonrpc_client_core::Transport,
+>(
+    address: String,
+    transport_func: F,
+) -> Result<(reactor::Core, Client<T>, ClientHandle)> {
+    let core = reactor::Core::new().chain_err(|| ErrorKind::TokioError)?;
+    let (client, client_handle) = transport_func(address, core.handle())?.into_client();
+    Ok((core, client, client_handle))
+}
 
-        let address = lines
-            .next()
-            .ok_or_else(|| ErrorKind::EmptyRpcFile(file_path.to_owned()))?
-            .chain_err(|| ErrorKind::ReadRpcFileError(file_path.to_owned()))?;
-        let credentials = lines
-            .next()
-            .ok_or_else(|| ErrorKind::MissingRpcCredentials(file_path.to_owned()))?
-            .chain_err(|| ErrorKind::ReadRpcFileError(file_path.to_owned()))?;
+pub struct DaemonRpcClient {
+    rpc_client: jsonrpc_client_core::ClientHandle,
+}
 
-        Ok((address, credentials))
-    }
 
-    pub fn auth(&mut self, credentials: &str) -> Result<()> {
-        self.call("auth", &[credentials])
+impl DaemonRpcClient {
+    pub fn new(rpc_client: ClientHandle) -> Self {
+        DaemonRpcClient { rpc_client }
     }
 
     pub fn connect(&mut self) -> Result<()> {
@@ -219,139 +209,43 @@ impl DaemonRpcClient {
         self.call("update_relay_settings", &[update])
     }
 
-    pub fn call<A, O>(&mut self, method: &str, args: &A) -> Result<O>
+    pub fn call<A, O>(&mut self, method: &'static str, args: &A) -> Result<O>
     where
-        A: Serialize,
-        O: for<'de> Deserialize<'de>,
+        A: Serialize + Send + 'static,
+        O: for<'de> Deserialize<'de> + Send + 'static,
     {
         self.rpc_client
-            .call(method, args)
+            .call_method(method, args)
+            .wait()
             .chain_err(|| ErrorKind::RpcCallError(method.to_owned()))
     }
 
     pub fn new_state_subscribe(&mut self) -> Result<mpsc::Receiver<DaemonState>> {
-        self.subscribe("new_state")
-    }
-
-    pub fn subscribe<T>(&mut self, event: &str) -> Result<mpsc::Receiver<T>>
-    where
-        T: for<'de> serde::Deserialize<'de> + Send + 'static,
-    {
-        let (event_tx, event_rx) = mpsc::channel();
-        let subscribe_method = format!("{}_subscribe", event);
-        let unsubscribe_method = format!("{}_unsubscribe", event);
-
-        self.rpc_client
-            .subscribe::<T, T>(subscribe_method, unsubscribe_method, event_tx)
-            .chain_err(|| ErrorKind::RpcSubscribeError(event.to_owned()))?;
-
-        Ok(event_rx)
-    }
-}
-
-#[cfg(unix)]
-mod platform_specific {
-    use std::os::unix::fs::MetadataExt;
-
-    use super::*;
-
-    pub fn ensure_written_by_admin<P: AsRef<Path>>(path: P) -> Result<()> {
-        let path = path.as_ref();
-        let metadata = path
-            .metadata()
-            .chain_err(|| ErrorKind::ReadRpcFileError(path.to_owned()))?;
-
-        let is_owned_by_root = metadata.uid() == 0;
-        let is_read_only_by_non_owner = (metadata.mode() & 0o022) == 0;
-
-        ensure!(
-            is_owned_by_root && is_read_only_by_non_owner,
-            ErrorKind::InsecureRpcFile(path.to_owned())
-        );
-
-        Ok(())
-    }
-}
-
-#[cfg(windows)]
-mod platform_specific {
-    extern crate winapi;
+        let client = self.rpc_client.clone();
+        let mut current_state = self.get_state()?;
+        let first_message = stream::once(Ok(current_state.clone()));
 
-    use std::iter::once;
-    use std::os::windows::ffi::OsStrExt;
-    use std::ptr;
+        let (tx, rx) = mpsc::channel();
 
-    use self::winapi::shared::winerror::ERROR_SUCCESS;
-    use self::winapi::um::accctrl::SE_FILE_OBJECT;
-    use self::winapi::um::aclapi::GetNamedSecurityInfoW;
-    use self::winapi::um::securitybaseapi::IsWellKnownSid;
-    use self::winapi::um::winbase::LocalFree;
-    use self::winapi::um::winnt::{
-        WinBuiltinAdministratorsSid, OWNER_SECURITY_INFORMATION, PSECURITY_DESCRIPTOR, PSID,
-    };
+        let polled = tokio_timer::wheel()
+            .build()
+            .interval(Duration::from_secs(1))
+            .then(move |_| client.call_method("get_state", &NO_ARGS));
 
-    use super::*;
-
-    mod errors {
-        error_chain! {
-            errors {
-                GetSecurityInfoError {
-                    description("Failed to get security information of RPC address file")
-                }
-
-                OwnerNotAdmin {
-                    description("Owner of RPC address file is not an administrator")
-                }
-            }
-        }
-    }
-    use self::errors::{ErrorKind as WinErrorKind, Result as WinResult};
-
-    pub fn ensure_written_by_admin<P: AsRef<Path>>(file_path: P) -> Result<()> {
-        let path = file_path.as_ref();
-
-        ensure_owned_by_admin(&path).chain_err(|| ErrorKind::InsecureRpcFile(path.to_owned()))?;
-
-        Ok(())
-    }
-
-    fn ensure_owned_by_admin<P: AsRef<Path>>(path: P) -> WinResult<()> {
-        let file_path: Vec<u16> = path
-            .as_ref()
-            .as_os_str()
-            .encode_wide()
-            .chain(once(0))
-            .collect();
-
-        unsafe {
-            let mut owner_sid: PSID = ptr::null_mut();
-            let mut security_descriptor: PSECURITY_DESCRIPTOR = ptr::null_mut();
-
-            let get_security_info_result = GetNamedSecurityInfoW(
-                file_path.as_ptr(),
-                SE_FILE_OBJECT,
-                OWNER_SECURITY_INFORMATION,
-                &mut owner_sid,
-                ptr::null_mut(),
-                ptr::null_mut(),
-                ptr::null_mut(),
-                &mut security_descriptor,
-            );
-
-            ensure!(
-                get_security_info_result == ERROR_SUCCESS,
-                WinErrorKind::GetSecurityInfoError
-            );
-
-            let sid_check_result = IsWellKnownSid(owner_sid, WinBuiltinAdministratorsSid);
-
-            if !LocalFree(security_descriptor as *mut _).is_null() {
-                panic!("Failed to deallocate security descriptor");
-            }
-
-            ensure!(sid_check_result != 0, WinErrorKind::OwnerNotAdmin);
-
-            Ok(())
-        }
+        thread::spawn(move || {
+            let _ = first_message
+                .chain(polled)
+                .for_each(move |state| {
+                    if state != current_state {
+                        current_state = state;
+                        if tx.send(state).is_err() {
+                            trace!("can't send new state to subscriber");
+                            return Err(jsonrpc_client_core::ErrorKind::Shutdown.into());
+                        };
+                    }
+                    Ok(())
+                }).wait();
+        });
+        Ok(rx)
     }
 }