diff --git a/aries/src/main.rs b/aries/src/main.rs index 6b139d0e4..06fa3aca5 100644 --- a/aries/src/main.rs +++ b/aries/src/main.rs @@ -50,7 +50,7 @@ async fn main() { ztm_agent.clone().start_ztm_agent(); thread::sleep(time::Duration::from_secs(3)); run_ztm_client( - "http://34.84.172.121/relay".to_string(), + "http://gitmono.org/relay".to_string(), config.clone(), peer_id, ztm_agent, diff --git a/gemini/src/http/handler.rs b/gemini/src/http/handler.rs index d97b329c7..adb346203 100644 --- a/gemini/src/http/handler.rs +++ b/gemini/src/http/handler.rs @@ -2,8 +2,10 @@ use common::model::CommonResult; use jupiter::context::Context; use crate::{ - util::{get_available_port, get_short_peer_id, repo_alias_to_identifier}, - ztm::{agent::share_repo, create_tunnel, send_get_request_to_peer_by_tunnel}, + util::repo_alias_to_identifier, + ztm::{ + agent::share_repo, get_or_create_remote_mega_tunnel, send_get_request_to_peer_by_tunnel, + }, RepoInfo, }; @@ -54,53 +56,11 @@ pub async fn repo_provide( Ok("success".to_string()) } -pub async fn repo_folk( - ztm_agent_port: u16, - identifier: String, - local_port: u16, -) -> Result { - let remote_peer_id = match get_peer_id_from_identifier(identifier.clone()) { - Ok(p) => p, - Err(e) => return Err(e), - }; - let remote_port = match get_remote_port_from_identifier(identifier.clone()) { - Ok(p) => p, - Err(e) => return Err(e), - }; - let git_path = match get_git_path_from_identifier(identifier) { - Ok(p) => p, - Err(e) => return Err(e), - }; - - let (peer_id, _) = vault::init(); - let bound_name = format!( - "{}_{}", - get_short_peer_id(peer_id), - get_short_peer_id(remote_peer_id.clone()) - ); - match create_tunnel( - ztm_agent_port, - remote_peer_id, - local_port, - remote_port, - bound_name, - ) - .await - { - Ok(_) => (), - Err(e) => return Err(e), - } - - let msg = format!("git clone http://localhost:{local_port}/{git_path}"); - Ok(msg) -} - pub async fn repo_folk_alias(ztm_agent_port: u16, identifier: String) -> Result { let remote_peer_id = match get_peer_id_from_identifier(identifier.clone()) { Ok(p) => p, Err(e) => return Err(e), }; - let remote_port = 8000; let alias = match get_alias_from_identifier(identifier) { Ok(p) => p, Err(e) => return Err(e), @@ -111,28 +71,14 @@ pub async fn repo_folk_alias(ztm_agent_port: u16, identifier: String) -> Result< Err(e) => return Err(e), }; - let peer_id = vault::get_peerid(); - let bound_name = format!( - "{}_{}", - get_short_peer_id(peer_id), - get_short_peer_id(remote_peer_id.clone()) - ); - let local_port = match get_available_port() { - Ok(p) => p, - Err(e) => return Err(e), + let local_port = get_or_create_remote_mega_tunnel(ztm_agent_port, remote_peer_id).await; + + let local_port = match local_port { + Ok(local_port) => local_port, + Err(e) => { + return Err(e); + } }; - match create_tunnel( - ztm_agent_port, - remote_peer_id, - local_port, - remote_port, - bound_name, - ) - .await - { - Ok(_) => (), - Err(e) => return Err(e), - } let msg = format!("http://localhost:{local_port}{path}.git"); Ok(msg) @@ -147,28 +93,6 @@ pub fn get_peer_id_from_identifier(identifier: String) -> Result 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_alias_from_identifier(identifier: String) -> Result { // p2p://wGg2inNE22LY1eHttDB63znw2MnsK8CPXeG2nfhpXs5a/serde_python let words: Vec<&str> = identifier.split('/').collect(); diff --git a/gemini/src/lib.rs b/gemini/src/lib.rs index a035aad1c..a60af9cf4 100644 --- a/gemini/src/lib.rs +++ b/gemini/src/lib.rs @@ -10,7 +10,7 @@ pub mod nostr; pub mod util; pub mod ztm; -const ZTM_APP_PROVIDER: &str = "mega"; + #[derive(Deserialize, Debug)] pub struct RelayGetParams { diff --git a/gemini/src/util.rs b/gemini/src/util.rs index cc2baf5ce..5af2cb575 100644 --- a/gemini/src/util.rs +++ b/gemini/src/util.rs @@ -52,3 +52,11 @@ pub fn repo_path_to_identifier(http_port: u16, repo_path: String) -> String { let (peer_id, _) = vault::init(); format!("p2p://{}/{http_port}{repo_path}.git", peer_id.clone()) } + +pub fn get_ztm_app_tunnel_bound_name(remote_peer_id: String) -> String { + format!( + "{}_{}", + get_short_peer_id(vault::get_peerid()), + get_short_peer_id(remote_peer_id) + ) +} diff --git a/gemini/src/ztm/agent.rs b/gemini/src/ztm/agent.rs index 14710f928..6dcb04057 100644 --- a/gemini/src/ztm/agent.rs +++ b/gemini/src/ztm/agent.rs @@ -10,9 +10,12 @@ use common::config::Config; use reqwest::{header::CONTENT_TYPE, Client}; use serde::{Deserialize, Serialize}; -use crate::{RepoInfo, ZTM_APP_PROVIDER}; +use crate::{ + ztm::{ZTM_APP_NAME, ZTM_APP_PROVIDER}, + RepoInfo, +}; -use super::{handle_response, hub::ZTMUserPermit}; +use super::{handle_response, hub::ZTMUserPermit, MESH_NAME}; #[derive(Deserialize, Serialize, Debug, Clone)] pub struct ZTMMesh { @@ -55,6 +58,20 @@ pub struct ZTMPortReq { pub struct ZTMPortService { pub service: String, } + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMAppInbound { + pub protocol: String, + pub name: String, + pub listens: Vec, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ZTMAppInboundListen { + pub ip: String, + pub port: u16, +} + #[async_trait] pub trait ZTMAgent { async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result; @@ -90,6 +107,14 @@ pub trait ZTMAgent { port: u16, ) -> Result; + async fn get_ztm_app_tunnel_inbound_port( + &self, + ep_id: String, + provider: String, + app_name: String, + bound_name: String, + ) -> Option; + async fn create_ztm_app_tunnel_outbound( &self, ep_id: String, @@ -120,8 +145,6 @@ impl LocalZTMAgent { } } -const MESH_NAME: &str = "relay_mesh"; - #[async_trait] impl ZTMAgent for LocalZTMAgent { async fn connect_ztm_hub(&self, permit: ZTMUserPermit) -> Result { @@ -335,6 +358,42 @@ impl ZTMAgent for LocalZTMAgent { Ok(response_text) } + async fn get_ztm_app_tunnel_inbound_port( + &self, + ep_id: String, + provider: String, + app_name: String, + bound_name: String, + ) -> Option { + //GET /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" + ); + tracing::debug!("get_ztm_app_tunnel_inbound url: {}", url); + let client = Client::new(); + let request_result = client.get(url).send().await; + let response_text = match handle_response(request_result).await { + Ok(s) => s, + Err(e) => { + tracing::error!("get_ztm_app_tunnel_inbound error: {}", e); + return None; + } + }; + let ztm_app_inbounds: Vec = + match serde_json::from_str(response_text.as_str()) { + Ok(inbounds) => inbounds, + Err(e) => { + tracing::error!("get_ztm_app_tunnel_inbound error: {}", e); + return None; + } + }; + let ztm_app_inbound = ztm_app_inbounds.iter().find(|x| x.name == bound_name)?; + let listen = ztm_app_inbound.listens.first()?; + Some(listen.port) + } + async fn create_ztm_app_tunnel_outbound( &self, ep_id: String, @@ -410,7 +469,7 @@ pub async fn run_ztm_client( .start_ztm_app( mesh.clone().agent.id, ZTM_APP_PROVIDER.to_string(), - "tunnel".to_string(), + ZTM_APP_NAME.to_string(), ) .await { @@ -422,23 +481,6 @@ pub async fn run_ztm_client( } tracing::info!("start tunnel app successfully"); - //test send msg - // sleep(Duration::from_secs(5)); - // match send_get_request_to_peer_by_tunnel( - // agent.agent_port, - // "nX7NRgitx7wUwAiJXxAVcec4iAoV8YwbzUn8FqwfFR4J".to_string(), - // "api/v1/ztm/repo_provide".to_string(), - // ) - // .await - // { - // Ok(s) => { - // tracing::info!("send_get_request_to_peer_by_tunnel successfully:{}", s); - // } - // Err(e) => { - // tracing::error!(e); - // } - // }; - // ping relay let peer_id_clone = peer_id.clone(); let bootstrap_node_clone = bootstrap_node.clone(); diff --git a/gemini/src/ztm/mod.rs b/gemini/src/ztm/mod.rs index 616405e8b..e55f0851e 100755 --- a/gemini/src/ztm/mod.rs +++ b/gemini/src/ztm/mod.rs @@ -1,15 +1,18 @@ use agent::{LocalZTMAgent, ZTMAgent}; use reqwest::{header::CONTENT_TYPE, Client}; -use crate::{ - util::{get_available_port, get_short_peer_id, handle_response}, - ZTM_APP_PROVIDER, -}; +use crate::util::{get_available_port, get_ztm_app_tunnel_bound_name, handle_response}; pub mod agent; pub mod hub; -pub async fn create_tunnel( +const MESH_NAME: &str = "relay_mesh"; + +const ZTM_APP_PROVIDER: &str = "mega"; + +const ZTM_APP_NAME: &str = "tunnel_punch"; + +async fn create_tunnel( ztm_agent_port: u16, remote_peer_id: String, local_port: u16, @@ -35,7 +38,7 @@ pub async fn create_tunnel( .create_ztm_app_tunnel_inbound( local_ep.id, ZTM_APP_PROVIDER.to_string(), - "tunnel".to_string(), + ZTM_APP_NAME.to_string(), bound_name.clone(), local_port, ) @@ -54,7 +57,7 @@ pub async fn create_tunnel( .create_ztm_app_tunnel_outbound( remote_ep.id, ZTM_APP_PROVIDER.to_string(), - "tunnel".to_string(), + ZTM_APP_NAME.to_string(), bound_name, remote_port, ) @@ -71,57 +74,102 @@ pub async fn create_tunnel( Ok(()) } -pub async fn send_get_request_to_peer_by_tunnel( +pub async fn get_or_create_remote_mega_tunnel( ztm_agent_port: u16, remote_peer_id: String, - path: String, -) -> Result { - //get a random port - let local_port = match get_available_port() { - Ok(port) => port, - Err(e) => { - return Err(e); - } - }; - let (peer_id, _) = vault::init(); - let bound_name = format!( - "get_{}_{}", - get_short_peer_id(peer_id), - get_short_peer_id(remote_peer_id.clone()) +) -> Result { + let bound_name = get_ztm_app_tunnel_bound_name(remote_peer_id.clone()); + + //Check if the tunnel exists + let local_port = search_tunnel_inbound_port(ztm_agent_port, bound_name.clone()).await; + + tracing::debug!( + "get_or_create_remote_mega_tunnel, local_port exist:{:?}", + local_port ); - let remote_port = 8000; - match create_tunnel( - ztm_agent_port, - remote_peer_id.clone(), - local_port, - remote_port, - bound_name.clone(), - ) - .await - { - Ok(_) => { - tracing::info!( - "create ztm tunnel successfully: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}", - bound_name, + + let local_port = match local_port { + Some(local_port) => local_port, + None => { + //create a new tunnel to remote peer + let local_port = match get_available_port() { + Ok(port) => port, + Err(e) => { + return Err(e); + } + }; + let remote_port = 8000; + match create_tunnel( + ztm_agent_port, + remote_peer_id.clone(), local_port, - remote_peer_id, remote_port, - ); + bound_name.clone(), + ) + .await + { + Ok(_) => { + tracing::info!( + "create ztm tunnel successfully: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}", + bound_name, + local_port, + remote_peer_id, + remote_port, + ); + } + Err(e) => { + tracing::error!( + "create ztm tunnel failed: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}\nerror:{}", + bound_name, + local_port, + remote_peer_id, + remote_port, + e.clone(), + ); + return Err(e); + } + } + local_port } + }; + Ok(local_port) +} + +async fn search_tunnel_inbound_port(ztm_agent_port: u16, bound_name: String) -> Option { + let agent: LocalZTMAgent = LocalZTMAgent { + agent_port: ztm_agent_port, + }; + let local_ep = match agent.get_ztm_local_endpoint().await { + Ok(ep) => ep, + Err(_) => return None, + }; + tracing::debug!("local_ep:{:?}", local_ep); + //search inbound + agent + .get_ztm_app_tunnel_inbound_port( + local_ep.id, + ZTM_APP_PROVIDER.to_string(), + ZTM_APP_NAME.to_string(), + bound_name.clone(), + ) + .await +} + +pub async fn send_get_request_to_peer_by_tunnel( + ztm_agent_port: u16, + remote_peer_id: String, + path: String, +) -> Result { + let local_port = get_or_create_remote_mega_tunnel(ztm_agent_port, remote_peer_id).await; + + let local_port = match local_port { + Ok(local_port) => local_port, Err(e) => { - tracing::error!( - "create ztm tunnel failed: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}\nerror:{}", - bound_name, - local_port, - remote_peer_id, - remote_port, - e.clone(), - ); return Err(e); } - } + }; + let url = format!("http://127.0.0.1:{local_port}/{path}"); - tracing::info!("send request to:\n{}", url); let request_result = reqwest::get(url.clone()).await; match handle_response(request_result).await { Ok(s) => { @@ -141,50 +189,15 @@ pub async fn send_post_request_to_peer_by_tunnel( path: String, body: String, ) -> Result { - //get a random port - let local_port = match get_available_port() { - Ok(port) => port, + let local_port = get_or_create_remote_mega_tunnel(ztm_agent_port, remote_peer_id).await; + + let local_port = match local_port { + Ok(local_port) => local_port, Err(e) => { return Err(e); } }; - let (peer_id, _) = vault::init(); - let bound_name = format!( - "post_{}_{}", - get_short_peer_id(peer_id), - get_short_peer_id(remote_peer_id.clone()) - ); - let remote_port = 8000; - match create_tunnel( - ztm_agent_port, - remote_peer_id.clone(), - local_port, - remote_port, - bound_name.clone(), - ) - .await - { - Ok(_) => { - tracing::info!( - "create ztm tunnel successfully: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}", - bound_name, - local_port, - remote_peer_id, - remote_port, - ); - } - Err(e) => { - tracing::error!( - "create ztm tunnel failed: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}\nerror:{}", - bound_name, - local_port, - remote_peer_id, - remote_port, - e.clone(), - ); - return Err(e); - } - } + let url = format!("http://127.0.0.1:{local_port}/{path}"); let client = Client::new(); diff --git a/neptune/libs/ztm b/neptune/libs/ztm index f3ba7aa56..2e0ac3ed1 160000 --- a/neptune/libs/ztm +++ b/neptune/libs/ztm @@ -1 +1 @@ -Subproject commit f3ba7aa56814c5e23b12bb3f7d951b86977fd260 +Subproject commit 2e0ac3ed15b8374aeb8227b6a65a83dc6f2a97dc