introduce pre-shared swarm key

7 files changed, 316 insertions, 1 deletions

diff --git a/src/main.rs b/src/main.rs
index ef9bf54..96758ec 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -15,7 +15,7 @@ use anyhow::{Context, Result};
 use automerge::AutoCommit;
 use clap::Parser;
 use tokio::sync::{broadcast, mpsc, watch, RwLock};
-use tracing::{debug, error, info, trace};
+use tracing::{debug, error, info, trace, warn};
 use url::Url;
 
 mod api;
@@ -75,6 +75,11 @@ struct Args {
     #[arg(long, value_name = "PATH")]
     htpasswd: String,
 
+    /// Pre-shared swarm key: only peers that know it may join (raw bytes, at least 16;
+    /// e.g. `head -c 32 /dev/urandom`). Without it, anyone who knows the swarm ID can.
+    #[arg(long, value_name = "PATH")]
+    swarm_key_file: Option<PathBuf>,
+
     /// Keepalive interval; dormancy, resync and probe timings derive from it.
     #[arg(long, value_name = "SECONDS", default_value_t = 60)]
     keepalive_secs: u64,
@@ -91,6 +96,15 @@ async fn main() -> Result<()> {
 
     let args = Args::parse();
 
+    // Read before anything starts, so a bad key file fails right away.
+    let swarm_key = match &args.swarm_key_file {
+        Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)),
+        None => {
+            warn!("No swarm key: any peer that learns the swarm ID can join");
+            None
+        }
+    };
+
     // First thing: until a handler is installed, SIGINT and SIGTERM kill the process on
     // the spot, and startup (spawning Mycelium, bootstrap) can take minutes.
     let stop = watch_for_shutdown().context("Installing signal handlers")?;
@@ -163,6 +177,7 @@ async fn main() -> Result<()> {
         change_events,
         timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)),
         remake: Arc::new(remake::RemakeState::default()),
+        swarm_key,
         stale: Default::default(),
     };
 
diff --git a/src/state.rs b/src/state.rs
index 736f55b..49f75dc 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -102,6 +102,10 @@ pub(crate) struct PeerState {
     /// Consecutive sync exchanges that changed nothing (livelock detection).
     pub(crate) stalled_exchanges: u32,
 
+    /// The peer proved the swarm key in its HELLO or HELLO_REPLY (always false and
+    /// irrelevant without a swarm key).
+    pub(crate) authenticated: bool,
+
     /// The answer we expect to our last sync message(s), if any.
     pub(crate) sync_reply: Option<SyncReplyWait>,
     /// How long to wait for it; doubles while answers stay out.
@@ -141,6 +145,7 @@ impl Default for PeerState {
             last_sync_rx: None,
             last_resync_at: None,
             stalled_exchanges: 0,
+            authenticated: false,
             sync_reply: None,
             sync_reply_timeout: SYNC_REPLY_TIMEOUT,
             epoch: None,
@@ -364,6 +369,10 @@ pub(crate) struct AppState {
     /// Per-node remake state (prebuilt snapshots, switch status, archive).
     pub(crate) remake: Arc<RemakeState>,
 
+    /// Pre-shared key every member must prove in the handshake; None admits anyone who
+    /// knows the swarm ID.
+    pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>,
+
     /// While rejoining a newer epoch, the previous document, served read-only (and
     /// marked stale) instead of answering 503.
     pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>,
diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs
new file mode 100644
index 0000000..8b1352c
--- /dev/null
+++ b/src/swarm/auth.rs
@@ -0,0 +1,193 @@
+//! Swarm membership by pre-shared key.
+//!
+//! Mycelium authenticates the sender of every message (`src_pk` can't be forged), so we
+//! always know *who* talks to us. With a swarm key configured, only peers that also know
+//! the key may take part: HELLO and HELLO_REPLY carry a tag proving it, and until a peer
+//! has sent a valid one, everything else it sends is ignored.
+//!
+//! The tag is `HMAC-SHA256(key, label ‖ swarm_id ‖ from_pk ‖ to_pk)`. It needs no
+//! challenge: it is bound to both keys, and only `from_pk` can send as `from_pk`, so a
+//! captured tag is useless to anyone else. Different labels for HELLO and HELLO_REPLY
+//! keep one from being passed off as the other. A HELLO has no swarm ID yet (a new node
+//! doesn't know it), so its tag binds an empty one.
+//!
+//! Limits of a shared key: every member holds the same secret, so a single member can't
+//! be revoked (forgetting it lasts until it says HELLO again), and changing the key
+//! means restarting every node. A member list in the document is the proper successor.
+
+use std::path::Path;
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+use anyhow::{bail, Context, Result};
+use sha2::{Digest, Sha256};
+
+const HELLO_LABEL: &[u8] = b"continuum/hello/1";
+const HELLO_REPLY_LABEL: &[u8] = b"continuum/hello-reply/1";
+
+/// The swarm's pre-shared key.
+pub(crate) struct SwarmKey {
+    key: Vec<u8>,
+    rejections: Mutex<Rejections>,
+}
+
+/// For rate limiting the log of rejected handshakes.
+#[derive(Default)]
+struct Rejections {
+    last_logged: Option<Instant>,
+    suppressed: u64,
+}
+
+/// At most one rejection log line per this interval.
+const REJECTION_LOG_INTERVAL: Duration = Duration::from_secs(10);
+
+impl SwarmKey {
+    /// Reads the key from a file; its bytes are the key, as they are. Generate one with
+    /// e.g. `head -c 32 /dev/urandom > swarm.key`.
+    pub(crate) fn from_file(path: &Path) -> Result<Self> {
+        let key = std::fs::read(path)
+            .with_context(|| format!("Reading the swarm key from {}", path.display()))?;
+        if key.len() < 16 {
+            bail!(
+                "The swarm key in {} has {} bytes; use at least 16 (32 random bytes recommended)",
+                path.display(),
+                key.len()
+            );
+        }
+        Ok(Self::new(key))
+    }
+
+    fn new(key: Vec<u8>) -> Self {
+        Self {
+            key,
+            rejections: Mutex::default(),
+        }
+    }
+
+    /// Records a rejected handshake. Returns how many went unlogged since the last log
+    /// line if this one should be logged, None if it should be suppressed.
+    pub(crate) fn note_rejection(&self) -> Option<u64> {
+        let mut rejections = self
+            .rejections
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let due = rejections
+            .last_logged
+            .is_none_or(|last| last.elapsed() >= REJECTION_LOG_INTERVAL);
+        if !due {
+            rejections.suppressed += 1;
+            return None;
+        }
+        rejections.last_logged = Some(Instant::now());
+        Some(std::mem::take(&mut rejections.suppressed))
+    }
+
+    /// Tag for a HELLO from `from` to `to`.
+    pub(crate) fn hello_tag(&self, from: &str, to: &str) -> Vec<u8> {
+        self.tag(HELLO_LABEL, "", from, to).to_vec()
+    }
+
+    /// Tag for a HELLO_REPLY from `from` to `to`, naming the swarm.
+    pub(crate) fn hello_reply_tag(&self, swarm_id: &str, from: &str, to: &str) -> Vec<u8> {
+        self.tag(HELLO_REPLY_LABEL, swarm_id, from, to).to_vec()
+    }
+
+    pub(crate) fn verify_hello(&self, tag: &[u8], from: &str, to: &str) -> bool {
+        constant_time_eq(tag, &self.tag(HELLO_LABEL, "", from, to))
+    }
+
+    pub(crate) fn verify_hello_reply(
+        &self,
+        tag: &[u8],
+        swarm_id: &str,
+        from: &str,
+        to: &str,
+    ) -> bool {
+        constant_time_eq(tag, &self.tag(HELLO_REPLY_LABEL, swarm_id, from, to))
+    }
+
+    fn tag(&self, label: &[u8], swarm_id: &str, from: &str, to: &str) -> [u8; 32] {
+        // Length-prefixed fields, so no two different inputs encode the same.
+        let mut message = Vec::new();
+        for field in [label, swarm_id.as_bytes(), from.as_bytes(), to.as_bytes()] {
+            message.extend_from_slice(&(field.len() as u32).to_be_bytes());
+            message.extend_from_slice(field);
+        }
+        hmac_sha256(&self.key, &message)
+    }
+}
+
+/// HMAC-SHA256 (RFC 2104), on top of the `sha2` crate we already use.
+fn hmac_sha256(key: &[u8], message: &[u8]) -> [u8; 32] {
+    const BLOCK: usize = 64;
+
+    let mut block_key = [0u8; BLOCK];
+    if key.len() > BLOCK {
+        block_key[..32].copy_from_slice(&Sha256::digest(key));
+    } else {
+        block_key[..key.len()].copy_from_slice(key);
+    }
+
+    let mut inner = Sha256::new();
+    inner.update(block_key.map(|byte| byte ^ 0x36));
+    inner.update(message);
+    let inner = inner.finalize();
+
+    let mut outer = Sha256::new();
+    outer.update(block_key.map(|byte| byte ^ 0x5c));
+    outer.update(inner);
+
+    let mut tag = [0u8; 32];
+    tag.copy_from_slice(&outer.finalize());
+    tag
+}
+
+fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
+    a.len() == b.len() && a.iter().zip(b).fold(0u8, |diff, (x, y)| diff | (x ^ y)) == 0
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn hex(bytes: &[u8]) -> String {
+        bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+    }
+
+    #[test]
+    fn hmac_matches_rfc_4231() {
+        // Test case 2.
+        assert_eq!(
+            hex(&hmac_sha256(b"Jefe", b"what do ya want for nothing?")),
+            "5bdcc146bf60754e6a042426089575c75a003f089d2739839dec58b964ec3843"
+        );
+        // Test case 6: a key longer than the block size.
+        assert_eq!(
+            hex(&hmac_sha256(
+                &[0xaa; 131],
+                b"Test Using Larger Than Block-Size Key - Hash Key First"
+            )),
+            "60e431591ee0b67f0d8a26aacbf5b77f8e0bc6213728c5140546040f0ee37f54"
+        );
+    }
+
+    #[test]
+    fn tags_bind_key_peers_and_kind() {
+        let key = SwarmKey::new(vec![1; 32]);
+        let other = SwarmKey::new(vec![2; 32]);
+
+        let hello = key.hello_tag("A", "B");
+        assert!(key.verify_hello(&hello, "A", "B"));
+        assert!(!other.verify_hello(&hello, "A", "B"));
+        assert!(!key.verify_hello(&hello, "B", "A"));
+        assert!(!key.verify_hello(&hello, "A", "C"));
+        assert!(!key.verify_hello(&[], "A", "B"));
+
+        let reply = key.hello_reply_tag("swarm", "A", "B");
+        assert!(key.verify_hello_reply(&reply, "swarm", "A", "B"));
+        assert!(!key.verify_hello_reply(&reply, "other-swarm", "A", "B"));
+        // A HELLO tag is no HELLO_REPLY tag, and the other way round.
+        assert!(!key.verify_hello_reply(&hello, "", "A", "B"));
+        assert!(!key.verify_hello(&key.hello_reply_tag("", "A", "B"), "A", "B"));
+    }
+}
diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs
index 7316146..a4277c7 100644
--- a/src/swarm/mod.rs
+++ b/src/swarm/mod.rs
@@ -1,6 +1,7 @@
 //! The swarm of peers: bootstrap, handshake, keepalive and dormancy (`peer`), and
 //! Automerge synchronization with every member (`sync`).
 
+pub(crate) mod auth;
 pub(crate) mod bootstrap;
 pub(crate) mod peer;
 pub(crate) mod sync;
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index 53a4731..d801b0f 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -237,6 +237,11 @@ async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result
             peer,
             &ParsedMessage::Hello(transport::HelloRequest {
                 epoch: state.local_epoch(),
+                auth: state
+                    .swarm_key
+                    .as_ref()
+                    .map(|key| key.hello_tag(state.local_id(), peer))
+                    .unwrap_or_default(),
             }),
         )
         .await;
@@ -321,10 +326,16 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> {
         }
     }
 
+    let auth = state
+        .swarm_key
+        .as_ref()
+        .map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer))
+        .unwrap_or_default();
     let reply = transport::HelloReply {
         swarm_id,
         peers,
         epoch: state.local_epoch(),
+        auth,
     };
 
     let ready = state.is_ready().await;
@@ -563,6 +574,76 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
     relation == EpochRelation::Same
 }
 
+// --- swarm key -----------------------------------------------------------------
+
+/// With a swarm key, only peers that proved it get past this point. A HELLO or
+/// HELLO_REPLY must carry a valid tag, and it admits its sender; anything else is only
+/// accepted from a peer admitted that way. Everything else is ignored without a reply,
+/// so strangers learn nothing, not even whether we know the key.
+async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool {
+    let Some(key) = &state.swarm_key else {
+        return true;
+    };
+    let local = state.local_id();
+
+    let proof = match message {
+        ParsedMessage::Hello(hello) => Some(key.verify_hello(&hello.auth, peer, local)),
+        ParsedMessage::HelloReply(reply) => {
+            Some(key.verify_hello_reply(&reply.auth, &reply.swarm_id, peer, local))
+        }
+        _ => None,
+    };
+
+    match proof {
+        Some(true) => {
+            remember_peer(state, peer).await;
+            if let Some(peer_state) = state.get_peer_state(peer).await {
+                let mut peer_state = peer_state.lock().await;
+                if !peer_state.authenticated {
+                    peer_state.authenticated = true;
+                    debug!(peer = %peer, "Peer proved the swarm key");
+                }
+            }
+            true
+        }
+        Some(false) => {
+            // A known peer with a wrong key (e.g. one that was given another key) loses
+            // its admission until it proves the right one, and with it its membership:
+            // we stop syncing our document to it, not just listening to it.
+            if let Some(peer_state) = state.get_peer_state(peer).await {
+                let mut peer_state = peer_state.lock().await;
+                if peer_state.authenticated {
+                    warn!(peer = %peer, "Peer no longer proves the swarm key; dropping its membership");
+                }
+                peer_state.authenticated = false;
+                peer_state.membership = PeerMembership::Unknown;
+                peer_state.hello_state = HelloState::NotSent;
+                peer_state.reset_sync();
+            }
+            // The one place an operator can see why a node can't join: rate limited,
+            // since anyone can send us handshakes.
+            if let Some(suppressed) = key.note_rejection() {
+                info!(
+                    peer = %peer,
+                    suppressed,
+                    "Rejected handshake without a valid swarm key proof"
+                );
+            }
+            false
+        }
+        None => {
+            let admitted = match state.get_peer_state(peer).await {
+                Some(peer_state) => peer_state.lock().await.authenticated,
+                None => false,
+            };
+            if !admitted {
+                trace!(peer = %peer, "Ignoring message from a peer that hasn't proven the swarm key");
+            }
+            admitted
+        }
+    }
+}
+
 // --- inbound dispatch --------------------------------------------------------
 
 pub(crate) async fn handle_message(
@@ -575,6 +656,10 @@ pub(crate) async fn handle_message(
         return Ok(());
     }
 
+    if !admit(state, &peer, &message).await {
+        return Ok(());
+    }
+
     remember_peer(state, &peer).await;
 
     // Any inbound message proves liveness.
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 0afb5ee..58db611 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -43,6 +43,8 @@ impl ParsedMessage {
 #[derive(Debug, Clone)]
 pub struct HelloRequest {
     pub epoch: Epoch,
+    /// Proof of the swarm key (`swarm::auth`); empty without one.
+    pub auth: Vec<u8>,
 }
 
 #[derive(Debug, Clone)]
@@ -50,6 +52,8 @@ pub struct HelloReply {
     pub swarm_id: String,
     pub peers: Vec<String>,
     pub epoch: Epoch,
+    /// Proof of the swarm key (`swarm::auth`); empty without one.
+    pub auth: Vec<u8>,
 }
 
 #[derive(Debug, Clone)]
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 194f998..057916f 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -851,6 +851,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
 pub struct HelloRequestShadow {
     #[serde(default)]
     epoch: EpochShadow,
+    #[serde(default)]
+    auth: ByteBuf,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -861,6 +863,8 @@ pub struct HelloReplyShadow<'a> {
     pub peers: Vec<String>,
     #[serde(default)]
     epoch: EpochShadow,
+    #[serde(default)]
+    auth: ByteBuf,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -905,12 +909,14 @@ mod automerge_msg_codec {
 pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&HelloRequestShadow {
         epoch: epoch_to_shadow(&msg.epoch),
+        auth: ByteBuf::from(msg.auth.clone()),
     })
 }
 pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
     let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
     Ok(HelloRequest {
         epoch: epoch_from_shadow(shadow.epoch)?,
+        auth: shadow.auth.into_vec(),
     })
 }
 
@@ -920,6 +926,7 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
         swarm_id: Cow::Borrowed(&msg.swarm_id),
         peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
         epoch: epoch_to_shadow(&msg.epoch),
+        auth: ByteBuf::from(msg.auth.clone()),
     };
     rmp_serde::to_vec(&shadow)
 }
@@ -929,6 +936,7 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError>
         swarm_id: shadow.swarm_id.into_owned(),
         peers: shadow.peers,
         epoch: epoch_from_shadow(shadow.epoch)?,
+        auth: shadow.auth.into_vec(),
     })
 }