feat: Add P2P functionality

Cargo.toml

@@ -66,6 +66,7 @@ color-print = "0.3.4"
 criterion = "0.5.1"
 digest = "0.10.7"
 docify = { version = "0.2.8" }
+ed25519-dalek = { version = "2.1.1", default-features = false }
 env_logger = "0.11.2"
 ff = "0.13.0"
 futures = "0.3.28"
@@ -78,6 +79,7 @@ ipld-core = "0.4.1"
 ipld-dagpb = "0.2.1"
 itertools = "0.13.0"
 jsonrpsee = { version = "0.24.7" }
+libp2p = { version = "0.54", default-features = false }
 log = { version = "0.4.21", default-features = false }
 multihash-codetable = { version = "0.1.1", default-features = false }
 num-bigint = { version = "0.4.5", default-features = false }
@@ -109,6 +111,7 @@ thiserror = { version = "2.0.3", default-features = false }
 tokio = "1.37.0"
 tokio-stream = "0.1.15"
 tokio-util = "0.7.11"
+toml = "0.8.19"
 tower = "0.4.13"
 tower-http = "0.5.2"
 tracing = "0.1.40"

storage-provider/client/Cargo.toml

@@ -20,8 +20,10 @@ bls12_381 = { workspace = true }
 cid = { workspace = true, features = ["std"] }
 clap = { workspace = true, features = ["derive"] }
 codec = { workspace = true }
+ed25519-dalek = { workspace = true, features = ["pem", "pkcs8"] }
 hex = { workspace = true }
 jsonrpsee = { workspace = true, features = ["http-client", "macros", "server", "ws-client"] }
+libp2p.workspace = true
 sc-cli = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

storage-provider/client/src/commands/mod.rs

@@ -1,8 +1,16 @@
 mod proofs;
 mod wallet;
 
+use std::{
+    fs,
+    io::{self, Write},
+    path::PathBuf,
+};
+
 use clap::Parser;
+use ed25519_dalek::pkcs8::{DecodePublicKey, PublicKeyBytes};
 use jsonrpsee::core::ClientError;
+use libp2p::{identity::ed25519::PublicKey as EdPubKey, PeerId};
 use polka_storage_provider_common::rpc::StorageProviderRpcClient;
 use storagext::{
     deser::DeserializablePath,
@@ -47,6 +55,12 @@ pub enum CliError {
 
     #[error("no signer key was provider")]
     NoSigner,
+
+    #[error(transparent)]
+    PubKeyError(#[from] ed25519_dalek::pkcs8::spki::Error),
+
+    #[error(transparent)]
+    DecodingError(#[from] libp2p::identity::DecodingError),
 }
 
 /// A CLI application that facilitates management operations over a running full
@@ -98,6 +112,18 @@ pub(crate) enum Cli {
         #[command(flatten)]
         signer_key: MultiPairArgs,
     },
+
+    /// Generate a Peer ID from a ED25519 public key pem file
+    GeneratePeerID {
+        /// Path to the ED25519 public key pem file
+        #[arg(long)]
+        pubkey: PathBuf,
+
+        /// Path to the output file to write the peer ID to.
+        /// If `None` the peer ID will be printed to stdout.
+        #[arg(long)]
+        file: Option<PathBuf>,
+    },
 }
 
 impl Cli {
@@ -131,6 +157,7 @@ impl Cli {
                 deal_proposal,
                 signer_key,
             } => Self::sign_deal(deal_proposal, signer_key),
+            Self::GeneratePeerID { pubkey, file } => Self::generate_peer_id(pubkey, file),
         }
     }
 
@@ -182,4 +209,17 @@ impl Cli {
         );
         Ok(())
     }
+
+    fn generate_peer_id(path: PathBuf, file: Option<PathBuf>) -> Result<(), CliError> {
+        let pubkey_bytes = PublicKeyBytes::read_public_key_pem_file(path)?;
+        let pubkey = EdPubKey::try_from_bytes(&pubkey_bytes.to_bytes())?;
+        let key = libp2p::identity::PublicKey::from(pubkey);
+        let peer_id = PeerId::from_public_key(&key).to_string();
+
+        match file {
+            None => io::stdout().lock().write_all(&peer_id.as_bytes())?,
+            Some(path) => fs::write(path, &peer_id.as_bytes())?,
+        }
+        Ok(())
+    }
 }

storage-provider/server/Cargo.toml

@@ -24,10 +24,12 @@ axum = { workspace = true, features = ["macros", "multipart"] }
 cid = { workspace = true, features = ["serde", "std"] }
 clap = { workspace = true, features = ["derive"] }
 codec = { workspace = true }
+ed25519-dalek = { workspace = true, features = ["pem", "std"] }
 futures = { workspace = true }
 hex = { workspace = true, features = ["std"] }
 hyper = { workspace = true }
 jsonrpsee = { workspace = true, features = ["http-client", "macros", "server", "ws-client"] }
+libp2p = { workspace = true, features = ["identify", "macros", "noise", "rendezvous", "tcp", "tokio", "yamux"] }
 rand = { workspace = true }
 rocksdb = { workspace = true }
 sc-cli = { workspace = true }
@@ -38,6 +40,7 @@ tempfile = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true, features = ["rt"] }
+toml.workspace = true
 tower = { workspace = true }
 tower-http = { workspace = true, features = ["trace"] }
 tracing = { workspace = true }

storage-provider/server/p2p-example-config/bootstrap.toml

@@ -0,0 +1,2 @@
+address = "/ip4/127.0.0.1/tcp/62649"
+key_path = "storage-provider/server/p2p-example-config/private.pem"

storage-provider/server/p2p-example-config/private.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIKo5QJZ40QKfNdrdMyjJeYagahGMgT7ibZsIuUeC3Yrw
+-----END PRIVATE KEY-----

storage-provider/server/p2p-example-config/public.pem

@@ -0,0 +1,3 @@
+-----BEGIN PUBLIC KEY-----
+MCowBQYDK2VwAyEA8noBMkYU3PmEVHWpu78JrIoeKjZcC8jX1SgRI8ZBPok=
+-----END PUBLIC KEY-----

storage-provider/server/p2p-example-config/register.toml

@@ -0,0 +1,3 @@
+key_path = "storage-provider/server/p2p-example-config/private.pem"
+rendezvous_point = "12D3KooWS8tjUoiMgh2hZ12EyDjnJuEoa1zxq7G7EQtqafRUT8fr"
+rendezvous_point_address = "/ip4/127.0.0.1/tcp/62649"

storage-provider/server/src/main.rs

@@ -3,13 +3,15 @@
 #![deny(clippy::unwrap_used)]
 
 mod db;
+mod p2p;
 mod pipeline;
 mod rpc;
 mod storage;
 
 use std::{env::temp_dir, net::SocketAddr, num::NonZero, path::PathBuf, sync::Arc, time::Duration};
 
 use clap::Parser;
+use p2p::{start_p2p_node, NodeType, P2PError};
 use pipeline::types::PipelineMessage;
 use polka_storage_proofs::{
     porep::{self, PoRepParameters},
@@ -165,6 +167,9 @@ pub enum ServerError {
 
     #[error(transparent)]
     Join(#[from] JoinError),
+
+    #[error(transparent)]
+    P2P(#[from] P2PError),
 }
 
 /// The server arguments, as passed by the user, unvalidated.
@@ -244,6 +249,14 @@ pub struct ServerArguments {
     /// E.g. With 2KiB sector sizes and 16GiB of RAM, it goes OOM at 4 parallel.
     #[arg(long, default_value = "2")]
     parallel_prove_commits: NonZero<usize>,
+
+    /// P2P Node type, can be either a bootstrap node or a registration node.
+    #[arg(long)]
+    node_type: NodeType,
+
+    /// Path to P2P config file
+    #[arg(long)]
+    p2p_config: PathBuf,
 }
 
 /// A valid server configuration. To be created using [`ServerConfiguration::try_from`].
@@ -286,6 +299,12 @@ pub struct ServerConfiguration {
 
     /// The number of prove commits to be run in parallel.
     parallel_prove_commits: usize,
+
+    /// P2P Network node type, can either be a bootstrap or registration node
+    node_type: NodeType,
+
+    /// P2P Network node configuration file path
+    p2p_config: PathBuf,
 }
 
 impl TryFrom<ServerArguments> for ServerConfiguration {
@@ -342,12 +361,16 @@ impl TryFrom<ServerArguments> for ServerConfiguration {
             porep_parameters,
             post_parameters,
             parallel_prove_commits: value.parallel_prove_commits.get(),
+            node_type: value.node_type,
+            p2p_config: value.p2p_config,
         })
     }
 }
 
 impl ServerConfiguration {
     pub async fn run(self) -> Result<(), ServerError> {
+        let config = self.p2p_config.clone();
+        let node_type = self.node_type;
         let SetupOutput {
             storage_state,
             rpc_state,
@@ -370,6 +393,11 @@ impl ServerConfiguration {
             pipeline_rx,
             cancellation_token.child_token(),
         ));
+        let p2p_task = tokio::spawn(start_p2p_node(
+            node_type,
+            config,
+            cancellation_token.child_token(),
+        ));
 
         // Wait for SIGTERM on the main thread and once received "unblock"
         tokio::signal::ctrl_c()
@@ -381,8 +409,8 @@ impl ServerConfiguration {
         tracing::info!("sent shutdown signal");
 
         // Wait for the tasks to finish
-        let (upload_result, rpc_task, pipeline_task) =
-            tokio::join!(storage_task, rpc_task, pipeline_task);
+        let (upload_result, rpc_task, pipeline_task, p2p_task) =
+            tokio::join!(storage_task, rpc_task, pipeline_task, p2p_task);
 
         // Log errors
         let upload_result = upload_result
@@ -402,10 +430,17 @@ impl ServerConfiguration {
                 let _ = ok.as_ref().inspect_err(|err| tracing::error!(%err));
             });
 
+        let p2p_task = p2p_task
+            .inspect_err(|err| tracing::error!(%err))
+            .inspect(|ok| {
+                let _ = ok.as_ref().inspect_err(|err| tracing::error!(%err));
+            });
+
         // Exit with error
         upload_result??;
         rpc_task??;
         pipeline_task??;
+        p2p_task??;
 
         Ok(())
     }