From 8cbc3c240359798f5617774fc7a1f798086ff1ff Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Wed, 8 Feb 2023 13:21:32 +0100 Subject: [PATCH] node: Move to NoiseXK Move from NoiseNN with a custom authentication protocol, to NoiseXK. Uses ECDH on the ed25519 curve via the `ec25519` library. --- Cargo.lock | 72 +++++----- radicle-cob/Cargo.toml | 1 - radicle-crypto/Cargo.toml | 4 +- radicle-crypto/src/lib.rs | 98 ++++--------- radicle-crypto/src/ssh/keystore.rs | 18 +-- radicle-crypto/src/test/signer.rs | 16 ++- radicle-node/Cargo.toml | 4 +- radicle-node/src/runtime.rs | 26 ++-- radicle-node/src/runtime/handle.rs | 12 +- radicle-node/src/service.rs | 4 +- radicle-node/src/service/message.rs | 6 +- radicle-node/src/test/environment.rs | 12 +- radicle-node/src/test/peer.rs | 4 +- radicle-node/src/wire/message.rs | 10 +- radicle-node/src/wire/protocol.rs | 208 ++++++++++++++++++--------- radicle-node/src/worker.rs | 12 +- radicle/Cargo.toml | 3 +- 17 files changed, 269 insertions(+), 241 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c1fa2f94..75088fcc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -31,9 +31,9 @@ dependencies = [ [[package]] name = "amplify" -version = "4.0.0-beta.7" +version = "4.0.0-beta.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1cc8da6203c5311cc0d74155341a5577564b8360976988eeb5ce7ab6903757f1" +checksum = "72eaf6c304959bf74ba94973ebea642729f583000ef77e7139844642cc066004" dependencies = [ "amplify_derive", "amplify_num", @@ -43,9 +43,9 @@ dependencies = [ [[package]] name = "amplify_derive" -version = "4.0.0-alpha.2" +version = "4.0.0-alpha.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2a258501c851f9dec1549e046d551b645b8b970e30aca85dbec2d6e36cda8e91" +checksum = "01c4835e964725149d7961ec5af2ca1302f6f68c8c738b4acb06185f596c3333" dependencies = [ "amplify_syn", "proc-macro2", @@ -55,15 +55,15 @@ dependencies = [ [[package]] name = "amplify_num" -version = "0.4.1" +version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f27d3d00d3d115395a7a8a4dc045feb7aa82b641e485f7e15f4e67ac16f4f56d" +checksum = "ddce3bc63e807ea02065e8d8b702695f3d302ae4158baddff8b0ce5c73947251" [[package]] name = "amplify_syn" -version = "1.1.6" +version = "2.0.0-beta.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da24db1445cc7bc3842fa072c2d51fe5b25b812b6a572d65842a4c72e87221ac" +checksum = "2e3db5cb96f8f18a9566a975c521b77485f6f8e3e8d866d0f52369446ba31274" dependencies = [ "proc-macro2", "quote", @@ -571,9 +571,9 @@ dependencies = [ [[package]] name = "cypheraddr" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe21bd66fbbd74b706a551b0f56d8ee2488300f5a7cff41febce5973fa05cbc7" +checksum = "41a05c461a9b86ba80542a5204924fd3cae3f47be011e00b1bbef9d71d95b3bb" dependencies = [ "amplify", "base32", @@ -583,21 +583,21 @@ dependencies = [ [[package]] name = "cyphergraphy" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89b94c89b484ddb937306c3ae8122b866749d1d47768b5c4a1e3e8e72ea8270" +checksum = "3c8e45921460ef188da680742fd641f9b2a85d0de6bce12ce26e64c0f4913b41" dependencies = [ "amplify", - "ed25519-compact", + "ec25519", "multibase", "sha2 0.10.6", ] [[package]] name = "cyphernet" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c479a27a1c33fe0a146b6e7048fd373f48ed11205bb6c956d05d090fe749d2fe" +checksum = "5c4cd4c5d6937b81f3df6bb26fe94b4d1c52dd2cfd85507d063d9892cd64448d" dependencies = [ "cypheraddr", "cyphergraphy", @@ -682,6 +682,17 @@ dependencies = [ "subtle", ] +[[package]] +name = "ec25519" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdfd533a2fc01178c738c99412ae1f7e1ad2cb37c2e14bfd87e9d4618171c825" +dependencies = [ + "ct-codecs", + "ed25519", + "getrandom 0.2.8", +] + [[package]] name = "ecdsa" version = "0.14.8" @@ -703,17 +714,6 @@ dependencies = [ "signature", ] -[[package]] -name = "ed25519-compact" -version = "2.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a3d382e8464107391c8706b4c14b087808ecb909f6c15c34114bc42e53a9e4c" -dependencies = [ - "ct-codecs", - "ed25519", - "getrandom 0.2.8", -] - [[package]] name = "ed25519-dalek" version = "1.0.1" @@ -730,9 +730,9 @@ dependencies = [ [[package]] name = "eidolon-auth" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc7d8d05034e0462d2df39fbbd8dbf24b0c0c4337541850ec17ace0b6b42d787" +checksum = "4a764411b1ee8bcacb5203d24e9f0b2d192be62b23ea1c16d0e3462e103f3ffc" dependencies = [ "amplify", "cyphergraphy", @@ -1435,13 +1435,12 @@ dependencies = [ [[package]] name = "netservices" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c55aa6eb316df39bd0d90c845bd02bc0c10629788aae523a8759c6d3632d568" +checksum = "dc700700b1a95cd1eafc4c25c48ac42f72a54ab8c5fa809eef630bebbb390312" dependencies = [ "amplify", "cyphernet", - "ed25519-compact", "io-reactor", "libc", "rand 0.8.5", @@ -1450,9 +1449,9 @@ dependencies = [ [[package]] name = "noise-framework" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ce256b1337e403d6576c5af917a2152441e94d430bb3202e13b7f952fa53e49" +checksum = "daedd5a3c07ef713cb9c4b6f263837822d70412517576dcacd050fa0d1ba99bd" dependencies = [ "amplify", "chacha20poly1305", @@ -1883,7 +1882,6 @@ dependencies = [ name = "radicle-cob" version = "0.1.0" dependencies = [ - "ed25519-compact", "fastrand", "git-commit", "git-ref-format", @@ -1923,7 +1921,7 @@ dependencies = [ "amplify", "base64", "cyphernet", - "ed25519-compact", + "ec25519", "fastrand", "git-ref-format", "multibase", @@ -2491,9 +2489,9 @@ dependencies = [ [[package]] name = "socks5-client" -version = "0.1.0" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d80f67376257f749cbf2835799705650d76833389e80dabd2986082fb2f42072" +checksum = "4091196d57cf9436ebecbec4c572b2be61373a7aaa632a3e93a5cb8555ec1b79" dependencies = [ "amplify", "cypheraddr", diff --git a/radicle-cob/Cargo.toml b/radicle-cob/Cargo.toml index 44259c61..f59cc914 100644 --- a/radicle-cob/Cargo.toml +++ b/radicle-cob/Cargo.toml @@ -42,7 +42,6 @@ version = "1.0" features = ["derive"] [dev-dependencies] -ed25519-compact = { version = "2.0.2", features = ["pem"] } fastrand = { version = "1.8.0", default-features = false } git-ref-format = { version = "0.2", features = ["macro"] } tempfile = { version = "3" } diff --git a/radicle-crypto/Cargo.toml b/radicle-crypto/Cargo.toml index bf4e2205..af2c102a 100644 --- a/radicle-crypto/Cargo.toml +++ b/radicle-crypto/Cargo.toml @@ -14,9 +14,9 @@ ssh = ["base64", "radicle-ssh", "ssh-key"] [dependencies] amplify = { version = "4.0.0-beta.4" } -ed25519-compact = { version = "2.0.2", features = ["pem"] } -cyphernet = { version = "0.1.0", optional = true } +cyphernet = { version = "0.2.0", optional = true } multibase = { version = "0.9.1" } +ec25519 = { version = "0.1.0", features = [] } serde = { version = "1", features = ["derive"] } sha2 = { version = "0.10.2" } sqlite = { version = "0.30.3", optional = true } diff --git a/radicle-crypto/src/lib.rs b/radicle-crypto/src/lib.rs index fdb1199b..354ea87a 100644 --- a/radicle-crypto/src/lib.rs +++ b/radicle-crypto/src/lib.rs @@ -2,13 +2,11 @@ use std::cmp::Ordering; use std::sync::Arc; use std::{fmt, ops::Deref, str::FromStr}; -#[cfg(feature = "cyphernet")] -use cyphernet::{EcSigInvalid, EcSkInvalid, EcVerifyError}; -use ed25519_compact as ed25519; +use ec25519 as ed25519; use serde::{Deserialize, Serialize}; use thiserror::Error; -pub use ed25519::{Error, KeyPair, Seed}; +pub use ed25519::{edwards25519, Error, KeyPair, Seed}; pub mod hash; #[cfg(feature = "ssh")] @@ -159,12 +157,6 @@ impl TryFrom for Signature { #[serde(into = "String", try_from = "String")] pub struct PublicKey(pub ed25519::PublicKey); -impl PublicKey { - pub fn from_pem(pem: &str) -> Result { - ed25519::PublicKey::from_pem(pem).map(Self) - } -} - #[cfg(feature = "cyphernet")] impl cyphernet::display::MultiDisplay for PublicKey { type Display = String; @@ -174,45 +166,10 @@ impl cyphernet::display::MultiDisplay for PublicKe } } -#[cfg(feature = "cyphernet")] -impl cyphernet::display::MultiDisplay for Signature { - type Display = String; - - fn display_fmt(&self, _: &cyphernet::display::Encoding) -> Self::Display { - self.to_string() - } -} - -#[cfg(feature = "cyphernet")] -impl cyphernet::EcSk for SecretKey { - type Pk = PublicKey; - - fn generate_keypair() -> (Self, Self::Pk) - where - Self: Sized, - { - let pair = KeyPair::generate(); - (pair.sk.into(), pair.pk.into()) - } - - fn to_pk(&self) -> Result { - Ok(self.public_key().into()) - } -} - -#[cfg(feature = "cyphernet")] -impl cyphernet::EcSign for SecretKey { - type Sig = Signature; - - fn sign(&self, msg: impl AsRef<[u8]>) -> Self::Sig { - self.0.sign(msg, None).into() - } -} - #[cfg(feature = "cyphernet")] impl cyphernet::EcPk for PublicKey { const COMPRESSED_LEN: usize = 32; - const CURVE_NAME: &'static str = "Ed25519"; + const CURVE_NAME: &'static str = "Curve25519"; type Compressed = [u8; 32]; @@ -235,35 +192,20 @@ impl cyphernet::EcPk for PublicKey { } } -#[cfg(feature = "cyphernet")] -impl cyphernet::EcSig for Signature { - const COMPRESSED_LEN: usize = 64; - type Pk = PublicKey; - type Compressed = [u8; 64]; - - fn to_sig_compressed(&self) -> Self::Compressed { - *self.0.deref() - } - - fn from_sig_compressed(sig: Self::Compressed) -> Result { - Ok(Signature::from(sig)) - } - - fn from_sig_compressed_slice(slice: &[u8]) -> Result { - ed25519::Signature::from_slice(slice) - .map_err(|_| EcSigInvalid::default()) - .map(Signature) - } - - fn verify(&self, pk: &Self::Pk, msg: impl AsRef<[u8]>) -> Result<(), EcVerifyError> { - self.0.verify(pk, msg) - } -} - /// The private/signing key. #[derive(Clone, Debug, Eq, PartialEq, Hash)] pub struct SecretKey(ed25519::SecretKey); +impl SecretKey { + /// Elliptic-curve Diffie-Hellman. + pub fn ecdh(&self, pk: &PublicKey) -> Result<[u8; 32], ed25519::Error> { + let scalar = self.seed().scalar(); + let ge = edwards25519::GeP3::from_bytes_vartime(pk).ok_or(Error::InvalidPublicKey)?; + + Ok(edwards25519::ge_scalarmult(&scalar, &ge).to_bytes()) + } +} + impl PartialOrd for SecretKey { fn partial_cmp(&self, other: &Self) -> Option { Some(self.cmp(other)) @@ -520,10 +462,22 @@ pub mod keypair { #[cfg(test)] mod tests { - use crate::PublicKey; + use super::KeyPair; + use crate::{PublicKey, SecretKey}; use qcheck_macros::quickcheck; use std::str::FromStr; + #[test] + fn test_e25519_dh() { + let kp_a = KeyPair::generate(); + let kp_b = KeyPair::generate(); + + let output_a = SecretKey::from(kp_b.sk).ecdh(&kp_a.pk.into()).unwrap(); + let output_b = SecretKey::from(kp_a.sk).ecdh(&kp_b.pk.into()).unwrap(); + + assert_eq!(output_a, output_b); + } + #[quickcheck] fn prop_encode_decode(input: PublicKey) { let encoded = input.to_string(); diff --git a/radicle-crypto/src/ssh/keystore.rs b/radicle-crypto/src/ssh/keystore.rs index b7348f9d..80961e96 100644 --- a/radicle-crypto/src/ssh/keystore.rs +++ b/radicle-crypto/src/ssh/keystore.rs @@ -4,7 +4,7 @@ use std::path::{Path, PathBuf}; use std::{fs, io}; #[cfg(feature = "cyphernet")] -use cyphernet::{EcSign, EcSk, EcSkInvalid}; +use cyphernet::{EcSk, EcSkInvalid, Ecdh}; use thiserror::Error; use zeroize::Zeroizing; @@ -163,21 +163,23 @@ impl EcSk for MemorySigner { where Self: Sized, { - // TODO(cloudhead): Do we need `EcSk` on `MemorySigner`? - todo!() + let ms = Self::gen(); + let pk = ms.public; + + (ms, pk) } fn to_pk(&self) -> Result { - Ok(*self.public_key()) + Ok(self.public) } } #[cfg(feature = "cyphernet")] -impl EcSign for MemorySigner { - type Sig = Signature; +impl Ecdh for MemorySigner { + type SharedSecret = [u8; 32]; - fn sign(&self, msg: impl AsRef<[u8]>) -> Self::Sig { - Signer::sign(self, msg.as_ref()) + fn ecdh(&self, pk: &Self::Pk) -> Result { + self.secret.ecdh(pk).map_err(cyphernet::EcdhError::from) } } diff --git a/radicle-crypto/src/test/signer.rs b/radicle-crypto/src/test/signer.rs index 947e9224..997beb21 100644 --- a/radicle-crypto/src/test/signer.rs +++ b/radicle-crypto/src/test/signer.rs @@ -76,23 +76,27 @@ impl Signer for MockSigner { impl cyphernet::EcSk for MockSigner { type Pk = PublicKey; + // TODO: Should be renamed to 'generate'. fn generate_keypair() -> (Self, Self::Pk) where Self: Sized, { - unimplemented! {} + let kp = Self::default(); + let pk = kp.pk; + + (kp, pk) } fn to_pk(&self) -> Result { - Ok(*self.public_key()) + Ok(self.pk) } } #[cfg(feature = "cyphernet")] -impl cyphernet::EcSign for MockSigner { - type Sig = Signature; +impl cyphernet::Ecdh for MockSigner { + type SharedSecret = [u8; 32]; - fn sign(&self, msg: impl AsRef<[u8]>) -> Self::Sig { - Signer::sign(self, msg.as_ref()) + fn ecdh(&self, pk: &Self::Pk) -> Result { + self.sk.ecdh(pk).map_err(|_| cyphernet::EcdhError::WeakPk) } } diff --git a/radicle-node/Cargo.toml b/radicle-node/Cargo.toml index 3e3618a0..923768bb 100644 --- a/radicle-node/Cargo.toml +++ b/radicle-node/Cargo.toml @@ -16,7 +16,7 @@ byteorder = { version = "1" } chrono = { version = "0.4.0" } colored = { version = "1.9.0" } crossbeam-channel = { version = "0.5.6" } -cyphernet = { version = "0.1.0", features = ["tor", "dns", "ed25519", "p2p-ed25519"] } +cyphernet = { version = "0.2.0", features = ["tor", "dns", "ed25519", "p2p-ed25519"] } fastrand = { version = "1.8.0" } git-ref-format = { version = "0.2", features = ["serde", "macro"] } io-reactor = { version = "0.1.0", features = ["popol"] } @@ -24,7 +24,7 @@ lexopt = { version = "0.2.1" } libc = { version = "0.2.137" } log = { version = "0.4.17", features = ["std"] } localtime = { version = "1.1.0" } -netservices = { version = "0.1.0", features = ["io-reactor", "socket2"] } +netservices = { version = "0.2.0", features = ["io-reactor", "socket2"] } nonempty = { version = "0.8.1", features = ["serialize"] } qcheck = { version = "1", default-features = false, optional = true } sqlite = { version = "0.30.3" } diff --git a/radicle-node/src/runtime.rs b/radicle-node/src/runtime.rs index 8ea179ab..df28674c 100644 --- a/radicle-node/src/runtime.rs +++ b/radicle-node/src/runtime.rs @@ -6,19 +6,20 @@ use std::path::PathBuf; use std::{fs, io, net, thread, time}; use crossbeam_channel as chan; -use cyphernet::{Cert, EcSign}; +use cyphernet::Ecdh; use netservices::resource::NetAccept; -use radicle::git; -use radicle::node::Handle as _; -use radicle::profile::Home; -use radicle::Storage; use reactor::poller::popol; use reactor::Reactor; use thiserror::Error; +use radicle::git; +use radicle::node::Handle as _; +use radicle::profile::Home; +use radicle::Storage; + use crate::address; use crate::control; -use crate::crypto::{Signature, Signer}; +use crate::crypto::Signer; use crate::node::NodeId; use crate::service::{routing, tracking}; use crate::wire; @@ -69,7 +70,7 @@ pub enum Error { } /// Holds join handles to the client threads, as well as a client handle. -pub struct Runtime { +pub struct Runtime { pub id: NodeId, pub home: Home, pub handle: Handle, @@ -81,7 +82,7 @@ pub struct Runtime { pub signals: chan::Receiver<()>, } -impl Runtime { +impl Runtime { /// Initialize the runtime. /// /// This function spawns threads. @@ -95,7 +96,7 @@ impl Runtime { signer: G, ) -> Result, Error> where - G: EcSign + Clone, + G: Ecdh + Clone, { let id = *signer.public_key(); let node_dir = home.node(); @@ -129,13 +130,8 @@ impl Runtime { rng, ); - let cert = Cert { - pk: id, - sig: EcSign::sign(&signer, id.as_slice()), - }; - let (worker_send, worker_recv) = chan::unbounded::>(); - let mut wire = Wire::new(service, worker_send, cert, signer, proxy, clock); + let mut wire = Wire::new(service, worker_send, signer, proxy, clock); let mut local_addrs = Vec::new(); for addr in listen { diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index 36329031..db2e7070 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use crossbeam_channel as chan; -use cyphernet::EcSign; +use cyphernet::Ecdh; use thiserror::Error; use crate::crypto::Signer; @@ -56,7 +56,7 @@ impl From> for Error { } } -pub struct Handle { +pub struct Handle { pub(crate) home: Home, pub(crate) controller: reactor::Controller>, @@ -64,13 +64,13 @@ pub struct Handle { shutdown: Arc, } -impl fmt::Debug for Handle { +impl fmt::Debug for Handle { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Handle").field("home", &self.home).finish() } } -impl Clone for Handle { +impl Clone for Handle { fn clone(&self) -> Self { Self { home: self.home.clone(), @@ -80,7 +80,7 @@ impl Clone for Handle { } } -impl Handle { +impl Handle { pub fn new(home: Home, controller: reactor::Controller>) -> Self { Self { home, @@ -104,7 +104,7 @@ impl Handle { } } -impl radicle::node::Handle for Handle { +impl radicle::node::Handle for Handle { type Sessions = Sessions; type Error = Error; type FetchResult = FetchResult; diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 6e85faaa..ae9738f7 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -953,7 +953,7 @@ where return Err(session::Error::Misbehavior); } - (session::State::Connected { initialized, .. }, Message::Initialize {}) => { + (session::State::Connected { initialized, .. }, Message::Initialize { .. }) => { // Already initialized! if *initialized { debug!( @@ -1537,7 +1537,7 @@ mod gossip { }; let mut msgs = vec![ - Message::init(), + Message::init(*signer.public_key()), Message::inventory(gossip::inventory(now, inventory), signer), Message::subscribe( filter, diff --git a/radicle-node/src/service/message.rs b/radicle-node/src/service/message.rs index 2a7418e6..9d2dbff5 100644 --- a/radicle-node/src/service/message.rs +++ b/radicle-node/src/service/message.rs @@ -304,7 +304,7 @@ impl Announcement { #[derive(Clone, PartialEq, Eq)] pub enum Message { /// The first message sent to a peer after connection. - Initialize {}, + Initialize { node_id: NodeId }, /// Subscribe to gossip messages matching the filter and time range. Subscribe(Subscribe), @@ -333,8 +333,8 @@ pub enum Message { } impl Message { - pub fn init() -> Self { - Self::Initialize {} + pub fn init(node_id: NodeId) -> Self { + Self::Initialize { node_id } } pub fn announcement( diff --git a/radicle-node/src/test/environment.rs b/radicle-node/src/test/environment.rs index 3c2bb74e..81bfbd34 100644 --- a/radicle-node/src/test/environment.rs +++ b/radicle-node/src/test/environment.rs @@ -10,7 +10,7 @@ use crossbeam_channel as chan; use radicle::crypto::ssh::{keystore::MemorySigner, Keystore}; use radicle::crypto::test::signer::MockSigner; -use radicle::crypto::{KeyPair, Seed, Signature, Signer}; +use radicle::crypto::{KeyPair, Seed, Signer}; use radicle::git; use radicle::git::refname; use radicle::identity::Id; @@ -104,7 +104,7 @@ pub struct Node { } /// Handle to a running node. -pub struct NodeHandle { +pub struct NodeHandle { pub id: NodeId, pub storage: Storage, pub signer: G, @@ -114,7 +114,7 @@ pub struct NodeHandle { pub handle: ManuallyDrop>, } -impl Drop for NodeHandle { +impl Drop for NodeHandle { fn drop(&mut self) { log::debug!(target: "test", "Node {} shutting down..", self.id); @@ -128,7 +128,7 @@ impl Drop for NodeHandle { } } -impl NodeHandle { +impl NodeHandle { /// Connect this node to another node, and wait for the connection to be established both ways. pub fn connect(&mut self, remote: &NodeHandle) -> &mut Self { self.handle.connect(remote.id, remote.addr.into()).unwrap(); @@ -245,7 +245,7 @@ impl Node { } } -impl + Signer + Clone> Node { +impl + Signer + Clone> Node { /// Spawn a node in its own thread. pub fn spawn(self, config: service::Config) -> NodeHandle { let listen = vec![([0, 0, 0, 0], 0).into()]; @@ -312,7 +312,7 @@ impl + Signer + Clone> Node( +pub fn converge<'a, G: Signer + cyphernet::Ecdh + 'static>( nodes: impl IntoIterator>, ) -> BTreeSet<(Id, NodeId)> { let nodes = nodes.into_iter().collect::>(); diff --git a/radicle-node/src/test/peer.rs b/radicle-node/src/test/peer.rs index 3e388843..c2e010b1 100644 --- a/radicle-node/src/test/peer.rs +++ b/radicle-node/src/test/peer.rs @@ -243,7 +243,7 @@ where self.initialize(); self.service.connected(remote_id, Link::Inbound); - self.receive(remote_id, Message::init()); + self.receive(remote_id, Message::init(remote_id)); let mut msgs = self.messages(remote_id); msgs.find(|m| matches!(m, Message::Initialize { .. })) @@ -282,7 +282,7 @@ where }) .expect("`inventory-announcement` must be sent"); - self.receive(remote_id, Message::init()); + self.receive(remote_id, Message::init(remote_id)); } pub fn elapse(&mut self, duration: LocalDuration) { diff --git a/radicle-node/src/wire/message.rs b/radicle-node/src/wire/message.rs index 2992fe78..8caf0893 100644 --- a/radicle-node/src/wire/message.rs +++ b/radicle-node/src/wire/message.rs @@ -194,7 +194,9 @@ impl wire::Encode for Message { let mut n = self.type_id().encode(writer)?; match self { - Self::Initialize {} => {} + Self::Initialize { node_id } => { + n += node_id.encode(writer)?; + } Self::Subscribe(Subscribe { filter, since, @@ -243,7 +245,11 @@ impl wire::Decode for Message { let type_id = reader.read_u16::()?; match MessageType::try_from(type_id) { - Ok(MessageType::Initialize) => Ok(Self::Initialize {}), + Ok(MessageType::Initialize) => { + let node_id = NodeId::decode(reader)?; + + Ok(Self::Initialize { node_id }) + } Ok(MessageType::Subscribe) => { let filter = Filter::decode(reader)?; let since = Timestamp::decode(reader)?; diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs index 6c05ffac..a68d31d9 100644 --- a/radicle-node/src/wire/protocol.rs +++ b/radicle-node/src/wire/protocol.rs @@ -11,15 +11,16 @@ use std::{fmt, io, net}; use amplify::Wrapper as _; use crossbeam_channel as chan; -use cyphernet::{Cert, Digest, EcSign, Sha256}; +use cyphernet::addr::{HostName, InetHost, NetAddr}; +use cyphernet::encrypt::noise::{HandshakePattern, Keyset, NoiseState}; +use cyphernet::proxy::socks5; +use cyphernet::{Digest, EcSk, Ecdh, Sha256}; use localtime::LocalTime; use netservices::resource::{ListenerEvent, NetAccept, NetTransport, SessionEvent}; -use netservices::session::ProtocolArtifact; -use netservices::session::{CypherReader, CypherSession, CypherWriter}; -use netservices::{NetConnection, NetSession}; +use netservices::session::{ProtocolArtifact, Socks5Session}; +use netservices::{NetConnection, NetProtocol, NetReader, NetSession, NetWriter}; use radicle::collections::HashMap; -use radicle::crypto::Signature; use radicle::node::NodeId; use radicle::storage::WriteStorage; @@ -32,16 +33,22 @@ use crate::worker::{Task, TaskResult}; use crate::Link; use crate::{address, service}; +/// NoiseXK handshake pattern. +pub const NOISE_XK: HandshakePattern = HandshakePattern { + initiator: cyphernet::encrypt::noise::InitiatorPattern::Xmitted, + responder: cyphernet::encrypt::noise::OneWayPattern::Known, +}; + #[allow(clippy::large_enum_variant)] /// Control message used internally between workers, users, and the service. -pub enum Control { +pub enum Control { /// Message from the user to the service. User(service::Command), /// Message from a worker to the service. Worker(TaskResult), } -impl fmt::Debug for Control { +impl fmt::Debug for Control { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::User(cmd) => cmd.fmt(f), @@ -51,19 +58,21 @@ impl fmt::Debug for Control { } /// Peer session type. -pub type WireSession = CypherSession; +pub type WireSession = NetProtocol, Socks5Session>; /// Peer session type (read-only). -pub type WireReader = CypherReader; +pub type WireReader = NetReader>; /// Peer session type (write-only). -pub type WireWriter = CypherWriter; +pub type WireWriter = NetWriter, Socks5Session>; /// Reactor action. type Action = reactor::Action>, NetTransport>>; /// Peer connection state machine. enum Peer { - /// The initial state before handshake is completed. - Connecting { link: Link, id: Option }, + /// The initial state of an inbound peer before handshake is completed. + Inbound {}, + /// The initial state of an outbound peer before handshake is completed. + Outbound { id: NodeId }, /// The state after handshake is completed. /// Peers in this state are handled by the underlying service. Connected { @@ -92,8 +101,8 @@ enum Peer { impl std::fmt::Debug for Peer { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Connecting { link, id: Some(id) } => write!(f, "Connecting({link:?}, {id})"), - Self::Connecting { link, id: None } => write!(f, "Connecting({link:?})"), + Self::Inbound {} => write!(f, "Inbound"), + Self::Outbound { id } => write!(f, "Outbound({id})"), Self::Connected { link, id, .. } => write!(f, "Connected({link:?}, {id})"), Self::Disconnected { .. } => write!(f, "Disconnected"), Self::Upgrading { @@ -121,18 +130,36 @@ impl Peer { } /// Return a new inbound connecting peer. - fn connecting(link: Link, id: Option) -> Self { - Self::Connecting { link, id } + fn inbound() -> Self { + Self::Inbound {} + } + + /// Return a new inbound connecting peer. + fn outbound(id: NodeId) -> Self { + Self::Outbound { id } } /// Switch to connected state. - fn connected(&mut self, id: NodeId) { - if let Self::Connecting { link, .. } = self { + fn connected(&mut self, id: NodeId) -> Link { + if let Self::Inbound {} = self { + let link = Link::Inbound; + *self = Self::Connected { - link: *link, + link, id, inbox: VecDeque::new(), }; + link + } else if let Self::Outbound { id: expected } = self { + assert_eq!(id, *expected); + let link = Link::Outbound; + + *self = Self::Connected { + link, + id, + inbox: VecDeque::new(), + }; + link } else { panic!("Peer::connected: session for {id} is already established"); } @@ -145,8 +172,13 @@ impl Peer { id: Some(*id), reason, }; - } else if let Self::Connecting { id, .. } = self { - *self = Self::Disconnected { id: *id, reason }; + } else if let Self::Inbound {} = self { + *self = Self::Disconnected { id: None, reason }; + } else if let Self::Outbound { id } = self { + *self = Self::Disconnected { + id: Some(*id), + reason, + }; } else { panic!("Peer::disconnected: session is not connected ({self:?})"); } @@ -162,7 +194,7 @@ impl Peer { inbox: inbox.clone(), }; } else { - panic!("Peer::upgrading: session is not connected"); + panic!("Peer::upgrading: session is not fully connected"); } } @@ -205,13 +237,11 @@ impl Peer { } /// Wire protocol implementation for a set of peers. -pub struct Wire { +pub struct Wire { /// Backing service instance. service: Service, /// Worker pool interface. worker: chan::Sender>, - /// Used for authentication; keeps local identity. - cert: Cert, /// Used for authentication. signer: G, /// Internal queue of actions to send to the reactor. @@ -227,12 +257,11 @@ where R: routing::Store, S: address::Store, W: WriteStorage + 'static, - G: Signer + EcSign, + G: Signer + Ecdh, { pub fn new( mut service: Service, worker: chan::Sender>, - cert: Cert, signer: G, proxy: net::SocketAddr, clock: LocalTime, @@ -244,7 +273,6 @@ where Self { service, worker, - cert, signer, proxy, actions: VecDeque::new(), @@ -284,8 +312,8 @@ where fn active(&self) -> impl Iterator { self.peers.iter().filter_map(|(fd, peer)| match peer { - Peer::Connecting { id: Some(id), .. } => Some((*fd, id)), - Peer::Connecting { id: None, .. } => None, + Peer::Inbound {} => None, + Peer::Outbound { id } => Some((*fd, id)), Peer::Connected { id, .. } => Some((*fd, id)), Peer::Upgrading { id, .. } => Some((*fd, id)), Peer::Upgraded { id, .. } => Some((*fd, id)), @@ -382,7 +410,7 @@ where R: routing::Store + Send, S: address::Store + Send, W: WriteStorage + Send + 'static, - G: Signer + EcSign + Clone + Send, + G: Signer + Ecdh + Clone + Send, { type Listener = NetAccept>; type Transport = NetTransport>; @@ -410,17 +438,9 @@ where "Accepting inbound peer connection from {}..", connection.remote_addr() ); - self.peers.insert( - connection.as_raw_fd(), - Peer::connecting(Link::Inbound, None), - ); + self.peers.insert(connection.as_raw_fd(), Peer::inbound()); - let session = WireSession::accept::<{ Sha256::OUTPUT_LEN }>( - connection, - self.cert, - vec![], - self.signer.clone(), - ); + let session = accept::(connection, self.signer.clone()); let transport = match NetTransport::with_session(session, Link::Inbound) { Ok(transport) => transport, Err(err) => { @@ -445,21 +465,21 @@ where _: Duration, ) { match event { - SessionEvent::Established(ProtocolArtifact { - state: Cert { pk: node_id, .. }, - .. - }) => { - log::debug!(target: "wire", "Session established with {node_id} (fd={fd})"); + SessionEvent::Established(ProtocolArtifact { state, .. }) => { + // SAFETY: With the NoiseXK protocol, there is always a remote static key. + let id: NodeId = state.remote_static_key.unwrap(); + + log::debug!(target: "wire", "Session established with {id} (fd={fd})"); let conflicting = self .active() - .filter(|(other, id)| **id == node_id && *other != fd) + .filter(|(other, d)| **d == id && *other != fd) .map(|(fd, _)| fd) .collect::>(); for fd in conflicting { log::warn!( - target: "wire", "Closing conflicting session with {node_id} (fd={fd})" + target: "wire", "Closing conflicting session with {id} (fd={fd})" ); self.disconnect( fd, @@ -473,19 +493,9 @@ where log::error!(target: "wire", "Session not found for fd {fd}"); return; }; - let Peer::Connecting { link, .. } = peer else { - log::error!( - target: "wire", - "Session for {node_id} was either not found, or in an invalid state" - ); - return; - }; - log::debug!(target: "wire", "Found connecting peer ({:?})..", link); + let link = peer.connected(id); - let link = *link; - - peer.connected(node_id); - self.service.connected(node_id, link); + self.service.connected(id, link); } SessionEvent::Data(data) => { if let Some(Peer::Connected { id, inbox, .. }) = self.peers.get_mut(&fd) { @@ -594,7 +604,7 @@ where log::debug!(target: "wire", "Received transport handover (fd={fd})"); match self.peers.get(&fd) { - Some(Peer::Disconnected { id, reason }) => { + Some(Peer::Disconnected { id, reason, .. }) => { // Disconnect TCP stream. drop(transport); @@ -623,7 +633,7 @@ where R: routing::Store, S: address::Store, W: WriteStorage + 'static, - G: Signer + EcSign, + G: Signer + Ecdh, { type Item = Action; @@ -664,10 +674,9 @@ where break; } - match WireSession::connect_nonblocking::<{ Sha256::OUTPUT_LEN }>( + match dial::( addr.to_inner(), - self.cert, - vec![node_id], + node_id, self.signer.clone(), self.proxy.into(), false, @@ -679,10 +688,8 @@ where self.service.attempted(node_id, &addr); // TODO: Keep track of peer address for when peer disconnects before // handshake is complete. - self.peers.insert( - transport.as_raw_fd(), - Peer::connecting(Link::Outbound, Some(node_id)), - ); + self.peers + .insert(transport.as_raw_fd(), Peer::outbound(node_id)); self.actions .push_back(reactor::Action::RegisterTransport(transport)); @@ -713,3 +720,66 @@ where self.actions.pop_front() } } + +/// Establish a new outgoing connection. +pub fn dial>( + remote_addr: NetAddr, + remote_id: ::Pk, + signer: G, + proxy_addr: NetAddr, + force_proxy: bool, +) -> io::Result> { + let connection = if force_proxy { + net::TcpStream::connect_nonblocking(proxy_addr)? + } else { + net::TcpStream::connect_nonblocking(remote_addr.connection_addr(proxy_addr))? + }; + Ok(session::( + remote_addr, + Some(remote_id), + connection, + signer, + force_proxy, + )) +} + +/// Accept a new connection. +pub fn accept>( + connection: net::TcpStream, + signer: G, +) -> WireSession { + session::( + connection.remote_addr().into(), + None, + connection, + signer, + false, + ) +} + +/// Create a new [`WireSession`]. +fn session>( + remote_addr: NetAddr, + remote_id: Option, + connection: net::TcpStream, + signer: G, + force_proxy: bool, +) -> WireSession { + let socks5 = socks5::Socks5::with(remote_addr, force_proxy); + let proxy = Socks5Session::with(connection, socks5); + let pair = G::generate_keypair(); + let keyset = Keyset { + e: pair.0, + s: Some(signer), + re: None, + rs: remote_id, + }; + + let noise = NoiseState::initialize::<{ Sha256::OUTPUT_LEN }>( + NOISE_XK, + remote_id.is_some(), + &[], + keyset, + ); + WireSession::with(proxy, noise) +} diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs index d0734068..3d477dcc 100644 --- a/radicle-node/src/worker.rs +++ b/radicle-node/src/worker.rs @@ -3,7 +3,7 @@ use std::thread::JoinHandle; use std::{env, io, net, process, thread, time}; use crossbeam_channel as chan; -use cyphernet::EcSign; +use cyphernet::Ecdh; use netservices::tunnel::Tunnel; use netservices::{NetSession, SplitIo}; @@ -58,21 +58,21 @@ impl FetchError { /// Task to be accomplished on a worker thread. /// This is either going to be an outgoing or incoming fetch. -pub struct Task { +pub struct Task { pub fetch: Fetch, pub session: WireSession, pub drain: Vec, } /// Worker response. -pub struct TaskResult { +pub struct TaskResult { pub fetch: Fetch, pub result: Result, FetchError>, pub session: WireSession, } /// A worker that replicates git objects. -struct Worker { +struct Worker { storage: Storage, tasks: chan::Receiver>, daemon: net::SocketAddr, @@ -82,7 +82,7 @@ struct Worker { name: String, } -impl Worker { +impl Worker { /// Waits for tasks and runs them. Blocks indefinitely unless there is an error receiving /// the next task. fn run(mut self) -> Result<(), chan::RecvError> { @@ -303,7 +303,7 @@ pub struct Pool { impl Pool { /// Create a new worker pool with the given parameters. - pub fn with( + pub fn with( tasks: chan::Receiver>, handle: Handle, config: Config, diff --git a/radicle/Cargo.toml b/radicle/Cargo.toml index 7dda9f9d..a73e8b3f 100644 --- a/radicle/Cargo.toml +++ b/radicle/Cargo.toml @@ -13,8 +13,7 @@ sql = ["sqlite"] [dependencies] amplify = { version = "4.0.0-beta.7", default-features = false, features = ["std"] } crossbeam-channel = { version = "0.5.6" } -# FIXME: Remove the extra cyphernet features once we're able to. -cyphernet = { version = "0.1.0", features = ["tor", "dns", "i2p", "nym", "ed25519"] } +cyphernet = { version = "0.2.0", features = ["tor", "dns", "ed25519"] } fastrand = { version = "1.8.0" } git-ref-format = { version = "0", features = ["serde", "macro"] } multibase = { version = "0.9.1" }