From b5ec374fa4c2454fd93660da26fa65b00d178e2d Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Fri, 19 Jul 2024 15:53:22 +0800 Subject: [PATCH 1/3] support ztm share repo --- common/src/model.rs | 11 + gateway/src/ca_server.rs | 156 +++++++ gateway/src/https_server.rs | 65 ++- gateway/src/lib.rs | 1 + gateway/src/relay_server.rs | 133 +++--- gemini/Cargo.toml | 2 + gemini/src/ca/mod.rs | 174 ++++++++ gemini/src/http/handler.rs | 207 ++++++++++ gemini/src/lib.rs | 19 + gemini/src/ztm/agent.rs | 568 ++++++++++++++++++++++++++ gemini/src/ztm/hub.rs | 122 ++++++ gemini/src/ztm/mod.rs | 463 +-------------------- jupiter/src/storage/git_db_storage.rs | 10 + neptune/src/lib.rs | 3 +- 14 files changed, 1400 insertions(+), 534 deletions(-) create mode 100755 gateway/src/ca_server.rs create mode 100755 gemini/src/ca/mod.rs create mode 100644 gemini/src/ztm/agent.rs create mode 100644 gemini/src/ztm/hub.rs mode change 100644 => 100755 gemini/src/ztm/mod.rs diff --git a/common/src/model.rs b/common/src/model.rs index 13374561e..0c7354ded 100644 --- a/common/src/model.rs +++ b/common/src/model.rs @@ -14,6 +14,15 @@ pub struct CommonOptions { #[arg(long, default_value_t = 8001)] pub relay_port: u16, + #[arg(long, default_value_t = 7777)] + pub ztm_agent_port: u16, + + #[arg(long, default_value_t = 8888)] + pub ztm_hub_port: u16, + + #[arg(long, default_value_t = 9999)] + pub ca_port: u16, + #[arg(long)] pub bootstrap_node: Option, } @@ -26,4 +35,6 @@ pub struct GetParams { pub path: Option, pub limit: Option, pub cursor: Option, + pub identifier: Option, + pub port: Option, } diff --git a/gateway/src/ca_server.rs b/gateway/src/ca_server.rs new file mode 100755 index 000000000..82277bbc1 --- /dev/null +++ b/gateway/src/ca_server.rs @@ -0,0 +1,156 @@ +use std::net::SocketAddr; +use std::str::FromStr; + +use axum::body::{to_bytes, Body}; +use axum::extract::{Query, State}; +use axum::http::{Request, Response, StatusCode, Uri}; +use axum::routing::get; +use axum::Router; +use common::config::Config; +use gemini::RelayGetParams; +use jupiter::context::Context; +use regex::Regex; +use tower::ServiceBuilder; +use tower_http::cors::{Any, CorsLayer}; +use tower_http::decompression::RequestDecompressionLayer; +use tower_http::trace::TraceLayer; + +use crate::api::api_router::{self}; +use crate::api::ApiServiceState; + +pub async fn run_ca_server(config: Config, _host: String, port: u16) { + let host = "127.0.0.1".to_string(); + let app = app(config.clone(), host.clone(), port).await; + + let server_url = format!("{}:{}", host, port); + tracing::info!("start ca server: {server_url}"); + let addr = SocketAddr::from_str(&server_url).unwrap(); + let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); + axum::serve(listener, app.into_make_service()) + .await + .unwrap(); +} + +#[derive(Clone)] +pub struct AppState { + pub context: Context, + pub host: String, + pub port: u16, +} + +pub async fn app(config: Config, host: String, port: u16) -> Router { + let state = AppState { + host, + port, + context: Context::new(config.clone()).await, + }; + + let api_state = ApiServiceState { + context: Context::new(config).await, + }; + + // add RequestDecompressionLayer for handle gzip encode + // add TraceLayer for log record + // add CorsLayer to add cors header + Router::new() + .nest("/api/", api_router::routers().with_state(api_state)) + .route( + "/*path", + get(get_method_router) + .post(post_method_router) + .delete(delete_method_router), + ) + .layer(ServiceBuilder::new().layer(CorsLayer::new().allow_origin(Any))) + .layer(TraceLayer::new_for_http()) + .layer(RequestDecompressionLayer::new()) + .with_state(state) +} + +async fn get_method_router( + _state: State, + Query(_params): Query, + uri: Uri, +) -> Result, (StatusCode, String)> { + if Regex::new(r"/certificates/[a-zA-Z0-9]+$") + .unwrap() + .is_match(uri.path()) + { + let name = match gemini::ca::get_cert_name_from_path(uri.path()) { + Some(n) => n, + None => { + return gemini::ca::response_error( + StatusCode::BAD_REQUEST.as_u16(), + "Bad request".to_string(), + ) + } + }; + return gemini::ca::get_certificate(name).await; + } + Err(( + StatusCode::NOT_FOUND, + String::from("Operation not supported\n"), + )) +} + +async fn post_method_router( + state: State, + uri: Uri, + req: Request, +) -> Result, (StatusCode, String)> { + let _ztm_config = state.context.config.ztm.clone(); + if Regex::new(r"/certificates/[a-zA-Z0-9]+$") + .unwrap() + .is_match(uri.path()) + { + let name = match gemini::ca::get_cert_name_from_path(uri.path()) { + Some(n) => n, + None => { + return gemini::ca::response_error( + StatusCode::BAD_REQUEST.as_u16(), + "Bad request".to_string(), + ) + } + }; + return gemini::ca::issue_certificate(name).await; + } else if Regex::new(r"/sign/hub/[a-zA-Z0-9]+$") + .unwrap() + .is_match(uri.path()) + { + let name = match gemini::ca::get_hub_name_from_path(uri.path()) { + Some(n) => n, + None => { + return gemini::ca::response_error( + StatusCode::BAD_REQUEST.as_u16(), + "Bad request".to_string(), + ) + } + }; + let bytes = to_bytes(req.into_body(), usize::MAX).await.unwrap(); + let pubkey = String::from_utf8(bytes.to_vec()).unwrap(); + return gemini::ca::sign_certificate(name, pubkey).await; + } + Err(( + StatusCode::NOT_FOUND, + String::from("Operation not supported\n"), + )) +} + +async fn delete_method_router( + _state: State, + uri: Uri, + _req: Request, +) -> Result, (StatusCode, String)> { + if Regex::new(r"/certificates/[a-zA-Z0-9]+$") + .unwrap() + .is_match(uri.path()) + { + return gemini::ca::delete_certificate(uri.path()).await; + } + Err(( + StatusCode::NOT_FOUND, + String::from("Operation not supported\n"), + )) +} + +#[cfg(test)] +mod tests {} diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index bffeef88e..4bf4c17af 100644 --- a/gateway/src/https_server.rs +++ b/gateway/src/https_server.rs @@ -3,6 +3,7 @@ use std::ops::Deref; use std::path::PathBuf; use std::str::FromStr; use std::sync::Arc; +use std::{thread, time}; use anyhow::Result; use axum::body::Body; @@ -14,7 +15,8 @@ use axum::Router; use axum_server::tls_rustls::RustlsConfig; use clap::Args; use common::enums::ZtmType; -use gemini::ztm::{run_ztm_client, LocalZTM}; +use gemini::ztm::agent::{run_ztm_client, LocalZTMAgent}; +use gemini::ztm::hub::LocalZTMHub; use regex::Regex; use tower::ServiceBuilder; use tower_http::cors::{Any, CorsLayer}; @@ -30,6 +32,7 @@ use jupiter::raw_storage::local_storage::LocalStorage; use crate::api::api_router::{self}; use crate::api::ApiServiceState; +use crate::ca_server::run_ca_server; use crate::lfs; use crate::relay_server::run_relay_server; @@ -62,6 +65,7 @@ pub struct AppState { pub context: Context, pub host: String, pub port: u16, + pub common: CommonOptions, } impl From for LfsConfig { @@ -92,9 +96,9 @@ pub async fn https_server(config: Config, options: HttpsOptions) { https_port, } = options.clone(); - check_run_with_ztm(config.clone(), options.common); + check_run_with_ztm(config.clone(), options.common.clone()); - let app = app(config, host.clone(), https_port).await; + let app = app(config, host.clone(), https_port, options.common.clone()).await; let server_url = format!("{}:{}", host, https_port); let addr = SocketAddr::from_str(&server_url).unwrap(); @@ -113,9 +117,9 @@ pub async fn http_server(config: Config, options: HttpOptions) { http_port, } = options.clone(); - check_run_with_ztm(config.clone(), options.common); + check_run_with_ztm(config.clone(), options.common.clone()); - let app = app(config, host.clone(), http_port).await; + let app = app(config, host.clone(), http_port, options.common.clone()).await; let server_url = format!("{}:{}", host, http_port); @@ -126,13 +130,14 @@ pub async fn http_server(config: Config, options: HttpOptions) { .unwrap(); } -pub async fn app(config: Config, host: String, port: u16) -> Router { +pub async fn app(config: Config, host: String, port: u16, common: CommonOptions) -> Router { let context = Context::new(config.clone()).await; context.services.mega_storage.init_monorepo().await; let state = AppState { host, port, context: context.clone(), + common: common.clone(), }; let api_state = ApiServiceState { context }; @@ -175,6 +180,19 @@ async fn get_method_router( TransportProtocol::Http, ); return ceres::http::handler::git_info_refs(params, pack_protocol).await; + } else if Regex::new(r"/ztm/repo_provide$") + .unwrap() + .is_match(uri.path()) + { + return gemini::http::handler::repo_provide( + state.port, + state.common.bootstrap_node.clone(), + state.context.clone(), + params, + ) + .await; + } else if Regex::new(r"/ztm/repo_folk$").unwrap().is_match(uri.path()) { + return gemini::http::handler::repo_folk(state.common.ztm_agent_port, params).await; } else { return Err(( StatusCode::NOT_FOUND, @@ -270,18 +288,37 @@ pub fn check_run_with_ztm(config: Config, common: CommonOptions) { } }; let (peer_id, _) = vault::init(); - let ztm: LocalZTM = LocalZTM { agent_port: 7778 }; - ztm.clone().start_ztm_agent(); - tokio::spawn(async move { run_ztm_client(bootstrap_node, config, peer_id, ztm).await }); + let ztm_agent: LocalZTMAgent = LocalZTMAgent { + agent_port: common.ztm_agent_port, + }; + ztm_agent.clone().start_ztm_agent(); + thread::sleep(time::Duration::from_secs(3)); + tokio::spawn(async move { + run_ztm_client(bootstrap_node, config, peer_id, ztm_agent).await + }); } ZtmType::Relay => { - //Start a sub thread to run relay server + //Start a sub thread to ca server let config_clone = config.clone(); let host_clone = common.host.clone(); - let relay_port = common.relay_port; - tokio::spawn( - async move { run_relay_server(config_clone, host_clone, relay_port).await }, - ); + let ca_port = common.ca_port; + tokio::spawn(async move { run_ca_server(config_clone, host_clone, ca_port).await }); + thread::sleep(time::Duration::from_secs(5)); + + //Start a sub thread to run ztm-hub + let ca = format!("localhost:{ca_port}"); + let ztm_hub: LocalZTMHub = LocalZTMHub { + hub_port: common.ztm_hub_port, + ca, + name: vec!["relay".to_string()], + }; + ztm_hub.clone().start_ztm_hub(); + thread::sleep(time::Duration::from_secs(5)); + + //Start a sub thread to run relay server + let config_clone = config.clone(); + let common: CommonOptions = common.clone(); + tokio::spawn(async move { run_relay_server(config_clone, common).await }); } } } diff --git a/gateway/src/lib.rs b/gateway/src/lib.rs index 85bba6b14..d267a80d6 100644 --- a/gateway/src/lib.rs +++ b/gateway/src/lib.rs @@ -1,4 +1,5 @@ pub mod api; +pub mod ca_server; mod git_protocol; pub mod https_server; pub mod init; diff --git a/gateway/src/relay_server.rs b/gateway/src/relay_server.rs index e8ec9de26..bb276d9b8 100644 --- a/gateway/src/relay_server.rs +++ b/gateway/src/relay_server.rs @@ -1,5 +1,6 @@ use std::net::SocketAddr; use std::str::FromStr; +use std::time::{Duration, SystemTime}; use axum::body::Body; use axum::extract::{FromRequest, Query, State}; @@ -7,8 +8,9 @@ use axum::http::{Request, Response, StatusCode, Uri}; use axum::routing::get; use axum::{Json, Router}; use callisto::{ztm_node, ztm_repo_info}; -use common::config::{Config, ZTMConfig}; -use gemini::ztm::{RemoteZTM, ZTMAgent, ZTMCA}; +use common::config::Config; +use common::model::CommonOptions; +use gemini::ztm::hub::{LocalHub, ZTMCA}; use gemini::{Node, RelayGetParams, RelayResultRes, RepoInfo}; use jupiter::context::Context; use regex::Regex; @@ -20,21 +22,14 @@ use tower_http::trace::TraceLayer; use crate::api::api_router::{self}; use crate::api::ApiServiceState; -pub async fn run_relay_server(config: Config, host: String, port: u16) { - let app = app(config.clone(), host.clone(), port).await; +pub async fn run_relay_server(config: Config, common: CommonOptions) { + let host = common.host.clone(); + let relay_port = common.relay_port; + let hub_port = common.ztm_hub_port; + let ca_port = common.ca_port; + let app = app(config.clone(), host.clone(), relay_port, hub_port, ca_port).await; - let ztm_config = config.ztm; - match relay_connect_ztm(ztm_config, port).await { - Ok(s) => { - tracing::info!("relay connect ztm success: {s}"); - } - Err(e) => { - tracing::error!("relay connect ztm failed : {e}"); - return; - } - } - - let server_url = format!("{}:{}", host, port); + let server_url = format!("{}:{}", host, relay_port); tracing::info!("start relay server: {server_url}"); let addr = SocketAddr::from_str(&server_url).unwrap(); let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); @@ -47,13 +42,23 @@ pub async fn run_relay_server(config: Config, host: String, port: u16) { pub struct AppState { pub context: Context, pub host: String, - pub port: u16, + pub relay_port: u16, + pub hub_port: u16, + pub ca_port: u16, } -pub async fn app(config: Config, host: String, port: u16) -> Router { +pub async fn app( + config: Config, + host: String, + relay_port: u16, + hub_port: u16, + ca_port: u16, +) -> Router { let state = AppState { host, - port, + relay_port, + hub_port, + ca_port, context: Context::new(config.clone()).await, }; @@ -61,6 +66,8 @@ pub async fn app(config: Config, host: String, port: u16) -> Router { context: Context::new(config).await, }; + let context = api_state.context.clone(); + tokio::spawn(async move { loop_running(context).await }); // add RequestDecompressionLayer for handle gzip encode // add TraceLayer for log record // add CorsLayer to add cors header @@ -82,11 +89,10 @@ async fn get_method_router( Query(params): Query, uri: Uri, ) -> Result, (StatusCode, String)> { - let ztm_config = state.context.config.ztm.clone(); if Regex::new(r"/hello$").unwrap().is_match(uri.path()) { return hello_relay(params).await; } else if Regex::new(r"/certificate$").unwrap().is_match(uri.path()) { - return certificate(ztm_config, params).await; + return certificate(state, params).await; } else if Regex::new(r"/ping$").unwrap().is_match(uri.path()) { return ping(state, params).await; } else if Regex::new(r"/node_list$").unwrap().is_match(uri.path()) { @@ -120,7 +126,7 @@ pub async fn hello_relay(_params: RelayGetParams) -> Result, (Sta } pub async fn certificate( - config: ZTMConfig, + state: State, params: RelayGetParams, ) -> Result, (StatusCode, String)> { if params.name.is_none() { @@ -128,7 +134,11 @@ pub async fn certificate( } let name = params.name.unwrap(); - let ztm: RemoteZTM = RemoteZTM { config }; + let ztm: LocalHub = LocalHub { + host: state.host.clone(), + hub_port: state.hub_port, + ca_port: state.ca_port, + }; let permit = match ztm.create_ztm_certificate(name.clone()).await { Ok(p) => p, Err(e) => { @@ -233,50 +243,57 @@ pub async fn repo_list( .into_iter() .map(|x| x.into()) .collect(); - let json_string = serde_json::to_string(&repo_info_list).unwrap(); + let nodelist: Vec = storage + .get_all_node() + .await + .unwrap() + .into_iter() + .map(|x| x.into()) + .collect(); + let mut repo_info_list_result = vec![]; + for mut repo in repo_info_list { + for node in &nodelist { + if repo.origin == node.peer_id { + repo.peer_online = node.online; + } + } + repo_info_list_result.push(repo.clone()); + } + let json_string = serde_json::to_string(&repo_info_list_result).unwrap(); Ok(Response::builder() .header("Content-Type", "application/json") .body(Body::from(json_string)) .unwrap()) } -pub async fn relay_connect_ztm(config: ZTMConfig, relay_port: u16) -> Result { - // 1. generate a permit for relay - let name = "relay".to_string(); - let ztm: RemoteZTM = RemoteZTM { config }; - match ztm.delete_ztm_certificate(name.clone()).await { - Ok(_s) => (), - Err(e) => { - return Err(e); - } - } - let permit = match ztm.create_ztm_certificate(name).await { - Ok(p) => p, - Err(e) => { - return Err(e); - } - }; +async fn loop_running(context: Context) { + let mut interval = tokio::time::interval(Duration::from_secs(60)); - // 2. connect to ZTM hub (join a mesh) - let mesh = match ztm.connect_ztm_hub(permit).await { - Ok(m) => m, - Err(e) => { - return Err(e); - } - }; + loop { + check_nodes_online(context.clone()).await; + interval.tick().await; + } +} - // 3. create a ZTM service - let response_text = match ztm - .create_ztm_service(mesh.agent.id, "relay".to_string(), relay_port) - .await - { - Ok(m) => m, - Err(e) => { - return Err(e); +async fn check_nodes_online(context: Context) { + let storage = context.services.ztm_storage.clone(); + let nodelist: Vec = + storage.get_all_node().await.unwrap().into_iter().collect(); + for mut node in nodelist { + //check online + let from_timestamp = Duration::from_millis(node.last_online_time as u64); + let now = SystemTime::now(); + let elapsed = match now.duration_since(SystemTime::UNIX_EPOCH) { + Ok(dur) => dur, + Err(_) => { + continue; + } + }; + if elapsed.as_secs() > from_timestamp.as_secs() + 60 { + node.online = false; + storage.update_node(node.clone()).await.unwrap(); } - }; - - Ok(response_text) + } } #[cfg(test)] diff --git a/gemini/Cargo.toml b/gemini/Cargo.toml index 8d87e9200..cf1784771 100644 --- a/gemini/Cargo.toml +++ b/gemini/Cargo.toml @@ -20,5 +20,7 @@ tracing = { workspace = true } tokio = { workspace = true, features = ["net"] } chrono = { workspace = true } jupiter = { workspace = true } +vault = { workspace = true } callisto = { workspace = true } +venus = { workspace = true } neptune = { workspace = true } \ No newline at end of file diff --git a/gemini/src/ca/mod.rs b/gemini/src/ca/mod.rs new file mode 100755 index 000000000..b9fe3db69 --- /dev/null +++ b/gemini/src/ca/mod.rs @@ -0,0 +1,174 @@ +use axum::{body::Body, response::Response}; +use reqwest::StatusCode; +use serde::{Deserialize, Serialize}; +use serde_json::json; + +#[derive(Deserialize, Serialize, Debug, Clone)] +struct ErrorResult { + status: u16, + #[serde(rename = "message")] + msg: String, +} + +impl ErrorResult { + pub fn to_json_string(&self) -> String { + serde_json::to_string(self).unwrap() + } + pub fn to_json_string_new(status: u16, msg: String) -> String { + let e = ErrorResult { status, msg }; + e.to_json_string() + } +} + +pub async fn get_certificate(name: String) -> Result { + if name == "ca" { + return Ok(Response::builder() + .body(Body::from(vault::pki::get_root_cert())) + .unwrap()); + } + let cert_option = get_from_vault(name); + match cert_option { + Some(cert) => Ok(Response::builder().body(Body::from(cert)).unwrap()), + None => response_error( + StatusCode::NOT_FOUND.as_u16(), + "Username not found".to_string(), + ), + } +} + +pub async fn issue_certificate(name: String) -> Result { + if is_reserved_name(name.clone()) { + return response_error( + StatusCode::FORBIDDEN.as_u16(), + "Reserved username".to_string(), + ); + } + // let cert_option = get_from_vault(name.clone()); + // if cert_option.is_some() { + // return response_error( + // StatusCode::CONFLICT.as_u16(), + // "Username already exists".to_string(), + // ); + // } + let (cert_pem, private_key) = vault::pki::issue_cert(json!({ + "ttl": "10d", + "common_name": name, + })); + //save cert to vault + save_to_vault(name, cert_pem); + Ok(Response::builder().body(Body::from(private_key)).unwrap()) +} + +pub async fn sign_certificate( + name: String, + pubkey: String, +) -> Result { + tracing::info!("sign_certificate,name:{name},pubkey:{pubkey}"); + if is_reserved_name(name.clone()) { + return response_error( + StatusCode::FORBIDDEN.as_u16(), + "Reserved username".to_string(), + ); + } + let cert_option = get_from_vault(name.clone()); + if cert_option.is_some() { + return response_error( + StatusCode::CONFLICT.as_u16(), + "Username already exists".to_string(), + ); + } + let (cert_pem, private_key) = vault::pki::issue_cert(json!({ + "ttl": "10d", + "common_name": name, + })); + //save cert to vault + save_to_vault(name, cert_pem); + Ok(Response::builder().body(Body::from(private_key)).unwrap()) +} + +pub async fn delete_certificate(path: &str) -> Result { + let name = match get_cert_name_from_path(path) { + Some(n) => n, + None => return response_error(StatusCode::BAD_REQUEST.as_u16(), "Bad request".to_string()), + }; + if is_reserved_name(name.clone()) { + return response_error( + StatusCode::FORBIDDEN.as_u16(), + "Reserved username".to_string(), + ); + } + delete_to_vault(name); + Ok(Response::builder() + .status(204) + .body(Body::from("")) + .unwrap()) +} + +pub fn get_cert_name_from_path(path: &str) -> Option { + let v: Vec<&str> = path.split('/').collect(); + v.get(3).map(|s| s.to_string()) +} + +pub fn get_hub_name_from_path(path: &str) -> Option { + let v: Vec<&str> = path.split('/').collect(); + v.get(4).map(|s| s.to_string()) +} + +fn is_reserved_name(name: String) -> bool { + if name == "ca" { + return true; + } + is_hub_name(name) +} + +fn is_hub_name(_name: String) -> bool { + // if name == "hub" || name.starts_with("hub/") { + // return true; + // } + // false + false +} + +fn save_to_vault(key: String, value: String) { + let key_f = format!("ca_{key}"); + let kv_data = json!({ + key_f.clone(): value, + }) + .as_object() + .unwrap() + .clone(); + vault::vault::write_secret(key_f.as_str(), Some(kv_data.clone())).unwrap(); +} + +fn get_from_vault(key: String) -> Option { + let key_f = format!("ca_{key}"); + let secret = match vault::vault::read_secret(key_f.as_str()).unwrap() { + Some(res) => res.data, + None => return None, + }; + + match secret { + Some(m) => { + let s = m.get(key_f.as_str()).unwrap().as_str().unwrap().to_string(); + let s = s.trim_matches(char::is_control).to_string(); + Some(s) + } + None => None, + } +} + +fn delete_to_vault(_key: String) { + // let key_f = format!("ca_{key}"); + // vault::vault::write_secret(key_f.as_str(), Some(Map).unwrap()); +} + +pub fn response_error(status: u16, message: String) -> Result { + Ok({ + let error_result = ErrorResult::to_json_string_new(status, message); + Response::builder() + .status(status) + .header("Content-Type", "application/json") + .body(Body::from(error_result)) + .unwrap() + }) +} diff --git a/gemini/src/http/handler.rs b/gemini/src/http/handler.rs index 5c6016f6a..d2f0f0a4a 100644 --- a/gemini/src/http/handler.rs +++ b/gemini/src/http/handler.rs @@ -1,2 +1,209 @@ +use axum::{body::Body, http::Response}; +use common::model::GetParams; +use jupiter::context::Context; +use reqwest::StatusCode; +use venus::import_repo::repo::Repo; + +use crate::{ + ztm::agent::{share_repo, LocalZTMAgent, ZTMAgent}, + RepoInfo, +}; + +pub async fn repo_provide( + port: u16, + bootstrap_node: Option, + context: Context, + params: GetParams, +) -> Result, (StatusCode, String)> { + let bootstrap_node_clone = match bootstrap_node { + Some(b) => b.clone(), + None => { + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + String::from("Bootstrap node not provide\n"), + )); + } + }; + let path = match params.path.clone() { + Some(p) => p, + None => { + return Err((StatusCode::BAD_REQUEST, String::from("Path not provide\n"))); + } + }; + let url = format!("{bootstrap_node_clone}/api/v1/repo_provide"); + let git_model = context + .services + .git_db_storage + .find_git_repo_by_path(path.as_str()) + .await; + + let git_model = match git_model { + Ok(r) => { + if let Some(m) = r { + m + } else { + return Err((StatusCode::BAD_REQUEST, String::from("Repo not found"))); + } + } + Err(_) => return Err((StatusCode::BAD_REQUEST, String::from("Repo not found"))), + }; + let repo: Repo = git_model.clone().into(); + let git_ref = context + .services + .git_db_storage + .get_default_ref(&repo) + .await + .unwrap() + .unwrap(); + + let name = git_model.repo_name; + let repo_path = git_model.repo_path; + let (peer_id, _) = vault::init(); + let identifier = format!("p2p://{}/{port}{repo_path}.git", peer_id.clone()); + let update_time = git_model.created_at.and_utc().timestamp(); + let repo_info = RepoInfo { + name, + identifier, + origin: peer_id.clone(), + update_time, + commit: git_ref.ref_hash, + peer_online: true, + }; + share_repo(url.clone(), repo_info).await; + Ok(Response::builder().body(Body::from("success")).unwrap()) +} + +pub async fn repo_folk( + ztm_agent_port: u16, + params: GetParams, +) -> Result, (StatusCode, String)> { + tracing::info!("params:{:?}", params); + let identifier = match params.identifier.clone() { + Some(i) => i, + None => { + return Err(( + StatusCode::BAD_REQUEST, + String::from("Identifier not provide\n"), + )); + } + }; + let port = match params.port { + Some(i) => i, + None => { + return Err((StatusCode::BAD_REQUEST, String::from("Port not provide\n"))); + } + }; + let remote_peer_id = match get_peer_id_from_identifier(identifier.clone()) { + Ok(p) => p, + Err(e) => return Err((StatusCode::BAD_REQUEST, e)), + }; + let remote_port = match get_remote_port_from_identifier(identifier.clone()) { + Ok(p) => p, + Err(e) => return Err((StatusCode::BAD_REQUEST, e)), + }; + let git_path = match get_git_path_from_identifier(identifier) { + Ok(p) => p, + Err(e) => return Err((StatusCode::BAD_REQUEST, e)), + }; + + let agent: LocalZTMAgent = LocalZTMAgent { + agent_port: ztm_agent_port, + }; + let local_ep = match agent.get_ztm_local_endpoint().await { + Ok(ep) => ep, + Err(e) => return Err((StatusCode::INTERNAL_SERVER_ERROR, e)), + }; + + let remote_ep = match agent.get_ztm_remote_endpoint(remote_peer_id.clone()).await { + Ok(ep) => ep, + Err(e) => return Err((StatusCode::INTERNAL_SERVER_ERROR, e)), + }; + + let (peer_id, _) = vault::init(); + let bound_name = format!( + "{}_{}", + get_short_peer_id(peer_id), + get_short_peer_id(remote_peer_id) + ); + //creata inbound + match agent + .create_ztm_app_tunnel_inbound( + local_ep.id, + "ztm".to_string(), + "tunnel".to_string(), + bound_name.clone(), + port, + ) + .await + { + Ok(_) => (), + Err(s) => { + tracing::error!("create app inbound, {s}"); + return Err((StatusCode::INTERNAL_SERVER_ERROR, s)); + } + } + tracing::info!("create app inbound successfully"); + + //creata outbound + match agent + .create_ztm_app_tunnel_outbound( + remote_ep.id, + "ztm".to_string(), + "tunnel".to_string(), + bound_name, + remote_port, + ) + .await + { + Ok(msg) => { + tracing::info!("create app outbound successfully,{}", msg); + } + Err(s) => { + tracing::error!("create app outbound, {s}"); + return Err((StatusCode::INTERNAL_SERVER_ERROR, s)); + } + } + let msg = format!("Success, you can try to clone the repo like this:\ngit clone http://localhost:{port}/{git_path}"); + Ok(Response::builder().body(Body::from(msg)).unwrap()) +} + +pub fn get_peer_id_from_identifier(identifier: String) -> Result { + // p2p://mrJ46F8gd2sa2Dx3iCYf6DauJ2WpAaepus7PwyZVebgD/8000/third-part/mega_143.git + let words: Vec<&str> = identifier.split('/').collect(); + if words.len() <= 2 { + return Err("invalid identifier".to_string()); + } + return Ok(words.get(2).unwrap().to_string()); +} + +pub fn get_remote_port_from_identifier(identifier: String) -> Result { + // p2p://mrJ46F8gd2sa2Dx3iCYf6DauJ2WpAaepus7PwyZVebgD/8000/third-part/mega_143.git + let words: Vec<&str> = identifier.split('/').collect(); + if words.len() <= 3 { + return Err("invalid identifier".to_string()); + } + match words.get(3).unwrap().parse::() { + Ok(number) => Ok(number), + Err(e) => Err(e.to_string()), + } +} + +pub fn get_git_path_from_identifier(identifier: String) -> Result { + // p2p://mrJ46F8gd2sa2Dx3iCYf6DauJ2WpAaepus7PwyZVebgD/8000/third-part/mega_143.git + let words: Vec<&str> = identifier.split('/').collect(); + if words.len() <= 4 { + return Err("invalid identifier".to_string()); + } + let path = words[4..].join("/"); + Ok(path) +} + +pub fn get_short_peer_id(peer_id: String) -> String { + if peer_id.len() <= 7 { + return peer_id; + } + peer_id[0..7].to_string() +} + #[cfg(test)] mod tests {} diff --git a/gemini/src/lib.rs b/gemini/src/lib.rs index 37171820d..a79159340 100644 --- a/gemini/src/lib.rs +++ b/gemini/src/lib.rs @@ -4,6 +4,7 @@ use callisto::{ztm_node, ztm_repo_info}; use chrono::Utc; use serde::{Deserialize, Serialize}; +pub mod ca; pub mod http; pub mod ztm; @@ -100,6 +101,22 @@ impl TryFrom for ztm_node::Model { } } +impl TryFrom for ztm_node::Model { + type Error = ConversionError; + + fn try_from(n: Node) -> Result { + Ok(ztm_node::Model { + peer_id: n.peer_id, + hub: n.hub, + agent_name: n.agent_name, + service_name: n.service_name, + r#type: n.mega_type, + online: n.online, + last_online_time: n.last_online_time, + }) + } +} + impl From for Node { fn from(n: ztm_node::Model) -> Self { Node { @@ -121,6 +138,7 @@ pub struct RepoInfo { pub origin: String, pub update_time: i64, pub commit: String, + pub peer_online: bool, } impl From for ztm_repo_info::Model { @@ -143,6 +161,7 @@ impl From for RepoInfo { origin: r.origin, update_time: r.update_time, commit: r.commit, + peer_online: false, } } } diff --git a/gemini/src/ztm/agent.rs b/gemini/src/ztm/agent.rs new file mode 100644 index 000000000..df5f84c0b --- /dev/null +++ b/gemini/src/ztm/agent.rs @@ -0,0 +1,568 @@ +use std::{ + collections::HashMap, + fs::{self, remove_dir_all}, + path::Path, + time::Duration, +}; + +use axum::async_trait; +use common::config::Config; +use jupiter::context::Context; +use reqwest::{header::CONTENT_TYPE, Client}; +use serde::{Deserialize, Serialize}; + +use crate::RepoInfo; + +use super::{handle_ztm_response, hub::ZTMUserPermit}; + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMMesh { + pub name: String, + pub ca: String, + pub agent: Agent, + pub bootstraps: Vec, + pub connected: bool, + pub errors: Vec, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct Agent { + pub id: String, + pub name: String, + pub username: String, + pub certificate: String, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMEndPoint { + pub id: String, + pub name: String, + pub username: String, + pub online: bool, + #[serde(rename = "isLocal")] + pub is_local: bool, +} +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMServiceReq { + pub host: String, + pub port: u16, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMPortReq { + pub target: ZTMPortService, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMPortService { + pub service: String, +} +#[async_trait] +pub trait ZTMAgent { + async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result; + async fn get_ztm_endpoints(&self) -> Result, String>; + async fn get_ztm_local_endpoint(&self) -> Result; + async fn get_ztm_remote_endpoint(&self, peer_id: String) -> Result; + async fn create_ztm_service( + &self, + ep_id: String, + service_name: String, + port: u16, + ) -> Result; + async fn create_ztm_port( + &self, + ep_id: String, + service_name: String, + port: u16, + ) -> Result; + + async fn start_ztm_app( + &self, + ep_id: String, + provider: String, + app_name: String, + ) -> Result; + + async fn create_ztm_app_tunnel_inbound( + &self, + ep_id: String, + provider: String, + app_name: String, + bound_name: String, + port: u16, + ) -> Result; + + async fn create_ztm_app_tunnel_outbound( + &self, + ep_id: String, + provider: String, + app_name: String, + bound_name: String, + port: u16, + ) -> Result; +} + +#[derive(Debug, Clone)] +pub struct LocalZTMAgent { + pub agent_port: u16, +} + +impl LocalZTMAgent { + pub fn start_ztm_agent(self) { + tokio::spawn(async move { + // neptune::start_agent("ztm_agent_db", self.agent_port); + let db_path = "ztm_agent_db"; + let path = Path::new(db_path); + if fs::metadata(path).is_ok() { + remove_dir_all(path).unwrap(); + } + + neptune::start_agent(db_path, self.agent_port); + }); + } +} + +const MESH_NAME: &str = "relay_mesh"; + +#[async_trait] +impl ZTMAgent for LocalZTMAgent { + async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result { + // POST {agent}/api/meshes/${meshName} + let permit_string = serde_json::to_string(&permit).unwrap(); + // tracing::info!("permit_string:{permit_string}"); + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!("{agent_address}/api/meshes/{MESH_NAME}"); + let client = Client::new(); + let request_result = client + .post(url) + .header(CONTENT_TYPE, "application/json") + .body(permit_string) + .send() + .await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + + let mesh: ZTMMesh = match serde_json::from_slice(response_text.as_bytes()) { + Ok(p) => p, + Err(e) => { + return Err(e.to_string()); + } + }; + Ok(mesh) + } + + async fn get_ztm_endpoints(&self) -> Result, String> { + //GET localhost:7777/api/meshes/{MESH_NAME}/endpoints + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!("{agent_address}/api/meshes/{MESH_NAME}/endpoints"); + let request_result = reqwest::get(url).await; + let eps: String = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + let ep_list: Vec = match serde_json::from_slice(eps.as_bytes()) { + Ok(p) => p, + Err(e) => { + return Err(e.to_string()); + } + }; + Ok(ep_list) + } + + async fn get_ztm_local_endpoint(&self) -> Result { + match self.get_ztm_endpoints().await { + Ok(ep_list) => { + for ele in ep_list { + if ele.is_local { + return Ok(ele); + } + } + return Err("Can not find local ztm endpoint".to_string()); + } + Err(e) => Err(e), + } + } + + async fn get_ztm_remote_endpoint(&self, peer_id: String) -> Result { + match self.get_ztm_endpoints().await { + Ok(ep_list) => { + for ele in ep_list { + if ele.online && ele.name == peer_id { + return Ok(ele); + } + } + return Err("Can not find remote ztm endpoint".to_string()); + } + Err(e) => Err(e), + } + } + + async fn create_ztm_service( + &self, + ep_id: String, + service_name: String, + port: u16, + ) -> Result { + // create a ZTM service + // POST {agent}/api/meshes/${mesh.name}/endpoints/${ep.id}/services/${svcName} + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!( + "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/services/tcp/{service_name}" + ); + let client = Client::new(); + let req = ZTMServiceReq { + host: "127.0.0.1".to_string(), + port, + }; + let req_string = serde_json::to_string(&req).unwrap(); + let request_result = client + .post(url) + .header(CONTENT_TYPE, "application/json") + .body(req_string) + .send() + .await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + Ok(response_text) + } + + async fn create_ztm_port( + &self, + ep_id: String, + service_name: String, + port: u16, + ) -> Result { + //POST {agent}/api/meshes/${mesh.name}/endpoints/${ep.id}/ports/127.0.0.1/tcp/{port} + // request body {"service":service_name} + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!( + "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/ports/127.0.0.1/tcp/{port}" + ); + let client = Client::new(); + let req = ZTMPortReq { + target: ZTMPortService { + service: service_name, + }, + }; + let req_string = serde_json::to_string(&req).unwrap(); + let request_result = client + .post(url) + .header(CONTENT_TYPE, "application/json") + .body(req_string) + .send() + .await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + Ok(response_text) + } + + async fn start_ztm_app( + &self, + ep_id: String, + provider: String, + app_name: String, + ) -> Result { + //POST /api/meshes/${mesh.name}/endpoints/${ep.id}/apps/${provider}/${name} + // request body { isRunning: true } + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!( + "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/apps/{provider}/{app_name}" + ); + let client = Client::new(); + let req = r#"{"isRunning":true}"#; + let request_result = client + .post(url) + .header(CONTENT_TYPE, "application/json") + .body(req) + .send() + .await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + Ok(response_text) + } + + async fn create_ztm_app_tunnel_inbound( + &self, + ep_id: String, + provider: String, + app_name: String, + bound_name: String, + port: u16, + ) -> Result { + //POST /api/meshes/{mesh.name}/apps/${provider}/${name}/api/endpoints/{ep}/inbound/{proto}/{name} + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!( + "{agent_address}/api/meshes/{MESH_NAME}/apps/{provider}/{app_name}/api/endpoints/{ep_id}/inbound/tcp/{bound_name}" + ); + let client = Client::new(); + let req = format!(r#"{{"listens": [{{"ip":"127.0.0.1","port":{port}}}]}}"#); + // let req = r#"{"listens":[{"ip":"127.0.0.1","port":8081}]}"#; + let request_result = client + .post(url) + .header(CONTENT_TYPE, "application/json") + .body(req) + .send() + .await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + Ok(response_text) + } + + async fn create_ztm_app_tunnel_outbound( + &self, + ep_id: String, + provider: String, + app_name: String, + bound_name: String, + port: u16, + ) -> Result { + //POST /api/meshes/{mesh.name}/apps/${provider}/${name}/api/endpoints/{ep}/outbound/{proto}/{name} + let agent_port = self.agent_port; + let agent_address = format!("http://127.0.0.1:{agent_port}"); + let url = format!( + "{agent_address}/api/meshes/{MESH_NAME}/apps/{provider}/{app_name}/api/endpoints/{ep_id}/outbound/tcp/{bound_name}" + ); + let client = Client::new(); + let req = format!(r#"{{"targets": [{{"host":"127.0.0.1","port":{port}}}]}}"#); + // let req = r#"{"targets":[{"host":"127.0.0.1","port":{}}]}"#; + let request_result = client + .post(url) + .header(CONTENT_TYPE, "application/json") + .body(req) + .send() + .await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + Ok(response_text) + } +} +pub async fn run_ztm_client( + bootstrap_node: String, + config: Config, + peer_id: String, + agent: LocalZTMAgent, +) { + let name = peer_id.clone(); + let _context = Context::new(config.clone()).await; + + // let local_permit_option = read_secret(peer_id.as_str()).unwrap(); + // let permit: ZTMUserPermit = match local_permit_option { + // Some(res) => { + // let p = ZTMUserPermit::from_json_map(res.data.unwrap()); + // tracing::info!("read ztm permit file from vault:{:?}", p); + // p + // } + // None => { + // //generate permit from bootstrap_node + // // 1. to get permit json from bootstrap_node + // // GET {bootstrap_node}/api/v1/certificate?name={name} + // let url = format!("{bootstrap_node}/api/v1/certificate?name={name}"); + // let request_result = reqwest::get(url).await; + // let response_text = match handle_ztm_response(request_result).await { + // Ok(s) => s, + // Err(s) => { + // tracing::error!( + // "GET {bootstrap_node}/api/v1/certificate?name={name} failed,{s}" + // ); + // return; + // } + // }; + // let permit: ZTMUserPermit = match serde_json::from_slice(response_text.as_bytes()) { + // Ok(p) => p, + // Err(e) => { + // tracing::error!("{}", e); + // return; + // } + // }; + // //save to vault + // write_secret(peer_id.clone().as_str(), Some(permit.to_json_map().clone())).unwrap(); + // permit + // } + // }; + + let url = format!("{bootstrap_node}/api/v1/certificate?name={name}"); + let request_result = reqwest::get(url).await; + let response_text = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + tracing::error!("GET {bootstrap_node}/api/v1/certificate?name={name} failed,{s}"); + return; + } + }; + let permit: ZTMUserPermit = match serde_json::from_slice(response_text.as_bytes()) { + Ok(p) => p, + Err(e) => { + tracing::error!("{}", e); + return; + } + }; + + // 2. join ztm mesh + let mesh = match agent.connect_ztm_hub(permit.clone()).await { + Ok(m) => m, + Err(s) => { + tracing::error!("join ztm mesh failed"); + tracing::error!(s); + return; + } + }; + tracing::info!("connect to ztm hub successfully"); + + // 3. start tunnel app + // /api/meshes/${mesh.name}/endpoints/${ep.id}/apps/${provider}/${name} + match agent + .start_ztm_app( + mesh.clone().agent.id, + "ztm".to_string(), + "tunnel".to_string(), + ) + .await + { + Ok(_) => (), + Err(s) => { + tracing::error!("start tunnel app failed, {s}"); + return; + } + } + tracing::info!("start tunnel app successfully"); + + // let ep_list = match agent.get_ztm_endpoints().await { + // Ok(eps) => { + // tracing::info!("eps:{:?}", eps); + // eps + // } + // Err(s) => { + // tracing::error!("get_ztm_endpoints, {s}"); + // return; + // } + // }; + + // //creata inbound + // match agent + // .create_ztm_app_tunnel_inbound( + // mesh.clone().agent.id, + // "ztm".to_string(), + // "tunnel".to_string(), + // "test".to_string(), + // ) + // .await + // { + // Ok(_) => (), + // Err(s) => { + // tracing::error!("create app inbound, {s}"); + // return; + // } + // } + // tracing::info!("create app inbound successfully"); + + // //creata outbound + // for ep in ep_list { + // if ep.online && ep.name == "ep-2" { + // tracing::info!("ep-2:{}", ep.id); + // match agent + // .create_ztm_app_tunnel_outbound( + // ep.id, + // "ztm".to_string(), + // "tunnel".to_string(), + // "test".to_string(), + // 8080, + // ) + // .await + // { + // Ok(msg) => { + // tracing::info!("create app outbound successfully,{}", msg); + // } + // Err(s) => { + // tracing::error!("create app outbound, {s}"); + // return; + // } + // } + // break; + // } + // } + + // ping relay + let peer_id_clone = peer_id.clone(); + let bootstrap_node_clone = bootstrap_node.clone(); + let mut interval = tokio::time::interval(Duration::from_secs(45)); + let url = format!("{bootstrap_node_clone}/api/v1/ping"); + loop { + ping( + url.clone(), + peer_id_clone.clone(), + permit.bootstraps.first().unwrap().to_string(), + ) + .await; + interval.tick().await; + } +} + +pub async fn ping(url: String, peer_id: String, hub: String) { + let mut params = HashMap::new(); + params.insert("peer_id", peer_id.clone()); + params.insert("hub", hub); + params.insert("agent_name", peer_id.clone()); + params.insert("service_name", peer_id.clone()); + let client = reqwest::Client::new(); + let response = client.get(url.clone()).query(¶ms).send().await; + let response_text = match handle_ztm_response(response).await { + Ok(s) => s, + Err(s) => { + tracing::error!("GET {url} failed,{s}"); + return; + } + }; + tracing::info!("Get {url}, response: {response_text}"); +} + +pub async fn share_repo(url: String, repo_info: RepoInfo) { + let client = Client::new(); + tracing::info!("share_repo ,{:?}", repo_info); + let req_string = serde_json::to_string(&repo_info).unwrap(); + let response = client + .post(url.clone()) + .header(CONTENT_TYPE, "application/json") + .body(req_string) + .send() + .await; + let response_text = match handle_ztm_response(response).await { + Ok(s) => s, + Err(s) => { + tracing::error!("POST {} failed,{}", url.clone(), s.clone()); + return; + } + }; + tracing::info!("POST {}, response: {}", url.clone(), response_text); +} diff --git a/gemini/src/ztm/hub.rs b/gemini/src/ztm/hub.rs new file mode 100644 index 000000000..7f5c9a97d --- /dev/null +++ b/gemini/src/ztm/hub.rs @@ -0,0 +1,122 @@ +use ::serde::{Deserialize, Serialize}; +use axum::async_trait; +use reqwest::Client; +use serde_json::Value; + +use crate::ztm::handle_ztm_response; + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct CertAgent { + pub name: String, + pub certificate: String, + #[serde(rename = "privateKey")] + pub private_key: String, +} + +#[derive(Debug, Clone)] +pub struct LocalZTMHub { + pub hub_port: u16, + pub ca: String, + pub name: Vec, +} + +impl LocalZTMHub { + pub fn start_ztm_hub(self) { + tokio::spawn(async move { + neptune::start_hub(self.hub_port, self.name, &self.ca); + }); + } +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMUserPermit { + pub ca: String, + pub agent: CertAgent, + pub bootstraps: Vec, +} + +impl ZTMUserPermit { + pub fn to_json_map(&self) -> serde_json::Map { + let value = serde_json::to_value(self.clone()).unwrap(); + serde_json::from_str(&value.to_string()).unwrap() + } + + pub fn from_json_map(map: serde_json::Map) -> ZTMUserPermit { + let permit: ZTMUserPermit = serde_json::from_value(Value::Object(map)).unwrap(); + permit + } +} + +#[async_trait] +pub trait ZTMCA { + async fn create_ztm_certificate(&self, name: String) -> Result; + async fn delete_ztm_certificate(&self, name: String) -> Result; +} + +pub struct LocalHub { + pub host: String, + pub hub_port: u16, + pub ca_port: u16, +} + +#[async_trait] +impl ZTMCA for LocalHub { + async fn create_ztm_certificate(&self, name: String) -> Result { + let ca_port = self.ca_port; + let ca_address = format!("http://localhost:{ca_port}"); + + //1. GET {ca}/api/certificates/ca -> ca certificate + let url = format!("{ca_address}/api/certificates/ca"); + let request_result = reqwest::get(url).await; + let ca_certificate = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + + //2. POST {ca}/api/certificates/{username} -> user private key + let url = format!("{ca_address}/api/certificates/{name}"); + let client = Client::new(); + let request_result = client.post(url).send().await; + let user_key = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + + //3. GET {ca}/api/certificates/{username} -> user certificate + let url = format!("{ca_address}/api/certificates/{name}"); + let request_result = reqwest::get(url).await; + let user_certificate = match handle_ztm_response(request_result).await { + Ok(s) => s, + Err(s) => { + return Err(s); + } + }; + + // Combine those into a json permit + let agent = CertAgent { + name: name.clone(), + certificate: user_certificate.clone(), + private_key: user_key.clone(), + }; + + let hub_address = format!("{}:{}", self.host, self.hub_port); + let permit = ZTMUserPermit { + ca: ca_certificate.clone(), + agent, + bootstraps: vec![hub_address], + }; + + let permit_json = serde_json::to_string(&permit).unwrap(); + tracing::info!("new permit [{name}]: {permit_json}"); + + Ok(permit) + } + + async fn delete_ztm_certificate(&self, _name: String) -> Result { + return Err("not allowed".to_string()); + } +} diff --git a/gemini/src/ztm/mod.rs b/gemini/src/ztm/mod.rs old mode 100644 new mode 100755 index e98a422ee..d99f6db9d --- a/gemini/src/ztm/mod.rs +++ b/gemini/src/ztm/mod.rs @@ -1,464 +1,5 @@ -use std::{collections::HashMap, thread::sleep, time::Duration}; - -use axum::async_trait; -use common::config::{Config, ZTMConfig}; -use reqwest::{header::CONTENT_TYPE, Client}; -use serde::{Deserialize, Serialize}; - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ZTMUserPermit { - pub ca: String, - pub agent: CertAgent, - pub bootstraps: Vec, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct CertAgent { - pub name: String, - pub certificate: String, - #[serde(rename = "privateKey")] - pub private_key: String, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ZTMMesh { - pub name: String, - pub ca: String, - pub agent: Agent, - pub bootstraps: Vec, - pub connected: bool, - pub errors: Vec, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct Agent { - pub id: String, - pub name: String, - pub username: String, - pub certificate: String, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ZTMEndPoint { - pub id: String, - pub name: String, - pub username: String, - pub online: bool, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ZTMServiceReq { - pub host: String, - pub port: u16, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ZTMPortReq { - pub target: ZTMPortService, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ZTMPortService { - pub service: String, -} - -const MESH_NAME: &str = "relay_mesh"; - -#[async_trait] -pub trait ZTMCA { - async fn create_ztm_certificate(&self, name: String) -> Result; - async fn delete_ztm_certificate(&self, name: String) -> Result; -} - -#[async_trait] -pub trait ZTMAgent { - async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result; - async fn create_ztm_service( - &self, - ep_id: String, - service_name: String, - port: u16, - ) -> Result; - async fn create_ztm_port( - &self, - ep_id: String, - service_name: String, - port: u16, - ) -> Result; -} - -pub struct RemoteZTM { - pub config: ZTMConfig, -} - -#[async_trait] -impl ZTMCA for RemoteZTM { - async fn create_ztm_certificate(&self, name: String) -> Result { - let ca_address = self.config.ca.clone(); - let hub_address = self.config.hub.clone(); - - //1. GET {ca}/api/certificates/ca -> ca certificate - let url = format!("{ca_address}/api/certificates/ca"); - let request_result = reqwest::get(url).await; - let ca_certificate = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - - //2. POST {ca}/api/certificates/{username} -> user private key - let url = format!("{ca_address}/api/certificates/{name}"); - let client = Client::new(); - let request_result = client.post(url).send().await; - let user_key = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - - //3. GET {ca}/api/certificates/{username} -> user certificate - let url = format!("{ca_address}/api/certificates/{name}"); - let request_result = reqwest::get(url).await; - let user_certificate = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - - // Combine those into a json permit - let agent = CertAgent { - name: name.clone(), - certificate: user_certificate.clone(), - private_key: user_key.clone(), - }; - - let hub_address = hub_address.replace("http://", ""); - let permit = ZTMUserPermit { - ca: ca_certificate.clone(), - agent, - bootstraps: vec![hub_address], - }; - - let permit_json = serde_json::to_string(&permit).unwrap(); - tracing::info!("new permit [{name}]: {permit_json}"); - - Ok(permit) - } - - async fn delete_ztm_certificate(&self, name: String) -> Result { - let ca_address = self.config.ca.clone(); - - //1. DELETE /api/certificates/${username} - let url = format!("{ca_address}/api/certificates/{name}"); - let client = Client::new(); - let request_result = client.delete(url).send().await; - let s: String = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - Ok(s) - } -} - -#[async_trait] -impl ZTMAgent for RemoteZTM { - async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result { - // POST {agent}/api/meshes/${meshName} - let permit_string = serde_json::to_string(&permit).unwrap(); - let agent_address = self.config.agent.clone(); - let url = format!("{agent_address}/api/meshes/{MESH_NAME}"); - let client = Client::new(); - let request_result = client - .post(url) - .header(CONTENT_TYPE, "application/json") - .body(permit_string) - .send() - .await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - - let mesh: ZTMMesh = match serde_json::from_slice(response_text.as_bytes()) { - Ok(p) => p, - Err(e) => { - return Err(e.to_string()); - } - }; - Ok(mesh) - } - - async fn create_ztm_service( - &self, - ep_id: String, - service_name: String, - port: u16, - ) -> Result { - // create a ZTM service - // POST {agent}/api/meshes/${mesh.name}/endpoints/${ep.id}/services/${svcName} - let agent_address = self.config.agent.clone(); - let url = format!( - "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/services/tcp/{service_name}" - ); - let client = Client::new(); - let req = ZTMServiceReq { - host: "127.0.0.1".to_string(), - port, - }; - let req_string = serde_json::to_string(&req).unwrap(); - let request_result = client - .post(url) - .header(CONTENT_TYPE, "application/json") - .body(req_string) - .send() - .await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - Ok(response_text) - } - - async fn create_ztm_port( - &self, - ep_id: String, - service_name: String, - port: u16, - ) -> Result { - //POST {agent}/api/meshes/${mesh.name}/endpoints/${ep.id}/ports/127.0.0.1/tcp/{port} - // request body {"service":service_name} - let agent_address = self.config.agent.clone(); - let url = format!( - "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/ports/127.0.0.1/tcp/{port}" - ); - let client = Client::new(); - let req = ZTMPortReq { - target: ZTMPortService { - service: service_name, - }, - }; - let req_string = serde_json::to_string(&req).unwrap(); - let request_result = client - .post(url) - .header(CONTENT_TYPE, "application/json") - .body(req_string) - .send() - .await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - Ok(response_text) - } -} - -#[derive(Debug, Clone)] -pub struct LocalZTM { - pub agent_port: u16, -} - -impl LocalZTM { - pub fn start_ztm_agent(self) { - tokio::spawn(async move { - neptune::start_agent("ztm_agent.db", self.agent_port); - }); - } -} - -#[async_trait] -impl ZTMAgent for LocalZTM { - async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result { - // POST {agent}/api/meshes/${meshName} - let permit_string = serde_json::to_string(&permit).unwrap(); - tracing::info!("permit_string:{permit_string}"); - let agent_port = self.agent_port; - let agent_address = format!("http://127.0.0.1:{agent_port}"); - let url = format!("{agent_address}/api/meshes/{MESH_NAME}"); - let client = Client::new(); - let request_result = client - .post(url) - .header(CONTENT_TYPE, "application/json") - .body(permit_string) - .send() - .await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - - let mesh: ZTMMesh = match serde_json::from_slice(response_text.as_bytes()) { - Ok(p) => p, - Err(e) => { - return Err(e.to_string()); - } - }; - Ok(mesh) - } - - async fn create_ztm_service( - &self, - ep_id: String, - service_name: String, - port: u16, - ) -> Result { - // create a ZTM service - // POST {agent}/api/meshes/${mesh.name}/endpoints/${ep.id}/services/${svcName} - let agent_port = self.agent_port; - let agent_address = format!("http://127.0.0.1:{agent_port}"); - let url = format!( - "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/services/tcp/{service_name}" - ); - let client = Client::new(); - let req = ZTMServiceReq { - host: "127.0.0.1".to_string(), - port, - }; - let req_string = serde_json::to_string(&req).unwrap(); - let request_result = client - .post(url) - .header(CONTENT_TYPE, "application/json") - .body(req_string) - .send() - .await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - Ok(response_text) - } - - async fn create_ztm_port( - &self, - ep_id: String, - service_name: String, - port: u16, - ) -> Result { - //POST {agent}/api/meshes/${mesh.name}/endpoints/${ep.id}/ports/127.0.0.1/tcp/{port} - // request body {"service":service_name} - let agent_port = self.agent_port; - let agent_address = format!("http://127.0.0.1:{agent_port}"); - let url = format!( - "{agent_address}/api/meshes/{MESH_NAME}/endpoints/{ep_id}/ports/127.0.0.1/tcp/{port}" - ); - let client = Client::new(); - let req = ZTMPortReq { - target: ZTMPortService { - service: service_name, - }, - }; - let req_string = serde_json::to_string(&req).unwrap(); - let request_result = client - .post(url) - .header(CONTENT_TYPE, "application/json") - .body(req_string) - .send() - .await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - return Err(s); - } - }; - Ok(response_text) - } -} - -pub async fn run_ztm_client( - bootstrap_node: String, - _config: Config, - peer_id: String, - agent: LocalZTM, -) { - let name = peer_id.clone(); - // let _context = Context::new(config.clone()).await; - - // 1. to get permit json from bootstrap_node - // GET {bootstrap_node}/api/v1/certificate?name={name} - let url = format!("{bootstrap_node}/api/v1/certificate?name={name}"); - let request_result = reqwest::get(url).await; - let response_text = match handle_ztm_response(request_result).await { - Ok(s) => s, - Err(s) => { - tracing::error!("GET {bootstrap_node}/api/v1/certificate?name={name} failed,{s}"); - return; - } - }; - let permit: ZTMUserPermit = match serde_json::from_slice(response_text.as_bytes()) { - Ok(p) => p, - Err(e) => { - tracing::error!("{}", e); - return; - } - }; - - // 2. join ztm mesh - let mesh = match agent.connect_ztm_hub(permit.clone()).await { - Ok(m) => m, - Err(s) => { - tracing::error!(s); - return; - } - }; - tracing::info!("connect to ztm hub successfully"); - - // 3. create a ztm port for relay - let ztm_port = 8002; - match agent - .create_ztm_port(mesh.agent.id, "relay".to_string(), ztm_port) - .await - { - Ok(_) => (), - Err(s) => { - tracing::error!("create a ztm port failed, {s}"); - return; - } - } - tracing::info!("create a ztm port successfully, port:{ztm_port}"); - let peer_id_clone = peer_id.clone(); - loop { - ping( - peer_id_clone.clone(), - permit.bootstraps.first().unwrap().to_string(), - ztm_port, - ) - .await; - sleep(Duration::from_secs(15)); - } -} - -pub async fn ping(peer_id: String, hub: String, ztm_port: u16) { - let url = format!("http://127.0.0.1:{ztm_port}/ping"); - let mut params = HashMap::new(); - params.insert("peer_id", peer_id.clone()); - params.insert("hub", hub); - params.insert("agent_name", peer_id.clone()); - params.insert("service_name", peer_id.clone()); - let client = reqwest::Client::new(); - let response = client.get(url.clone()).query(¶ms).send().await; - let response_text = match handle_ztm_response(response).await { - Ok(s) => s, - Err(s) => { - tracing::error!("GET {url} failed,{s}"); - return; - } - }; - tracing::info!("Get {url}, response: {response_text}"); -} +pub mod agent; +pub mod hub; pub async fn handle_ztm_response( request_result: Result, diff --git a/jupiter/src/storage/git_db_storage.rs b/jupiter/src/storage/git_db_storage.rs index 08e9e639b..b9498c80f 100644 --- a/jupiter/src/storage/git_db_storage.rs +++ b/jupiter/src/storage/git_db_storage.rs @@ -219,6 +219,16 @@ impl GitDbStorage { Ok(result) } + pub async fn find_git_repo_by_path( + &self, + repo_path: &str, + ) -> Result, MegaError> { + let query = git_repo::Entity::find().filter(git_repo::Column::RepoPath.eq(repo_path)); + tracing::debug!("{}", query.build(DbBackend::Postgres).to_string()); + let result = query.one(self.get_connection()).await?; + Ok(result) + } + pub async fn save_git_repo(&self, repo: Repo) -> Result<(), MegaError> { let model: git_repo::Model = repo.into(); let a_model = model.into_active_model(); diff --git a/neptune/src/lib.rs b/neptune/src/lib.rs index 18b1a2554..e6ba7fa97 100644 --- a/neptune/src/lib.rs +++ b/neptune/src/lib.rs @@ -19,7 +19,8 @@ pub fn start_agent(database: &str, listen_port: u16) { CString::new("ztm-pipy").unwrap(), CString::new("repo://ztm/agent").unwrap(), CString::new("--args").unwrap(), - CString::new(format!("--database={}", database)).unwrap(), + // CString::new(format!("--database={}", database)).unwrap(), + CString::new(format!("--data={}", database)).unwrap(), CString::new(format!("--listen=0.0.0.0:{}", listen_port)).unwrap(), ]; let c_args: Vec<*const c_char> = args.iter().map(|arg| arg.as_ptr()).collect(); From 672d8ebe95201794d5b7637830656dcbb53c372e Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Sat, 20 Jul 2024 14:09:21 +0800 Subject: [PATCH 2/3] update ztm lib --- .gitignore | 1 + gemini/src/http/handler.rs | 6 +- gemini/src/lib.rs | 2 + gemini/src/ztm/agent.rs | 59 +-- neptune/build.rs | 15 + neptune/hub/ca.js | 78 ++++ neptune/hub/cmdline.js | 245 +++++++++++ neptune/hub/db.js | 81 ++++ neptune/hub/main.js | 788 ++++++++++++++++++++++++++++++++++++ neptune/hub/options.js | 50 +++ neptune/mega/tunnel/cli.js | 144 +++---- neptune/mega/tunnel/main.js | 9 + neptune/src/lib.rs | 10 +- 13 files changed, 1329 insertions(+), 159 deletions(-) create mode 100644 neptune/hub/ca.js create mode 100644 neptune/hub/cmdline.js create mode 100644 neptune/hub/db.js create mode 100755 neptune/hub/main.js create mode 100644 neptune/hub/options.js diff --git a/.gitignore b/.gitignore index 219301b08..d5dc7dc5e 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,7 @@ Cargo.lock # zmt agent data file ztm_agent.db +ztm_agent_db* # buck2 out buck-out \ No newline at end of file diff --git a/gemini/src/http/handler.rs b/gemini/src/http/handler.rs index d2f0f0a4a..fea6e738d 100644 --- a/gemini/src/http/handler.rs +++ b/gemini/src/http/handler.rs @@ -6,7 +6,7 @@ use venus::import_repo::repo::Repo; use crate::{ ztm::agent::{share_repo, LocalZTMAgent, ZTMAgent}, - RepoInfo, + RepoInfo, ZTM_APP_PROVIDER, }; pub async fn repo_provide( @@ -129,7 +129,7 @@ pub async fn repo_folk( match agent .create_ztm_app_tunnel_inbound( local_ep.id, - "ztm".to_string(), + ZTM_APP_PROVIDER.to_string(), "tunnel".to_string(), bound_name.clone(), port, @@ -148,7 +148,7 @@ pub async fn repo_folk( match agent .create_ztm_app_tunnel_outbound( remote_ep.id, - "ztm".to_string(), + ZTM_APP_PROVIDER.to_string(), "tunnel".to_string(), bound_name, remote_port, diff --git a/gemini/src/lib.rs b/gemini/src/lib.rs index a79159340..826179cd5 100644 --- a/gemini/src/lib.rs +++ b/gemini/src/lib.rs @@ -8,6 +8,8 @@ pub mod ca; pub mod http; pub mod ztm; +const ZTM_APP_PROVIDER: &str = "mega"; + #[derive(Deserialize, Debug)] pub struct RelayGetParams { pub peer_id: Option, diff --git a/gemini/src/ztm/agent.rs b/gemini/src/ztm/agent.rs index df5f84c0b..46f0c8620 100644 --- a/gemini/src/ztm/agent.rs +++ b/gemini/src/ztm/agent.rs @@ -11,7 +11,7 @@ use jupiter::context::Context; use reqwest::{header::CONTENT_TYPE, Client}; use serde::{Deserialize, Serialize}; -use crate::RepoInfo; +use crate::{RepoInfo, ZTM_APP_PROVIDER}; use super::{handle_ztm_response, hub::ZTMUserPermit}; @@ -445,7 +445,7 @@ pub async fn run_ztm_client( match agent .start_ztm_app( mesh.clone().agent.id, - "ztm".to_string(), + ZTM_APP_PROVIDER.to_string(), "tunnel".to_string(), ) .await @@ -458,61 +458,6 @@ pub async fn run_ztm_client( } tracing::info!("start tunnel app successfully"); - // let ep_list = match agent.get_ztm_endpoints().await { - // Ok(eps) => { - // tracing::info!("eps:{:?}", eps); - // eps - // } - // Err(s) => { - // tracing::error!("get_ztm_endpoints, {s}"); - // return; - // } - // }; - - // //creata inbound - // match agent - // .create_ztm_app_tunnel_inbound( - // mesh.clone().agent.id, - // "ztm".to_string(), - // "tunnel".to_string(), - // "test".to_string(), - // ) - // .await - // { - // Ok(_) => (), - // Err(s) => { - // tracing::error!("create app inbound, {s}"); - // return; - // } - // } - // tracing::info!("create app inbound successfully"); - - // //creata outbound - // for ep in ep_list { - // if ep.online && ep.name == "ep-2" { - // tracing::info!("ep-2:{}", ep.id); - // match agent - // .create_ztm_app_tunnel_outbound( - // ep.id, - // "ztm".to_string(), - // "tunnel".to_string(), - // "test".to_string(), - // 8080, - // ) - // .await - // { - // Ok(msg) => { - // tracing::info!("create app outbound successfully,{}", msg); - // } - // Err(s) => { - // tracing::error!("create app outbound, {s}"); - // return; - // } - // } - // break; - // } - // } - // ping relay let peer_id_clone = peer_id.clone(); let bootstrap_node_clone = bootstrap_node.clone(); diff --git a/neptune/build.rs b/neptune/build.rs index 2ce4a0ab0..cacc5ccf3 100644 --- a/neptune/build.rs +++ b/neptune/build.rs @@ -20,6 +20,20 @@ fn copy_mega_apps() { }); } +/// copy `hub/main.js` to `libs/ztm/hub/main.js` +fn copy_mega_ztm_hub() { + let path = Path::new("hub"); + if fs::metadata(path).is_err() { + println!("neptune/hub does not exist, skip to copy"); + return; + } + let dst = Path::new("libs/ztm/hub"); + copy_dir_all(path, dst).unwrap_or_else(|e| { + fs::remove_dir(dst).expect("failed to remove ztm/hub"); + panic!("failed to copy neptune/hub to neptune/libs/ztm/hub: {}", e); + }); +} + /// use npm to build agent ui in `libs/ztm/agent/gui` fn npm_build_agent_ui() { let ui = Path::new("libs/ztm/agent/gui"); @@ -211,6 +225,7 @@ fn main() { return_if_change(); copy_mega_apps(); + copy_mega_ztm_hub(); let dst = build(); parse_link_args_to_rustc(&dst); copy_lib_to_target(&dst); // optional, didn't work in all cases diff --git a/neptune/hub/ca.js b/neptune/hub/ca.js new file mode 100644 index 000000000..0d64da18b --- /dev/null +++ b/neptune/hub/ca.js @@ -0,0 +1,78 @@ +import db from './db.js' + +var caCert +var caKey +var caURL + +function init(url) { + if (url) { + caURL = new URL(url) + } else { + if (!db.getKey('ca')) { + var key = new crypto.PrivateKey({ type: 'rsa', bits: 2048 }) + var cert = new crypto.Certificate({ + subject: { CN: 'ca' }, + privateKey: key, + publicKey: new crypto.PublicKey(key), + days: 365, + }) + db.setCert('ca', cert.toPEM().toString()) + db.setKey('ca', key.toPEM().toString()) + } + + caCert = new crypto.Certificate(db.getCert('ca')) + caKey = new crypto.PrivateKey(db.getKey('ca')) + } + return Promise.resolve() +} + +function getCertificate(name) { + if (caURL) { + var url = caURL + var ha = new http.Agent(url.host) + return ha.request( + 'GET', os.path.join(url.path, name) + ).then(res => { + var status = res?.head?.status + if (200 <= status && status <= 299) { + return new crypto.Certificate(res.body.toString()) + } + return null + }) + } else { + var pem = db.getCert(name) + var crt = pem ? new crypto.Certificate(pem) : null + return Promise.resolve(crt) + } +} + +function signCertificate(name, pkey) { + if (caURL) { + var url = caURL + var ha = new http.Agent(url.host) + return ha.request( + 'POST', os.path.join(url.path, name), null, pkey.toPEM() + ).then(res => { + var status = res?.head?.status + if (200 <= status && status <= 299) { + return new crypto.Certificate(res.body.toString()) + } + return null + }) + } else { + var crt = new crypto.Certificate({ + subject: { CN: name }, + days: 365, + issuer: caCert, + privateKey: caKey, + publicKey: pkey, + }) + return Promise.resolve(crt) + } +} + +export default { + init, + getCertificate, + signCertificate, +} diff --git a/neptune/hub/cmdline.js b/neptune/hub/cmdline.js new file mode 100644 index 000000000..9f389f65c --- /dev/null +++ b/neptune/hub/cmdline.js @@ -0,0 +1,245 @@ +export default function (argv, { commands, notes, help, fallback }) { + var program = argv[0] + argv = argv.slice(1) + + var isHelpCommand = false + if (argv[0] === 'help') { + isHelpCommand = true + argv.shift() + } + + var patterns = commands.map(cmd => cmd.usage ? tokenize(cmd.usage) : []) + var best = 0 + var bestIndex = -1 + + patterns.forEach((pattern, i) => { + var empty = !pattern.some(t => !t.startsWith('<') && !t.startsWith('[')) + var len = matchCommand(pattern, argv) + if (len > best || empty) { + best = len + bestIndex = i + } + }) + + var cmd = commands[bestIndex] + var pattern = patterns[bestIndex] + + if (isHelpCommand) { + var lines = [''] + if (cmd) { + var usage = program + if (cmd.usage) usage += ' ' + cmd.usage + if (cmd.options) usage += ' [options...]' + if (cmd.title) lines.push(cmd.title, '') + lines.push(`Usage: ${usage}`, '') + if (cmd.options) lines.push(`Options:`, '', stripIndentation(cmd.options, 2), '') + if (cmd.notes) lines.push(stripIndentation(cmd.notes), '') + } else { + var commandList = [] + commands.forEach(cmd => { + if (!cmd.usage) return + var tokens = tokenize(cmd.usage) + var firstArg = tokens.findIndex(t => t.startsWith('<') || t.startsWith('[')) + if (firstArg >= 0) { + commandList.push([ + tokens.slice(0, firstArg).join(' '), + tokens.slice(firstArg).join(' '), + cmd.title || '' + ]) + } else { + commandList.push([tokens.join(' '), '', cmd.title || '']) + } + }) + lines.push(`Commands:`, '') + var cmdWidth = commandList.reduce((max, cmd) => Math.max(max, cmd[0].length), 0) + var argWidth = commandList.reduce((max, cmd) => Math.max(max, cmd[1].length), 0) + commandList.forEach(cmd => { + lines.push(` ${program} ${cmd[0].padEnd(cmdWidth)} ${cmd[1].padEnd(argWidth)} ${cmd[2]}`) + }) + lines.push('') + if (notes) { + lines.push(stripIndentation(notes), '') + } + lines.push(`Type '${program} help ' for detailed info.`, '') + } + if (typeof help === 'function') { + return help(lines.join('\n'), cmd) + } else { + return lines.forEach(l => println(l)) + } + } + + if (!cmd) { + if (typeof fallback === 'function') { + return fallback(argv) + } else { + throw `Unknown command. Type '${program} help' for help info.` + } + } + + try { + var rest = [] + var args = parseCommand(pattern, argv, rest) + var opts = parseOptions(cmd.options, rest) + + } catch (err) { + var command = pattern.slice(0, best).join(' ') + throw `${err}. Type '${program} help ${command}' for help info.` + } + + return cmd.action({ ...args, ...opts }) +} + +function matchCommand(pattern, argv) { + var i = argv.findIndex( + (arg, i) => arg.startsWith('-') || arg !== pattern[i] + ) + return i < 0 ? argv.length : i +} + +function parseCommand(pattern, argv, rest) { + var values = {} + var i = argv.findIndex((arg, i) => { + var tok = pattern[i] + if (arg.startsWith('-')) return true + if (!tok) throw `Excessive positional argument: ${arg}` + if (tok.startsWith('[') || tok.startsWith('<')) { + values[tok] = arg + } + }) + if (i < 0) i = argv.length + argv.slice(i).forEach(arg => rest.push(arg)) + var tok = pattern[i] + if (!tok || tok.startsWith('[')) return values + if (!tok.startsWith('<')) return null + throw `Missing argument ${tok}` +} + +function parseOptions(format, argv) { + var options = {} + + if (format) { + format.split('\n').map( + line => line.trim() + ).filter( + line => line.startsWith('-') + ).forEach(line => { + var aliases = [] + var type + tokenize(line).find(t => { + if (t.startsWith('-')) { + if (t.endsWith(',')) t = t.substring(0, t.length - 1) + aliases.push(t) + } else { + if (t.endsWith('...]') || t.endsWith('...>')) type = 'array' + else if (t.startsWith('[')) type = 'optional string' + else if (t.startsWith('<')) type = 'string' + else type = 'boolean' + return true + } + }) + var option = { type } + aliases.forEach(name => options[name] = option) + }) + } + + var currentOption + + argv.forEach(arg => { + if (currentOption) { + if (arg.startsWith('-')) { + endOption(currentOption) + currentOption = undefined + } else { + addOption(currentOption, arg) + return + } + } + if (arg.startsWith('--')) { + currentOption = arg + } else if (arg.startsWith('-')) { + if (arg.length === 2) { + currentOption = arg + } else { + addOption(arg.substring(0, 2), arg.substring(2)) + } + } else { + throw `Excessive positional argument: ${arg}` + } + }) + + if (currentOption) endOption(currentOption) + + function addOption(name, value) { + var option = options[name] + if (!option) throw `Unrecognized option: ${name}` + switch (option.type) { + case 'boolean': + throw `Excessive positional argument: ${value}` + case 'string': + case 'optional string': + if ('value' in option) throw `Duplicated option value: ${value}` + option.value = value + break + case 'array': + option.value ??= [] + option.value.push(value) + break + } + } + + function endOption(name) { + var option = options[name] + if (!option) throw `Unrecognized option: ${name}` + switch (option.type) { + case 'boolean': + option.value = true + break + case 'string': + if (!('value' in option)) throw `Missing option value: ${name}` + break + case 'optional string': + if (!('value' in option)) option.value = '' + break + } + } + + var values = {} + + Object.entries(options).forEach( + ([name, option]) => { + if ('value' in option) { + values[name] = option.value + } + } + ) + + return values +} + +function tokenize(str) { + var tokens = str.split(' ').reduce( + (a, b) => { + if (typeof a === 'string') a = [a] + var last = a.pop() + if (last.startsWith('<')) { + a.push(`${last} ${b}`) + } else if (last.startsWith('[')) { + a.push(`${last} ${b}`) + } else { + a.push(last, b) + } + if (b.endsWith('>') || b.endsWith(']')) a.push('') + return a + } + ) + return tokens instanceof Array ? tokens.filter(t => t) : [tokens] +} + +function stripIndentation(s, indent) { + var lines = s.split('\n') + if (lines[0].trim() === '') lines.shift() + var depth = lines[0].length - lines[0].trimStart().length + var padding = ' '.repeat(indent || 0) + return lines.map(l => padding + l.substring(depth)).join('\n') +} diff --git a/neptune/hub/db.js b/neptune/hub/db.js new file mode 100644 index 000000000..61e6c354d --- /dev/null +++ b/neptune/hub/db.js @@ -0,0 +1,81 @@ +var db = null + +function open(pathname) { + db = sqlite(pathname) + + db.exec(` + CREATE TABLE IF NOT EXISTS certificates ( + name TEXT PRIMARY KEY, + data TEXT NOT NULL + ) + `) + + db.exec(` + CREATE TABLE IF NOT EXISTS keys ( + name TEXT PRIMARY KEY, + data TEXT NOT NULL + ) + `) +} + +function getCert(name) { + return db.sql(`SELECT data FROM certificates WHERE name = ?`) + .bind(1, name) + .exec()[0]?.data +} + +function setCert(name, data) { + if (getKey(name)) { + db.sql(`UPDATE certificates SET data = ? WHERE name = ?`) + .bind(1, data) + .bind(2, name) + .exec() + } else { + db.sql(`INSERT INTO certificates(name, data) VALUES(?, ?)`) + .bind(1, name) + .bind(2, data) + .exec() + } +} + +function delCert(name) { + db.sql(`DELETE FROM certificates WHERE name = ?`) + .bind(1, name) + .exec() +} + +function getKey(name) { + return db.sql(`SELECT data FROM keys WHERE name = ?`) + .bind(1, name) + .exec()[0]?.data +} + +function setKey(name, data) { + if (getKey(name)) { + db.sql(`UPDATE keys SET data = ? WHERE name = ?`) + .bind(1, data) + .bind(2, name) + .exec() + } else { + db.sql(`INSERT INTO keys(name, data) VALUES(?, ?)`) + .bind(1, name) + .bind(2, data) + .exec() + } +} + +function delKey(name) { + db.sql(`DELETE FROM keys WHERE name = ?`) + .bind(1, name) + .exec() +} + +export default { + open, + getCert, + setCert, + delCert, + getKey, + setKey, + delKey, +} diff --git a/neptune/hub/main.js b/neptune/hub/main.js new file mode 100755 index 000000000..d6f4e2c15 --- /dev/null +++ b/neptune/hub/main.js @@ -0,0 +1,788 @@ +#!/usr/bin/env -S pipy --args + +import db from './db.js' +import ca from './ca.js' +import cmdline from './cmdline.js' + +var routes = Object.entries({ + + '/api/status': { + 'POST': () => findCurrentEndpointSession() ? postStatus : noSession, + }, + + '/api/sign/{name}': { + 'POST': () => signCertificate, + }, + + '/api/endpoints': { + 'GET': () => getEndpoints, + }, + + '/api/endpoints/{ep}': { + 'GET': () => getEndpoint, + 'CONNECT': () => connectEndpoint, + }, + + '/api/endpoints/{ep}/file-data/{hash}': { + 'GET': () => getFileData, + }, + + '/api/endpoints/{ep}/services': { + 'GET': () => getServices, + }, + + '/api/endpoints/{ep}/apps/{app}': { + 'CONNECT': () => connectApp, + }, + + '/api/endpoints/{ep}/apps/{provider}/{app}': { + 'CONNECT': () => connectApp, + }, + + '/api/endpoints/{ep}/services/{proto}/{svc}': { + 'CONNECT': () => connectService, + }, + + '/api/filesystem': { + 'GET': () => getFilesystem, + 'POST': () => findCurrentEndpointSession() ? postFilesystem : noSession, + }, + + '/api/filesystem/*': { + 'GET': () => getFileInfo, + }, + + '/api/apps': { + 'POST': () => findCurrentEndpointSession() ? postAppStates : noSession, + }, + + '/api/apps/{app}': { + 'GET': () => getAppState, + }, + + '/api/apps/{provider}/{app}': { + 'GET': () => getAppState, + }, + + '/api/services': { + 'GET': () => getServices, + 'POST': () => findCurrentEndpointSession() ? postServices : noSession, + }, + + '/api/services/{proto}/{svc}': { + 'GET': () => getService, + }, + + '/api/forward/{ep}/*': { + 'GET': () => forwardRequest, + 'POST': () => forwardRequest, + 'DELETE': () => forwardRequest, + }, + +}).map( + function ([path, methods]) { + var match = new http.Match(path) + var handler = function (params, req) { + var f = methods[req.head.method] + if (f) return f(params, req) + return notSupported + } + return { match, handler } + } +) + +var endpoints = {} +var sessions = {} + +var caCert = null +var myCert = null +var myKey = null +var myNames = [] + +function main() { + return cmdline(pipy.argv, { + commands: [{ + title: 'ZTM Hub Service', + options: ` + -d, --data Specify the location of ZTM storage (default: ~/.ztm) + -l, --listen Specify the service listening port (default: 0.0.0.0:8888) + -n, --names Specify one or more hub names (host:port) that are accessible to agents + --ca Specify the location of an external CA service if any + `, + action: (args) => { + var dbPath = args['--data'] || '~/.ztm' + if (dbPath.startsWith('~/')) { + dbPath = os.home() + dbPath.substring(1) + } + + try { + dbPath = os.path.resolve(dbPath) + var st = os.stat(dbPath) + if (st) { + if (!st.isDirectory()) { + throw `directory path already exists as a regular file: ${dbPath}` + } + } else { + os.mkdir(dbPath, { recursive: true }) + } + + db.open(os.path.join(dbPath, 'ztm-hub.db')) + + } catch (e) { + if (e.stack) println(e.stack) + println('ztm:', e.toString()) + return Promise.reject() + } + + myNames = args['--names'] || [] + + return ca.init(args['--ca']).then(() => { + myKey = new crypto.PrivateKey({ type: 'rsa', bits: 2048 }) + var pkey = new crypto.PublicKey(myKey) + return Promise.all([ + ca.getCertificate('ca').then(crt => caCert = crt), + ca.signCertificate('hub', pkey).then(crt => myCert = crt), + ]) + }).then(() => { + start(args['--listen'] || '0.0.0.0:8888') + }) + } + }], + help: text => Promise.reject(println(text)) + }) +} + +try { + main().catch(error) +} catch (err) { error(err) } + +function error(err) { + if (err) { + if (err.stack) println(err.stack) + println(`ztm: ${err.toString()}`) + } + pipy.exit(-1) +} + +function endpointName(id) { + var ep = endpoints[id] + return ep?.name ? `${ep.name} (uuid = ${id})` : id +} + +function isEndpointOnline(ep) { + if (!sessions[ep.id]?.size) return false + if (ep.heartbeat + 30*1000 < Date.now()) return false + return true +} + +function isEndpointOutdated(ep) { + return (ep.heartbeat + 120*1000 < Date.now()) +} + +var $ctx = null +var $params = null +var $endpoint = null +var $hub = null +var $hubSelected = null +var $pingID + +function start(listen) { + + + var username = "hub0" + var caAgent = new http.Agent('localhost:9999') + caAgent.request('GET', '/api/certificates/ca').then( + function (res) { + if (res.head.status !== 200) { + println('cannot retreive the CA certificate') + pipy.exit(-1) + return + } + caCert = new crypto.Certificate(res.body) + println('==============') + println('CA certificate') + println('==============') + println(res.body.toString()) + println(caCert) + + // myKey = new crypto.PrivateKey({ type: 'rsa', bits: 2048 }) + // var pkey = new crypto.PublicKey(myKey) + return caAgent.request('POST', `/api/certificates/${username}`, null, "") + } + ).then( + function (res) { + if (res.head.status !== 200) { + println('error signing a hub certificate') + pipy.exit(-1) + return + } + myKey = new crypto.PrivateKey(res.body.toString()) + println('===============') + println('Hub Key') + println('===============') + println(res.body.toString()) + // myKey = res.body.toString(); + // myKey = new crypto.PrivateKey({ type: 'rsa', bits: 2048 }) + return caAgent.request('GET', `/api/certificates/${username}`) + } + ).then( + function (res) { + if (res.head.status !== 200) { + println('error signing a hub certificate') + pipy.exit(-1) + return + } + myCert = new crypto.Certificate(res.body) + println('===============') + println('Hub certificate') + println('===============') + println(res.body.toString()) + + pipy.listen(listen, $=>$ + .onStart( + function (conn) { + $ctx = { + ip: conn.remoteAddress, + port: conn.remotePort, + via: `${conn.localAddress}:${conn.localPort}`, + } + } + ) + .acceptTLS({ + certificate: { + cert: myCert, + key: myKey, + }, + trusted: [caCert], + onState: (tls) => { + if (tls.state === 'connected') { + $ctx.username = tls.peer?.subject?.commonName + } + } + }).to($=>$ + .demuxHTTP().to($=>$ + .pipe( + function (evt) { + if (evt instanceof MessageStart) { + var path = evt.head.path + var route = routes.find(r => Boolean($params = r.match(path))) + if (route) return route.handler($params, evt) + return notFound + } + } + ) + ) + ) + ) + + console.info('Hub started at', listen) + + } + ) + + + function clearOutdatedEndpoints() { + Object.values(endpoints).filter(ep => isEndpointOutdated(ep)).forEach( + (ep) => { + console.info(`Endpoint ${ep.name} (uuid = ${ep.id}) outdated`) + if (sessions[ep.id]?.size === 0) { + delete endpoints[ep.id] + } + } + ) + new Timeout(60).wait().then(clearOutdatedEndpoints) + } + + clearOutdatedEndpoints() +} + +var postStatus = pipeline($=>$ + .replaceMessage( + function (req) { + var info = JSON.decode(req.body) + Object.assign( + $endpoint, { + name: info.name, + heartbeat: Date.now(), + } + ) + return new Message({ status: 201 }) + } + ) +) + +var signCertificate = pipeline($=>$ + .replaceMessage( + function (req) { + var user = $ctx.username + var name = $params.name + if (name !== user && user !== 'root') return response(403) + var pkey = new crypto.PublicKey(req.body) + return ca.signCertificate(name, pkey).then( + cert => response(201, cert.toPEM().toString()) + ) + } + ) +) + +var getEndpoints = pipeline($=>$ + .replaceData() + .replaceMessage( + () => response(200, Object.values(endpoints).map( + ep => ({ + id: ep.id, + name: ep.name, + username: ep.username, + ip: ep.ip, + port: ep.port, + heartbeat: ep.heartbeat, + online: isEndpointOnline(ep), + }) + )) + ) +) + +var getEndpoint = pipeline($=>$ + .replaceData() + .replaceMessage( + function () { + var ep = endpoints[$params.ep] + if (!ep) return response(404) + return response(200, { + id: ep.id, + name: ep.name, + username: ep.username, + certificate: ep.certificate, + ip: ep.ip, + port: ep.port, + hubs: ep.hubs, + heartbeat: ep.heartbeat, + online: isEndpointOnline(ep), + }) + } + ) +) + +var getFilesystem = pipeline($=>$ + .replaceData() + .replaceMessage( + function () { + var fs = {} + Object.values(endpoints).forEach(ep => { + if (!ep.files) return + ep.files.forEach(f => { + updateFileInfo(fs, f, ep.id) + }) + }) + return response(200, fs) + } + ) +) + +var postFilesystem = pipeline($=>$ + .replaceMessage( + function (req) { + var body = JSON.decode(req.body) + $endpoint.files = Object.entries(body).map( + ([k, v]) => ({ + pathname: k, + time: v['T'], + hash: v['#'], + size: v['$'], + }) + ) + return new Message({ status: 201 }) + } + ) +) + +var postAppStates = pipeline($=>$ + .replaceMessage( + function (req) { + $endpoint.apps = JSON.decode(req.body) + return new Message({ status: 201 }) + } + ) +) + +var getFileInfo = pipeline($=>$ + .replaceData() + .replaceMessage( + function () { + var fs = {} + var pathname = '/' + $params['*'] + Object.values(endpoints).forEach(ep => { + if (!ep.files) return + ep.files.forEach(f => { + if (f.pathname === pathname) { + updateFileInfo(fs, f, ep.id) + } + }) + }) + var info = fs[pathname] + return info ? response(200, info) : response(404) + } + ) +) + +var getFileData = pipeline($=>$ + .pipe( + function (req) { + if (req instanceof MessageStart) { + var id = $params.ep + var ep = endpoints[id] + if (!ep) return notFound + sessions[id]?.forEach?.(h => $hubSelected = h) + if (!$hubSelected) return notFound + var hash = $params.hash + req.head.path = `/api/file-data/${hash}` + return muxToAgent + } + } + ) +) + +var getAppState = pipeline($=>$ + .replaceData() + .replaceMessage( + function () { + var provider = $params.provider + var name = $params.app + var runners = [] + Object.values(endpoints).forEach(ep => { + if (!isEndpointOnline(ep)) return + if (!ep.apps) return + var app = ep.apps.find( + a => a.name === name && (!provider || a.provider === provider) + ) + if (app) { + runners.push({ + id: ep.id, + name: ep.name, + username: app.username, + }) + } + }) + return response(200, { name, provider, endpoints: runners }) + } + ) +) + +var getServices = pipeline($=>$ + .replaceData() + .replaceMessage( + function () { + var services = [] + var collect = (ep) => { + ep.services?.forEach?.( + function (svc) { + var name = svc.name + var protocol = svc.protocol + var s = services.find(s => s.name === name && s.protocol === protocol) + if (!s) services.push(s = { name, protocol, endpoints: [] }) + s.endpoints.push({ id: ep.id, name: ep.name, users: svc.users }) + } + ) + } + if ($params.ep) { + var ep = endpoints[$params.ep] + if (ep && isEndpointOnline(ep)) collect(ep) + } else { + Object.values(endpoints).filter(isEndpointOnline).forEach(collect) + } + return response(200, services) + } + ) +) + +var postServices = pipeline($=>$ + .replaceMessage( + function (req) { + var body = JSON.decode(req.body) + var time = body.time + var last = $endpoint.servicesUpdateTime + if (!last || last <= time) { + var services = body.services + var oldList = $endpoint.services || [] + var newList = services instanceof Array ? services : [] + var who = endpointName($endpoint.id) + console.info(`Received service list (length = ${newList.length}) from ${who}`) + newList.forEach(({ name, protocol }) => { + if (!oldList.some(s => s.name === name && s.protocol === protocol)) { + console.info(`Service ${name} published by ${who}`) + } + }) + oldList.forEach(({ name, protocol }) => { + if (!newList.some(s => s.name === name && s.protocol === protocol)) { + console.info(`Service ${name} deleted by ${who}`) + } + }) + $endpoint.services = newList + $endpoint.servicesUpdateTime = time + } + return new Message({ status: 201 }) + } + ) +) + +var getService = pipeline($=>$ + .replaceData() + .replaceMessage( + function () { + var name = $params.svc + var protocol = $params.proto + var providers = Object.values(endpoints).filter( + ep => isEndpointOnline(ep) && ep.services.some(s => s.name === name && s.protocol === protocol) + ) + if (providers.length === 0) return response(404) + return response(200, { + name, + protocol, + endpoints: providers.map(({ id, name }) => ({ id, name })), + }) + } + ) +) + +var muxToAgent = pipeline($=>$ + .muxHTTP(() => $hubSelected, { version: 2 }).to($=>$ + .swap(() => $hubSelected) + ) +) + +var connectEndpoint = pipeline($=>$ + .acceptHTTPTunnel( + function () { + var id = $params.ep + $ctx.id = id + $hub = new pipeline.Hub + sessions[id] ??= new Set + sessions[id].add($hub) + collectMyNames($ctx.via) + console.info(`Endpoint ${endpointName(id)} joined, connections = ${sessions[id].size}`) + return response(200) + } + ).to($=>$ + .onStart(new Data) + .swap(() => $hub) + .onEnd(() => { + var id = $ctx.id + sessions[id]?.delete?.($hub) + console.info(`Endpoint ${endpointName(id)} left, connections = ${sessions[id]?.size || 0}`) + }) + ) +) + +var connectApp = pipeline($=>$ + .acceptHTTPTunnel( + function (req) { + var app = $params.app + var id = $params.ep + var ep = endpoints[id] + if (!ep) return response(404, 'Endpoint not found') + sessions[id]?.forEach?.(h => $hubSelected = h) + if (!$hubSelected) return response(404, 'Agent not found') + $params.query = new URL(req.head.path).searchParams.toObject() + console.info(`Forward to app ${app} at ${endpointName(id)}`) + return response(200) + } + ).to($=>$ + .connectHTTPTunnel(() => { + var src = $params.query.src + var ip = $ctx.ip + var port = $ctx.port + var username = URL.encodeComponent($ctx.username) + var q = `?src=${src}&ip=${ip}&port=${port}&username=${username}` + return new Message({ + method: 'CONNECT', + path: $params.provider ? `/api/apps/${$params.provider}/${$params.app}${q}` : `/api/apps/${$params.app}${q}`, + }) + }).to(muxToAgent) + ) +) + +var connectService = pipeline($=>$ + .acceptHTTPTunnel( + function () { + var svc = $params.svc + var proto = $params.proto + var id = $params.ep + var ep = endpoints[id] + if (!ep) return response(404, 'Endpoint not found') + if (!canConnect($ctx.username, ep, proto, svc)) return response(403) + if (!ep.services.some(s => s.name === svc && s.protocol === proto)) return response(404, 'Service not found') + sessions[id]?.forEach?.(h => $hubSelected = h) + if (!$hubSelected) return response(404, 'Agent not found') + console.info(`Forward to ${svc} at ${endpointName(id)}`) + return response(200) + } + ).to($=>$ + .connectHTTPTunnel( + () => new Message({ + method: 'CONNECT', + path: `/api/services/${$params.proto}/${$params.svc}`, + }) + ).to(muxToAgent) + ) +) + +var forwardRequest = pipeline($=>$ + .pipe( + function (req) { + if (req instanceof MessageStart) { + var id = $params.ep + var ep = endpoints[id] + if (!ep) return notFound + if (!canOperate($ctx.username, ep)) return notAllowed + sessions[id]?.forEach?.(h => $hubSelected = h) + if (!$hubSelected) return notFound + var path = $params['*'] + req.head.path = `/api/${path}` + return muxToAgent + } + } + ) +) + +// +// Ping agents regularly +// + +pipeline($=>$ + .onStart(new Message({ path: '/api/ping' })) + .repeat(() => new Timeout(15).wait().then(true)).to($=>$ + .forkJoin(() => Object.keys(sessions)).to($=>$ + .onStart(id => { $pingID = id }) + .forkJoin(() => { + var hubs = [] + sessions[$pingID].forEach(h => hubs.push(h)) + return hubs + }).to($=>$ + .onStart(hub => { $hubSelected = hub}) + .pipe(muxToAgent) + .replaceData() + .replaceMessage( + res => { + var hubs = sessions[$pingID] + if (res.head.status !== 200) { + hubs?.delete?.($hubSelected) + console.info(`Endpoint ${endpointName($pingID)} ping failure, connections = ${hubs?.size || 0}`) + } + return new StreamEnd + } + ) + ) + .replaceMessage(new StreamEnd) + ) + .replaceMessage(new StreamEnd) + ) +).spawn() + +var notFound = pipeline($=>$ + .replaceData() + .replaceMessage(response(404)) +) + +var notSupported = pipeline($=>$ + .replaceData() + .replaceMessage(response(405)) +) + +var notAllowed = pipeline($=>$ + .replaceData() + .replaceMessage(response(403)) +) + +var noSession = pipeline($=>$ + .replaceData() + .replaceMessage(response(404, 'No agent session established yet')) +) + +function collectMyNames(addr) { + if (myNames.indexOf(addr) < 0) { + myNames.push(addr) + Object.values(endpoints).forEach( + ep => { + if (ep.isConnected) ep.hubs.push(addr) + } + ) + } +} + +function findCurrentEndpointSession() { + var id = $ctx.id + if (!id) return false + $endpoint = endpoints[id] + if (!$endpoint) { + $endpoint = endpoints[id] = { + id, + username: $ctx.username, + ip: $ctx.ip, + port: $ctx.port, + via: $ctx.via, + services: [], + hubs: [...myNames] + } + } + $endpoint.isConnected = true + return true +} + +function updateFileInfo(fs, f, ep) { + var e = (fs[f.pathname] ??= { + 'T': 0, + '$': 0, + '#': null, + '@': null, + }) + var t1 = e['T'] + var h1 = e['#'] + var t2 = f.time + var h2 = f.hash + if (h2 === h1) { + e['@'].push(ep) + e['T'] = Math.max(t1, t2) + } else if (t2 > t1) { + e['#'] = h2 + e['$'] = f.size + e['@'] = [ep] + e['T'] = t2 + } +} + +function canSee(username, ep) { + if (username === 'root') return true + if (username === ep.username) return true + return false +} + +function canOperate(username, ep) { + if (username === 'root') return true + if (username === ep.username) return true + return false +} + +function canConnect(username, ep, proto, svc) { + if (username === 'root') return true + if (username === ep.username) return true + var s = ep.services.find(({ protocol, name }) => (protocol === proto && name === svc)) + if (!s) return false + if (!s.users) return true + return s.users.includes(username) +} + +function response(status, body) { + if (!body) return new Message({ status }) + if (typeof body === 'string') return responseCT(status, 'text/plain', body) + return responseCT(status, 'application/json', JSON.encode(body)) +} + +function responseCT(status, ct, body) { + return new Message( + { + status, + headers: { 'content-type': ct } + }, + body + ) +} diff --git a/neptune/hub/options.js b/neptune/hub/options.js new file mode 100644 index 000000000..74b797fd1 --- /dev/null +++ b/neptune/hub/options.js @@ -0,0 +1,50 @@ +export default function (argv, { defaults, shorthands }) { + var args = [] + var opts = {} + var lastOption + + argv.forEach( + function (term, i) { + if (i === 0) return + if (lastOption) { + if (term.startsWith('-')) throw `Value missing for option ${lastOption}` + addOption(lastOption, term) + lastOption = undefined + } else if (term.startsWith('--')) { + var kv = term.split('=') + processOption(kv[0], kv[1]) + } else if (term.startsWith('-')) { + if (term.length === 2) { + processOption(term) + } else { + processOption(term.substring(0, 2), term.substring(2)) + } + } else { + args.push(term) + } + } + ) + + function processOption(name, value) { + var k = shorthands[name] || name + if (!(k in defaults)) throw `invalid option ${k}` + if (typeof defaults[k] === 'boolean') { + opts[k] = true + } else if (value) { + addOption(k, value) + } else { + lastOption = name + } + } + + function addOption(name, value) { + var k = shorthands[name] || name + switch (typeof defaults[k]) { + case 'number': opts[k] = Number.parseFloat(value); break + case 'string': opts[k] = value; break + case 'object': (opts[k] ??= []).push(value); break + } + } + + return { args, ...defaults, ...opts } +} diff --git a/neptune/mega/tunnel/cli.js b/neptune/mega/tunnel/cli.js index b1068ab52..6f2416dfb 100644 --- a/neptune/mega/tunnel/cli.js +++ b/neptune/mega/tunnel/cli.js @@ -1,9 +1,9 @@ -export default function ({ app, api, utils }) { +export default function ({ api, utils }) { return pipeline($=>$ - .onStart(argv => main(argv)) + .onStart(ctx => main(ctx)) ) - function main(argv) { + function main({ argv, endpoint }) { var buffer = new Data function output(str) { @@ -27,21 +27,6 @@ export default function ({ app, api, utils }) { return api.allEndpoints().then(list => (endpoints = list)) } - function selectEndpoint(name) { - if (name) { - return allEndpoints().then(endpoints => { - var ep = endpoints.find(ep => ep.id === name) - if (ep) return ep - var list = endpoints.filter(ep => ep.name === name) - if (list.length === 1) return list[0] - if (list.length === 0) throw `Endpoint '${name}' not found` - throw `Ambiguous endpoint name '${name}'` - }) - } else { - return Promise.resolve(app.endpoint) - } - } - function lookupEndpointNames(list) { return allEndpoints().then(endpoints => ( list.map(id => { @@ -72,16 +57,11 @@ export default function ({ app, api, utils }) { { title: 'List objects of the specified type', usage: 'get ', - options: ` - --mesh Specify a mesh by name - --ep Specify an endpoint by name or UUID - `, notes: objectTypeNotes, action: (args) => { - var ep = args['--ep'] switch (validateObjectType(args, 'get')) { - case 'inbound': return getInbound(ep) - case 'outbound': return getOutbound(ep) + case 'inbound': return getInbound() + case 'outbound': return getOutbound() } } }, @@ -89,17 +69,12 @@ export default function ({ app, api, utils }) { { title: 'Show detailed info of the specified object', usage: 'describe ', - options: ` - --mesh Specify a mesh by name - --ep Specify an endpoint by name or UUID - `, notes: objectTypeNotes + objectNameNotes, action: (args) => { - var ep = args['--ep'] var name = args[''] switch (validateObjectType(args, 'describe')) { - case 'inbound': return describeInbound(ep, name) - case 'outbound': return describeOutbound(ep, name) + case 'inbound': return describeInbound(name) + case 'outbound': return describeOutbound(name) } } }, @@ -108,9 +83,6 @@ export default function ({ app, api, utils }) { title: 'Create an object of the specified type', usage: 'open ', options: ` - --mesh Specify a mesh by name - --ep Specify an endpoint by name or UUID - For inbound end: --listen <[ip:]port ...> Set local ports to listen on @@ -123,11 +95,10 @@ export default function ({ app, api, utils }) { `, notes: objectTypeNotes + objectNameNotes, action: (args) => { - var ep = args['--ep'] var name = args[''] switch (validateObjectType(args, 'open')) { - case 'inbound': return openInbound(ep, name, args['--listen'], args['--exit']) - case 'outbound': return openOutbound(ep, name, args['--target'], args['--entrance']) + case 'inbound': return openInbound(name, args['--listen'], args['--exit']) + case 'outbound': return openOutbound(name, args['--target'], args['--entrance']) } } }, @@ -135,17 +106,12 @@ export default function ({ app, api, utils }) { { title: 'Delete the specified object', usage: 'close ', - options: ` - --mesh Specify a mesh by name - --ep Specify an endpoint by name or UUID - `, notes: objectTypeNotes + objectNameNotes, action: (args) => { - var ep = args['--ep'] var name = args[''] switch (validateObjectType(args, 'close')) { - case 'inbound': return closeInbound(ep, name) - case 'outbound': return closeOutbound(ep, name) + case 'inbound': return closeInbound(name) + case 'outbound': return closeOutbound(name) } } }, @@ -161,10 +127,8 @@ export default function ({ app, api, utils }) { return Promise.resolve(flush()) } - function getInbound(epName) { - return selectEndpoint(epName).then(ep => - api.allInbound(ep.id) - ).then(list => ( + function getInbound() { + return api.allInbound(endpoint.id).then(list => ( Promise.all(list.map(i => lookupEndpointNames(i.exits || []).then(exits => ({ ...i, @@ -180,10 +144,8 @@ export default function ({ app, api, utils }) { )) } - function getOutbound(epName) { - return selectEndpoint(epName).then(ep => - api.allOutbound(ep.id) - ).then(list => ( + function getOutbound() { + return api.allOutbound(endpoint.id).then(list => ( Promise.all(list.map(o => lookupEndpointNames(o.entrances || []).then(entrances => ({ ...o, @@ -199,76 +161,64 @@ export default function ({ app, api, utils }) { )) } - function describeInbound(epName, tunnelName) { + function describeInbound(tunnelName) { var obj = validateObjectName(tunnelName) - return selectEndpoint(epName).then(ep => - api.getInbound(ep.id, obj.protocol, obj.name).then(obj => { - if (!obj) return - return lookupEndpointNames(obj.exits || []).then(exits => { - output(`Inbound ${obj.protocol}/${obj.name}\n`) - output(`Endpoint: ${ep.name} (${ep.id})\n`) - output(`Listens:\n`) - obj.listens.forEach(l => output(` ${l.ip}:${l.port}\n`)) - output(`Exits:\n`) - exits.forEach(e => output(` ${e}\n`)) - if (exits.length === 0) output(` (all endpoints)\n`) - }) + return api.getInbound(endpoint.id, obj.protocol, obj.name).then(obj => { + if (!obj) return + return lookupEndpointNames(obj.exits || []).then(exits => { + output(`Inbound ${obj.protocol}/${obj.name}\n`) + output(`Endpoint: ${endpoint.name} (${endpoint.id})\n`) + output(`Listens:\n`) + obj.listens.forEach(l => output(` ${l.ip}:${l.port}\n`)) + output(`Exits:\n`) + exits.forEach(e => output(` ${e}\n`)) + if (exits.length === 0) output(` (all endpoints)\n`) }) - ) + }) } - function describeOutbound(epName, tunnelName) { + function describeOutbound(tunnelName) { var obj = validateObjectName(tunnelName) - return selectEndpoint(epName).then(ep => - api.getOutbound(ep.id, obj.protocol, obj.name).then(obj => { - if (!obj) return - return lookupEndpointNames(obj.entrances || []).then(entrances => { - output(`Outbound ${obj.protocol}/${obj.name}\n`) - output(`Endpoint: ${ep.name} (${ep.id})\n`) - output(`Targets:\n`) - obj.targets.forEach(t => output(` ${t.host}:${t.port}\n`)) - output(`Entrances:\n`) - entrances.forEach(e => output(` ${e}\n`)) - if (entrances.length === 0) output(` (all endpoints)\n`) - }) + return api.getOutbound(endpoint.id, obj.protocol, obj.name).then(obj => { + if (!obj) return + return lookupEndpointNames(obj.entrances || []).then(entrances => { + output(`Outbound ${obj.protocol}/${obj.name}\n`) + output(`Endpoint: ${endpoint.name} (${endpoint.id})\n`) + output(`Targets:\n`) + obj.targets.forEach(t => output(` ${t.host}:${t.port}\n`)) + output(`Entrances:\n`) + entrances.forEach(e => output(` ${e}\n`)) + if (entrances.length === 0) output(` (all endpoints)\n`) }) - ) + }) } - function openInbound(epName, tunnelName, listens, exits) { + function openInbound(tunnelName, listens, exits) { var obj = validateObjectName(tunnelName) if (!listens || listens.length === 0) throw `Option '--listen' is required` listens = listens.map(l => validateHostPort(l)).map(({ host, port }) => ({ ip: host, port })) return lookupEndpointIDs(exits || []).then( - exits => selectEndpoint(epName).then( - ep => api.setInbound(ep.id, obj.protocol, obj.name, listens, exits) - ) + exits => api.setInbound(endpoint.id, obj.protocol, obj.name, listens, exits) ) } - function openOutbound(epName, tunnelName, targets, entrances) { + function openOutbound(tunnelName, targets, entrances) { var obj = validateObjectName(tunnelName) if (!targets || targets.length === 0) throw `Option '--target' is required` targets = targets.map(t => validateHostPort(t)) return lookupEndpointIDs(entrances || []).then( - entrances => selectEndpoint(epName).then( - ep => api.setOutbound(ep.id, obj.protocol, obj.name, targets, entrances) - ) + entrances => api.setOutbound(endpoint.id, obj.protocol, obj.name, targets, entrances) ) } - function closeInbound(epName, tunnelName) { + function closeInbound(tunnelName) { var obj = validateObjectName(tunnelName) - return selectEndpoint(epName).then(ep => - api.deleteInbound(ep.id, obj.protocol, obj.name) - ) + return api.deleteInbound(endpoint.id, obj.protocol, obj.name) } - function closeOutbound(epName, tunnelName) { + function closeOutbound(tunnelName) { var obj = validateObjectName(tunnelName) - return selectEndpoint(epName).then(ep => - api.deleteOutbound(ep.id, obj.protocol, obj.name) - ) + return api.deleteOutbound(endpoint.id, obj.protocol, obj.name) } function validateObjectType(args, command) { diff --git a/neptune/mega/tunnel/main.js b/neptune/mega/tunnel/main.js index 3c2e6b39d..c899ed385 100644 --- a/neptune/mega/tunnel/main.js +++ b/neptune/mega/tunnel/main.js @@ -16,6 +16,15 @@ export default function ({ app, mesh, utils }) { 'CONNECT': utils.createCLIResponder(cli), }, + '/api/appinfo': { + 'GET': responder(() => Promise.resolve(response(200, { + name: app.name, + provider: app.provider, + username: app.username, + endpoint: app.endpoint, + }))) + }, + '/api/endpoints': { 'GET': responder(() => api.allEndpoints().then( ret => ret ? response(200, ret) : response(404) diff --git a/neptune/src/lib.rs b/neptune/src/lib.rs index e6ba7fa97..82d7c506c 100644 --- a/neptune/src/lib.rs +++ b/neptune/src/lib.rs @@ -39,8 +39,8 @@ pub fn start_hub(listen_port: u16, name: Vec, ca: &str) { CString::new("ztm-pipy").unwrap(), CString::new("repo://ztm/hub").unwrap(), CString::new("--args").unwrap(), - CString::new(format!("--listen=0.0.0.0:{}", listen_port)).unwrap(), - CString::new(format!("--ca={}", ca)).unwrap(), + // CString::new(format!("--listen=0.0.0.0:{}", listen_port)).unwrap(), + // CString::new(format!("--ca={}", ca)).unwrap(), ]; let c_args: Vec<*const c_char> = args.iter().map(|arg| arg.as_ptr()).collect(); unsafe { @@ -80,6 +80,12 @@ mod tests { tracing::info!("ztm agent exit success"); } + #[tokio::test] + async fn test_start_hub() { + start_hub(8888, vec![], "localhost:9999"); + thread::sleep(std::time::Duration::from_secs(3)); + } + #[tokio::test] #[should_panic] /// didn't support multiple agent From de30d33eb9852bd6a6a960753d4ad7d2c05caf00 Mon Sep 17 00:00:00 2001 From: wujian0327 <353981613@qq.com> Date: Sun, 21 Jul 2024 19:25:55 +0800 Subject: [PATCH 3/3] update hub start function --- neptune/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/neptune/src/lib.rs b/neptune/src/lib.rs index 82d7c506c..b41193d2d 100644 --- a/neptune/src/lib.rs +++ b/neptune/src/lib.rs @@ -32,7 +32,7 @@ pub fn start_agent(database: &str, listen_port: u16) { /// start ztm hub, like run pipy repo://ztm/hub --listen=0.0.0.0:listen_port --name=name --ca=ca /// ! only support to start one agent or one hub at one process -pub fn start_hub(listen_port: u16, name: Vec, ca: &str) { +pub fn start_hub(listen_port: u16, name: Vec, _ca: &str) { let _ = name; // TODO: ignore name tracing::info!("start pipy with port: {}", listen_port); let args = [