bin/tau: add workspace feature

Cargo.lock

@@ -4124,12 +4124,14 @@ dependencies = [
  "async-executor",
  "async-std",
  "async-trait",
+ "bs58",
  "chrono",
  "crypto_box",
  "ctrlc-async",
  "darkfi",
  "easy-parallel",
  "futures",
+ "fxhash",
  "hex",
  "log",
  "notify",
@@ -4141,6 +4143,7 @@ dependencies = [
  "structopt",
  "structopt-toml",
  "thiserror",
+ "toml",
  "url",
 ]
 

bin/tau/tau-cli/src/main.rs

@@ -70,6 +70,12 @@ enum TauSubcommand {
 
     /// Get task info by ID
     Info { task_id: u64 },
+
+    /// Switch workspace
+    Switch {
+        /// Tau workspace
+        workspace: String,
+    },
 }
 
 pub struct Tau {
@@ -142,6 +148,11 @@ async fn main() -> Result<()> {
                 let task = tau.get_task_by_id(task_id).await?;
                 print_task_info(task)
             }
+
+            TauSubcommand::Switch { workspace } => {
+                tau.switch_ws(workspace).await?;
+                Ok(())
+            }
         },
         None => {
             let task_ids = tau.get_ids().await?;

bin/tau/tau-cli/src/rpc.rs

@@ -69,4 +69,14 @@ impl Tau {
 
         Ok(serde_json::from_value(rep)?)
     }
+
+    /// Switch workspace.
+    pub async fn switch_ws(&self, workspace: String) -> Result<()> {
+        let req = JsonRequest::new("switch_ws", json!([workspace]));
+        let rep = self.rpc_client.request(req).await?;
+
+        debug!("Got reply: {:?}", rep);
+
+        Ok(())
+    }
 }

bin/tau/taud/Cargo.toml

@@ -30,6 +30,7 @@ chrono = "0.4.19"
 thiserror = "1.0.31"
 ctrlc-async = {version= "3.2.2", default-features = false, features = ["async-std", "termination"]}
 url = "2.2.2"
+fxhash = "0.2.1"
 
 # Encoding and parsing
 serde = {version = "1.0.138", features = ["derive"]}
@@ -39,3 +40,5 @@ structopt-toml = "0.5.0"
 crypto_box = {version = "0.7.2", features = ["std"]}
 hex = "0.4.3"
 notify = "4.0.17"
+bs58 = "0.4.0"
+toml = "0.5.9"

bin/tau/taud/src/jsonrpc.rs

@@ -1,7 +1,9 @@
+use async_std::sync::{Arc, Mutex};
 use std::path::PathBuf;
 
 use async_trait::async_trait;
-use log::debug;
+use fxhash::FxHashMap;
+use log::{debug, warn};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
 
@@ -17,11 +19,14 @@ use crate::{
     error::{to_json_result, TaudError, TaudResult},
     month_tasks::MonthTasks,
     task_info::{Comment, TaskInfo},
+    util::Workspace,
 };
 
 pub struct JsonRpcInterface {
     dataset_path: PathBuf,
     nickname: String,
+    workspace: Arc<Mutex<String>>,
+    configured_ws: FxHashMap<String, Workspace>,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -50,6 +55,7 @@ impl RequestHandler for JsonRpcInterface {
             Some("set_state") => self.set_state(params).await,
             Some("set_comment") => self.set_comment(params).await,
             Some("get_task_by_id") => self.get_task_by_id(params).await,
+            Some("switch_ws") => self.switch_ws(params).await,
             Some(_) | None => return JsonError::new(ErrorCode::MethodNotFound, None, req.id).into(),
         };
 
@@ -58,8 +64,13 @@ impl RequestHandler for JsonRpcInterface {
 }
 
 impl JsonRpcInterface {
-    pub fn new(dataset_path: PathBuf, nickname: String) -> Self {
-        Self { dataset_path, nickname }
+    pub fn new(
+        dataset_path: PathBuf,
+        nickname: String,
+        workspace: Arc<Mutex<String>>,
+        configured_ws: FxHashMap<String, Workspace>,
+    ) -> Self {
+        Self { dataset_path, nickname, workspace, configured_ws }
     }
 
     // RPCAPI:
@@ -81,7 +92,9 @@ impl JsonRpcInterface {
         debug!(target: "tau", "JsonRpc::add() params {:?}", params);
 
         let task: BaseTaskInfo = serde_json::from_value(params[0].clone())?;
+        let ws = self.workspace.lock().await.clone();
         let mut new_task: TaskInfo = TaskInfo::new(
+            ws,
             &task.title,
             &task.desc,
             &self.nickname,
@@ -102,7 +115,8 @@ impl JsonRpcInterface {
     // <-- {"jsonrpc": "2.0", "result": [task_id, ...], "id": 1}
     async fn get_ids(&self, params: &[Value]) -> TaudResult<Value> {
         debug!(target: "tau", "JsonRpc::get_ids() params {:?}", params);
-        let tasks = MonthTasks::load_current_open_tasks(&self.dataset_path)?;
+        let ws = self.workspace.lock().await.clone();
+        let tasks = MonthTasks::load_current_open_tasks(&self.dataset_path, ws)?;
         let task_ids: Vec<u32> = tasks.iter().map(|task| task.get_id()).collect();
         Ok(json!(task_ids))
     }
@@ -117,8 +131,9 @@ impl JsonRpcInterface {
         if params.len() != 2 {
             return Err(TaudError::InvalidData("len of params should be 2".into()))
         }
+        let ws = self.workspace.lock().await.clone();
 
-        let task = self.check_params_for_update(&params[0], &params[1])?;
+        let task = self.check_params_for_update(&params[0], &params[1], ws)?;
         task.save(&self.dataset_path)?;
         Ok(json!(true))
     }
@@ -138,8 +153,9 @@ impl JsonRpcInterface {
         }
 
         let state: String = serde_json::from_value(params[1].clone())?;
+        let ws = self.workspace.lock().await.clone();
 
-        let mut task: TaskInfo = self.load_task_by_id(&params[0])?;
+        let mut task: TaskInfo = self.load_task_by_id(&params[0], ws)?;
 
         if states.contains(&state.as_str()) {
             task.set_state(&state);
@@ -162,8 +178,9 @@ impl JsonRpcInterface {
         }
 
         let comment_content: String = serde_json::from_value(params[1].clone())?;
+        let ws = self.workspace.lock().await.clone();
 
-        let mut task: TaskInfo = self.load_task_by_id(&params[0])?;
+        let mut task: TaskInfo = self.load_task_by_id(&params[0], ws)?;
         task.set_comment(Comment::new(&comment_content, &self.nickname));
 
         task.save(&self.dataset_path)?;
@@ -181,23 +198,55 @@ impl JsonRpcInterface {
         if params.len() != 1 {
             return Err(TaudError::InvalidData("len of params should be 1".into()))
         }
+        let ws = self.workspace.lock().await.clone();
 
-        let task: TaskInfo = self.load_task_by_id(&params[0])?;
+        let task: TaskInfo = self.load_task_by_id(&params[0], ws)?;
 
         Ok(json!(task))
     }
 
-    fn load_task_by_id(&self, task_id: &Value) -> TaudResult<TaskInfo> {
-        let task_id: u64 = serde_json::from_value(task_id.clone())?;
+    // RPCAPI:
+    // Switch tasks workspace.
+    // --> {"jsonrpc": "2.0", "method": "switch_ws", "params": [workspace], "id": 1}
+    // <-- {"jsonrpc": "2.0", "result": "true", "id": 1}
+    async fn switch_ws(&self, params: &[Value]) -> TaudResult<Value> {
+        debug!(target: "tau", "JsonRpc::switch_ws() params {:?}", params);
+
+        if params.len() != 1 {
+            return Err(TaudError::InvalidData("len of params should be 1".into()))
+        }
+
+        if !params[0].is_string() {
+            return Err(TaudError::InvalidData("Invalid workspace".into()))
+        }
 
-        let tasks = MonthTasks::load_current_open_tasks(&self.dataset_path)?;
+        let ws = params[0].as_str().unwrap().to_string();
+        let mut s = self.workspace.lock().await;
+
+        if self.configured_ws.contains_key(&ws) {
+            *s = ws
+        } else {
+            warn!("Workspace \"{}\" is not configured", ws);
+        }
+
+        Ok(json!(true))
+    }
+
+    fn load_task_by_id(&self, task_id: &Value, ws: String) -> TaudResult<TaskInfo> {
+        let task_id: u64 = serde_json::from_value(task_id.clone())?;
+        let tasks = MonthTasks::load_current_open_tasks(&self.dataset_path, ws)?;
         let task = tasks.into_iter().find(|t| (t.get_id() as u64) == task_id);
 
         task.ok_or(TaudError::InvalidId)
     }
 
-    fn check_params_for_update(&self, task_id: &Value, fields: &Value) -> TaudResult<TaskInfo> {
-        let mut task: TaskInfo = self.load_task_by_id(task_id)?;
+    fn check_params_for_update(
+        &self,
+        task_id: &Value,
+        fields: &Value,
+        ws: String,
+    ) -> TaudResult<TaskInfo> {
+        let mut task: TaskInfo = self.load_task_by_id(task_id, ws)?;
 
         if !fields.is_object() {
             return Err(TaudError::InvalidData("Invalid task's data".into()))

bin/tau/taud/src/main.rs

@@ -2,8 +2,9 @@ use async_std::sync::{Arc, Mutex};
 use std::{env, fs::create_dir_all, sync::mpsc, time::Duration};
 
 use async_executor::Executor;
-use crypto_box::{aead::Aead, Box, SecretKey, KEY_SIZE};
+use crypto_box::{aead::Aead, Box, SecretKey};
 use futures::{select, FutureExt};
+use fxhash::FxHashMap;
 use log::{debug, error, info, warn};
 use notify::{DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher};
 use smol::future;
@@ -34,39 +35,37 @@ use crate::{
     jsonrpc::JsonRpcInterface,
     settings::{Args, CONFIG_FILE, CONFIG_FILE_CONTENTS},
     task_info::TaskInfo,
-    util::{load, save},
+    util::{parse_workspaces, Workspace},
 };
 
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
 pub struct EncryptedTask {
+    workspace: String,
     nonce: Vec<u8>,
     payload: Vec<u8>,
 }
 
 fn encrypt_task(
     task: &TaskInfo,
-    secret_key: &SecretKey,
+    workspace: &String,
+    salsa_box: &Box,
     rng: &mut crypto_box::rand_core::OsRng,
 ) -> TaudResult<EncryptedTask> {
     debug!("start encrypting task");
-    let public_key = secret_key.public_key();
-    let msg_box = Box::new(&public_key, secret_key);
 
     let nonce = crypto_box::generate_nonce(rng);
     let payload = &serialize(task)[..];
-    let payload = msg_box.encrypt(&nonce, payload)?;
+    let payload = salsa_box.encrypt(&nonce, payload)?;
 
     let nonce = nonce.to_vec();
-    Ok(EncryptedTask { nonce, payload })
+    Ok(EncryptedTask { workspace: workspace.to_string(), nonce, payload })
 }
 
-fn decrypt_task(encrypt_task: &EncryptedTask, secret_key: &SecretKey) -> TaudResult<TaskInfo> {
+fn decrypt_task(encrypt_task: &EncryptedTask, salsa_box: &Box) -> TaudResult<TaskInfo> {
     debug!("start decrypting task");
-    let public_key = secret_key.public_key();
-    let msg_box = Box::new(&public_key, secret_key);
 
     let nonce = encrypt_task.nonce.as_slice();
-    let decrypted_task = msg_box.decrypt(nonce.into(), &encrypt_task.payload[..])?;
+    let decrypted_task = salsa_box.decrypt(nonce.into(), &encrypt_task.payload[..])?;
 
     let task = deserialize(&decrypted_task)?;
 
@@ -91,7 +90,7 @@ async fn start_sync_loop(
     raft_msgs_sender: async_channel::Sender<EncryptedTask>,
     commits_recv: async_channel::Receiver<EncryptedTask>,
     datastore_path: std::path::PathBuf,
-    secret_key: SecretKey,
+    configured_ws: FxHashMap<String, Workspace>,
     mut rng: crypto_box::rand_core::OsRng,
 ) -> TaudResult<()> {
     loop {
@@ -99,24 +98,33 @@ async fn start_sync_loop(
             task = broadcast_rcv.recv().fuse() => {
                 let tk = task.map_err(Error::from)?;
                 info!(target: "tau", "Save the task: ref: {}", tk.ref_id);
-                let encrypted_task = encrypt_task(&tk, &secret_key,&mut rng)?;
-                raft_msgs_sender.send(encrypted_task).await.map_err(Error::from)?;
+                if configured_ws.contains_key(&tk.workspace) {
+                    let ws_info = configured_ws.get(&tk.workspace).unwrap();
+                    if let Some(salsa_box) = &ws_info.encryption {
+                        let encrypted_task = encrypt_task(&tk, &tk.workspace, salsa_box, &mut rng)?;
+                        raft_msgs_sender.send(encrypted_task).await.map_err(Error::from)?;
+                    }
+                }
             }
             task = commits_recv.recv().fuse() => {
                 let recv = task.map_err(Error::from)?;
-                let task = decrypt_task(&recv, &secret_key);
-
-                if let Err(e) = task {
-                    warn!("unable to decrypt the task: {}", e);
-                    continue
+                if configured_ws.contains_key(&recv.workspace) {
+                    let ws_info = configured_ws.get(&recv.workspace).unwrap();
+                    if let Some(salsa_box) = &ws_info.encryption {
+                        let task = decrypt_task(&recv, salsa_box);
+                        if let Err(e) = task {
+                            warn!("unable to decrypt the task: {}", e);
+                            continue
+                        }
+
+                        let task = task.unwrap();
+                        if !commits_received.lock().await.contains(&task.ref_id) {
+                            commits_received.lock().await.push(task.ref_id.clone());
+                        }
+                        info!(target: "tau", "Update the task: ref: {}", task.ref_id);
+                        task.save(&datastore_path)?;
+                    }
                 }
-
-                let task = task.unwrap();
-                if !commits_received.lock().await.contains(&task.ref_id) {
-                    commits_received.lock().await.push(task.ref_id.clone());
-                }
-                info!(target: "tau", "Update the task: ref: {}", task.ref_id);
-                task.save(&datastore_path)?;
             }
         }
     }
@@ -192,44 +200,43 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         return Ok(())
     }
 
-    // mkdir datastore_path if not exists
-    create_dir_all(datastore_path.join("month"))?;
-    create_dir_all(datastore_path.join("task"))?;
-
     let mut rng = crypto_box::rand_core::OsRng;
 
-    let secret_key = if settings.key_gen {
+    if settings.key_gen {
         info!(target: "tau", "Generating a new secret key");
-        let secret = SecretKey::generate(&mut rng);
-        let sk_string = hex::encode(secret.as_bytes());
-        save::<String>(&datastore_path.join("secret_key"), &sk_string)?;
-        secret
-    } else {
-        let loaded_key = load::<String>(&datastore_path.join("secret_key"));
-
-        if loaded_key.is_err() {
-            error!(
-                "Could not load secret key from file, \
-                 Please run \"taud --help\" for more information"
-            );
-            return Ok(())
-        }
+        let secret_key = SecretKey::generate(&mut rng);
+        let encoded = bs58::encode(secret_key.as_bytes());
+        println!("Secret key: {}", encoded.into_string());
+        return Ok(())
+    }
+
+    // Pick up workspace settings from the TOML configuration
+    let cfg_path = get_config_path(settings.config, CONFIG_FILE)?;
+    let configured_ws = parse_workspaces(&cfg_path)?;
+
+    // mkdir datastore_path if not exists
+    create_dir_all(datastore_path.join("month"))?;
+    create_dir_all(datastore_path.join("task"))?;
 
-        let sk_bytes = hex::decode(loaded_key.unwrap())?;
-        let sk_bytes: [u8; KEY_SIZE] = sk_bytes.as_slice().try_into()?;
-        SecretKey::try_from(sk_bytes)?
-    };
+    // start at the first configured workspace
+    let key = configured_ws.keys().next().ok_or(Error::ConfigInvalid)?;
+    let workspace = Arc::new(Mutex::new(key.to_owned()));
 
     let (broadcast_snd, broadcast_rcv) = async_channel::unbounded::<TaskInfo>();
 
     //
     // RPC
     //
-    let rpc_interface = Arc::new(JsonRpcInterface::new(datastore_path.clone(), nickname.unwrap()));
+    let rpc_interface = Arc::new(JsonRpcInterface::new(
+        datastore_path.clone(),
+        nickname.unwrap(),
+        workspace,
+        configured_ws.clone(),
+    ));
     executor.spawn(listen_and_serve(settings.rpc_listen.clone(), rpc_interface)).detach();
 
     //
-    //Raft
+    // Raft
     //
     let net_settings = settings.net;
     let seen_net_msgs = Arc::new(Mutex::new(vec![]));
@@ -250,7 +257,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
             raft.get_msgs_channel(),
             raft.get_commits_channel(),
             datastore_path.clone(),
-            secret_key,
+            configured_ws,
             rng,
         ))
         .detach();

bin/tau/taud/src/month_tasks.rs

@@ -120,9 +120,14 @@ impl MonthTasks {
         }
     }
 
-    pub fn load_current_open_tasks(dataset_path: &Path) -> TaudResult<Vec<TaskInfo>> {
+    pub fn load_current_open_tasks(dataset_path: &Path, ws: String) -> TaudResult<Vec<TaskInfo>> {
         let mt = Self::load_or_create(None, dataset_path)?;
-        Ok(mt.objects(dataset_path)?.into_iter().filter(|t| t.get_state() != "stop").collect())
+        Ok(mt
+            .objects(dataset_path)?
+            .into_iter()
+            .filter(|t| t.get_state() != "stop")
+            .filter(|t| t.workspace == ws)
+            .collect())
     }
 }
 
@@ -156,8 +161,15 @@ mod tests {
         // load and save TaskInfo
         ///////////////////////
 
-        let mut task =
-            TaskInfo::new("test_title", "test_desc", "NICKNAME", None, 0.0, &dataset_path)?;
+        let mut task = TaskInfo::new(
+            "darkfi".to_string(),
+            "test_title",
+            "test_desc",
+            "NICKNAME",
+            None,
+            0.0,
+            &dataset_path,
+        )?;
 
         task.save(&dataset_path)?;
 
@@ -197,8 +209,15 @@ mod tests {
         // activate task
         ///////////////////////
 
-        let task =
-            TaskInfo::new("test_title_3", "test_desc", "NICKNAME", None, 0.0, &dataset_path)?;
+        let task = TaskInfo::new(
+            "darkfi".to_string(),
+            "test_title_3",
+            "test_desc",
+            "NICKNAME",
+            None,
+            0.0,
+            &dataset_path,
+        )?;
 
         task.save(&dataset_path)?;
 

bin/tau/taud/src/task_info.rs

@@ -58,6 +58,7 @@ pub struct TaskAssigns(Vec<String>);
 #[derive(Clone, Debug, Serialize, Deserialize, SerialEncodable, SerialDecodable, PartialEq)]
 pub struct TaskInfo {
     pub(crate) ref_id: String,
+    pub(crate) workspace: String,
     id: u32,
     title: String,
     desc: String,
@@ -73,6 +74,7 @@ pub struct TaskInfo {
 
 impl TaskInfo {
     pub fn new(
+        workspace: String,
         title: &str,
         desc: &str,
         owner: &str,
@@ -86,7 +88,10 @@ impl TaskInfo {
         let created_at = Timestamp::current_time();
 
         let task_ids: Vec<u32> =
-            MonthTasks::load_current_open_tasks(dataset_path)?.into_iter().map(|t| t.id).collect();
+            MonthTasks::load_current_open_tasks(dataset_path, workspace.clone())?
+                .into_iter()
+                .map(|t| t.id)
+                .collect();
 
         let id: u32 = find_free_id(&task_ids);
 
@@ -98,6 +103,7 @@ impl TaskInfo {
 
         Ok(Self {
             ref_id,
+            workspace,
             id,
             title: title.into(),
             desc: desc.into(),

bin/tau/taud/src/util.rs

@@ -1,10 +1,58 @@
-use std::{fs::File, io::BufReader, path::Path};
-
+use std::{
+    fs::File,
+    io::BufReader,
+    path::{Path, PathBuf},
+};
+
+use fxhash::FxHashMap;
+use log::info;
 use rand::{distributions::Alphanumeric, thread_rng, Rng};
 use serde::{de::DeserializeOwned, Serialize};
 
 use darkfi::Result;
 
+#[derive(Clone)]
+pub struct Workspace {
+    pub encryption: Option<crypto_box::Box>,
+}
+
+impl Workspace {
+    pub fn new() -> Result<Self> {
+        Ok(Self { encryption: None })
+    }
+}
+
+/// Parse the configuration file for any configured workspaces and return
+/// a map containing said configurations.
+pub fn parse_workspaces(config_file: &PathBuf) -> Result<FxHashMap<String, Workspace>> {
+    let toml_contents = std::fs::read_to_string(config_file)?;
+    let mut ret = FxHashMap::default();
+
+    if let toml::Value::Table(map) = toml::from_str(&toml_contents)? {
+        if map.contains_key("workspace") && map["workspace"].is_table() {
+            for ws in map["workspace"].as_table().unwrap() {
+                info!("Found configuration for workspace {}", ws.0);
+                let mut workspace_info = Workspace::new()?;
+
+                if ws.1.as_table().unwrap().contains_key("secret") {
+                    // Build the NaCl box
+                    let s = ws.1["secret"].as_str().unwrap();
+                    let bytes: [u8; 32] = bs58::decode(s).into_vec()?.try_into().unwrap();
+                    let secret = crypto_box::SecretKey::from(bytes);
+                    let public = secret.public_key();
+                    let msg_box = crypto_box::Box::new(&public, &secret);
+                    workspace_info.encryption = Some(msg_box);
+                    info!("Instantiated NaCl box for workspace {}", ws.0);
+                }
+
+                ret.insert(ws.0.to_string(), workspace_info);
+            }
+        }
+    };
+
+    Ok(ret)
+}
+
 pub fn random_ref_id() -> String {
     thread_rng().sample_iter(&Alphanumeric).take(30).map(char::from).collect()
 }

bin/tau/taud_config.toml

@@ -30,3 +30,11 @@
 #connect_timeout_seconds=10
 #channel_handshake_seconds=4
 #channel_heartbeat_seconds=10
+
+## Per-workspace settings
+[workspace."darkfi"]
+## Create with `taud --key-gen`
+secret = "7CkVuFgwTUpJn5Sv67Q3fyEDpa28yrSeL5Hg2GqQ4jfM"
+[workspace."general"]
+## Create with `taud --key-gen`
+secret = "6ZkvojUcSRML7wSGc3AMmM5meyyEXLoykT23cyUUB6GM"