diff --git a/seaweed-vector/Cargo.lock b/seaweed-vector/Cargo.lock index 9b2bc8e4d1..7b9d7d8f8d 100644 --- a/seaweed-vector/Cargo.lock +++ b/seaweed-vector/Cargo.lock @@ -6644,6 +6644,8 @@ dependencies = [ "bytes", "clap", "futures", + "http", + "hyper-util", "lance", "lance-arrow", "lance-linalg", @@ -6664,6 +6666,7 @@ dependencies = [ "tonic", "tonic-prost", "tonic-prost-build", + "tower", "tracing", "tracing-subscriber", "weed-lance-worker", diff --git a/seaweed-vector/crates/server/Cargo.toml b/seaweed-vector/crates/server/Cargo.toml index 30084008dd..853aee0db8 100644 --- a/seaweed-vector/crates/server/Cargo.toml +++ b/seaweed-vector/crates/server/Cargo.toml @@ -39,6 +39,9 @@ toml = "0.8" rustls = "0.23" tokio-rustls = "0.26" x509-parser = "0.16" +tower = "0.5" +hyper-util = { version = "0.1", features = ["tokio"] } +http = "1" tokio.workspace = true tokio-stream.workspace = true tracing.workspace = true diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index c883a4c87d..94f89563ad 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -42,13 +42,14 @@ struct Args { filer_signing_key: Option, /// The cluster's security.toml, for gRPC TLS: this server's identity is /// [grpc.vector], the masters are dialed with [grpc.client]. By default - /// the file weed processes find in ., ~/.seaweedfs and /etc/seaweedfs. + /// the file weed processes find in -config_dir, ., ~/.seaweedfs and + /// /etc/seaweedfs. #[arg(long, env = "WEED_VECTOR_SECURITY_TOML")] security_toml: Option, - /// The name the masters' TLS certificates are checked against, when it is - /// not their host. - #[arg(long, env = "WEED_VECTOR_TLS_SERVER_NAME")] - tls_server_name: Option, + /// The directory weed's -config_dir names, searched for security.toml + /// first. + #[arg(long, alias = "config_dir", env = "WEED_VECTOR_CONFIG_DIR")] + config_dir: Option, } fn parse_option(s: &str) -> Result<(String, String), String> { @@ -119,12 +120,12 @@ async fn main() -> Result<()> { tracing_subscriber::fmt::init(); let args = Args::parse_from(normalize_args(std::env::args().collect())); weed_lance_worker::tls::install_default_crypto_provider(); - let security = Security::load(args.security_toml.as_deref())?; + let security = Security::load(args.security_toml.as_deref(), args.config_dir.as_deref())?; if let Some(path) = security.path() { tracing::info!("security settings from {}", path.display()); } let master_tls = match &args.masters { - Some(_) => security.grpc_client_tls("grpc.client", args.tls_server_name.clone())?, + Some(_) => security.grpc_client_tls("grpc.client", "grpc.master")?, None => None, }; let server_tls = security diff --git a/seaweed-vector/crates/server/src/master_client.rs b/seaweed-vector/crates/server/src/master_client.rs index 2ce8e383c2..6cee66ef7f 100644 --- a/seaweed-vector/crates/server/src/master_client.rs +++ b/seaweed-vector/crates/server/src/master_client.rs @@ -7,7 +7,7 @@ use std::time::Duration; use anyhow::{Context, Result}; use tokio::time::sleep; -use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity}; +use tonic::transport::{Channel, Endpoint}; use crate::master_pb::{KeepConnectedRequest, seaweed_client::SeaweedClient}; use crate::security::GrpcTls; @@ -55,32 +55,6 @@ pub fn grpc_addr(master: &str) -> Result { Ok(format!("{scheme}://{rest}")) } -/// The master's gRPC endpoint, over TLS with `grpc.client`'s credentials when -/// the cluster has them; the master's certificate is checked against the host -/// dialed unless the credentials name another. -pub async fn master_endpoint(master: &str, tls: Option<&GrpcTls>) -> Result { - let addr = grpc_addr(master)?; - let Some(tls) = tls else { - return Ok(Endpoint::new(addr)?); - }; - weed_lance_worker::tls::install_default_crypto_provider(); - let endpoint = Endpoint::new(addr.replacen("http://", "https://", 1))?; - let host = dialed_host(&endpoint).with_context(|| format!("master {master} has no host"))?; - let (ca, cert, key) = tls.pems().await?; - let config = ClientTlsConfig::new() - .ca_certificate(Certificate::from_pem(ca)) - .identity(Identity::from_pem(cert, key)) - .domain_name(tls.server_name.clone().unwrap_or(host)); - Ok(endpoint - .tls_config(config) - .with_context(|| format!("TLS for master {master}"))?) -} - -fn dialed_host(endpoint: &Endpoint) -> Option { - let host = endpoint.uri().host()?; - Some(host.trim_matches(|c| c == '[' || c == ']').to_string()) -} - // `host:http.grpc` normalizes to `host:http` so a leader hint naming the // dialed master does not read as a redirect. fn http_addr(master: &str) -> String { @@ -141,9 +115,21 @@ where /// black-holed or silent master otherwise holds up the other masters. const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); -async fn connect(endpoint: impl Future>) -> Result { +/// A channel to the master's gRPC, over TLS with `grpc.client`'s credentials +/// when the cluster has them. +pub async fn connect(master: &str, tls: Option<&GrpcTls>) -> Result { + let addr = grpc_addr(master)?; tokio::time::timeout(CONNECT_TIMEOUT, async { - Ok(endpoint.await?.connect().await?) + Ok(match tls { + None => Endpoint::new(addr)?.connect().await?, + // The connector does the TLS; tonic adds its own only for an + // https URI, which it refuses without a tls_config. + Some(tls) => { + Endpoint::new(addr.replacen("https://", "http://", 1))? + .connect_with_connector(crate::tls::connector(tls.clone())) + .await? + } + }) }) .await .context("master did not complete the connection")? @@ -154,7 +140,7 @@ async fn stream_to( client_addr: &str, tls: Option<&GrpcTls>, ) -> Result> { - let channel = connect(master_endpoint(master, tls)).await?; + let channel = connect(master, tls).await?; // A master answers at once, a follower with its leader hint. let answer_by = tokio::time::Instant::now() + CONNECT_TIMEOUT; let mut client = SeaweedClient::new(channel); @@ -273,6 +259,24 @@ mod tests { } } + /// `tests/tls`: a CA and the `engine-a` leaf it signed. + fn credentials() -> GrpcTls { + let fixture = |name: &str| { + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("tests/tls") + .join(name) + .display() + .to_string() + }; + GrpcTls { + ca_path: fixture("ca.pem"), + cert_path: fixture("a.pem"), + key_path: fixture("a.key"), + allowed_server_names: Vec::new(), + allowed_server_wildcard_domain: String::new(), + } + } + #[tokio::test] async fn a_master_that_stalls_the_tls_handshake_is_bounded() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -281,27 +285,47 @@ mod tests { let (mut conn, _) = listener.accept().await.unwrap(); let _ = tokio::io::copy(&mut conn, &mut tokio::io::sink()).await; }); - weed_lance_worker::tls::install_default_crypto_provider(); - let endpoint = Endpoint::new(format!("https://127.0.0.1:{port}")) - .unwrap() - .tls_config(ClientTlsConfig::new().domain_name("localhost")) - .unwrap(); - match tokio::time::timeout(CONNECT_TIMEOUT * 3, connect(async { Ok(endpoint) })).await { + let master = format!("127.0.0.1:9333.{port}"); + let tls = credentials(); + let open = stream_to(&master, "127.0.0.1:8080", Some(&tls)); + match tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await { Ok(Err(e)) => assert!(format!("{e:#}").contains("did not complete"), "{e:#}"), other => panic!("handshake was not bounded: {other:?}"), } } - #[test] - fn the_tls_name_is_the_dialed_host() { - for (master, host) in [ - ("m1:9333", "m1"), - ("https://master.example", "master.example"), - ("https://master.example:9333", "master.example"), - ("[::1]:9333", "::1"), - ] { - let endpoint = Endpoint::new(grpc_addr(master).unwrap()).unwrap(); - assert_eq!(dialed_host(&endpoint).as_deref(), Some(host), "{master}"); + #[cfg(unix)] + #[tokio::test] + async fn a_credential_read_that_stalls_is_bounded() { + let dir = tempfile::tempdir().unwrap(); + let fifo = dir.path().join("ca.pem"); + let made = std::process::Command::new("mkfifo").arg(&fifo).status(); + assert!(made.unwrap().success()); + // Every connection is dropped at once, so a dial that skipped the read + // fails its handshake instead of timing out. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + while let Ok((conn, _)) = listener.accept().await { + drop(conn); + } + }); + let mut tls = credentials(); + tls.ca_path = fifo.display().to_string(); + let master = format!("127.0.0.1:9333.{port}"); + let open = stream_to(&master, "127.0.0.1:8080", Some(&tls)); + let result = tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await; + // A read+write open of a FIFO does not block, and it lets the + // blocked read finish so the runtime can shut down. + drop( + std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(&fifo), + ); + match result { + Ok(Err(e)) => assert!(format!("{e:#}").contains("did not complete"), "{e:#}"), + other => panic!("credential read was not bounded: {other:?}"), } } diff --git a/seaweed-vector/crates/server/src/security.rs b/seaweed-vector/crates/server/src/security.rs index 9f66f7d47b..7b3b269956 100644 --- a/seaweed-vector/crates/server/src/security.rs +++ b/seaweed-vector/crates/server/src/security.rs @@ -1,6 +1,6 @@ //! The cluster's `security.toml`, found and read the way weed processes do: -//! the file in `.`, `~/.seaweedfs`, `/usr/local/etc/seaweedfs` or -//! `/etc/seaweedfs`, with `WEED_*` environment variables over it. +//! the file in `-config_dir`, `.`, `~/.seaweedfs`, `/usr/local/etc/seaweedfs` +//! or `/etc/seaweedfs`, with `WEED_*` environment variables over it. use std::path::{Path, PathBuf}; @@ -8,10 +8,12 @@ use anyhow::{Context, Result, bail}; const FILE_NAME: &str = "security.toml"; -/// Credentials for dialing a weed gRPC server under TLS, as -/// `security.LoadClientTLS` builds them: the server's certificate must chain -/// to `grpc.ca` and name the host dialed, and this client presents its own. -#[derive(Clone)] +/// What a weed gRPC server is dialed with, as `security.LoadClientTLS` reads +/// it: the files, re-read at every dial, and the server identities allowed. +/// The server's certificate must chain to `grpc.ca`; like weed clients, this +/// one does not check it against the host dialed, only against the allowlist +/// when one is set. +#[derive(Clone, Debug, PartialEq, Eq)] pub struct GrpcTls { /// `grpc.ca` pub ca_path: String, @@ -19,9 +21,12 @@ pub struct GrpcTls { /// `key`. pub cert_path: String, pub key_path: String, - /// The name the server's certificate is checked against, when it is not - /// the host in the server's address. - pub server_name: Option, + /// `allowed_server_commonNames`: when set, the server certificate's common + /// name or a DNS SAN must be one of these or lie under the wildcard + /// domain; a URI SAN must be one of these exactly. + pub allowed_server_names: Vec, + /// `allowed_server_wildcard_domain` + pub allowed_server_wildcard_domain: String, } impl GrpcTls { @@ -41,14 +46,6 @@ impl GrpcTls { } } -impl std::fmt::Debug for GrpcTls { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("GrpcTls") - .field("server_name", &self.server_name) - .finish_non_exhaustive() - } -} - /// What a gRPC listener serves with, as `security.LoadServerTLS` reads it: /// the files, re-read when they change, and the client identities allowed. #[derive(Clone, Debug, PartialEq, Eq)] @@ -78,8 +75,23 @@ impl std::fmt::Debug for Security { } } -fn candidates(current_dir: Option<&Path>, home: Option<&Path>) -> Vec { +// weed's util.ResolvePath for "~" and "~/..." +fn resolve_path(path: &Path, home: Option<&Path>) -> PathBuf { + match (path.strip_prefix("~"), home) { + (Ok(rest), Some(home)) => home.join(rest), + _ => path.to_path_buf(), + } +} + +fn candidates( + config_dir: Option<&Path>, + current_dir: Option<&Path>, + home: Option<&Path>, +) -> Vec { let mut found = Vec::new(); + if let Some(dir) = config_dir { + found.push(resolve_path(dir, home).join(FILE_NAME)); + } if let Some(dir) = current_dir { found.push(dir.join(FILE_NAME)); } @@ -91,6 +103,17 @@ fn candidates(current_dir: Option<&Path>, home: Option<&Path>) -> Vec { found } +/// A comma-separated list such as `allowed_commonNames`, trimmed, without +/// empties. +fn names(list: Option) -> Vec { + list.unwrap_or_default() + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(str::to_string) + .collect() +} + // viper keys are case-insensitive; of two spellings, a set value wins over an // empty one fn entry<'a>(table: &'a toml::Table, key: &str) -> Option<&'a toml::Value> { @@ -105,16 +128,21 @@ fn entry<'a>(table: &'a toml::Table, key: &str) -> Option<&'a toml::Value> { } impl Security { - /// Reads `path`, or the first `security.toml` where weed looks; without - /// one, only the environment applies. - pub fn load(path: Option<&Path>) -> Result { + /// Reads `path`, or the first `security.toml` where weed looks: the + /// `-config_dir`, then `.`, `~/.seaweedfs`, `/usr/local/etc/seaweedfs` + /// and `/etc/seaweedfs`. Without one, only the environment applies. + pub fn load(path: Option<&Path>, config_dir: Option<&Path>) -> Result { let path = match path { Some(path) => Some(path.to_path_buf()), None => { let home = std::env::var_os("HOME").map(PathBuf::from); - candidates(std::env::current_dir().ok().as_deref(), home.as_deref()) - .into_iter() - .find(|p| p.is_file()) + candidates( + config_dir, + std::env::current_dir().ok().as_deref(), + home.as_deref(), + ) + .into_iter() + .find(|p| p.is_file()) } }; Self::from_file(path, Box::new(|name| std::env::var(name).ok())) @@ -180,28 +208,15 @@ impl Security { } } - /// weed checks a dialed server's identity against these on top of - /// `grpc.ca`; the engine's master dial cannot, so it refuses to start - /// rather than accept every master certificate the CA signed. - fn refuse_server_allowlists(&self, keys: &[String]) -> Result<()> { - if let Some(key) = keys.iter().find(|key| self.get(key).is_some()) { - bail!( - "{key} is set, but weed-vector does not check master certificates against it; \ - pass a --security-toml without it to accept any master certificate under grpc.ca" - ); - } - Ok(()) - } - - /// The credentials `component` (such as `grpc.client`) dials servers with, - /// following `security.LoadClientTLS`: a dedicated client certificate is - /// preferred over the component's own, and once either pair is set - /// `grpc.ca` is required. `None` keeps plaintext. - pub fn grpc_client_tls( - &self, - component: &str, - server_name: Option, - ) -> Result> { + /// The credentials `component` (such as `grpc.client`) dials `target` + /// (such as `grpc.master`) with, following `security.LoadClientTLS`: a + /// dedicated client certificate is preferred over the component's own, + /// and once either pair is set `grpc.ca` is required. The server + /// allowlist comes from `._allowed_server_commonNames`, + /// else `.allowed_server_commonNames`, else + /// `grpc.allowed_server_commonNames`, and the wildcard domain likewise. + /// `None` keeps plaintext. + pub fn grpc_client_tls(&self, component: &str, target: &str) -> Result> { let client = ( format!("{component}.client_cert"), format!("{component}.client_key"), @@ -217,20 +232,30 @@ impl Security { let ca = self .get("grpc.ca") .with_context(|| format!("{component} cert and key are set but grpc.ca is not"))?; - self.refuse_server_allowlists(&[ - format!("{component}.allowed_server_commonNames"), - "grpc.allowed_server_commonNames".into(), - format!("{component}.allowed_server_wildcard_domain"), - "grpc.allowed_server_wildcard_domain".into(), - ])?; self.read("grpc.ca", &ca)?; self.read(&format!("{component} cert"), &cert)?; self.read(&format!("{component} key"), &key)?; + let prefix = if target == component { + String::new() + } else { + format!("{}_", target.trim_start_matches("grpc.")) + }; + let first = |suffix: &str| { + [ + format!("{component}.{prefix}{suffix}"), + format!("{component}.{suffix}"), + format!("grpc.{suffix}"), + ] + .iter() + .find_map(|key| self.get(key)) + }; Ok(Some(GrpcTls { ca_path: ca, cert_path: cert, key_path: key, - server_name, + allowed_server_names: names(first("allowed_server_commonNames")), + allowed_server_wildcard_domain: first("allowed_server_wildcard_domain") + .unwrap_or_default(), })) } @@ -256,14 +281,7 @@ impl Security { cert_path, key_path, ca_path, - allowed_common_names: self - .get(&allowed) - .unwrap_or_default() - .split(',') - .map(str::trim) - .filter(|name| !name.is_empty()) - .map(str::to_string) - .collect(), + allowed_common_names: names(self.get(&allowed)), allowed_wildcard_domain: self.get("grpc.allowed_wildcard_domain").unwrap_or_default(), })) } @@ -346,7 +364,7 @@ mod tests { let (_dir, ca, cert, key, client) = files(); assert!( security("", &[]) - .grpc_client_tls("grpc.client", None) + .grpc_client_tls("grpc.client", "grpc.master") .unwrap() .is_none() ); @@ -355,14 +373,23 @@ mod tests { &format!("[grpc]\nca = {ca:?}\n[grpc.client]\ncert = {cert:?}\nkey = {key:?}\n"), &[], ) - .grpc_client_tls("grpc.client", Some("master.internal".into())) + .grpc_client_tls("grpc.client", "grpc.master") .unwrap() .unwrap(); + assert_eq!( + tls, + GrpcTls { + ca_path: ca.clone(), + cert_path: cert.clone(), + key_path: key.clone(), + allowed_server_names: Vec::new(), + allowed_server_wildcard_domain: String::new(), + } + ); assert_eq!( tls.pems().await.unwrap(), (b"CA".to_vec(), b"CERT".to_vec(), b"KEY".to_vec()) ); - assert_eq!(tls.server_name.as_deref(), Some("master.internal")); // Each dial reads the files, so a rotated pair is picked up. std::fs::write(&cert, "ROTATED").unwrap(); assert_eq!(tls.pems().await.unwrap().1, b"ROTATED"); @@ -376,7 +403,7 @@ mod tests { ("WEED_GRPC_CLIENT_CLIENT_KEY", &key), ], ) - .grpc_client_tls("grpc.client", None) + .grpc_client_tls("grpc.client", "grpc.master") .unwrap() .unwrap(); assert_eq!(tls.pems().await.unwrap().1, b"CLIENT"); @@ -385,24 +412,54 @@ mod tests { format!("[grpc.client]\ncert = {cert:?}\nkey = {key:?}\n"), format!("[grpc]\nca = {ca:?}\n[grpc.client]\ncert = {cert:?}\n"), format!("[grpc]\nca = {ca:?}\n[grpc.client]\nclient_key = {key:?}\n"), - format!( - "[grpc]\nca = {ca:?}\nallowed_server_wildcard_domain = \".internal\"\n\ - [grpc.client]\ncert = {cert:?}\nkey = {key:?}\n" - ), - format!( - "[grpc]\nca = {ca:?}\n[grpc.client]\ncert = {cert:?}\nkey = {key:?}\n\ - allowed_server_commonNames = \"master\"\n" - ), ] { assert!( security(&broken, &[]) - .grpc_client_tls("grpc.client", None) + .grpc_client_tls("grpc.client", "grpc.master") .is_err(), "{broken}" ); } } + #[test] + fn server_allowlists_fall_back_from_target_to_component_to_grpc() { + let (_dir, ca, cert, key, _client) = files(); + let base = format!("[grpc.client]\ncert = {cert:?}\nkey = {key:?}\n"); + let load = |grpc: &str, client_extra: &str| { + security( + &format!("[grpc]\nca = {ca:?}\n{grpc}{base}{client_extra}"), + &[], + ) + .grpc_client_tls("grpc.client", "grpc.master") + .unwrap() + .unwrap() + }; + let tls = load( + "allowed_server_commonNames = \"any\"\nallowed_server_wildcard_domain = \".any\"\n", + "", + ); + assert_eq!(tls.allowed_server_names, ["any"]); + assert_eq!(tls.allowed_server_wildcard_domain, ".any"); + let tls = load( + "allowed_server_commonNames = \"any\"\n", + "allowed_server_commonNames = \"client, wide\"\n", + ); + assert_eq!(tls.allowed_server_names, ["client", "wide"]); + let tls = load( + "allowed_server_commonNames = \"any\"\n", + "allowed_server_commonNames = \"client\"\nmaster_allowed_server_commonNames = \"master\"\n", + ); + assert_eq!(tls.allowed_server_names, ["master"]); + assert!(tls.allowed_server_wildcard_domain.is_empty()); + // An empty spelling falls through, as firstConfigString skips "". + let tls = load( + "allowed_server_commonNames = \"any\"\n", + "master_allowed_server_commonNames = \"\"\n", + ); + assert_eq!(tls.allowed_server_names, ["any"]); + } + #[test] fn server_identity_needs_both_halves_and_the_ca() { let (_dir, ca, cert, key, _client) = files(); @@ -456,10 +513,15 @@ mod tests { #[test] fn the_file_is_searched_where_weed_looks() { - let found = candidates(Some(Path::new("/cwd")), Some(Path::new("/home/u"))); + let found = candidates( + Some(Path::new("/etc/weed")), + Some(Path::new("/cwd")), + Some(Path::new("/home/u")), + ); assert_eq!( found, [ + "/etc/weed/security.toml", "/cwd/security.toml", "/home/u/.seaweedfs/security.toml", "/usr/local/etc/seaweedfs/security.toml", @@ -467,5 +529,32 @@ mod tests { ] .map(PathBuf::from) ); + let home = Some(Path::new("/home/u")); + for (dir, want) in [ + ("~", "/home/u/security.toml"), + ("~/cluster", "/home/u/cluster/security.toml"), + ("~other/cluster", "~other/cluster/security.toml"), + ("/srv/~/cluster", "/srv/~/cluster/security.toml"), + ] { + assert_eq!( + candidates(Some(Path::new(dir)), None, home)[0], + PathBuf::from(want), + "{dir}" + ); + } + } + + #[test] + fn load_takes_the_config_dir_before_the_usual_places() { + let dir = tempfile::tempdir().unwrap(); + let in_config_dir = dir.path().join(FILE_NAME); + std::fs::write(&in_config_dir, "").unwrap(); + let loaded = Security::load(None, Some(dir.path())).unwrap(); + assert_eq!(loaded.path(), Some(in_config_dir.as_path())); + // An explicit file still wins. + let explicit = dir.path().join("other.toml"); + std::fs::write(&explicit, "").unwrap(); + let loaded = Security::load(Some(&explicit), Some(dir.path())).unwrap(); + assert_eq!(loaded.path(), Some(explicit.as_path())); } } diff --git a/seaweed-vector/crates/server/src/tls.rs b/seaweed-vector/crates/server/src/tls.rs index 30bc876e26..e5b8eaf827 100644 --- a/seaweed-vector/crates/server/src/tls.rs +++ b/seaweed-vector/crates/server/src/tls.rs @@ -1,8 +1,15 @@ -//! The engine's gRPC listener under TLS, run the way weed servers run -//! theirs: every client must present a certificate chaining to `grpc.ca`, +//! gRPC TLS the way weed processes do it. +//! +//! Serving: every client must present a certificate chaining to `grpc.ca`, //! `allowed_commonNames` and `allowed_wildcard_domain` restrict which client //! identities are served, and the certificate, key and CA files are re-read //! on a timer and on SIGHUP so rotated files take effect without a restart. +//! +//! Dialing: the server's certificate must chain to `grpc.ca`; it is not +//! checked against the host dialed (weed clients verify the chain only, as +//! `advancedtls.CertVerification`), and `allowed_server_commonNames` / +//! `allowed_server_wildcard_domain` restrict which server identities are +//! accepted. The files are re-read at every dial. use std::collections::HashSet; use std::sync::{Arc, Mutex, RwLock}; @@ -10,21 +17,26 @@ use std::time::Duration; use anyhow::{Context, Result, bail}; use futures::Stream; -use rustls::DistinguishedName; -use rustls::RootCertStore; -use rustls::ServerConfig; -use rustls::client::danger::HandshakeSignatureValid; +use http::Uri; +use hyper_util::rt::TokioIo; +use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}; +use rustls::client::verify_server_cert_signed_by_trust_anchor; +use rustls::crypto::{CryptoProvider, aws_lc_rs}; use rustls::pki_types::pem::PemObject; -use rustls::pki_types::{CertificateDer, PrivateKeyDer, UnixTime}; -use rustls::server::WebPkiClientVerifier; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}; use rustls::server::danger::{ClientCertVerified, ClientCertVerifier}; -use rustls::{DigitallySignedStruct, SignatureScheme}; +use rustls::server::{ParsedCertificate, WebPkiClientVerifier}; +use rustls::{ + ClientConfig, DigitallySignedStruct, DistinguishedName, RootCertStore, ServerConfig, + SignatureScheme, +}; use tokio::net::{TcpListener, TcpStream}; -use tokio_rustls::TlsAcceptor; use tokio_rustls::server::TlsStream; +use tokio_rustls::{TlsAcceptor, TlsConnector}; +use x509_parser::extensions::GeneralName; use x509_parser::prelude::{FromDer, X509Certificate}; -use crate::security::GrpcServerTls; +use crate::security::{GrpcServerTls, GrpcTls}; /// `WEED_TLS_CERT_REFRESH_INTERVAL`, the cadence weed's PEM providers re-read /// certificate files at; a Go duration such as `30m` or `500ms`. @@ -42,38 +54,82 @@ pub fn refresh_interval() -> Duration { .unwrap_or(DEFAULT_REFRESH_INTERVAL) } -/// `time.ParseDuration`'s grammar: an optional `+`, then a sequence of -/// ``. A negative value is `None`, as weed ignores one. +/// `time.ParseDuration`, with its arithmetic: an optional `+`, then a +/// sequence of ``, summed in whole nanoseconds (a fraction of +/// one is dropped) up to `i64::MAX`. A negative value is `None`, as weed +/// ignores one. fn parse_go_duration(s: &str) -> Option { - let s = s.trim(); + const LIMIT: u64 = 1 << 63; + let digits = |s: &str| s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len()); let mut rest = s.strip_prefix('+').unwrap_or(s); + if rest == "0" { + return Some(Duration::ZERO); + } if rest.is_empty() { return None; } - let mut total = Duration::ZERO; + let mut total: u64 = 0; while !rest.is_empty() { - let digits = rest - .find(|c: char| !c.is_ascii_digit() && c != '.') - .unwrap_or(rest.len()); - let (number, tail) = rest.split_at(digits); - let value: f64 = number.parse().ok()?; + let (whole, tail) = rest.split_at(digits(rest)); + let mut v: u64 = 0; + for c in whole.bytes() { + if v > LIMIT / 10 { + return None; + } + v = v * 10 + u64::from(c - b'0'); + if v > LIMIT { + return None; + } + } + let (mut fraction, mut scale) = (0u64, 1f64); + let (fraction_digits, tail) = match tail.strip_prefix('.') { + Some(tail) => tail.split_at(digits(tail)), + None => ("", tail), + }; + let mut overflow = false; + for c in fraction_digits.bytes() { + if overflow || fraction > (LIMIT - 1) / 10 { + overflow = true; + continue; + } + let next = fraction * 10 + u64::from(c - b'0'); + if next > LIMIT { + overflow = true; + continue; + } + fraction = next; + scale *= 10.0; + } + if whole.is_empty() && fraction_digits.is_empty() { + return None; + } let unit_len = tail - .find(|c: char| c.is_ascii_digit() || c == '.') + .find(|c: char| c == '.' || c.is_ascii_digit()) .unwrap_or(tail.len()); let (unit, tail) = tail.split_at(unit_len); - let seconds = match unit { - "h" => 3600.0, - "m" => 60.0, - "s" => 1.0, - "ms" => 1e-3, - "us" | "\u{b5}s" | "\u{3bc}s" => 1e-6, - "ns" => 1e-9, + let unit: u64 = match unit { + "ns" => 1, + "us" | "\u{b5}s" | "\u{3bc}s" => 1_000, + "ms" => 1_000_000, + "s" => 1_000_000_000, + "m" => 60_000_000_000, + "h" => 3_600_000_000_000, _ => return None, }; - total = total.checked_add(Duration::try_from_secs_f64(value * seconds).ok()?)?; + if v > LIMIT / unit { + return None; + } + v *= unit; + if fraction > 0 { + v += (fraction as f64 * (unit as f64 / scale)) as u64; + if v > LIMIT { + return None; + } + } + total = total.checked_add(v).filter(|&total| total <= LIMIT)?; rest = tail; } - Some(total) + (total < LIMIT).then(|| Duration::from_nanos(total)) } /// The listener's TLS configuration, rebuilt from the files whenever they @@ -93,6 +149,10 @@ impl std::fmt::Debug for ServerTls { } } +fn read(key: &str, path: &str) -> Result> { + std::fs::read(path).with_context(|| format!("{key} = {path:?}")) +} + fn read_all(files: &GrpcServerTls) -> Result> { let mut all = Vec::new(); for (key, path) in [ @@ -100,36 +160,43 @@ fn read_all(files: &GrpcServerTls) -> Result> { ("key", &files.key_path), ("grpc.ca", &files.ca_path), ] { - all.extend(std::fs::read(path).with_context(|| format!("{key} = {path:?}"))?); + all.extend(read(key, path)?); all.push(0); } Ok(all) } +fn cert_chain(key: &str, path: &str, pem: &[u8]) -> Result>> { + let chain: Vec> = CertificateDer::pem_slice_iter(pem) + .collect::>() + .with_context(|| format!("{key} = {path:?}"))?; + if chain.is_empty() { + bail!("{key} = {path:?} holds no certificate"); + } + Ok(chain) +} + +fn private_key(key: &str, path: &str, pem: &[u8]) -> Result> { + PrivateKeyDer::from_pem_slice(pem).with_context(|| format!("{key} = {path:?}")) +} + +fn root_store(path: &str, pem: &[u8]) -> Result { + let mut roots = RootCertStore::empty(); + for cert in cert_chain("grpc.ca", path, pem)? { + roots + .add(cert) + .with_context(|| format!("grpc.ca = {path:?}"))?; + } + Ok(roots) +} + /// Builds the configuration `security.LoadServerTLS` would: the identity in /// `cert`/`key`, every client verified against `grpc.ca`, and the client's /// common name checked against the allowlist when one is set. fn build_config(files: &GrpcServerTls) -> Result { - let cert_chain: Vec> = CertificateDer::pem_file_iter(&files.cert_path) - .with_context(|| format!("cert = {:?}", files.cert_path))? - .collect::>() - .with_context(|| format!("cert = {:?}", files.cert_path))?; - if cert_chain.is_empty() { - bail!("cert = {:?} holds no certificate", files.cert_path); - } - let key = PrivateKeyDer::from_pem_file(&files.key_path) - .with_context(|| format!("key = {:?}", files.key_path))?; - let mut roots = RootCertStore::empty(); - for cert in CertificateDer::pem_file_iter(&files.ca_path) - .with_context(|| format!("grpc.ca = {:?}", files.ca_path))? - { - roots - .add(cert.with_context(|| format!("grpc.ca = {:?}", files.ca_path))?) - .with_context(|| format!("grpc.ca = {:?}", files.ca_path))?; - } - if roots.is_empty() { - bail!("grpc.ca = {:?} holds no certificate", files.ca_path); - } + let cert_chain = cert_chain("cert", &files.cert_path, &read("cert", &files.cert_path)?)?; + let key = private_key("key", &files.key_path, &read("key", &files.key_path)?)?; + let roots = root_store(&files.ca_path, &read("grpc.ca", &files.ca_path)?)?; let verifier = WebPkiClientVerifier::builder(Arc::new(roots)) .build() .context("client certificate verifier")?; @@ -327,6 +394,172 @@ impl ClientCertVerifier for CommonNameVerifier { } } +/// The client configuration `security.LoadClientTLS` would build, from the +/// CA, certificate and key as `pems` holds them. +fn client_config( + files: &GrpcTls, + (ca, cert, key): (Vec, Vec, Vec), +) -> Result { + weed_lance_worker::tls::install_default_crypto_provider(); + let verifier = ServerIdentityVerifier { + roots: Arc::new(root_store(&files.ca_path, &ca)?), + provider: Arc::new(aws_lc_rs::default_provider()), + allowed_names: files.allowed_server_names.iter().cloned().collect(), + allowed_wildcard_domain: files.allowed_server_wildcard_domain.clone(), + }; + let mut config = ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(verifier)) + .with_client_auth_cert( + cert_chain("cert", &files.cert_path, &cert)?, + private_key("key", &files.key_path, &key)?, + ) + .context("client certificate")?; + config.alpn_protocols = vec![b"h2".to_vec()]; + Ok(config) +} + +/// The host of a gRPC address, without an IPv6 address's brackets. +fn dialed_host(uri: &Uri) -> Option<&str> { + Some(uri.host()?.trim_matches(|c| c == '[' || c == ']')) +} + +/// A tonic connector for `Endpoint::connect_with_connector`: the files read +/// as they are now, so rotated credentials are picked up, then TCP to the +/// URI's host and port and the handshake under [`client_config`]. +pub fn connector( + files: GrpcTls, +) -> impl tower::Service< + Uri, + Response = TokioIo>, + Error = Box, + Future = impl Send, +> + Clone ++ Send ++ 'static { + tower::service_fn(move |uri: Uri| { + let files = files.clone(); + async move { + let host = dialed_host(&uri).ok_or("no host in gRPC address")?; + let port = uri.port_u16().unwrap_or(443); + let config = Arc::new(client_config(&files, files.pems().await?)?); + let stream = TcpStream::connect((host, port)).await?; + let _ = stream.set_nodelay(true); + let name = ServerName::try_from(host.to_string())?; + let stream = TlsConnector::from(config).connect(name, stream).await?; + Ok(TokioIo::new(stream)) + } + }) +} + +/// `security.Authenticator.AuthenticateServer`: the chain must reach +/// `grpc.ca`, the host dialed is not checked, and when an allowlist is set the +/// common name or a DNS SAN must be allowed or lie under the wildcard domain, +/// while a URI SAN must be allowed exactly. +#[derive(Debug)] +struct ServerIdentityVerifier { + roots: Arc, + provider: Arc, + allowed_names: HashSet, + allowed_wildcard_domain: String, +} + +impl ServerCertVerifier for ServerIdentityVerifier { + fn verify_server_cert( + &self, + end_entity: &CertificateDer<'_>, + intermediates: &[CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp_response: &[u8], + now: UnixTime, + ) -> Result { + let cert = ParsedCertificate::try_from(end_entity)?; + verify_server_cert_signed_by_trust_anchor( + &cert, + &self.roots, + intermediates, + now, + self.provider.signature_verification_algorithms.all, + )?; + if self.allowed_names.is_empty() && self.allowed_wildcard_domain.is_empty() { + return Ok(ServerCertVerified::assertion()); + } + let (names, uris) = server_identities(end_entity) + .map_err(|e| rustls::Error::General(format!("parse server certificate failed: {e}")))?; + if names.iter().any(|name| { + self.allowed_names.contains(name) + || wildcard_domain_matches(name, &self.allowed_wildcard_domain) + }) || uris.iter().any(|uri| self.allowed_names.contains(uri)) + { + return Ok(ServerCertVerified::assertion()); + } + Err(rustls::Error::General(format!( + "AuthenticateServer: untrusted server identity {:?}", + names.into_iter().chain(uris).collect::>() + ))) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls12_signature( + message, + cert, + dss, + &self.provider.signature_verification_algorithms, + ) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls13_signature( + message, + cert, + dss, + &self.provider.signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + self.provider + .signature_verification_algorithms + .supported_schemes() + } +} + +/// The common name and DNS SANs, and the URI SANs, of a certificate. +fn server_identities(cert: &CertificateDer<'_>) -> Result<(Vec, Vec)> { + let (_, certificate) = X509Certificate::from_der(cert.as_ref()) + .map_err(|e| anyhow::anyhow!("parse X.509 certificate failed: {e}"))?; + let mut names = vec![ + certificate + .subject() + .iter_common_name() + .next() + .and_then(|common_name| common_name.as_str().ok()) + .unwrap_or_default() + .to_string(), + ]; + let mut uris = Vec::new(); + if let Ok(Some(san)) = certificate.subject_alternative_name() { + for name in &san.value.general_names { + match name { + GeneralName::DNSName(dns) => names.push(dns.to_string()), + GeneralName::URI(uri) => uris.push(uri.to_string()), + _ => {} + } + } + } + Ok((names, uris)) +} + fn parse_common_name(cert: &CertificateDer<'_>) -> Result { let (_, certificate) = X509Certificate::from_der(cert.as_ref()) .map_err(|e| anyhow::anyhow!("parse X.509 certificate failed: {e}"))?; @@ -521,6 +754,129 @@ mod tests { assert_eq!(parse_common_name(&leaf("b.pem")).unwrap(), "engine-b"); } + #[test] + fn server_identities_follow_the_allowlist() { + weed_lance_worker::tls::install_default_crypto_provider(); + let leaf = |name: &str| CertificateDer::from_pem_file(fixture(name)).unwrap(); + let verifier = |allowed: &[&str], wildcard: &str| ServerIdentityVerifier { + roots: Arc::new( + root_store("ca.pem", &std::fs::read(fixture("ca.pem")).unwrap()).unwrap(), + ), + provider: Arc::new(aws_lc_rs::default_provider()), + allowed_names: allowed.iter().map(|s| s.to_string()).collect(), + allowed_wildcard_domain: wildcard.into(), + }; + let name = ServerName::try_from("nothing.example").unwrap(); + let check = |v: &ServerIdentityVerifier, cert: &str| { + v.verify_server_cert(&leaf(cert), &[], &name, &[], UnixTime::now()) + }; + // The chain is checked; the host dialed is not. + assert!(check(&verifier(&[], ""), "a.pem").is_ok()); + assert!(check(&verifier(&[], ""), "b.pem").is_ok()); + assert!( + check(&verifier(&[], ""), "ca.pem").is_err(), + "a root is not a server" + ); + // The common name or a DNS SAN must be allowed. + assert!(check(&verifier(&["engine-a"], ""), "a.pem").is_ok()); + assert!(check(&verifier(&["localhost"], ""), "b.pem").is_ok()); + let err = check(&verifier(&["engine-a"], ""), "b.pem").unwrap_err(); + assert!(err.to_string().contains("engine-b"), "{err}"); + assert!(check(&verifier(&[], ".example"), "b.pem").is_err()); + assert_eq!( + server_identities(&leaf("a.pem")).unwrap(), + ( + vec!["engine-a".to_string(), "localhost".to_string()], + vec![] + ) + ); + } + + #[test] + fn the_dialed_host_drops_ipv6_brackets() { + for (addr, host) in [ + ("http://m1:19333", "m1"), + ("http://master.example", "master.example"), + ("http://127.0.0.1:19333", "127.0.0.1"), + ("http://[::1]:19333", "::1"), + ] { + let uri: Uri = addr.parse().unwrap(); + assert_eq!(dialed_host(&uri), Some(host), "{addr}"); + } + } + + fn client(leaf: &str, allowed: &[&str]) -> GrpcTls { + GrpcTls { + ca_path: fixture("ca.pem").display().to_string(), + cert_path: fixture(&format!("{leaf}.pem")).display().to_string(), + key_path: fixture(&format!("{leaf}.key")).display().to_string(), + allowed_server_names: allowed.iter().map(|s| s.to_string()).collect(), + allowed_server_wildcard_domain: String::new(), + } + } + + async fn handshake(port: u16, files: &GrpcTls) -> Result<()> { + let config = Arc::new(client_config(files, files.pems().await?)?); + let stream = TcpStream::connect(("127.0.0.1", port)).await?; + let name = ServerName::try_from("127.0.0.1").unwrap(); + let mut stream = TlsConnector::from(config).connect(name, stream).await?; + // A server that rejected the client certificate says so after the + // handshake, on the first read. + let mut byte = [0u8; 1]; + match tokio::time::timeout( + Duration::from_secs(1), + tokio::io::AsyncReadExt::read(&mut stream, &mut byte), + ) + .await + { + Ok(Err(err)) => Err(err.into()), + _ => Ok(()), + } + } + + #[tokio::test] + async fn both_sides_check_identities() { + use futures::StreamExt; + let dir = tempfile::tempdir().unwrap(); + let mut files = install(dir.path(), "a"); + files.allowed_common_names = vec!["engine-b".into()]; + let tls = ServerTls::load(files).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut incoming = Box::pin(tls.incoming(listener)); + + // engine-b, trusting the CA, accepts engine-a by name, dialed by IP. + let allowed = client("b", &["engine-a"]); + let (served, dialed) = tokio::join!( + tokio::time::timeout(Duration::from_secs(5), incoming.next()), + handshake(port, &allowed) + ); + assert!(dialed.is_ok(), "{dialed:?}"); + assert!(matches!(served, Ok(Some(Ok(_))))); + + // The client refuses a server outside its allowlist. + let err = handshake(port, &client("b", &["engine-z"])) + .await + .unwrap_err(); + assert!( + err.to_string().contains("untrusted server identity"), + "{err}" + ); + + // The server refuses a client outside its allowlist: engine-a is not + // engine-b. + let outsider = client("a", &[]); + let (served, dialed) = tokio::join!( + tokio::time::timeout(Duration::from_secs(1), incoming.next()), + handshake(port, &outsider) + ); + assert!( + dialed.is_err(), + "the server must cut off a client it does not allow" + ); + assert!(served.is_err(), "nothing is served for a refused client"); + } + #[test] fn wildcard_domain_matches_like_go() { let none = HashSet::new(); @@ -564,5 +920,29 @@ mod tests { ); assert_eq!(parse_go_duration("2d"), None); assert_eq!(parse_go_duration(""), None); + assert_eq!(parse_go_duration("0"), Some(Duration::ZERO)); + assert_eq!(parse_go_duration("0.3s"), Some(Duration::from_millis(300))); + assert_eq!(parse_go_duration("1.1s"), Some(Duration::from_millis(1100))); + assert_eq!(parse_go_duration(".5h"), Some(Duration::from_secs(1800))); + assert_eq!(parse_go_duration("5.s"), Some(Duration::from_secs(5))); + assert_eq!(parse_go_duration(".s"), None); + assert_eq!(parse_go_duration("1.2.3s"), None); + assert_eq!(parse_go_duration(" 30m"), None); + // A fraction of a nanosecond is dropped, so this is zero and the + // default applies, as in weed. + assert_eq!(parse_go_duration("0.9ns"), Some(Duration::ZERO)); + assert_eq!(parse_go_duration("1.9ns"), Some(Duration::from_nanos(1))); + // The longest time.Duration, and past it. + assert_eq!( + parse_go_duration("2562047h47m16.854775807s"), + Some(Duration::from_nanos(i64::MAX as u64)) + ); + assert_eq!( + parse_go_duration("9223372036854775807ns"), + Some(Duration::from_nanos(i64::MAX as u64)) + ); + assert_eq!(parse_go_duration("9223372036854775808ns"), None); + assert_eq!(parse_go_duration("2562048h"), None); + assert_eq!(parse_go_duration("3000000h"), None); } } diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index 33a18a553e..c60ccf1fa1 100644 --- a/test/s3vectors/conftest.py +++ b/test/s3vectors/conftest.py @@ -109,7 +109,10 @@ def write_tls(directory): "-out", "outsider.pem", "-days", "2", "-extfile", "ext.cnf") cert, key, ca = (f'{key} = {json.dumps(str(directory / name), ensure_ascii=False)}\n' for key, name in (("cert", "cert.pem"), ("key", "key.pem"), ("ca", "ca.pem"))) pem = cert + key + ca - grpc = "".join(f"[grpc.{component}]\n{cert}{key}\n" for component in ("master", "volume", "filer", "s3", "client")) + grpc = "".join(f"[grpc.{component}]\n{cert}{key}\n" for component in ("master", "volume", "filer", "s3")) + # Clients (the gateway, the engine) accept only servers named localhost; + # the engine serves only clients named localhost. + grpc += f"[grpc.client]\n{cert}{key}allowed_server_commonNames = \"localhost\"\n" grpc += f"[grpc.vector]\n{cert}{key}allowed_commonNames = \"localhost\"\n" (directory / "security.toml").write_text(f"[https.volume]\n{pem}\n[https.client]\nenabled = true\n{pem}\n[grpc]\n{ca}\n{grpc}")