moved message structs to its own dir

.gitignore

@@ -6,4 +6,5 @@ data/
 certs/*
 *.log
 echoer
-.env.private
+.env.private
+*.lock

Makefile

@@ -1,4 +1,4 @@
-.PHONY: _pre_build debug release run clean unit-tests integration-tests test
+.PHONY: _pre_build debug release run clean unit integration test
 
 CARGO=./scripts/cargo_musl.sh
 
@@ -17,10 +17,10 @@ release: _pre_build
 run: build
 	${CARGO} run
 
-unit-tests:
+unit:
 	${CARGO} test --lib
 
-integration-tests:
+integration:
 	cd ./integration && ./integration_tests.sh
 
-test: unit-tests integration-tests
+test: unit integration

bin/u_agent/src/lib.rs

@@ -2,8 +2,6 @@
 // поддержка питона
 // резолв адреса управляющего сервера через DoT
 // кроссплатформенность (реализовать интерфейс для винды и никсов)
-// проверка обнов
-// самоуничтожение
 
 #[macro_use]
 extern crate log;

certs/gen_certs.sh

@@ -1,17 +1,18 @@
 set -ex
 DIR=.
-V3_CFG=v3.ext
+V3_CFG=$DIR/v3.ext
 
-cat > $DIR/$V3_CFG << EOF
+cat > $V3_CFG << EOF
 authorityKeyIdentifier=keyid,issuer
 basicConstraints=CA:FALSE
 keyUsage = digitalSignature, nonRepudiation, keyEncipherment, dataEncipherment, keyAgreement, keyCertSign, cRLSign
 EOF
 
-
 openssl req -x509 -newkey rsa:4096 -keyout $DIR/ca.key -out $DIR/ca.crt -nodes -days 365 -subj "/CN=root"
 openssl req -newkey rsa:4096 -keyout $DIR/alice.key -out $DIR/alice.csr -nodes -days 365 -subj "/CN=alice"
 openssl req -newkey rsa:4096 -keyout $DIR/server.key -out $DIR/server.csr -nodes -days 365 -subj "/CN=u_server"
-openssl x509 -req -in $DIR/alice.csr -CA $DIR/ca.crt -CAkey $DIR/ca.key -out $DIR/alice.crt -set_serial 01 -days 365 -extfile $DIR/$V3_CFG
+openssl x509 -req -in $DIR/alice.csr -CA $DIR/ca.crt -CAkey $DIR/ca.key -out $DIR/alice.crt -set_serial 01 -days 365 -extfile $V3_CFG
-openssl x509 -req -in $DIR/server.csr -CA $DIR/ca.crt -CAkey $DIR/ca.key -out $DIR/server.crt -set_serial 01 -days 365 -extfile $DIR/$V3_CFG
+openssl x509 -req -in $DIR/server.csr -CA $DIR/ca.crt -CAkey $DIR/ca.key -out $DIR/server.crt -set_serial 01 -days 365 -extfile $V3_CFG
-openssl pkcs12 -export -out $DIR/alice.p12 -inkey $DIR/alice.key -in $DIR/alice.crt -passin pass: -passout pass:
+openssl pkcs12 -export -out $DIR/alice.p12 -inkey $DIR/alice.key -in $DIR/alice.crt -passin pass: -passout pass:
+
+rm $V3_CFG

integration/Cargo.toml

@@ -16,6 +16,7 @@ serde_json = "1.0"
 serde = { version = "1.0.114", features = ["derive"] }
 futures = "0.3.5"
 shlex = "1.0.0"
+rstest = "0.11"
 
 [dependencies.u_lib]
 path = "../lib/u_lib"
@@ -24,4 +25,4 @@ version = "*"
 
 [[test]]
 name = "integration"
-path = "tests/tests.rs"
+path = "tests/tests.rs"

integration/tests/behaviour.rs

@@ -1 +1,46 @@
+use crate::fixtures::agent::*;
+use crate::helpers::Panel;
 
+use rstest::rstest;
+use std::error::Error;
+use std::thread::sleep;
+use std::time::Duration;
+use u_lib::models::*;
+use uuid::Uuid;
+
+type TestResult<R = ()> = Result<R, Box<dyn Error>>;
+
+#[rstest]
+#[tokio::test]
+async fn test_registration(#[future] register_agent: RegisteredAgent) -> TestResult {
+    let agent = register_agent.await;
+    let agents: Vec<Agent> = Panel::check_output("agents list");
+    let found = agents.iter().find(|v| v.id == agent.uid);
+    assert!(found.is_some());
+    //teardown
+    Panel::check_status::<i32>(&format!("agents delete {}", agent.uid));
+    Ok(()) //TODO: ______^^^^^ REMOV
+}
+
+#[tokio::test]
+async fn test_setup_tasks() -> TestResult {
+    //some independent agents should present
+    let agents: Vec<Agent> = Panel::check_output("agents list");
+    let agent_uid = agents[0].id;
+    let job_alias = "passwd_contents";
+    let cmd = format!("jobs add --alias {} 'cat /etc/passwd'", job_alias);
+    Panel::check_status::<Empty>(&cmd);
+    let cmd = format!("jobmap add {} {}", agent_uid, job_alias);
+    let assigned_uids: Vec<Uuid> = Panel::check_output(cmd);
+    for _ in 0..3 {
+        let result: Vec<AssignedJob> =
+            Panel::check_output(format!("jobmap list {}", assigned_uids[0]));
+        if result[0].state == JobState::Finished {
+            return Ok(());
+        } else {
+            sleep(Duration::from_secs(5));
+            eprintln!("waiting for task");
+        }
+    }
+    panic!("Job didn't appear in the job map");
+}

integration/tests/fixtures/agent.rs

@@ -0,0 +1,36 @@
+use u_lib::{api::ClientHandler, models::*};
+use uuid::Uuid;
+
+pub struct RegisteredAgent {
+    pub uid: Uuid,
+}
+
+impl RegisteredAgent {
+    pub async fn unregister(self) {
+        let cli = ClientHandler::new(None);
+        cli.del(Some(self.uid)).await.unwrap();
+    }
+}
+
+#[fixture]
+pub async fn register_agent() -> RegisteredAgent {
+    let cli = ClientHandler::new(None);
+    let agent_uid = Uuid::new_v4();
+    let resp = cli
+        .get_personal_jobs(Some(agent_uid))
+        .await
+        .unwrap()
+        .pop()
+        .unwrap();
+    let job_id = resp.job_id;
+    let resp = cli.get_jobs(Some(job_id)).await.unwrap().pop().unwrap();
+    assert_eq!(resp.alias, Some("agent_hello".to_string()));
+    let agent_data = Agent {
+        id: agent_uid,
+        ..Default::default()
+    };
+    cli.report(&vec![ExecResult::Agent(agent_data)])
+        .await
+        .unwrap();
+    RegisteredAgent { uid: agent_uid }
+}

integration/tests/fixtures/mod.rs

@@ -0,0 +1 @@
+pub mod agent;

integration/tests/helpers/client.rs

@@ -1,48 +0,0 @@
-use reqwest::{Client, RequestBuilder, Url};
-use serde::Serialize;
-use serde_json::{from_str, json, Value};
-
-const SERVER: &str = "u_server";
-const PORT: &str = "63714";
-
-pub struct AgentClient {
-    client: Client,
-    base_url: Url,
-}
-
-impl AgentClient {
-    pub fn new() -> Self {
-        Self {
-            client: Client::new(),
-            base_url: Url::parse(&format!("http://{}:{}", SERVER, PORT)).unwrap(),
-        }
-    }
-
-    async fn process_request(&self, req: RequestBuilder, resp_needed: bool) -> Value {
-        let resp = req.send().await.unwrap();
-        if let Err(e) = resp.error_for_status_ref() {
-            panic!(
-                "Server responded with code {}\nError: {}",
-                e.status()
-                    .map(|s| s.to_string())
-                    .unwrap_or(String::from("<none>")),
-                e.to_string()
-            );
-        }
-        if !resp_needed {
-            return json!([]);
-        }
-        let resp: Value = from_str(&resp.text().await.unwrap()).unwrap();
-        resp.get("inner").unwrap().get(0).unwrap().clone()
-    }
-
-    pub async fn get<S: AsRef<str>>(&self, url: S) -> Value {
-        let req = self.client.get(self.base_url.join(url.as_ref()).unwrap());
-        self.process_request(req, true).await
-    }
-
-    pub async fn post<S: AsRef<str>, B: Serialize>(&self, url: S, body: &B) -> Value {
-        let req = self.client.post(self.base_url.join(url.as_ref()).unwrap());
-        self.process_request(req.json(body), false).await
-    }
-}

integration/tests/tests.rs

@@ -1,67 +1,6 @@
+mod behaviour;
+mod fixtures;
 mod helpers;
 
-use helpers::Panel;
+#[macro_use]
-
+extern crate rstest;
-use std::error::Error;
-use std::thread::sleep;
-use std::time::Duration;
-use u_lib::{api::ClientHandler, models::*};
-use uuid::Uuid;
-
-type TestResult<R = ()> = Result<R, Box<dyn Error>>;
-
-async fn register_agent() -> Uuid {
-    let cli = ClientHandler::new(None);
-    let agent_uid = Uuid::new_v4();
-    let resp = cli
-        .get_personal_jobs(Some(agent_uid))
-        .await
-        .unwrap()
-        .pop()
-        .unwrap();
-    let job_id = resp.job_id;
-    let resp = cli.get_jobs(Some(job_id)).await.unwrap().pop().unwrap();
-    assert_eq!(resp.alias, Some("agent_hello".to_string()));
-    let agent_data = Agent {
-        id: agent_uid,
-        ..Default::default()
-    };
-    cli.report(&vec![ExecResult::Agent(agent_data)])
-        .await
-        .unwrap();
-    agent_uid
-}
-
-#[tokio::test]
-async fn test_registration() -> TestResult {
-    let agent_uid = register_agent().await;
-    let agents: Vec<Agent> = Panel::check_output("agents list");
-    let found = agents.iter().find(|v| v.id == agent_uid);
-    assert!(found.is_some());
-    //teardown
-    Panel::check_status::<i32>(&format!("agents delete {}", agent_uid));
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_setup_tasks() -> TestResult {
-    //some independent agents should present
-    let agents: Vec<Agent> = Panel::check_output("agents list");
-    let agent_uid = agents[0].id;
-    let job_alias = "passwd_contents";
-    let cmd = format!("jobs add --alias {} 'cat /etc/passwd'", job_alias);
-    Panel::check_status::<Empty>(&cmd);
-    let cmd = format!("jobmap add {} {}", agent_uid, job_alias);
-    let assigned_uids: Vec<Uuid> = Panel::check_output(cmd);
-    for _ in 0..3 {
-        let result: Vec<AssignedJob> =
-            Panel::check_output(format!("jobmap list {}", assigned_uids[0]));
-        if result[0].state == JobState::Finished {
-            return Ok(());
-        } else {
-            sleep(Duration::from_secs(5));
-            eprintln!("waiting for task");
-        }
-    }
-    panic!()
-}

lib/u_lib/src/api.rs

@@ -1,4 +1,5 @@
-#[allow(non_upper_case_globals)]
+use crate::messaging;
+//#[allow(non_upper_case_globals)]
 use crate::{
     config::{MASTER_PORT, MASTER_SERVER},
     messaging::{AsMsg, BaseMessage},
@@ -61,34 +62,42 @@ impl ClientHandler {
     //
     // get jobs for client
     #[api_route("GET")]
-    fn get_personal_jobs(&self, url_param: Option<Uuid>) -> VecDisplay<models::AssignedJob> {}
+    async fn get_personal_jobs(&self, url_param: Option<Uuid>) -> VecDisplay<models::AssignedJob> {}
     //
     // send something to server
     #[api_route("POST")]
-    fn report<M: AsMsg>(&self, payload: &M) -> models::Empty {}
+    async fn report<M: AsMsg>(&self, payload: &M) -> messaging::Empty {}
+    //
+    // download file
+    #[api_route("GET")]
+    async fn dl(&self, url_param: Option<Uuid>) -> Vec<u8> {}
+    //
+    // request download
+    #[api_route("POST")]
+    async fn dlr(&self, url_param: Option<String>) -> messaging::DownloadInfo {}
 
     //##########// Admin area //##########//
     /// client listing
     #[api_route("GET")]
-    fn get_agents(&self, url_param: Option<Uuid>) -> VecDisplay<models::Agent> {}
+    async fn get_agents(&self, url_param: Option<Uuid>) -> VecDisplay<models::Agent> {}
     //
     // get all available jobs
     #[api_route("GET")]
-    fn get_jobs(&self, url_param: Option<Uuid>) -> VecDisplay<models::JobMeta> {}
+    async fn get_jobs(&self, url_param: Option<Uuid>) -> VecDisplay<models::JobMeta> {}
     //
     // create and upload job
     #[api_route("POST")]
-    fn upload_jobs(&self, payload: &[models::JobMeta]) -> models::Empty {}
+    async fn upload_jobs(&self, payload: &[models::JobMeta]) -> messaging::Empty {}
     //
     // delete something
     #[api_route("GET")]
-    fn del(&self, url_param: Option<Uuid>) -> i32 {}
+    async fn del(&self, url_param: Option<Uuid>) -> i32 {}
     //
     // set jobs for any client
     #[api_route("POST")]
-    fn set_jobs(&self, url_param: Option<Uuid>, payload: &[String]) -> VecDisplay<Uuid> {}
+    async fn set_jobs(&self, url_param: Option<Uuid>, payload: &[String]) -> VecDisplay<Uuid> {}
     //
     // get jobs for any client
     #[api_route("GET")]
-    fn get_agent_jobs(&self, url_param: Option<Uuid>) -> VecDisplay<models::AssignedJob> {}
+    async fn get_agent_jobs(&self, url_param: Option<Uuid>) -> VecDisplay<models::AssignedJob> {}
 }

lib/u_lib/src/builder.rs

@@ -1,15 +1,13 @@
 use crate::{
-    cache::JobCache,
+    UError, UResult, cache::JobCache, executor::{Waiter, DynFut}, 
-    executor::{Waiter, DynFut},
+    models::{Agent, AssignedJob, JobMeta, JobType, ExecResult}, 
-    models::{Agent, AssignedJob, JobMeta, JobType, ExecResult},
+    utils::{CombinedResult, OneOrVec}
-    utils::{CombinedResult, OneOrVec},
-    UError,
 };
 use guess_host_triple::guess_host_triple;
 use std::collections::HashMap;
 
 pub struct JobBuilder {
-    jobs: Waiter,
+    waiter: Waiter,
 }
 
 impl JobBuilder {
@@ -25,11 +23,12 @@ impl JobBuilder {
             }
             let job_meta = job_meta.unwrap();
             //waiting for try-blocks stabilization
-            let built_req = (|| {
+            let built_req = (|| -> UResult<()> {
                 Ok(match job_meta.exec_type {
                     JobType::Shell => {
                         let meta = JobCache::get(&req.job_id).ok_or(UError::NoJob(req.job_id))?;
                         let curr_platform = guess_host_triple().unwrap_or("unknown").to_string();
+                        //extend platform checking (partial check)
                         if meta.platform != curr_platform {
                             return Err(UError::InsuitablePlatform(
                                 meta.platform.clone(),
@@ -48,7 +47,7 @@ impl JobBuilder {
             }
         }
         result.ok(Self {
-            jobs: Waiter::new(prepared),
+            waiter: Waiter::new(prepared),
         });
         result
     }
@@ -68,18 +67,18 @@ impl JobBuilder {
 
     /// Spawn jobs and pop results later
     pub async fn spawn(mut self) -> Self {
-        self.jobs = self.jobs.spawn().await;
+        self.waiter = self.waiter.spawn().await;
         self
     }
 
     /// Spawn jobs and wait for result
     pub async fn wait(self) -> Vec<ExecResult> {
-        self.jobs.spawn().await.wait().await
+        self.waiter.spawn().await.wait().await
     }
 
     /// Spawn one job and wait for result
     pub async fn wait_one(self) -> ExecResult {
-        self.jobs.spawn().await.wait().await.pop().unwrap()
+        self.waiter.spawn().await.wait().await.pop().unwrap()
     }
 }
 

lib/u_lib/src/messaging/files.rs

@@ -0,0 +1,8 @@
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Serialize, Deserialize, Clone, Debug, Default)]
+pub struct DownloadInfo {
+    hashsum: String,
+    dl_fid: Uuid,
+}

lib/u_lib/src/messaging/mod.rs

@@ -0,0 +1,29 @@
+mod base;
+mod files;
+
+use crate::models::*;
+pub use base::{AsMsg, BaseMessage};
+pub use files::*;
+use serde::{Deserialize, Serialize};
+use std::fmt;
+use uuid::Uuid;
+
+impl AsMsg for Agent {}
+impl AsMsg for AssignedJob {}
+impl AsMsg for DownloadInfo {}
+impl AsMsg for ExecResult {}
+impl AsMsg for JobMeta {}
+impl AsMsg for String {}
+impl AsMsg for Uuid {}
+impl AsMsg for Empty {}
+impl AsMsg for i32 {}
+impl AsMsg for u8 {}
+
+#[derive(Serialize, Deserialize, Clone, Default, Debug)]
+pub struct Empty;
+
+impl fmt::Display for Empty {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "<empty>")
+    }
+}

lib/u_lib/src/models/jobs/meta.rs

@@ -41,12 +41,12 @@ impl fmt::Display for JobMeta {
         out += &format!("\nPlatform: {}", self.platform);
         if self.exec_type == JobType::Shell && self.payload.is_some() {
             let payload = self.payload.as_ref().unwrap();
-            let pld_len = {
+            let (pld_len, large) = {
                 let pl = payload.len();
                 if pl > 20 {
-                    20
+                    (20, true)
                 } else {
-                    pl
+                    (pl, false)
                 }
             };
             let pld_beginning = payload
@@ -57,7 +57,7 @@ impl fmt::Display for JobMeta {
             out += &format!(
                 "\nPayload: {}{}",
                 String::from_utf8_lossy(&pld_beginning),
-                if pld_len <= 20 { "" } else { " <...>" }
+                if large { "" } else { " <...>" }
             );
         }
         write!(f, "{}", out)
@@ -69,7 +69,7 @@ impl Default for JobMeta {
         Self {
             id: Uuid::new_v4(),
             alias: None,
-            argv: String::from("/bin/bash -c {}"),
+            argv: String::new(),
             exec_type: JobType::Shell,
             platform: guess_host_triple().unwrap_or("unknown").to_string(),
             payload: None,
@@ -111,9 +111,12 @@ impl JobMetaBuilder {
     }
 
     pub fn build(self) -> UResult<JobMeta> {
-        let inner = self.inner;
+        let mut inner = self.inner;
         match inner.exec_type {
             JobType::Shell => {
+                if inner.argv == "" {
+                    inner.argv = String::from("/bin/bash -c {}")
+                }
                 let argv_parts =
                     shlex::split(&inner.argv).ok_or(UError::JobArgsError("Shlex failed".into()))?;
                 let empty_err = UError::JobArgsError("Empty argv".into());

lib/u_lib/src/models/mod.rs

@@ -1,29 +1,13 @@
 mod agent;
 pub mod jobs;
-mod result;
 pub mod schema;
 
-use crate::messaging::AsMsg;
-pub use crate::models::result::ExecResult;
 pub use crate::models::{agent::*, jobs::*};
 use serde::{Deserialize, Serialize};
-use std::fmt;
-use uuid::Uuid;
 
-impl AsMsg for Agent {}
+#[derive(Serialize, Deserialize, Clone, PartialEq)]
-impl AsMsg for AssignedJob {}
+pub enum ExecResult {
-impl AsMsg for ExecResult {}
+    Assigned(AssignedJob),
-impl AsMsg for JobMeta {}
+    Agent(Agent),
-impl AsMsg for String {}
+    Dummy,
-impl AsMsg for Uuid {}
-impl AsMsg for Empty {}
-impl AsMsg for i32 {}
-
-#[derive(Serialize, Deserialize, Clone, Default, Debug)]
-pub struct Empty;
-
-impl fmt::Display for Empty {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "<empty>")
-    }
 }

lib/u_lib/src/models/result.rs

@@ -1,9 +0,0 @@
-use crate::models::{Agent, AssignedJob};
-use serde::{Deserialize, Serialize};
-
-#[derive(Serialize, Deserialize, Clone, PartialEq)]
-pub enum ExecResult {
-    Assigned(AssignedJob),
-    Agent(Agent),
-    Dummy,
-}

lib/u_lib/src/utils/hexlify.rs

@@ -0,0 +1,12 @@
+use std::fmt;
+
+pub struct Hexlify<'b>(pub &'b [u8]);
+
+impl<'a> fmt::LowerHex for Hexlify<'a> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        for byte in self.0.iter() {
+            write!(f, "{:02x}", byte)?;
+        }
+        Ok(())
+    }
+}

lib/u_lib/src/utils/mod.rs

@@ -1,11 +1,15 @@
-pub mod combined_result;
+mod combined_result;
-pub mod conv;
+mod conv;
-pub mod misc;
+mod hexlify;
-pub mod tempfile;
+mod misc;
-pub mod vec_display;
+mod storage;
+mod tempfile;
+mod vec_display;
 
 pub use combined_result::*;
 pub use conv::*;
+pub use hexlify::*;
 pub use misc::*;
+pub use storage::*;
 pub use tempfile::*;
 pub use vec_display::*;

lib/u_lib/src/utils/storage.rs

@@ -0,0 +1,39 @@
+use once_cell::sync::Lazy;
+use std::cmp::Eq;
+use std::collections::HashMap;
+use std::hash::Hash;
+use std::ops::Deref;
+use std::sync::Arc;
+use std::sync::{Mutex, MutexGuard};
+
+//improve this later, replace job cacher with it
+//possibly add different backends (memory, disk)
+pub struct SharedStorage<Key, Val>(Arc<Mutex<HashMap<Key, Val>>>);
+
+impl<Key: Eq + Hash, Val> SharedStorage<Key, Val> {
+    pub fn new() -> Lazy<SharedStorage<Key, Val>> {
+        Lazy::new(|| SharedStorage(Arc::new(Mutex::new(HashMap::new()))))
+    }
+
+    pub fn lock(&self) -> MutexGuard<'_, HashMap<Key, Val>> {
+        self.0.lock().unwrap()
+    }
+
+    pub fn get<'get, 'slf: 'get>(&'slf self, key: &'get Key) -> Option<RefHolder<'get, Key, Val>> {
+        if !self.lock().contains_key(key) {
+            return None;
+        }
+        let lock = self.lock();
+        Some(RefHolder(lock, key))
+    }
+}
+
+pub struct RefHolder<'h, Key, Val>(pub MutexGuard<'h, HashMap<Key, Val>>, pub &'h Key);
+
+impl<'h, Key: Eq + Hash, Val> Deref for RefHolder<'h, Key, Val> {
+    type Target = Val;
+
+    fn deref(&self) -> &Self::Target {
+        self.0.get(self.1).unwrap()
+    }
+}