janky but runs

Cargo.lock

@@ -364,6 +364,15 @@ version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "325918d6fe32f23b19878fe4b34794ae41fc19ddbe53b10571a4874d44ffd39b"
 
+[[package]]
+name = "cbor4ii"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "472931dd4dfcc785075b09be910147f9c6258883fc4591d0dac6116392b2daa6"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "cc"
 version = "1.2.6"
@@ -1945,6 +1954,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1356c9e376a94a75ae830c42cdaea3d4fe1290ba409a22c809033d1b7dcab0a6"
 dependencies = [
  "async-trait",
+ "cbor4ii",
  "futures",
  "futures-bounded",
  "futures-timer",
@@ -1952,6 +1962,7 @@ dependencies = [
  "libp2p-identity",
  "libp2p-swarm",
  "rand",
+ "serde",
  "smallvec",
  "tracing",
  "void",

Cargo.toml

@@ -8,13 +8,17 @@ publish = false
 [package.metadata.release]
 release = false
 
-[[bin]] # Bin to run the server
-name = "cli"
-path = "app/cli.rs"
-# 
-# [[bin]] # Bin to run the client
-# name = "client"
-# path = "app/client.rs"
+[[bin]] # Admin
+name = "admin"
+path = "app/admin.rs"
+
+[[bin]] # Server
+name = "server"
+path = "app/server.rs"
+
+[[bin]] # Client
+name = "client"
+path = "app/client.rs"
 
 [dependencies]
 anyhow = "1.0.95"
@@ -23,7 +27,7 @@ cryptoxide = "0.4.4"
 ed25519-dalek = { version = "2.1.1", features = ["digest"] }
 futures = "0.3.30"
 hex = "0.4.3"
-libp2p = { version = "0.54.1", features = ["tokio", "gossipsub", "mdns", "noise", "macros", "tcp", "yamux", "quic", "identify", "ping", "relay", "dcutr", "rendezvous", "kad", "request-response"] }
+libp2p = { version = "0.54.1", features = ["tokio", "gossipsub", "mdns", "noise", "macros", "tcp", "yamux", "quic", "identify", "ping", "relay", "dcutr", "rendezvous", "kad", "request-response", "cbor"] }
 libp2p-identity = { version = "0.2.9", features = ["ed25519", "peerid"] }
 owo-colors = "4.1.0"
 quick-protobuf = "0.8.1"

README.md

@@ -9,6 +9,24 @@ establish our key dependencies.
 
 This repo use nix flakes with a shell available. Otherwise ymmv.
 
+Build:
+
+```sh
+cargo build
+```
+
+Run binary `x`:
+
+```sh
+cargo run --bin <x>
+```
+
+There are three binaries:
+
+- `admin` - admin controls: create, add, revoke ephemeral keys
+- `server` - run the signing server
+- `client` - run a client
+
 ## Context
 
 CL depends on the use on signing/verification key pair cryptography. The signing

app/admin.rs

@@ -1,10 +1,10 @@
-use std::env;
-
 use clap::{Parser, Subcommand};
 use ed25519_dalek::{Signature, SigningKey};
-use sqlx::sqlite::SqlitePool;
 
-use cll2v0::{db, keys};
+use cll2v0::{
+    db,
+    keys::{self},
+};
 
 /// cll2v0 is a playground for rust libraries.
 /// This is signing service.
@@ -46,15 +46,7 @@ enum Command {
 async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
-    let keys_path = env::var("CLL2V0_KEYS").expect("Expect `CLL2V0_KEYS` to be set");
-    let keychain = keys::mk_keychain(&keys_path);
-
-    let db_url = env::var("DATABASE_URL").expect("Expect `DATABASE_URL` to be set");
-    let pool = SqlitePool::connect(&db_url).await?;
-    for vkey_bytes in keychain.keys() {
-        let vkey = keys::from_bytes(vkey_bytes.into())?;
-        let _ = db::add_persistent_key(&pool, &vkey).await;
-    }
+    let (_, pool) = db::start_db().await?;
 
     match args.cmd {
         Some(Command::Add {
@@ -78,7 +70,11 @@ async fn main() -> anyhow::Result<()> {
             let res = db::list_persistent_keys(&pool).await?;
             println!("Persistent keys:");
             res.into_iter()
-                .for_each(|key| println!("{}", keys::to_hex(&key)))
+                .for_each(|key| println!("{}", keys::to_hex(&key)));
+            let res = db::list_ephemeral_keys(&pool).await?;
+            println!("ephemeral keys:");
+            res.into_iter()
+                .for_each(|(e, p, t)| println!("{} {} {t}", keys::to_hex(&e), keys::to_hex(&p)))
         }
         Some(Command::Gen { seed }) => {
             let ekey = SigningKey::from_bytes(&[

app/client.rs

@@ -0,0 +1,159 @@
+use ed25519_dalek::ed25519::signature::SignerMut;
+use std::error::Error;
+use tokio::{io, io::AsyncBufReadExt, select};
+
+use futures::prelude::*;
+use libp2p::{
+    mdns,
+    request_response::{self, ProtocolSupport},
+    swarm::SwarmEvent,
+    Multiaddr,
+};
+use tracing_subscriber::EnvFilter;
+
+use cll2v0::{
+    keys,
+    messages::MyRequest,
+    protocol::{mk_swarm, MyBehaviourEvent},
+};
+
+use clap::{Parser, Subcommand};
+
+/// cll2v0 is a playground for rust libraries.
+/// This is signing service.
+#[derive(Parser)]
+#[command(arg_required_else_help(true), version, about)]
+struct Args {
+    #[command(subcommand)]
+    cmd: Option<Command>,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Start server
+    Start(ClientParams),
+}
+
+#[derive(clap::Args, Debug)]
+pub struct ClientParams {
+    /// Root of the multiaddr
+    #[arg(
+        long,
+        default_value = "/ip4/0.0.0.0/tcp/52321/p2p/12D3KooWPADMrTD3njNGBwyiqFEaDSarki8bQJdiFtofo8sGXs1o"
+    )]
+    pub server_addr: String,
+    // pub listen_on : String,
+    /// If used in prod, this should be an envvar.
+    #[arg(
+        long,
+        default_value = "0000000000000000000000000000000000000000000000000000000000000000"
+    )]
+    pub skey: String,
+    // FIXME :: rust does not like this option /// If turn trace on
+    // pub trace: bool,
+}
+
+fn main() -> Result<(), Box<dyn Error>> {
+    match Args::parse().cmd {
+        Some(Command::Start(sp)) => tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(start(sp)),
+        _ => {
+            panic!("oops")
+        }
+    }
+}
+
+pub async fn start(params: ClientParams) -> Result<(), Box<dyn Error>> {
+    println!("params {:?}", params);
+    let ClientParams {
+        server_addr,
+        skey: skey_hex,
+    } = params;
+    if true {
+        let _ = tracing_subscriber::fmt()
+            .with_env_filter(EnvFilter::from_default_env())
+            .try_init();
+    }
+
+    let key = keys::signing_key_from_hex(&skey_hex)?;
+    println!("VERIFYING : {}", keys::to_hex(&key.verifying_key()));
+
+    let keypair = keys::libp2p_kp_from_hex(skey_hex)?;
+
+    let mut swarm = mk_swarm(&keypair.clone().into(), ProtocolSupport::Outbound)?;
+    swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
+
+    // Read full lines from stdin
+    let mut stdin = io::BufReader::new(io::stdin()).lines();
+
+    let server_addr: libp2p::Multiaddr = server_addr.parse()?;
+    let server_peer_id = match server_addr.clone().pop() {
+        Some(libp2p::multiaddr::Protocol::P2p(peer_id)) => peer_id,
+        _ => panic!("must end in peer"),
+    };
+    println!("server addr : {:?}, {:?}", server_addr, server_peer_id);
+    let _res = swarm.dial(server_addr)?;
+
+    loop {
+        select! {
+            Ok(Some(line)) = stdin.next_line() => {
+                println!("MESSAGE: {}", line);
+                let bytes = line.into_bytes();
+                let req = MyRequest {
+                        key: keypair.clone().public().to_bytes(),
+                        body: bytes.clone(),
+                        sig: keys::sign(&mut key.clone(), bytes).to_vec(),
+                    };
+                println!("REQ : {:?}", req);
+                swarm.behaviour_mut().req_res.send_request(
+                    &server_peer_id,
+                    req,
+                );
+            }
+            event = swarm.select_next_some() => match event {
+                // MDNS BEHAVIOR
+                SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Discovered(list))) => {
+                    for (peer_id, multiaddr) in list {
+                        println!("add peer {:?}, {:?}", peer_id, multiaddr.clone());
+                        // swarm.dial(multiaddr)?;
+                    }
+                }
+                // SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Expired(list))) => {
+                //     for (peer_id, _multiaddr) in list {
+                //         println!("mDNS discover peer has expired: {peer_id}");
+                //         // swarm.behaviour_mut().gossipsub.remove_explicit_peer(&peer_id);
+                //     }
+                // }
+                // SwarmEvent::Behaviour(MyBehaviourEvent::ReqRes(request_response::Event::Message {
+                //     peer: _peer,
+                //     message:
+                //         libp2p::request_response::Message::Request {
+                //             request, channel, ..
+                //         },
+                // })) => {
+                //     // println!("Req : {:?} {:?} {:?}", peer, channel, hex::encode(request.body.clone()), );
+                //     println!("REQSIG : {}", hex::encode(request.sig.clone()),);
+                //     let _ = swarm.behaviour_mut().req_res.send_response(
+                //         channel,
+                //         MyResponse {
+                //             sig: keypair.sign(&request.body),
+                //         },
+                //     );
+                // }
+                SwarmEvent::Behaviour(MyBehaviourEvent::ReqRes(request_response::Event::Message {
+                    peer: _peer,
+                    message: libp2p::request_response::Message::Response { response, .. },
+                })) => {
+                    // println!("response : {:?} {:?} {:?}", peer, request_id, "" );
+                    println!("RESSIG : {}", hex::encode(response.sig.clone()),);
+                }
+                e => {
+                    println!("OTHER {:?}", e)
+                }
+            }
+        }
+    }
+}

app/server.rs

@@ -0,0 +1,129 @@
+use std::error::Error;
+
+use ed25519_dalek::{ed25519::signature::SignerMut, Signature};
+use futures::prelude::*;
+use libp2p::{
+    mdns,
+    request_response::{self, ProtocolSupport},
+    swarm::SwarmEvent,
+};
+use tracing_subscriber::EnvFilter;
+
+use cll2v0::{
+    db, keys,
+    messages::{MyRequest, MyResponse},
+    protocol::{mk_swarm, MyBehaviourEvent},
+};
+
+use clap::{Parser, Subcommand};
+
+/// cll2v0 is a playground for rust libraries.
+/// This is signing service.
+#[derive(Parser)]
+#[command(arg_required_else_help(true), version, about)]
+struct Args {
+    #[command(subcommand)]
+    cmd: Option<Command>,
+}
+
+#[derive(Subcommand)]
+enum Command {
+    /// Start server
+    Start(ServerParams),
+}
+
+#[derive(clap::Args, Debug)]
+pub struct ServerParams {
+    /// Server listening address
+    #[arg(long, default_value = "/ip4/0.0.0.0/tcp/52321")]
+    pub listen_on: String,
+    /// Signing key of server libp2p part
+    #[arg(
+        long,
+        default_value = "deadbeef00000000000000000000000000000000000000000000000000000000"
+    )]
+    pub skey: String,
+}
+
+fn main() -> Result<(), Box<dyn Error>> {
+    match Args::parse().cmd {
+        Some(Command::Start(params)) => tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+            .block_on(start(params)),
+        _ => {
+            panic!("oops")
+        }
+    }
+}
+
+pub async fn start(params: ServerParams) -> Result<(), Box<dyn Error>> {
+    let ServerParams {
+        skey: skey_hex,
+        listen_on,
+    } = params;
+    if true {
+        let _ = tracing_subscriber::fmt()
+            .with_env_filter(EnvFilter::from_default_env())
+            .try_init();
+    }
+    let keypair = keys::libp2p_kp_from_hex(skey_hex)?;
+    let peer_id = libp2p_identity::PublicKey::from(keypair.public()).to_peer_id();
+    println!("PEER_ID : {}", peer_id);
+
+    let (keychain, pool) = db::start_db().await?;
+
+    let mut swarm = mk_swarm(&keypair.clone().into(), ProtocolSupport::Inbound)?;
+    swarm.listen_on(listen_on.parse()?)?;
+
+    loop {
+        match swarm.select_next_some().await {
+            // MDNS BEHAVIOR
+            SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Expired(list))) => {
+                for (peer_id, _multiaddr) in list {
+                    println!("mDNS discover peer has expired: {peer_id}");
+                    // swarm.behaviour_mut().gossipsub.remove_explicit_peer(&peer_id);
+                }
+            }
+            SwarmEvent::Behaviour(MyBehaviourEvent::ReqRes(request_response::Event::Message {
+                peer: _peer,
+                message:
+                    libp2p::request_response::Message::Request {
+                        request, channel, ..
+                    },
+            })) => {
+                // println!("Req : {:?} {:?} {:?}", peer, channel, hex::encode(request.body.clone()), );
+                println!("REQSIG : {}", hex::encode(request.sig.clone()),);
+                let MyRequest { key, body, sig } = request;
+
+                // FIXME :: MAP ERROR.
+                let sig_arr: [u8; 64] = sig.clone().try_into().unwrap();
+
+                if let Ok(ekey) = keys::from_bytes(key.into()) {
+                    if let Ok(_) = ekey.verify_strict(&body, &Signature::from_bytes(&sig_arr)) {
+                        if let Ok(pkey) = db::get_persistent_key(&pool, &ekey).await {
+                            if let Some(skey) = keychain.get(&pkey.to_bytes()) {
+                                let _ = swarm.behaviour_mut().req_res.send_response(
+                                    channel,
+                                    MyResponse {
+                                        sig: skey.clone().sign(&body).to_bytes().into(),
+                                    },
+                                );
+                            } else {
+                                println!("err0")
+                            };
+                        } else {
+                            println!("err1");
+                        }
+                    } else {
+                        println!("err2");
+                    }
+                }
+            }
+            e => {
+                println!("OTHER {:?}", e)
+            }
+        }
+    }
+}

src/arr.rs

@@ -0,0 +1,8 @@
+pub fn arr32(b: Vec<u8>) -> anyhow::Result<[u8; 32]> {
+    println!("{}", b.len());
+    b.try_into().map_err(|_| anyhow::anyhow!("bad input"))
+}
+
+pub fn arr64(b: Vec<u8>) -> anyhow::Result<[u8; 64]> {
+    b.try_into().map_err(|_| anyhow::anyhow!("bad input"))
+}

src/db.rs

@@ -1,7 +1,22 @@
+use std::env;
+
 use ed25519_dalek::VerifyingKey;
 use sqlx::sqlite::SqlitePool;
 
-use crate::keys;
+use crate::keys::{self, Keychain};
+
+pub async fn start_db() -> Result<(Keychain, SqlitePool), anyhow::Error> {
+    let keys_path = env::var("CLL2V0_KEYS").expect("Expect `CLL2V0_KEYS` to be set");
+    let keychain = keys::mk_keychain(&keys_path);
+
+    let db_url = env::var("DATABASE_URL").expect("Expect `DATABASE_URL` to be set");
+    let pool = SqlitePool::connect(&db_url).await?;
+    for vkey_bytes in keychain.keys() {
+        let vkey = keys::from_bytes(vkey_bytes.into())?;
+        let _ = add_persistent_key(&pool, &vkey).await;
+    }
+    Ok((keychain, pool))
+}
 
 pub async fn add_persistent_key(pool: &SqlitePool, vkey: &VerifyingKey) -> anyhow::Result<i64> {
     let mut conn = pool.acquire().await?;
@@ -53,7 +68,8 @@ pub async fn get_persistent_key(
 ) -> anyhow::Result<VerifyingKey> {
     let b = keys::to_bytes(ephemeral_key);
     let rec = sqlx::query!(
-        r#" SELECT persistent_key FROM ephemeral_keys WHERE id = ? AND expires_at > date(); "#,
+        r#" SELECT persistent_key FROM ephemeral_keys WHERE id = ?;"#,
+        // r#" SELECT persistent_key FROM ephemeral_keys WHERE id = ? AND expires_at > date(); "#,
         b,
     )
     .fetch_one(pool)
@@ -72,3 +88,33 @@ pub async fn list_persistent_keys(pool: &SqlitePool) -> anyhow::Result<Vec<Verif
         .collect::<Vec<VerifyingKey>>();
     Ok(keys)
 }
+
+pub async fn list_ephemeral_keys(
+    pool: &SqlitePool,
+) -> anyhow::Result<Vec<(VerifyingKey, VerifyingKey, i64)>> {
+    let recs =
+        sqlx::query!(r#" SELECT id, persistent_key, expires_at FROM ephemeral_keys ORDER BY id "#)
+            .fetch_all(pool)
+            .await?;
+    let now: i64 = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_millis()
+        .try_into()
+        .unwrap();
+    let keys = recs
+        .into_iter()
+        .map(|b| {
+            (
+                keys::from_bytes(b.id).expect("illegal"),
+                keys::from_bytes(b.persistent_key).expect("illegal"),
+                if b.expires_at > now {
+                    <i64>::try_from(b.expires_at - now).unwrap()
+                } else {
+                    0 - <i64>::try_from(now - b.expires_at).unwrap()
+                },
+            )
+        })
+        .collect::<Vec<(VerifyingKey, VerifyingKey, i64)>>();
+    Ok(keys)
+}

src/keys.rs

@@ -2,7 +2,10 @@ use std::{collections::HashMap, fs::read_to_string};
 
 use ed25519_dalek::{ed25519::signature::SignerMut, Signature, SigningKey, VerifyingKey};
 
+use crate::arr::arr32;
+
 pub fn from_bytes(bytes: Vec<u8>) -> Result<VerifyingKey, anyhow::Error> {
+    println!("{}", bytes.len());
     let arr: [u8; 32] = TryInto::try_into(bytes).expect("");
     Ok(VerifyingKey::from_bytes(&arr)?)
 }
@@ -54,3 +57,16 @@ pub fn mk_keychain(keys_path: &str) -> Keychain {
     }
     keychain
 }
+
+pub fn libp2p_kp_from_hex(s: String) -> Result<libp2p_identity::ed25519::Keypair, anyhow::Error> {
+    let skey_arr = hex::decode(s)?; // .expect("Cannot fromdecode hex");
+    let skey = libp2p_identity::ed25519::SecretKey::try_from_bytes(skey_arr)?;
+    Ok(libp2p_identity::ed25519::Keypair::from(skey))
+}
+
+pub fn libp2p_pubkey_from_hex(
+    s: &String,
+) -> Result<libp2p_identity::ed25519::PublicKey, anyhow::Error> {
+    let pk = libp2p_identity::ed25519::PublicKey::try_from_bytes(&arr32(hex::decode(s)?)?)?;
+    Ok(pk)
+}

src/lib.rs

@@ -1,5 +1,7 @@
+pub mod arr;
 pub mod db;
 pub mod hash;
 pub mod keys;
-pub mod protobuf;
-pub mod server;
+pub mod messages;
+//pub mod protobuf;
+pub mod protocol;

src/messages.rs

@@ -0,0 +1,14 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct MyRequest {
+    pub key: [u8; 32],
+    pub body: Vec<u8>,
+    pub sig: Vec<u8>, // FIXME :: fixed length array not supported by serde [u8; 64],
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct MyResponse {
+    // sig: [u8; 64],
+    pub sig: Vec<u8>, // FIXME :: fixed length array not supported by serde [u8; 64],
+}

src/ping.rs

@@ -0,0 +1,301 @@
+// Copyright 2020 Parity Technologies (UK) Ltd.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the "Software"),
+// to deal in the Software without restriction, including without limitation
+// the rights to use, copy, modify, merge, publish, distribute, sublicense,
+// and/or sell copies of the Software, and to permit persons to whom the
+// Software is furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+//! Integration tests for the `Behaviour`.
+
+use std::{io, iter};
+
+use futures::prelude::*;
+use libp2p::identity::PeerId;
+use libp2p::request_response;
+use libp2p::request_response::ProtocolSupport;
+use libp2p::swarm::{StreamProtocol, Swarm, SwarmEvent};
+// use libp2p::swarm_test::SwarmExt;
+use rand::Rng;
+use serde::{Deserialize, Serialize};
+use tracing_subscriber::EnvFilter;
+
+async fn is_response_outbound() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env())
+        .try_init();
+    let ping = Ping("ping".to_string().into_bytes());
+    let offline_peer = PeerId::random();
+
+    let mut swarm1 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(
+            [(
+                StreamProtocol::new("/ping/1"),
+                request_response::ProtocolSupport::Full,
+            )],
+            request_response::Config::default(),
+        )
+    });
+
+    let request_id1 = swarm1
+        .behaviour_mut()
+        .send_request(&offline_peer, ping.clone());
+
+    match swarm1
+        .next_swarm_event()
+        .await
+        .try_into_behaviour_event()
+        .unwrap()
+    {
+        request_response::Event::OutboundFailure {
+            peer,
+            request_id: req_id,
+            error: _error,
+            ..
+        } => {
+            assert_eq!(&offline_peer, &peer);
+            assert_eq!(req_id, request_id1);
+        }
+        e => panic!("Peer: Unexpected event: {e:?}"),
+    }
+
+    let request_id2 = swarm1.behaviour_mut().send_request(&offline_peer, ping);
+
+    assert!(!swarm1
+        .behaviour()
+        .is_pending_outbound(&offline_peer, &request_id1));
+    assert!(swarm1
+        .behaviour()
+        .is_pending_outbound(&offline_peer, &request_id2));
+}
+
+async fn ping_protocol() {
+    let ping = Ping("ping".to_string().into_bytes());
+    let pong = Pong("pong".to_string().into_bytes());
+
+    let protocols = iter::once((StreamProtocol::new("/ping/1"), ProtocolSupport::Full));
+    let cfg = request_response::Config::default();
+
+    let mut swarm1 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(protocols.clone(), cfg.clone())
+    });
+    let peer1_id = *swarm1.local_peer_id();
+    let mut swarm2 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(protocols, cfg)
+    });
+    let peer2_id = *swarm2.local_peer_id();
+
+    swarm1.listen().with_memory_addr_external().await;
+    swarm2.connect(&mut swarm1).await;
+
+    let expected_ping = ping.clone();
+    let expected_pong = pong.clone();
+
+    let peer1 = async move {
+        loop {
+            match swarm1.next_swarm_event().await.try_into_behaviour_event() {
+                Ok(request_response::Event::Message {
+                    peer,
+                    message:
+                        request_response::Message::Request {
+                            request, channel, ..
+                        },
+                    ..
+                }) => {
+                    assert_eq!(&request, &expected_ping);
+                    assert_eq!(&peer, &peer2_id);
+                    swarm1
+                        .behaviour_mut()
+                        .send_response(channel, pong.clone())
+                        .unwrap();
+                }
+                Ok(request_response::Event::ResponseSent { peer, .. }) => {
+                    assert_eq!(&peer, &peer2_id);
+                }
+                Ok(e) => {
+                    panic!("Peer1: Unexpected event: {e:?}")
+                }
+                Err(..) => {}
+            }
+        }
+    };
+
+    let num_pings: u8 = rand::thread_rng().gen_range(1..100);
+
+    let peer2 = async {
+        let mut count = 0;
+
+        let mut req_id = swarm2.behaviour_mut().send_request(&peer1_id, ping.clone());
+        assert!(swarm2.behaviour().is_pending_outbound(&peer1_id, &req_id));
+
+        loop {
+            match swarm2
+                .next_swarm_event()
+                .await
+                .try_into_behaviour_event()
+                .unwrap()
+            {
+                request_response::Event::Message {
+                    peer,
+                    message:
+                        request_response::Message::Response {
+                            request_id,
+                            response,
+                        },
+                    ..
+                } => {
+                    count += 1;
+                    assert_eq!(&response, &expected_pong);
+                    assert_eq!(&peer, &peer1_id);
+                    assert_eq!(req_id, request_id);
+                    if count >= num_pings {
+                        return;
+                    } else {
+                        req_id = swarm2.behaviour_mut().send_request(&peer1_id, ping.clone());
+                    }
+                }
+                e => panic!("Peer2: Unexpected event: {e:?}"),
+            }
+        }
+    };
+
+    async_std::task::spawn(Box::pin(peer1));
+    peer2.await;
+}
+
+async fn emits_inbound_connection_closed_failure() {
+    let ping = Ping("ping".to_string().into_bytes());
+
+    let protocols = iter::once((StreamProtocol::new("/ping/1"), ProtocolSupport::Full));
+    let cfg = request_response::Config::default();
+
+    let mut swarm1 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(protocols.clone(), cfg.clone())
+    });
+    let peer1_id = *swarm1.local_peer_id();
+    let mut swarm2 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(protocols, cfg)
+    });
+    let peer2_id = *swarm2.local_peer_id();
+
+    swarm1.listen().with_memory_addr_external().await;
+    swarm2.connect(&mut swarm1).await;
+
+    swarm2.behaviour_mut().send_request(&peer1_id, ping.clone());
+
+    // Wait for swarm 1 to receive request by swarm 2.
+    let _channel = loop {
+        futures::select!(
+            event = swarm1.select_next_some() => match event {
+                SwarmEvent::Behaviour(request_response::Event::Message {
+                    peer,
+                    message: request_response::Message::Request { request, channel, .. },
+                    ..
+                }) => {
+                    assert_eq!(&request, &ping);
+                    assert_eq!(&peer, &peer2_id);
+                    break channel;
+                },
+                SwarmEvent::Behaviour(ev) => panic!("Peer1: Unexpected event: {ev:?}"),
+                _ => {}
+            },
+            event = swarm2.select_next_some() => {
+                if let SwarmEvent::Behaviour(ev) = event {
+                    panic!("Peer2: Unexpected event: {ev:?}");
+                }
+            }
+        )
+    };
+
+    // Drop swarm 2 in order for the connection between swarm 1 and 2 to close.
+    drop(swarm2);
+
+    loop {
+        match swarm1.select_next_some().await {
+            SwarmEvent::Behaviour(request_response::Event::InboundFailure {
+                error: request_response::InboundFailure::ConnectionClosed,
+                ..
+            }) => break,
+            SwarmEvent::Behaviour(e) => panic!("Peer1: Unexpected event: {e:?}"),
+            _ => {}
+        }
+    }
+}
+
+/// We expect the substream to be properly closed when response channel is dropped.
+/// Since the ping protocol used here expects a response, the sender considers this
+/// early close as a protocol violation which results in the connection being closed.
+/// If the substream were not properly closed when dropped, the sender would instead
+/// run into a timeout waiting for the response.
+async fn emits_inbound_connection_closed_if_channel_is_dropped() {
+    let ping = Ping("ping".to_string().into_bytes());
+
+    let protocols = iter::once((StreamProtocol::new("/ping/1"), ProtocolSupport::Full));
+    let cfg = request_response::Config::default();
+
+    let mut swarm1 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(protocols.clone(), cfg.clone())
+    });
+    let peer1_id = *swarm1.local_peer_id();
+    let mut swarm2 = Swarm::new_ephemeral(|_| {
+        request_response::cbor::Behaviour::<Ping, Pong>::new(protocols, cfg)
+    });
+    let peer2_id = *swarm2.local_peer_id();
+
+    swarm1.listen().with_memory_addr_external().await;
+    swarm2.connect(&mut swarm1).await;
+
+    swarm2.behaviour_mut().send_request(&peer1_id, ping.clone());
+
+    // Wait for swarm 1 to receive request by swarm 2.
+    let event = loop {
+        futures::select!(
+            event = swarm1.select_next_some() => {
+                if let SwarmEvent::Behaviour(request_response::Event::Message {
+                    peer,
+                    message: request_response::Message::Request { request, channel, .. },
+                    ..
+                }) = event {
+                    assert_eq!(&request, &ping);
+                    assert_eq!(&peer, &peer2_id);
+
+                    drop(channel);
+                    continue;
+                }
+            },
+            event = swarm2.select_next_some() => {
+                if let SwarmEvent::Behaviour(ev) = event {
+                    break ev;
+                }
+            },
+        )
+    };
+
+    let error = match event {
+        request_response::Event::OutboundFailure { error, .. } => error,
+        e => panic!("unexpected event from peer 2: {e:?}"),
+    };
+
+    assert!(matches!(
+        error,
+        request_response::OutboundFailure::Io(e) if e.kind() == io::ErrorKind::UnexpectedEof,
+    ));
+}
+
+// Simple Ping-Pong Protocol
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+struct Ping(Vec<u8>);
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+struct Pong(Vec<u8>);

src/protocol.rs

@@ -0,0 +1,42 @@
+use std::{error::Error, time::Duration};
+
+use libp2p::{
+    mdns, noise,
+    request_response::{self, ProtocolSupport},
+    swarm::NetworkBehaviour,
+    tcp, yamux, StreamProtocol, Swarm,
+};
+use libp2p_identity::ed25519::Keypair;
+
+use crate::messages::{MyRequest, MyResponse};
+
+#[derive(NetworkBehaviour)]
+pub struct MyBehaviour {
+    pub mdns: mdns::tokio::Behaviour,
+    pub req_res: request_response::cbor::Behaviour<MyRequest, MyResponse>,
+}
+
+pub fn mk_swarm(
+    kp: &Keypair,
+    protocol_support: ProtocolSupport,
+) -> Result<Swarm<MyBehaviour>, Box<dyn Error>> {
+    let swarm = libp2p::SwarmBuilder::with_existing_identity(kp.clone().into())
+        .with_tokio()
+        .with_tcp(
+            tcp::Config::default(),
+            noise::Config::new,
+            yamux::Config::default,
+        )?
+        .with_behaviour(|key| {
+            let mdns =
+                mdns::tokio::Behaviour::new(mdns::Config::default(), key.public().to_peer_id())?;
+            let protocol = [(StreamProtocol::new("/sign-me/1"), protocol_support)];
+            let config = request_response::Config::default();
+            let req_res =
+                request_response::cbor::Behaviour::<MyRequest, MyResponse>::new(protocol, config);
+            Ok(MyBehaviour { req_res, mdns })
+        })?
+        .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(u64::MAX)))
+        .build();
+    Ok(swarm)
+}

src/server.rs

@@ -1,206 +0,0 @@
-use std::{error::Error, time::Duration};
-
-use futures::prelude::*;
-use libp2p::{
-    mdns, noise, ping,
-    request_response::{self, Codec},
-    swarm::SwarmEvent,
-    tcp, yamux, Multiaddr,
-};
-use tracing_subscriber::EnvFilter;
-
-// We create a custom network behaviour
-
-#[derive(NetworkBehaviour)]
-struct MyBehaviour {
-    req_res: request_response::Behaviour,
-    mdns: mdns::tokio::Behaviour,
-}
-
-struct MyRequest {
-    key: [u8; 32],
-    body: Vec<u8>,
-    sig: [u8; 64],
-}
-
-struct MyResponse {
-    sig: [u8; 64],
-}
-
-struct MyCodec;
-
-impl Codec for MyCodec {
-    #[doc = " The type of protocol(s) or protocol versions being negotiated."]
-    type Protocol = String;
-
-    #[doc = " The type of inbound and outbound requests."]
-    type Request = MyRequest;
-
-    #[doc = " The type of inbound and outbound responses."]
-    type Response = MyResponse;
-
-    #[doc = " Reads a request from the given I/O stream according to the"]
-    #[doc = " negotiated protocol."]
-    #[must_use]
-    #[allow(
-        elided_named_lifetimes,
-        clippy::type_complexity,
-        clippy::type_repetition_in_bounds
-    )]
-    fn read_request<'life0, 'life1, 'life2, 'async_trait, T>(
-        &'life0 mut self,
-        _protocol: &'life1 Self::Protocol,
-        io: &'life2 mut T,
-    ) -> ::core::pin::Pin<
-        Box<
-            dyn ::core::future::Future<Output = io::Result<Self::Request>>
-                + ::core::marker::Send
-                + 'async_trait,
-        >,
-    >
-    where
-        T: AsyncRead + Unpin + Send,
-        T: 'async_trait,
-        'life0: 'async_trait,
-        'life1: 'async_trait,
-        'life2: 'async_trait,
-        Self: 'async_trait,
-    {
-        todo!()
-    }
-
-    #[doc = " Reads a response from the given I/O stream according to the"]
-    #[doc = " negotiated protocol."]
-    #[must_use]
-    #[allow(
-        elided_named_lifetimes,
-        clippy::type_complexity,
-        clippy::type_repetition_in_bounds
-    )]
-    fn read_response<'life0, 'life1, 'life2, 'async_trait, T>(
-        &'life0 mut self,
-        protocol: &'life1 Self::Protocol,
-        io: &'life2 mut T,
-    ) -> ::core::pin::Pin<
-        Box<
-            dyn ::core::future::Future<Output = io::Result<Self::Response>>
-                + ::core::marker::Send
-                + 'async_trait,
-        >,
-    >
-    where
-        T: AsyncRead + Unpin + Send,
-        T: 'async_trait,
-        'life0: 'async_trait,
-        'life1: 'async_trait,
-        'life2: 'async_trait,
-        Self: 'async_trait,
-    {
-        todo!()
-    }
-
-    #[doc = " Writes a request to the given I/O stream according to the"]
-    #[doc = " negotiated protocol."]
-    #[must_use]
-    #[allow(
-        elided_named_lifetimes,
-        clippy::type_complexity,
-        clippy::type_repetition_in_bounds
-    )]
-    fn write_request<'life0, 'life1, 'life2, 'async_trait, T>(
-        &'life0 mut self,
-        protocol: &'life1 Self::Protocol,
-        io: &'life2 mut T,
-        req: Self::Request,
-    ) -> ::core::pin::Pin<
-        Box<
-            dyn ::core::future::Future<Output = io::Result<()>>
-                + ::core::marker::Send
-                + 'async_trait,
-        >,
-    >
-    where
-        T: AsyncWrite + Unpin + Send,
-        T: 'async_trait,
-        'life0: 'async_trait,
-        'life1: 'async_trait,
-        'life2: 'async_trait,
-        Self: 'async_trait,
-    {
-        todo!()
-    }
-
-    #[doc = " Writes a response to the given I/O stream according to the"]
-    #[doc = " negotiated protocol."]
-    #[must_use]
-    #[allow(
-        elided_named_lifetimes,
-        clippy::type_complexity,
-        clippy::type_repetition_in_bounds
-    )]
-    fn write_response<'life0, 'life1, 'life2, 'async_trait, T>(
-        &'life0 mut self,
-        protocol: &'life1 Self::Protocol,
-        io: &'life2 mut T,
-        res: Self::Response,
-    ) -> ::core::pin::Pin<
-        Box<
-            dyn ::core::future::Future<Output = io::Result<()>>
-                + ::core::marker::Send
-                + 'async_trait,
-        >,
-    >
-    where
-        T: AsyncWrite + Unpin + Send,
-        T: 'async_trait,
-        'life0: 'async_trait,
-        'life1: 'async_trait,
-        'life2: 'async_trait,
-        Self: 'async_trait,
-    {
-        todo!()
-    }
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    let _ = tracing_subscriber::fmt()
-        .with_env_filter(EnvFilter::from_default_env())
-        .try_init();
-
-    let mut swarm = libp2p::SwarmBuilder::with_new_identity()
-        .with_tokio()
-        .with_tcp(
-            tcp::Config::default(),
-            noise::Config::new,
-            yamux::Config::default,
-        )?
-        .with_behaviour(|key| {
-            let mdns =
-                mdns::tokio::Behaviour::new(mdns::Config::default(), key.public().to_peer_id())?;
-            let req_res = request_response::Behaviour::with_codec(MyCodec, protocol, config);
-            Ok(MyBehaviour { req_res, mdns })
-        })?
-        .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(u64::MAX)))
-        .build();
-
-    // Tell the swarm to listen on all interfaces and a random, OS-assigned
-    // port.
-    swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
-
-    // Dial the peer identified by the multi-address given as the second
-    // command-line argument, if any.
-    if let Some(addr) = std::env::args().nth(1) {
-        let remote: Multiaddr = addr.parse()?;
-        swarm.dial(remote)?;
-        println!("Dialed {addr}")
-    }
-
-    loop {
-        match swarm.select_next_some().await {
-            SwarmEvent::NewListenAddr { address, .. } => println!("Listening on {address:?}"),
-            SwarmEvent::Behaviour(event) => println!("{event:?}"),
-            _ => {}
-        }
-    }
-}