seaweed-worker: mint the admin bearer token like the Go worker

The admin control stream now requires mTLS or a token signed by
jwt.filer_signing.key; the Rust worker gets --filer-signing-key and
--filer-signing-key-expires-after-seconds, mirroring the Go worker's
config names, and attaches the token to the stream metadata.

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"

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

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<TlsOptions>,
+    /// 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<String>,
+    /// 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,
         }
     }
 }

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 {

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(),

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<String> {
+    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"{}");
+    }
+}

seaweed-worker/crates/lance/src/main.rs

@@ -61,6 +61,18 @@ struct Args {
     #[arg(long)]
     tls_server_name: Option<String>,
 
+    /// 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<String>,
+
+    /// 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()
     };