Merge branch 'daemon-cleanup'

8 files changed, 74 insertions, 281 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 6c54d1219c..750463b577 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -147,6 +147,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "num 0.1.39 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)",
  "time 0.1.38 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -537,6 +538,7 @@ dependencies = [
 name = "mullvad-types"
 version = "0.1.0"
 dependencies = [
+ "chrono 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_derive 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -676,7 +678,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 [[package]]
 name = "pfctl"
 version = "0.1.0"
-source = "git+ssh://git@github.com/mullvad/pfctl-rs.git#19f9b6104cf4ba903b7f1d96cec87779eff7ae08"
+source = "git+https://github.com/mullvad/pfctl-rs.git#7d4c6ec799e3b6c256f1db84a0daec3bb0cd180a"
 dependencies = [
  "bindgen 0.29.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "derive_builder 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -919,7 +921,7 @@ dependencies = [
  "log 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "mktemp 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "openvpn-plugin 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "pfctl 0.1.0 (git+ssh://git@github.com/mullvad/pfctl-rs.git)",
+ "pfctl 0.1.0 (git+https://github.com/mullvad/pfctl-rs.git)",
  "talpid-ipc 0.1.0",
 ]
 
@@ -1223,7 +1225,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum parking_lot_core 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "0ad2c4d148942b3560034785bf19df586ebba53351e8c78f84984147d5795eef"
 "checksum peeking_take_while 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099"
 "checksum percent-encoding 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "de154f638187706bde41d9b4738748933d64e6b37bdbffc0b47a97d16a6ae356"
-"checksum pfctl 0.1.0 (git+ssh://git@github.com/mullvad/pfctl-rs.git)" = "<none>"
+"checksum pfctl 0.1.0 (git+https://github.com/mullvad/pfctl-rs.git)" = "<none>"
 "checksum quasi 0.32.0 (registry+https://github.com/rust-lang/crates.io-index)" = "18c45c4854d6d1cf5d531db97c75880feb91c958b0720f4ec1057135fec358b3"
 "checksum quasi_codegen 0.32.0 (registry+https://github.com/rust-lang/crates.io-index)" = "51b9e25fa23c044c1803f43ca59c98dac608976dd04ce799411edd58ece776d4"
 "checksum quote 0.3.15 (registry+https://github.com/rust-lang/crates.io-index)" = "7a6e920b65c65f10b2ae65c831a81a073a89edd28c7cce89475bff467ab4167a"
diff --git a/mullvad-daemon/src/main.rs b/mullvad-daemon/src/main.rs
index 3479a231a1..4b7b0f5745 100644
--- a/mullvad-daemon/src/main.rs
+++ b/mullvad-daemon/src/main.rs
@@ -31,7 +31,7 @@ mod settings;
 mod shutdown;
 
 use error_chain::ChainedError;
-use jsonrpc_core::futures::sync;
+use jsonrpc_core::futures::sync::oneshot::Sender as OneshotSender;
 use management_interface::{ManagementInterfaceServer, TunnelCommand};
 use mullvad_types::states::{DaemonState, SecurityState, TargetState};
 use std::io;
@@ -294,14 +294,12 @@ impl Daemon {
         }
     }
 
-    fn on_get_state(&self, tx: sync::oneshot::Sender<DaemonState>) {
-        if let Err(_) = tx.send(self.last_broadcasted_state) {
-            warn!("Unable to send current state to management interface client",);
-        }
+    fn on_get_state(&self, tx: OneshotSender<DaemonState>) {
+        Self::oneshot_send(tx, self.last_broadcasted_state, "current state");
     }
 
     fn on_set_account(&mut self,
-                      tx: sync::oneshot::Sender<()>,
+                      tx: OneshotSender<()>,
                       account_token: Option<String>)
                       -> Result<()> {
 
@@ -309,10 +307,7 @@ impl Daemon {
 
         match save_result.chain_err(|| "Unable to save settings") {
             Ok(account_changed) => {
-                if let Err(_) = tx.send(()) {
-                    warn!("Unable to send response to management interface client");
-                }
-
+                Self::oneshot_send(tx, (), "set_account response");
                 let tunnel_needs_restart = self.state == TunnelState::Connecting ||
                                            self.state == TunnelState::Connected;
                 if account_changed && tunnel_needs_restart {
@@ -325,9 +320,13 @@ impl Daemon {
         Ok(())
     }
 
-    fn on_get_account(&self, tx: sync::oneshot::Sender<Option<String>>) {
-        if let Err(_) = tx.send(self.settings.get_account_token()) {
-            warn!("Unable to send current account to management interface client");
+    fn on_get_account(&self, tx: OneshotSender<Option<String>>) {
+        Self::oneshot_send(tx, self.settings.get_account_token(), "current account")
+    }
+
+    fn oneshot_send<T>(tx: OneshotSender<T>, t: T, msg: &'static str) {
+        if let Err(_) = tx.send(t) {
+            warn!("Unable to send {} to management interface client", msg);
         }
     }
 
diff --git a/mullvad-daemon/src/management_interface.rs b/mullvad-daemon/src/management_interface.rs
index f78f991c0e..9401f4ad5e 100644
--- a/mullvad-daemon/src/management_interface.rs
+++ b/mullvad-daemon/src/management_interface.rs
@@ -2,9 +2,12 @@ use error_chain;
 
 use jsonrpc_core::{Error, ErrorCode, Metadata};
 use jsonrpc_core::futures::{BoxFuture, Future, future, sync};
+use jsonrpc_core::futures::sync::oneshot::Sender as OneshotSender;
 use jsonrpc_macros::pubsub;
 use jsonrpc_pubsub::{PubSubHandler, PubSubMetadata, Session, SubscriptionId};
 use jsonrpc_ws_server;
+use mullvad_types::account::{AccountData, AccountToken};
+use mullvad_types::location::{CountryCode, Location};
 use mullvad_types::states::{DaemonState, TargetState};
 
 use serde;
@@ -19,22 +22,6 @@ use talpid_ipc;
 use uuid;
 
 
-pub type AccountToken = String;
-pub type CountryCode = String;
-
-#[derive(Serialize)]
-pub struct AccountData {
-    pub paid_until: String,
-}
-
-#[derive(Serialize)]
-pub struct Location {
-    pub latlong: [f64; 2],
-    pub country: String,
-    pub city: String,
-}
-
-
 build_rpc_trait! {
     pub trait ManagementInterfaceApi {
         type Metadata;
@@ -65,13 +52,13 @@ build_rpc_trait! {
         fn set_autoconnect(&self, bool) -> Result<(), Error>;
 
         /// Try to connect if disconnected, or do nothing if already connecting/connected.
-        #[rpc(name = "connect")]
-        fn connect(&self) -> Result<(), Error>;
+        #[rpc(async, name = "connect")]
+        fn connect(&self) -> BoxFuture<(), Error>;
 
         /// Disconnect the VPN tunnel if it is connecting/connected. Does nothing if already
         /// disconnected.
-        #[rpc(name = "disconnect")]
-        fn disconnect(&self) -> Result<(), Error>;
+        #[rpc(async, name = "disconnect")]
+        fn disconnect(&self) -> BoxFuture<(), Error>;
 
         /// Returns the current state of the Mullvad client. Changes to this state will
         /// be announced to subscribers of `new_state`.
@@ -116,11 +103,11 @@ pub enum TunnelCommand {
     /// Change target state.
     SetTargetState(TargetState),
     /// Request the current state.
-    GetState(sync::oneshot::Sender<DaemonState>),
+    GetState(OneshotSender<DaemonState>),
     /// Set which account token to use for subsequent connection attempts.
-    SetAccount(sync::oneshot::Sender<()>, Option<AccountToken>),
+    SetAccount(OneshotSender<()>, Option<AccountToken>),
     /// Request the current account token being used.
-    GetAccount(sync::oneshot::Sender<Option<AccountToken>>),
+    GetAccount(OneshotSender<Option<AccountToken>>),
 }
 
 #[derive(Default)]
@@ -245,6 +232,13 @@ impl<T: From<TunnelCommand> + 'static + Send> ManagementInterface<T> {
         };
         result.boxed()
     }
+
+    /// Sends a command to the daemon and maps the error to an RPC error.
+    fn send_command_to_daemon(&self, command: TunnelCommand) -> BoxFuture<(), Error> {
+        future::result(self.tx.lock().unwrap().send(command))
+            .map_err(|_| Error::internal_error())
+            .boxed()
+    }
 }
 
 impl<T: From<TunnelCommand> + 'static + Send> ManagementInterfaceApi for ManagementInterface<T> {
@@ -252,7 +246,12 @@ impl<T: From<TunnelCommand> + 'static + Send> ManagementInterfaceApi for Managem
 
     fn get_account_data(&self, _account_token: AccountToken) -> Result<AccountData, Error> {
         trace!("get_account_data");
-        Ok(AccountData { paid_until: "2018-12-31T16:00:00.000Z".to_owned() },)
+        // Just mock implementation, so locally importing temporarily.
+        use chrono::DateTime;
+        use chrono::offset::Utc;
+        use std::str::FromStr;
+        let expiry: DateTime<Utc> = DateTime::from_str("2018-12-31T16:00:00.000Z").unwrap();
+        Ok(AccountData { expiry })
     }
 
     fn get_countries(&self) -> Result<HashMap<CountryCode, String>, Error> {
@@ -263,19 +262,17 @@ impl<T: From<TunnelCommand> + 'static + Send> ManagementInterfaceApi for Managem
     fn set_account(&self, account_token: Option<AccountToken>) -> BoxFuture<(), Error> {
         trace!("set_account");
         let (tx, rx) = sync::oneshot::channel();
-        match self.tx.lock().unwrap().send(TunnelCommand::SetAccount(tx, account_token)) {
-            Ok(()) => rx.map_err(|_| Error::internal_error()).boxed(),
-            Err(_) => future::err(Error::internal_error()).boxed(),
-        }
+        self.send_command_to_daemon(TunnelCommand::SetAccount(tx, account_token))
+            .and_then(|_| rx.map_err(|_| Error::internal_error()))
+            .boxed()
     }
 
     fn get_account(&self) -> BoxFuture<Option<AccountToken>, Error> {
         trace!("get_account");
         let (tx, rx) = sync::oneshot::channel();
-        match self.tx.lock().unwrap().send(TunnelCommand::GetAccount(tx)) {
-            Ok(()) => rx.map_err(|_| Error::internal_error()).boxed(),
-            Err(_) => future::err(Error::internal_error()).boxed(),
-        }
+        self.send_command_to_daemon(TunnelCommand::GetAccount(tx))
+            .and_then(|_| rx.map_err(|_| Error::internal_error()))
+            .boxed()
     }
 
     fn set_country(&self, _country_code: CountryCode) -> Result<(), Error> {
@@ -288,31 +285,22 @@ impl<T: From<TunnelCommand> + 'static + Send> ManagementInterfaceApi for Managem
         Ok(())
     }
 
-    fn connect(&self) -> Result<(), Error> {
+    fn connect(&self) -> BoxFuture<(), Error> {
         trace!("connect");
-        self.tx
-            .lock()
-            .unwrap()
-            .send(TunnelCommand::SetTargetState(TargetState::Secured))
-            .map_err(|_| Error::internal_error())
+        self.send_command_to_daemon(TunnelCommand::SetTargetState(TargetState::Secured))
     }
 
-    fn disconnect(&self) -> Result<(), Error> {
+    fn disconnect(&self) -> BoxFuture<(), Error> {
         trace!("disconnect");
-        self.tx
-            .lock()
-            .unwrap()
-            .send(TunnelCommand::SetTargetState(TargetState::Unsecured))
-            .map_err(|_| Error::internal_error())
+        self.send_command_to_daemon(TunnelCommand::SetTargetState(TargetState::Unsecured))
     }
 
     fn get_state(&self) -> BoxFuture<DaemonState, Error> {
         trace!("get_state");
         let (state_tx, state_rx) = sync::oneshot::channel();
-        match self.tx.lock().unwrap().send(TunnelCommand::GetState(state_tx)) {
-            Ok(()) => state_rx.map_err(|_| Error::internal_error()).boxed(),
-            Err(_) => future::err(Error::internal_error()).boxed(),
-        }
+        self.send_command_to_daemon(TunnelCommand::GetState(state_tx))
+            .and_then(|_| state_rx.map_err(|_| Error::internal_error()))
+            .boxed()
     }
 
     fn get_ip(&self) -> Result<IpAddr, Error> {
diff --git a/mullvad-daemon/src/mock_ipc.rs b/mullvad-daemon/src/mock_ipc.rs
deleted file mode 100644
index 90859401c2..0000000000
--- a/mullvad-daemon/src/mock_ipc.rs
+++ /dev/null
@@ -1,217 +0,0 @@
-use ipc_api::*;
-
-use jsonrpc_core::{self, Error, ErrorCode, Metadata};
-use jsonrpc_core::futures::{BoxFuture, Future, future};
-use jsonrpc_macros::pubsub;
-use jsonrpc_pubsub::{PubSubHandler, PubSubMetadata, Session, SubscriptionId};
-use jsonrpc_ws_server;
-
-use std::collections::HashMap;
-use std::net::{IpAddr, Ipv4Addr};
-use std::sync::{Arc, RwLock, atomic};
-
-use talpid_ipc;
-
-type ActiveSubscriptions = Arc<RwLock<HashMap<SubscriptionId, pubsub::Sink<String>>>>;
-
-pub struct IpcServer {
-    server: talpid_ipc::IpcServer,
-}
-
-impl IpcServer {
-    pub fn start() -> talpid_ipc::Result<Self> {
-        let active_subscriptions = ActiveSubscriptions::default();
-        let rpc = MockIpcApi::new(active_subscriptions.clone());
-        let mut io = PubSubHandler::default();
-        io.extend_with(rpc.to_delegate());
-        let server = talpid_ipc::IpcServer::start_with_metadata(io.into(), meta_extractor)?;
-
-        Self::spawn_broadcast_thread(active_subscriptions);
-        Ok(IpcServer { server })
-    }
-
-    pub fn address(&self) -> &str {
-        self.server.address()
-    }
-
-    pub fn wait(self) -> talpid_ipc::Result<()> {
-        self.server.wait()
-    }
-
-    // TODO(linus): This thread will never die. But this is just mock anyway so not important.
-    fn spawn_broadcast_thread(active_subscriptions: ActiveSubscriptions) {
-        ::std::thread::spawn(
-            move || loop {
-                {
-                    let subscribers = active_subscriptions.read().unwrap();
-                    for sink in subscribers.values() {
-                        let _ = sink.notify(Ok("Hello World!".into())).wait();
-                    }
-                }
-                ::std::thread::sleep(::std::time::Duration::from_secs(1));
-            },
-        );
-    }
-}
-
-
-
-/// The metadata type. There is one instance associated with each connection. In this pubsub
-/// scenario they are created by `From<Sender<String>>::from` by the server on each new incoming
-/// connection.
-#[derive(Clone, Debug, Default)]
-pub struct Meta {
-    session: Option<Arc<Session>>,
-}
-
-/// Make the `Meta` type possible to use as jsonrpc metadata type.
-impl Metadata for Meta {}
-
-/// Make the `Meta` type possible to use as a pubsub metadata type.
-impl PubSubMetadata for Meta {
-    fn session(&self) -> Option<Arc<Session>> {
-        self.session.clone()
-    }
-}
-
-/// Metadata extractor function for `Meta`.
-fn meta_extractor(context: &jsonrpc_ws_server::RequestContext) -> Meta {
-    Meta { session: Some(Arc::new(Session::new(context.sender()))) }
-}
-
-/// A mock implementation of the Mullvad frontend API. A very simplified explanation is that for
-/// the real implementation `tunnel_is_up` should be replaced with some kind of handle (or proxy to
-/// a handle) to the jsonrpc client talking with `talpid_core`.
-pub struct MockIpcApi {
-    next_subscription_id: atomic::AtomicUsize,
-    active: ActiveSubscriptions,
-    country: RwLock<CountryCode>,
-    tunnel_is_up: atomic::AtomicBool,
-}
-
-impl MockIpcApi {
-    pub fn new(active: ActiveSubscriptions) -> Self {
-        MockIpcApi {
-            next_subscription_id: atomic::AtomicUsize::new(0),
-            active: active,
-            country: RwLock::new("se".to_owned()),
-            tunnel_is_up: atomic::AtomicBool::new(false),
-        }
-    }
-}
-
-impl IpcApi for MockIpcApi {
-    type Metadata = Meta;
-
-    fn get_account_data(&self, account_token: AccountToken) -> Result<AccountData, Error> {
-        debug!("Login for {}", account_token);
-
-        let paid_until = if account_token.starts_with("1111") {
-            // accounts starting with 1111 expire in one month
-            Ok("2018-12-31T16:00:00.000Z".to_owned())
-        } else if account_token.starts_with("2222") {
-            Ok("2012-12-31T16:00:00.000Z".to_owned())
-        } else if account_token.starts_with("3333") {
-            Ok("2037-12-31T16:00:00.000Z".to_owned())
-        } else {
-            Err(jsonrpc_core::Error::invalid_params("You are not welcome"))
-        }?;
-        Ok(AccountData { paid_until: paid_until })
-    }
-
-    fn get_countries(&self) -> Result<HashMap<CountryCode, String>, Error> {
-        let mut countries = HashMap::new();
-        countries.insert("se".to_owned(), "Sweden".to_owned());
-        countries.insert("de".to_owned(), "Denmark".to_owned());
-        countries.insert("na".to_owned(), "Narnia".to_owned());
-        Ok(countries)
-    }
-
-    fn set_account(&self, _account_token: AccountToken) -> Result<(), Error> {
-        Ok(())
-    }
-
-    fn set_country(&self, country_code: CountryCode) -> Result<(), Error> {
-        *self.country.write().unwrap() = country_code;
-        Ok(())
-    }
-
-    fn set_autoconnect(&self, _autoconnect: bool) -> Result<(), Error> {
-        Ok(())
-    }
-
-    fn connect(&self) -> Result<(), Error> {
-        if self.country.read().unwrap().starts_with("se") {
-            Err(jsonrpc_core::Error::invalid_params("Invalid server"))
-        } else {
-            self.tunnel_is_up.store(true, atomic::Ordering::SeqCst);
-            Ok(())
-        }
-    }
-
-    fn disconnect(&self) -> Result<(), Error> {
-        self.tunnel_is_up.store(false, atomic::Ordering::SeqCst);
-        Ok(())
-    }
-
-    fn get_state(&self) -> Result<SecurityState, Error> {
-        if self.tunnel_is_up.load(atomic::Ordering::SeqCst) {
-            Ok(SecurityState::Secured)
-        } else {
-            Ok(SecurityState::Unsecured)
-        }
-    }
-
-    fn get_ip(&self) -> Result<IpAddr, Error> {
-        let ip = if self.tunnel_is_up.load(atomic::Ordering::SeqCst) {
-                IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4))
-            } else {
-                IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2))
-            }
-            .to_owned();
-        Ok(ip)
-    }
-
-    fn get_location(&self) -> Result<Location, Error> {
-        Ok(
-            if self.tunnel_is_up.load(atomic::Ordering::SeqCst) {
-                Location {
-                    latlong: [1.0, 2.0],
-                    country: "narnia".to_owned(),
-                    city: "Le city".to_owned(),
-                }
-            } else {
-                Location {
-                    latlong: [60.0, 61.0],
-                    country: "sweden".to_owned(),
-                    city: "bollebygd".to_owned(),
-                }
-            },
-        )
-    }
-
-    fn subscribe(&self, _meta: Self::Metadata, subscriber: pubsub::Subscriber<String>) {
-        let id = self.next_subscription_id.fetch_add(1, atomic::Ordering::SeqCst);
-        let sub_id = SubscriptionId::Number(id as u64);
-        if let Ok(sink) = subscriber.assign_id(sub_id.clone()) {
-            debug!("Accepting new subscription with id {}", id);
-            self.active.write().unwrap().insert(sub_id, sink);
-        }
-    }
-
-    fn unsubscribe(&self, id: SubscriptionId) -> BoxFuture<bool, Error> {
-        debug!("Unsubscribing id {:?}", id);
-        if self.active.write().unwrap().remove(&id).is_some() {
-            future::ok(true).boxed()
-        } else {
-            future::err(
-                Error {
-                    code: ErrorCode::InvalidParams,
-                    message: "Invalid subscription.".into(),
-                    data: None,
-                },
-            )
-                    .boxed()
-        }
-    }
-}
diff --git a/mullvad-types/Cargo.toml b/mullvad-types/Cargo.toml
index 9f5e3a4202..cedea3d517 100644
--- a/mullvad-types/Cargo.toml
+++ b/mullvad-types/Cargo.toml
@@ -4,5 +4,6 @@ version = "0.1.0"
 authors = ["Linus Färnstrand <linus@mullvad.net>"]
 
 [dependencies]
+chrono = { version = "0.4", features = ["serde"] }
 serde_derive = "1.0"
 serde = "1.0"
diff --git a/mullvad-types/src/account.rs b/mullvad-types/src/account.rs
new file mode 100644
index 0000000000..5b335514f1
--- /dev/null
+++ b/mullvad-types/src/account.rs
@@ -0,0 +1,9 @@
+use chrono::DateTime;
+use chrono::offset::Utc;
+
+pub type AccountToken = String;
+
+#[derive(Debug, Clone, Eq, PartialEq, Hash, Serialize, Deserialize)]
+pub struct AccountData {
+    pub expiry: DateTime<Utc>,
+}
diff --git a/mullvad-types/src/lib.rs b/mullvad-types/src/lib.rs
index 602a43cb29..4c74f3814c 100644
--- a/mullvad-types/src/lib.rs
+++ b/mullvad-types/src/lib.rs
@@ -1,5 +1,8 @@
+extern crate chrono;
 #[macro_use]
 extern crate serde_derive;
 extern crate serde;
 
+pub mod account;
+pub mod location;
 pub mod states;
diff --git a/mullvad-types/src/location.rs b/mullvad-types/src/location.rs
new file mode 100644
index 0000000000..72146998b3
--- /dev/null
+++ b/mullvad-types/src/location.rs
@@ -0,0 +1,8 @@
+pub type CountryCode = String;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Location {
+    pub latlong: [f64; 2],
+    pub country: String,
+    pub city: String,
+}