From 92f17d71c3e851174913a47f8c4addfaafa80c8d Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Mon, 29 Aug 2022 17:39:56 +0200 Subject: [PATCH] node: Implement auto-fetching of projects Signed-off-by: Alexis Sellier --- node/src/protocol.rs | 124 +++++++++++++++++++++++++-------------- node/src/storage.rs | 5 ++ node/src/storage/git.rs | 8 +++ node/src/test/peer.rs | 23 +++++--- node/src/test/storage.rs | 12 +++- node/src/test/tests.rs | 40 ++++++++++--- 6 files changed, 148 insertions(+), 64 deletions(-) diff --git a/node/src/protocol.rs b/node/src/protocol.rs index 859195ba..9b6fb7e3 100644 --- a/node/src/protocol.rs +++ b/node/src/protocol.rs @@ -180,6 +180,13 @@ impl Config { self.connect.contains(addr) } + pub fn is_tracking(&self, proj: &ProjId) -> bool { + match &self.project_tracking { + ProjectTracking::All { blocked } => !blocked.contains(proj), + ProjectTracking::Allowed(projs) => projs.contains(proj), + } + } + /// Track a project. Returns whether the policy was updated. pub fn track(&mut self, proj: ProjId) -> bool { match &mut self.project_tracking { @@ -686,7 +693,7 @@ pub struct Context { impl Context where - T: storage::ReadStorage, + T: storage::ReadStorage + storage::WriteStorage, { pub(crate) fn new( config: Config, @@ -706,22 +713,25 @@ where } } - /// Get current local timestamp. - pub(crate) fn timestamp(&self) -> Timestamp { - self.clock.local_time().as_secs() - } - /// Process a peer inventory announcement by updating our routing table. - fn process_inventory(&mut self, inventory: &Inventory, from: PeerId) { + fn process_inventory(&mut self, inventory: &Inventory, from: PeerId, remote: &Url) { for proj_id in inventory { let inventory = self .routing .entry(proj_id.clone()) .or_insert_with(|| HashSet::with_hasher(self.rng.clone().into())); - // TODO: If we're tracking this project, check the refs to see if we need to - // fetch updates from this peer. + if self.config.is_tracking(proj_id) { + // TODO: Verify refs before adding them to storage. + let mut repo = self.storage.repository(proj_id).unwrap(); + repo.fetch(&Url { + path: format!("/{}", proj_id).into(), + ..remote.clone() + }) + .unwrap(); + } + // TODO: Fire an event on routing update. inventory.insert(from); } } @@ -733,6 +743,11 @@ where } impl Context { + /// Get current local timestamp. + pub(crate) fn timestamp(&self) -> Timestamp { + self.clock.local_time().as_secs() + } + /// Connect to a peer. fn connect(&mut self, addr: net::SocketAddr) { // TODO: Make sure we don't try to connect more than once to the same address. @@ -816,6 +831,8 @@ pub enum PeerError { WrongVersion(u32), #[error("invalid inventory timestamp: {0}")] InvalidTimestamp(u64), + #[error("peer misbehaved")] + Misbehavior, } #[derive(Debug)] @@ -866,19 +883,22 @@ impl Peer { ctx: &mut Context, ) -> Result, PeerError> where - T: storage::ReadStorage, + T: storage::ReadStorage + storage::WriteStorage, { if envelope.magic != NETWORK_MAGIC { return Err(PeerError::WrongMagic(envelope.magic)); } debug!("Received {:?} from {}", &envelope.msg, self.id()); - match envelope.msg { - Message::Hello { - timestamp, - version, - git, - } => { + match (&self.state, envelope.msg) { + ( + PeerState::Initial, + Message::Hello { + timestamp, + version, + git, + }, + ) => { let now = ctx.timestamp(); if timestamp.abs_diff(now) > MAX_TIME_DELTA.as_secs() { @@ -887,37 +907,43 @@ impl Peer { if version != PROTOCOL_VERSION { return Err(PeerError::WrongVersion(version)); } - if let PeerState::Initial = self.state { - // Nb. This is a very primitive handshake. Eventually we should have anyhow - // extra "acknowledgment" message sent when the `Hello` is well received. - if self.link.is_inbound() { - let git = ctx.config.git_url.clone(); - ctx.write_all( - self.addr, - [Message::hello(now, git), Message::get_inventory([])], - ); - } - // Nb. we don't set the peer timestamp here, since it is going to be - // 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 { - since: ctx.clock.local_time(), - git, - }; - } else { - // TODO: Handle misbehavior. + // Nb. This is a very primitive handshake. Eventually we should have anyhow + // extra "acknowledgment" message sent when the `Hello` is well received. + if self.link.is_inbound() { + let git = ctx.config.git_url.clone(); + ctx.write_all( + self.addr, + [Message::hello(now, git), Message::get_inventory([])], + ); } + // Nb. we don't set the peer timestamp here, since it is going to be + // 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 { + since: ctx.clock.local_time(), + git, + }; } - Message::GetInventory { .. } => { + (PeerState::Initial, _) => { + debug!( + "Disconnecting peer {} for sending us a message before handshake", + self.id() + ); + return Err(PeerError::Misbehavior); + } + (PeerState::Negotiated { .. }, Message::GetInventory { .. }) => { // TODO: Handle partial inventory requests. let inventory = Message::inventory(ctx).unwrap(); ctx.write(self.addr, inventory); } - Message::Inventory { - timestamp, - inv, - origin, - } => { + ( + PeerState::Negotiated { git, .. }, + Message::Inventory { + timestamp, + inv, + origin, + }, + ) => { let now = ctx.clock.local_time(); let last = self.timestamp; @@ -932,7 +958,7 @@ impl Peer { } else { return Ok(None); } - ctx.process_inventory(&inv, origin.unwrap_or_else(|| self.id())); + ctx.process_inventory(&inv, origin.unwrap_or_else(|| self.id()), git); if ctx.config.relay { return Ok(Some(Message::Inventory { @@ -942,14 +968,24 @@ impl Peer { })); } } - Message::GetAddrs => { + (PeerState::Negotiated { .. }, Message::GetAddrs) => { // TODO: Send peer addresses. todo!(); } - Message::Addrs { .. } => { + (PeerState::Negotiated { .. }, Message::Addrs { .. }) => { // TODO: Update address book. todo!(); } + (PeerState::Negotiated { .. }, Message::Hello { .. }) => { + debug!( + "Disconnecting peer {} for sending us a redundant handshake message", + self.id() + ); + return Err(PeerError::Misbehavior); + } + (PeerState::Disconnected { .. }, msg) => { + debug!("Ignoring {:?} from disconnected peer {}", msg, self.id()); + } } Ok(None) } diff --git a/node/src/storage.rs b/node/src/storage.rs index fd5be0a4..e4271a79 100644 --- a/node/src/storage.rs +++ b/node/src/storage.rs @@ -116,6 +116,7 @@ impl Remote { } pub trait ReadStorage { + fn url(&self) -> Url; fn get(&self, proj: &ProjId) -> Result>, Error>; fn inventory(&self) -> Result; } @@ -140,6 +141,10 @@ where T: Deref, S: ReadStorage, { + fn url(&self) -> Url { + self.deref().url() + } + fn inventory(&self) -> Result { self.deref().inventory() } diff --git a/node/src/storage/git.rs b/node/src/storage/git.rs index 2d615b00..4f17f497 100644 --- a/node/src/storage/git.rs +++ b/node/src/storage/git.rs @@ -41,6 +41,14 @@ impl fmt::Debug for Storage { } impl ReadStorage for Storage { + fn url(&self) -> Url { + Url { + scheme: git_url::Scheme::File, + host: Some(self.path.to_string_lossy().to_string()), + ..Url::default() + } + } + fn get(&self, _id: &ProjId) -> Result>, Error> { todo!() } diff --git a/node/src/test/peer.rs b/node/src/test/peer.rs index 0039c460..bafff94f 100644 --- a/node/src/test/peer.rs +++ b/node/src/test/peer.rs @@ -63,7 +63,10 @@ where pub fn new(name: &'static str, ip: impl Into, storage: S) -> Self { Self::config( name, - Config::default(), + Config { + git_url: storage.url(), + ..Config::default() + }, ip, vec![], storage, @@ -143,20 +146,22 @@ where .expect("`get-inventory` is sent"); } - pub fn connect_to(&mut self, remote: &net::SocketAddr) { - self.initialize(); - self.protocol.attempted(remote); - self.protocol - .connected(*remote, &self.local_addr, Link::Outbound); + pub fn connect_to(&mut self, peer: &Self) { + let remote = simulator::Peer::>::addr(peer); - let mut msgs = self.messages(remote); + self.initialize(); + self.protocol.attempted(&remote); + self.protocol + .connected(remote, &self.local_addr, Link::Outbound); + + let mut msgs = self.messages(&remote); msgs.find(|m| matches!(m, Message::Hello { .. })) .expect("`hello` is sent"); msgs.find(|m| matches!(m, Message::GetInventory { .. })) .expect("`get-inventory` is sent"); - let git = self.config().git_url.clone(); - self.receive(remote, Message::hello(self.local_time().as_secs(), git)); + let git = peer.config().git_url.clone(); + self.receive(&remote, Message::hello(self.local_time().as_secs(), git)); } /// Drain outgoing messages sent from this peer to the remote address. diff --git a/node/src/test/storage.rs b/node/src/test/storage.rs index 2f11c6b1..750073ca 100644 --- a/node/src/test/storage.rs +++ b/node/src/test/storage.rs @@ -23,6 +23,14 @@ impl MockStorage { } impl ReadStorage for MockStorage { + fn url(&self) -> Url { + Url { + scheme: git_url::Scheme::Radicle, + host: Some("mock".to_string()), + ..Url::default() + } + } + fn get(&self, proj: &ProjId) -> Result>, Error> { if let Some((_, refs)) = self.inventory.iter().find(|(id, _)| id == proj) { return Ok(Some(refs.clone())); @@ -45,7 +53,7 @@ impl WriteStorage for MockStorage { type Repository = MockRepository; fn repository(&self, _proj: &ProjId) -> Result { - todo!() + Ok(MockRepository {}) } } @@ -53,7 +61,7 @@ pub struct MockRepository {} impl WriteRepository for MockRepository { fn fetch(&mut self, _url: &Url) -> Result<(), git2::Error> { - todo!() + Ok(()) } fn namespace( diff --git a/node/src/test/tests.rs b/node/src/test/tests.rs index f34a7d63..f83cf13d 100644 --- a/node/src/test/tests.rs +++ b/node/src/test/tests.rs @@ -31,8 +31,8 @@ fn test_outbound_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_to(&bob.addr()); - alice.connect_to(&eve.addr()); + alice.connect_to(&bob); + alice.connect_to(&eve); let peers = alice .protocol @@ -120,20 +120,42 @@ fn test_wrong_peer_magic() { } #[test] -fn test_inventory_fetch() { - let mut alice = Peer::new("alice", [7, 7, 7, 7], MockStorage::empty()); - let bob = Peer::new("bob", [8, 8, 8, 8], MockStorage::empty()); +fn test_inventory_sync() { + let tmp = tempfile::tempdir().unwrap(); + let mut alice = Peer::new( + "alice", + [7, 7, 7, 7], + Storage::open(tmp.path().join("alice")).unwrap(), + ); + let bob_storage = fixtures::storage(tmp.path().join("bob")); + let bob = Peer::new("bob", [8, 8, 8, 8], bob_storage); let now = LocalTime::now().as_secs(); + let projs = bob.storage().inventory().unwrap(); - alice.connect_to(&bob.addr()); + alice.connect_to(&bob); alice.receive( &bob.addr(), Message::Inventory { timestamp: now, - inv: vec![], + inv: projs.clone(), origin: None, }, ); + + for proj in &projs { + let providers = alice.routing().get(proj).unwrap(); + assert!(providers.contains(&bob.ip)); + } + + let a = alice + .storage() + .inventory() + .unwrap() + .into_iter() + .collect::>(); + let b = projs.into_iter().collect::>(); + + assert_eq!(a, b); } #[test] @@ -143,7 +165,7 @@ fn test_inventory_relay_bad_timestamp() { let two_hours = 3600 * 2; let timestamp = alice.local_time.as_secs() - two_hours; - alice.connect_to(&bob.addr()); + alice.connect_to(&bob); alice.receive( &bob.addr(), Message::Inventory { @@ -169,7 +191,7 @@ fn test_inventory_relay() { let now = LocalTime::now().as_secs(); // Inventory from Bob relayed to Eve. - alice.connect_to(&bob.addr()); + alice.connect_to(&bob); alice.connect_from(&eve.addr()); alice.receive( &bob.addr(),