partially implemented some structs, but not the behaviour

8 changed files with 221 additions and 8 deletions
--- a/bin/u_server/Cargo.toml
+++ b/bin/u_server/Cargo.toml
@ -15,6 +15,7 @@ hyper = "0.14"
 mockall = "0.9.1"
 mockall_double = "0.2"
 openssl = "*"
+sha2 = "0.9.5"

 [dependencies.diesel]
 features = ["postgres", "uuid"]
@ -41,4 +42,4 @@ path = "src/u_server.rs"

 [[bin]]
 name = "u_server"
-path = "src/main.rs"
+path = "src/main.rs"
--- a/bin/u_server/src/filters.rs
+++ b/bin/u_server/src/filters.rs
@ -1,5 +1,6 @@
 use crate::handlers::Endpoints;
 use serde::de::DeserializeOwned;
+use std::convert::Infallible;
 use std::env;
 use u_lib::{
    messaging::{AsMsg, BaseMessage},
@ -16,7 +17,7 @@ where
 }

 pub fn make_filters() -> impl Filter<Extract = (impl Reply,), Error = Rejection> + Clone {
-    let infallible_none = |_| async { Ok::<(Option<Uuid>,), std::convert::Infallible>((None,)) };
+    let infallible_none = |_| async { Ok::<(Option<Uuid>,), Infallible>((None,)) };

    let get_agents = warp::get()
        .and(warp::path("get_agents"))
@ -68,7 +69,18 @@ pub fn make_filters() -> impl Filter<Extract = (impl Reply,), Error = Rejection>

    let report = warp::post()
        .and(warp::path("report"))
-        .and(get_content::<Vec<ExecResult>>().and_then(Endpoints::report));
+        .and(get_content::<Vec<ExecResult>>())
+        .and_then(Endpoints::report);
+
+    let download = warp::get()
+        .and(warp::path("dl"))
+        .and(warp::path::param::<Uuid>())
+        .and_then(Endpoints::download);
+
+    let download_request = warp::post()
+        .and(warp::path("dlr"))
+        .and(get_content::<String>())
+        .and_then(Endpoints::download_request);

    let auth_token = format!("Bearer {}", env::var("ADMIN_AUTH_TOKEN").unwrap()).into_boxed_str();
    let auth_header = warp::header::exact("authorization", Box::leak(auth_token));
@ -81,7 +93,11 @@ pub fn make_filters() -> impl Filter<Extract = (impl Reply,), Error = Rejection>
        .or(get_agent_jobs))
    .and(auth_header);

-    let agent_zone = get_jobs.clone().or(get_personal_jobs).or(report);
+    let agent_zone = get_jobs
+        .or(get_personal_jobs)
+        .or(report)
+        .or(download)
+        .or(download_request);

    auth_zone.or(agent_zone)
 }
--- a/bin/u_server/src/handlers.rs
+++ b/bin/u_server/src/handlers.rs
@ -1,9 +1,12 @@
-use crate::db::UDB;
+use crate::{
+    db::UDB,
+    storages::{FileIndex, Mapper},
+};
 use diesel::SaveChangesDsl;
 use hyper::Body;
 use serde::Serialize;
 use u_lib::{
-    messaging::{AsMsg, BaseMessage},
+    messaging::{AsMsg, BaseMessage, DownloadInfo},
    models::{Agent, AgentState, AssignedJob, ExecResult, JobMeta, JobState},
    ULocalError,
 };
@ -25,6 +28,10 @@ pub fn build_err<S: ToString>(body: S) -> Response<Body> {
    build_response(StatusCode::BAD_REQUEST, body.to_string())
 }

+pub fn build_not_found() -> Response<Body> {
+    build_response(StatusCode::NOT_FOUND, "")
+}
+
 pub fn build_message<M: AsMsg + Serialize>(m: M) -> Response<Body> {
    warp::reply::json(&m.as_message()).into_response()
 }
@ -168,4 +175,23 @@ impl Endpoints {
        }
        Ok(build_ok(""))
    }
+
+    pub async fn download_request(
+        msg: BaseMessage<'static, String>,
+    ) -> Result<Response<Body>, Rejection> {
+        let agent_id = msg.id;
+        FileIndex::
+    }
+
+    pub async fn download(uid: Uuid) -> Result<Response<Body>, Rejection> {
+        let data = match Mapper::get(uid) {
+            Some(mut r) => {
+                let mut buf = vec![];
+                r.read_to_end(&mut buf).unwrap();
+                build_ok(buf)
+            }
+            None => build_not_found(),
+        };
+        Ok(data)
+    }
 }
--- a/bin/u_server/src/storages/files.rs
+++ b/bin/u_server/src/storages/files.rs
@ -0,0 +1,65 @@
+use crate::worker::Worker;
+use once_cell::sync::Lazy;
+use sha2::{Digest, Sha512};
+use std::fs;
+use std::path::PathBuf;
+use u_lib::utils::{Hexlify, SharedStorage};
+
+pub struct FileMeta {
+    hashsum: String,
+    path: PathBuf,
+    name: String,
+}
+
+static FILES: Lazy<SharedStorage<String, FileMeta>> = SharedStorage::new();
+
+pub struct FileIndex;
+
+impl FileIndex {
+    pub fn search_by_name(pat: impl Into<String>) -> Option<String> {
+        let name = pat.into();
+        let (key, _) = FILES.lock().iter().find(|(_, v)| v.name == name)?;
+        Some(*key)
+    }
+
+    pub fn get(key: impl Into<String>) -> Option<FileMeta> {
+        let name = key.into();
+        FILES.get(&name).and_then(|r| Some(*r))
+    }
+}
+
+impl Worker for FileIndex {
+    fn process(&mut self) {
+        let hexlify = |data: &[u8]| format!("{:x}", Hexlify(data));
+        let mut hasher = Sha512::new();
+        for file in fs::read_dir("./files/hosted/").unwrap() {
+            let file = match file {
+                Ok(f) => f.path(),
+                Err(e) => {
+                    warn!("Error reading file: {}", e);
+                    continue;
+                }
+            };
+            if file.is_file() {
+                let file_data = fs::read(file).unwrap();
+                hasher.update(file_data);
+                let hashsum = hasher.finalize_reset();
+                let shrinked_sum = hashsum
+                    .iter()
+                    .cloned()
+                    .take(32)
+                    .map(|b| b ^ 0x5a)
+                    .collect::<Vec<u8>>();
+                let shrinked_sum = hexlify(&shrinked_sum);
+                let hashsum = hexlify(&hashsum);
+                let name = file.file_name().unwrap().to_string_lossy().into_owned();
+                let meta = FileMeta {
+                    hashsum,
+                    name,
+                    path: file,
+                };
+                FILES.lock().insert(shrinked_sum, meta);
+            }
+        }
+    }
+}
--- a/bin/u_server/src/storages/mapper.rs
+++ b/bin/u_server/src/storages/mapper.rs
@ -0,0 +1,78 @@
+use once_cell::sync::Lazy;
+use std::collections::HashSet;
+use std::io::Read;
+use std::iter::Iterator;
+use std::time::{Duration, SystemTime};
+use u_lib::utils::SharedStorage;
+use uuid::Uuid;
+
+type Data = Box<dyn Read + Send + Sync>;
+
+/// This struct holds uids of various readable objects (files, streams, etc...)
+/// with timeout and permissions for any uids to read data
+struct MappedData {
+    //ids that allowed to read the data
+    for_ids: HashSet<Uuid>,
+    created: SystemTime,
+    timeout: Duration,
+    data: Data,
+}
+
+impl MappedData {
+    pub fn new(
+        ids: impl Iterator<Item = Uuid>,
+        data: impl Read + Send + Sync + 'static,
+        timeout: Duration,
+    ) -> Self {
+        MappedData {
+            for_ids: ids.collect(),
+            created: SystemTime::now(),
+            timeout,
+            data: Box::new(data),
+        }
+    }
+}
+
+static MAPPER: Lazy<SharedStorage<Uuid, MappedData>> = SharedStorage::new();
+
+pub struct Mapper;
+
+impl Mapper {
+    fn remove_overdue() {
+        MAPPER
+            .lock()
+            .retain(|_, v| v.created.elapsed().unwrap() < v.timeout);
+    }
+
+    pub fn get(item_uid: Uuid, getter_uid: Uuid) -> Option<Data> {
+        Self::remove_overdue();
+        let allowed_ids = MAPPER.lock().get(&item_uid)?.for_ids;
+        if allowed_ids.contains(&getter_uid) {
+            MAPPER.lock().remove(&item_uid).map(|d| d.data)
+        } else {
+            None
+        }
+    }
+
+    pub fn set(
+        for_ids: impl Iterator<Item = Uuid>,
+        data: impl Read + Send + Sync + 'static,
+        timeout: Option<Duration>,
+    ) -> Uuid {
+        Self::remove_overdue();
+        let uid = Uuid::new_v4();
+        let timeout = timeout.unwrap_or(Duration::from_secs(60));
+        let mapped = MappedData::new(for_ids, data, timeout);
+        MAPPER.lock().insert(uid, mapped);
+        uid
+    }
+}
+
+// init:
+// fill index struct with files allowed to download (probably auto-refresh index on file adding)
+// hashmap<stripped_hash_of_file_name, PathBuf>
+// download:
+// 1) agent sends download_request with hash of wanted file
+// 2) handler checks if this agent is allowed to dl file
+// 3) if all ok, create in entry in mapper and return its uid and hashsum to agent
+// 4) agent downloads file by uid, uid destructed after that
--- a/bin/u_server/src/storages/mod.rs
+++ b/bin/u_server/src/storages/mod.rs
@ -0,0 +1,5 @@
+pub mod files;
+pub mod mapper;
+
+pub use files::FileIndex;
+pub use mapper::Mapper;
--- a/bin/u_server/src/u_server.rs
+++ b/bin/u_server/src/u_server.rs
@ -14,13 +14,25 @@ extern crate diesel;
 mod db;
 mod filters;
 mod handlers;
+mod storages;
+mod worker;

 use db::UDB;
 use filters::make_filters;
+use std::thread;
+use std::time::Duration;
+use storages::FileIndex;
 use u_lib::{config::MASTER_PORT, models::*, utils::init_env};
 use warp::Filter;
+use worker::Worker;

-const LOGFILE: &str = "u_server.log";
+fn init_workers() {
+    let workers: Vec<Box<dyn Worker + Send>> = vec![Box::new(FileIndex)];
+    thread::spawn(move || loop {
+        workers.into_iter().for_each(|w| w.process());
+        thread::sleep(Duration::from_secs(5));
+    });
+}

 fn prefill_jobs() {
    let agent_hello = JobMeta::builder()
@ -28,12 +40,19 @@ fn prefill_jobs() {
        .with_alias("agent_hello")
        .build()
        .unwrap();
-    UDB::lock_db().insert_jobs(&[agent_hello]).ok();
+    let update_agent = JobMeta::builder()
+        .with_type(misc::JobType::Update)
+        .with_alias("update_agent")
+        .build()
+        .unwrap();
+    let all = [agent_hello, update_agent];
+    UDB::lock_db().insert_jobs(&all).ok();
 }

 fn init_logger() {
    use simplelog::*;
    use std::fs::OpenOptions;
+    const LOGFILE: &str = "u_server.log";
    let log_cfg = ConfigBuilder::new()
        .set_time_format_str("%x %X")
        .set_time_to_local(true)
--- a/bin/u_server/src/worker.rs
+++ b/bin/u_server/src/worker.rs
@ -0,0 +1,3 @@
+pub trait Worker {
+    fn process(&mut self);
+}