p2p checkpoint

Cargo.toml

@@ -21,13 +21,14 @@ bincode = "1.3"
 bs58 = "0.4"
 cached = "0.40"
 derive_builder = "0.11"
+either = "1.8"
 futures = { default-features = false, version = "0.3", features = ["alloc", "std"] }
 hex = "0.4"
 hyper = "0.14" # Prometheus server
 iri-string = "0.5"
 lazy_static = "1.4.0"
 libipld = { version = "0.14", default-features = false, features = ["dag-cbor", "dag-json", "derive", "serde-codec"]}
-libp2p = { default-features = false, features = ["tcp", "mplex", "noise", "ping", "yamux", "dns", "tokio"], version = "0.51.0", git = "https://github.com/kckeiks/rust-libp2p/", branch = "use-quick-protobuf" }
+libp2p = { default-features = false, features = ["autonat", "websocket", "wasm-ext", "gossipsub", "identify", "kad", "tcp", "mplex", "noise", "ping", "yamux", "dns", "tokio", "relay", "dcutr", "macros"], version = "0.51.1" }
 opentelemetry = { version = "0.17.0", features = ["rt-tokio"] }
 opentelemetry-jaeger = { version = "0.16.0", features = ["rt-tokio", "reqwest_collector_client"] }
 pin-project = "1"
@@ -58,7 +59,6 @@ percent-encoding = "2.1"
 path = "lib/"
 
 [workspace]
-
 members = [
     "lib",
     "sdk-wasm",

Dockerfile

@@ -24,6 +24,7 @@ COPY --from=builder /app/target/x86_64-unknown-linux-musl/release/kepler /usr/lo
 USER kepler
 COPY ./kepler.toml ./
 ENV ROCKET_ADDRESS=0.0.0.0
+ENV KEPLER_RELAY_ADDRESSES=[\"/ip4/0.0.0.0/tcp/8081\"]
 EXPOSE 8000
 EXPOSE 8001
 EXPOSE 8081

README.md

@@ -40,14 +40,15 @@ Kepler instances are configured by the [kepler.toml](kepler.toml) configuration
 
 The following common options are available:
 
-| Option               | env var                     | description                                                    |
-|:---------------------|:----------------------------|:---------------------------------------------------------------|
-| log_level            | KEPLER_LOG_LEVEL            | Set the level of logging output, options are "normal", "debug" |
-| address              | KEPLER_ADDRESS              | Set the listening address of the kepler instance               |
-| port                 | KEPLER_PORT                 | Set the listening TCP port for the kepler instance             |
-| storage.blocks.type  | KEPLER_STORAGE_BLOCKS_TYPE  | Set the mode of block storage, options are "Local" and "S3"    |
-| storage.indexes.type | KEPLER_STORAGE_INDEXES_TYPE | Set the type of the index store, options are "Local" and "DynamoDB" |
-| orbits.allowlist     | KEPLER_ORBITS_ALLOWLIST     | Set the URL of an allowlist service for gating the creation of Orbit Peers                                                               |
+| Option               | env var                     | description                                                                |
+|:---------------------|:----------------------------|:---------------------------------------------------------------------------|
+| log_level            | KEPLER_LOG_LEVEL            | Set the level of logging output, options are "normal", "debug"             |
+| address              | KEPLER_ADDRESS              | Set the listening address of the kepler instance                           |
+| port                 | KEPLER_PORT                 | Set the listening TCP port for the kepler instance                         |
+| storage.blocks.type  | KEPLER_STORAGE_BLOCKS_TYPE  | Set the mode of block storage, options are "Local" and "S3"                |
+| storage.indexes.type | KEPLER_STORAGE_INDEXES_TYPE | Set the type of the index store, options are "Local" and "DynamoDB"        |
+| orbits.allowlist     | KEPLER_ORBITS_ALLOWLIST     | Set the URL of an allowlist service for gating the creation of Orbit Peers |
+| relay.addresses      | KEPLER_RELAY_ADDRESSES      | Set the listening multiaddresses for the relay node                        |
 
 ### Storage Config
 

kepler.toml

@@ -17,3 +17,6 @@
 [global.orbits]
 ## Orbit allow list api endpoint
 # allowlist = "http://localhost:10000"
+
+[global.relay]
+# addresses = ["/ip4/127.0.0.1/tcp/8081"]

lib/Cargo.toml

@@ -30,7 +30,7 @@ uuid = { version = "1", features = ["v4", "js"] }
 uuid = { version = "1", features = ["v4"] }
 
 [dependencies.ssi]
-version = "0.5"
+version = "0.6"
 default-features = false
 features = ["ed25519", "rsa", "secp256k1"]
 

src/auth_guards.rs

@@ -2,7 +2,7 @@ use crate::authorization::{Delegation, Invocation, Verifiable};
 use crate::capabilities::store::{InvokeError, ToBlock, Updates};
 use crate::config;
 use crate::orbit::{create_orbit, load_orbit, Orbit};
-use crate::relay::RelayNode;
+use crate::p2p::relay::RelayNode;
 use crate::routes::Metadata;
 use crate::BlockStores;
 use anyhow::Result;
@@ -12,8 +12,9 @@ use kepler_lib::{
     resource::{OrbitId, ResourceId},
 };
 use libp2p::{
-    core::{Multiaddr, PeerId},
-    identity::ed25519::Keypair as Ed25519Keypair,
+    core::{multiaddr::multiaddr, Multiaddr},
+    identity::Keypair,
+    PeerId,
 };
 use rocket::{
     futures::future::try_join_all,
@@ -75,14 +76,16 @@ pub fn check_orbit_and_service(
 }
 
 fn get_state(req: &Request<'_>) -> Result<(config::Config, (PeerId, Multiaddr))> {
+    let config = req
+        .rocket()
+        .state::<config::Config>()
+        .cloned()
+        .ok_or_else(|| anyhow!("Could not retrieve config"))?;
     Ok((
-        req.rocket()
-            .state::<config::Config>()
-            .cloned()
-            .ok_or_else(|| anyhow!("Could not retrieve config"))?,
+        config,
         req.rocket()
             .state::<RelayNode>()
-            .map(|r| (r.id, r.internal()))
+            .map(|r| (*r.id(), multiaddr!(Memory(1u8))))
             .ok_or_else(|| anyhow!("Could not retrieve relay node information"))?,
     ))
 }
@@ -153,17 +156,15 @@ impl<'l> FromRequest<'l> for DelegateAuthWrapper {
                         .await,
                     ) {
                         (Some(p), Ok(None)) => {
-                            let keys = match req
-                                .rocket()
-                                .state::<RwLock<HashMap<PeerId, Ed25519Keypair>>>()
-                            {
-                                Some(k) => k,
-                                _ => {
-                                    return Err(internal_server_error(anyhow!(
-                                        "could not retrieve open key set"
-                                    )))
-                                }
-                            };
+                            let keys =
+                                match req.rocket().state::<RwLock<HashMap<PeerId, Keypair>>>() {
+                                    Some(k) => k,
+                                    _ => {
+                                        return Err(internal_server_error(anyhow!(
+                                            "could not retrieve open key set"
+                                        )))
+                                    }
+                                };
 
                             if let Err(e) = token
                                 .verify(

src/authorization.rs

@@ -514,7 +514,7 @@ mod test {
         },
     };
 
-    async fn gen(
+    async fn _gen(
         iss: &JWK,
         aud: String,
         caps: Vec<Capability>,
@@ -530,10 +530,10 @@ mod test {
                 .await
                 .1
                 .unwrap(),
-            gen_ucan((iss, did), aud, caps, exp, prf).await,
+            _gen_ucan((iss, did), aud, caps, exp, prf).await,
         )
     }
-    async fn gen_ucan(
+    async fn _gen_ucan(
         iss: (&JWK, String),
         audience: String,
         attenuation: Vec<Capability>,

src/config.rs

@@ -3,6 +3,7 @@ use crate::{
     storage::{file_system::FileSystemConfig, s3::S3BlockConfig},
     BlockConfig,
 };
+use libp2p::{build_multiaddr, Multiaddr};
 use rocket::http::hyper::Uri;
 use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DisplayFromStr, FromInto};
@@ -81,8 +82,7 @@ pub struct DynamoStorage {
 
 #[derive(Serialize, Deserialize, Debug, Clone, Hash, PartialEq, Eq)]
 pub struct Relay {
-    pub address: String,
-    pub port: u16,
+    pub addresses: Vec<Multiaddr>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, Hash, PartialEq, Eq)]
@@ -122,8 +122,7 @@ impl Default for LocalIndexStorage {
 impl Default for Relay {
     fn default() -> Self {
         Self {
-            address: "127.0.0.1".into(),
-            port: 8081,
+            addresses: vec![build_multiaddr!(Ip4([127, 0, 0, 1]), Tcp(8081u16))],
         }
     }
 }

src/lib.rs

@@ -14,26 +14,23 @@ pub mod allow_list;
 pub mod auth_guards;
 pub mod authorization;
 pub mod capabilities;
-pub mod cas;
 pub mod config;
 pub mod indexes;
 pub mod kv;
 pub mod manifest;
 pub mod orbit;
+pub mod p2p;
 pub mod prometheus;
-pub mod relay;
 pub mod routes;
 pub mod storage;
 mod tracing;
 pub mod transport;
 
 use config::{BlockStorage, Config};
-use libp2p::{
-    identity::{ed25519::Keypair as Ed25519Keypair, Keypair},
-    PeerId,
-};
+use libp2p::{build_multiaddr, identity::Keypair, PeerId};
 use orbit::ProviderUtils;
-use relay::RelayNode;
+use p2p::relay::Config as RelayConfig;
+use p2p::transport::{Both, MemoryConfig, TcpConfig};
 use routes::{delegate, invoke, open_host_key, relay_addr, util_routes::*};
 use std::{collections::HashMap, sync::RwLock};
 use storage::{
@@ -76,9 +73,22 @@ pub async fn app(config: &Figment) -> Result<Rocket<Build>> {
     tracing::tracing_try_init(&kepler_config.log);
 
     storage::KV::healthcheck(kepler_config.storage.indexes.clone()).await?;
-    let kp = kepler_config.storage.blocks.relay_key_pair().await?;
 
-    let relay_node = RelayNode::new(kepler_config.relay.port, Keypair::Ed25519(kp)).await?;
+    let mut relay_node = RelayConfig::default().launch(
+        kepler_config.storage.blocks.relay_key_pair().await?,
+        Both::<MemoryConfig, TcpConfig>::default(),
+    )?;
+
+    relay_node
+        .listen_on(
+            kepler_config
+                .relay
+                .addresses
+                .iter()
+                .cloned()
+                .chain([build_multiaddr!(Memory(1u64))]),
+        )
+        .await?;
 
     let routes = routes![
         healthcheck,
@@ -96,7 +106,7 @@ pub async fn app(config: &Figment) -> Result<Rocket<Build>> {
             header_name: kepler_config.log.tracing.traceheader,
         })
         .manage(relay_node)
-        .manage(RwLock::new(HashMap::<PeerId, Ed25519Keypair>::new()));
+        .manage(RwLock::new(HashMap::<PeerId, Keypair>::new()));
 
     if kepler_config.cors {
         Ok(rocket.attach(AdHoc::on_response("CORS", |_, resp| {

src/orbit.rs

@@ -13,11 +13,7 @@ use kepler_lib::libipld::cid::{
     Cid,
 };
 use kepler_lib::resource::OrbitId;
-use libp2p::{
-    core::Multiaddr,
-    identity::{ed25519::Keypair as Ed25519Keypair, PublicKey},
-    PeerId,
-};
+use libp2p::{core::Multiaddr, identity::Keypair, PeerId};
 use rocket::tokio::task::JoinHandle;
 
 use cached::proc_macro::cached;
@@ -56,7 +52,7 @@ pub struct Orbit<B> {
 #[derive(Clone, Debug, Builder)]
 pub struct OrbitPeerConfig<B, I = config::IndexStorage> {
     #[builder(setter(into))]
-    identity: Ed25519Keypair,
+    identity: Keypair,
     #[builder(setter(into))]
     manifest: Manifest,
     #[builder(setter(into, strip_option), default)]
@@ -78,7 +74,7 @@ where
         B::Error: 'static,
     {
         let id = config.manifest.id().get_cid();
-        let _local_peer_id = PeerId::from_public_key(&PublicKey::Ed25519(config.identity.public()));
+        let _local_peer_id = PeerId::from_public_key(&config.identity.public());
         let _relay = &config.relay;
 
         let blocks = match config.blocks.open(config.manifest.id()).await? {
@@ -106,7 +102,7 @@ where
         B::Error: 'static,
     {
         let id = config.manifest.id().get_cid();
-        let _local_peer_id = PeerId::from_public_key(&PublicKey::Ed25519(config.identity.public()));
+        let _local_peer_id = PeerId::from_public_key(&config.identity.public());
         let _relay = &config.relay;
 
         let blocks = config.blocks.create(config.manifest.id()).await?;
@@ -129,9 +125,9 @@ where
 pub trait ProviderUtils {
     type Error: StdError;
     async fn exists(&self, orbit: &OrbitId) -> Result<bool, Self::Error>;
-    async fn relay_key_pair(&self) -> Result<Ed25519Keypair, Self::Error>;
-    async fn key_pair(&self, orbit: &OrbitId) -> Result<Option<Ed25519Keypair>, Self::Error>;
-    async fn setup_orbit(&self, orbit: &OrbitId, key: &Ed25519Keypair) -> Result<(), Self::Error>;
+    async fn relay_key_pair(&self) -> Result<Keypair, Self::Error>;
+    async fn key_pair(&self, orbit: &OrbitId) -> Result<Option<Keypair>, Self::Error>;
+    async fn setup_orbit(&self, orbit: &OrbitId, key: &Keypair) -> Result<(), Self::Error>;
 }
 
 // Using Option to distinguish when the orbit already exists from a hard error
@@ -140,7 +136,7 @@ pub async fn create_orbit(
     store_config: &BlockConfig,
     index_config: &config::IndexStorage,
     relay: (PeerId, Multiaddr),
-    kp: Ed25519Keypair,
+    kp: Keypair,
 ) -> Result<Option<Orbit<BlockStores>>> {
     match Manifest::resolve_dyn(id, None).await? {
         Some(_) => {}
@@ -254,7 +250,7 @@ mod tests {
         Ok((
             Orbit::create(
                 &OrbitPeerConfigBuilder::<BlockConfig, IndexStorage>::default()
-                    .identity(Ed25519Keypair::generate())
+                    .identity(Keypair::generate_ed25519())
                     .manifest(md)
                     .blocks(BlockConfig::B(FileSystemConfig::new(dir.path())))
                     .index(IndexStorage::Local(LocalIndexStorage {
@@ -284,6 +280,6 @@ mod tests {
 
         let md = Manifest::resolve_dyn(&oid, None).await.unwrap().unwrap();
 
-        let (orbit, dir) = op(md).await.unwrap();
+        let (_, _) = op(md).await.unwrap();
     }
 }