Swarm: TLS between peers, pinned by the hub ecce3719da80

- **Runners share images (#47, docs/jetstream.md §8.1).** A runner with a `[runner.swarm]` section serves the image chunks it holds to other runners in its site, and fetches an image it lacks from them and the hub together, rarest chunks first. The hub is asked only for chunks no peer has or is fetching, so runners starting together split its work. Peers talk over TLS: each runner keeps its own certificate, the hub hands its fingerprint on with the peer, and a fetcher trusts no other. Peers admit each other with a ticket the hub signs, and every chunk is checked against its hash. Uploads are capped (`upload_mib_s`, default 50) at 16 connections. Without the section a runner fetches from the hub alone, as before.

name = "asn1-rs"

version = "0.7.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8"

dependencies = [

 "asn1-rs-derive",

 "asn1-rs-impl",

 "displaydoc",

 "nom",

 "num-traits",

 "rusticata-macros",

 "thiserror 2.0.21",

 "time",

]

[[package]]

name = "asn1-rs-derive"

version = "0.6.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 2.0.119",

 "synstructure 0.13.2",

]

[[package]]

name = "asn1-rs-impl"

version = "0.2.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 2.0.119",

]

[[package]]

[[package]]

name = "base64"

version = "0.23.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"

name = "bit-vec"

version = "0.9.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51"

dependencies = [

 "serde",

]

[[package]]

name = "data-encoding"

version = "2.11.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"

[[package]]

]

[[package]]

name = "der-parser"

version = "10.0.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6"

dependencies = [

 "asn1-rs",

 "displaydoc",

 "nom",

 "num-bigint",

 "num-traits",

 "rusticata-macros",

name = "deranged"

version = "0.5.8"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"

[[package]]

 "rcgen",

 "rustls",

 "thiserror 1.0.69",

 "thiserror 1.0.69",

name = "lazy_static"

version = "1.5.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "20870f649af7073d53e38067b2a84312175d56ea15217e1b15bc83506ec50afb"

[[package]]

[[package]]

name = "minimal-lexical"

version = "0.2.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"

]

[[package]]

name = "nom"

version = "7.1.3"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"

dependencies = [

 "memchr",

 "minimal-lexical",

]

[[package]]

name = "num-bigint"

version = "0.4.8"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367"

dependencies = [

 "num-integer",

 "num-traits",

]

[[package]]

name = "num-conv"

version = "0.2.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441"

[[package]]

name = "num-integer"

version = "0.1.47"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "7ce2d95d4b3734dc35aa2f45e1aa22cd416814592a4f9d9205e11affd5b8e10b"

dependencies = [

 "num-traits",

]

[[package]]

name = "oid-registry"

version = "0.8.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7"

dependencies = [

 "asn1-rs",

 "thiserror 1.0.69",

]

[[package]]

name = "pem"

version = "4.0.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "d354a98a3d1251555de99e8fdd8afda05573c31b82f59063a7b0a29b5527f120"

dependencies = [

 "base64 0.23.1",

 "serde_core",

[[package]]

name = "powerfmt"

version = "0.2.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "4a6394b9e965e73d0a289ee54f589087e2c676aedf60885baf52c76b771e4958"

name = "rcgen"

version = "0.14.10"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "8774e05a7d0de114588e6a28fe7e71694b82614ed569d86d8b389dfbc98b8ad8"

dependencies = [

 "pem",

 "ring",

 "rustls-pki-types",

 "time",

 "x509-parser",

 "yasna",

]

[[package]]

]

[[package]]

name = "rusticata-macros"

version = "4.1.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632"

dependencies = [

 "nom",

version = "0.13.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 2.0.119",

]

[[package]]

name = "synstructure"

dependencies = [

 "thiserror-impl 1.0.69",

]

[[package]]

name = "thiserror"

version = "2.0.21"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e"

 "thiserror-impl 2.0.21",

]

[[package]]

name = "thiserror-impl"

version = "2.0.21"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 3.0.6",

]

[[package]]

name = "time"

version = "0.3.55"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134"

dependencies = [

 "deranged",

 "num-conv",

 "powerfmt",

 "serde_core",

 "time-core",

 "time-macros",

name = "time-core"

version = "0.1.9"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109"

[[package]]

name = "time-macros"

version = "0.2.32"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "7e689342a48d2ea927c87ea50cabf8594854bf940e9310208848d680d668ed85"

dependencies = [

 "num-conv",

 "time-core",

]

[[package]]

 "base64 0.22.1",

[[package]]

name = "x509-parser"

version = "0.18.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202"

dependencies = [

 "asn1-rs",

 "data-encoding",

 "der-parser",

 "lazy_static",

 "nom",

 "oid-registry",

 "ring",

 "rusticata-macros",

 "thiserror 2.0.21",

 "time",

]

[[package]]

name = "yasna"

version = "0.6.0"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282"

dependencies = [

 "bit-vec",

 "time",

]

 "synstructure 0.14.0",

 "synstructure 0.14.0",

/// A runner's place in the image swarm: the address peers reach it on,

/// the site it is in (peers are only ever in the same site), and the

/// BLAKE3 of its TLS certificate, which peers check it by.

    #[serde(default)]

    pub cert: String,

    /// The BLAKE3 of its TLS certificate: a peer that shows another is not

    /// this runner.

    pub cert: String,

rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] }

rcgen = "0.14"

//! The server is a small HTTPS endpoint on the swarm address: a peer

//!

//! Each runner makes its own TLS certificate on first use and reports its

//! BLAKE3 with its swarm address; the hub hands that on with the peer, and

//! a fetcher talks only to a peer that shows that certificate and holds

//! its key. Nobody on the network reads the chunks, tickets or sessions.

use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};

use rustls::crypto::CryptoProvider;

use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName, UnixTime};

use rustls::{DigitallySignedStruct, SignatureScheme};

/// Connections served at once; more are told 503 and come back shortly.

/// Connections past this are closed without a word.

const MAX_OPEN: usize = 4 * MAX_CONNS;

    tls: Arc<rustls::ServerConfig>,

        let (cert, key) = certificate(&home.root)?;

        let fingerprint = blake3::hash(&cert).to_hex().to_string();

        let tls = server_tls(cert, key)?;

            tls,

        let addr = SwarmAddr { addr: sc.advertise.clone().unwrap_or_else(|| sc.listen.clone()), site: sc.site.clone(), cert: fingerprint };

        eprintln!(

            "swarm: serving image chunks on {} over TLS (certificate {}, site {:?}, up to {} MiB/s)",

            sc.listen,

            &addr.cert[..12],

            sc.site,

            sc.upload_mib_s.unwrap_or(50)

        );

        Ok(Some(Swarm { state, addr }))

                            // A busy peer is asked again shortly.

                            (Some(_), _) if e.is::<Busy>() => std::thread::sleep(Duration::from_millis(200)),

                            Err(e) if e.is::<Busy>() => {}

                continue;

            }

            if !is_hash(&sp.cert) {

                eprintln!("swarm: peer {} gave no certificate (an older flow); not asking it", sp.name);

            let joined = hello(&sp.addr, &sp.cert, ticket).and_then(|(agent, session)| {

                    base: format!("https://{}/swarm/v1", sp.addr),

                    agent,

        let r = match p.agent.get(&url).set("Authorization", &format!("Bearer {}", p.session)).call() {

        let r = p.agent.get(&url).set("Authorization", &format!("Bearer {}", p.session)).call().map_err(|e| peer_error(&url, e))?;

/// A peer this fetch asks for chunks, over a connection that trusts only

/// its certificate.

    agent: ureq::Agent,

/// A peer at its connection limit: ask it again shortly.

#[derive(Debug)]

struct Busy;

impl std::fmt::Display for Busy {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        f.write_str("the peer is busy")

    }

}

impl std::error::Error for Busy {}

        ureq::Error::Status(503, _) => anyhow::Error::new(Busy),

    }

}

/// Show `ticket` to the peer at `addr`, trusting only the certificate

/// `cert`: a connection to it and a session.

fn hello(addr: &str, cert: &str, ticket: &str) -> Result<(ureq::Agent, String)> {

    let agent = peer_agent(cert)?;

    let url = format!("https://{addr}/swarm/v1/hello");

    let r: serde_json::Value = agent.post(&url).send_string(ticket).map_err(|e| peer_error(&url, e))?.into_json()?;

    let session = r["session"].as_str().map(String::from).context("the peer gave no session")?;

    Ok((agent, session))

}

fn provider() -> Arc<CryptoProvider> {

    Arc::new(rustls::crypto::ring::default_provider())

}

/// A client that trusts one certificate, by its BLAKE3, whatever name

/// the peer goes by: the one the hub said this peer has.

fn peer_agent(cert: &str) -> Result<ureq::Agent> {

    let provider = provider();

    let cfg = rustls::ClientConfig::builder_with_provider(provider.clone())

        .with_safe_default_protocol_versions()?

        .dangerous()

        .with_custom_certificate_verifier(Arc::new(Pinned { cert: cert.to_string(), provider }))

        .with_no_client_auth();

    Ok(ureq::AgentBuilder::new().tls_config(Arc::new(cfg)).timeout_connect(Duration::from_secs(2)).timeout(Duration::from_secs(20)).build())

}

/// Trusts the certificate with this BLAKE3 and no other. The handshake's

/// signature is still checked against it, so only the holder of its key

/// passes.

#[derive(Debug)]

struct Pinned {

    cert: String,

    provider: Arc<CryptoProvider>,

}

impl ServerCertVerifier for Pinned {

    fn verify_server_cert(

        &self,

        end_entity: &CertificateDer<'_>,

        _intermediates: &[CertificateDer<'_>],

        _name: &ServerName<'_>,

        _ocsp: &[u8],

        _now: UnixTime,

    ) -> std::result::Result<ServerCertVerified, rustls::Error> {

        if blake3::hash(end_entity.as_ref()).to_hex().as_str() == self.cert {

            Ok(ServerCertVerified::assertion())

        } else {

            Err(rustls::Error::General("not the peer the hub named: its certificate is another".into()))

        }

    }

    fn verify_tls12_signature(

        &self,

        message: &[u8],

        cert: &CertificateDer<'_>,

        dss: &DigitallySignedStruct,

    ) -> std::result::Result<HandshakeSignatureValid, rustls::Error> {

        rustls::crypto::verify_tls12_signature(message, cert, dss, &self.provider.signature_verification_algorithms)

    }

    fn verify_tls13_signature(

        &self,

        message: &[u8],

        cert: &CertificateDer<'_>,

        dss: &DigitallySignedStruct,

    ) -> std::result::Result<HandshakeSignatureValid, rustls::Error> {

        rustls::crypto::verify_tls13_signature(message, cert, dss, &self.provider.signature_verification_algorithms)

    }

    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {

        self.provider.signature_verification_algorithms.supported_schemes()

/// The runner's swarm certificate and key (`swarm/cert.der`, and

/// `swarm/key.der` readable by the runner alone), made on first use and

/// kept, so its fingerprint holds across restarts.

fn certificate(root: &Path) -> Result<(Vec<u8>, Vec<u8>)> {

    let dir = root.join("swarm");

    let (cert_path, key_path) = (dir.join("cert.der"), dir.join("key.der"));

    if let (Ok(c), Ok(k)) = (std::fs::read(&cert_path), std::fs::read(&key_path)) {

        return Ok((c, k));

    }

    std::fs::create_dir_all(&dir)?;

    let made = rcgen::generate_simple_self_signed(vec!["flow-swarm".to_string()]).context("make the swarm's certificate")?;

    let (cert, key) = (made.cert.der().to_vec(), made.signing_key.serialize_der());

    {

        use std::os::unix::fs::OpenOptionsExt;

        let mut f = std::fs::OpenOptions::new().write(true).create(true).truncate(true).mode(0o600).open(&key_path)?;

        f.write_all(&key)?;

    }

    std::fs::write(&cert_path, &cert)?;

    Ok((cert, key))

}

fn server_tls(cert: Vec<u8>, key: Vec<u8>) -> Result<Arc<rustls::ServerConfig>> {

    let cfg = rustls::ServerConfig::builder_with_provider(provider())

        .with_safe_default_protocol_versions()?

        .with_no_client_auth()

        .with_single_cert(vec![CertificateDer::from(cert)], PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key)))

        .context("the swarm's certificate")?;

    Ok(Arc::new(cfg))

}

        let Ok(conn) = conn else { continue };

        if state.conns.load(Ordering::Relaxed) >= MAX_OPEN {

            // Past the limit a peer is still answered, with 503, so it

            // comes back rather than giving up on this runner.

            let busy = state.conns.fetch_add(1, Ordering::Relaxed) >= MAX_CONNS;

            serve_tls(conn, &state, busy).ok();

fn serve_tls(conn: TcpStream, state: &State, busy: bool) -> std::io::Result<()> {

    conn.set_read_timeout(Some(Duration::from_secs(30)))?;

    conn.set_write_timeout(Some(Duration::from_secs(30)))?;

    conn.set_nodelay(true)?;

    let tls = rustls::ServerConnection::new(state.tls.clone()).map_err(std::io::Error::other)?;

    serve_conn(rustls::StreamOwned::new(tls, conn), state, busy)

}

fn respond(conn: &mut impl Write, code: u16, kind: &str, body: &[u8]) -> std::io::Result<()> {

fn serve_conn<S: Read + Write>(conn: S, state: &State, busy: bool) -> std::io::Result<()> {

            return respond(r.get_mut(), 400, "text/plain", b"too large");

        if busy {

            return respond(r.get_mut(), 503, "text/plain", b"busy: ask again shortly");

        }

        respond(r.get_mut(), code, kind, &reply)?;

        return match admit(state, &String::from_utf8_lossy(body)) {

fn admit(state: &State, ticket: &str) -> std::result::Result<String, String> {

        hub.runners.lock().unwrap().push((

            token.into(),

            SwarmPeer {

                runner: id,

                name: format!("r{id}"),

                addr: format!("127.0.0.1:{port}"),

                cert: swarm.addr.cert.clone(),

                complete: true,

            },

        ));

        let err = hello(&format!("127.0.0.1:{port}"), &o.addr.cert, &ticket).unwrap_err();

        let url = format!("https://127.0.0.1:{port}/swarm/v1/chunks/{}", m.chunks[0].hash());

        assert!(matches!(peer_agent(&o.addr.cert).unwrap().get(&url).call(), Err(ureq::Error::Status(401, _))));

        // Nor any answer at all without TLS.

        assert!(ureq::get(&url.replacen("https", "http", 1)).call().is_err());

    fn a_peer_with_another_certificate_is_not_trusted() {

        let bytes = image();

        let m = jet::images::scan(&bytes[..], None, |_, _| Ok(())).unwrap();

        let hub = fake_hub(m.clone(), bytes.clone(), "lab");

        let (a_dir, b_dir) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

        std::fs::create_dir_all(a_dir.path().join("cas")).unwrap();

        std::fs::write(a_dir.path().join("cas").join(format!("{}.ext4", m.hash)), &bytes).unwrap();

        let (_a_home, a) = runner(a_dir.path(), &hub, 1, "ta", free_port(), "lab");

        while !a.state.offers.lock().unwrap().contains_key(&m.hash) {

            std::thread::sleep(Duration::from_millis(20));

        }

        let (b_home, b) = runner(b_dir.path(), &hub, 2, "tb", free_port(), "lab");

        // The hub names A, but with B's certificate: whoever answers at A's

        // address is not who the hub vouched for.

        hub.runners.lock().unwrap()[0].1.cert = b.addr.cert.clone();

        let moved = b.fetch(&b_home, &m.hash).unwrap().unwrap();

        assert_eq!(moved.peer_bytes, 0, "nothing was taken from the stranger");

        assert_eq!(hub.chunks_served.load(Ordering::Relaxed), m.needed().len());

        assert_eq!(std::fs::read(b_home.rootfs(&m.hash)).unwrap(), bytes);

        // The certificate is kept: the same runner, the same fingerprint.

        let mut cfg = b_home.config.runner.clone().unwrap();

        cfg.swarm.as_mut().unwrap().listen = "127.0.0.1:0".into();

        let again = Swarm::start(&b_home, &cfg).unwrap().unwrap();

        assert_eq!(again.addr.cert, b.addr.cert);

    }

    #[test]

        let dir = tempfile::tempdir().unwrap();

        let (cert, key) = certificate(dir.path()).unwrap();

            tls: server_tls(cert, key).unwrap(),

- **The hub is the tracker.** A runner reports its swarm address, its site

  and its certificate's fingerprint with every claim. `POST /api/v1/flow/swarm/announce {image, state}`

- **Encrypted, and pinned by the hub.** Each runner makes its own TLS

  certificate on first use (`swarm/cert.der`, its key readable by the

  runner alone) and keeps it, so its fingerprint (BLAKE3 of the

  certificate) holds across restarts. The tracker hands each peer out with

  that fingerprint, and a fetcher talks only to a peer showing that

  certificate and proving its key in the handshake; no CA is involved. A

  peer at another address, or a stranger at a peer's, gets nothing and

  sees nothing: chunks, tickets and sessions all travel inside TLS.

- **The peer protocol** is HTTPS on the runner's swarm address:

    connection limit, and the fetcher asks it again shortly.

  runner neither listens nor asks peers. Peers are not offered across

  sites, so a site is the network its runners can reach each other on.