diff --git a/node/src/crypto.rs b/node/src/crypto.rs index 9e3e17ba..9e262ac8 100644 --- a/node/src/crypto.rs +++ b/node/src/crypto.rs @@ -4,10 +4,19 @@ use ed25519_consensus as ed25519; use serde::{Deserialize, Serialize}; use thiserror::Error; -#[derive(Serialize, Deserialize, Eq, Debug, Clone)] +pub trait Signer { + /// Return this signer's public/verification key. + fn public_key(&self) -> &PublicKey; +} + +/// The public/verification key. +#[derive(Serialize, Deserialize, Eq, Debug, Copy, Clone)] #[serde(transparent)] pub struct PublicKey(pub ed25519::VerificationKey); +/// The private/signing key. +pub type SecretKey = ed25519::SigningKey; + #[derive(Error, Debug)] pub enum PublicKeyError { #[error("invalid length {0}")] @@ -36,6 +45,12 @@ impl PartialEq for PublicKey { } } +impl From for PublicKey { + fn from(other: ed25519::VerificationKey) -> Self { + Self(other) + } +} + impl PublicKey { pub fn encode(&self) -> String { multibase::encode(multibase::Base::Base58Btc, &self.0) diff --git a/node/src/git.rs b/node/src/git.rs index b10a79c2..9940370c 100644 --- a/node/src/git.rs +++ b/node/src/git.rs @@ -37,7 +37,7 @@ pub fn list_remotes(url: &Url) -> Result, ListRefsError> { for r in refs { let (id, refname) = parse_ref::(r.name())?; let entry = remotes - .entry(id.clone()) + .entry(id) .or_insert_with(|| Remote::new(id, HashMap::default())); entry.refs.insert(refname.to_string(), r.oid().into()); diff --git a/node/src/identity.rs b/node/src/identity.rs index 3aa8541c..c554cab5 100644 --- a/node/src/identity.rs +++ b/node/src/identity.rs @@ -123,8 +123,8 @@ mod test { let mut hm = HashSet::new(); - assert!(hm.insert(a.clone())); - assert!(hm.insert(b.clone())); + assert!(hm.insert(a)); + assert!(hm.insert(b)); assert!(!hm.insert(a)); assert!(!hm.insert(b)); } diff --git a/node/src/protocol.rs b/node/src/protocol.rs index 0845ef6f..9f530e3d 100644 --- a/node/src/protocol.rs +++ b/node/src/protocol.rs @@ -16,15 +16,16 @@ use crate::address_book::AddressBook; use crate::address_manager::AddressManager; use crate::clock::RefClock; use crate::collections::{HashMap, HashSet}; +use crate::crypto; use crate::decoder::Decoder; use crate::identity::{ProjId, UserId}; use crate::storage::{self, WriteRepository}; use crate::storage::{Inventory, ReadStorage, Remotes, Unverified, WriteStorage}; -/// Network peer identifier. -pub type PeerId = IpAddr; +/// Network node identifier. +pub type NodeId = crypto::PublicKey; /// Network routing table. Keeps track of where projects are hosted. -pub type Routing = HashMap>; +pub type Routing = HashMap>; /// Seconds since epoch. pub type Timestamp = u64; @@ -69,6 +70,8 @@ pub struct Envelope { pub enum Message { /// Say hello to a peer. This is the first message sent to a peer after connection. Hello { + // TODO: This is currently untrusted. + id: NodeId, timestamp: Timestamp, version: u32, addrs: Vec
, @@ -87,7 +90,7 @@ pub enum Message { timestamp: Timestamp, /// Original peer this inventory came from. We don't set this when we /// are the originator, only when relaying. - origin: Option, + origin: Option, }, } @@ -101,8 +104,9 @@ impl From for Envelope { } impl Message { - pub fn hello(timestamp: Timestamp, addrs: Vec
, git: Url) -> Self { + pub fn hello(id: NodeId, timestamp: Timestamp, addrs: Vec
, git: Url) -> Self { Self::Hello { + id, timestamp, version: PROTOCOL_VERSION, addrs, @@ -110,7 +114,7 @@ impl Message { } } - pub fn inventory(ctx: &mut Context) -> Result + pub fn inventory(ctx: &mut Context) -> Result where T: storage::ReadStorage, { @@ -219,11 +223,11 @@ impl Config { } #[derive(Debug)] -pub struct Protocol { +pub struct Protocol { /// Peers currently or recently connected. peers: Peers, /// Protocol state that isn't peer-specific. - context: Context, + context: Context, /// Whether our local inventory no long represents what we have announced to the network. out_of_sync: bool, /// Last time the protocol was idle. @@ -238,12 +242,19 @@ pub struct Protocol { start_time: LocalTime, } -impl Protocol { - pub fn new(config: Config, clock: RefClock, storage: T, addresses: S, rng: Rng) -> Self { +impl Protocol { + pub fn new( + config: Config, + clock: RefClock, + storage: T, + addresses: S, + signer: G, + rng: Rng, + ) -> Self { let addrmgr = AddressManager::new(addresses); Self { - context: Context::new(config, clock, storage, addrmgr, rng.clone()), + context: Context::new(config, clock, storage, addrmgr, signer, rng.clone()), peers: Peers::new(rng), out_of_sync: false, last_idle: LocalTime::default(), @@ -254,15 +265,15 @@ impl Protocol { } } - pub fn disconnect(&mut self, peer: &PeerId, reason: DisconnectReason) { - if let Some(addr) = self.peers.get(peer).map(|p| p.addr) { + pub fn disconnect(&mut self, remote: &IpAddr, reason: DisconnectReason) { + if let Some(addr) = self.peers.get(remote).map(|p| p.addr) { self.context.disconnect(addr, reason); } } pub fn providers(&self, proj: &ProjId) -> Box + '_> { if let Some(peers) = self.routing.get(proj) { - Box::new(peers.iter().filter_map(|id| self.peers.get(id))) + Box::new(peers.iter().filter_map(|id| self.peers.by_id(id))) } else { Box::new(std::iter::empty()) } @@ -302,7 +313,7 @@ impl Protocol { /// Returns a sorted list from the closest peer to the furthest. /// Peers with more trackings in common score score higher. #[allow(unused)] - pub fn closest_peers(&self, n: usize) -> Vec { + pub fn closest_peers(&self, n: usize) -> Vec { todo!() } @@ -348,7 +359,7 @@ impl Protocol { .context .routing .get(proj) - .map_or(vec![], |r| r.iter().copied().collect()), + .map_or(vec![], |r| r.iter().cloned().collect()), } } @@ -403,10 +414,11 @@ impl Protocol { } } -impl nakamoto::Protocol for Protocol +impl nakamoto::Protocol for Protocol where T: ReadStorage + WriteStorage + 'static, S: address_book::Store, + G: crypto::Signer, { type Event = (); type Command = Command; @@ -488,11 +500,11 @@ where } fn attempted(&mut self, addr: &std::net::SocketAddr) { - let id = addr.ip(); + let ip = addr.ip(); let persistent = self.context.config.is_persistent(addr); let mut peer = self .peers - .entry(id) + .entry(ip) .or_insert_with(|| Peer::new(*addr, Link::Outbound, persistent)); peer.attempts += 1; @@ -504,9 +516,9 @@ where _local_addr: &std::net::SocketAddr, link: Link, ) { - let id = addr.ip(); + let ip = addr.ip(); - debug!("Connected to {} ({:?})", id, link); + debug!("Connected to {} ({:?})", ip, link); // For outbound connections, we are the first to say "Hello". // For inbound connections, we wait for the remote to say "Hello" first. @@ -514,11 +526,12 @@ where if link.is_outbound() { let git = self.config.git_url.clone(); - if let Some(peer) = self.peers.get_mut(&id) { + if let Some(peer) = self.peers.get_mut(&ip) { self.context.write_all( peer.addr, [ Message::hello( + self.context.id(), self.context.timestamp(), self.context.config.listen.clone(), git, @@ -531,7 +544,7 @@ where } } else { self.peers.insert( - id, + ip, Peer::new( addr, Link::Inbound, @@ -547,11 +560,11 @@ where reason: nakamoto::DisconnectReason, ) { let since = self.local_time(); - let id = addr.ip(); + let ip = addr.ip(); - debug!("Disconnected from {} ({})", id, reason); + debug!("Disconnected from {} ({})", ip, reason); - if let Some(peer) = self.peers.get_mut(&id) { + if let Some(peer) = self.peers.get_mut(&ip) { peer.state = PeerState::Disconnected { since }; // Attempt to re-connect to persistent peers. @@ -566,7 +579,7 @@ where } // TODO: Eventually we want a delay before attempting a reconnection, // with exponential back-off. - debug!("Reconnecting to {} (attempts={})...", id, peer.attempts); + debug!("Reconnecting to {} (attempts={})...", ip, peer.attempts); // TODO: Try to reconnect only if the peer was attempted. A disconnect without // even a successful attempt means that we're unlikely to be able to reconnect. @@ -614,7 +627,7 @@ where Ok(Some(msg)) => { let peers = negotiated .iter() - .filter(|(id, _)| *id != peer.id()) + .filter(|(ip, _)| *ip != peer.ip()) .map(|(_, addr)| *addr) .collect::>(); @@ -629,15 +642,15 @@ where } } -impl Deref for Protocol { - type Target = Context; +impl Deref for Protocol { + type Target = Context; fn deref(&self) -> &Self::Target { &self.context } } -impl DerefMut for Protocol { +impl DerefMut for Protocol { fn deref_mut(&mut self) -> &mut Self::Target { &mut self.context } @@ -673,7 +686,7 @@ impl fmt::Display for DisconnectReason { } } -impl Iterator for Protocol { +impl Iterator for Protocol { type Item = Io<(), DisconnectReason>; fn next(&mut self) -> Option { @@ -687,14 +700,16 @@ pub struct Lookup { /// Whether the project was found locally or not. pub local: Option>, /// A list of remote peers on which the project is known to exist. - pub remote: Vec, + pub remote: Vec, } /// Global protocol state used across peers. #[derive(Debug)] -pub struct Context { +pub struct Context { /// Protocol configuration. config: Config, + /// Our cryptographic signer and key. + signer: G, /// Tracks the location of projects. routing: Routing, /// Outgoing I/O queue. @@ -709,19 +724,22 @@ pub struct Context { rng: Rng, } -impl Context +impl Context where T: storage::ReadStorage + storage::WriteStorage, + G: crypto::Signer, { pub(crate) fn new( config: Config, clock: RefClock, storage: T, addrmgr: AddressManager, + signer: G, rng: Rng, ) -> Self { Self { config, + signer, clock, routing: HashMap::with_hasher(rng.clone().into()), io: VecDeque::new(), @@ -731,8 +749,12 @@ where } } + pub(crate) fn id(&self) -> NodeId { + *self.signer.public_key() + } + /// Process a peer inventory announcement by updating our routing table. - fn process_inventory(&mut self, inventory: &Inventory, from: PeerId, remote: &Url) { + fn process_inventory(&mut self, inventory: &Inventory, from: NodeId, remote: &Url) { for proj_id in inventory { let inventory = self .routing @@ -760,7 +782,7 @@ where } } -impl Context { +impl Context { /// Get current local timestamp. pub(crate) fn timestamp(&self) -> Timestamp { self.clock.local_time().as_secs() @@ -800,13 +822,24 @@ impl Context { } #[derive(Debug)] -pub struct Peers(AddressBook); +/// Holds currently (or recently) connected peers. +pub struct Peers(AddressBook); impl Peers { pub fn new(rng: Rng) -> Self { Self(AddressBook::new(rng)) } + pub fn by_id(&self, id: &NodeId) -> Option<&Peer> { + self.0.values().find(|p| { + if let PeerState::Negotiated { id: _id, .. } = &p.state { + _id == id + } else { + false + } + }) + } + /// Iterator over fully negotiated peers. pub fn negotiated(&self) -> impl Iterator + Clone { self.0.iter().filter(move |(_, p)| p.is_negotiated()) @@ -814,7 +847,7 @@ impl Peers { } impl Deref for Peers { - type Target = AddressBook; + type Target = AddressBook; fn deref(&self) -> &Self::Target { &self.0 @@ -828,6 +861,7 @@ impl DerefMut for Peers { } #[derive(Debug, Default)] +#[allow(clippy::large_enum_variant)] enum PeerState { /// Initial peer state. For outgoing peers this /// means we've attempted a connection. For incoming @@ -837,6 +871,8 @@ enum PeerState { Initial, /// State after successful handshake. Negotiated { + /// The peer's unique identifier. + id: NodeId, since: LocalTime, /// Addresses this peer is reachable on. addrs: Vec
, @@ -892,7 +928,7 @@ impl Peer { } } - fn id(&self) -> PeerId { + fn ip(&self) -> IpAddr { self.addr.ip() } @@ -900,23 +936,25 @@ impl Peer { matches!(self.state, PeerState::Negotiated { .. }) } - fn received( + fn received( &mut self, envelope: Envelope, - ctx: &mut Context, + ctx: &mut Context, ) -> Result, PeerError> where T: storage::ReadStorage + storage::WriteStorage, + G: crypto::Signer, { if envelope.magic != NETWORK_MAGIC { return Err(PeerError::WrongMagic(envelope.magic)); } - debug!("Received {:?} from {}", &envelope.msg, self.id()); + debug!("Received {:?} from {}", &envelope.msg, self.ip()); match (&self.state, envelope.msg) { ( PeerState::Initial, Message::Hello { + id, timestamp, version, addrs, @@ -938,7 +976,7 @@ impl Peer { ctx.write_all( self.addr, [ - Message::hello(now, ctx.config.listen.clone(), git), + Message::hello(ctx.id(), now, ctx.config.listen.clone(), git), Message::get_inventory([]), ], ); @@ -947,6 +985,7 @@ impl Peer { // set after the first message is received only. Setting it here would // mean that messages received right after the handshake could be ignored. self.state = PeerState::Negotiated { + id, since: ctx.clock.local_time(), addrs, git, @@ -955,7 +994,7 @@ impl Peer { (PeerState::Initial, _) => { debug!( "Disconnecting peer {} for sending us a message before handshake", - self.id() + self.ip() ); return Err(PeerError::Misbehavior); } @@ -965,7 +1004,7 @@ impl Peer { ctx.write(self.addr, inventory); } ( - PeerState::Negotiated { git, .. }, + PeerState::Negotiated { id, git, .. }, Message::Inventory { timestamp, inv, @@ -986,13 +1025,13 @@ impl Peer { } else { return Ok(None); } - ctx.process_inventory(&inv, origin.unwrap_or_else(|| self.id()), git); + ctx.process_inventory(&inv, origin.unwrap_or(*id), git); if ctx.config.relay { return Ok(Some(Message::Inventory { timestamp, inv, - origin: origin.or_else(|| Some(self.id())), + origin: origin.or(Some(*id)), })); } } @@ -1007,12 +1046,12 @@ impl Peer { (PeerState::Negotiated { .. }, Message::Hello { .. }) => { debug!( "Disconnecting peer {} for sending us a redundant handshake message", - self.id() + self.ip() ); return Err(PeerError::Misbehavior); } (PeerState::Disconnected { .. }, msg) => { - debug!("Ignoring {:?} from disconnected peer {}", msg, self.id()); + debug!("Ignoring {:?} from disconnected peer {}", msg, self.ip()); } } Ok(None) diff --git a/node/src/storage/git.rs b/node/src/storage/git.rs index 4f17f497..fc7da8b7 100644 --- a/node/src/storage/git.rs +++ b/node/src/storage/git.rs @@ -164,7 +164,7 @@ impl ReadRepository for Repository { let name = r.name().ok_or(Error::InvalidRef)?; let (id, refname) = git::parse_ref::(name)?; let entry = remotes - .entry(id.clone()) + .entry(id) .or_insert_with(|| Remote::new(id, HashMap::default())); let oid = r.target().ok_or(Error::InvalidRef)?; diff --git a/node/src/test.rs b/node/src/test.rs index 300ad876..f0565852 100644 --- a/node/src/test.rs +++ b/node/src/test.rs @@ -1,5 +1,6 @@ pub(crate) mod arbitrary; pub(crate) mod assert; +pub(crate) mod crypto; pub(crate) mod fixtures; pub(crate) mod logger; pub(crate) mod peer; diff --git a/node/src/test/crypto.rs b/node/src/test/crypto.rs new file mode 100644 index 00000000..3577b824 --- /dev/null +++ b/node/src/test/crypto.rs @@ -0,0 +1,38 @@ +use crate::crypto::{PublicKey, SecretKey, Signer}; + +#[derive(Debug)] +pub struct MockSigner { + key: PublicKey, +} + +impl MockSigner { + pub fn new(rng: &mut fastrand::Rng) -> Self { + let mut bytes: [u8; 32] = [0; 32]; + + for byte in &mut bytes { + *byte = rng.u8(..); + } + let sk = SecretKey::from(bytes); + + Self { + key: sk.verification_key().into(), + } + } +} + +impl Default for MockSigner { + fn default() -> Self { + let bytes: [u8; 32] = [0; 32]; + let sk = SecretKey::from(bytes); + + Self { + key: sk.verification_key().into(), + } + } +} + +impl Signer for MockSigner { + fn public_key(&self) -> &PublicKey { + &self.key + } +} diff --git a/node/src/test/peer.rs b/node/src/test/peer.rs index 02bf012b..eb33f5f2 100644 --- a/node/src/test/peer.rs +++ b/node/src/test/peer.rs @@ -12,10 +12,11 @@ use crate::collections::HashMap; use crate::decoder::Decoder; use crate::protocol::*; use crate::storage::{ReadStorage, WriteStorage}; +use crate::test::crypto::MockSigner; use crate::*; /// Protocol instantiation used for testing. -pub type Protocol = crate::protocol::Protocol, S>; +pub type Protocol = crate::protocol::Protocol, S, MockSigner>; #[derive(Debug)] pub struct Peer { @@ -80,7 +81,7 @@ where ip: impl Into, addrs: Vec<(net::SocketAddr, Source)>, storage: S, - rng: fastrand::Rng, + mut rng: fastrand::Rng, ) -> Self { let addrs = addrs .into_iter() @@ -88,7 +89,8 @@ where .collect(); let local_time = LocalTime::now(); let clock = RefClock::from(local_time); - let protocol = Protocol::new(config, clock, storage, addrs, rng.clone()); + let signer = MockSigner::new(&mut rng); + let protocol = Protocol::new(config, clock, storage, addrs, signer, rng.clone()); let ip = ip.into(); let local_addr = net::SocketAddr::new(ip, rng.u16(..)); @@ -120,6 +122,10 @@ where self.config().git_url.clone() } + pub fn id(&self) -> NodeId { + self.protocol.id() + } + pub fn receive(&mut self, peer: &net::SocketAddr, msg: Message) { let bytes = serde_json::to_vec(&Envelope { magic: NETWORK_MAGIC, @@ -130,23 +136,25 @@ where self.protocol.received_bytes(peer, &bytes); } - pub fn connect_from(&mut self, remote: &net::SocketAddr) { + pub fn connect_from(&mut self, peer: &Self) { + let remote = simulator::Peer::>::addr(peer); let local = net::SocketAddr::new(self.ip, self.rng.u16(..)); let git = format!("file://{}.git", remote.ip()); let git = Url::from_bytes(git.as_bytes()).unwrap(); self.initialize(); - self.protocol.connected(*remote, &local, Link::Inbound); + self.protocol.connected(remote, &local, Link::Inbound); self.receive( - remote, + &remote, Message::hello( + peer.id(), self.local_time().as_secs(), - vec![Address::Tcp { addr: *remote }], + vec![Address::Tcp { addr: remote }], git, ), ); - let mut msgs = self.messages(remote); + let mut msgs = self.messages(&remote); msgs.find(|m| matches!(m, Message::Hello { .. })) .expect("`hello` is sent"); msgs.find(|m| matches!(m, Message::GetInventory { .. })) @@ -171,6 +179,7 @@ where self.receive( &remote, Message::hello( + peer.id(), self.local_time().as_secs(), peer.config().listen.clone(), git, diff --git a/node/src/test/tests.rs b/node/src/test/tests.rs index 3fc8e8f8..bf9ac12c 100644 --- a/node/src/test/tests.rs +++ b/node/src/test/tests.rs @@ -51,8 +51,8 @@ fn test_inbound_connection() { let bob = Peer::new("bob", [9, 9, 9, 9], MockStorage::empty()); let eve = Peer::new("eve", [7, 7, 7, 7], MockStorage::empty()); - alice.connect_from(&bob.addr()); - alice.connect_from(&eve.addr()); + alice.connect_from(&bob); + alice.connect_from(&eve); let peers = alice .protocol @@ -108,7 +108,12 @@ fn test_handshake_invalid_timestamp() { alice.connected(bob.addr(), &local, Link::Inbound); alice.receive( &bob.addr(), - Message::hello(alice.timestamp() - time_delta, vec![], bob.git_url()), + Message::hello( + bob.id(), + alice.timestamp() - time_delta, + vec![], + bob.git_url(), + ), ); assert_matches!(alice.outbox().next(), Some(Io::Disconnect(addr, _)) if addr == bob.addr()); } @@ -144,7 +149,7 @@ fn test_inventory_sync() { for proj in &projs { let providers = alice.routing().get(proj).unwrap(); - assert!(providers.contains(&bob.ip)); + assert!(providers.contains(&bob.id())); } let a = alice @@ -192,7 +197,7 @@ fn test_inventory_relay() { // Inventory from Bob relayed to Eve. alice.connect_to(&bob); - alice.connect_from(&eve.addr()); + alice.connect_from(&eve); alice.receive( &bob.addr(), Message::Inventory { @@ -204,7 +209,7 @@ fn test_inventory_relay() { assert_matches!( alice.messages(&eve.addr()).next(), Some(Message::Inventory { timestamp, origin, .. }) - if origin == Some(bob.ip) && timestamp == now + if origin == Some(bob.id()) && timestamp == now ); assert_matches!( alice.messages(&bob.addr()).next(), @@ -237,7 +242,7 @@ fn test_inventory_relay() { assert_matches!( alice.messages(&eve.addr()).next(), Some(Message::Inventory { timestamp, origin, .. }) - if origin == Some(bob.ip) && timestamp == now + 1, + if origin == Some(bob.id()) && timestamp == now + 1, "Sending a new inventory does trigger the relay" ); @@ -253,7 +258,7 @@ fn test_inventory_relay() { assert_matches!( alice.messages(&bob.addr()).next(), Some(Message::Inventory { timestamp, origin, .. }) - if origin == Some(eve.ip) && timestamp == now + if origin == Some(eve.id()) && timestamp == now ); } @@ -334,9 +339,9 @@ fn prop_inventory_exchange_dense() { let mut routing = Routing::with_hasher(rng.clone().into()); for (inv, peer) in &[ - (alice_inv.inventory, alice.addr().ip()), - (bob_inv.inventory, bob.addr().ip()), - (eve_inv.inventory, eve.addr().ip()), + (alice_inv.inventory, alice.id()), + (bob_inv.inventory, bob.id()), + (eve_inv.inventory, eve.id()), ] { for (proj, _) in inv { routing @@ -352,7 +357,7 @@ fn prop_inventory_exchange_dense() { eve.command(Command::Connect(alice.addr())); eve.command(Command::Connect(bob.addr())); - let mut peers: HashMap<_, _> = [(alice.ip, alice), (bob.ip, bob), (eve.ip, eve)] + let mut peers: HashMap<_, _> = [(alice.id(), alice), (bob.id(), bob), (eve.id(), eve)] .into_iter() .collect(); let mut simulator = Simulation::new(LocalTime::now(), rng, simulator::Options::default())