add identity layer

2 files changed, 154 insertions, 0 deletions

diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
new file mode 100644
index 0000000..a33783f
--- /dev/null
+++ b/src/simtest/identity.rs
@@ -0,0 +1,128 @@
+//! Nodes under the identity layer (`transport::identity`): named by node IDs, found
+//! through addresses (seeds, HELLO_REPLYs, the member list), and believed only once a
+//! node proved itself at an address.
+
+use std::time::Duration;
+
+use serde_json::json;
+
+use super::{Sim, read, sim_key, write};
+use crate::identity::{self, NodeKey};
+use crate::state::AppState;
+use crate::swarm::bootstrap::Start;
+use crate::swarm::members;
+use crate::transport::identity::{WHO_LABEL, signed};
+use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage};
+
+const SOON: Duration = Duration::from_secs(20);
+
+async fn peers_are_node_ids(node: &AppState) {
+    for (peer, _) in node.peers_snapshot().await {
+        assert!(identity::is_node_id(&peer), "peer {peer}");
+    }
+}
+
+/// Seeds by name (resolved by the net) and by address; the third node finds the second
+/// only through what the first lists in its HELLO_REPLY.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn nodes_are_their_node_ids() {
+    let sim = Sim::with_identity(51);
+    let a = sim.start("a", &[], Start::Genesis);
+    let b = sim.start("b", &["a"], Start::Join);
+    let c = sim.start("c", &["sim:a"], Start::Join);
+    sim.wait_ready(&[&a, &b, &c], SOON).await;
+    assert_eq!(a.local_id(), sim_key("a").id());
+
+    write(&b, "x", json!(1)).await;
+    sim.wait_converged(&[&a, &b, &c], SOON).await;
+    assert_eq!(read(&c, "x").await, Some(json!(1)));
+    for node in [&a, &b, &c] {
+        peers_are_node_ids(node).await;
+    }
+    let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned());
+    let a = &a;
+    sim.wait_until(
+        "b and c are registered with their addresses",
+        SOON,
+        move || {
+            let (b_id, c_id) = (b_id.clone(), c_id.clone());
+            async move {
+                let replica = a.doc.read().await;
+                members::addresses(&replica.doc, &b_id) == ["sim:b"]
+                    && members::addresses(&replica.doc, &c_id) == ["sim:c"]
+            }
+        },
+    )
+    .await;
+}
+
+/// With two neighbours each, most pairs of six nodes never meet through seeds: they
+/// find each other's addresses in the member list.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn members_find_each_other_through_the_member_list() {
+    let sim = Sim::with_identity(52);
+    let two = |config: &mut crate::node::NodeConfig| config.neighbours_k = 2;
+    let a = sim.start_with("a", &[], Start::Genesis, two);
+    let others: Vec<AppState> = ["b", "c", "d", "e", "f"]
+        .into_iter()
+        .map(|id| sim.start_with(id, &["a"], Start::Join, two))
+        .collect();
+    let mut all: Vec<&AppState> = vec![&a];
+    all.extend(others.iter());
+    sim.wait_ready(&all, SOON).await;
+
+    write(&others[4], "from/f", json!("f")).await;
+    write(&others[0], "from/b", json!("b")).await;
+    sim.wait_converged(&all, SOON).await;
+    for node in &all {
+        assert_eq!(read(node, "from/f").await, Some(json!("f")));
+        assert_eq!(read(node, "from/b").await, Some(json!("b")));
+        peers_are_node_ids(node).await;
+    }
+}
+
+fn ping(node: &AppState) -> ParsedMessage {
+    ParsedMessage::Ping(Keepalive {
+        heads: HeadsSummary::of(Vec::new()),
+        epoch: node.local_epoch(),
+        epoch_seq: 0,
+        ready: false,
+        syncs_with_you: false,
+    })
+}
+
+fn who(key: &NodeKey, from: &str, to: &str) -> ParsedMessage {
+    ParsedMessage::Who(IdentityProof {
+        key: key.public(),
+        signature: key.sign(&signed(WHO_LABEL, from, to)),
+    })
+}
+
+/// A proof counts only for the address it was made at: messages from an address with
+/// a proof made elsewhere are dropped, and the node never hears of them.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_proof_holds_only_at_its_address() {
+    let sim = Sim::with_identity(53);
+    let a = sim.start("a", &[], Start::Genesis);
+    sim.wait_ready(&[&a], SOON).await;
+    let ghost = NodeKey::from_seed([9; 32]);
+
+    // Made for another sender address, or for another recipient.
+    for forged in [
+        who(&ghost, "sim:elsewhere", "sim:a"),
+        who(&ghost, "sim:ghost", "sim:b"),
+    ] {
+        sim.net.inject("ghost", "a", &forged).await.unwrap();
+        sim.net.inject("ghost", "a", &ping(&a)).await.unwrap();
+        assert!(a.get_peer_state(&ghost.id()).await.is_none());
+    }
+
+    sim.net
+        .inject("ghost", "a", &who(&ghost, "sim:ghost", "sim:a"))
+        .await
+        .unwrap();
+    sim.net.inject("ghost", "a", &ping(&a)).await.unwrap();
+    assert!(a.get_peer_state(&ghost.id()).await.is_some());
+    // Known by its node ID, not by its address.
+    assert!(a.get_peer_state("ghost").await.is_none());
+}
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 531d1c6..4d7374a 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -10,6 +10,7 @@
 //! (all nodes interleaved; messages name peers by ID).
 
 mod deltas;
+mod identity;
 mod remakes;
 mod soak;
 mod timing;
@@ -26,10 +27,12 @@ use tokio::time::Instant;
 use crate::document::replica::Epoch;
 use crate::document::store;
 use crate::files::Files;
+use crate::identity::NodeKey;
 use crate::locks::RwLockExt;
 use crate::node::{self, Node, NodeConfig};
 use crate::state::{AppState, PeerState, Timing};
 use crate::swarm::bootstrap::{self, Start};
+use crate::transport::identity::IdentityConfig;
 use crate::transport::sim::SimNet;
 
 /// Keepalive of simulated nodes: dormancy after 1 s, probes of dormant peers every
@@ -45,6 +48,9 @@ pub(crate) const DELIVERY_LIMIT: Duration = Duration::from_secs(3);
 /// A simulated swarm: the net and the nodes started on it.
 pub(crate) struct Sim {
     pub(crate) net: SimNet,
+    /// Nodes under the identity layer (`transport::identity`), named by node IDs; else
+    /// by their names on the net.
+    identity: bool,
     /// Every node started, for the description of a failed wait.
     nodes: std::sync::Mutex<Vec<(String, AppState)>>,
 }
@@ -54,10 +60,20 @@ impl Sim {
         init_logging();
         Self {
             net: SimNet::new(seed),
+            identity: false,
             nodes: Default::default(),
         }
     }
 
+    /// [`new`](Self::new), with every node under the identity layer: its key derived
+    /// from its name, its node ID as `local_id`.
+    pub(crate) fn with_identity(seed: u64) -> Self {
+        Self {
+            identity: true,
+            ..Self::new(seed)
+        }
+    }
+
     /// Starts node `id` with the test defaults: builds it, starts its workers, and
     /// bootstraps it in the background (through `seeds`, other nodes' IDs). Returns at
     /// once; [`wait_ready`] waits for it.
@@ -75,6 +91,10 @@ impl Sim {
     ) -> AppState {
         let mut config = NodeConfig {
             transport: self.net.transport(id),
+            identity: self.identity.then(|| IdentityConfig {
+                key: Arc::new(sim_key(id)),
+                scheme: "sim",
+            }),
             initial_epoch: Epoch::GENESIS,
             initial_doc: AutoCommit::new(),
             earlier_actors: Vec::new(),
@@ -239,6 +259,12 @@ impl Sim {
     }
 }
 
+/// The node key of simulated node `id`.
+pub(crate) fn sim_key(id: &str) -> NodeKey {
+    use sha2::Digest;
+    NodeKey::from_seed(sha2::Sha256::digest(id.as_bytes()).into())
+}
+
 fn path(path: &str) -> Vec<String> {
     path.split('/').map(str::to_owned).collect()
 }