add identity layer

5 files changed, 610 insertions, 27 deletions

diff --git a/src/transport/identity.rs b/src/transport/identity.rs
new file mode 100644
index 0000000..712e1f9
--- /dev/null
+++ b/src/transport/identity.rs
@@ -0,0 +1,510 @@
+//! The identity layer: between the node and a transport whose identities are addresses
+//! (Mycelium's public keys), so that the rest of the node deals in node IDs
+//! (`crate::identity`) only, whatever transport reaches a peer.
+//!
+//! A node proves who it is to an address with WHO (which asks back) or ME (the answer):
+//! its public node key, and a signature over its own address and the recipient's
+//! (`signed`). The transport authenticates the sender's address, so a proof is good for
+//! that pair of addresses only: replayed from another address, or to another recipient,
+//! it fails. Once a node proved itself at an address, messages from there are its.
+//!
+//! * A message from an address no node proved itself at is dropped, and the address is
+//!   asked WHO (at most every `ASK_INTERVAL`): after a restart, the first message from
+//!   a peer brings its proof back, and the peer's node retries what was lost.
+//! * A seed (`resolve_seed`) is an address, or something the transport resolves to one;
+//!   it is asked WHO, and its ME says which node is there.
+//! * Where other nodes are comes from the member list and HELLO_REPLYs
+//!   (`learn_addresses`): hints, not proofs. A message to a node goes to the address it
+//!   proved itself at; with only hints, those are asked WHO, and the send is refused
+//!   until one answers (the node retries refused sends).
+//!
+//! Addresses are `<scheme>:<transport identity>`, e.g. `mycelium:<public key>`.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use anyhow::{Result, anyhow, bail};
+use async_trait::async_trait;
+use tokio::sync::oneshot;
+use tokio::time::Instant;
+use tracing::{debug, info};
+
+use crate::identity::{self, NodeKey};
+use crate::locks::MutexExt;
+use crate::transport::traffic::Traffic;
+use crate::transport::{
+    EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind, ParsedMessage, Transport,
+};
+
+pub(crate) const WHO_LABEL: &[u8] = b"continuum/who/1";
+pub(crate) const ME_LABEL: &[u8] = b"continuum/me/1";
+/// An address is asked WHO at most this often, unless a seed is being resolved.
+const ASK_INTERVAL: Duration = Duration::from_secs(5);
+/// How long resolving a seed waits for its ME.
+const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
+/// Addresses kept per node from hints: about one per transport.
+const MAX_HINTS: usize = 8;
+/// Addresses with a proven node, nodes with hints, and addresses asked, each at most:
+/// anyone can send us proofs (keys are free), so the books are bounded.
+const MAX_ENTRIES: usize = 4096;
+
+/// What a node under the identity layer needs (`node::NodeConfig::identity`).
+#[derive(Clone)]
+pub(crate) struct IdentityConfig {
+    pub(crate) key: Arc<NodeKey>,
+    /// The transport's address scheme, e.g. `mycelium`.
+    pub(crate) scheme: &'static str,
+}
+
+/// The bytes a WHO (`label`: `WHO_LABEL`) or ME proof signs: the label, then the
+/// sender's and the recipient's address, each with its length.
+pub(crate) fn signed(label: &[u8], from: &str, to: &str) -> Vec<u8> {
+    let mut bytes = label.to_vec();
+    for address in [from, to] {
+        bytes.extend_from_slice(&(address.len() as u32).to_be_bytes());
+        bytes.extend_from_slice(address.as_bytes());
+    }
+    bytes
+}
+
+pub(crate) struct IdentityTransport {
+    shared: Arc<Shared>,
+}
+
+struct Shared {
+    inner: Arc<dyn Transport>,
+    scheme: &'static str,
+    key: Arc<NodeKey>,
+    node_id: String,
+    /// WHO and ME are counted here, under the address (`traffic`).
+    traffic: Arc<Traffic>,
+    // A std mutex: every access is short and never spans an await.
+    book: Mutex<Book>,
+}
+
+/// Who is where. Keys are transport identities (an address without its scheme).
+#[derive(Default)]
+struct Book {
+    /// The node that proved itself at a transport identity, and when.
+    at: HashMap<String, (String, Instant)>,
+    /// The transport identity each node last proved itself at.
+    proven: HashMap<String, String>,
+    /// Where others say a node is.
+    hints: HashMap<String, Vec<String>>,
+    /// When each transport identity was last asked WHO.
+    asked: HashMap<String, Instant>,
+    /// Seeds being resolved: who waits for a ME from where.
+    waiting: HashMap<String, Vec<oneshot::Sender<String>>>,
+}
+
+impl Book {
+    /// `node` proved itself at `target`. Returns whether that is news.
+    fn bind(&mut self, target: &str, node: &str, now: Instant) -> bool {
+        let news = self.at.get(target).is_none_or(|(known, _)| known != node);
+        // One address per node and one node per address: a node that moved, or an
+        // address taken over by another node, leaves no stale entry behind.
+        if let Some(previous) = self.proven.insert(node.to_owned(), target.to_owned())
+            && previous != target
+        {
+            self.at.remove(&previous);
+        }
+        if let Some((previous, _)) = self.at.insert(target.to_owned(), (node.to_owned(), now))
+            && previous != node
+            && self.proven.get(&previous).is_some_and(|at| at == target)
+        {
+            self.proven.remove(&previous);
+        }
+        if self.at.len() > MAX_ENTRIES
+            && let Some(oldest) = self
+                .at
+                .iter()
+                .min_by_key(|(_, (_, at))| *at)
+                .map(|(target, _)| target.clone())
+            && let Some((node, _)) = self.at.remove(&oldest)
+        {
+            self.proven.remove(&node);
+        }
+        for waiter in self.waiting.remove(target).into_iter().flatten() {
+            let _ = waiter.send(node.to_owned());
+        }
+        news
+    }
+
+    /// Whether to ask `target` WHO now (always, if `force`), noting that it was asked.
+    fn ask(&mut self, target: &str, now: Instant, force: bool) -> bool {
+        if !force
+            && self
+                .asked
+                .get(target)
+                .is_some_and(|at| now.duration_since(*at) < ASK_INTERVAL)
+        {
+            return false;
+        }
+        if self.asked.len() >= MAX_ENTRIES {
+            self.asked
+                .retain(|_, at| now.duration_since(*at) < ASK_INTERVAL);
+        }
+        self.asked.insert(target.to_owned(), now);
+        true
+    }
+}
+
+impl IdentityTransport {
+    pub(crate) fn new(
+        inner: Arc<dyn Transport>,
+        config: IdentityConfig,
+        traffic: Arc<Traffic>,
+    ) -> Self {
+        let node_id = config.key.id();
+        Self {
+            shared: Arc::new(Shared {
+                inner,
+                scheme: config.scheme,
+                key: config.key,
+                node_id,
+                traffic,
+                book: Mutex::default(),
+            }),
+        }
+    }
+}
+
+impl Shared {
+    fn lock(&self) -> MutexGuard<'_, Book> {
+        self.book.lock_or_recover()
+    }
+
+    fn address(&self, target: &str) -> String {
+        format!("{}:{target}", self.scheme)
+    }
+
+    /// The transport identity in `address`, if it is one of this transport's.
+    fn target_of<'a>(&self, address: &'a str) -> Option<&'a str> {
+        address
+            .strip_prefix(self.scheme)
+            .and_then(|rest| rest.strip_prefix(':'))
+            .filter(|target| self.inner.is_peer_id(target))
+    }
+
+    fn proof(&self, label: &[u8], target: &str) -> IdentityProof {
+        let message = signed(
+            label,
+            &self.address(self.inner.local_id()),
+            &self.address(target),
+        );
+        IdentityProof {
+            key: self.key.public(),
+            signature: self.key.sign(&message),
+        }
+    }
+
+    /// Notes the node `proof` proves to be at `sender`; None if it proves nothing.
+    fn prove(&self, sender: &str, label: &[u8], proof: &IdentityProof) -> Option<String> {
+        let message = signed(
+            label,
+            &self.address(sender),
+            &self.address(self.inner.local_id()),
+        );
+        if !identity::verify(&proof.key, &message, &proof.signature) {
+            debug!(from = %self.address(sender), "Ignoring an identity proof that doesn't hold");
+            return None;
+        }
+        let node = identity::node_id_of(&proof.key);
+        if node == self.node_id {
+            debug!(from = %self.address(sender), "Ignoring a proof of our own node key");
+            return None;
+        }
+        if self.lock().bind(sender, &node, Instant::now()) {
+            debug!(node = %node, address = %self.address(sender), "Node proved its address");
+        }
+        Some(node)
+    }
+
+    /// Asks `target` WHO, unless it was asked recently (`force`: anyway).
+    fn ask(self: &Arc<Self>, target: &str, force: bool) {
+        if !self.lock().ask(target, Instant::now(), force) {
+            return;
+        }
+        let shared = Arc::clone(self);
+        let target = target.to_owned();
+        tokio::spawn(async move { shared.send_proof(MessageKind::Who, &target).await });
+    }
+
+    /// Sends a WHO or ME (`kind`) to `target`.
+    async fn send_proof(&self, kind: MessageKind, target: &str) {
+        let message = if kind == MessageKind::Who {
+            ParsedMessage::Who(self.proof(WHO_LABEL, target))
+        } else {
+            ParsedMessage::Me(self.proof(ME_LABEL, target))
+        };
+        let encoded = match self.inner.encode(target, &message) {
+            Ok(encoded) => encoded,
+            Err(error) => {
+                debug!(to = %target, error = %format!("{error:#}"), "Couldn't encode an identity proof");
+                return;
+            }
+        };
+        let address = self.address(target);
+        match self.inner.send_encoded(&encoded).await {
+            Ok(()) => self
+                .traffic
+                .sent(&address, encoded.kind, encoded.payload_len),
+            Err(error) => {
+                self.traffic
+                    .refused(&address, encoded.kind, encoded.payload_len);
+                debug!(to = %address, error = %format!("{error:#}"), "Couldn't send an identity proof");
+            }
+        }
+    }
+
+    async fn receive(
+        self: Arc<Self>,
+        sender: String,
+        message: ParsedMessage,
+        payload_len: usize,
+        handler: &HandlerFn,
+    ) -> Result<()> {
+        match message {
+            ParsedMessage::Who(proof) => {
+                self.traffic
+                    .received(&self.address(&sender), MessageKind::Who, payload_len);
+                if self.prove(&sender, WHO_LABEL, &proof).is_some() {
+                    let shared = Arc::clone(&self);
+                    tokio::spawn(async move { shared.send_proof(MessageKind::Me, &sender).await });
+                }
+                Ok(())
+            }
+            ParsedMessage::Me(proof) => {
+                self.traffic
+                    .received(&self.address(&sender), MessageKind::Me, payload_len);
+                self.prove(&sender, ME_LABEL, &proof);
+                Ok(())
+            }
+            message => {
+                let node = self.lock().at.get(&sender).map(|(node, _)| node.clone());
+                match node {
+                    Some(node) => handler(node, message, payload_len).await,
+                    None => {
+                        debug!(
+                            from = %self.address(&sender),
+                            kind = ?message.kind(),
+                            "Dropping a message from an address no node proved itself at; asking who is there"
+                        );
+                        self.ask(&sender, false);
+                        Ok(())
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl Transport for IdentityTransport {
+    fn local_id(&self) -> &str {
+        &self.shared.node_id
+    }
+
+    fn is_peer_id(&self, id: &str) -> bool {
+        identity::is_node_id(id)
+    }
+
+    fn delivery_limit(&self) -> Duration {
+        self.shared.inner.delivery_limit()
+    }
+
+    fn local_addresses(&self) -> Vec<String> {
+        vec![self.shared.address(self.shared.inner.local_id())]
+    }
+
+    fn addresses_of(&self, peer: &str) -> Vec<String> {
+        let book = self.shared.lock();
+        match book.proven.get(peer) {
+            Some(target) => vec![self.shared.address(target)],
+            None => book
+                .hints
+                .get(peer)
+                .into_iter()
+                .flatten()
+                .map(|target| self.shared.address(target))
+                .collect(),
+        }
+    }
+
+    fn learn_addresses(&self, peer: &str, addresses: &[String]) {
+        if peer == self.shared.node_id || !identity::is_node_id(peer) {
+            return;
+        }
+        let local = self.shared.inner.local_id();
+        let mut targets: Vec<String> = addresses
+            .iter()
+            .filter_map(|address| self.shared.target_of(address))
+            .filter(|target| *target != local)
+            .map(str::to_owned)
+            .collect();
+        targets.dedup();
+        targets.truncate(MAX_HINTS);
+        if targets.is_empty() {
+            return;
+        }
+        let mut book = self.shared.lock();
+        if book.hints.len() < MAX_ENTRIES || book.hints.contains_key(peer) {
+            book.hints.insert(peer.to_owned(), targets);
+        }
+    }
+
+    /// A node ID resolves to itself if this node knows where it is; anything else is
+    /// an address (`<scheme>:...`) or what the transport resolves to one, asked WHO.
+    async fn resolve_seed(&self, seed: &str) -> Result<String> {
+        if identity::is_node_id(seed) {
+            let book = self.shared.lock();
+            if book.proven.contains_key(seed) || book.hints.contains_key(seed) {
+                return Ok(seed.to_owned());
+            }
+            bail!("a node ID alone doesn't say where that node is; give its address");
+        }
+        let target = match self.shared.target_of(seed) {
+            Some(target) => target.to_owned(),
+            None => self.shared.inner.resolve_seed(seed).await?,
+        };
+        if target == self.shared.inner.local_id() {
+            bail!("{seed} is this node");
+        }
+        let receiver = {
+            let mut book = self.shared.lock();
+            if let Some((node, _)) = book.at.get(&target) {
+                return Ok(node.clone());
+            }
+            let (sender, receiver) = oneshot::channel();
+            let waiting = book.waiting.entry(target.clone()).or_default();
+            waiting.retain(|waiter| !waiter.is_closed());
+            waiting.push(sender);
+            receiver
+        };
+        self.shared.ask(&target, true);
+        match tokio::time::timeout(RESOLVE_TIMEOUT, receiver).await {
+            Ok(Ok(node)) => {
+                info!(seed = %seed, node = %node, "Seed is node");
+                Ok(node)
+            }
+            _ => Err(anyhow!(
+                "no node answered at {} within {RESOLVE_TIMEOUT:?}",
+                self.shared.address(&target)
+            )),
+        }
+    }
+
+    /// For the address `peer` proved itself at, or else one it is said to be at (a send
+    /// there is refused until it proves itself, `send_encoded`).
+    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+        let target = {
+            let book = self.shared.lock();
+            book.proven
+                .get(peer)
+                .or_else(|| book.hints.get(peer).and_then(|hints| hints.first()))
+                .cloned()
+        };
+        let Some(target) = target else {
+            bail!("no address known for {peer}");
+        };
+        let mut encoded = self.shared.inner.encode(&target, message)?;
+        encoded.peer = peer.to_owned();
+        Ok(encoded)
+    }
+
+    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
+        let hints = {
+            let book = self.shared.lock();
+            if book.proven.get(&message.peer) == Some(&message.address) {
+                None
+            } else {
+                Some(book.hints.get(&message.peer).cloned().unwrap_or_default())
+            }
+        };
+        if let Some(hints) = hints {
+            for target in hints {
+                self.shared.ask(&target, false);
+            }
+            bail!(
+                "{} hasn't proven where it is yet; asked where it is said to be",
+                message.peer
+            );
+        }
+        let mut inner = message.clone();
+        inner.peer = message.address.clone();
+        self.shared.inner.send_encoded(&inner).await
+    }
+
+    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
+        let handler: Arc<HandlerFn> = Arc::from(handler);
+        let shared = Arc::clone(&self.shared);
+        self.shared
+            .inner
+            .start_consumer(Box::new(move |sender, message, payload_len| {
+                let shared = Arc::clone(&shared);
+                let handler = Arc::clone(&handler);
+                Box::pin(async move {
+                    shared
+                        .receive(sender, message, payload_len, &*handler)
+                        .await
+                })
+            }))
+            .await
+    }
+
+    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+        self.shared.inner.wait_for_exit().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn signed_bytes_say_both_addresses() {
+        // A length before each address: "a:bc" to "d" isn't "a:b" to "cd".
+        assert_ne!(
+            signed(WHO_LABEL, "a:bc", "d"),
+            signed(WHO_LABEL, "a:b", "cd")
+        );
+        assert_ne!(signed(WHO_LABEL, "a", "b"), signed(ME_LABEL, "a", "b"));
+        assert_ne!(signed(WHO_LABEL, "a", "b"), signed(WHO_LABEL, "b", "a"));
+    }
+
+    #[test]
+    fn one_address_per_node_and_one_node_per_address() {
+        let now = Instant::now();
+        let mut book = Book::default();
+        assert!(book.bind("x", "n1", now));
+        assert!(!book.bind("x", "n1", now));
+        // n1 moved to y: x is no longer n1's.
+        assert!(book.bind("y", "n1", now));
+        assert!(!book.at.contains_key("x"));
+        assert_eq!(book.proven["n1"], "y");
+        // n2 took over y: n1 is nowhere proven.
+        assert!(book.bind("y", "n2", now));
+        assert!(!book.proven.contains_key("n1"));
+        assert_eq!(book.at["y"].0, "n2");
+    }
+
+    #[test]
+    fn asking_is_rate_limited() {
+        let now = Instant::now();
+        let mut book = Book::default();
+        assert!(book.ask("x", now, false));
+        assert!(!book.ask("x", now, false));
+        assert!(book.ask("x", now, true));
+        assert!(book.ask("x", now + ASK_INTERVAL, false));
+    }
+
+    #[tokio::test]
+    async fn a_waiting_seed_learns_the_node() {
+        let mut book = Book::default();
+        let (sender, receiver) = oneshot::channel();
+        book.waiting.insert("x".into(), vec![sender]);
+        book.bind("x", "n1", Instant::now());
+        assert_eq!(receiver.await.unwrap(), "n1");
+    }
+}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 29c48d6..52db406 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -12,6 +12,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 pub(crate) use crate::document::replica::Epoch;
 
 pub(crate) mod firewall;
+pub(crate) mod identity;
 pub(crate) mod mycelium;
 #[cfg(test)]
 pub(crate) mod sim;
@@ -137,6 +138,31 @@ pub(crate) enum ParsedMessage {
     Ping(Keepalive),
     Pong(Keepalive),
     Blob(BlobMessage),
+    /// Asks the recipient who it is, saying who the sender is (`transport::identity`).
+    Who(IdentityProof),
+    /// Answers a WHO.
+    Me(IdentityProof),
+}
+
+/// A node's proof that it holds the key of its node ID (`identity`), sending from its
+/// transport address to the recipient's: a signature over both addresses
+/// (`transport::identity::signed`). WHO and ME end at the identity layer; the rest of
+/// the node never sees them.
+#[derive(Clone, Serialize, Deserialize)]
+pub(crate) struct IdentityProof {
+    /// The sender's public node key.
+    #[serde(with = "crate::transport::wire::codec::bytes32")]
+    pub(crate) key: [u8; 32],
+    #[serde(with = "crate::transport::wire::codec::bytes64")]
+    pub(crate) signature: [u8; 64],
+}
+
+impl std::fmt::Debug for IdentityProof {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("IdentityProof")
+            .field("key", &Hex(&self.key))
+            .finish_non_exhaustive()
+    }
 }
 
 /// Content-addressed bytes that any node holding them can serve, in chunks
@@ -237,6 +263,8 @@ pub(crate) enum MessageKind {
     Ping,
     Pong,
     Blob,
+    Who,
+    Me,
 }
 
 impl ParsedMessage {
@@ -248,28 +276,23 @@ impl ParsedMessage {
             ParsedMessage::Ping(_) => MessageKind::Ping,
             ParsedMessage::Pong(_) => MessageKind::Pong,
             ParsedMessage::Blob(_) => MessageKind::Blob,
+            ParsedMessage::Who(_) => MessageKind::Who,
+            ParsedMessage::Me(_) => MessageKind::Me,
         }
     }
 
-    /// Every message carries the sender's epoch.
-    pub(crate) fn epoch(&self) -> Epoch {
-        match self {
-            ParsedMessage::Hello(hello) => hello.epoch,
-            ParsedMessage::HelloReply(reply) => reply.epoch,
-            ParsedMessage::Sync(envelope) => envelope.epoch,
-            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
-            ParsedMessage::Blob(blob) => blob.epoch,
-        }
-    }
-
-    /// `SyncEnvelope::epoch_seq`, as every message carries it.
-    pub(crate) fn epoch_seq(&self) -> u32 {
+    /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them
+    /// but WHO and ME (which the node never sees).
+    pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> {
         match self {
-            ParsedMessage::Hello(hello) => hello.epoch_seq,
-            ParsedMessage::HelloReply(reply) => reply.epoch_seq,
-            ParsedMessage::Sync(envelope) => envelope.epoch_seq,
-            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq,
-            ParsedMessage::Blob(blob) => blob.epoch_seq,
+            ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)),
+            ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)),
+            ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)),
+            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => {
+                Some((keepalive.epoch, keepalive.epoch_seq))
+            }
+            ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
+            ParsedMessage::Who(_) | ParsedMessage::Me(_) => None,
         }
     }
 }
@@ -472,6 +495,9 @@ impl std::fmt::Debug for SyncPayload {
 #[derive(Clone, Debug)]
 pub(crate) struct EncodedMessage {
     pub(crate) peer: String,
+    /// Where the transport sends it: `peer` itself for a transport whose identities are
+    /// addresses, the peer's address under the identity layer (`transport::identity`).
+    pub(crate) address: String,
     pub(crate) body: bytes::Bytes,
     pub(crate) kind: MessageKind,
     /// Bytes the message takes on the wire (without the transport's own framing).
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 17344c7..354aebc 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -446,6 +446,7 @@ pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<Enco
     let (outbound, payload_len) = format_message(peer, message)?;
     Ok(EncodedMessage {
         peer: peer.to_owned(),
+        address: peer.to_owned(),
         body: serde_json::to_vec(&outbound)?.into(),
         kind: message.kind(),
         payload_len,
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index c27f0d7..59acaf2 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -40,6 +40,9 @@ pub(crate) struct ByKind {
     pub(crate) ping: Counter,
     pub(crate) pong: Counter,
     pub(crate) blob: Counter,
+    /// WHO and ME (`transport::identity`), counted by the identity layer under the
+    /// address they went to or came from.
+    pub(crate) identity: Counter,
 }
 
 impl ByKind {
@@ -51,6 +54,7 @@ impl ByKind {
             MessageKind::Ping => &mut self.ping,
             MessageKind::Pong => &mut self.pong,
             MessageKind::Blob => &mut self.blob,
+            MessageKind::Who | MessageKind::Me => &mut self.identity,
         }
     }
 }
@@ -92,15 +96,15 @@ impl Traffic {
         }
     }
 
-    fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) {
+    pub(crate) fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) {
         self.count(peer, bytes, |t| t.sent.get_mut(kind));
     }
 
-    fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) {
+    pub(crate) fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) {
         self.count(peer, bytes, |t| t.refused.get_mut(kind));
     }
 
-    fn received(&self, peer: &str, kind: MessageKind, bytes: usize) {
+    pub(crate) fn received(&self, peer: &str, kind: MessageKind, bytes: usize) {
         self.count(peer, bytes, |t| t.received.get_mut(kind));
     }
 
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index ceeaf87..5577f42 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage};
 
 /// The wire format's version, the first byte of every payload. Bump it with every change
 /// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 4;
+pub(crate) const VERSION: u8 = 5;
 
 /// What the second byte of a payload says.
 fn tag(kind: MessageKind) -> u8 {
@@ -25,6 +25,8 @@ fn tag(kind: MessageKind) -> u8 {
         MessageKind::Ping => 3,
         MessageKind::Pong => 4,
         MessageKind::Blob => 5,
+        MessageKind::Who => 6,
+        MessageKind::Me => 7,
     }
 }
 
@@ -57,6 +59,9 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
             rmp_serde::encode::write(&mut payload, keepalive)?
         }
         ParsedMessage::Blob(blob) => rmp_serde::encode::write(&mut payload, blob)?,
+        ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => {
+            rmp_serde::encode::write(&mut payload, proof)?
+        }
     }
     Ok(payload)
 }
@@ -81,6 +86,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
         ParsedMessage::Pong(rmp_serde::from_slice(body)?)
     } else if is(MessageKind::Blob) {
         ParsedMessage::Blob(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Who) {
+        ParsedMessage::Who(rmp_serde::from_slice(body)?)
+    } else if is(MessageKind::Me) {
+        ParsedMessage::Me(rmp_serde::from_slice(body)?)
     } else {
         return Err(WireError::UnknownKind(kind));
     })
@@ -125,6 +134,19 @@ pub(crate) mod codec {
         }
     }
 
+    /// A 64-byte signature.
+    pub(crate) mod bytes64 {
+        use super::*;
+
+        pub(crate) fn serialize<S: Serializer>(value: &[u8; 64], s: S) -> Result<S::Ok, S::Error> {
+            Bytes::new(value).serialize(s)
+        }
+
+        pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 64], D::Error> {
+            fixed(&ByteBuf::deserialize(d)?, "a signature")
+        }
+    }
+
     /// 32-byte hashes.
     pub(crate) mod hashes32 {
         use super::*;
@@ -299,8 +321,8 @@ mod tests {
     use super::*;
     use crate::document::replica::Epoch;
     use crate::transport::{
-        BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, Keepalive,
-        ListedPeer, MAX_LISTED_HEADS,
+        BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof,
+        Keepalive, ListedPeer, MAX_LISTED_HEADS,
     };
     use automerge::ChangeHash;
 
@@ -447,7 +469,7 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&hello).unwrap()),
-            format!("040097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+            format!("050097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
         );
 
         let reply = ParsedMessage::HelloReply(HelloReply {
@@ -467,7 +489,7 @@ mod tests {
         assert_eq!(
             hex(&encode(&reply).unwrap()),
             format!(
-                "040199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
+                "050199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
                 run("04"),
                 run("06")
             )
@@ -482,7 +504,27 @@ mod tests {
         });
         assert_eq!(
             hex(&encode(&ping).unwrap()),
-            format!("0403959290c400{epoch_hex}01c3c2")
+            format!("0503959290c400{epoch_hex}01c3c2")
         );
+
+        let who = ParsedMessage::Who(IdentityProof {
+            key: [8; 32],
+            signature: [9; 64],
+        });
+        assert_eq!(
+            hex(&encode(&who).unwrap()),
+            format!("050692c420{}c440{}", "08".repeat(32), "09".repeat(64))
+        );
+        let ParsedMessage::Me(proof) = decode(
+            &encode(&ParsedMessage::Me(IdentityProof {
+                key: [8; 32],
+                signature: [9; 64],
+            }))
+            .unwrap(),
+        )
+        .unwrap() else {
+            panic!("not a ME");
+        };
+        assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64]));
     }
 }