diff --git a/seaweed-worker/Cargo.lock b/seaweed-worker/Cargo.lock index 033bf3ed4..3dfd02f29 100644 --- a/seaweed-worker/Cargo.lock +++ b/seaweed-worker/Cargo.lock @@ -485,6 +485,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" dependencies = [ "aws-lc-sys", + "untrusted 0.7.1", "zeroize", ] @@ -5124,7 +5125,7 @@ dependencies = [ "cfg-if", "getrandom 0.2.17", "libc", - "untrusted", + "untrusted 0.9.0", "windows-sys 0.52.0", ] @@ -5260,7 +5261,7 @@ dependencies = [ "aws-lc-rs", "ring", "rustls-pki-types", - "untrusted", + "untrusted 0.9.0", ] [[package]] @@ -5342,7 +5343,9 @@ version = "0.1.0" dependencies = [ "anyhow", "async-trait", + "aws-lc-rs", "axum", + "base64 0.22.1", "prometheus", "prost", "prost-types", @@ -6213,6 +6216,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254" +[[package]] +name = "untrusted" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" + [[package]] name = "untrusted" version = "0.9.0" diff --git a/seaweed-worker/crates/core/Cargo.toml b/seaweed-worker/crates/core/Cargo.toml index 63619e8f9..d75d5b9a4 100644 --- a/seaweed-worker/crates/core/Cargo.toml +++ b/seaweed-worker/crates/core/Cargo.toml @@ -12,6 +12,10 @@ name = "seaweed_worker_core" # Helpers the Rust volume server needs as well. A path dependency because # the two trees are separate cargo workspaces with no common root manifest. seaweed-common = { path = "../../../seaweed-common" } +# Already in the graph through tonic's `tls-aws-lc` feature; named here so the +# admin token can sign with the same HMAC the Go worker uses. +aws-lc-rs = "1" +base64 = "0.22" anyhow.workspace = true async-trait.workspace = true prost.workspace = true diff --git a/seaweed-worker/crates/core/src/config.rs b/seaweed-worker/crates/core/src/config.rs index c8fcb726d..2941c5d45 100644 --- a/seaweed-worker/crates/core/src/config.rs +++ b/seaweed-worker/crates/core/src/config.rs @@ -17,6 +17,15 @@ pub struct WorkerOptions { /// section of security.toml. None means plaintext, which is the default the /// Go worker also takes when no certificates are configured. pub tls: Option, + /// The shared secret the admin stream requires when the cluster configures + /// `jwt.filer_signing.key`; mirrored from that config value. Empty means no + /// token is attached, which only works against admins that either bound + /// their worker listener to loopback or authenticate with mTLS instead. + pub signing_key: Option, + /// Seconds the minted admin token lives, mirroring + /// `jwt.filer_signing.expires_after_seconds`. Zero mints a token without an + /// expiry, matching the Go default. + pub signing_key_expires_after_seconds: u64, } /// Certificates for the control stream. All three are required together: the @@ -44,6 +53,8 @@ impl Default for WorkerOptions { max_detection_concurrency: 1, max_execution_concurrency: 1, tls: None, + signing_key: None, + signing_key_expires_after_seconds: 0, } } } diff --git a/seaweed-worker/crates/core/src/lib.rs b/seaweed-worker/crates/core/src/lib.rs index ffe8ea453..8a9a548dc 100644 --- a/seaweed-worker/crates/core/src/lib.rs +++ b/seaweed-worker/crates/core/src/lib.rs @@ -13,6 +13,7 @@ pub mod metrics; pub mod registry; pub mod senders; pub mod stream; +pub mod token; /// Generated plugin.proto types. pub mod pb { diff --git a/seaweed-worker/crates/core/src/stream.rs b/seaweed-worker/crates/core/src/stream.rs index 67fe8fa83..63eeeea69 100644 --- a/seaweed-worker/crates/core/src/stream.rs +++ b/seaweed-worker/crates/core/src/stream.rs @@ -176,10 +176,19 @@ async fn serve_once( metadata: Default::default(), }))?; - let mut inbound = client - .worker_stream(UnboundedReceiverStream::new(rx)) - .await? - .into_inner(); + let mut request = tonic::Request::new(UnboundedReceiverStream::new(rx)); + if let Some(token) = crate::token::admin_token( + options.signing_key.as_deref().unwrap_or("").as_bytes(), + options.signing_key_expires_after_seconds, + ) { + request.metadata_mut().insert( + "authorization", + format!("Bearer {token}") + .parse() + .context("minted admin token is not valid metadata")?, + ); + } + let mut inbound = client.worker_stream(request).await?.into_inner(); let heartbeat = spawn_heartbeat( sender.clone(), diff --git a/seaweed-worker/crates/core/src/token.rs b/seaweed-worker/crates/core/src/token.rs new file mode 100644 index 000000000..3df69398e --- /dev/null +++ b/seaweed-worker/crates/core/src/token.rs @@ -0,0 +1,61 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + +use aws_lc_rs::hmac; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; + +/// Mints the HS256 Bearer token the admin control stream requires when +/// jwt.filer_signing.key is configured, matching the SeaweedFilerAdminClaims +/// the Go worker sends: registered claims only, no session id. +pub fn admin_token(signing_key: &[u8], expires_after_secs: u64) -> Option { + if signing_key.is_empty() { + return None; + } + let claims = if expires_after_secs > 0 { + let exp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or_default() + + expires_after_secs; + format!("{{\"exp\":{exp}}}") + } else { + "{}".to_string() + }; + let signing_input = format!( + "{}.{}", + URL_SAFE_NO_PAD.encode(b"{\"alg\":\"HS256\",\"typ\":\"JWT\"}"), + URL_SAFE_NO_PAD.encode(claims.as_bytes()) + ); + let key = hmac::Key::new(hmac::HMAC_SHA256, signing_key); + let signature = hmac::sign(&key, signing_input.as_bytes()); + Some(format!( + "{signing_input}.{}", + URL_SAFE_NO_PAD.encode(signature.as_ref()) + )) +} + +#[cfg(test)] +mod tests { + use super::admin_token; + + #[test] + fn no_key_no_token() { + assert!(admin_token(b"", 0).is_none()); + } + + #[test] + fn minted_token_verifies_shape() { + let token = admin_token(b"secret", 60).unwrap(); + let parts: Vec<&str> = token.split('.').collect(); + assert_eq!(parts.len(), 3); + assert!(!parts.iter().any(|p| p.is_empty())); + } + + #[test] + fn no_expiry_has_empty_claims() { + let token = admin_token(b"secret", 0).unwrap(); + let payload = token.split('.').nth(1).unwrap(); + use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; + let claims = URL_SAFE_NO_PAD.decode(payload).unwrap(); + assert_eq!(claims, b"{}"); + } +} diff --git a/seaweed-worker/crates/lance/src/main.rs b/seaweed-worker/crates/lance/src/main.rs index 1ff6db40a..4c268c7c1 100644 --- a/seaweed-worker/crates/lance/src/main.rs +++ b/seaweed-worker/crates/lance/src/main.rs @@ -61,6 +61,18 @@ struct Args { #[arg(long)] tls_server_name: Option, + /// Shared secret for the admin control stream, the `jwt.filer_signing.key` + /// value the Go worker reads from its config file. Required by admins that + /// authenticate the worker stream with a signing key. + #[arg(long, env = "WEED_FILER_SIGNING_KEY")] + filer_signing_key: Option, + + /// Lifetime of the token minted from the signing key, mirroring + /// `jwt.filer_signing.expires_after_seconds`. Zero mints a token with no + /// expiry. + #[arg(long, default_value = "0")] + filer_signing_key_expires_after_seconds: u64, + /// Serve /health, /ready and /metrics on this port, the way /// `weed worker -metricsPort` does. Zero, the default, serves nothing. #[arg(long, default_value = "0", env = "WEED_METRICS_PORT")] @@ -131,6 +143,8 @@ async fn main() -> Result<()> { max_detection_concurrency: args.max_concurrency, max_execution_concurrency: args.max_concurrency, tls, + signing_key: args.filer_signing_key, + signing_key_expires_after_seconds: args.filer_signing_key_expires_after_seconds, ..Default::default() };