gitmono-dev · genedna · Jul 22, 2024 · Jul 19, 2024 · Jul 19, 2024 · Jul 20, 2024
diff --git a/.gitignore b/.gitignore
@@ -21,6 +21,7 @@ Cargo.lock
 
 # zmt agent data file
 ztm_agent.db
+ztm_agent_db*
 
 # buck2 out
 buck-out
diff --git 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<String>,
 }
@@ -26,4 +35,6 @@ pub struct GetParams {
     pub path: Option<String>,
     pub limit: Option<String>,
     pub cursor: Option<String>,
+    pub identifier: Option<String>,
+    pub port: Option<u16>,
 }
diff --git a/gateway/src/ca_server.rs 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<AppState>,
+    Query(_params): Query<RelayGetParams>,
+    uri: Uri,
+) -> Result<Response<Body>, (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<AppState>,
+    uri: Uri,
+    req: Request<Body>,
+) -> Result<Response<Body>, (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<AppState>,
+    uri: Uri,
+    _req: Request<Body>,
+) -> Result<Response<Body>, (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
@@ -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<AppState> 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
@@ -1,4 +1,5 @@
 pub mod api;
+pub mod ca_server;
 mod git_protocol;
 pub mod https_server;
 pub mod init;