diff --git a/README.md b/README.md index 817b0bf..abbc1fe 100644 --- a/README.md +++ b/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 `. 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 ` 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`. diff --git a/crates/beacon-core/src/crypto.rs b/crates/beacon-core/src/crypto.rs index a527bed..b314145 100644 --- a/crates/beacon-core/src/crypto.rs +++ b/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, 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, 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()); + } +} diff --git a/crates/beacon-core/src/schema.rs b/crates/beacon-core/src/schema.rs index 5405c4b..d6bc914 100644 --- a/crates/beacon-core/src/schema.rs +++ b/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)] diff --git a/crates/oh-coordinator/src/lib.rs b/crates/oh-coordinator/src/lib.rs index ae68986..f9b58ac 100644 --- a/crates/oh-coordinator/src/lib.rs +++ b/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)?); } diff --git a/crates/oh-coordinator/tests/encrypted_transport.rs b/crates/oh-coordinator/tests/encrypted_transport.rs new file mode 100644 index 0000000..8a039da --- /dev/null +++ b/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 { + 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>>) { + 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>>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let capture: Arc>> = 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 = 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")); +} diff --git a/crates/oh-coordinator/tests/loopback.rs b/crates/oh-coordinator/tests/loopback.rs index c20381d..f194b7c 100644 --- a/crates/oh-coordinator/tests/loopback.rs +++ b/crates/oh-coordinator/tests/loopback.rs @@ -2,7 +2,7 @@ use beacon_core::Keypair; use oh_coordinator::{Coordinator, RoundReport, Settings, Verdict}; -use oh_core::{read_frame, Ledger, Update, WorkUnit}; +use oh_core::{read_frame, volunteer_handshake, CoordinatorPin, Ledger, Update, WorkUnit}; use oh_volunteer::{run_once, Outcome}; use std::collections::HashSet; use std::io::{Cursor, Read, Write}; @@ -86,18 +86,30 @@ fn fresh_ledger(name: &str) -> PathBuf { } /// Starts one volunteer thread that takes `rounds` work units from `addr`. +/// With `pin`, each connection opens the encrypted transport pinned to that +/// coordinator key (as `Coordinator::serve` expects); without, it speaks +/// plain frames, for tests that drive `run_round` directly and read them. fn spawn_volunteer( addr: SocketAddr, rounds: u32, seed: u64, + pin: Option<[u8; 32]>, ) -> thread::JoinHandle<([u8; 32], Vec)> { thread::spawn(move || { let keypair = Keypair::generate(); let settings = volunteer_settings(seed); let outcomes = (0..rounds) .map(|_| { - let mut stream = TcpStream::connect(addr).unwrap(); - run_once(&mut stream, &keypair, &settings).unwrap() + let stream = TcpStream::connect(addr).unwrap(); + match pin { + Some(key) => { + let mut secure = + volunteer_handshake(stream, &keypair, CoordinatorPin::Key(key)) + .unwrap(); + run_once(&mut secure, &keypair, &settings).unwrap() + } + None => run_once(&mut { stream }, &keypair, &settings).unwrap(), + } }) .collect(); (keypair.pubkey_bytes(), outcomes) @@ -125,7 +137,9 @@ fn test_two_volunteers_two_rounds_over_loopback() { .unwrap(); reports }); - let volunteers: Vec<_> = (0..2).map(|i| spawn_volunteer(addr, 2, 10 + i)).collect(); + let volunteers: Vec<_> = (0..2) + .map(|i| spawn_volunteer(addr, 2, 10 + i, Some(key))) + .collect(); let results: Vec<_> = volunteers.into_iter().map(|v| v.join().unwrap()).collect(); let reports = server.join().unwrap(); @@ -224,7 +238,9 @@ fn test_no_heldout_bytes_in_any_frame_sent() { coordinator.run_round(&mut conns).unwrap(); } }); - let volunteers: Vec<_> = (0..2).map(|i| spawn_volunteer(addr, 2, 20 + i)).collect(); + let volunteers: Vec<_> = (0..2) + .map(|i| spawn_volunteer(addr, 2, 20 + i, None)) + .collect(); for v in volunteers { v.join().unwrap(); } diff --git a/crates/oh-coordinator/tests/teaching.rs b/crates/oh-coordinator/tests/teaching.rs index 3060c2a..ca6b8b0 100644 --- a/crates/oh-coordinator/tests/teaching.rs +++ b/crates/oh-coordinator/tests/teaching.rs @@ -3,9 +3,12 @@ use beacon_core::Keypair; use oh_coordinator::{Coordinator, RoundReport, Settings, TeachingVerdict}; -use oh_core::{Ledger, Teacher, TeacherError, TeachingBatch}; +use oh_core::{ + volunteer_handshake, CoordinatorPin, EncryptedStream, Ledger, Teacher, TeacherError, + TeachingBatch, +}; use oh_volunteer::{run_once, teach_once}; -use std::net::{TcpListener, TcpStream}; +use std::net::{SocketAddr, TcpListener, TcpStream}; use std::path::PathBuf; use std::thread; @@ -96,6 +99,17 @@ fn good_text(seed: u64) -> String { String::from_utf8(corpus(1500, seed)).unwrap() } +/// Connects to a coordinator started with `serve` and opens the encrypted +/// transport, pinned to `key`. +fn secure(addr: SocketAddr, node: &Keypair, key: [u8; 32]) -> EncryptedStream { + volunteer_handshake( + TcpStream::connect(addr).unwrap(), + node, + CoordinatorPin::Key(key), + ) + .unwrap() +} + #[test] fn test_teacher_and_trainer_share_a_round_over_loopback() { let ledger_path = fresh_ledger("shared-round"); @@ -121,7 +135,7 @@ fn test_teacher_and_trainer_share_a_round_over_loopback() { let trainer = thread::spawn(move || { let keypair = Keypair::generate(); for _ in 0..2 { - let mut stream = TcpStream::connect(addr).unwrap(); + let mut stream = secure(addr, &keypair, key); run_once(&mut stream, &keypair, &volunteer_settings(11)).unwrap(); } }); @@ -129,7 +143,7 @@ fn test_teacher_and_trainer_share_a_round_over_loopback() { let mut outcomes = Vec::new(); for round in 0..2u64 { let teacher = FakeTeacher(good_text(100 + round)); - let mut stream = TcpStream::connect(addr).unwrap(); + let mut stream = secure(addr, &teacher_node, key); outcomes.push( teach_once( &mut stream, diff --git a/crates/oh-core/src/lib.rs b/crates/oh-core/src/lib.rs index 852b330..0937bf6 100644 --- a/crates/oh-core/src/lib.rs +++ b/crates/oh-core/src/lib.rs @@ -8,11 +8,13 @@ pub mod learner; pub mod scoring; pub mod teacher; +pub mod transport; pub mod update; pub mod wire; pub use learner::*; pub use scoring::*; pub use teacher::*; +pub use transport::*; pub use update::*; pub use wire::*; diff --git a/crates/oh-core/src/transport.rs b/crates/oh-core/src/transport.rs new file mode 100644 index 0000000..bba00c2 --- /dev/null +++ b/crates/oh-core/src/transport.rs @@ -0,0 +1,509 @@ +//! Encrypted transport between the coordinator and a volunteer. +//! +//! Handshake (one message each way, volunteer first), each hello is +//! `version[8] || identity[32] || ephemeral[32] || sig[64]`: +//! +//! 1. Volunteer: a fresh X25519 key, signed with its Ed25519 identity over +//! `VOLUNTEER_CONTEXT || identity || ephemeral`. +//! 2. Coordinator: a fresh X25519 key, signed with its Ed25519 identity over +//! `COORDINATOR_CONTEXT || volunteer hello || identity || ephemeral`, so +//! the signature is bound to this session and cannot be replayed into +//! another. +//! +//! The volunteer checks the coordinator identity against a pinned key (or +//! accepts it on first use when the caller allows that). Both sides derive +//! one key per direction with BLAKE3 `derive_key` from the X25519 shared +//! secret and a hash of both hellos. +//! +//! After the handshake every byte goes in sealed records: +//! `len[4, big endian] || ChaCha20-Poly1305(plaintext)`, with the length +//! header as associated data. The nonce is the record number in this +//! direction (4 zero bytes then the 64-bit little-endian counter); it is +//! never sent, never reused, and the receiver only accepts the next number, +//! so a replayed, dropped or reordered record fails to open and closes the +//! stream. All primitives come from beacon-core (x25519-dalek, +//! chacha20poly1305) and blake3; nothing here is new cryptography. + +use crate::wire::MAX_FRAME_BYTES; +use beacon_core::{verify_signature, EphemeralKeyAgreement, Keypair, SessionCipher}; +use std::io::{self, Read, Write}; + +/// Protocol tag at the start of each hello. +pub const TRANSPORT_VERSION: &[u8; 8] = b"OHSEC001"; +const VOLUNTEER_CONTEXT: &[u8] = b"open-humanity transport v1: volunteer hello"; +const COORDINATOR_CONTEXT: &[u8] = b"open-humanity transport v1: coordinator hello"; +const KEY_V2C: &str = "open-humanity transport v1 2026-10-16 volunteer to coordinator"; +const KEY_C2V: &str = "open-humanity transport v1 2026-10-16 coordinator to volunteer"; +const HELLO_LEN: usize = 8 + 32 + 32 + 64; +const TAG_LEN: usize = 16; +/// Largest plaintext in one record: one full wire frame with its header. +pub const MAX_RECORD_PLAINTEXT: usize = MAX_FRAME_BYTES as usize + 4; + +#[derive(Debug, thiserror::Error)] +pub enum TransportError { + #[error("transport i/o error: {0}")] + Io(#[from] io::Error), + #[error("peer does not speak this transport version")] + BadVersion, + #[error("peer's hello signature does not verify")] + BadSignature, + #[error("key agreement failed")] + KeyAgreement, + #[error("coordinator key {got} is not the pinned key {expected}")] + WrongCoordinator { expected: String, got: String }, +} + +/// Which coordinator identity a volunteer accepts. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CoordinatorPin { + /// Only this Ed25519 public key. + Key([u8; 32]), + /// Whatever key the coordinator presents. An attacker in the middle of + /// the first connection can present their own; the caller should warn. + TrustFirstUse, +} + +struct Hello { + bytes: [u8; HELLO_LEN], +} + +impl Hello { + fn new(identity: &Keypair, ephemeral: &[u8; 32], signed_prefix: &[u8]) -> Self { + let mut bytes = [0u8; HELLO_LEN]; + bytes[..8].copy_from_slice(TRANSPORT_VERSION); + bytes[8..40].copy_from_slice(&identity.pubkey_bytes()); + bytes[40..72].copy_from_slice(ephemeral); + let mut msg = signed_prefix.to_vec(); + msg.extend_from_slice(&bytes[8..72]); + bytes[72..].copy_from_slice(&identity.sign(&msg)); + Self { bytes } + } + + fn read(r: &mut R) -> Result { + let mut bytes = [0u8; HELLO_LEN]; + r.read_exact(&mut bytes)?; + if &bytes[..8] != TRANSPORT_VERSION { + return Err(TransportError::BadVersion); + } + Ok(Self { bytes }) + } + + fn identity(&self) -> [u8; 32] { + self.bytes[8..40].try_into().expect("32 bytes") + } + + fn ephemeral(&self) -> [u8; 32] { + self.bytes[40..72].try_into().expect("32 bytes") + } + + fn verify(&self, signed_prefix: &[u8]) -> Result<(), TransportError> { + let mut msg = signed_prefix.to_vec(); + msg.extend_from_slice(&self.bytes[8..72]); + let sig: [u8; 64] = self.bytes[72..].try_into().expect("64 bytes"); + verify_signature(&self.identity(), &msg, &sig).map_err(|_| TransportError::BadSignature) + } +} + +fn coordinator_prefix(volunteer_hello: &Hello) -> Vec { + let mut p = COORDINATOR_CONTEXT.to_vec(); + p.extend_from_slice(&volunteer_hello.bytes); + p +} + +/// (volunteer to coordinator key, coordinator to volunteer key). +fn session_keys(shared: &[u8; 32], v: &Hello, c: &Hello) -> ([u8; 32], [u8; 32]) { + let mut transcript = blake3::Hasher::new(); + transcript.update(&v.bytes); + transcript.update(&c.bytes); + let mut material = shared.to_vec(); + material.extend_from_slice(transcript.finalize().as_bytes()); + ( + blake3::derive_key(KEY_V2C, &material), + blake3::derive_key(KEY_C2V, &material), + ) +} + +/// Volunteer side of the handshake. Sends nothing after the hello unless +/// the coordinator's signature verifies and its key passes `pin`. +pub fn volunteer_handshake( + mut stream: S, + identity: &Keypair, + pin: CoordinatorPin, +) -> Result, TransportError> { + let eph = EphemeralKeyAgreement::generate(); + let mine = Hello::new(identity, &eph.public_bytes(), VOLUNTEER_CONTEXT); + stream.write_all(&mine.bytes)?; + stream.flush()?; + let theirs = Hello::read(&mut stream)?; + theirs.verify(&coordinator_prefix(&mine))?; + if let CoordinatorPin::Key(expected) = pin { + if theirs.identity() != expected { + return Err(TransportError::WrongCoordinator { + expected: hex(&expected), + got: hex(&theirs.identity()), + }); + } + } + let shared = eph + .agree(&theirs.ephemeral()) + .map_err(|_| TransportError::KeyAgreement)?; + let (v2c, c2v) = session_keys(&shared, &mine, &theirs); + Ok(EncryptedStream::new(stream, &v2c, &c2v, theirs.identity())) +} + +/// Coordinator side of the handshake. +pub fn coordinator_handshake( + mut stream: S, + identity: &Keypair, +) -> Result, TransportError> { + let theirs = Hello::read(&mut stream)?; + theirs.verify(VOLUNTEER_CONTEXT)?; + let eph = EphemeralKeyAgreement::generate(); + let mine = Hello::new(identity, &eph.public_bytes(), &coordinator_prefix(&theirs)); + stream.write_all(&mine.bytes)?; + stream.flush()?; + let shared = eph + .agree(&theirs.ephemeral()) + .map_err(|_| TransportError::KeyAgreement)?; + let (v2c, c2v) = session_keys(&shared, &theirs, &mine); + Ok(EncryptedStream::new(stream, &c2v, &v2c, theirs.identity())) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +fn nonce(counter: u64) -> [u8; 12] { + let mut n = [0u8; 12]; + n[4..].copy_from_slice(&counter.to_le_bytes()); + n +} + +/// A stream whose bytes travel in sealed records. Writes are buffered and +/// sealed on `flush`; data written but not flushed is not sent. +pub struct EncryptedStream { + inner: S, + send: SessionCipher, + recv: SessionCipher, + send_counter: u64, + recv_counter: u64, + write_buf: Vec, + read_buf: Vec, + read_pos: usize, + peer_identity: [u8; 32], + /// Set after a record fails to open; every later read fails. + broken: bool, +} + +impl EncryptedStream { + fn new(inner: S, send_key: &[u8; 32], recv_key: &[u8; 32], peer: [u8; 32]) -> Self { + Self { + inner, + send: SessionCipher::new(send_key), + recv: SessionCipher::new(recv_key), + send_counter: 0, + recv_counter: 0, + write_buf: Vec::new(), + read_buf: Vec::new(), + read_pos: 0, + peer_identity: peer, + broken: false, + } + } + + /// The Ed25519 identity the peer proved in the handshake. + pub fn peer_identity(&self) -> [u8; 32] { + self.peer_identity + } + + pub fn get_ref(&self) -> &S { + &self.inner + } + + /// The raw stream underneath. Bytes written or read here bypass the + /// encryption; meant for tests and socket settings. + pub fn get_mut(&mut self) -> &mut S { + &mut self.inner + } +} + +fn invalid(msg: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, msg) +} + +impl EncryptedStream { + /// Reads and opens the next record. `Ok(false)` on a clean end of stream + /// between records. + fn fill(&mut self) -> io::Result { + let mut header = [0u8; 4]; + let mut got = 0; + while got < 4 { + match self.inner.read(&mut header[got..]) { + Ok(0) if got == 0 => return Ok(false), + Ok(0) => return Err(io::ErrorKind::UnexpectedEof.into()), + Ok(n) => got += n, + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + let len = u32::from_be_bytes(header) as usize; + if !(TAG_LEN..=MAX_RECORD_PLAINTEXT + TAG_LEN).contains(&len) { + self.broken = true; + return Err(invalid("encrypted record has an impossible length")); + } + let mut sealed = vec![0u8; len]; + self.inner.read_exact(&mut sealed)?; + match self.recv.open(&nonce(self.recv_counter), &header, &sealed) { + Ok(plain) => { + self.recv_counter += 1; + self.read_buf = plain; + self.read_pos = 0; + Ok(true) + } + Err(_) => { + self.broken = true; + Err(invalid( + "encrypted record failed to open: tampered, replayed or out of order", + )) + } + } + } +} + +impl Read for EncryptedStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + while self.read_pos == self.read_buf.len() { + if self.broken { + return Err(invalid("encrypted stream is closed after a bad record")); + } + if !self.fill()? { + return Ok(0); + } + } + let n = buf.len().min(self.read_buf.len() - self.read_pos); + buf[..n].copy_from_slice(&self.read_buf[self.read_pos..self.read_pos + n]); + self.read_pos += n; + Ok(n) + } +} + +impl Write for EncryptedStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.write_buf.extend_from_slice(buf); + Ok(buf.len()) + } + + /// Seals everything written since the last flush, in records of at most + /// `MAX_RECORD_PLAINTEXT` bytes, and sends them. + fn flush(&mut self) -> io::Result<()> { + let pending = std::mem::take(&mut self.write_buf); + for chunk in pending.chunks(MAX_RECORD_PLAINTEXT) { + if self.send_counter == u64::MAX { + return Err(io::Error::other("record counter exhausted")); + } + let len = (chunk.len() + TAG_LEN) as u32; + let header = len.to_be_bytes(); + let sealed = self + .send + .seal(&nonce(self.send_counter), &header, chunk) + .map_err(|_| io::Error::other("encryption failed"))?; + self.send_counter += 1; + self.inner.write_all(&header)?; + self.inner.write_all(&sealed)?; + } + self.inner.flush() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{TcpListener, TcpStream}; + use std::thread; + + fn tcp_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let a = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let (b, _) = listener.accept().unwrap(); + (a, b) + } + + /// (volunteer side, coordinator side, coordinator identity). + fn session( + pin: impl FnOnce(&Keypair) -> CoordinatorPin, + ) -> ( + Result, TransportError>, + Result, TransportError>, + ) { + let (v, c) = tcp_pair(); + let coord = Keypair::generate(); + let pin = pin(&coord); + let handle = thread::spawn(move || coordinator_handshake(c, &coord)); + let vol = volunteer_handshake(v, &Keypair::generate(), pin); + (vol, handle.join().unwrap()) + } + + fn pinned() -> (EncryptedStream, EncryptedStream) { + let (v, c) = session(|k| CoordinatorPin::Key(k.pubkey_bytes())); + (v.unwrap(), c.unwrap()) + } + + /// Reads one raw record (header and sealed bytes) off the wire. + fn raw_record(s: &mut TcpStream) -> Vec { + let mut header = [0u8; 4]; + s.read_exact(&mut header).unwrap(); + let mut rec = header.to_vec(); + let mut body = vec![0u8; u32::from_be_bytes(header) as usize]; + s.read_exact(&mut body).unwrap(); + rec.extend_from_slice(&body); + rec + } + + fn send_frame(s: &mut EncryptedStream, msg: &[u8]) { + s.write_all(msg).unwrap(); + s.flush().unwrap(); + } + + #[test] + fn test_round_trip_both_directions() { + let (mut v, mut c) = pinned(); + send_frame(&mut v, b"update from the volunteer"); + let mut got = [0u8; 25]; + c.read_exact(&mut got).unwrap(); + assert_eq!(&got, b"update from the volunteer"); + send_frame(&mut c, b"receipt"); + let mut got = [0u8; 7]; + v.read_exact(&mut got).unwrap(); + assert_eq!(&got, b"receipt"); + assert_eq!(c.peer_identity().len(), 32); + // Frames of the wire format pass through unchanged. + let unit = crate::wire::BaseModel { + dim: 2, + params: vec![1.0, 2.0], + }; + crate::wire::send(&mut c, &unit).unwrap(); + let back: crate::wire::BaseModel = crate::wire::recv(&mut v).unwrap(); + assert_eq!(back.params, unit.params); + } + + #[test] + fn test_ciphertext_on_the_wire_is_not_the_plaintext() { + let (mut v, mut c) = pinned(); + send_frame(&mut v, b"secret shard bytes"); + let rec = raw_record(c.get_mut()); + assert!(!rec.windows(6).any(|w| w == b"secret")); + } + + #[test] + fn test_tampered_ciphertext_is_rejected() { + let (mut v, mut c) = pinned(); + send_frame(&mut v, b"hello coordinator"); + let mut rec = raw_record(c.get_mut()); + rec[10] ^= 0x01; + v.get_mut().write_all(&rec).unwrap(); + let mut buf = [0u8; 17]; + let err = c.read_exact(&mut buf).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + // The stream stays closed afterwards. + assert!(c.read(&mut buf).is_err()); + } + + #[test] + fn test_tampered_length_header_is_rejected() { + let (mut v, mut c) = pinned(); + send_frame(&mut v, b"hello coordinator"); + let mut rec = raw_record(c.get_mut()); + // Shorter length, same bytes: the header is authenticated. + let len = u32::from_be_bytes(rec[..4].try_into().unwrap()) - 1; + rec[..4].copy_from_slice(&len.to_be_bytes()); + v.get_mut().write_all(&rec).unwrap(); + let mut buf = [0u8; 4]; + assert!(c.read_exact(&mut buf).is_err()); + } + + #[test] + fn test_replayed_record_is_rejected() { + let (mut v, mut c) = pinned(); + send_frame(&mut v, b"one"); + let rec = raw_record(c.get_mut()); + v.get_mut().write_all(&rec).unwrap(); + let mut buf = [0u8; 3]; + c.read_exact(&mut buf).unwrap(); + assert_eq!(&buf, b"one"); + v.get_mut().write_all(&rec).unwrap(); + let err = c.read_exact(&mut buf).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + #[test] + fn test_reordered_records_are_rejected() { + let (mut v, mut c) = pinned(); + send_frame(&mut v, b"first"); + send_frame(&mut v, b"second"); + let r1 = raw_record(c.get_mut()); + let r2 = raw_record(c.get_mut()); + v.get_mut().write_all(&r2).unwrap(); + v.get_mut().write_all(&r1).unwrap(); + let mut buf = [0u8; 6]; + assert!(c.read_exact(&mut buf).is_err()); + } + + #[test] + fn test_record_from_the_other_direction_is_rejected() { + // Each direction has its own key: a record the coordinator sent + // cannot be reflected back to it. + let (mut v, mut c) = pinned(); + send_frame(&mut c, b"receipt"); + let rec = raw_record(v.get_mut()); + v.get_mut().write_all(&rec).unwrap(); + let mut buf = [0u8; 7]; + assert!(c.read_exact(&mut buf).is_err()); + } + + #[test] + fn test_wrong_pinned_key_is_refused() { + let other = Keypair::generate().pubkey_bytes(); + let (v, _c) = session(|_| CoordinatorPin::Key(other)); + assert!(matches!(v, Err(TransportError::WrongCoordinator { .. }))); + } + + #[test] + fn test_trust_first_use_accepts_and_reports_the_key() { + let (v, c) = session(|_| CoordinatorPin::TrustFirstUse); + let (v, _c) = (v.unwrap(), c.unwrap()); + assert_ne!(v.peer_identity(), [0u8; 32]); + } + + #[test] + fn test_forged_coordinator_hello_is_refused() { + // A hello signed for a different volunteer hello (a replay from + // another session) does not verify. + let (v, mut fake) = tcp_pair(); + let coord = Keypair::generate(); + let pin = CoordinatorPin::Key(coord.pubkey_bytes()); + let handle = thread::spawn(move || { + let mut theirs = [0u8; HELLO_LEN]; + fake.read_exact(&mut theirs).unwrap(); + let eph = EphemeralKeyAgreement::generate(); + let stale = Hello::new(&coord, &eph.public_bytes(), COORDINATOR_CONTEXT); + fake.write_all(&stale.bytes).unwrap(); + }); + let res = volunteer_handshake(v, &Keypair::generate(), pin); + handle.join().unwrap(); + assert!(matches!(res, Err(TransportError::BadSignature))); + } + + #[test] + fn test_wrong_version_is_refused() { + let (mut v, c) = tcp_pair(); + let handle = thread::spawn(move || coordinator_handshake(c, &Keypair::generate())); + v.write_all(&[0u8; HELLO_LEN]).unwrap(); + assert!(matches!( + handle.join().unwrap(), + Err(TransportError::BadVersion) + )); + } +} diff --git a/crates/oh-volunteer/src/main.rs b/crates/oh-volunteer/src/main.rs index bfe33b7..a04a665 100644 --- a/crates/oh-volunteer/src/main.rs +++ b/crates/oh-volunteer/src/main.rs @@ -1,11 +1,18 @@ //! `oh-volunteer`: lend this machine's time to train the shared model. //! //! Usage: oh-volunteer --coordinator [--rounds ] [--seed ] -//! [--keep ] [--coordinator-key ] +//! [--keep ] (--coordinator-key | --trust-first-use) //! //! oh-volunteer teach --coordinator [--teacher ] //! [--model ] [--batches ] [--max-tokens ] [--seed ] -//! [--coordinator-key ] +//! (--coordinator-key | --trust-first-use) +//! +//! Every connection is encrypted (`oh_core::volunteer_handshake`). The +//! coordinator must prove it holds `--coordinator-key` (printed by +//! oh-coordinator at start as "coordinator key "). Without that key the +//! volunteer refuses to connect unless `--trust-first-use` is given, which +//! accepts the first key seen, prints a warning, and pins it for the rest +//! of the run. //! //! `teach` asks a language model running on this machine (default //! 127.0.0.1:18091; the model id is read from its /v1/models unless @@ -13,7 +20,9 @@ //! Personal Data Firewall, and sends only the firewalled text. use beacon_core::Keypair; -use oh_core::{discover_model, LocalHttpTeacher}; +use oh_core::{ + discover_model, volunteer_handshake, CoordinatorPin, EncryptedStream, LocalHttpTeacher, +}; use oh_volunteer::{ parse_key, run_once, summary_line, teach_once, teach_summary_line, Settings, VolunteerError, DEFAULT_KEEP_FRACTION, DEFAULT_TEACHER_ADDR, DEFAULT_TEACH_MAX_TOKENS, @@ -29,6 +38,7 @@ struct Args { rounds: u32, settings: Settings, teach: Option, + trust_first_use: bool, } struct TeachArgs { @@ -47,6 +57,7 @@ fn parse_args() -> Result { coordinator_key: None, }, teach: None, + trust_first_use: false, }; let mut it = std::env::args().skip(1).peekable(); if it.peek().map(String::as_str) == Some("teach") { @@ -58,6 +69,10 @@ fn parse_args() -> Result { }); } while let Some(flag) = it.next() { + if flag == "--trust-first-use" { + args.trust_first_use = true; + continue; + } let value = it.next().ok_or(format!("{flag} needs a value"))?; let bad = || format!("bad value for {flag}: {value}"); match (flag.as_str(), args.teach.as_mut()) { @@ -90,9 +105,44 @@ fn parse_args() -> Result { if args.coordinator.is_empty() { return Err("--coordinator is required".into()); } + if args.settings.coordinator_key.is_none() && !args.trust_first_use { + return Err("refusing to connect without the coordinator's key: pass \ + --coordinator-key (oh-coordinator prints it at start as \ + \"coordinator key \"), or --trust-first-use to accept the first \ + key seen" + .into()); + } Ok(args) } +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// Connects and opens the encrypted transport. Pinned to +/// `--coordinator-key`, or to the key accepted on first use; after the +/// first connection the key is kept, so work units signed by any other key +/// are refused as well. +fn open(args: &mut Args, keypair: &Keypair) -> Result, String> { + let stream = connect(&args.coordinator)?; + let pin = match args.settings.coordinator_key { + Some(key) => CoordinatorPin::Key(key), + None => CoordinatorPin::TrustFirstUse, + }; + let secure = volunteer_handshake(stream, keypair, pin) + .map_err(|e| format!("secure connection to {} failed: {e}", args.coordinator))?; + if pin == CoordinatorPin::TrustFirstUse { + let key = hex(&secure.peer_identity()); + eprintln!( + "WARNING: trusting coordinator key {key} on first use. Anyone between this \ + machine and the coordinator could have supplied it. Check it against the \ + coordinator's \"coordinator key\" line and pass --coordinator-key {key} next time." + ); + } + args.settings.coordinator_key = Some(secure.peer_identity()); + Ok(secure) +} + /// The coordinator may not be listening yet; keep trying for a minute. fn connect(addr: &str) -> Result { let deadline = Instant::now() + Duration::from_secs(60); @@ -114,15 +164,11 @@ fn connect(addr: &str) -> Result { fn train(args: &mut Args, keypair: &Keypair) -> ExitCode { for _ in 0..args.rounds { - let result = connect(&args.coordinator).and_then(|mut stream| { + let result = open(args, keypair).and_then(|mut stream| { run_once(&mut stream, keypair, &args.settings).map_err(|e| e.to_string()) }); match result { - Ok(outcome) => { - println!("{}", summary_line(&outcome)); - // Trust the first coordinator key seen for the rest of the run. - args.settings.coordinator_key = Some(outcome.coordinator_key); - } + Ok(outcome) => println!("{}", summary_line(&outcome)), Err(e) => { eprintln!("oh-volunteer: work unit failed: {e}"); return ExitCode::FAILURE; @@ -148,7 +194,7 @@ fn teach(args: &mut Args, keypair: &Keypair) -> ExitCode { let teacher = LocalHttpTeacher::new(&t.teacher, &model); let max_tokens = t.max_tokens; for batch in 1..=args.rounds { - let mut stream = match connect(&args.coordinator) { + let mut stream = match open(args, keypair) { Ok(s) => s, Err(e) => { eprintln!("oh-volunteer: {e}"); @@ -156,10 +202,7 @@ fn teach(args: &mut Args, keypair: &Keypair) -> ExitCode { } }; match teach_once(&mut stream, keypair, &args.settings, &teacher, max_tokens) { - Ok(outcome) => { - println!("{}", teach_summary_line(batch, &outcome)); - args.settings.coordinator_key = Some(outcome.coordinator_key); - } + Ok(outcome) => println!("{}", teach_summary_line(batch, &outcome)), // A refusal is this batch's answer, not a failure of the run. Err(e @ (VolunteerError::Firewall(_) | VolunteerError::EmptyTeaching)) => { println!("Batch {batch}: not sent, {e}"); diff --git a/docs/design/qwen-teacher-v0.md b/docs/design/qwen-teacher-v0.md index 2054e36..7a13fb2 100644 --- a/docs/design/qwen-teacher-v0.md +++ b/docs/design/qwen-teacher-v0.md @@ -31,7 +31,7 @@ The teacher must be on the same machine. `LocalHttpTeacher` refuses any address The hint itself comes from the coordinator's own training shard, so sending it to a local teacher reveals nothing new. A model that echoes its hint will send those (at most 200) shard bytes back inside its passage; tests check that no other shard bytes leave. -Frames: an update frame is unchanged. A teaching frame is the 8 bytes `OHTEACH1` followed by the bincode batch (`send_teaching`, `recv_submission`). An update frame begins with its run id, a BLAKE3-derived value, so it collides with the tag with probability 2^-64; existing volunteers keep working unmodified. +Frames: an update frame is unchanged. A teaching frame is the 8 bytes `OHTEACH1` followed by the bincode batch (`send_teaching`, `recv_submission`). An update frame begins with its run id, a BLAKE3-derived value, so it collides with the tag with probability 2^-64. Since follow-up F3 every frame, teaching or update, travels inside the encrypted transport ([volunteer-compute-v0.md](volunteer-compute-v0.md#encrypted-transport)), so the passage is readable only by the coordinator; `teach` takes `--coordinator-key ` or `--trust-first-use` like training mode. Volunteers built before F3 cannot connect. ## Scoring diff --git a/docs/design/volunteer-compute-v0.md b/docs/design/volunteer-compute-v0.md index 26640db..f8d6a7b 100644 --- a/docs/design/volunteer-compute-v0.md +++ b/docs/design/volunteer-compute-v0.md @@ -1,6 +1,6 @@ # Volunteer compute, version 0 (design) -Status: design for the first working slice. Pure Rust, CPU only. +Status: built and merged (#17); encrypted transport added in follow-up F3. Pure Rust, CPU only. ## Goal @@ -64,7 +64,22 @@ Signing bytes for a struct are the bincode bytes of that struct with its `sig` f - `Receipt { run_id, round, update_digest: [u8;32], turings: f64, accepted: bool, coordinator_pubkey, sig: Vec }` The coordinator's signed answer: the score and whether the update was kept. -Over TCP, each message travels in a frame: a 4-byte big-endian length, then that many bytes of bincode. Frames larger than 64 MiB are refused before any memory is set aside for them. +Each message travels in a frame: a 4-byte big-endian length, then that many bytes of bincode. Frames larger than 64 MiB are refused before any memory is set aside for them. Frames never go on the network in the clear: they travel inside the encrypted transport below. + +## Encrypted transport + +Every connection between `Coordinator::serve` and a volunteer starts with a handshake and then carries only sealed records (`oh_core::transport`, follow-up F3). The primitives are beacon-core's: X25519 key agreement with a fresh one-time key per connection (`EphemeralKeyAgreement`, x25519-dalek) and ChaCha20-Poly1305 (`SessionCipher`, chacha20poly1305). Keys are derived with BLAKE3 `derive_key`. No cryptographic math is written here. + +Handshake, one message each way, volunteer first. Each hello is `"OHSEC001" || identity[32] || ephemeral[32] || sig[64]` (136 bytes): + +1. The volunteer sends a fresh X25519 public key, signed with its Ed25519 identity over `"open-humanity transport v1: volunteer hello" || identity || ephemeral`. +2. The coordinator sends its own fresh X25519 public key, signed with the run's Ed25519 key over `"open-humanity transport v1: coordinator hello" || the volunteer's whole hello || identity || ephemeral`. Because the signature covers the volunteer's fresh key, a coordinator hello recorded from one session does not verify in another. +3. The volunteer checks the signature and then the coordinator identity against its pin. With `--coordinator-key ` (printed by `oh-coordinator` at start as `coordinator key `) any other key is refused before a single record is sent. Without a key the volunteer refuses to connect unless `--trust-first-use` is given; then it accepts the first key, prints a warning naming it, and pins it for the rest of the run. Work units must also be signed by the key proven in the handshake. +4. Both sides compute the X25519 shared secret (an all-zero result from a bad public key is refused) and derive two keys: `derive_key("open-humanity transport v1 2026-10-16 volunteer to coordinator", shared || BLAKE3(both hellos))` and the same with "coordinator to volunteer". + +Records: `len[4, big endian] || ChaCha20-Poly1305 ciphertext and tag`, with the 4 length bytes as associated data, so changing the length also fails. Everything written between two flushes becomes one record (one wire frame, in practice). The 96-bit nonce is 4 zero bytes and then a 64-bit little-endian record counter for that direction. It is never sent and never reused, since each direction has its own key and counter. The receiver only accepts the next counter value, so a replayed, dropped, reordered or reflected record fails to open and the stream is closed. A connection that fails the handshake is closed and does not take a volunteer slot; the handshake must finish within 30 seconds. + +Tests: `crates/oh-core/src/transport.rs` (round trip, tampered ciphertext, tampered length, replayed record, reordered records, reflected record, wrong pinned key, trust on first use, forged coordinator hello, wrong version) and `crates/oh-coordinator/tests/encrypted_transport.rs`. In that test a relay records every byte of a full round (one training and one teaching volunteer) and finds no 16-byte run of the training shards and none of the teaching text. A negative control shows the same check finds both in plain frames. Teaching mode adds one message, `TeachingBatch { run_id, round, base_digest, node_pubkey, text, text_digest, sig }`, signed the same way and sent in a frame tagged `OHTEACH1`. The coordinator scores it by training a copy of the base on the text and measuring held-out bits, and trains the shared model on accepted text after the round's update aggregate step, keeping each text only if held-out bits drop again. Details: [qwen-teacher-v0.md](qwen-teacher-v0.md). @@ -77,8 +92,10 @@ Teaching mode adds one message, `TeachingBatch { run_id, round, base_digest, nod - 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. Version 0 is meant for loopback only; the shard a volunteer receives and the update it returns are readable by anyone on the path. -- Pinned coordinator identity by default. Unless `--coordinator-key` is given, a volunteer trusts the first coordinator key it sees and holds to it for the rest of its run. +- A lasting coordinator identity. The coordinator makes a fresh key at every start, so volunteers need the new key (out of band) for each run, and trust on first use only protects a run, not the next one. +- Hiding sizes and timing. Record lengths and timing are visible on the wire, so an observer can tell a teaching batch from an update and estimate the model size. +- Checking volunteer identities. The coordinator learns each volunteer's key in the handshake but accepts any key; volunteers stay anonymous by design. +- Protection against stalled handshakes. A connection that never finishes the handshake holds up the accept loop for up to 30 seconds. - Protection against slow or silent volunteers. The coordinator waits for one connection per shard and reads each update with a 10 minute timeout, so one stalled volunteer holds up the round. - A ledger line for dropped updates. Updates with a bad signature, or that answer the wrong work unit, get no receipt and no ledger line; they are only logged. Accepted updates get a credit line and every other update that gets a receipt gets a rejected line. diff --git a/scripts/demo-local.sh b/scripts/demo-local.sh index 3b94910..e64a875 100755 --- a/scripts/demo-local.sh +++ b/scripts/demo-local.sh @@ -45,6 +45,24 @@ echo "Corpus: LICENSE, $(wc -c < "$CORPUS") bytes, sha256 $(sha256sum "$CORPUS" echo "Settings: rounds=$ROUNDS volunteers=$VOLUNTEERS steps=$STEPS dim=$DIM seeds=\"$SEEDS\"" echo +# Prints the key from the coordinator's "coordinator key " line, +# waiting up to 30 seconds for it to appear in log $1 (process $2). +wait_for_key() { + local tries=0 key="" + while [ -z "$key" ]; do + key="$(sed -n 's/^coordinator key \([0-9a-f]\{64\}\)$/\1/p' "$1" 2>/dev/null | head -n 1)" + if [ -n "$key" ]; then break; fi + if ! kill -0 "$2" 2>/dev/null || [ "$tries" -ge 300 ]; then + echo "The coordinator did not start:" >&2 + cat "$1" >&2 + exit 1 + fi + tries=$((tries + 1)) + sleep 0.1 + done + echo "$key" +} + common=(--corpus "$CORPUS" --rounds "$ROUNDS" --volunteers-per-round "$VOLUNTEERS" --steps "$STEPS" --dim "$DIM") dist_results=() @@ -56,11 +74,15 @@ for seed in $SEEDS; do "$BIN/oh-coordinator" "${common[@]}" --seed "$seed" \ --listen "127.0.0.1:$port" --ledger "$WORK/ledger-$seed.jsonl" > "$coord_log" 2>&1 & coord_pid=$! + # Volunteers pin the key the coordinator prints, so the encrypted link + # only opens to this coordinator. + key="$(wait_for_key "$coord_log" "$coord_pid")" vol_pids=() for i in $(seq 1 "$VOLUNTEERS"); do "$BIN/oh-volunteer" --coordinator "127.0.0.1:$port" --rounds "$ROUNDS" \ - --seed $((seed + i)) > "$WORK/volunteer-$seed-$i.log" 2>&1 & + --seed $((seed + i)) --coordinator-key "$key" \ + > "$WORK/volunteer-$seed-$i.log" 2>&1 & vol_pids+=($!) done for pid in "${vol_pids[@]}"; do wait "$pid"; done diff --git a/scripts/demo-teach.sh b/scripts/demo-teach.sh index 805686f..ff91cf8 100755 --- a/scripts/demo-teach.sh +++ b/scripts/demo-teach.sh @@ -61,9 +61,24 @@ coord_log="$WORK/coordinator.log" --ledger "$WORK/ledger.jsonl" > "$coord_log" 2>&1 & coord_pid=$! +# The volunteer pins the key the coordinator prints, so the encrypted link +# only opens to this coordinator. Wait up to 30 seconds for that line. +key="" +for _ in $(seq 1 300); do + key="$(sed -n 's/^coordinator key \([0-9a-f]\{64\}\)$/\1/p' "$coord_log" | head -n 1)" + if [ -n "$key" ] || ! kill -0 "$coord_pid" 2>/dev/null; then break; fi + sleep 0.1 +done +if [ -z "$key" ]; then + echo "The coordinator did not start:" + cat "$coord_log" + exit 1 +fi + vol_log="$WORK/volunteer.log" "$BIN/oh-volunteer" teach --coordinator "127.0.0.1:$PORT" --teacher "$TEACHER" \ - --batches "$BATCHES" --max-tokens "$MAX_TOKENS" --seed "$SEED" > "$vol_log" 2>&1 & + --batches "$BATCHES" --max-tokens "$MAX_TOKENS" --seed "$SEED" \ + --coordinator-key "$key" > "$vol_log" 2>&1 & vol_pid=$! vol_status=0