From eeb8aefe706c5d60a83b5499c71e0e830c574438 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 11:42:49 -0800 Subject: [PATCH] weed-vector: gRPC TLS to the masters and for the gateway (#12368) * weed-vector: read the cluster's security.toml The engine finds security.toml where weed processes do (., ~/.seaweedfs, /usr/local/etc/seaweedfs, /etc/seaweedfs) with WEED_* environment variables over it, and builds gRPC client and server credentials by the rules of security.LoadClientTLS and LoadServerTLS: a dedicated client certificate is preferred, both halves of a pair must be set, and grpc.ca is required once either is. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * weed-vector: gRPC TLS to the masters and from the gateway The masters are dialed with [grpc.client]'s credentials when the cluster has them, the master's certificate checked against the host dialed or --tls-server-name. The engine's own listener takes its identity from [grpc.vector] and then requires a client certificate chaining to grpc.ca, which is how the gateway, dialing with [grpc.client], reaches it. The rustls provider is pinned before the first TLS use, as the other binaries linking both aws-lc-rs and ring do. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * scaffold: document [grpc.vector] Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/s3vectors: put every gRPC hop under mutual TLS in the hardened leg VECTOR_TLS=1 now also writes [grpc.master], [grpc.volume], [grpc.filer], [grpc.s3], [grpc.client] and [grpc.vector], and probes the master's and the engine's gRPC ports: plaintext HTTP/2 and a client without a certificate are refused, the client certificate gets a SETTINGS frame. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/s3vectors: give the test volume server 64 volume slots Every collection grows 7 volumes and test_vector_bucket_encryption opens two buckets while the previous test's bucket is still being torn down, so 16 slots were marginal: the Go legs failed CreateIndex now and then with "failed to find writable volumes" and master.log showing "create 7 volume, created 4: Not enough data nodes found!". Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * weed-vector: reload the listener's certificates and enforce allowed_commonNames The listener now runs the way weed servers run theirs. The certificate, key and CA files are re-read every WEED_TLS_CERT_REFRESH_INTERVAL (weed's default of 5h) and on SIGHUP, and new handshakes use the changed files without a restart; a file that no longer parses is logged and the working configuration stays. [grpc.vector] allowed_commonNames and grpc.allowed_wildcard_domain restrict which client certificates are served, checked against the subject common name as security.Authenticator and the Rust volume server's CommonNameVerifier do. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * scaffold: allowed_commonNames and the refresh cadence for [grpc.vector] Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * test/s3vectors: the engine's allowlist keeps an outsider certificate out The hardened leg sets [grpc.vector] allowed_commonNames = "localhost" and probes the engine with a second certificate the same CA signed for CN=outsider: the handshake must be refused, while the gateway's certificate still gets an HTTP/2 SETTINGS frame. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * weed-vector: take the master's TLS name from the parsed endpoint A master given without a port (https://master.example) had the whole URL checked against its certificate. Co-Authored-By: Claude Opus 5.5 * weed-vector: bound the master TLS handshake connect_timeout covers only the TCP connect, so a master that accepted the connection and never answered the handshake held up the other masters. Co-Authored-By: Claude Opus 5.5 * weed-vector: look up security.toml keys the way viper does An empty WEED_* variable no longer hides the file's value, and keys match regardless of case, as for weed processes. Co-Authored-By: Claude Opus 5.5 * weed-vector: refuse master allowlists it cannot enforce weed checks a dialed server against allowed_server_commonNames and allowed_server_wildcard_domain on top of grpc.ca. The engine's master dial accepted every certificate the CA signed even when one was set; it now fails at startup and names the key. Co-Authored-By: Claude Opus 5.5 * test/s3vectors: read the whole frame header in the gRPC TLS probe Co-Authored-By: Claude Opus 5.5 * test/s3vectors: probe the volume, filer and S3 gRPC listeners too Co-Authored-By: Claude Opus 5.5 * weed-vector: read the master dial credentials only when given masters A standalone engine never dials with grpc.client, so its settings no longer stop it from starting. Co-Authored-By: Claude Opus 5.5 * weed-vector: run each listener TLS handshake on its own task The accept loop awaited every handshake before accepting the next connection, so a peer that connected and sent nothing stopped every gateway connection after it. Handshakes now run concurrently, each under a 10s timeout. Co-Authored-By: Claude Opus 5.5 * weed-vector: read the master dial's TLS files at every dial The client certificate, key and CA were read once at startup, so after the files rotated a reconnect still presented the old pair. Each dial now reads them, as the listener re-reads its own. Co-Authored-By: Claude Opus 5.5 * weed-vector: read every unit Go durations accept in the refresh interval WEED_TLS_CERT_REFRESH_INTERVAL values in ns, us or a leading + fell back to five hours in the engine while weed servers honored them. Co-Authored-By: Claude Opus 5.5 * weed-vector: fall back on a refresh interval too long to hold An oversized WEED_TLS_CERT_REFRESH_INTERVAL panicked the engine at startup; weed servers reject it and keep the default. Co-Authored-By: Claude Opus 5.5 * weed-vector: refuse allowed_commonNames without the listener's certificate With [grpc.vector] allowed_commonNames set but no cert and key, the listener served plaintext and checked no client at all. Co-Authored-By: Claude Opus 5.5 * weed-vector: read the master dial's credentials inside its connect timeout The CA, certificate and key were read synchronously before the timeout started, so a stalled credential path held up every other master. Co-Authored-By: Claude Opus 5.5 * weed-vector: let a set key win over an empty spelling of it With allowed_server_commonNames = "" beside allowed_server_commonnames = "master", the lookup stopped at the empty exact match and the allowlist went unnoticed. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Fable 5.1 --- .github/workflows/rust-vector-tests.yml | 5 +- seaweed-vector/Cargo.lock | 190 +++++- seaweed-vector/crates/server/Cargo.toml | 4 + seaweed-vector/crates/server/src/lib.rs | 2 + seaweed-vector/crates/server/src/main.rs | 69 ++- .../crates/server/src/master_client.rs | 95 ++- seaweed-vector/crates/server/src/security.rs | 471 +++++++++++++++ seaweed-vector/crates/server/src/tls.rs | 568 ++++++++++++++++++ seaweed-vector/crates/server/tests/tls/a.key | 3 + seaweed-vector/crates/server/tests/tls/a.pem | 11 + seaweed-vector/crates/server/tests/tls/b.key | 3 + seaweed-vector/crates/server/tests/tls/b.pem | 11 + seaweed-vector/crates/server/tests/tls/ca.pem | 10 + test/s3vectors/conftest.py | 65 +- weed/command/scaffold/security.toml | 10 + 15 files changed, 1484 insertions(+), 33 deletions(-) create mode 100644 seaweed-vector/crates/server/src/security.rs create mode 100644 seaweed-vector/crates/server/src/tls.rs create mode 100644 seaweed-vector/crates/server/tests/tls/a.key create mode 100644 seaweed-vector/crates/server/tests/tls/a.pem create mode 100644 seaweed-vector/crates/server/tests/tls/b.key create mode 100644 seaweed-vector/crates/server/tests/tls/b.pem create mode 100644 seaweed-vector/crates/server/tests/tls/ca.pem diff --git a/.github/workflows/rust-vector-tests.yml b/.github/workflows/rust-vector-tests.yml index ac6371a89a..e00dc2f12b 100644 --- a/.github/workflows/rust-vector-tests.yml +++ b/.github/workflows/rust-vector-tests.yml @@ -83,8 +83,9 @@ jobs: fail-fast: false matrix: volume: [go, rust] - # Hardened: volume servers behind mutual TLS, read JWTs required, - # and every chunk encrypted by the gateway. + # Hardened: volume servers behind mutual TLS, every gRPC hop (master, + # volume, filer, gateway, engine) under mutual TLS, read JWTs + # required, and every chunk encrypted by the gateway. hardened: ["", "1"] env: CARGO_PROFILE_DEV_DEBUG: "0" diff --git a/seaweed-vector/Cargo.lock b/seaweed-vector/Cargo.lock index da09df9035..9b2bc8e4d1 100644 --- a/seaweed-vector/Cargo.lock +++ b/seaweed-vector/Cargo.lock @@ -348,6 +348,45 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "asn1-rs" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5493c3bedbacf7fd7382c6346bbd66687d12bbaad3a89a2d2c303ee6cf20b048" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom 7.1.3", + "num-traits", + "rusticata-macros", + "thiserror 1.0.69", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "965c2d33e53cb6b267e148a4cb0760bc01f4904c1cd4bb4002a085bb016d1490" +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]] name = "async-channel" version = "2.5.0" @@ -1468,6 +1507,12 @@ dependencies = [ "parking_lot_core", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "datafusion" version = "54.1.0" @@ -2106,6 +2151,20 @@ dependencies = [ "thiserror 2.0.21", ] +[[package]] +name = "der-parser" +version = "9.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5cd0a5c643689626bec213c4d8bd4d96acc8ffdb4ad4bb6bc16abf27d5f4b553" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom 7.1.3", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.9" @@ -3146,7 +3205,7 @@ dependencies = [ "fast-float2", "itoa", "jiff", - "nom", + "nom 8.0.0", "num-traits", "ordered-float", "rand 0.9.5", @@ -3969,6 +4028,12 @@ dependencies = [ "unicase", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.9.1" @@ -4037,6 +4102,16 @@ dependencies = [ "rawpointer", ] +[[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 = "nom" version = "8.0.0" @@ -4164,6 +4239,15 @@ dependencies = [ "tokio", ] +[[package]] +name = "oid-registry" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8d8034d9489cdaf79228eb9f6a3b8d7bb32ba00d6645ebd48eef4077ceb5bd9" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -5159,6 +5243,15 @@ dependencies = [ "semver", ] +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "rustix" version = "1.1.5" @@ -5452,6 +5545,15 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -5739,6 +5841,17 @@ dependencies = [ "futures-core", ] +[[package]] +name = "synstructure" +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" version = "0.14.0" @@ -5953,6 +6066,47 @@ dependencies = [ "tokio", ] +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap 2.14.2", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + [[package]] name = "tonic" version = "0.14.6" @@ -6499,11 +6653,14 @@ dependencies = [ "prost-types", "protoc-bin-vendored", "reqwest 0.12.28", + "rustls", "serde", "serde_json", "tempfile", "tokio", + "tokio-rustls", "tokio-stream", + "toml", "tonic", "tonic-prost", "tonic-prost-build", @@ -6511,6 +6668,7 @@ dependencies = [ "tracing-subscriber", "weed-lance-worker", "weed-vector-core", + "x509-parser", ] [[package]] @@ -6663,6 +6821,15 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winnow" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" +dependencies = [ + "memchr", +] + [[package]] name = "wit-bindgen" version = "0.57.1" @@ -6684,6 +6851,23 @@ dependencies = [ "tap", ] +[[package]] +name = "x509-parser" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcbc162f30700d6f3f82a24bf7cc62ffe7caea42c0b2cba8bf7f3ae50cf51f69" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom 7.1.3", + "oid-registry", + "rusticata-macros", + "thiserror 1.0.69", + "time", +] + [[package]] name = "xmlparser" version = "0.13.6" @@ -6716,7 +6900,7 @@ dependencies = [ "proc-macro2", "quote", "syn 3.0.9", - "synstructure", + "synstructure 0.14.0", ] [[package]] @@ -6757,7 +6941,7 @@ dependencies = [ "proc-macro2", "quote", "syn 3.0.9", - "synstructure", + "synstructure 0.14.0", ] [[package]] diff --git a/seaweed-vector/crates/server/Cargo.toml b/seaweed-vector/crates/server/Cargo.toml index 8d65d590d1..30084008dd 100644 --- a/seaweed-vector/crates/server/Cargo.toml +++ b/seaweed-vector/crates/server/Cargo.toml @@ -35,6 +35,10 @@ moka = { version = "0.12", features = ["sync"] } object_store = "0.13" serde = { version = "1", features = ["derive"] } serde_json = "1" +toml = "0.8" +rustls = "0.23" +tokio-rustls = "0.26" +x509-parser = "0.16" tokio.workspace = true tokio-stream.workspace = true tracing.workspace = true diff --git a/seaweed-vector/crates/server/src/lib.rs b/seaweed-vector/crates/server/src/lib.rs index 84db0b59cd..074dcdc697 100644 --- a/seaweed-vector/crates/server/src/lib.rs +++ b/seaweed-vector/crates/server/src/lib.rs @@ -13,6 +13,8 @@ pub mod master_pb { tonic::include_proto!("master_pb"); } pub mod master_client; +pub mod security; +pub mod tls; pub mod pb { pub mod vector { diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index 42bbc9ca3b..c883a4c87d 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -1,12 +1,16 @@ +use std::path::PathBuf; + use anyhow::{Context, Result}; use clap::Parser; +use tokio_stream::wrappers::TcpListenerStream; use tonic::transport::Server; -use tonic::transport::server::TcpIncoming; use weed_lance_worker::catalog::{NamespaceClient, SigningKey}; use weed_lance_worker::dataset::FallbackOptions; use weed_vector_server::pb::vector::vector_service_server::VectorServiceServer; use weed_vector_server::read_cache::ReadCache; +use weed_vector_server::security::Security; use weed_vector_server::service::VectorServer; +use weed_vector_server::tls::{self, ServerTls}; /// The VectorService the S3 gateway fans vector calls out to. #[derive(Parser)] @@ -36,6 +40,15 @@ struct Args { /// credentials that write index data only to a caller holding it. #[arg(long, env = "WEED_FILER_SIGNING_KEY")] 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. + #[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, } fn parse_option(s: &str) -> Result<(String, String), String> { @@ -105,11 +118,24 @@ fn advertised_addr(listen: &str, masters: &[String]) -> String { 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())?; + 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())?, + None => None, + }; + let server_tls = security + .grpc_server_tls("grpc.vector")? + .map(ServerTls::load) + .transpose()?; // Bind before registering, so the master never lists an engine that is // about to exit because its port is taken. - let incoming = TcpIncoming::bind(args.listen.parse()?) - .with_context(|| format!("listen on {}", args.listen))? - .with_nodelay(Some(true)); + let listener = tokio::net::TcpListener::bind(&args.listen) + .await + .with_context(|| format!("listen on {}", args.listen))?; if let Some(masters) = &args.masters { let masters: Vec = masters.split(',').map(|s| s.trim().to_string()).collect(); for master in &masters { @@ -117,7 +143,9 @@ async fn main() -> Result<()> { } let advertise = advertised_addr(&args.listen, &masters); tokio::spawn(weed_vector_server::master_client::keep_connected( - masters, advertise, + masters, + advertise, + master_tls.clone(), )); } let fallback: FallbackOptions = args.storage_options.into_iter().collect(); @@ -129,16 +157,37 @@ async fn main() -> Result<()> { if args.read_cache_mb > 0 { service = service.with_read_cache(ReadCache::new(args.read_cache_mb << 20)); } - tracing::info!(namespace = %args.namespace, listen = %args.listen, "weed-vector"); - Server::builder() + tracing::info!( + namespace = %args.namespace, + listen = %args.listen, + "weed-vector; serving gRPC {}, dialing masters {}", + match &server_tls { + Some(tls) if tls.files().allowed_common_names.is_empty() + && tls.files().allowed_wildcard_domain.is_empty() => + "over TLS with client certificates required", + Some(_) => "over TLS with allowed client certificates required", + None => "in plaintext", + }, + if master_tls.is_some() { "over TLS" } else { "in plaintext" }, + ); + let router = Server::builder() // Match the gateway's 1 GiB cap — put batches over the tonic 4 MiB // default otherwise get rejected before they reach the handler. .add_service( VectorServiceServer::new(service) .max_decoding_message_size(1 << 30) .max_encoding_message_size(1 << 30), - ) - .serve_with_incoming(incoming) - .await?; + ); + match server_tls { + Some(tls) => { + tokio::spawn(tls.clone().refresh(tls::refresh_interval())); + router.serve_with_incoming(tls.incoming(listener)).await?; + } + None => { + router + .serve_with_incoming(TcpListenerStream::new(listener)) + .await? + } + } Ok(()) } diff --git a/seaweed-vector/crates/server/src/master_client.rs b/seaweed-vector/crates/server/src/master_client.rs index a1c357beda..2ce8e383c2 100644 --- a/seaweed-vector/crates/server/src/master_client.rs +++ b/seaweed-vector/crates/server/src/master_client.rs @@ -7,9 +7,10 @@ use std::time::Duration; use anyhow::{Context, Result}; use tokio::time::sleep; -use tonic::transport::Endpoint; +use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity}; use crate::master_pb::{KeepConnectedRequest, seaweed_client::SeaweedClient}; +use crate::security::GrpcTls; fn registration(client_addr: &str) -> KeepConnectedRequest { KeepConnectedRequest { @@ -54,6 +55,32 @@ 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 { @@ -68,11 +95,12 @@ fn http_addr(master: &str) -> String { /// Retry forever: try each master until a stream opens, hold it until it /// errors, then rediscover — same shape as wdclient.KeepConnectedToMaster. -pub async fn keep_connected(masters: Vec, client_addr: String) { +pub async fn keep_connected(masters: Vec, client_addr: String, tls: Option) { let client_addr = client_addr.as_str(); + let tls = tls.as_ref(); loop { try_all_masters(&masters, |master| async move { - stream_to(&master, client_addr).await + stream_to(&master, client_addr, tls).await }) .await; sleep(Duration::from_secs(3)).await; @@ -108,16 +136,25 @@ where } } -/// Bounds the TCP connect and then the master's first answer, the 5s wdclient -/// gives a master call; a black-holed or silent master otherwise holds up the -/// other masters. +/// Bounds the credential reads and the connect, TLS handshake included, and +/// then the master's first answer, the 5s wdclient gives a master call; a +/// black-holed or silent master otherwise holds up the other masters. const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); -async fn stream_to(master: &str, client_addr: &str) -> Result> { - let channel = Endpoint::new(grpc_addr(master)?)? - .connect_timeout(CONNECT_TIMEOUT) - .connect() - .await?; +async fn connect(endpoint: impl Future>) -> Result { + tokio::time::timeout(CONNECT_TIMEOUT, async { + Ok(endpoint.await?.connect().await?) + }) + .await + .context("master did not complete the connection")? +} + +async fn stream_to( + master: &str, + client_addr: &str, + tls: Option<&GrpcTls>, +) -> Result> { + let channel = connect(master_endpoint(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); @@ -215,7 +252,7 @@ mod tests { #[tokio::test] async fn dial_to_a_black_holed_master_is_bounded() { // TEST-NET-1 (RFC 5737) is reserved, so nothing answers the SYN. - let dial = stream_to("192.0.2.1:9333", "127.0.0.1:8080"); + let dial = stream_to("192.0.2.1:9333", "127.0.0.1:8080", None); let result = tokio::time::timeout(CONNECT_TIMEOUT * 3, dial).await; assert!(matches!(result, Ok(Err(_))), "dial was not bounded"); } @@ -229,13 +266,45 @@ mod tests { let _ = tokio::io::copy(&mut conn, &mut tokio::io::sink()).await; }); let master = format!("127.0.0.1:9333.{port}"); - let open = stream_to(&master, "127.0.0.1:8080"); + let open = stream_to(&master, "127.0.0.1:8080", None); match tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await { Ok(Err(e)) => assert!(format!("{e:#}").contains("did not answer"), "{e:#}"), other => panic!("open was not bounded: {other:?}"), } } + #[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(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + 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 { + 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}"); + } + } + #[test] fn grpc_addr_refuses_a_port_past_65535() { assert_eq!(grpc_addr("m1:9333").unwrap(), "http://m1:19333"); diff --git a/seaweed-vector/crates/server/src/security.rs b/seaweed-vector/crates/server/src/security.rs new file mode 100644 index 0000000000..9f66f7d47b --- /dev/null +++ b/seaweed-vector/crates/server/src/security.rs @@ -0,0 +1,471 @@ +//! 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. + +use std::path::{Path, PathBuf}; + +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)] +pub struct GrpcTls { + /// `grpc.ca` + pub ca_path: String, + /// The component's `client_cert` and `client_key`, else its `cert` and + /// `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, +} + +impl GrpcTls { + /// The CA, certificate and key as the files hold them now, so a dial + /// after the files rotate presents the new pair. + pub async fn pems(&self) -> Result<(Vec, Vec, Vec)> { + async fn read(path: &str) -> Result> { + tokio::fs::read(path) + .await + .with_context(|| format!("reading {path}")) + } + Ok(( + read(&self.ca_path).await?, + read(&self.cert_path).await?, + read(&self.key_path).await?, + )) + } +} + +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)] +pub struct GrpcServerTls { + pub cert_path: String, + pub key_path: String, + /// `grpc.ca`, which every client certificate must chain to. + pub ca_path: String, + /// `.allowed_commonNames`: when set, the client certificate's + /// common name must be one of these, or lie under the wildcard domain. + pub allowed_common_names: Vec, + /// `grpc.allowed_wildcard_domain` + pub allowed_wildcard_domain: String, +} + +pub struct Security { + path: Option, + table: toml::Table, + env: Box Option + Send + Sync>, +} + +impl std::fmt::Debug for Security { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Security") + .field("path", &self.path) + .finish_non_exhaustive() + } +} + +fn candidates(current_dir: Option<&Path>, home: Option<&Path>) -> Vec { + let mut found = Vec::new(); + if let Some(dir) = current_dir { + found.push(dir.join(FILE_NAME)); + } + if let Some(home) = home { + found.push(home.join(".seaweedfs").join(FILE_NAME)); + } + found.push(PathBuf::from("/usr/local/etc/seaweedfs").join(FILE_NAME)); + found.push(PathBuf::from("/etc/seaweedfs").join(FILE_NAME)); + found +} + +// 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> { + let set = |v: &&toml::Value| v.as_str() != Some(""); + table.get(key).filter(set).or_else(|| { + table + .iter() + .filter(|(k, _)| k.eq_ignore_ascii_case(key)) + .map(|(_, v)| v) + .find(set) + }) +} + +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 { + 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()) + } + }; + Self::from_file(path, Box::new(|name| std::env::var(name).ok())) + } + + fn from_file( + path: Option, + env: Box Option + Send + Sync>, + ) -> Result { + let table = match &path { + Some(path) => std::fs::read_to_string(path) + .with_context(|| format!("reading {}", path.display()))? + .parse::() + .with_context(|| format!("parsing {}", path.display()))?, + None => toml::Table::new(), + }; + Ok(Self { path, table, env }) + } + + pub fn path(&self) -> Option<&Path> { + self.path.as_deref() + } + + fn raw(&self, key: &str) -> Option { + let env_name = format!("WEED_{}", key.to_uppercase().replace('.', "_")); + // viper treats an empty variable as unset + if let Some(value) = (self.env)(&env_name).filter(|v| !v.is_empty()) { + return Some(value); + } + let mut parts = key.split('.'); + let mut value = entry(&self.table, parts.next()?)?; + for part in parts { + value = entry(value.as_table()?, part)?; + } + match value { + toml::Value::String(s) => Some(s.clone()), + toml::Value::Boolean(b) => Some(b.to_string()), + toml::Value::Integer(i) => Some(i.to_string()), + _ => None, + } + } + + /// A dotted key such as `grpc.ca`, from `WEED_GRPC_CA` when set, else + /// the file; empty counts as unset. + pub fn get(&self, key: &str) -> Option { + self.raw(key).filter(|v| !v.is_empty()) + } + + pub fn get_bool(&self, key: &str) -> bool { + self.raw(key).is_some_and(|v| v.trim() == "true") + } + + /// The file a key names, read whole. + pub fn read(&self, key: &str, path: &str) -> Result> { + std::fs::read(path).with_context(|| format!("{key} = {path:?}")) + } + + fn pair(&self, cert: &str, key: &str) -> Result> { + match (self.get(cert), self.get(key)) { + (Some(cert), Some(key)) => Ok(Some((cert, key))), + (None, None) => Ok(None), + _ => bail!("{cert} and {key} must both be set"), + } + } + + /// 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> { + let client = ( + format!("{component}.client_cert"), + format!("{component}.client_key"), + ); + let own = (format!("{component}.cert"), format!("{component}.key")); + let (cert, key) = match self.pair(&client.0, &client.1)? { + Some(pair) => pair, + None => match self.pair(&own.0, &own.1)? { + Some(pair) => pair, + None => return Ok(None), + }, + }; + 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)?; + Ok(Some(GrpcTls { + ca_path: ca, + cert_path: cert, + key_path: key, + server_name, + })) + } + + /// The identity `component` (such as `grpc.vector`) serves with, following + /// `security.LoadServerTLS`: its `cert` and `key`, with `grpc.ca` as the + /// root every client certificate must chain to, and the component's + /// `allowed_commonNames`. `None` keeps plaintext. The files are read by + /// [`crate::tls::ServerTls`], which keeps re-reading them. + pub fn grpc_server_tls(&self, component: &str) -> Result> { + let allowed = format!("{component}.allowed_commonNames"); + let Some((cert_path, key_path)) = + self.pair(&format!("{component}.cert"), &format!("{component}.key"))? + else { + if self.get(&allowed).is_some() { + bail!("{allowed} is set but {component}.cert and {component}.key are not"); + } + return Ok(None); + }; + let ca_path = self + .get("grpc.ca") + .with_context(|| format!("{component} cert and key are set but grpc.ca is not"))?; + Ok(Some(GrpcServerTls { + 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_wildcard_domain: self.get("grpc.allowed_wildcard_domain").unwrap_or_default(), + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + fn security(toml: &str, env: &[(&str, &str)]) -> Security { + let env: HashMap = env + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(); + Security { + path: None, + table: toml.parse().unwrap(), + env: Box::new(move |name| env.get(name).cloned()), + } + } + + /// Placeholder credential files: the loader reads them, it does not + /// parse them. + fn files() -> (tempfile::TempDir, String, String, String, String) { + let dir = tempfile::tempdir().unwrap(); + let mut paths = Vec::new(); + for (name, body) in [ + ("ca.pem", "CA"), + ("cert.pem", "CERT"), + ("key.pem", "KEY"), + ("client.pem", "CLIENT"), + ] { + let path = dir.path().join(name); + std::fs::write(&path, body).unwrap(); + paths.push(path.display().to_string()); + } + let [ca, cert, key, client]: [String; 4] = paths.try_into().unwrap(); + (dir, ca, cert, key, client) + } + + #[test] + fn the_environment_wins_over_the_file() { + let s = security( + "[jwt.signing.read]\nkey = \"from-file\"\n[https.client]\nenabled = true\n", + &[("WEED_JWT_SIGNING_READ_KEY", "from-env")], + ); + assert_eq!(s.get("jwt.signing.read.key").as_deref(), Some("from-env")); + assert!(s.get_bool("https.client.enabled")); + let s = security("[jwt.signing.read]\nkey = \"from-file\"\n", &[]); + assert_eq!(s.get("jwt.signing.read.key").as_deref(), Some("from-file")); + let s = security( + "[jwt.signing.read]\nkey = \"from-file\"\n", + &[("WEED_JWT_SIGNING_READ_KEY", "")], + ); + assert_eq!(s.get("jwt.signing.read.key").as_deref(), Some("from-file")); + let s = security("[grpc.vector]\nallowed_commonnames = \"s3\"\n", &[]); + assert_eq!( + s.get("grpc.vector.allowed_commonNames").as_deref(), + Some("s3") + ); + let s = security( + "[grpc.client]\nallowed_server_commonNames = \"\"\nallowed_server_commonnames = \"master\"\n", + &[], + ); + assert_eq!( + s.get("grpc.client.allowed_server_commonNames").as_deref(), + Some("master") + ); + assert!(security("", &[]).get("jwt.signing.read.key").is_none()); + assert!( + security("[grpc]\nca = \"\"\n", &[]) + .get("grpc.ca") + .is_none() + ); + } + + #[tokio::test] + async fn client_credentials_follow_the_gateways_rules() { + let (_dir, ca, cert, key, client) = files(); + assert!( + security("", &[]) + .grpc_client_tls("grpc.client", None) + .unwrap() + .is_none() + ); + + let tls = security( + &format!("[grpc]\nca = {ca:?}\n[grpc.client]\ncert = {cert:?}\nkey = {key:?}\n"), + &[], + ) + .grpc_client_tls("grpc.client", Some("master.internal".into())) + .unwrap() + .unwrap(); + 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"); + + // A dedicated client certificate is preferred, and the environment + // can supply it. + let tls = security( + &format!("[grpc]\nca = {ca:?}\n[grpc.client]\ncert = {cert:?}\nkey = {key:?}\n"), + &[ + ("WEED_GRPC_CLIENT_CLIENT_CERT", &client), + ("WEED_GRPC_CLIENT_CLIENT_KEY", &key), + ], + ) + .grpc_client_tls("grpc.client", None) + .unwrap() + .unwrap(); + assert_eq!(tls.pems().await.unwrap().1, b"CLIENT"); + + for broken in [ + 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) + .is_err(), + "{broken}" + ); + } + } + + #[test] + fn server_identity_needs_both_halves_and_the_ca() { + let (_dir, ca, cert, key, _client) = files(); + assert!( + security("", &[]) + .grpc_server_tls("grpc.vector") + .unwrap() + .is_none() + ); + let tls = security( + &format!("[grpc]\nca = {ca:?}\n[grpc.vector]\ncert = {cert:?}\nkey = {key:?}\n"), + &[], + ) + .grpc_server_tls("grpc.vector") + .unwrap() + .unwrap(); + assert_eq!( + tls, + GrpcServerTls { + cert_path: cert.clone(), + key_path: key.clone(), + ca_path: ca.clone(), + allowed_common_names: Vec::new(), + allowed_wildcard_domain: String::new(), + } + ); + let tls = security( + &format!( + "[grpc]\nca = {ca:?}\nallowed_wildcard_domain = \".internal\"\n[grpc.vector]\ncert = {cert:?}\nkey = {key:?}\nallowed_commonNames = \"gateway, s3 ,\"\n" + ), + &[], + ) + .grpc_server_tls("grpc.vector") + .unwrap() + .unwrap(); + assert_eq!(tls.allowed_common_names, ["gateway", "s3"]); + assert_eq!(tls.allowed_wildcard_domain, ".internal"); + for broken in [ + format!("[grpc.vector]\ncert = {cert:?}\nkey = {key:?}\n"), + format!("[grpc]\nca = {ca:?}\n[grpc.vector]\nkey = {key:?}\n"), + format!("[grpc]\nca = {ca:?}\n[grpc.vector]\nallowed_commonNames = \"s3\"\n"), + ] { + assert!( + security(&broken, &[]) + .grpc_server_tls("grpc.vector") + .is_err(), + "{broken}" + ); + } + } + + #[test] + fn the_file_is_searched_where_weed_looks() { + let found = candidates(Some(Path::new("/cwd")), Some(Path::new("/home/u"))); + assert_eq!( + found, + [ + "/cwd/security.toml", + "/home/u/.seaweedfs/security.toml", + "/usr/local/etc/seaweedfs/security.toml", + "/etc/seaweedfs/security.toml" + ] + .map(PathBuf::from) + ); + } +} diff --git a/seaweed-vector/crates/server/src/tls.rs b/seaweed-vector/crates/server/src/tls.rs new file mode 100644 index 0000000000..30bc876e26 --- /dev/null +++ b/seaweed-vector/crates/server/src/tls.rs @@ -0,0 +1,568 @@ +//! The engine's gRPC listener under TLS, run the way weed servers run +//! theirs: 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. + +use std::collections::HashSet; +use std::sync::{Arc, Mutex, RwLock}; +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 rustls::pki_types::pem::PemObject; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, UnixTime}; +use rustls::server::WebPkiClientVerifier; +use rustls::server::danger::{ClientCertVerified, ClientCertVerifier}; +use rustls::{DigitallySignedStruct, SignatureScheme}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_rustls::TlsAcceptor; +use tokio_rustls::server::TlsStream; +use x509_parser::prelude::{FromDer, X509Certificate}; + +use crate::security::GrpcServerTls; + +/// `WEED_TLS_CERT_REFRESH_INTERVAL`, the cadence weed's PEM providers re-read +/// certificate files at; a Go duration such as `30m` or `500ms`. +pub const REFRESH_INTERVAL_ENV: &str = "WEED_TLS_CERT_REFRESH_INTERVAL"; +const DEFAULT_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 3600); +const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10); + +/// How often the files are re-read: the environment's value, else weed's +/// default of five hours. +pub fn refresh_interval() -> Duration { + std::env::var(REFRESH_INTERVAL_ENV) + .ok() + .and_then(|s| parse_go_duration(&s)) + .filter(|d| !d.is_zero()) + .unwrap_or(DEFAULT_REFRESH_INTERVAL) +} + +/// `time.ParseDuration`'s grammar: an optional `+`, then a sequence of +/// ``. A negative value is `None`, as weed ignores one. +fn parse_go_duration(s: &str) -> Option { + let s = s.trim(); + let mut rest = s.strip_prefix('+').unwrap_or(s); + if rest.is_empty() { + return None; + } + let mut total = Duration::ZERO; + 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 unit_len = tail + .find(|c: char| c.is_ascii_digit() || c == '.') + .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, + _ => return None, + }; + total = total.checked_add(Duration::try_from_secs_f64(value * seconds).ok()?)?; + rest = tail; + } + Some(total) +} + +/// The listener's TLS configuration, rebuilt from the files whenever they +/// change. +pub struct ServerTls { + files: GrpcServerTls, + config: RwLock>, + /// The file contents the current configuration was built from. + loaded: Mutex>, +} + +impl std::fmt::Debug for ServerTls { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ServerTls") + .field("files", &self.files) + .finish_non_exhaustive() + } +} + +fn read_all(files: &GrpcServerTls) -> Result> { + let mut all = Vec::new(); + for (key, path) in [ + ("cert", &files.cert_path), + ("key", &files.key_path), + ("grpc.ca", &files.ca_path), + ] { + all.extend(std::fs::read(path).with_context(|| format!("{key} = {path:?}"))?); + all.push(0); + } + Ok(all) +} + +/// 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 verifier = WebPkiClientVerifier::builder(Arc::new(roots)) + .build() + .context("client certificate verifier")?; + let verifier: Arc = + if files.allowed_common_names.is_empty() && files.allowed_wildcard_domain.is_empty() { + verifier + } else { + Arc::new(CommonNameVerifier { + inner: verifier, + allowed_common_names: files.allowed_common_names.iter().cloned().collect(), + allowed_wildcard_domain: files.allowed_wildcard_domain.clone(), + }) + }; + let mut config = ServerConfig::builder() + .with_client_cert_verifier(verifier) + .with_single_cert(cert_chain, key) + .context("server certificate")?; + config.alpn_protocols = vec![b"h2".to_vec()]; + Ok(config) +} + +impl ServerTls { + pub fn load(files: GrpcServerTls) -> Result> { + weed_lance_worker::tls::install_default_crypto_provider(); + let loaded = read_all(&files)?; + let config = build_config(&files)?; + Ok(Arc::new(Self { + files, + config: RwLock::new(Arc::new(config)), + loaded: Mutex::new(loaded), + })) + } + + pub fn files(&self) -> &GrpcServerTls { + &self.files + } + + /// Re-reads the files; when they changed, the next handshake uses them. + /// Returns whether they had. A file that no longer parses is an error, + /// and the configuration in use stays. + pub fn reload(&self) -> Result { + let current = read_all(&self.files)?; + let mut loaded = self.loaded.lock().unwrap_or_else(|e| e.into_inner()); + if *loaded == current { + return Ok(false); + } + let config = build_config(&self.files)?; + *self.config.write().unwrap_or_else(|e| e.into_inner()) = Arc::new(config); + *loaded = current; + Ok(true) + } + + pub fn acceptor(&self) -> TlsAcceptor { + TlsAcceptor::from( + self.config + .read() + .unwrap_or_else(|e| e.into_inner()) + .clone(), + ) + } + + /// Connections accepted on `listener` and taken through the TLS + /// handshake with the configuration current at that moment. Each + /// handshake runs on its own task under `HANDSHAKE_TIMEOUT`, so a peer + /// that connects and stays silent holds up nobody else; a failed + /// handshake is logged and skipped, as the Rust volume server does. + pub fn incoming( + self: &Arc, + listener: TcpListener, + ) -> impl Stream>> + Send + 'static { + let tls = self.clone(); + let (tx, rx) = tokio::sync::mpsc::channel(64); + tokio::spawn(async move { + loop { + let accepted = tokio::select! { + _ = tx.closed() => return, + accepted = listener.accept() => accepted, + }; + let (stream, peer) = match accepted { + Ok(accepted) => accepted, + Err(err) => { + let _ = tx.send(Err(err)).await; + continue; + } + }; + let _ = stream.set_nodelay(true); + let acceptor = tls.acceptor(); + let tx = tx.clone(); + tokio::spawn(async move { + match tokio::time::timeout(HANDSHAKE_TIMEOUT, acceptor.accept(stream)).await { + Ok(Ok(stream)) => { + let _ = tx.send(Ok(stream)).await; + } + Ok(Err(err)) => tracing::debug!("gRPC TLS handshake from {peer}: {err}"), + Err(_) => tracing::debug!("gRPC TLS handshake from {peer} timed out"), + } + }); + } + }); + tokio_stream::wrappers::ReceiverStream::new(rx) + } + + /// Re-reads the files every `every`, and at once on SIGHUP. + pub async fn refresh(self: Arc, every: Duration) { + #[cfg(unix)] + let mut hangup = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup()).ok(); + loop { + #[cfg(unix)] + { + tokio::select! { + _ = tokio::time::sleep(every) => {} + _ = async { match hangup.as_mut() { Some(h) => h.recv().await, None => std::future::pending().await } } => { + tracing::info!("SIGHUP: re-reading gRPC TLS files"); + } + } + } + #[cfg(not(unix))] + tokio::time::sleep(every).await; + match self.reload() { + Ok(true) => tracing::info!("gRPC TLS files changed; new handshakes use them"), + Ok(false) => {} + Err(err) => tracing::warn!("gRPC TLS files not reloaded: {err:#}"), + } + } + } +} + +/// `security.Authenticator` for client certificates: after the chain is +/// verified, the subject common name must be allowed, or lie under the +/// wildcard domain. +#[derive(Debug)] +struct CommonNameVerifier { + inner: Arc, + allowed_common_names: HashSet, + allowed_wildcard_domain: String, +} + +impl ClientCertVerifier for CommonNameVerifier { + fn offer_client_auth(&self) -> bool { + self.inner.offer_client_auth() + } + + fn client_auth_mandatory(&self) -> bool { + self.inner.client_auth_mandatory() + } + + fn root_hint_subjects(&self) -> &[DistinguishedName] { + self.inner.root_hint_subjects() + } + + fn verify_client_cert( + &self, + end_entity: &CertificateDer<'_>, + intermediates: &[CertificateDer<'_>], + now: UnixTime, + ) -> Result { + self.inner + .verify_client_cert(end_entity, intermediates, now)?; + let common_name = parse_common_name(end_entity).map_err(|e| { + rustls::Error::General(format!("parse client certificate common name failed: {e}")) + })?; + if common_name_is_allowed( + &common_name, + &self.allowed_common_names, + &self.allowed_wildcard_domain, + ) { + return Ok(ClientCertVerified::assertion()); + } + Err(rustls::Error::General(format!( + "Authenticate: invalid subject client common name: {common_name}" + ))) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + self.inner.verify_tls12_signature(message, cert, dss) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &DigitallySignedStruct, + ) -> Result { + self.inner.verify_tls13_signature(message, cert, dss) + } + + fn supported_verify_schemes(&self) -> Vec { + self.inner.supported_verify_schemes() + } +} + +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}"))?; + Ok(certificate + .subject() + .iter_common_name() + .next() + .and_then(|common_name| common_name.as_str().ok()) + .map(str::to_string) + .unwrap_or_default()) +} + +fn common_name_is_allowed( + common_name: &str, + allowed_common_names: &HashSet, + allowed_wildcard_domain: &str, +) -> bool { + allowed_common_names.contains(common_name) + || wildcard_domain_matches(common_name, allowed_wildcard_domain) +} + +// Mirrors the Go wildcard check: the name must end with the domain, carry a +// non-empty label in front of it, and be a valid DNS name so a CA-signed name +// like ".example.com" cannot satisfy the suffix check. A leading "*." label +// is allowed so wildcard certs keep matching. One trailing root dot is +// ignored on both sides so absolute DNS names still match. +fn wildcard_domain_matches(name: &str, wildcard_domain: &str) -> bool { + let name = name.strip_suffix('.').unwrap_or(name); + let wildcard_domain = wildcard_domain.strip_suffix('.').unwrap_or(wildcard_domain); + if wildcard_domain.is_empty() + || name.len() <= wildcard_domain.len() + || !name.ends_with(wildcard_domain) + { + return false; + } + if !wildcard_domain.starts_with('.') + && name.as_bytes()[name.len() - wildcard_domain.len() - 1] != b'.' + { + return false; + } + is_domain_name(name.strip_prefix("*.").unwrap_or(name)) +} + +// Mirrors Go's net.IsDomainName: non-empty labels of letters, digits, '-', +// '_', each up to 63 chars and not '-'-edged, with at least one non-digit so +// a bare IP cannot pass a domain wildcard. +fn is_domain_name(name: &str) -> bool { + if name.is_empty() || name.len() > 253 { + return false; + } + let mut non_numeric = false; + let valid = name.split('.').all(|label| { + non_numeric |= label.bytes().any(|c| !c.is_ascii_digit()); + !label.is_empty() + && label.len() <= 63 + && !label.starts_with('-') + && !label.ends_with('-') + && label + .bytes() + .all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_') + }); + valid && non_numeric +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::{Path, PathBuf}; + + /// `tests/tls`: a CA and two leaves it signed, `engine-a` and `engine-b`. + fn fixture(name: &str) -> PathBuf { + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("tests/tls") + .join(name) + } + + fn install(dir: &Path, leaf: &str) -> GrpcServerTls { + for (from, to) in [ + (format!("{leaf}.pem"), "cert.pem"), + (format!("{leaf}.key"), "key.pem"), + ("ca.pem".to_string(), "ca.pem"), + ] { + std::fs::copy(fixture(&from), dir.join(to)).unwrap(); + } + GrpcServerTls { + cert_path: dir.join("cert.pem").display().to_string(), + key_path: dir.join("key.pem").display().to_string(), + ca_path: dir.join("ca.pem").display().to_string(), + allowed_common_names: Vec::new(), + allowed_wildcard_domain: String::new(), + } + } + + #[test] + fn rotated_files_are_picked_up_and_broken_ones_are_not() { + let dir = tempfile::tempdir().unwrap(); + let tls = ServerTls::load(install(dir.path(), "a")).unwrap(); + assert!(!tls.reload().unwrap(), "unchanged files are not a reload"); + let before = tls.acceptor(); + + install(dir.path(), "b"); + assert!(tls.reload().unwrap()); + let after = tls.acceptor(); + assert!(!Arc::ptr_eq(before.config(), after.config())); + + std::fs::write(dir.path().join("key.pem"), "not a key").unwrap(); + assert!(tls.reload().is_err()); + assert!( + Arc::ptr_eq(after.config(), tls.acceptor().config()), + "a broken file leaves the working configuration in place" + ); + } + + #[tokio::test] + async fn a_silent_peer_does_not_hold_up_the_next_handshake() { + use futures::StreamExt; + weed_lance_worker::tls::install_default_crypto_provider(); + let dir = tempfile::tempdir().unwrap(); + let tls = ServerTls::load(install(dir.path(), "a")).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let mut incoming = Box::pin(tls.incoming(listener)); + let _silent = TcpStream::connect(addr).await.unwrap(); + + let mut roots = RootCertStore::empty(); + for cert in CertificateDer::pem_file_iter(fixture("ca.pem")).unwrap() { + roots.add(cert.unwrap()).unwrap(); + } + let config = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_client_auth_cert( + vec![CertificateDer::from_pem_file(fixture("a.pem")).unwrap()], + PrivateKeyDer::from_pem_file(fixture("a.key")).unwrap(), + ) + .unwrap(); + let client = tokio::spawn(async move { + let tcp = TcpStream::connect(addr).await.unwrap(); + tokio_rustls::TlsConnector::from(Arc::new(config)) + .connect("localhost".try_into().unwrap(), tcp) + .await + }); + let served = tokio::time::timeout(Duration::from_secs(5), incoming.next()).await; + assert!( + matches!(served, Ok(Some(Ok(_)))), + "the handshake waited behind a silent peer" + ); + client.await.unwrap().unwrap(); + } + + #[test] + fn missing_or_empty_files_are_refused() { + let dir = tempfile::tempdir().unwrap(); + let mut files = install(dir.path(), "a"); + files.ca_path = dir.path().join("absent.pem").display().to_string(); + assert!(ServerTls::load(files.clone()).is_err()); + std::fs::write(dir.path().join("empty.pem"), "").unwrap(); + files.ca_path = dir.path().join("empty.pem").display().to_string(); + assert!(ServerTls::load(files).is_err()); + } + + #[test] + fn client_common_names_follow_the_allowlist() { + weed_lance_worker::tls::install_default_crypto_provider(); + let mut roots = RootCertStore::empty(); + for cert in CertificateDer::pem_file_iter(fixture("ca.pem")).unwrap() { + roots.add(cert.unwrap()).unwrap(); + } + let inner = WebPkiClientVerifier::builder(Arc::new(roots)) + .build() + .unwrap(); + let leaf = |name: &str| CertificateDer::from_pem_file(fixture(name)).unwrap(); + let now = UnixTime::now(); + + // Without an allowlist any certificate the CA signed is served. + assert!(inner.verify_client_cert(&leaf("a.pem"), &[], now).is_ok()); + assert!(inner.verify_client_cert(&leaf("b.pem"), &[], now).is_ok()); + + let verifier = CommonNameVerifier { + inner, + allowed_common_names: ["engine-a".to_string()].into_iter().collect(), + allowed_wildcard_domain: String::new(), + }; + assert!( + verifier + .verify_client_cert(&leaf("a.pem"), &[], now) + .is_ok() + ); + let err = verifier + .verify_client_cert(&leaf("b.pem"), &[], now) + .unwrap_err(); + assert!(err.to_string().contains("engine-b"), "{err}"); + assert_eq!(parse_common_name(&leaf("b.pem")).unwrap(), "engine-b"); + } + + #[test] + fn wildcard_domain_matches_like_go() { + let none = HashSet::new(); + assert!(common_name_is_allowed("a.internal", &none, ".internal")); + assert!(common_name_is_allowed("a.internal", &none, "internal")); + assert!(common_name_is_allowed("*.internal", &none, ".internal")); + assert!(!common_name_is_allowed("internal", &none, ".internal")); + assert!(!common_name_is_allowed("xinternal", &none, "internal")); + assert!(!common_name_is_allowed("10.0.0.1", &none, "0.0.1")); + assert!(!common_name_is_allowed("a.internal", &none, "")); + let names: HashSet = ["gateway".to_string()].into_iter().collect(); + assert!(common_name_is_allowed("gateway", &names, "")); + assert!(!common_name_is_allowed("other", &names, "")); + } + + #[test] + fn refresh_interval_reads_go_durations() { + assert_eq!(parse_go_duration("30m"), Some(Duration::from_secs(1800))); + assert_eq!(parse_go_duration("500ms"), Some(Duration::from_millis(500))); + assert_eq!(parse_go_duration("1h30m"), Some(Duration::from_secs(5400))); + assert_eq!(parse_go_duration("2.5s"), Some(Duration::from_millis(2500))); + assert_eq!(parse_go_duration("+2s"), Some(Duration::from_secs(2))); + assert_eq!( + parse_go_duration("1500us"), + Some(Duration::from_micros(1500)) + ); + assert_eq!( + parse_go_duration("3\u{b5}s"), + Some(Duration::from_micros(3)) + ); + assert_eq!( + parse_go_duration("3\u{3bc}s"), + Some(Duration::from_micros(3)) + ); + assert_eq!(parse_go_duration("1s20ns"), Some(Duration::new(1, 20))); + assert_eq!(parse_go_duration("-5m"), None); + assert_eq!(parse_go_duration("99999999999999999999h"), None); + assert_eq!( + parse_go_duration("4000000000000000h4000000000000000h"), + None + ); + assert_eq!(parse_go_duration("2d"), None); + assert_eq!(parse_go_duration(""), None); + } +} diff --git a/seaweed-vector/crates/server/tests/tls/a.key b/seaweed-vector/crates/server/tests/tls/a.key new file mode 100644 index 0000000000..51551061ec --- /dev/null +++ b/seaweed-vector/crates/server/tests/tls/a.key @@ -0,0 +1,3 @@ +-----BEGIN PRIVATE KEY----- +MC4CAQAwBQYDK2VwBCIEIIe8UCQQm2tMev77AnUJM5AQHPGBJ06euM8nGjDkILso +-----END PRIVATE KEY----- diff --git a/seaweed-vector/crates/server/tests/tls/a.pem b/seaweed-vector/crates/server/tests/tls/a.pem new file mode 100644 index 0000000000..deb43fb85a --- /dev/null +++ b/seaweed-vector/crates/server/tests/tls/a.pem @@ -0,0 +1,11 @@ +-----BEGIN CERTIFICATE----- +MIIBfTCCAS+gAwIBAgIUTqmeCb9ijj6cVkoZU6tXADadFnEwBQYDK2VwMB4xHDAa +BgNVBAMME3dlZWQtdmVjdG9yIHRlc3QgQ0EwHhcNMjYxMTAzMTgwNjI5WhcNMzYx +MDMxMTgwNjI5WjATMREwDwYDVQQDDAhlbmdpbmUtYTAqMAUGAytlcAMhACDaGO8F +ITSUkCId23gLzafauk7Z6bzZOpD732Zq5p/Uo4GJMIGGMBoGA1UdEQQTMBGCCWxv +Y2FsaG9zdIcEfwAAATAdBgNVHSUEFjAUBggrBgEFBQcDAQYIKwYBBQUHAwIwCQYD +VR0TBAIwADAdBgNVHQ4EFgQUr386MhXMqJ19631S4E70aTmtJ7wwHwYDVR0jBBgw +FoAUY0IfV1/n+b5dGE81HUcczB3Z5e4wBQYDK2VwA0EAalLd1JyRIForesFYzdVZ +x4xMjUqXmZHjIZfrnwli6jLwXHhOBD57Kg8diozVVWFviTqaw58q/615qikJn6hO +AA== +-----END CERTIFICATE----- diff --git a/seaweed-vector/crates/server/tests/tls/b.key b/seaweed-vector/crates/server/tests/tls/b.key new file mode 100644 index 0000000000..7832fd81a7 --- /dev/null +++ b/seaweed-vector/crates/server/tests/tls/b.key @@ -0,0 +1,3 @@ +-----BEGIN PRIVATE KEY----- +MC4CAQAwBQYDK2VwBCIEIE/JTETGtdVKPupXUWAvw6QMjiKZQvXvClUSyOt3etEt +-----END PRIVATE KEY----- diff --git a/seaweed-vector/crates/server/tests/tls/b.pem b/seaweed-vector/crates/server/tests/tls/b.pem new file mode 100644 index 0000000000..a87d74959d --- /dev/null +++ b/seaweed-vector/crates/server/tests/tls/b.pem @@ -0,0 +1,11 @@ +-----BEGIN CERTIFICATE----- +MIIBfTCCAS+gAwIBAgIUTqmeCb9ijj6cVkoZU6tXADadFnIwBQYDK2VwMB4xHDAa +BgNVBAMME3dlZWQtdmVjdG9yIHRlc3QgQ0EwHhcNMjYxMTAzMTgwNjI5WhcNMzYx +MDMxMTgwNjI5WjATMREwDwYDVQQDDAhlbmdpbmUtYjAqMAUGAytlcAMhAJ+akJpq +yTvnOqNrXU9gAVTZ6UZksVQ8dLrYKruicLy4o4GJMIGGMBoGA1UdEQQTMBGCCWxv +Y2FsaG9zdIcEfwAAATAdBgNVHSUEFjAUBggrBgEFBQcDAQYIKwYBBQUHAwIwCQYD +VR0TBAIwADAdBgNVHQ4EFgQUq90KSsHhZ+BFewHdX/j8D7/Ow0kwHwYDVR0jBBgw +FoAUY0IfV1/n+b5dGE81HUcczB3Z5e4wBQYDK2VwA0EAJUIkaJjItZ15e+ais3NQ +nMa4r/Xk+goY48Kw+GXapcT6goxMqiMOcbg3i1Ct3Guywew5VPTLlw12g8rFbYiU +CQ== +-----END CERTIFICATE----- diff --git a/seaweed-vector/crates/server/tests/tls/ca.pem b/seaweed-vector/crates/server/tests/tls/ca.pem new file mode 100644 index 0000000000..401d0d96c7 --- /dev/null +++ b/seaweed-vector/crates/server/tests/tls/ca.pem @@ -0,0 +1,10 @@ +-----BEGIN CERTIFICATE----- +MIIBYTCCAROgAwIBAgIUTmlwPR0OgCN9BqG8FpVZDesTFz0wBQYDK2VwMB4xHDAa +BgNVBAMME3dlZWQtdmVjdG9yIHRlc3QgQ0EwHhcNMjYxMTAzMTgwNjI5WhcNMzYx +MDMxMTgwNjI5WjAeMRwwGgYDVQQDDBN3ZWVkLXZlY3RvciB0ZXN0IENBMCowBQYD +K2VwAyEAYHNDiIBy6Y4b3hBO71kc80611yE53/ZCCfszFpsV06ijYzBhMB0GA1Ud +DgQWBBRjQh9XX+f5vl0YTzUdRxzMHdnl7jAfBgNVHSMEGDAWgBRjQh9XX+f5vl0Y +TzUdRxzMHdnl7jAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQEAwIBBjAFBgMr +ZXADQQAdFveXGKtqg1whYn7dKVnHyJp8+pbSDHhOzjkj5zMXssT6I9l/ydb00jpA +0d9mGnOB2cgywGgeAOpqypbCnxIG +-----END CERTIFICATE----- diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index a127c34251..33a18a553e 100644 --- a/test/s3vectors/conftest.py +++ b/test/s3vectors/conftest.py @@ -89,8 +89,9 @@ def open_cluster(request, tmp_path_factory): def write_tls(directory): - """A CA and one certificate for localhost; volume servers serve HTTPS - requiring client certificates, and every client presents the same one.""" + """A CA and one certificate for localhost. Volume servers serve HTTPS and + every gRPC listener (master, volume, filer, gateway, engine) serves TLS, + all requiring client certificates, and every client presents the same one.""" def openssl(*args): subprocess.run(["openssl", *args], check=True, capture_output=True, cwd=directory) openssl("req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "ca.key", "-out", "ca.pem", @@ -101,8 +102,56 @@ def write_tls(directory): "subjectAltName=DNS:localhost,IP:127.0.0.1\nextendedKeyUsage=serverAuth,clientAuth\nbasicConstraints=CA:FALSE\n") openssl("x509", "-req", "-in", "req.csr", "-CA", "ca.pem", "-CAkey", "ca.key", "-CAcreateserial", "-out", "cert.pem", "-days", "2", "-extfile", "ext.cnf") - pem = "".join(f'{key} = {json.dumps(str(directory / name), ensure_ascii=False)}\n' for key, name in (("cert", "cert.pem"), ("key", "key.pem"), ("ca", "ca.pem"))) - (directory / "security.toml").write_text(f"[https.volume]\n{pem}\n[https.client]\nenabled = true\n{pem}") + # A second identity the same CA signed, which the engine's allowlist + # must keep out. + openssl("req", "-newkey", "rsa:2048", "-nodes", "-keyout", "outsider.key", "-out", "outsider.csr", "-subj", "/CN=outsider") + openssl("x509", "-req", "-in", "outsider.csr", "-CA", "ca.pem", "-CAkey", "ca.key", "-CAcreateserial", + "-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 += 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}") + + +def probe_grpc_tls(name, port, directory, outsider=False): + """A TLS gRPC listener must refuse plaintext HTTP/2 and a connection + without the client certificate, and answer the HTTP/2 preface with it. + With `outsider`, a certificate the CA signed for a name outside the + listener's allowed_commonNames must be refused too.""" + preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" + with socket.create_connection(("127.0.0.1", port), timeout=5) as sock: + sock.sendall(preface) + with contextlib.suppress(ConnectionError, TimeoutError): + answer = sock.recv(16) + # Nothing, or a TLS alert record; a plaintext server sends SETTINGS. + assert not answer or answer[0] == 0x15, f"{name} gRPC answered plaintext HTTP/2: {answer!r}" + context = ssl.create_default_context(cafile=directory / "ca.pem") + context.set_alpn_protocols(["h2"]) + with contextlib.suppress(ConnectionError, TimeoutError, ssl.SSLError): + with socket.create_connection(("127.0.0.1", port), timeout=5) as sock: + with context.wrap_socket(sock, server_hostname="localhost") as tls: + tls.sendall(preface) + assert not tls.recv(16), f"{name} gRPC served a client without a certificate" + if outsider: + other = ssl.create_default_context(cafile=directory / "ca.pem") + other.set_alpn_protocols(["h2"]) + other.load_cert_chain(directory / "outsider.pem", directory / "outsider.key") + with contextlib.suppress(ConnectionError, TimeoutError, ssl.SSLError): + with socket.create_connection(("127.0.0.1", port), timeout=5) as sock: + with other.wrap_socket(sock, server_hostname="localhost") as tls: + tls.sendall(preface) + assert not tls.recv(16), f"{name} gRPC served a certificate outside allowed_commonNames" + context.load_cert_chain(directory / "cert.pem", directory / "key.pem") + with socket.create_connection(("127.0.0.1", port), timeout=5) as sock: + with context.wrap_socket(sock, server_hostname="localhost") as tls: + tls.sendall(preface) + header = b"" + while len(header) < 9 and (chunk := tls.recv(9 - len(header))): + header += chunk + # The first server frame is SETTINGS (type 4). + assert len(header) == 9 and header[3] == 4, \ + f"{name} gRPC did not answer the HTTP/2 preface over TLS: {header!r}" def read_token(key, fid): @@ -222,7 +271,7 @@ def run_cluster(param, tmp_path_factory, identities): start("volume", prefix + [f"{flag}{key}={value}" for key, value in { "master": master_addr, "ip": "127.0.0.1", "ip.bind": "127.0.0.1", "port": vol, "port.grpc": vol_grpc, "dir": directory / "volume", - "max": 16, "preStopSeconds": 0, + "max": 64, "preStopSeconds": 0, }.items()]) wait_for(lambda: listening(vol_grpc), processes, "volume not ready") start("filer", [weed, "filer", f"-master={master_addr}", "-ip=127.0.0.1", @@ -250,6 +299,12 @@ def run_cluster(param, tmp_path_factory, identities): start("vector", [engine, "--namespace", f"http://127.0.0.1:{lance}", "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", *credentials], unset) wait_for(lambda: listening(vector), processes, "vector engine not ready") + if os.environ.get("VECTOR_TLS"): + wait_for(lambda: listening(s3_grpc), processes, "s3 gRPC not ready") + for name, port in (("master", master_grpc), ("volume", vol_grpc), ("filer", filer_grpc), + ("s3", s3_grpc)): + probe_grpc_tls(name, port, directory) + probe_grpc_tls("engine", vector, directory, outsider=True) wait_for( lambda: f"Cluster: vector node 127.0.0.1:{vector} added to group ''" diff --git a/weed/command/scaffold/security.toml b/weed/command/scaffold/security.toml index b347d818a2..2bebdc45f4 100644 --- a/weed/command/scaffold/security.toml +++ b/weed/command/scaffold/security.toml @@ -174,6 +174,16 @@ client_cert = "" client_key = "" allowed_commonNames = "" # comma-separated SSL certificate common names +# weed-vector gRPC options (server-side) +# Enables mTLS for incoming gRPC connections to the vector engine; the S3 +# gateway dials it with [grpc.client], and the engine dials the masters with +# [grpc.client] too. The files are re-read every WEED_TLS_CERT_REFRESH_INTERVAL +# (default 5h) and on SIGHUP. +[grpc.vector] +cert = "" +key = "" +allowed_commonNames = "" # comma-separated SSL certificate common names + # gRPC client configuration for outgoing gRPC connections # Used by clients (S3, mount, backup, benchmark, filer.copy, filer.replicate, upload, etc.) # when connecting to any gRPC server (master, volume, filer)