aboutsummaryrefslogtreecommitdiffstats
path: root/src/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport')
-rw-r--r--src/transport/libp2p.rs35
-rw-r--r--src/transport/mod.rs34
-rw-r--r--src/transport/mycelium/mod.rs73
-rw-r--r--src/transport/tor.rs22
4 files changed, 156 insertions, 8 deletions
diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs
index 610b94e..2d67d2c 100644
--- a/src/transport/libp2p.rs
+++ b/src/transport/libp2p.rs
@@ -205,6 +205,41 @@ pub(crate) struct Libp2pConfig {
}
impl Libp2pConfig {
+ /// The configuration `settings` give; `relay_only`: a relay that is no node
+ /// (`--relay-only`).
+ pub(crate) fn from_settings(
+ settings: &crate::config::Settings,
+ relay_only: bool,
+ ) -> Result<Self> {
+ let parse = |option: &str, addresses: &[String]| {
+ addresses
+ .iter()
+ .map(|address| {
+ address
+ .parse::<Multiaddr>()
+ .with_context(|| format!("{option} {address:?}"))
+ })
+ .collect::<Result<Vec<_>>>()
+ };
+ Ok(Self {
+ key: Self::load_key(settings.libp2p_key_file.as_deref())?,
+ listen: parse("libp2p-listen", &settings.libp2p_listen)?,
+ announce: parse("libp2p-announce", &settings.libp2p_announce)?,
+ relays: parse("libp2p-relays", &settings.libp2p_relays)?,
+ relay: settings.libp2p_relay || relay_only,
+ continuum: !relay_only,
+ mdns: settings.libp2p_mdns,
+ upnp: settings.libp2p_upnp,
+ ignore: settings
+ .libp2p_ignore
+ .iter()
+ .map(|range| range.parse().context("libp2p-ignore"))
+ .collect::<Result<Vec<_>>>()?,
+ public: settings.libp2p_public,
+ memory: false,
+ })
+ }
+
/// The key in `path` (32 raw bytes, an Ed25519 secret key), or a new one each start
/// without a file: the peer ID only says where a node is, so a new one costs nothing
/// but the WHO that finds it again (`transport::identity`).
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b10c799..9d43e5a 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -28,6 +28,40 @@ pub(crate) mod tor;
pub(crate) mod traffic;
pub(crate) mod wire;
+/// Starts the transport `name` (one of `config::TRANSPORTS`) as `settings` configure
+/// it: its scheme and the transport, or None if `stop` came first (some take a while to
+/// start: a supervised Mycelium daemon, arti). `key`: the node key, which some prove
+/// themselves.
+pub(crate) async fn start(
+ name: &str,
+ settings: &crate::config::Settings,
+ key: &std::sync::Arc<crate::identity::NodeKey>,
+ stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<Option<(&'static str, std::sync::Arc<dyn Transport>)>> {
+ use std::sync::Arc;
+ let _ = key;
+ let started: Option<(&'static str, Arc<dyn Transport>)> = match name {
+ "mycelium" => mycelium::start(settings, stop)
+ .await?
+ .map(|transport| (mycelium::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
+ #[cfg(feature = "libp2p")]
+ "libp2p" => {
+ let _ = stop;
+ let config = libp2p::Libp2pConfig::from_settings(settings, false)?;
+ Some((
+ libp2p::SCHEME,
+ Arc::new(libp2p::Libp2pTransport::start(config)?) as Arc<dyn Transport>,
+ ))
+ }
+ #[cfg(feature = "tor")]
+ "tor" => tor::TorTransport::from_settings(settings, key, stop)
+ .await?
+ .map(|transport| (tor::SCHEME, Arc::new(transport) as Arc<dyn Transport>)),
+ other => anyhow::bail!("transports: no {other} in this build"),
+ };
+ Ok(started)
+}
+
/// Random identifier of one run of a node (from start to exit).
pub(crate) type RunId = [u8; 16];
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index d33e323..836eca9 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -91,6 +91,70 @@ pub(crate) struct MyceliumTransport {
local_id: String,
}
+/// The transport as `settings` configure it (`transport::start`): its daemon supervised
+/// (started, which can take a while) or one running elsewhere. None if `stop` came
+/// first.
+pub(crate) async fn start(
+ settings: &crate::config::Settings,
+ mut stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<Option<MyceliumTransport>> {
+ let spawn_config = MyceliumSpawnConfig {
+ rest_url_opt: settings.mycelium_rest_url(),
+ key_file_opt: settings.mycelium_key_file.clone(),
+ network_key_file: settings.mycelium_network_key_file.clone(),
+ network_name: settings.mycelium_network_name.clone(),
+ tcp_port: settings.mycelium_tcp_listen_port,
+ peers: settings.mycelium_peers.clone(),
+ };
+ let setup = tokio::select! {
+ setup = MyceliumSetup::bootstrap(spawn_config) => setup?,
+ _ = stop.wait_for(|stop| *stop) => return Ok(None),
+ };
+ // Without a password it may carry (the option is secret).
+ let mut shown_url = setup.rest_url.clone();
+ let _ = shown_url.set_password(None);
+ tracing::debug!(url = shown_url.to_string(), "Using Mycelium REST API");
+ tokio::select! {
+ transport = MyceliumTransport::connect(setup) => Ok(Some(transport?)),
+ _ = stop.wait_for(|stop| *stop) => Ok(None),
+ }
+}
+
+/// How long the daemon's API may take to come up: a daemon just started listens a
+/// moment later (`admin_info`).
+const API_READY_WITHIN: Duration = Duration::from_secs(30);
+
+/// The daemon's identity, from its API at `url` (`/api/v1/admin`), once the API answers:
+/// a daemon we just started doesn't listen yet, so refused connections are tried again
+/// (50 ms, doubling up to a second) for `API_READY_WITHIN`; anything else is an error.
+async fn admin_info(client: &reqwest::Client, url: Url) -> Result<MyceliumAdminInfo> {
+ let deadline = tokio::time::Instant::now() + API_READY_WITHIN;
+ let mut wait = Duration::from_millis(50);
+ loop {
+ match client.get(url.clone()).send().await {
+ Ok(response) => {
+ return response
+ .error_for_status()?
+ .json()
+ .await
+ .context("Querying local Mycelium identity");
+ }
+ Err(error) if error.is_connect() && tokio::time::Instant::now() + wait < deadline => {
+ tracing::debug!(url = %url, error = %error, "Mycelium's API doesn't answer yet; trying again");
+ tokio::time::sleep(wait).await;
+ wait = (wait * 2).min(Duration::from_secs(1));
+ }
+ Err(error) => {
+ return Err(error).with_context(|| {
+ format!(
+ "Querying local Mycelium identity (no answer within {API_READY_WITHIN:?})"
+ )
+ });
+ }
+ }
+ }
+}
+
impl MyceliumTransport {
pub(crate) async fn connect(setup: MyceliumSetup) -> Result<Self> {
let rest_url = setup.rest_url.clone();
@@ -102,14 +166,7 @@ impl MyceliumTransport {
.timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30))
.build()?;
- let admin: MyceliumAdminInfo = client
- .get(rest_url.join("/api/v1/admin")?)
- .send()
- .await?
- .error_for_status()?
- .json()
- .await
- .context("Querying local Mycelium identity")?;
+ let admin = admin_info(&client, rest_url.join("/api/v1/admin")?).await?;
let transport = Self {
setup: tokio::sync::Mutex::new(setup),
diff --git a/src/transport/tor.rs b/src/transport/tor.rs
index 2cf3249..8eb1d02 100644
--- a/src/transport/tor.rs
+++ b/src/transport/tor.rs
@@ -198,6 +198,28 @@ impl TorTransport {
/// while: it fetches what it needs to know of the Tor network). Until it is done,
/// dials wait for it (arti bootstraps on demand), and the onion service is published
/// once it is: the node starts without waiting for Tor.
+ /// The transport as `settings` configure it (`transport::start`), proving `key`.
+ /// None if `stop` came first (bootstrapping arti can take a while).
+ pub(crate) async fn from_settings(
+ settings: &crate::config::Settings,
+ key: &Arc<NodeKey>,
+ mut stop: watch::Receiver<bool>,
+ ) -> Result<Option<Self>> {
+ let state_dir = settings
+ .tor_state_dir
+ .clone()
+ .context("tor needs tor-state-dir")?;
+ let config = TorConfig {
+ key: Arc::clone(key),
+ state_dir,
+ port: settings.tor_port,
+ };
+ tokio::select! {
+ transport = Self::start(config) => transport.map(Some),
+ _ = stop.wait_for(|stop| *stop) => Ok(None),
+ }
+ }
+
pub(crate) async fn start(config: TorConfig) -> Result<Self> {
let tor_config = TorClientConfigBuilder::from_directories(
config.state_dir.join("state"),