From cdaee30c14137e975e4aefb6b46489636f038189 Mon Sep 17 00:00:00 2001 From: ejc3 Date: Sun, 25 Oct 2026 11:10:03 +0000 Subject: [PATCH] fleet aggregate: serve /frames with TLS on public listeners An https page cannot open ws://, so a region stream a browser reads must be served with TLS. `fcvm fleet aggregate --tls-listen ` (repeatable, with `--tls-file`) serves GET /frames and GET /healthz with TLS on that address. POST/GET /layout and /stats stay on --listen only, so a public address never accepts a layout. - src/fleet/aggregate/tls.rs: the certificate is the agent's renderer file (tls.json, {"cert", "key"} PEM). It is read again every 2 s and a new version is used for the connections that follow; a version that does not load is logged and the previous one kept. The listener accepts and runs each handshake on its own task with a 10 s deadline (at most 256 at once), so a stalled client delays no other. - public_router: /frames and /healthz only. - Cargo.toml: rustls and tokio-rustls as direct dependencies (ring, no aws-lc-rs; both were already in the lock through reqwest and tokio-tungstenite). - fleet/PROTOCOL.md: the two flags. Tests: the_public_listener_serves_frames_with_tls_and_no_control_routes (tests/test_fleet_aggregate.rs) reads the region header and a frame over wss, gets 404 for /layout and /stats over TLS, and after a renewal sees the new certificate on the next connection while a client trusting only the old one is refused. With public_router returning the full router it fails at the 404 assertion. a_file_without_a_usable_identity_is_refused covers missing key, empty cert, bad key and a mismatched key. --- Cargo.lock | 2 + Cargo.toml | 2 + fleet/PROTOCOL.md | 7 ++ src/cli/args.rs | 10 ++ src/fleet/aggregate/mod.rs | 42 ++++++- src/fleet/aggregate/tls.rs | 225 ++++++++++++++++++++++++++++++++++ tests/test_fleet_aggregate.rs | 134 ++++++++++++++++++++ 7 files changed, 420 insertions(+), 2 deletions(-) create mode 100644 src/fleet/aggregate/tls.rs diff --git a/Cargo.lock b/Cargo.lock index 88d041cea..ed521e9e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -834,6 +834,7 @@ dependencies = [ "rcgen", "regex", "reqwest", + "rustls", "serde", "serde_json", "serde_norway", @@ -844,6 +845,7 @@ dependencies = [ "tar", "tempfile", "tokio", + "tokio-rustls", "tokio-tungstenite", "tokio-util", "toml", diff --git a/Cargo.toml b/Cargo.toml index 51c738d19..a47e37c9f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,8 @@ exec-proto = { path = "exec-proto" } fleet-core = { path = "fleet/core" } futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] } tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } +tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } instant-acme = { version = "0.8", default-features = false, features = ["ring", "hyper-rustls", "rcgen"] } rcgen = { version = "0.14", default-features = false, features = ["crypto", "pem", "ring", "x509-parser"] } x509-parser = "0.18" diff --git a/fleet/PROTOCOL.md b/fleet/PROTOCOL.md index fd97cf916..c5f44f970 100644 --- a/fleet/PROTOCOL.md +++ b/fleet/PROTOCOL.md @@ -862,6 +862,13 @@ processes; the host needs the `ffmpeg` CLI with libx264. `after_last_tile_ms` (last tile frame used to the region frame out of the encoder), `after_first_tile_ms`, `encode_ms`, `tile_lateness_ms` and `out_lateness_ms` (wall clock minus the slot's scene time), and per tile whether it is streaming. +- `--tls-listen ` (repeatable, with `--tls-file `) also serves `GET + /frames` and `GET /healthz` with TLS on that address, for browsers on an https page. + `/layout` and `/stats` are served on `--listen` only, which the agent and the + coordinator keep on an address of their own. `--tls-file` is the file the agent writes + for its renderers (`tls.json`, `{"cert", "key"}`, PEM; see the agent's `--tls-secret`): + it is read again every 2 s and a new version is used for the connections that follow; + a version that does not load is logged and the previous one kept. ## Renderer diff --git a/src/cli/args.rs b/src/cli/args.rs index 0ed53a640..440cc4c9d 100644 --- a/src/cli/args.rs +++ b/src/cli/args.rs @@ -110,6 +110,16 @@ pub struct FleetAggregateArgs { /// Initial layout (JSON), instead of waiting for POST /layout #[arg(long)] pub layout: Option, + + /// Also serve /frames and /healthz with TLS on this address, for browsers (repeatable). + /// /layout and /stats are served on --listen only. + #[arg(long = "tls-listen", requires = "tls_file")] + pub tls_listen: Vec, + + /// The certificate of --tls-listen: a JSON object {"cert", "key"} (PEM), the file the + /// agent writes for its renderers; read again every 2 s + #[arg(long, requires = "tls_listen")] + pub tls_file: Option, } #[derive(Clone, Copy, Debug, PartialEq, Eq, ValueEnum)] diff --git a/src/fleet/aggregate/mod.rs b/src/fleet/aggregate/mod.rs index d9ca8dcbf..e3ff1784e 100644 --- a/src/fleet/aggregate/mod.rs +++ b/src/fleet/aggregate/mod.rs @@ -23,6 +23,7 @@ pub mod codec; pub mod compose; pub mod frame; mod tile; +pub mod tls; use crate::cli::{FleetAggregateArgs, FleetAggregateCodec}; use crate::fleet::now_ms; @@ -732,6 +733,15 @@ pub fn router(shared: Arc) -> Router { .with_state(shared) } +/// What a public listener (`--tls-listen`) serves: the region stream and the health check, +/// never the layout or the stats. +pub fn public_router(shared: Arc) -> Router { + Router::new() + .route("/frames", get(frames_route)) + .route("/healthz", get(|| async { "ok" })) + .with_state(shared) +} + pub async fn cmd(args: FleetAggregateArgs) -> Result<()> { let codec: Arc = match args.codec { FleetAggregateCodec::Ffmpeg => Arc::new(codec::Ffmpeg { bin: args.ffmpeg }), @@ -755,15 +765,43 @@ pub async fn cmd(args: FleetAggregateArgs) -> Result<()> { .with_context(|| format!("reading {}", path.display()))?; shared.set_layout(l); } + // The public listeners are bound before the private one, so a started aggregator + // (the private `/healthz` answers) is serving every address it was given. + let mut public = tokio::task::JoinSet::new(); + if !args.tls_listen.is_empty() { + let file = args + .tls_file + .as_deref() + .context("--tls-listen needs --tls-file")?; + let certs = tls::CertFile::load(file)?; + tokio::spawn(certs.clone().watch(tls::RELOAD)); + for addr in &args.tls_listen { + let l = tls::TlsListener::bind(addr, certs.clone()).await?; + eprintln!( + "fleet aggregator serving /frames with TLS on {}", + l.address() + ); + let app = public_router(shared.clone()); + let addr = addr.clone(); + public.spawn(async move { (addr, axum::serve(l, app).await) }); + } + } let cancel = CancellationToken::new(); tokio::spawn(run(shared.clone(), codec, opts, cancel.clone())); let listener = tokio::net::TcpListener::bind(&args.listen) .await .with_context(|| format!("listening on {}", args.listen))?; eprintln!("fleet aggregator listening on {}", listener.local_addr()?); - axum::serve(listener, router(shared)).await?; + let result = tokio::select! { + r = axum::serve(listener, router(shared)) => r.map_err(anyhow::Error::from), + Some(r) = public.join_next() => match r { + Ok((addr, Ok(()))) => Err(anyhow::anyhow!("the listener on {addr} stopped")), + Ok((addr, Err(e))) => Err(anyhow::Error::from(e).context(format!("serving on {addr}"))), + Err(e) => Err(anyhow::Error::from(e)), + }, + }; cancel.cancel(); - Ok(()) + result } #[cfg(test)] diff --git a/src/fleet/aggregate/tls.rs b/src/fleet/aggregate/tls.rs new file mode 100644 index 000000000..f2b58d8f9 --- /dev/null +++ b/src/fleet/aggregate/tls.rs @@ -0,0 +1,225 @@ +//! The aggregator's public listeners (`--tls-listen`): browsers on an https page open +//! `wss://`, so a region stream they read is served with TLS. The certificate is the +//! fleet's wildcard in the file the agent writes for its renderers (`--tls-file`, +//! `tls.json`: a JSON object `{"cert", "key"}`, PEM; src/fleet/agent/tls.rs). The file is +//! read again every 2 s, and a new version is used for the connections that follow, so a +//! renewal needs no restart; a version that does not load is logged and the previous one +//! kept. +//! +//! A public listener serves only `/frames` and `/healthz` ([`super::public_router`]): +//! `/layout` and `/stats` stay on `--listen`, which the agent and the coordinator keep on +//! an address of their own. + +use anyhow::{bail, Context, Result}; +use rustls::pki_types::pem::PemObject; +use rustls::pki_types::{CertificateDer, PrivateKeyDer}; +use rustls::ServerConfig; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, RwLock}; +use std::time::Duration; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{mpsc, Semaphore}; +use tokio_rustls::server::TlsStream; +use tokio_rustls::TlsAcceptor; + +/// How often the certificate file is read again. +pub const RELOAD: Duration = Duration::from_secs(2); +/// A client that has not finished its handshake by then is dropped. +const HANDSHAKE: Duration = Duration::from_secs(10); +/// Handshakes in progress at once per listener; further connections wait to be accepted. +const HANDSHAKES: usize = 256; + +/// A server configuration for the certificate file's contents. +pub fn server_config(bytes: &[u8]) -> Result> { + #[derive(serde::Deserialize)] + struct File { + cert: String, + key: String, + } + let f: File = + serde_json::from_slice(bytes).context("not a JSON object with `cert` and `key`")?; + let certs = CertificateDer::pem_slice_iter(f.cert.as_bytes()) + .collect::, _>>() + .context("`cert` is not PEM certificates")?; + if certs.is_empty() { + bail!("`cert` holds no certificate"); + } + let key = PrivateKeyDer::from_pem_slice(f.key.as_bytes()) + .context("`key` is not a PEM private key")?; + let mut c = + ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions()? + .with_no_client_auth() + .with_single_cert(certs, key) + .context("the certificate and key do not form a server identity")?; + // WebSocket upgrades are HTTP/1.1. + c.alpn_protocols = vec![b"http/1.1".to_vec()]; + Ok(Arc::new(c)) +} + +/// The certificate file and the configuration loaded from its current version. +pub struct CertFile { + path: PathBuf, + current: RwLock<(Vec, Arc)>, +} + +impl CertFile { + /// Load the file; one that cannot be read or does not load is an error. + pub fn load(path: &Path) -> Result> { + let bytes = std::fs::read(path).with_context(|| format!("reading {}", path.display()))?; + let config = + server_config(&bytes).with_context(|| format!("loading {}", path.display()))?; + Ok(Arc::new(CertFile { + path: path.to_path_buf(), + current: RwLock::new((bytes, config)), + })) + } + + /// The configuration new connections get. + pub fn config(&self) -> Arc { + self.current.read().unwrap().1.clone() + } + + /// Read the file again. Returns whether a new version was loaded; a version that does + /// not load is an error and the previous one stays in use. + pub fn reload(&self) -> Result { + let bytes = std::fs::read(&self.path) + .with_context(|| format!("reading {}", self.path.display()))?; + if self.current.read().unwrap().0 == bytes { + return Ok(false); + } + let config = + server_config(&bytes).with_context(|| format!("loading {}", self.path.display()))?; + *self.current.write().unwrap() = (bytes, config); + Ok(true) + } + + /// Reload every `every` for as long as the process runs. + pub async fn watch(self: Arc, every: Duration) { + loop { + tokio::time::sleep(every).await; + match self.reload() { + Ok(true) => { + tracing::info!(file = %self.path.display(), "aggregator certificate updated") + } + Ok(false) => {} + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "aggregator certificate not reloaded; keeping the previous one") + } + } + } + } +} + +/// A TLS listener for `axum::serve`. Connections are accepted and their handshakes run on +/// a task of its own, each bounded by a deadline, so a client that stalls its handshake +/// delays no other. +pub struct TlsListener { + rx: mpsc::Receiver<(TlsStream, SocketAddr)>, + local: SocketAddr, + acceptor: tokio::task::JoinHandle<()>, +} + +impl Drop for TlsListener { + fn drop(&mut self) { + self.acceptor.abort(); + } +} + +impl TlsListener { + /// The address it is bound to. + pub fn address(&self) -> SocketAddr { + self.local + } + + pub async fn bind(addr: &str, certs: Arc) -> Result { + let tcp = TcpListener::bind(addr) + .await + .with_context(|| format!("listening on {addr}"))?; + let local = tcp.local_addr()?; + let (tx, rx) = mpsc::channel(64); + let acceptor = tokio::spawn(accept_loop(tcp, certs, tx)); + Ok(TlsListener { + rx, + local, + acceptor, + }) + } +} + +async fn accept_loop( + tcp: TcpListener, + certs: Arc, + tx: mpsc::Sender<(TlsStream, SocketAddr)>, +) { + let slots = Arc::new(Semaphore::new(HANDSHAKES)); + loop { + let Ok(permit) = slots.clone().acquire_owned().await else { + return; + }; + let (stream, peer) = match tcp.accept().await { + Ok(c) => c, + Err(e) => { + // EMFILE and the like: back off instead of spinning. + tracing::warn!(error = %e, "aggregator TLS accept failed"); + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + }; + let acceptor = TlsAcceptor::from(certs.config()); + let tx = tx.clone(); + tokio::spawn(async move { + let _permit = permit; + match tokio::time::timeout(HANDSHAKE, acceptor.accept(stream)).await { + Ok(Ok(s)) => { + let _ = tx.send((s, peer)).await; + } + Ok(Err(e)) => tracing::debug!(%peer, error = %e, "TLS handshake failed"), + Err(_) => tracing::debug!(%peer, "TLS handshake timed out"), + } + }); + } +} + +impl axum::serve::Listener for TlsListener { + type Io = TlsStream; + type Addr = SocketAddr; + + async fn accept(&mut self) -> (Self::Io, Self::Addr) { + match self.rx.recv().await { + Some(c) => c, + // The accept task only ends when the listener is dropped. + None => std::future::pending().await, + } + } + + fn local_addr(&self) -> std::io::Result { + Ok(self.local) + } +} + +#[cfg(test)] +mod tests { + use super::server_config; + + #[test] + fn a_file_without_a_usable_identity_is_refused() { + let ck = rcgen::generate_simple_self_signed(vec!["agg.example.test".into()]).unwrap(); + let good = + serde_json::json!({"cert": ck.cert.pem(), "key": ck.signing_key.serialize_pem()}); + assert!(server_config(good.to_string().as_bytes()).is_ok()); + for bad in [ + serde_json::json!({"cert": ck.cert.pem()}), + serde_json::json!({"cert": "", "key": ck.signing_key.serialize_pem()}), + serde_json::json!({"cert": ck.cert.pem(), "key": "not a key"}), + ] { + assert!(server_config(bad.to_string().as_bytes()).is_err(), "{bad}"); + } + // A key that is not the certificate's. + let other = rcgen::generate_simple_self_signed(vec!["x.example.test".into()]).unwrap(); + let mixed = + serde_json::json!({"cert": ck.cert.pem(), "key": other.signing_key.serialize_pem()}); + assert!(server_config(mixed.to_string().as_bytes()).is_err()); + } +} diff --git a/tests/test_fleet_aggregate.rs b/tests/test_fleet_aggregate.rs index 5e85b76af..d34f8c442 100644 --- a/tests/test_fleet_aggregate.rs +++ b/tests/test_fleet_aggregate.rs @@ -692,3 +692,137 @@ async fn a_re_tile_of_streaming_tiles_skips_no_slot() { ); cancel.cancel(); } + +/// A self-signed certificate for `name`, as the agent's `tls.json` holds it, and the +/// certificate alone (PEM) for a client to trust. +fn cert_file(name: &str) -> (String, String) { + let ck = rcgen::generate_simple_self_signed(vec![name.to_string()]).unwrap(); + let body = serde_json::json!({"cert": ck.cert.pem(), "key": ck.signing_key.serialize_pem()}); + (body.to_string(), ck.cert.pem()) +} + +fn trusting(pem: &str) -> Arc { + use rustls::pki_types::pem::PemObject; + let mut roots = rustls::RootCertStore::empty(); + for c in rustls::pki_types::CertificateDer::pem_slice_iter(pem.as_bytes()) { + roots.add(c.unwrap()).unwrap(); + } + Arc::new( + rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::ring::default_provider(), + )) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(roots) + .with_no_client_auth(), + ) +} + +/// Open `/frames` at `addr` over TLS as `wss:///frames`, trusting only `pem`. +async fn open_wss(addr: std::net::SocketAddr, name: &str, pem: &str) -> anyhow::Result { + let tcp = tokio::net::TcpStream::connect(addr).await?; + let (ws, _) = tokio_tungstenite::client_async_tls_with_config( + format!("wss://{name}:{}/frames", addr.port()), + tcp, + None, + Some(tokio_tungstenite::Connector::Rustls(trusting(pem))), + ) + .await?; + Ok(ws) +} + +/// The public listener (`--tls-listen`) serves the region stream with the certificate +/// file's identity and a renewed file to the connections that follow; it does not serve +/// the layout or the stats, which stay on the private listener. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_public_listener_serves_frames_with_tls_and_no_control_routes() { + use fcvm::fleet::aggregate::public_router; + use fcvm::fleet::aggregate::tls::{CertFile, TlsListener}; + const NAME: &str = "agg.fleet.test"; + let shared = Shared::new(); + let region = Rect { + x: 0.5, + y: 0.0, + w: 0.5, + h: 1.0, + }; + shared.set_layout(Layout { + epoch_unix_ms: now_ms(), + image_aspect: 16.0 / 9.0, + region, + tiles: vec![TileSpec { + id: "ec2-1".into(), + rect: region, + url: None, + }], + generation: 1, + from_frame: 0, + }); + let cancel = CancellationToken::new(); + let opts = Options { + image_height: 36, + ..Options::default() + }; + tokio::spawn(run(shared.clone(), Arc::new(Raw), opts, cancel.clone())); + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("tls.json"); + let (first, first_pem) = cert_file(NAME); + std::fs::write(&path, &first).unwrap(); + let certs = CertFile::load(&path).unwrap(); + let l = TlsListener::bind("127.0.0.1:0", certs.clone()) + .await + .unwrap(); + let addr = l.address(); + let app = public_router(shared.clone()); + tokio::spawn(async move { axum::serve(l, app).await.unwrap() }); + + let mut ws = open_wss(addr, NAME, &first_pem).await.unwrap(); + let h: serde_json::Value = match next_msg(&mut ws).await { + Message::Text(t) => serde_json::from_str(&t).unwrap(), + other => panic!("header first, got {other:?}"), + }; + assert_eq!((&h["x0"], &h["height"]), (&32.into(), &36.into())); + assert!(matches!(next_msg(&mut ws).await, Message::Binary(_))); + + let http = |pem: &str| { + reqwest::Client::builder() + .use_rustls_tls() + .tls_built_in_root_certs(false) + .add_root_certificate(reqwest::Certificate::from_pem(pem.as_bytes()).unwrap()) + .resolve(NAME, addr) + .no_proxy() + .build() + .unwrap() + }; + let base = format!("https://{NAME}:{}", addr.port()); + let c = http(&first_pem); + let r = c.get(format!("{base}/healthz")).send().await.unwrap(); + assert_eq!(r.status(), 200); + assert_eq!(r.text().await.unwrap(), "ok"); + for (method, path) in [("GET", "/layout"), ("POST", "/layout"), ("GET", "/stats")] { + let r = c + .request(method.parse().unwrap(), format!("{base}{path}")) + .json(&serde_json::json!({})) + .send() + .await + .unwrap(); + assert_eq!(r.status(), 404, "{method} {path} over the public listener"); + } + + // A renewal: the next connection gets the new certificate, and one that trusts only + // the old certificate is refused. + let (second, second_pem) = cert_file(NAME); + std::fs::write(&path, &second).unwrap(); + assert!(certs.reload().unwrap(), "a new file is a new version"); + assert!(!certs.reload().unwrap(), "the same file is not"); + open_wss(addr, NAME, &second_pem).await.unwrap(); + assert!(open_wss(addr, NAME, &first_pem).await.is_err()); + // A file that does not load keeps the version in use. + std::fs::write(&path, "{}").unwrap(); + assert!(certs.reload().is_err()); + open_wss(addr, NAME, &second_pem).await.unwrap(); + // The stream that was open before the renewal keeps going. + assert!(matches!(next_msg(&mut ws).await, Message::Binary(_))); + cancel.cancel(); +}