simplified folder structure

rustfmt.toml

@@ -1,3 +1,3 @@
 reorder_impl_items = true
 use_small_heuristics = "Max"
-imports_granularity = "Crate"
+imports_granularity = "Module"

src/bin/brain.rs

@@ -6,9 +6,8 @@ use detee_shared::vm_proto::brain_vm_daemon_server::BrainVmDaemonServer;
 use dotenv::dotenv;
 use surreal_brain::constants::{BRAIN_GRPC_ADDR, CERT_KEY_PATH, CERT_PATH};
 use surreal_brain::db;
-use surreal_brain::grpc::prelude::{
+use surreal_brain::grpc::general::GeneralCliServer;
-    BrainGeneralCliForReal, BrainVmCliForReal, BrainVmDaemonForReal,
+use surreal_brain::grpc::vm::{VmCliServer, VmDaemonServer};
-};
 use tonic::transport::{Identity, Server, ServerTlsConfig};
 
 #[tokio::main]
@@ -27,10 +26,9 @@ async fn main() {
 
     let addr = BRAIN_GRPC_ADDR.parse().unwrap();
 
-    let snp_daemon_server = BrainVmDaemonServer::new(BrainVmDaemonForReal::new(db_arc.clone()));
+    let snp_daemon_server = BrainVmDaemonServer::new(VmDaemonServer::new(db_arc.clone()));
-    let snp_cli_server = BrainVmCliServer::new(BrainVmCliForReal::new(db_arc.clone()));
+    let snp_cli_server = BrainVmCliServer::new(VmCliServer::new(db_arc.clone()));
-    let general_service_server =
+    let general_service_server = BrainGeneralCliServer::new(GeneralCliServer::new(db_arc.clone()));
-        BrainGeneralCliServer::new(BrainGeneralCliForReal::new(db_arc.clone()));
 
     let cert = std::fs::read_to_string(CERT_PATH).unwrap();
     let key = std::fs::read_to_string(CERT_KEY_PATH).unwrap();

src/db/app.rs

@@ -4,7 +4,9 @@ use crate::db::general::Report;
 use super::Error;
 use crate::old_brain;
 use serde::{Deserialize, Serialize};
-use surrealdb::{engine::remote::ws::Client, sql::Datetime, RecordId, Surreal};
+use surrealdb::engine::remote::ws::Client;
+use surrealdb::sql::Datetime;
+use surrealdb::{RecordId, Surreal};
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AppNode {

src/db/common/mod.rs

@@ -1,114 +0,0 @@
-pub use crate::constants::{
-    ACCOUNT, ACTIVE_APP, ACTIVE_VM, DB_SCHEMA_FILE, DELETED_VM, ID_ALPHABET, NEW_VM_REQ,
-    UPDATE_VM_REQ, VM_CONTRACT, VM_NODE,
-};
-
-use crate::db::prelude::*;
-use crate::old_brain;
-use serde::{Deserialize, Serialize};
-use surrealdb::Notification;
-use surrealdb::{
-    engine::remote::ws::{Client, Ws},
-    opt::auth::Root,
-    Surreal,
-};
-use tokio::sync::mpsc::Sender;
-use tokio_stream::StreamExt;
-
-#[derive(thiserror::Error, Debug)]
-pub enum Error {
-    #[error("Internal DB error: {0}")]
-    DataBase(#[from] surrealdb::Error),
-    #[error("Daemon channel got closed: {0}")]
-    DaemonConnection(#[from] tokio::sync::mpsc::error::SendError<DaemonNotification>),
-    #[error(transparent)]
-    StdIo(#[from] std::io::Error),
-    #[error(transparent)]
-    TimeOut(#[from] tokio::time::error::Elapsed),
-}
-
-pub async fn db_connection(
-    db_address: &str,
-    username: &str,
-    password: &str,
-    ns: &str,
-    db: &str,
-) -> Result<Surreal<Client>, Error> {
-    let db_connection: Surreal<Client> = Surreal::init();
-    db_connection.connect::<Ws>(db_address).await?;
-    // Sign in to the server
-    db_connection.signin(Root { username, password }).await?;
-    db_connection.use_ns(ns).use_db(db).await?;
-    Ok(db_connection)
-}
-
-pub async fn migration0(
-    db: &Surreal<Client>,
-    old_data: &old_brain::BrainData,
-) -> Result<(), Error> {
-    let accounts: Vec<Account> = old_data.into();
-    let vm_nodes: Vec<VmNode> = old_data.into();
-    let app_nodes: Vec<AppNode> = old_data.into();
-    let vm_contracts: Vec<ActiveVm> = old_data.into();
-
-    let schema = std::fs::read_to_string(DB_SCHEMA_FILE)?;
-    db.query(schema).await?;
-
-    println!("Inserting accounts...");
-    let _: Vec<Account> = db.insert(()).content(accounts).await?;
-    println!("Inserting vm nodes...");
-    let _: Vec<VmNode> = db.insert(()).content(vm_nodes).await?;
-    println!("Inserting app nodes...");
-    let _: Vec<AppNode> = db.insert(()).content(app_nodes).await?;
-    println!("Inserting vm contracts...");
-    let _: Vec<ActiveVm> = db.insert("vm_contract").relation(vm_contracts).await?;
-
-    Ok(())
-}
-
-pub async fn upsert_record<SomeRecord: Serialize + 'static>(
-    db: &Surreal<Client>,
-    table: &str,
-    id: &str,
-    my_record: SomeRecord,
-) -> Result<(), Error> {
-    #[derive(Deserialize)]
-    struct Wrapper {}
-    let _: Option<Wrapper> = db.create((table, id)).content(my_record).await?;
-    Ok(())
-}
-
-pub async fn listen_for_node<
-    T: Into<DaemonNotification> + std::marker::Unpin + for<'de> Deserialize<'de>,
->(
-    db: &Surreal<Client>,
-    node: &str,
-    tx: Sender<DaemonNotification>,
-) -> Result<(), Error> {
-    let table_name = match std::any::type_name::<T>() {
-        "surreal_brain::db::NewVmReq" => NEW_VM_REQ.to_string(),
-        "surreal_brain::db::UpdateVmReq" => UPDATE_VM_REQ.to_string(),
-        "surreal_brain::db::DeletedVm" => DELETED_VM.to_string(),
-        wat => {
-            log::error!("listen_for_node: T has type {wat}");
-            String::from("wat")
-        }
-    };
-    let mut resp =
-        db.query(format!("live select * from {table_name} where out = vm_node:{node};")).await?;
-    let mut live_stream = resp.stream::<Notification<T>>(0)?;
-    while let Some(result) = live_stream.next().await {
-        match result {
-            Ok(notification) => {
-                if notification.action == surrealdb::Action::Create {
-                    tx.send(notification.data.into()).await?
-                }
-            }
-            Err(e) => {
-                log::warn!("listen_for_deletion DB stream failed for {node}: {e}");
-                return Err(Error::from(e));
-            }
-        }
-    }
-    Ok(())
-}

src/db/general.rs

@@ -1,11 +1,12 @@
 use crate::constants::ACCOUNT;
-use crate::db::app::AppNodeWithReports;
+use crate::db::prelude::*;
-use crate::db::vm::VmNodeWithReports;
 
 use super::Error;
 use crate::old_brain;
 use serde::{Deserialize, Serialize};
-use surrealdb::{engine::remote::ws::Client, sql::Datetime, RecordId, Surreal};
+use surrealdb::engine::remote::ws::Client;
+use surrealdb::sql::Datetime;
+use surrealdb::{RecordId, Surreal};
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Account {

src/db/mod.rs

@@ -1,21 +1,118 @@
 pub mod app;
-pub mod common;
 pub mod general;
 pub mod vm;
 
-pub use prelude::*;
+use crate::constants::{DELETED_VM, NEW_VM_REQ, UPDATE_VM_REQ};
-pub mod prelude {
+use crate::old_brain;
-    pub use super::app::{ActiveAppWithNode, AppNode, AppNodeWithReports};
+use prelude::*;
-    pub use super::common::listen_for_node;
+use serde::{Deserialize, Serialize};
-    pub use super::common::{db_connection, migration0, upsert_record, Error};
+use surrealdb::engine::remote::ws::{Client, Ws};
-    pub use super::general::{Account, Operator, Report};
+use surrealdb::opt::auth::Root;
-    pub use super::vm::{
+use surrealdb::{Notification, Surreal};
-        ActiveVm, ActiveVmWithNode, DaemonNotification, DeletedVm, NewVmReq, NewVmResp,
+use tokio::sync::mpsc::Sender;
-        UpdateVmReq, VmNode, VmNodeResources, VmNodeWithReports,
+use tokio_stream::StreamExt;
-    };
+
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("Internal DB error: {0}")]
+    DataBase(#[from] surrealdb::Error),
+    #[error("Daemon channel got closed: {0}")]
+    VmDaemonConnection(#[from] tokio::sync::mpsc::error::SendError<VmDaemonMsgs>),
+    #[error(transparent)]
+    StdIo(#[from] std::io::Error),
+    #[error(transparent)]
+    TimeOut(#[from] tokio::time::error::Elapsed),
 }
 
-pub use crate::constants::{
+pub mod prelude {
-    ACCOUNT, ACTIVE_APP, ACTIVE_VM, DB_SCHEMA_FILE, DELETED_VM, ID_ALPHABET, NEW_VM_REQ,
+    pub use super::app::*;
-    UPDATE_VM_REQ, VM_CONTRACT, VM_NODE,
+    pub use super::general::*;
-};
+    pub use super::vm::*;
+    pub use super::*;
+}
+
+pub async fn db_connection(
+    db_address: &str,
+    username: &str,
+    password: &str,
+    ns: &str,
+    db: &str,
+) -> Result<Surreal<Client>, Error> {
+    let db_connection: Surreal<Client> = Surreal::init();
+    db_connection.connect::<Ws>(db_address).await?;
+    // Sign in to the server
+    db_connection.signin(Root { username, password }).await?;
+    db_connection.use_ns(ns).use_db(db).await?;
+    Ok(db_connection)
+}
+
+pub async fn migration0(
+    db: &Surreal<Client>,
+    old_data: &old_brain::BrainData,
+) -> Result<(), Error> {
+    let accounts: Vec<Account> = old_data.into();
+    let vm_nodes: Vec<VmNode> = old_data.into();
+    let app_nodes: Vec<AppNode> = old_data.into();
+    let vm_contracts: Vec<ActiveVm> = old_data.into();
+
+    let schema = std::fs::read_to_string(crate::constants::DB_SCHEMA_FILE)?;
+    db.query(schema).await?;
+
+    println!("Inserting accounts...");
+    let _: Vec<Account> = db.insert(()).content(accounts).await?;
+    println!("Inserting vm nodes...");
+    let _: Vec<VmNode> = db.insert(()).content(vm_nodes).await?;
+    println!("Inserting app nodes...");
+    let _: Vec<AppNode> = db.insert(()).content(app_nodes).await?;
+    println!("Inserting vm contracts...");
+    let _: Vec<ActiveVm> = db.insert("vm_contract").relation(vm_contracts).await?;
+
+    Ok(())
+}
+
+pub async fn upsert_record<SomeRecord: Serialize + 'static>(
+    db: &Surreal<Client>,
+    table: &str,
+    id: &str,
+    my_record: SomeRecord,
+) -> Result<(), Error> {
+    #[derive(Deserialize)]
+    struct Wrapper {}
+    let _: Option<Wrapper> = db.create((table, id)).content(my_record).await?;
+    Ok(())
+}
+
+pub async fn live_vmnode_msgs<
+    T: Into<vm::VmDaemonMsgs> + std::marker::Unpin + for<'de> Deserialize<'de>,
+>(
+    db: &Surreal<Client>,
+    node: &str,
+    tx: Sender<vm::VmDaemonMsgs>,
+) -> Result<(), Error> {
+    let table_name = match std::any::type_name::<T>() {
+        "surreal_brain::db::NewVmReq" => NEW_VM_REQ.to_string(),
+        "surreal_brain::db::UpdateVmReq" => UPDATE_VM_REQ.to_string(),
+        "surreal_brain::db::DeletedVm" => DELETED_VM.to_string(),
+        wat => {
+            log::error!("listen_for_node: T has type {wat}");
+            String::from("wat")
+        }
+    };
+    let mut resp =
+        db.query(format!("live select * from {table_name} where out = vm_node:{node};")).await?;
+    let mut live_stream = resp.stream::<Notification<T>>(0)?;
+    while let Some(result) = live_stream.next().await {
+        match result {
+            Ok(notification) => {
+                if notification.action == surrealdb::Action::Create {
+                    tx.send(notification.data.into()).await?
+                }
+            }
+            Err(e) => {
+                log::warn!("listen_for_deletion DB stream failed for {node}: {e}");
+                return Err(Error::from(e));
+            }
+        }
+    }
+    Ok(())
+}

src/db/vm.rs

@@ -1,11 +1,14 @@
-use std::{str::FromStr, time::Duration};
+use std::str::FromStr;
+use std::time::Duration;
 
 use super::Error;
 use crate::constants::{ACCOUNT, ACTIVE_VM, DELETED_VM, NEW_VM_REQ, VM_NODE};
 use crate::db::general::Report;
 use crate::old_brain;
 use serde::{Deserialize, Serialize};
-use surrealdb::{engine::remote::ws::Client, sql::Datetime, Notification, RecordId, Surreal};
+use surrealdb::engine::remote::ws::Client;
+use surrealdb::sql::Datetime;
+use surrealdb::{Notification, RecordId, Surreal};
 use tokio_stream::StreamExt as _;
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -115,25 +118,25 @@ impl VmNodeWithReports {
     }
 }
 
-pub enum DaemonNotification {
+pub enum VmDaemonMsgs {
     Create(NewVmReq),
     Update(UpdateVmReq),
     Delete(DeletedVm),
 }
 
-impl From<NewVmReq> for DaemonNotification {
+impl From<NewVmReq> for VmDaemonMsgs {
     fn from(value: NewVmReq) -> Self {
         Self::Create(value)
     }
 }
 
-impl From<UpdateVmReq> for DaemonNotification {
+impl From<UpdateVmReq> for VmDaemonMsgs {
     fn from(value: UpdateVmReq) -> Self {
         Self::Update(value)
     }
 }
 
-impl From<DeletedVm> for DaemonNotification {
+impl From<DeletedVm> for VmDaemonMsgs {
     fn from(value: DeletedVm) -> Self {
         Self::Delete(value)
     }

src/grpc/common/mod.rs

@@ -1,173 +0,0 @@
-pub mod types;
-
-use crate::constants::ADMIN_ACCOUNTS;
-use detee_shared::{
-    common_proto::{Empty, Pubkey},
-    general_proto::{AirdropReq, BanUserReq, KickReq, RegOperatorReq, ReportNodeReq, SlashReq},
-    vm_proto::{ListVmContractsReq, *},
-};
-use tonic::{Request, Status};
-
-pub trait PubkeyGetter {
-    fn get_pubkey(&self) -> Option<String>;
-}
-
-macro_rules! impl_pubkey_getter {
-    ($t:ty, $field:ident) => {
-        impl PubkeyGetter for $t {
-            fn get_pubkey(&self) -> Option<String> {
-                Some(self.$field.clone())
-            }
-        }
-    };
-    ($t:ty) => {
-        impl PubkeyGetter for $t {
-            fn get_pubkey(&self) -> Option<String> {
-                None
-            }
-        }
-    };
-}
-
-impl_pubkey_getter!(Pubkey, pubkey);
-impl_pubkey_getter!(NewVmReq, admin_pubkey);
-impl_pubkey_getter!(DeleteVmReq, admin_pubkey);
-impl_pubkey_getter!(UpdateVmReq, admin_pubkey);
-impl_pubkey_getter!(ExtendVmReq, admin_pubkey);
-impl_pubkey_getter!(ReportNodeReq, admin_pubkey);
-impl_pubkey_getter!(ListVmContractsReq, wallet);
-impl_pubkey_getter!(RegisterVmNodeReq, node_pubkey);
-impl_pubkey_getter!(RegOperatorReq, pubkey);
-impl_pubkey_getter!(KickReq, operator_wallet);
-impl_pubkey_getter!(BanUserReq, operator_wallet);
-
-impl_pubkey_getter!(VmNodeFilters);
-impl_pubkey_getter!(Empty);
-impl_pubkey_getter!(AirdropReq);
-impl_pubkey_getter!(SlashReq);
-
-// impl_pubkey_getter!(NewAppReq, admin_pubkey);
-// impl_pubkey_getter!(DelAppReq, admin_pubkey);
-// impl_pubkey_getter!(ListAppContractsReq, admin_pubkey);
-//
-// impl_pubkey_getter!(RegisterAppNodeReq);
-// impl_pubkey_getter!(AppNodeFilters);
-
-pub fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Result<T, Status> {
-    let time = match req.metadata().get("timestamp") {
-        Some(t) => t.clone(),
-        None => return Err(Status::unauthenticated("Timestamp not found in metadata.")),
-    };
-    let time = time
-        .to_str()
-        .map_err(|_| Status::unauthenticated("Timestamp in metadata is not a string"))?;
-
-    let now = chrono::Utc::now();
-    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
-        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
-    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
-    if !(-4..=4).contains(&seconds_elapsed) {
-        return Err(Status::unauthenticated(format!(
-            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
-            parsed_time, now
-        )));
-    }
-
-    let signature = match req.metadata().get("request-signature") {
-        Some(t) => t,
-        None => return Err(Status::unauthenticated("signature not found in metadata.")),
-    };
-    let signature = bs58::decode(signature)
-        .into_vec()
-        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
-    let signature = ed25519_dalek::Signature::from_bytes(
-        signature
-            .as_slice()
-            .try_into()
-            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
-    );
-
-    let pubkey_value = match req.metadata().get("pubkey") {
-        Some(p) => p.clone(),
-        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
-    };
-    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
-        &bs58::decode(&pubkey_value)
-            .into_vec()
-            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
-            .try_into()
-            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
-    )
-    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
-
-    let req = req.into_inner();
-    let message = format!("{time}{req:?}");
-    use ed25519_dalek::Verifier;
-    pubkey
-        .verify(message.as_bytes(), &signature)
-        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
-    if let Some(req_pubkey) = req.get_pubkey() {
-        if *pubkey_value.to_str().unwrap() != req_pubkey {
-            return Err(Status::unauthenticated(
-                "pubkey of signature does not match pubkey of request",
-            ));
-        }
-    }
-    Ok(req)
-}
-
-pub fn check_sig_from_parts(pubkey: &str, time: &str, msg: &str, sig: &str) -> Result<(), Status> {
-    let now = chrono::Utc::now();
-    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
-        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
-    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
-    if !(-4..=4).contains(&seconds_elapsed) {
-        return Err(Status::unauthenticated(format!(
-            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
-            parsed_time, now
-        )));
-    }
-
-    let signature = bs58::decode(sig)
-        .into_vec()
-        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
-    let signature = ed25519_dalek::Signature::from_bytes(
-        signature
-            .as_slice()
-            .try_into()
-            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
-    );
-
-    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
-        &bs58::decode(&pubkey)
-            .into_vec()
-            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
-            .try_into()
-            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
-    )
-    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
-
-    let msg = time.to_string() + msg;
-    use ed25519_dalek::Verifier;
-    pubkey
-        .verify(msg.as_bytes(), &signature)
-        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
-
-    Ok(())
-}
-
-pub fn check_admin_key<T>(req: &Request<T>) -> Result<(), Status> {
-    let pubkey = match req.metadata().get("pubkey") {
-        Some(p) => p.clone(),
-        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
-    };
-    let pubkey = pubkey
-        .to_str()
-        .map_err(|_| Status::unauthenticated("could not parse pubkey metadata to str"))?;
-
-    if !ADMIN_ACCOUNTS.contains(&pubkey) {
-        return Err(Status::unauthenticated("This operation is reserved to admin accounts"));
-    }
-
-    Ok(())
-}

src/grpc/general.rs

@@ -1,16 +1,15 @@
-use crate::db;
+use crate::db::prelude as db;
-use crate::grpc::prelude::*;
+use crate::grpc::{check_admin_key, check_sig_from_req};
 use detee_shared::app_proto::AppContract;
-use detee_shared::{
+use detee_shared::common_proto::{Empty, Pubkey};
-    common_proto::{Empty, Pubkey},
+use detee_shared::general_proto::brain_general_cli_server::BrainGeneralCli;
-    general_proto::{
+use detee_shared::general_proto::{
-        brain_general_cli_server::BrainGeneralCli, Account, AccountBalance, AirdropReq, BanUserReq,
+    Account, AccountBalance, AirdropReq, BanUserReq, InspectOperatorResp, KickReq, KickResp,
-        InspectOperatorResp, KickReq, KickResp, ListOperatorsResp, RegOperatorReq, ReportNodeReq,
+    ListOperatorsResp, RegOperatorReq, ReportNodeReq, SlashReq,
-        SlashReq,
-    },
-    vm_proto::VmContract,
 };
-use surrealdb::{engine::remote::ws::Client, Surreal};
+use detee_shared::vm_proto::VmContract;
+use surrealdb::engine::remote::ws::Client;
+use surrealdb::Surreal;
 
 use std::pin::Pin;
 use std::sync::Arc;
@@ -19,18 +18,18 @@ use tokio_stream::wrappers::ReceiverStream;
 use tokio_stream::Stream;
 use tonic::{Request, Response, Status};
 
-pub struct BrainGeneralCliForReal {
+pub struct GeneralCliServer {
     pub db: Arc<Surreal<Client>>,
 }
 
-impl BrainGeneralCliForReal {
+impl GeneralCliServer {
     pub fn new(db: Arc<Surreal<Client>>) -> Self {
         Self { db }
     }
 }
 
 #[tonic::async_trait]
-impl BrainGeneralCli for BrainGeneralCliForReal {
+impl BrainGeneralCli for GeneralCliServer {
     type ListAccountsStream = Pin<Box<dyn Stream<Item = Result<Account, Status>> + Send>>;
     type ListAllAppContractsStream =
         Pin<Box<dyn Stream<Item = Result<AppContract, Status>> + Send>>;
@@ -53,17 +52,24 @@ impl BrainGeneralCli for BrainGeneralCliForReal {
                 {
                     (vm_contract.admin, vm_contract.vm_node.id, vm_contract.id.to_string())
                 }
-                _ => match db::ActiveAppWithNode::get_by_uuid(&self.db, &req.contract).await? {
+                _ => {
-                    Some(app_contract)
+                    match db::app::ActiveAppWithNode::get_by_uuid(&self.db, &req.contract).await? {
-                        if app_contract.admin.key().to_string() == req.admin_pubkey
+                        Some(app_contract)
-                            && app_contract.app_node.id.key().to_string() == req.node_pubkey =>
+                            if app_contract.admin.key().to_string() == req.admin_pubkey
-                    {
+                                && app_contract.app_node.id.key().to_string()
-                        (app_contract.admin, app_contract.app_node.id, app_contract.id.to_string())
+                                    == req.node_pubkey =>
+                        {
+                            (
+                                app_contract.admin,
+                                app_contract.app_node.id,
+                                app_contract.id.to_string(),
+                            )
+                        }
+                        _ => {
+                            return Err(Status::unauthenticated("No contract found by this ID."));
+                        }
                     }
-                    _ => {
+                }
-                        return Err(Status::unauthenticated("No contract found by this ID."));
-                    }
-                },
             };
         db::Report::create(&self.db, account, node, req.reason, contract_id).await?;
         Ok(Response::new(Empty {}))

src/grpc/mod.rs

@@ -1,12 +1,176 @@
 pub mod app;
-pub mod common;
 pub mod general;
+pub mod types;
 pub mod vm;
 
-pub mod prelude {
+use crate::constants::ADMIN_ACCOUNTS;
-    pub use super::common::{
+use detee_shared::common_proto::{Empty, Pubkey};
-        check_admin_key, check_sig_from_parts, check_sig_from_req, PubkeyGetter,
+use detee_shared::general_proto::{
-    };
+    AirdropReq, BanUserReq, KickReq, RegOperatorReq, ReportNodeReq, SlashReq,
-    pub use super::general::BrainGeneralCliForReal;
+};
-    pub use super::vm::{BrainVmCliForReal, BrainVmDaemonForReal};
+use detee_shared::vm_proto::{ListVmContractsReq, *};
+use tonic::{Request, Status};
+
+pub trait PubkeyGetter {
+    fn get_pubkey(&self) -> Option<String>;
+}
+
+macro_rules! impl_pubkey_getter {
+    ($t:ty, $field:ident) => {
+        impl PubkeyGetter for $t {
+            fn get_pubkey(&self) -> Option<String> {
+                Some(self.$field.clone())
+            }
+        }
+    };
+    ($t:ty) => {
+        impl PubkeyGetter for $t {
+            fn get_pubkey(&self) -> Option<String> {
+                None
+            }
+        }
+    };
+}
+
+impl_pubkey_getter!(Pubkey, pubkey);
+impl_pubkey_getter!(NewVmReq, admin_pubkey);
+impl_pubkey_getter!(DeleteVmReq, admin_pubkey);
+impl_pubkey_getter!(UpdateVmReq, admin_pubkey);
+impl_pubkey_getter!(ExtendVmReq, admin_pubkey);
+impl_pubkey_getter!(ReportNodeReq, admin_pubkey);
+impl_pubkey_getter!(ListVmContractsReq, wallet);
+impl_pubkey_getter!(RegisterVmNodeReq, node_pubkey);
+impl_pubkey_getter!(RegOperatorReq, pubkey);
+impl_pubkey_getter!(KickReq, operator_wallet);
+impl_pubkey_getter!(BanUserReq, operator_wallet);
+
+impl_pubkey_getter!(VmNodeFilters);
+impl_pubkey_getter!(Empty);
+impl_pubkey_getter!(AirdropReq);
+impl_pubkey_getter!(SlashReq);
+
+// impl_pubkey_getter!(NewAppReq, admin_pubkey);
+// impl_pubkey_getter!(DelAppReq, admin_pubkey);
+// impl_pubkey_getter!(ListAppContractsReq, admin_pubkey);
+//
+// impl_pubkey_getter!(RegisterAppNodeReq);
+// impl_pubkey_getter!(AppNodeFilters);
+
+pub fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Result<T, Status> {
+    let time = match req.metadata().get("timestamp") {
+        Some(t) => t.clone(),
+        None => return Err(Status::unauthenticated("Timestamp not found in metadata.")),
+    };
+    let time = time
+        .to_str()
+        .map_err(|_| Status::unauthenticated("Timestamp in metadata is not a string"))?;
+
+    let now = chrono::Utc::now();
+    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
+        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
+    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
+    if !(-4..=4).contains(&seconds_elapsed) {
+        return Err(Status::unauthenticated(format!(
+            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
+            parsed_time, now
+        )));
+    }
+
+    let signature = match req.metadata().get("request-signature") {
+        Some(t) => t,
+        None => return Err(Status::unauthenticated("signature not found in metadata.")),
+    };
+    let signature = bs58::decode(signature)
+        .into_vec()
+        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
+    let signature = ed25519_dalek::Signature::from_bytes(
+        signature
+            .as_slice()
+            .try_into()
+            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
+    );
+
+    let pubkey_value = match req.metadata().get("pubkey") {
+        Some(p) => p.clone(),
+        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
+    };
+    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
+        &bs58::decode(&pubkey_value)
+            .into_vec()
+            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
+            .try_into()
+            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
+    )
+    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
+
+    let req = req.into_inner();
+    let message = format!("{time}{req:?}");
+    use ed25519_dalek::Verifier;
+    pubkey
+        .verify(message.as_bytes(), &signature)
+        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
+    if let Some(req_pubkey) = req.get_pubkey() {
+        if *pubkey_value.to_str().unwrap() != req_pubkey {
+            return Err(Status::unauthenticated(
+                "pubkey of signature does not match pubkey of request",
+            ));
+        }
+    }
+    Ok(req)
+}
+
+pub fn check_sig_from_parts(pubkey: &str, time: &str, msg: &str, sig: &str) -> Result<(), Status> {
+    let now = chrono::Utc::now();
+    let parsed_time = chrono::DateTime::parse_from_rfc3339(time)
+        .map_err(|_| Status::unauthenticated("Coult not parse timestamp"))?;
+    let seconds_elapsed = now.signed_duration_since(parsed_time).num_seconds();
+    if !(-4..=4).contains(&seconds_elapsed) {
+        return Err(Status::unauthenticated(format!(
+            "Date is not within 4 sec of the time of the server: CLI {} vs Server {}",
+            parsed_time, now
+        )));
+    }
+
+    let signature = bs58::decode(sig)
+        .into_vec()
+        .map_err(|_| Status::unauthenticated("signature is not a bs58 string"))?;
+    let signature = ed25519_dalek::Signature::from_bytes(
+        signature
+            .as_slice()
+            .try_into()
+            .map_err(|_| Status::unauthenticated("could not parse ed25519 signature"))?,
+    );
+
+    let pubkey = ed25519_dalek::VerifyingKey::from_bytes(
+        &bs58::decode(&pubkey)
+            .into_vec()
+            .map_err(|_| Status::unauthenticated("pubkey is not a bs58 string"))?
+            .try_into()
+            .map_err(|_| Status::unauthenticated("pubkey does not have the correct size."))?,
+    )
+    .map_err(|_| Status::unauthenticated("could not parse ed25519 pubkey"))?;
+
+    let msg = time.to_string() + msg;
+    use ed25519_dalek::Verifier;
+    pubkey
+        .verify(msg.as_bytes(), &signature)
+        .map_err(|_| Status::unauthenticated("the signature is not valid"))?;
+
+    Ok(())
+}
+
+pub fn check_admin_key<T>(req: &Request<T>) -> Result<(), Status> {
+    let pubkey = match req.metadata().get("pubkey") {
+        Some(p) => p.clone(),
+        None => return Err(Status::unauthenticated("pubkey not found in metadata.")),
+    };
+    let pubkey = pubkey
+        .to_str()
+        .map_err(|_| Status::unauthenticated("could not parse pubkey metadata to str"))?;
+
+    if !ADMIN_ACCOUNTS.contains(&pubkey) {
+        return Err(Status::unauthenticated("This operation is reserved to admin accounts"));
+    }
+
+    Ok(())
 }

src/grpc/types.rs

@@ -1,9 +1,8 @@
-use crate::db;
+use crate::constants::{ACCOUNT, ID_ALPHABET, NEW_VM_REQ, VM_NODE};
+use crate::db::prelude as db;
 use detee_shared::app_proto::AppNodeListResp;
-use detee_shared::{
+use detee_shared::general_proto::{AccountBalance, ListOperatorsResp};
-    general_proto::{AccountBalance, ListOperatorsResp},
+use detee_shared::vm_proto::*;
-    vm_proto::*,
-};
 use nanoid::nanoid;
 
 use surrealdb::RecordId;
@@ -17,9 +16,9 @@ impl From<db::Account> for AccountBalance {
 impl From<NewVmReq> for db::NewVmReq {
     fn from(new_vm_req: NewVmReq) -> Self {
         Self {
-            id: RecordId::from((db::NEW_VM_REQ, nanoid!(40, &db::ID_ALPHABET))),
+            id: RecordId::from((NEW_VM_REQ, nanoid!(40, &ID_ALPHABET))),
-            admin: RecordId::from((db::ACCOUNT, new_vm_req.admin_pubkey)),
+            admin: RecordId::from((ACCOUNT, new_vm_req.admin_pubkey)),
-            vm_node: RecordId::from((db::VM_NODE, new_vm_req.node_pubkey)),
+            vm_node: RecordId::from((VM_NODE, new_vm_req.node_pubkey)),
             hostname: new_vm_req.hostname,
             extra_ports: new_vm_req.extra_ports,
             public_ipv4: new_vm_req.public_ipv4,
@@ -102,16 +101,16 @@ impl From<db::DeletedVm> for DeleteVmReq {
     }
 }
 
-impl From<db::DaemonNotification> for BrainVmMessage {
+impl From<db::VmDaemonMsgs> for BrainVmMessage {
-    fn from(notification: db::DaemonNotification) -> Self {
+    fn from(notification: db::VmDaemonMsgs) -> Self {
         match notification {
-            db::DaemonNotification::Create(new_vm_req) => {
+            db::VmDaemonMsgs::Create(new_vm_req) => {
                 BrainVmMessage { msg: Some(brain_vm_message::Msg::NewVmReq(new_vm_req.into())) }
             }
-            db::DaemonNotification::Update(update_vm_req) => BrainVmMessage {
+            db::VmDaemonMsgs::Update(update_vm_req) => BrainVmMessage {
                 msg: Some(brain_vm_message::Msg::UpdateVmReq(update_vm_req.into())),
             },
-            db::DaemonNotification::Delete(deleted_vm) => {
+            db::VmDaemonMsgs::Delete(deleted_vm) => {
                 BrainVmMessage { msg: Some(brain_vm_message::Msg::DeleteVm(deleted_vm.into())) }
             }
         }

src/grpc/vm.rs

@@ -1,15 +1,13 @@
 #![allow(dead_code)]
 use crate::constants::{ACCOUNT, VM_NODE};
-use crate::db;
+use crate::db::prelude as db;
-use crate::grpc::prelude::*;
+use crate::grpc::{check_sig_from_parts, check_sig_from_req};
-use detee_shared::{
+use detee_shared::common_proto::Empty;
-    common_proto::Empty,
+use detee_shared::vm_proto::brain_vm_cli_server::BrainVmCli;
-    vm_proto::{
+use detee_shared::vm_proto::brain_vm_daemon_server::BrainVmDaemon;
-        brain_vm_cli_server::BrainVmCli, brain_vm_daemon_server::BrainVmDaemon, ListVmContractsReq,
+use detee_shared::vm_proto::{ListVmContractsReq, *};
-        *,
+use surrealdb::engine::remote::ws::Client;
-    },
+use surrealdb::Surreal;
-};
-use surrealdb::{engine::remote::ws::Client, Surreal};
 
 use log::info;
 use std::pin::Pin;
@@ -19,19 +17,21 @@ use tokio_stream::wrappers::ReceiverStream;
 use tokio_stream::{Stream, StreamExt};
 use tonic::{Request, Response, Status, Streaming};
 
-pub struct BrainVmDaemonForReal {
+pub struct VmDaemonServer {
     pub db: Arc<Surreal<Client>>,
 }
 
-impl BrainVmDaemonForReal {
+impl VmDaemonServer {
     pub fn new(db: Arc<Surreal<Client>>) -> Self {
         Self { db }
     }
 }
 
 #[tonic::async_trait]
-impl BrainVmDaemon for BrainVmDaemonForReal {
+impl BrainVmDaemon for VmDaemonServer {
+    type BrainMessagesStream = Pin<Box<dyn Stream<Item = Result<BrainVmMessage, Status>> + Send>>;
     type RegisterVmNodeStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
+
     async fn register_vm_node(
         &self,
         req: Request<RegisterVmNodeReq>,
@@ -70,7 +70,6 @@ impl BrainVmDaemon for BrainVmDaemonForReal {
         Ok(Response::new(Box::pin(output_stream) as Self::RegisterVmNodeStream))
     }
 
-    type BrainMessagesStream = Pin<Box<dyn Stream<Item = Result<BrainVmMessage, Status>> + Send>>;
     async fn brain_messages(
         &self,
         req: Request<DaemonStreamAuth>,
@@ -91,10 +90,10 @@ impl BrainVmDaemon for BrainVmDaemonForReal {
             let pubkey = pubkey.clone();
             let tx = tx.clone();
             tokio::spawn(async move {
-                match db::listen_for_node::<db::DeletedVm>(&db, &pubkey, tx).await {
+                match db::live_vmnode_msgs::<db::DeletedVm>(&db, &pubkey, tx).await {
-                    Ok(()) => log::info!("db::VmContract::listen_for_node ended for {pubkey}"),
+                    Ok(()) => log::info!("live_vmnode_msgs ended for {pubkey}"),
                     Err(e) => {
-                        log::warn!("db::VmContract::listen_for_node errored for {pubkey}: {e}")
+                        log::warn!("live_vmnode_msgs errored for {pubkey}: {e}")
                     }
                 };
             });
@@ -104,7 +103,7 @@ impl BrainVmDaemon for BrainVmDaemonForReal {
             let pubkey = pubkey.clone();
             let tx = tx.clone();
             tokio::spawn(async move {
-                let _ = db::listen_for_node::<db::NewVmReq>(&db, &pubkey, tx.clone()).await;
+                let _ = db::live_vmnode_msgs::<db::NewVmReq>(&db, &pubkey, tx.clone()).await;
             });
         }
         {
@@ -112,7 +111,7 @@ impl BrainVmDaemon for BrainVmDaemonForReal {
             let pubkey = pubkey.clone();
             let tx = tx.clone();
             tokio::spawn(async move {
-                let _ = db::listen_for_node::<db::UpdateVmReq>(&db, &pubkey, tx.clone()).await;
+                let _ = db::live_vmnode_msgs::<db::UpdateVmReq>(&db, &pubkey, tx.clone()).await;
             });
         }
 
@@ -190,25 +189,25 @@ impl BrainVmDaemon for BrainVmDaemonForReal {
     }
 }
 
-pub struct BrainVmCliForReal {
+pub struct VmCliServer {
     pub db: Arc<Surreal<Client>>,
 }
 
-impl BrainVmCliForReal {
+impl VmCliServer {
     pub fn new(db: Arc<Surreal<Client>>) -> Self {
         Self { db }
     }
 }
 
 #[tonic::async_trait]
-impl BrainVmCli for BrainVmCliForReal {
+impl BrainVmCli for VmCliServer {
     type ListVmContractsStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
     type ListVmNodesStream = Pin<Box<dyn Stream<Item = Result<VmNodeListResp, Status>> + Send>>;
 
     async fn new_vm(&self, req: Request<NewVmReq>) -> Result<Response<NewVmResp>, Status> {
         let req = check_sig_from_req(req)?;
         info!("New VM requested via CLI: {req:?}");
-        if db::Account::is_banned_by_node(&self.db, &req.admin_pubkey, &req.node_pubkey).await? {
+        if db::general::Account::is_banned_by_node(&self.db, &req.admin_pubkey, &req.node_pubkey).await? {
             return Err(Status::permission_denied("This operator banned you. What did you do?"));
         }
 

tests/common/prepare_test_env.rs

@@ -1,17 +1,19 @@
+use detee_shared::general_proto::brain_general_cli_server::BrainGeneralCliServer;
 use detee_shared::vm_proto::brain_vm_cli_server::BrainVmCliServer;
-use detee_shared::{
+use detee_shared::vm_proto::brain_vm_daemon_server::BrainVmDaemonServer;
-    general_proto::brain_general_cli_server::BrainGeneralCliServer,
-    vm_proto::brain_vm_daemon_server::BrainVmDaemonServer,
-};
 use dotenv::dotenv;
 use hyper_util::rt::TokioIo;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use surreal_brain::grpc::{BrainGeneralCliForReal, BrainVmCliForReal, BrainVmDaemonForReal};
+use surreal_brain::grpc::{
+    general::GeneralCliServer,
+    vm::{VmCliServer, VmDaemonServer},
+};
 use surrealdb::engine::remote::ws::Client;
 use surrealdb::Surreal;
 use tokio::io::DuplexStream;
-use tokio::{net::TcpListener, sync::OnceCell};
+use tokio::net::TcpListener;
+use tokio::sync::OnceCell;
 use tonic::transport::{Channel, Endpoint, Server, Uri};
 use tower::service_fn;
 
@@ -58,9 +60,9 @@ pub async fn run_service_in_background() -> SocketAddr {
         let db_arc = Arc::new(db);
 
         Server::builder()
-            .add_service(BrainGeneralCliServer::new(BrainGeneralCliForReal::new(db_arc.clone())))
+            .add_service(BrainGeneralCliServer::new(GeneralCliServer::new(db_arc.clone())))
-            .add_service(BrainVmCliServer::new(BrainVmCliForReal::new(db_arc.clone())))
+            .add_service(BrainVmCliServer::new(VmCliServer::new(db_arc.clone())))
-            .add_service(BrainVmDaemonServer::new(BrainVmDaemonForReal::new(db_arc.clone())))
+            .add_service(BrainVmDaemonServer::new(VmDaemonServer::new(db_arc.clone())))
             .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener))
             .await
             .unwrap();
@@ -88,9 +90,9 @@ pub async fn run_service_for_stream_server() -> DuplexStream {
         let db_arc = Arc::new(db);
 
         tonic::transport::Server::builder()
-            .add_service(BrainGeneralCliServer::new(BrainGeneralCliForReal::new(db_arc.clone())))
+            .add_service(BrainGeneralCliServer::new(GeneralCliServer::new(db_arc.clone())))
-            .add_service(BrainVmCliServer::new(BrainVmCliForReal::new(db_arc.clone())))
+            .add_service(BrainVmCliServer::new(VmCliServer::new(db_arc.clone())))
-            .add_service(BrainVmDaemonServer::new(BrainVmDaemonForReal::new(db_arc.clone())))
+            .add_service(BrainVmDaemonServer::new(VmDaemonServer::new(db_arc.clone())))
             .serve_with_incoming(tokio_stream::once(Ok::<_, std::io::Error>(server)))
             .await
     });

tests/common/test_utils.rs

@@ -1,7 +1,6 @@
 use anyhow::Result;
 use detee_shared::vm_proto as snp_proto;
-use ed25519_dalek::Signer;
+use ed25519_dalek::{Signer, SigningKey};
-use ed25519_dalek::SigningKey;
 use tonic::metadata::AsciiMetadataValue;
 use tonic::Request;
 

tests/common/vm_cli_utils.rs

@@ -2,7 +2,7 @@ use super::test_utils::Key;
 use detee_shared::vm_proto;
 use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient;
 use surreal_brain::constants::{ACTIVE_VM, NEW_VM_REQ};
-use surreal_brain::db;
+use surreal_brain::db::prelude as db;
 use surrealdb::engine::remote::ws::Client;
 use surrealdb::Surreal;
 use tonic::transport::Channel;

tests/grpc_test.rs

@@ -1,19 +1,17 @@
-use common::{
+use common::prepare_test_env::{
-    prepare_test_env::{prepare_test_db, run_service_for_stream, run_service_in_background},
+    prepare_test_db, run_service_for_stream, run_service_in_background,
-    test_utils::Key,
-    vm_cli_utils::create_new_vm,
-    vm_daemon_utils::mock_vm_daemon,
 };
-use detee_shared::common_proto::Empty;
+use common::test_utils::Key;
+use common::vm_cli_utils::create_new_vm;
+use common::vm_daemon_utils::mock_vm_daemon;
+use detee_shared::common_proto::{Empty, Pubkey};
+use detee_shared::general_proto::brain_general_cli_client::BrainGeneralCliClient;
 use detee_shared::general_proto::ReportNodeReq;
 use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient;
 use detee_shared::vm_proto::ListVmContractsReq;
-use detee_shared::{
-    common_proto::Pubkey, general_proto::brain_general_cli_client::BrainGeneralCliClient,
-};
 use futures::StreamExt;
 use surreal_brain::constants::VM_NODE;
-use surreal_brain::db::VmNodeWithReports;
+use surreal_brain::db::vm::VmNodeWithReports;
 
 mod common;
 

tests/grpc_vm_daemon_test.rs

@@ -1,8 +1,8 @@
-use common::{
+use common::prepare_test_env::{
-    prepare_test_env::{prepare_test_db, run_service_for_stream, run_service_in_background},
+    prepare_test_db, run_service_for_stream, run_service_in_background,
-    test_utils::Key,
-    vm_daemon_utils::{mock_vm_daemon, register_vm_node},
 };
+use common::test_utils::Key;
+use common::vm_daemon_utils::{mock_vm_daemon, register_vm_node};
 use detee_shared::vm_proto;
 use detee_shared::vm_proto::brain_vm_cli_client::BrainVmCliClient;
 use detee_shared::vm_proto::brain_vm_daemon_client::BrainVmDaemonClient;