feat(transport): encrypt every frame between coordinator and volunteers (F3)

README.md

@@ -36,6 +36,7 @@ Volunteers lend spare computer time to train one open model. Design: [docs/desig
 - **What leaves the volunteer's machine**: only the signed, compressed update (the largest 5% of the parameter changes, each stored as an 8-bit number), its digest, the ids of the run and work unit it answers, the volunteer's public key and a signature. No training text and no personal data. A test checks the bytes the volunteer sends and finds none of its shard in them.
 - **Turing scoring**: each update is scored as held-out bits before minus held-out bits after applying it, measured in Turings. An update scoring zero or less is rejected. The coordinator sends a signed receipt and records accepted Turings in an append-only ledger against the volunteer's public key.
 - **The two programs**: `oh-coordinator` (hands out work, scores, aggregates, keeps the ledger; `--baseline` trains the same work in one process for comparison) and `oh-volunteer` (takes work units, trains locally, prints "Round N: update accepted, earned X Turings" or "rejected"). Volunteer keys are generated in memory and never saved.
+- **Encrypted in transit**: every connection between the two is encrypted (X25519 and ChaCha20-Poly1305 from `beacon-core`), with fresh keys per connection and per direction. Tampered, replayed or reordered messages are refused. The volunteer checks that it is talking to the right coordinator: pass the key the coordinator prints at start with `--coordinator-key <hex>`. Without it the volunteer refuses to connect, unless `--trust-first-use` is given, which accepts the first key and prints a warning. A test that records everything on the wire finds no training text and no teaching text in it.
 - **Demo**: `scripts/demo-local.sh` runs the coordinator and three volunteers on one machine for several seeds and compares them with the single-process baseline. Measured results: [docs/research/2026-10-16-v0-loopback-results.md](docs/research/2026-10-16-v0-loopback-results.md). In short, the loop works, and the volunteer version currently ends 0.33 to 0.36 bits per byte behind the single-process baseline.
 
 Not built in version 0 (from the design doc):
@@ -47,7 +48,8 @@ Not built in version 0 (from the design doc):
 - The DiLoCo outer optimizer. Version 0 combines accepted updates with a plain average.
 - Sybil resistance. One person can run many keys, and could split one update across them.
 - Protection against held-out leakage from repeated scoring. A volunteer who submits many updates and watches the scores could slowly learn about the held-out set.
-- Transport encryption. Frames travel in the clear over TCP, so version 0 is for loopback only. The design doc lists the remaining gaps (coordinator key pinning, stalled volunteers, dropped-update records).
+- A lasting coordinator identity. The coordinator makes a new key each time it starts, so volunteers need that run's key.
+- Hiding message sizes and timing from someone watching the network. The design doc lists the other remaining gaps (stalled volunteers and handshakes, dropped-update records).
 
 **Teaching mode.** A volunteer who runs an open language model on their own machine can lend that model instead of training: `oh-volunteer teach --coordinator <addr>` asks the local model (by default the chat server at 127.0.0.1:18091; it refuses any teacher that is not on the same machine) to write a short passage on a hint taken from the work unit, runs the passage through the Personal Data Firewall, and sends only the firewalled text, signed. Text that looks like it holds a password, key or token is not sent at all. The coordinator scores the text in Turings the same way as an update (train a copy of the model on it and measure held-out bits) and, if it helps, trains the shared model on it after the round's update step. Design: [docs/design/qwen-teacher-v0.md](docs/design/qwen-teacher-v0.md). First run against a local Qwen3-30B: [docs/research/2026-10-16-qwen-teacher-v0-results.md](docs/research/2026-10-16-qwen-teacher-v0-results.md). Demo: `scripts/demo-teach.sh`.
 

crates/beacon-core/src/crypto.rs

@@ -4,14 +4,14 @@
 //! end-to-end authenticated payload encryption.
 
 use crate::schema::BeaconError;
-use chacha20poly1305::aead::{Aead, KeyInit};
+use chacha20poly1305::aead::{Aead, KeyInit, Payload};
 use chacha20poly1305::{ChaCha20Poly1305, Key, Nonce};
 use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
 use rand::rngs::OsRng;
 use rand::RngCore;
 use std::collections::HashSet;
 use uuid::Uuid;
-use x25519_dalek::{PublicKey as X25519PublicKey, StaticSecret};
+use x25519_dalek::{EphemeralSecret, PublicKey as X25519PublicKey, StaticSecret};
 
 pub const MAX_TIMESTAMP_DRIFT_SECS: u64 = 300;
 
@@ -104,6 +104,85 @@ pub fn decrypt_from_peer(
         .map_err(|_| BeaconError::DecryptionFailed)
 }
 
+/// One side of an X25519 key agreement with a one-time secret, for setting
+/// up a session. The secret is used once (`agree` consumes it) and is wiped
+/// when dropped (x25519-dalek zeroizes it).
+pub struct EphemeralKeyAgreement {
+    secret: EphemeralSecret,
+    public: [u8; 32],
+}
+
+impl EphemeralKeyAgreement {
+    pub fn generate() -> Self {
+        let secret = EphemeralSecret::random_from_rng(OsRng);
+        let public = *X25519PublicKey::from(&secret).as_bytes();
+        Self { secret, public }
+    }
+
+    pub fn public_bytes(&self) -> [u8; 32] {
+        self.public
+    }
+
+    /// The X25519 shared secret with `peer_public`. Refuses a peer key that
+    /// gives an all-zero (non-contributory) result, such as a low-order point.
+    pub fn agree(self, peer_public: &[u8; 32]) -> Result<[u8; 32], BeaconError> {
+        let shared = self
+            .secret
+            .diffie_hellman(&X25519PublicKey::from(*peer_public));
+        if !shared.was_contributory() {
+            return Err(BeaconError::KeyAgreementFailed);
+        }
+        Ok(*shared.as_bytes())
+    }
+}
+
+/// ChaCha20-Poly1305 under one fixed key, with the nonce chosen by the
+/// caller. The caller must never use the same nonce twice under one key.
+pub struct SessionCipher(ChaCha20Poly1305);
+
+impl SessionCipher {
+    pub fn new(key: &[u8; 32]) -> Self {
+        Self(ChaCha20Poly1305::new(Key::from_slice(key)))
+    }
+
+    /// Encrypts `plaintext` and binds `aad` (sent in the clear) to it.
+    pub fn seal(
+        &self,
+        nonce: &[u8; 12],
+        aad: &[u8],
+        plaintext: &[u8],
+    ) -> Result<Vec<u8>, BeaconError> {
+        self.0
+            .encrypt(
+                Nonce::from_slice(nonce),
+                Payload {
+                    msg: plaintext,
+                    aad,
+                },
+            )
+            .map_err(|_| BeaconError::Serialization("Encryption failed".into()))
+    }
+
+    /// Decrypts and checks `ciphertext` against `nonce` and `aad`. Any
+    /// change to the ciphertext, nonce or aad fails.
+    pub fn open(
+        &self,
+        nonce: &[u8; 12],
+        aad: &[u8],
+        ciphertext: &[u8],
+    ) -> Result<Vec<u8>, BeaconError> {
+        self.0
+            .decrypt(
+                Nonce::from_slice(nonce),
+                Payload {
+                    msg: ciphertext,
+                    aad,
+                },
+            )
+            .map_err(|_| BeaconError::DecryptionFailed)
+    }
+}
+
 pub fn verify_timestamp_drift(
     timestamp: u64,
     current_time: u64,
@@ -160,3 +239,38 @@ impl ReplayProtector {
         self.seen.is_empty()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ephemeral_agreement_matches_on_both_sides() {
+        let a = EphemeralKeyAgreement::generate();
+        let b = EphemeralKeyAgreement::generate();
+        let (a_pub, b_pub) = (a.public_bytes(), b.public_bytes());
+        assert_eq!(a.agree(&b_pub).unwrap(), b.agree(&a_pub).unwrap());
+    }
+
+    #[test]
+    fn test_ephemeral_agreement_refuses_low_order_point() {
+        let a = EphemeralKeyAgreement::generate();
+        assert!(matches!(
+            a.agree(&[0u8; 32]),
+            Err(BeaconError::KeyAgreementFailed)
+        ));
+    }
+
+    #[test]
+    fn test_session_cipher_round_trip_and_tamper() {
+        let c = SessionCipher::new(&[7u8; 32]);
+        let nonce = [1u8; 12];
+        let sealed = c.seal(&nonce, b"hdr", b"hello").unwrap();
+        assert_eq!(c.open(&nonce, b"hdr", &sealed).unwrap(), b"hello");
+        let mut bad = sealed.clone();
+        bad[0] ^= 1;
+        assert!(c.open(&nonce, b"hdr", &bad).is_err());
+        assert!(c.open(&[2u8; 12], b"hdr", &sealed).is_err());
+        assert!(c.open(&nonce, b"HDR", &sealed).is_err());
+    }
+}

crates/beacon-core/src/schema.rs

@@ -42,6 +42,8 @@ pub enum BeaconError {
     TimestampDrift { drift_secs: u64, max_allowed: u64 },
     #[error("Replay attack detected for packet ID: {0}")]
     ReplayDetected(Uuid),
+    #[error("Key agreement failed: the peer's public key gives no shared secret")]
+    KeyAgreementFailed,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]

crates/oh-coordinator/src/lib.rs

@@ -11,8 +11,8 @@
 
 use beacon_core::Keypair;
 use oh_core::{
-    apply, recv_submission, send, BaseModel, Learner, Ledger, LedgerError, Receipt, Scorer,
-    Submission, TeachingBatch, TinyByteModel, Update, WorkUnit,
+    apply, coordinator_handshake, recv_submission, send, BaseModel, Learner, Ledger, LedgerError,
+    Receipt, Scorer, Submission, TeachingBatch, TinyByteModel, Update, WorkUnit,
 };
 use std::io::{Read, Write};
 use std::net::TcpListener;
@@ -27,6 +27,12 @@ pub const EPSILON: f64 = 1e-3;
 /// Largest teaching batch the coordinator will train on, in bytes.
 pub const MAX_TEACHING_TEXT_BYTES: usize = 1024 * 1024;
 
+/// How long a new connection has to finish the transport handshake.
+pub const HANDSHAKE_TIMEOUT_SECS: u64 = 30;
+
+/// Read and write timeout on a connection once the round is under way.
+pub const ROUND_TIMEOUT_SECS: u64 = 600;
+
 #[derive(Debug, Clone)]
 pub struct Settings {
     pub volunteers_per_round: usize,
@@ -494,7 +500,10 @@ impl Coordinator {
     }
 
     /// Runs `rounds` rounds on `listener`, waiting each round for one
-    /// connection per shard, and passes each round's report to `log`.
+    /// connection per shard, and passes each round's report to `log`. Every
+    /// connection starts with the encrypted-transport handshake
+    /// (`oh_core::coordinator_handshake`, signed with this run's key); one
+    /// that fails it is closed and not counted.
     pub fn serve(
         &mut self,
         listener: &TcpListener,
@@ -504,15 +513,26 @@ impl Coordinator {
         for _ in 0..rounds {
             let mut conns = Vec::with_capacity(self.shards.len());
             while conns.len() < self.shards.len() {
-                let (stream, _) = listener.accept().map_err(|e| e.to_string())?;
-                let timeout = Some(Duration::from_secs(600));
-                stream
-                    .set_read_timeout(timeout)
-                    .map_err(|e| e.to_string())?;
-                stream
-                    .set_write_timeout(timeout)
-                    .map_err(|e| e.to_string())?;
-                conns.push(stream);
+                let (stream, peer) = listener.accept().map_err(|e| e.to_string())?;
+                let set_timeout = |secs| {
+                    let t = Some(Duration::from_secs(secs));
+                    stream
+                        .set_read_timeout(t)
+                        .and_then(|_| stream.set_write_timeout(t))
+                        .map_err(|e| e.to_string())
+                };
+                set_timeout(HANDSHAKE_TIMEOUT_SECS)?;
+                match coordinator_handshake(stream, &self.keypair) {
+                    Ok(conn) => {
+                        let s = conn.get_ref();
+                        let t = Some(Duration::from_secs(ROUND_TIMEOUT_SECS));
+                        s.set_read_timeout(t)
+                            .and_then(|_| s.set_write_timeout(t))
+                            .map_err(|e| e.to_string())?;
+                        conns.push(conn);
+                    }
+                    Err(e) => eprintln!("connection from {peer}: handshake failed: {e}"),
+                }
             }
             log(&self.run_round(&mut conns)?);
         }

crates/oh-coordinator/tests/encrypted_transport.rs

@@ -0,0 +1,154 @@
+//! An eavesdropper on the loopback link between `Coordinator::serve` and
+//! its volunteers sees no shard bytes and no teaching text.
+//!
+//! A relay sits between the volunteers and the coordinator and records every
+//! byte in both directions for one full round: one training volunteer and
+//! one teaching volunteer, both pinned to the coordinator key.
+
+use beacon_core::Keypair;
+use oh_coordinator::{Coordinator, RoundReport, Settings};
+use oh_core::{
+    send, send_teaching, volunteer_handshake, CoordinatorPin, Ledger, Teacher, TeacherError,
+    TeachingBatch, WorkUnit,
+};
+use oh_volunteer::{run_once, teach_once};
+use std::collections::HashSet;
+use std::io::{Read, Write};
+use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
+use std::sync::{Arc, Mutex};
+use std::thread;
+
+const LICENSE: &[u8] = include_bytes!("../../../LICENSE");
+const WINDOW: usize = 16;
+const TEACHING_TEXT: &str = "Every person who receives a copy of the program may read the \
+source, change it to suit their own needs, and pass the improved version on to their friends.";
+
+struct FakeTeacher;
+
+impl Teacher for FakeTeacher {
+    fn generate(&self, _prompt: &str, _max_tokens: u32) -> Result<String, TeacherError> {
+        Ok(TEACHING_TEXT.into())
+    }
+}
+
+/// Copies bytes from `from` to `to`, appending each chunk to `log`.
+fn pump(mut from: TcpStream, mut to: TcpStream, log: Arc<Mutex<Vec<u8>>>) {
+    let mut buf = [0u8; 16 * 1024];
+    loop {
+        match from.read(&mut buf) {
+            Ok(0) | Err(_) => break,
+            Ok(n) => {
+                log.lock().unwrap().extend_from_slice(&buf[..n]);
+                if to.write_all(&buf[..n]).is_err() {
+                    break;
+                }
+            }
+        }
+    }
+    let _ = to.shutdown(Shutdown::Write);
+}
+
+/// Relays `connections` connections from a new listener to `target` and
+/// records everything. Returns the relay address and the capture.
+fn eavesdropper(target: SocketAddr, connections: usize) -> (SocketAddr, Arc<Mutex<Vec<u8>>>) {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let addr = listener.local_addr().unwrap();
+    let capture: Arc<Mutex<Vec<u8>>> = Arc::default();
+    let log = capture.clone();
+    thread::spawn(move || {
+        for _ in 0..connections {
+            let (client, _) = listener.accept().unwrap();
+            let server = TcpStream::connect(target).unwrap();
+            let (c2, s2) = (client.try_clone().unwrap(), server.try_clone().unwrap());
+            let (l1, l2) = (log.clone(), log.clone());
+            thread::spawn(move || pump(client, server, l1));
+            thread::spawn(move || pump(s2, c2, l2));
+        }
+    });
+    (addr, capture)
+}
+
+/// True when any `WINDOW`-byte run of `needle` appears in `haystack`.
+fn leaks(haystack: &[u8], needle: &[u8]) -> bool {
+    let seen: HashSet<&[u8]> = haystack.windows(WINDOW).collect();
+    needle.windows(WINDOW).any(|w| seen.contains(w))
+}
+
+#[test]
+fn test_eavesdropper_sees_no_shard_or_teaching_bytes() {
+    let dir = std::env::temp_dir().join(format!("oh-encrypted-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    let settings = Settings {
+        volunteers_per_round: 2,
+        steps: 20,
+        heldout_fraction: 0.1,
+        seed: 3,
+        dim: 8,
+    };
+    let mut coordinator = Coordinator::new(
+        LICENSE,
+        &settings,
+        Ledger::open(&dir.join("ledger.jsonl")).unwrap(),
+    )
+    .unwrap();
+    let key = coordinator.pubkey();
+    let train = LICENSE[..coordinator.train_len()].to_vec();
+
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    let (relay, capture) = eavesdropper(listener.local_addr().unwrap(), 2);
+    let server = thread::spawn(move || {
+        let mut reports: Vec<RoundReport> = Vec::new();
+        coordinator
+            .serve(&listener, 1, |r| reports.push(r.clone()))
+            .unwrap();
+        reports
+    });
+
+    let vol_settings = move |seed| oh_volunteer::Settings {
+        seed,
+        keep_fraction: 0.05,
+        coordinator_key: Some(key),
+    };
+    let trainer = thread::spawn(move || {
+        let node = Keypair::generate();
+        let stream = TcpStream::connect(relay).unwrap();
+        let mut s = volunteer_handshake(stream, &node, CoordinatorPin::Key(key)).unwrap();
+        run_once(&mut s, &node, &vol_settings(1)).unwrap()
+    });
+    let node = Keypair::generate();
+    let stream = TcpStream::connect(relay).unwrap();
+    let mut s = volunteer_handshake(stream, &node, CoordinatorPin::Key(key)).unwrap();
+    let taught = teach_once(&mut s, &node, &vol_settings(2), &FakeTeacher, 64).unwrap();
+    drop(s);
+    trainer.join().unwrap();
+    let reports = server.join().unwrap();
+    assert_eq!(reports.len(), 1);
+    assert_eq!(reports[0].taught, 1, "{:?}", reports[0]);
+    assert!(taught.accepted);
+
+    let captured = capture.lock().unwrap().clone();
+    // Both shards (about 15 KB each) and the base model crossed the link.
+    assert!(
+        captured.len() > train.len(),
+        "capture {} bytes",
+        captured.len()
+    );
+
+    // Negative control: the same check finds both in plain frames,
+    // so a clean result below means the bytes really were hidden.
+    let unit = WorkUnit::new_signed(&node, [0; 16], 1, [0; 32], train[..64].to_vec(), 1);
+    let batch = TeachingBatch::new_signed(&node, &unit, TEACHING_TEXT.as_bytes().to_vec());
+    let mut plain = Vec::new();
+    send(&mut plain, &unit).unwrap();
+    send_teaching(&mut plain, &batch).unwrap();
+    assert!(leaks(&plain, TEACHING_TEXT.as_bytes()));
+    assert!(leaks(&plain, &train[..64]));
+
+    assert!(!leaks(&captured, &train), "shard bytes visible on the wire");
+    assert!(
+        !leaks(&captured, TEACHING_TEXT.as_bytes()),
+        "teaching text visible on the wire"
+    );
+    assert!(!captured.windows(8).any(|w| w == b"OHTEACH1"));
+}