mab122
/
radicle-heartwood-lfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

node: Move to NoiseXK 

Move from NoiseNN with a custom authentication protocol, to NoiseXK.
Uses ECDH on the ed25519 curve via the `ec25519` library.
This commit is contained in:
Alexis Sellier 2023-02-08 13:21:32 +01:00
parent 76d23a3fa9
commit 8cbc3c2403
No known key found for this signature in database
17 changed files with 269 additions and 241 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
Cargo.lock generated
View File

@ -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",

1
radicle-cob/Cargo.toml
View File

@ -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" }

4
radicle-crypto/Cargo.toml
View File

@ -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 }

98
radicle-crypto/src/lib.rs
View File

@ -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<String> for Signature {
#[serde(into = "String", try_from = "String")]
pub struct PublicKey(pub ed25519::PublicKey);
impl PublicKey {
pub fn from_pem(pem: &str) -> Result<Self, ed25519::Error> {
ed25519::PublicKey::from_pem(pem).map(Self)
}
}
#[cfg(feature = "cyphernet")]
impl cyphernet::display::MultiDisplay<cyphernet::display::Encoding> for PublicKey {
type Display = String;
@ -174,45 +166,10 @@ impl cyphernet::display::MultiDisplay<cyphernet::display::Encoding> for PublicKe
}
}
#[cfg(feature = "cyphernet")]
impl cyphernet::display::MultiDisplay<cyphernet::display::Encoding> 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<Self::Pk, EcSkInvalid> {
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<Self, EcSigInvalid> {
Ok(Signature::from(sig))
}
fn from_sig_compressed_slice(slice: &[u8]) -> Result<Self, EcSigInvalid> {
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<Ordering> {
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();

18
radicle-crypto/src/ssh/keystore.rs
View File

@ -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<Self::Pk, EcSkInvalid> {
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::SharedSecret, cyphernet::EcdhError> {
self.secret.ecdh(pk).map_err(cyphernet::EcdhError::from)
}
}

16
radicle-crypto/src/test/signer.rs
View File

@ -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<Self::Pk, cyphernet::EcSkInvalid> {
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::SharedSecret, cyphernet::EcdhError> {
self.sk.ecdh(pk).map_err(|_| cyphernet::EcdhError::WeakPk)
}
}

4
radicle-node/Cargo.toml
View File

@ -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" }

26
radicle-node/src/runtime.rs
View File

@ -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<G: Signer + EcSign> {
pub struct Runtime<G: Signer + Ecdh> {
pub id: NodeId,
pub home: Home,
pub handle: Handle<G>,
@ -81,7 +82,7 @@ pub struct Runtime<G: Signer + EcSign> {
pub signals: chan::Receiver<()>,
}
impl<G: Signer + EcSign + 'static> Runtime<G> {
impl<G: Signer + Ecdh + 'static> Runtime<G> {
/// Initialize the runtime.
///
/// This function spawns threads.
@ -95,7 +96,7 @@ impl<G: Signer + EcSign + 'static> Runtime<G> {
signer: G,
) -> Result<Runtime<G>, Error>
where
G: EcSign<Sig = Signature, Pk = NodeId> + Clone,
G: Ecdh<Pk = NodeId> + Clone,
{
let id = *signer.public_key();
let node_dir = home.node();
@ -129,13 +130,8 @@ impl<G: Signer + EcSign + 'static> Runtime<G> {
rng,
);
let cert = Cert {
pk: id,
sig: EcSign::sign(&signer, id.as_slice()),
};
let (worker_send, worker_recv) = chan::unbounded::<worker::Task<G>>();
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 {

12
radicle-node/src/runtime/handle.rs
View File

@ -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<T> From<chan::SendError<T>> for Error {
}
}
pub struct Handle<G: Signer + EcSign> {
pub struct Handle<G: Signer + Ecdh> {
pub(crate) home: Home,
pub(crate) controller: reactor::Controller<wire::Control<G>>,
@ -64,13 +64,13 @@ pub struct Handle<G: Signer + EcSign> {
shutdown: Arc<AtomicBool>,
}
impl<G: Signer + EcSign> fmt::Debug for Handle<G> {
impl<G: Signer + Ecdh> fmt::Debug for Handle<G> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Handle").field("home", &self.home).finish()
}
}
impl<G: Signer + EcSign> Clone for Handle<G> {
impl<G: Signer + Ecdh> Clone for Handle<G> {
fn clone(&self) -> Self {
Self {
home: self.home.clone(),
@ -80,7 +80,7 @@ impl<G: Signer + EcSign> Clone for Handle<G> {
}
}
impl<G: Signer + EcSign + 'static> Handle<G> {
impl<G: Signer + Ecdh + 'static> Handle<G> {
pub fn new(home: Home, controller: reactor::Controller<wire::Control<G>>) -> Self {
Self {
home,
@ -104,7 +104,7 @@ impl<G: Signer + EcSign + 'static> Handle<G> {
}
}
impl<G: Signer + EcSign + 'static> radicle::node::Handle for Handle<G> {
impl<G: Signer + Ecdh + 'static> radicle::node::Handle for Handle<G> {
type Sessions = Sessions;
type Error = Error;
type FetchResult = FetchResult;

4
radicle-node/src/service.rs
View File

@ -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,

6
radicle-node/src/service/message.rs
View File

@ -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(

12
radicle-node/src/test/environment.rs
View File

@ -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<G> {
}
/// Handle to a running node.
pub struct NodeHandle<G: Signer + cyphernet::EcSign + 'static> {
pub struct NodeHandle<G: Signer + cyphernet::Ecdh + 'static> {
pub id: NodeId,
pub storage: Storage,
pub signer: G,
@ -114,7 +114,7 @@ pub struct NodeHandle<G: Signer + cyphernet::EcSign + 'static> {
pub handle: ManuallyDrop<Handle<G>>,
}
impl<G: Signer + cyphernet::EcSign + 'static> Drop for NodeHandle<G> {
impl<G: Signer + cyphernet::Ecdh + 'static> Drop for NodeHandle<G> {
fn drop(&mut self) {
log::debug!(target: "test", "Node {} shutting down..", self.id);
@ -128,7 +128,7 @@ impl<G: Signer + cyphernet::EcSign + 'static> Drop for NodeHandle<G> {
}
}
impl<G: Signer + cyphernet::EcSign> NodeHandle<G> {
impl<G: Signer + cyphernet::Ecdh> NodeHandle<G> {
/// Connect this node to another node, and wait for the connection to be established both ways.
pub fn connect(&mut self, remote: &NodeHandle<G>) -> &mut Self {
self.handle.connect(remote.id, remote.addr.into()).unwrap();
@ -245,7 +245,7 @@ impl Node<MockSigner> {
}
}
impl<G: cyphernet::EcSign<Pk = NodeId, Sig = Signature> + Signer + Clone> Node<G> {
impl<G: cyphernet::Ecdh<Pk = NodeId> + Signer + Clone> Node<G> {
/// Spawn a node in its own thread.
pub fn spawn(self, config: service::Config) -> NodeHandle<G> {
let listen = vec![([0, 0, 0, 0], 0).into()];
@ -312,7 +312,7 @@ impl<G: cyphernet::EcSign<Pk = NodeId, Sig = Signature> + Signer + Clone> Node<G
/// Checks whether the nodes have converged in their routing tables.
#[track_caller]
pub fn converge<'a, G: Signer + cyphernet::EcSign + 'static>(
pub fn converge<'a, G: Signer + cyphernet::Ecdh + 'static>(
nodes: impl IntoIterator<Item = &'a NodeHandle<G>>,
) -> BTreeSet<(Id, NodeId)> {
let nodes = nodes.into_iter().collect::<Vec<_>>();

4
radicle-node/src/test/peer.rs
View File

@ -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) {

10
radicle-node/src/wire/message.rs
View File

@ -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::<NetworkEndian>()?;
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)?;

208
radicle-node/src/wire/protocol.rs
View File

@ -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<G: Signer + EcSign> {
pub enum Control<G: Signer + Ecdh> {
/// Message from the user to the service.
User(service::Command),
/// Message from a worker to the service.
Worker(TaskResult<G>),
}
impl<G: Signer + EcSign> fmt::Debug for Control<G> {
impl<G: Signer + Ecdh> fmt::Debug for Control<G> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::User(cmd) => cmd.fmt(f),
@ -51,19 +58,21 @@ impl<G: Signer + EcSign> fmt::Debug for Control<G> {
}
/// Peer session type.
pub type WireSession<G> = CypherSession<G, Sha256>;
pub type WireSession<G> = NetProtocol<NoiseState<G, Sha256>, Socks5Session<net::TcpStream>>;
/// Peer session type (read-only).
pub type WireReader = CypherReader<Sha256>;
pub type WireReader = NetReader<Socks5Session<net::TcpStream>>;
/// Peer session type (write-only).
pub type WireWriter<G> = CypherWriter<G, Sha256>;
pub type WireWriter<G> = NetWriter<NoiseState<G, Sha256>, Socks5Session<net::TcpStream>>;
/// Reactor action.
type Action<G> = reactor::Action<NetAccept<WireSession<G>>, NetTransport<WireSession<G>>>;
/// Peer connection state machine.
enum Peer {
/// The initial state before handshake is completed.
Connecting { link: Link, id: Option<NodeId> },
/// 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<NodeId>) -> 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<R, S, W, G: Signer + EcSign> {
pub struct Wire<R, S, W, G: Signer + Ecdh> {
/// Backing service instance.
service: Service<R, S, W, G>,
/// Worker pool interface.
worker: chan::Sender<Task<G>>,
/// Used for authentication; keeps local identity.
cert: Cert<Signature>,
/// 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<Pk = NodeId>,
{
pub fn new(
mut service: Service<R, S, W, G>,
worker: chan::Sender<Task<G>>,
cert: Cert<Signature>,
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<Item = (RawFd, &NodeId)> {
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<Pk = NodeId, Sig = Signature> + Clone + Send,
G: Signer + Ecdh<Pk = NodeId> + Clone + Send,
{
type Listener = NetAccept<WireSession<G>>;
type Transport = NetTransport<WireSession<G>>;
@ -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::<G>(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::<Vec<_>>();
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<Pk = NodeId, Sig = Signature>,
G: Signer + Ecdh<Pk = NodeId>,
{
type Item = Action<G>;
@ -664,10 +674,9 @@ where
break;
}
match WireSession::connect_nonblocking::<{ Sha256::OUTPUT_LEN }>(
match dial::<G>(
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<G: Signer + Ecdh<Pk = NodeId>>(
remote_addr: NetAddr<HostName>,
remote_id: <G as EcSk>::Pk,
signer: G,
proxy_addr: NetAddr<InetHost>,
force_proxy: bool,
) -> io::Result<WireSession<G>> {
let connection = if force_proxy {
net::TcpStream::connect_nonblocking(proxy_addr)?
} else {
net::TcpStream::connect_nonblocking(remote_addr.connection_addr(proxy_addr))?
};
Ok(session::<G>(
remote_addr,
Some(remote_id),
connection,
signer,
force_proxy,
))
}
/// Accept a new connection.
pub fn accept<G: Signer + Ecdh<Pk = NodeId>>(
connection: net::TcpStream,
signer: G,
) -> WireSession<G> {
session::<G>(
connection.remote_addr().into(),
None,
connection,
signer,
false,
)
}
/// Create a new [`WireSession`].
fn session<G: Signer + Ecdh<Pk = NodeId>>(
remote_addr: NetAddr<HostName>,
remote_id: Option<NodeId>,
connection: net::TcpStream,
signer: G,
force_proxy: bool,
) -> WireSession<G> {
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)
}

12
radicle-node/src/worker.rs
View File

@ -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<G: Signer + EcSign> {
pub struct Task<G: Signer + Ecdh> {
pub fetch: Fetch,
pub session: WireSession<G>,
pub drain: Vec<u8>,
}
/// Worker response.
pub struct TaskResult<G: Signer + EcSign> {
pub struct TaskResult<G: Signer + Ecdh> {
pub fetch: Fetch,
pub result: Result<Vec<RefUpdate>, FetchError>,
pub session: WireSession<G>,
}
/// A worker that replicates git objects.
struct Worker<G: Signer + EcSign> {
struct Worker<G: Signer + Ecdh> {
storage: Storage,
tasks: chan::Receiver<Task<G>>,
daemon: net::SocketAddr,
@ -82,7 +82,7 @@ struct Worker<G: Signer + EcSign> {
name: String,
}
impl<G: Signer + EcSign + 'static> Worker<G> {
impl<G: Signer + Ecdh + 'static> Worker<G> {
/// 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<G: Signer + EcSign + 'static>(
pub fn with<G: Signer + Ecdh + 'static>(
tasks: chan::Receiver<Task<G>>,
handle: Handle<G>,
config: Config,

3
radicle/Cargo.toml
View File

@ -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" }