forked from ghe0/brain-to-surreal
Compare commits
8 Commits
55549ba10d
...
b3e3e8c996
Author | SHA1 | Date | |
---|---|---|---|
b3e3e8c996 | |||
c8a215c8d3 | |||
34bd7617bc | |||
351f0c5462 | |||
0ece617496 | |||
6a99c146ce | |||
d9f4df2c3d | |||
ee75412bb0 |
@ -1,10 +1,11 @@
|
||||
DEFINE TABLE account SCHEMAFULL;
|
||||
DEFINE FIELD balance ON TABLE account TYPE int;
|
||||
DEFINE FIELD tmp_locked ON TABLE account TYPE int;
|
||||
DEFINE FIELD escrow ON TABLE account TYPE int;
|
||||
DEFINE FIELD email ON TABLE account TYPE string;
|
||||
DEFINE FIELD balance ON TABLE account TYPE int DEFAULT 0;
|
||||
DEFINE FIELD tmp_locked ON TABLE account TYPE int DEFAULT 0;
|
||||
DEFINE FIELD escrow ON TABLE account TYPE int DEFAULT 0;
|
||||
DEFINE FIELD email ON TABLE account TYPE string DEFAULT "";
|
||||
|
||||
DEFINE TABLE vm_node SCHEMAFULL;
|
||||
DEFINE FIELD operator ON TABLE vm_node TYPE record<account>;
|
||||
DEFINE FIELD country ON TABLE vm_node TYPE string;
|
||||
DEFINE FIELD region ON TABLE vm_node TYPE string;
|
||||
DEFINE FIELD city ON TABLE vm_node TYPE string;
|
||||
@ -37,6 +38,7 @@ DEFINE FIELD locked_nano ON TABLE vm_contract TYPE int;
|
||||
DEFINE FIELD collected_at ON TABLE vm_contract TYPE datetime;
|
||||
|
||||
DEFINE TABLE app_node SCHEMAFULL;
|
||||
DEFINE FIELD operator ON TABLE app_node TYPE record<account>;
|
||||
DEFINE FIELD country ON TABLE app_node TYPE string;
|
||||
DEFINE FIELD region ON TABLE app_node TYPE string;
|
||||
DEFINE FIELD city ON TABLE app_node TYPE string;
|
||||
@ -75,7 +77,5 @@ DEFINE FIELD reason ON TABLE kick TYPE string;
|
||||
DEFINE FIELD contract ON TABLE kick TYPE record<vm_contract|app_contract>;
|
||||
|
||||
DEFINE TABLE report TYPE RELATION FROM account TO vm_node|app_node;
|
||||
DEFINE FIELD created_at ON TABLE ban TYPE datetime;
|
||||
DEFINE FIELD reason ON TABLE ban TYPE string;
|
||||
|
||||
DEFINE TABLE operator TYPE RELATION FROM account TO vm_node|app_node;
|
||||
DEFINE FIELD created_at ON TABLE report TYPE datetime;
|
||||
DEFINE FIELD reason ON TABLE report TYPE string;
|
||||
|
@ -4,8 +4,8 @@ use surreal_brain::constants::{
|
||||
BRAIN_GRPC_ADDR, CERT_KEY_PATH, CERT_PATH, DB_ADDRESS, DB_NAME, DB_NS,
|
||||
};
|
||||
use surreal_brain::db;
|
||||
use surreal_brain::grpc::BrainGeneralCliMock;
|
||||
use surreal_brain::grpc::BrainVmCliMock;
|
||||
use surreal_brain::grpc::BrainGeneralCliForReal;
|
||||
use surreal_brain::grpc::BrainVmCliForReal;
|
||||
use tonic::transport::{Identity, Server, ServerTlsConfig};
|
||||
|
||||
#[tokio::main]
|
||||
@ -14,8 +14,8 @@ async fn main() {
|
||||
db::init(DB_ADDRESS, DB_NS, DB_NAME).await.unwrap();
|
||||
let addr = BRAIN_GRPC_ADDR.parse().unwrap();
|
||||
|
||||
let snp_cli_server = BrainVmCliServer::new(BrainVmCliMock {});
|
||||
let general_service_server = BrainGeneralCliServer::new(BrainGeneralCliMock {});
|
||||
let snp_cli_server = BrainVmCliServer::new(BrainVmCliForReal {});
|
||||
let general_service_server = BrainGeneralCliServer::new(BrainGeneralCliForReal {});
|
||||
|
||||
let cert = std::fs::read_to_string(CERT_PATH).unwrap();
|
||||
let key = std::fs::read_to_string(CERT_KEY_PATH).unwrap();
|
||||
|
194
src/db.rs
194
src/db.rs
@ -1,3 +1,4 @@
|
||||
use crate::constants::{ACCOUNT, DB_PASS, DB_USER, VM_CONTRACT, VM_NODE};
|
||||
use crate::old_brain;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::LazyLock;
|
||||
@ -8,8 +9,6 @@ use surrealdb::{
|
||||
RecordId, Surreal,
|
||||
};
|
||||
|
||||
use crate::constants::{ACCOUNT, DB_PASS, DB_USER, OPERATOR, VM_CONTRACT, VM_NODE};
|
||||
|
||||
pub static DB: LazyLock<Surreal<Client>> = LazyLock::new(Surreal::init);
|
||||
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
@ -31,7 +30,6 @@ pub async fn migration0(old_data: &old_brain::BrainData) -> surrealdb::Result<()
|
||||
let vm_nodes: Vec<VmNode> = old_data.into();
|
||||
let app_nodes: Vec<AppNode> = old_data.into();
|
||||
let vm_contracts: Vec<VmContract> = old_data.into();
|
||||
let operators: Vec<OperatorRelation> = old_data.into();
|
||||
|
||||
println!("Inserting accounts...");
|
||||
let _: Vec<Account> = DB.insert(()).content(accounts).await?;
|
||||
@ -41,8 +39,6 @@ pub async fn migration0(old_data: &old_brain::BrainData) -> surrealdb::Result<()
|
||||
let _: Vec<AppNode> = DB.insert(()).content(app_nodes).await?;
|
||||
println!("Inserting vm contracts...");
|
||||
let _: Vec<VmContract> = DB.insert("vm_contract").relation(vm_contracts).await?;
|
||||
println!("Inserting operators...");
|
||||
let _: Vec<OperatorRelation> = DB.insert(OPERATOR).relation(operators).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@ -81,6 +77,7 @@ impl Account {
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct VmNode {
|
||||
pub id: RecordId,
|
||||
pub operator: RecordId,
|
||||
pub country: String,
|
||||
pub region: String,
|
||||
pub city: String,
|
||||
@ -96,6 +93,33 @@ pub struct VmNode {
|
||||
pub offline_minutes: u64,
|
||||
}
|
||||
|
||||
impl VmNode {
|
||||
pub async fn register(self) -> Result<(), Error> {
|
||||
let _: Option<VmNode> = DB.upsert(self.id.clone()).content(self).await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct VmNodeWithReports {
|
||||
pub id: RecordId,
|
||||
pub operator: RecordId,
|
||||
pub country: String,
|
||||
pub region: String,
|
||||
pub city: String,
|
||||
pub ip: String,
|
||||
pub avail_mem_mb: u32,
|
||||
pub avail_vcpus: u32,
|
||||
pub avail_storage_gbs: u32,
|
||||
pub avail_ipv4: u32,
|
||||
pub avail_ipv6: u32,
|
||||
pub avail_ports: u32,
|
||||
pub max_ports_per_vm: u32,
|
||||
pub price: u64,
|
||||
pub offline_minutes: u64,
|
||||
pub reports: Vec<Report>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct VmContract {
|
||||
pub id: RecordId,
|
||||
@ -138,6 +162,8 @@ impl VmContract {
|
||||
}
|
||||
}
|
||||
|
||||
impl VmContract {}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct VmContractWithNode {
|
||||
pub id: RecordId,
|
||||
@ -177,6 +203,14 @@ impl VmContractWithNode {
|
||||
Ok(contracts)
|
||||
}
|
||||
|
||||
pub async fn list_by_node(admin: &str) -> Result<Vec<Self>, Error> {
|
||||
let mut result = DB
|
||||
.query(format!("select * from {VM_CONTRACT} where out = {VM_NODE}:{admin} fetch out;"))
|
||||
.await?;
|
||||
let contracts: Vec<Self> = result.take(0)?;
|
||||
Ok(contracts)
|
||||
}
|
||||
|
||||
pub async fn list_by_operator(operator: &str) -> Result<Vec<Self>, Error> {
|
||||
let mut result = DB
|
||||
.query(format!(
|
||||
@ -217,18 +251,37 @@ impl VmContractWithNode {
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct AppNode {
|
||||
id: RecordId,
|
||||
country: String,
|
||||
region: String,
|
||||
city: String,
|
||||
ip: String,
|
||||
avail_mem_mb: u32,
|
||||
avail_vcpus: u32,
|
||||
avail_storage_gbs: u32,
|
||||
avail_ports: u32,
|
||||
max_ports_per_app: u32,
|
||||
price: u64,
|
||||
offline_minutes: u64,
|
||||
pub id: RecordId,
|
||||
pub operator: RecordId,
|
||||
pub country: String,
|
||||
pub region: String,
|
||||
pub city: String,
|
||||
pub ip: String,
|
||||
pub avail_mem_mb: u32,
|
||||
pub avail_vcpus: u32,
|
||||
pub avail_storage_gbs: u32,
|
||||
pub avail_ports: u32,
|
||||
pub max_ports_per_app: u32,
|
||||
pub price: u64,
|
||||
pub offline_minutes: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct AppNodeWithReports {
|
||||
pub id: RecordId,
|
||||
pub operator: RecordId,
|
||||
pub country: String,
|
||||
pub region: String,
|
||||
pub city: String,
|
||||
pub ip: String,
|
||||
pub avail_mem_mb: u32,
|
||||
pub avail_vcpus: u32,
|
||||
pub avail_storage_gbs: u32,
|
||||
pub avail_ports: u32,
|
||||
pub max_ports_per_app: u32,
|
||||
pub price: u64,
|
||||
pub offline_minutes: u64,
|
||||
pub reports: Vec<Report>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
@ -284,7 +337,7 @@ pub struct Report {
|
||||
#[serde(rename = "out")]
|
||||
to_node: RecordId,
|
||||
created_at: Datetime,
|
||||
reason: String,
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
impl Report {
|
||||
@ -302,23 +355,6 @@ impl Report {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct OperatorRelation {
|
||||
#[serde(rename = "in")]
|
||||
pub account: RecordId,
|
||||
#[serde(rename = "out")]
|
||||
pub node: RecordId,
|
||||
}
|
||||
|
||||
impl OperatorRelation {
|
||||
fn new(account: &str, vm_node: &str) -> Self {
|
||||
Self {
|
||||
account: RecordId::from(("account", account.to_string())),
|
||||
node: RecordId::from(("vm_node", vm_node.to_string())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// This is the operator obtained from the DB,
|
||||
/// however the relation is defined using OperatorRelation
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
@ -335,21 +371,62 @@ impl Operator {
|
||||
pub async fn list() -> Result<Vec<Self>, Error> {
|
||||
let mut result = DB
|
||||
.query(format!(
|
||||
"select *,
|
||||
in as account,
|
||||
<-account.email[0] as email,
|
||||
<-account.escrow[0] as escrow,
|
||||
count(->vm_node) as vm_nodes,
|
||||
count(->app_node) as app_nodes,
|
||||
(select in from <-account->operator->vm_node<-report).len() +
|
||||
(select in from <-account->operator->app_node<-report).len()
|
||||
as reports
|
||||
from operator group by in;"
|
||||
"array::distinct(array::flatten( [
|
||||
(select operator from vm_node group by operator).operator,
|
||||
(select operator from app_node group by operator).operator
|
||||
]));"
|
||||
))
|
||||
.await?;
|
||||
let operators: Vec<Self> = result.take(0)?;
|
||||
let operator_accounts: Vec<RecordId> = result.take(0)?;
|
||||
let mut operators: Vec<Self> = Vec::new();
|
||||
for account in operator_accounts.iter() {
|
||||
if let Some(operator) = Self::inspect(&account.key().to_string()).await? {
|
||||
operators.push(operator);
|
||||
}
|
||||
}
|
||||
Ok(operators)
|
||||
}
|
||||
|
||||
pub async fn inspect(account: &str) -> Result<Option<Self>, Error> {
|
||||
let mut result = DB
|
||||
.query(format!(
|
||||
"$vm_nodes = (select id from vm_node where operator = account:{account}).id;
|
||||
$app_nodes = (select id from app_node where operator = account:{account}).id;
|
||||
select *,
|
||||
id as account,
|
||||
email,
|
||||
escrow,
|
||||
$vm_nodes.len() as vm_nodes,
|
||||
$app_nodes.len() as app_nodes,
|
||||
(select id from report where $vm_nodes contains out).len() +
|
||||
(select id from report where $app_nodes contains out).len()
|
||||
as reports
|
||||
from account where id = account:{account};"
|
||||
))
|
||||
.await?;
|
||||
let operator: Option<Self> = result.take(2)?;
|
||||
Ok(operator)
|
||||
}
|
||||
|
||||
pub async fn inspect_nodes(
|
||||
account: &str,
|
||||
) -> Result<(Option<Self>, Vec<VmNodeWithReports>, Vec<AppNodeWithReports>), Error> {
|
||||
let operator = Self::inspect(account).await?;
|
||||
let mut result = DB
|
||||
.query(format!(
|
||||
"select *, operator, <-report.* as reports from vm_node
|
||||
where operator = account:{account};"
|
||||
))
|
||||
.query(format!(
|
||||
"select *, operator, <-report.* as reports from app_node
|
||||
where operator = account:{account};"
|
||||
))
|
||||
.await?;
|
||||
let vm_nodes: Vec<VmNodeWithReports> = result.take(0)?;
|
||||
let app_nodes: Vec<AppNodeWithReports> = result.take(1)?;
|
||||
|
||||
Ok((operator, vm_nodes, app_nodes))
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: delete all of these From implementation after migration 0 gets executed
|
||||
@ -359,7 +436,8 @@ impl From<&old_brain::BrainData> for Vec<VmNode> {
|
||||
let mut nodes = Vec::new();
|
||||
for old_node in old_data.vm_nodes.iter() {
|
||||
nodes.push(VmNode {
|
||||
id: RecordId::from(("vm_node", old_node.public_key.clone())),
|
||||
id: RecordId::from((VM_NODE, old_node.public_key.clone())),
|
||||
operator: RecordId::from((ACCOUNT, old_node.operator_wallet.clone())),
|
||||
country: old_node.country.clone(),
|
||||
region: old_node.region.clone(),
|
||||
city: old_node.city.clone(),
|
||||
@ -388,7 +466,11 @@ impl From<&old_brain::BrainData> for Vec<VmContract> {
|
||||
mapped_ports.push((*port, 8080 as u32));
|
||||
}
|
||||
contracts.push(VmContract {
|
||||
id: RecordId::from((VM_CONTRACT, old_c.uuid.replace("-", ""))),
|
||||
id: RecordId::from((
|
||||
VM_CONTRACT,
|
||||
old_c.node_pubkey.chars().take(20).collect::<String>()
|
||||
+ &old_c.uuid.replace("-", "").chars().take(20).collect::<String>(),
|
||||
)),
|
||||
admin: RecordId::from((ACCOUNT, old_c.admin_pubkey.clone())),
|
||||
vm_node: RecordId::from((VM_NODE, old_c.node_pubkey.clone())),
|
||||
state: "active".to_string(),
|
||||
@ -418,6 +500,7 @@ impl From<&old_brain::BrainData> for Vec<AppNode> {
|
||||
for old_node in old_data.app_nodes.iter() {
|
||||
nodes.push(AppNode {
|
||||
id: RecordId::from(("app_node", old_node.node_pubkey.clone())),
|
||||
operator: RecordId::from((ACCOUNT, old_node.operator_wallet.clone())),
|
||||
country: old_node.country.clone(),
|
||||
region: old_node.region.clone(),
|
||||
city: old_node.city.clone(),
|
||||
@ -455,18 +538,3 @@ impl From<&old_brain::BrainData> for Vec<Account> {
|
||||
accounts
|
||||
}
|
||||
}
|
||||
|
||||
impl From<&old_brain::BrainData> for Vec<OperatorRelation> {
|
||||
fn from(old_data: &old_brain::BrainData) -> Self {
|
||||
let mut operator_entries = Vec::new();
|
||||
for operator in old_data.operators.clone() {
|
||||
for vm_node in operator.1.vm_nodes.iter() {
|
||||
operator_entries.push(OperatorRelation::new(&operator.0, vm_node));
|
||||
}
|
||||
for app_node in operator.1.app_nodes.iter() {
|
||||
operator_entries.push(OperatorRelation::new(&operator.0, app_node));
|
||||
}
|
||||
}
|
||||
operator_entries
|
||||
}
|
||||
}
|
||||
|
238
src/grpc.rs
238
src/grpc.rs
@ -1,7 +1,7 @@
|
||||
#![allow(dead_code)]
|
||||
use crate::constants::ADMIN_ACCOUNTS;
|
||||
use crate::constants::{ACCOUNT, ADMIN_ACCOUNTS, VM_NODE};
|
||||
use crate::db;
|
||||
use detee_shared::app_proto::AppContract;
|
||||
use detee_shared::app_proto::{AppContract, AppNodeListResp};
|
||||
use detee_shared::{
|
||||
common_proto::{Empty, Pubkey},
|
||||
general_proto::{
|
||||
@ -9,7 +9,10 @@ use detee_shared::{
|
||||
InspectOperatorResp, KickReq, KickResp, ListOperatorsResp, RegOperatorReq, ReportNodeReq,
|
||||
SlashReq,
|
||||
},
|
||||
vm_proto::{brain_vm_cli_server::BrainVmCli, ListVmContractsReq, *},
|
||||
vm_proto::{
|
||||
brain_vm_cli_server::BrainVmCli, brain_vm_daemon_server::BrainVmDaemon, ListVmContractsReq,
|
||||
*,
|
||||
},
|
||||
};
|
||||
|
||||
use log::info;
|
||||
@ -19,9 +22,9 @@ use tokio_stream::wrappers::ReceiverStream;
|
||||
// use tokio::sync::mpsc;
|
||||
// use tokio_stream::{wrappers::ReceiverStream, Stream};
|
||||
use tokio_stream::Stream;
|
||||
use tonic::{Request, Response, Status};
|
||||
use tonic::{Request, Response, Status, Streaming};
|
||||
|
||||
pub struct BrainGeneralCliMock {}
|
||||
pub struct BrainGeneralCliForReal {}
|
||||
|
||||
impl From<db::Account> for AccountBalance {
|
||||
fn from(account: db::Account) -> Self {
|
||||
@ -85,8 +88,172 @@ impl From<db::Operator> for ListOperatorsResp {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<db::VmNodeWithReports> for VmNodeListResp {
|
||||
fn from(vm_node: db::VmNodeWithReports) -> Self {
|
||||
Self {
|
||||
operator: vm_node.operator.key().to_string(),
|
||||
node_pubkey: vm_node.id.key().to_string(),
|
||||
country: vm_node.country,
|
||||
region: vm_node.region,
|
||||
city: vm_node.city,
|
||||
ip: vm_node.ip,
|
||||
reports: vm_node.reports.iter().map(|n| n.reason.clone()).collect(),
|
||||
price: vm_node.price,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<db::AppNodeWithReports> for AppNodeListResp {
|
||||
fn from(app_node: db::AppNodeWithReports) -> Self {
|
||||
Self {
|
||||
operator: app_node.operator.key().to_string(),
|
||||
node_pubkey: app_node.id.key().to_string(),
|
||||
country: app_node.country,
|
||||
region: app_node.region,
|
||||
city: app_node.city,
|
||||
ip: app_node.ip,
|
||||
reports: app_node.reports.iter().map(|n| n.reason.clone()).collect(),
|
||||
price: app_node.price,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct BrainVmDaemonForReal {}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl BrainGeneralCli for BrainGeneralCliMock {
|
||||
impl BrainVmDaemon for BrainVmDaemonForReal {
|
||||
type RegisterVmNodeStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
|
||||
async fn register_vm_node(
|
||||
&self,
|
||||
req: Request<RegisterVmNodeReq>,
|
||||
) -> Result<Response<Self::RegisterVmNodeStream>, Status> {
|
||||
let req = check_sig_from_req(req)?;
|
||||
info!("Starting registration process for {:?}", req);
|
||||
db::VmNode {
|
||||
id: surrealdb::RecordId::from((VM_NODE, req.node_pubkey.clone())),
|
||||
operator: surrealdb::RecordId::from((ACCOUNT, req.operator_wallet)),
|
||||
country: req.country,
|
||||
region: req.region,
|
||||
city: req.city,
|
||||
ip: req.main_ip,
|
||||
price: req.price,
|
||||
avail_mem_mb: 0,
|
||||
avail_vcpus: 0,
|
||||
avail_storage_gbs: 0,
|
||||
avail_ipv4: 0,
|
||||
avail_ipv6: 0,
|
||||
avail_ports: 0,
|
||||
max_ports_per_vm: 0,
|
||||
offline_minutes: 0,
|
||||
}
|
||||
.register()
|
||||
.await?;
|
||||
|
||||
info!("Sending existing contracts to {}", req.node_pubkey);
|
||||
let contracts = db::VmContractWithNode::list_by_node(&req.node_pubkey).await?;
|
||||
let (tx, rx) = mpsc::channel(6);
|
||||
tokio::spawn(async move {
|
||||
for contract in contracts {
|
||||
let _ = tx.send(Ok(contract.into())).await;
|
||||
}
|
||||
});
|
||||
let output_stream = ReceiverStream::new(rx);
|
||||
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>,
|
||||
) -> Result<Response<Self::BrainMessagesStream>, Status> {
|
||||
todo!();
|
||||
let auth = req.into_inner();
|
||||
let pubkey = auth.pubkey.clone();
|
||||
check_sig_from_parts(
|
||||
&pubkey,
|
||||
&auth.timestamp,
|
||||
&format!("{:?}", auth.contracts),
|
||||
&auth.signature,
|
||||
)?;
|
||||
info!("Daemon {} connected to receive brain messages", pubkey);
|
||||
// A surreql query that listens live for changes:
|
||||
// live select * from vm_contract where meta::id(id) in "3zRxiGRnf46vd3zAEmpa".."3zRxiGRnf46vd3zAEmpb";
|
||||
//
|
||||
// And the same from Rust:
|
||||
// let mut stream = db
|
||||
// .select("vm_contract")
|
||||
// .range("3zRxiGRnf46vd3zAEmpa".."3zRxiGRnf46vd3zAEmpb")
|
||||
// .live()
|
||||
// .await?;
|
||||
//
|
||||
// fn handle(result: Result<Notification<VmContract>, surrealdb::Error>) {
|
||||
// println!("Received notification: {:?}", result);
|
||||
// }
|
||||
//
|
||||
// while let Some(result) = stream.next().await {
|
||||
// handle(result);
|
||||
// }
|
||||
//
|
||||
|
||||
// let (tx, rx) = mpsc::channel(6);
|
||||
//self.data.add_daemon_tx(&pubkey, tx);
|
||||
//let output_stream = ReceiverStream::new(rx).map(|msg| Ok(msg));
|
||||
//Ok(Response::new(Box::pin(output_stream) as Self::BrainMessagesStream))
|
||||
}
|
||||
|
||||
async fn daemon_messages(
|
||||
&self,
|
||||
_req: Request<Streaming<VmDaemonMessage>>,
|
||||
) -> Result<Response<Empty>, Status> {
|
||||
todo!();
|
||||
// let mut req_stream = req.into_inner();
|
||||
// let pubkey: String;
|
||||
// if let Some(Ok(msg)) = req_stream.next().await {
|
||||
// log::debug!("demon_messages received the following auth message: {:?}", msg.msg);
|
||||
// if let Some(vm_daemon_message::Msg::Auth(auth)) = msg.msg {
|
||||
// pubkey = auth.pubkey.clone();
|
||||
// check_sig_from_parts(
|
||||
// &pubkey,
|
||||
// &auth.timestamp,
|
||||
// &format!("{:?}", auth.contracts),
|
||||
// &auth.signature,
|
||||
// )?;
|
||||
// } else {
|
||||
// return Err(Status::unauthenticated(
|
||||
// "Could not authenticate the daemon: could not extract auth signature",
|
||||
// ));
|
||||
// }
|
||||
// } else {
|
||||
// return Err(Status::unauthenticated("Could not authenticate the daemon"));
|
||||
// }
|
||||
|
||||
// // info!("Received a message from daemon {pubkey}: {daemon_message:?}");
|
||||
// while let Some(daemon_message) = req_stream.next().await {
|
||||
// match daemon_message {
|
||||
// Ok(msg) => match msg.msg {
|
||||
// Some(vm_daemon_message::Msg::NewVmResp(new_vm_resp)) => {
|
||||
// self.data.submit_newvm_resp(new_vm_resp).await;
|
||||
// }
|
||||
// Some(vm_daemon_message::Msg::UpdateVmResp(update_vm_resp)) => {
|
||||
// self.data.submit_updatevm_resp(update_vm_resp).await;
|
||||
// }
|
||||
// Some(vm_daemon_message::Msg::VmNodeResources(node_resources)) => {
|
||||
// self.data.submit_node_resources(node_resources);
|
||||
// }
|
||||
// _ => {}
|
||||
// },
|
||||
// Err(e) => {
|
||||
// log::warn!("Daemon disconnected: {e:?}");
|
||||
// self.data.del_daemon_tx(&pubkey);
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// Ok(Response::new(Empty {}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl BrainGeneralCli for BrainGeneralCliForReal {
|
||||
type ListAccountsStream = Pin<Box<dyn Stream<Item = Result<Account, Status>> + Send>>;
|
||||
type ListAllAppContractsStream =
|
||||
Pin<Box<dyn Stream<Item = Result<AppContract, Status>> + Send>>;
|
||||
@ -135,13 +302,16 @@ impl BrainGeneralCli for BrainGeneralCliMock {
|
||||
|
||||
async fn inspect_operator(
|
||||
&self,
|
||||
_req: Request<Pubkey>,
|
||||
req: Request<Pubkey>,
|
||||
) -> Result<Response<InspectOperatorResp>, Status> {
|
||||
todo!();
|
||||
// match self.data.inspect_operator(&req.into_inner().pubkey) {
|
||||
// Some(op) => Ok(Response::new(op.into())),
|
||||
// None => Err(Status::not_found("The wallet you specified is not an operator")),
|
||||
// }
|
||||
match db::Operator::inspect_nodes(&req.into_inner().pubkey).await? {
|
||||
(Some(op), vm_nodes, app_nodes) => Ok(Response::new(InspectOperatorResp {
|
||||
operator: Some(op.into()),
|
||||
vm_nodes: vm_nodes.into_iter().map(|n| n.into()).collect(),
|
||||
app_nodes: app_nodes.into_iter().map(|n| n.into()).collect(),
|
||||
})),
|
||||
(None, _, _) => Err(Status::not_found("The wallet you specified is not an operator")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn register_operator(
|
||||
@ -245,10 +415,10 @@ impl BrainGeneralCli for BrainGeneralCliMock {
|
||||
}
|
||||
}
|
||||
|
||||
pub struct BrainVmCliMock {}
|
||||
pub struct BrainVmCliForReal {}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl BrainVmCli for BrainVmCliMock {
|
||||
impl BrainVmCli for BrainVmCliForReal {
|
||||
type ListVmContractsStream = Pin<Box<dyn Stream<Item = Result<VmContract, Status>> + Send>>;
|
||||
type ListVmNodesStream = Pin<Box<dyn Stream<Item = Result<VmNodeListResp, Status>> + Send>>;
|
||||
|
||||
@ -500,6 +670,46 @@ fn check_sig_from_req<T: std::fmt::Debug + PubkeyGetter>(req: Request<T>) -> Res
|
||||
Ok(req)
|
||||
}
|
||||
|
||||
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 seconds_elapsed > 4 || seconds_elapsed < -4 {
|
||||
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(())
|
||||
}
|
||||
|
||||
fn check_admin_key<T>(req: &Request<T>) -> Result<(), Status> {
|
||||
let pubkey = match req.metadata().get("pubkey") {
|
||||
Some(p) => p.clone(),
|
||||
|
Loading…
Reference in New Issue
Block a user