From b9bc42c217280ad2a1c806312973f7730c697073 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 1 Nov 2026 18:49:01 -0800 Subject: [PATCH] s3vectors: vend write credentials to index data only to the vector engine and Lance worker (#12316) * security: share the filer admin token check CheckGrpcAdminPeer verifies an admin Bearer token inline. Move the check into IsFilerAdminToken so an HTTP endpoint can accept the same token. Co-Authored-By: Claude Opus 5.5 * lance: recognize the cluster's own components by a signed token A request whose X-Seaweedfs-Cluster-Token header carries an admin token signed with jwt.filer_signing.key comes from a component that holds the cluster key: the vector engine or the Lance maintenance worker. Mark such a request as trusted. It still authenticates as an identity, which stays the principal for vended credentials. A dedicated header leaves the Authorization header to Lance OAuth clients. A gateway with no key trusts no one. Co-Authored-By: Claude Opus 5.5 * lance: vend write credentials to a vector bucket's indexes only to trusted callers An index in a vector bucket's vector namespace is also a Lance table, so a caller granted s3tables:UpdateTable could describe it, or declare it again, and get credentials that write the index data without passing the vector bucket policy. In that namespace both now vend read-only credentials unless the caller is trusted, and a declare from anyone else no longer writes the dataset markers. Co-Authored-By: Claude Opus 5.5 * lance worker: prove the cluster key to the Lance namespace The gateway now vends write credentials to a vector bucket's indexes only to callers that send a token signed with jwt.filer_signing.key. The worker already takes that key for the admin stream; its namespace client now mints a token from it for every catalog request. Co-Authored-By: Claude Opus 5.5 * weed-vector: describe the index the gateway created and prove the cluster key CreateIndex declared a table the gateway had just created, which only worked because declare is idempotent. Describe it instead. The Lance namespace vends write credentials to a vector bucket's indexes only to callers that send a token signed with jwt.filer_signing.key, so take the key with --filer-signing-key, the same flag the Lance worker has. Co-Authored-By: Claude Opus 5.5 * lance worker: keep the cluster key out of Debug output Co-Authored-By: Claude Opus 5.5 * lance: answer a declare of an existing table with its own location CreateTable succeeds for a name that already exists without looking at the location requested, and declare kept using that location for the dataset markers and the vended credentials. Declaring an existing table with another table's location, such as a vector index, got credentials to that table's data. Use the location the catalog holds, which CreateTable now also reports for a new table. Co-Authored-By: Claude Opus 5.5 * lance: sign cluster tokens with a key derived for the catalog A cluster token was a filer admin token signed with jwt.filer_signing.key and, by default, never expired. Sent to the Lance namespace on every request, often over plain HTTP, a captured one would also authorize the filer's IAM service. Sign it with a key derived from the cluster key for this one use, which the admin services do not accept, and mint it with a five minute lifetime. weed-vector drops its token lifetime flag. Co-Authored-By: Claude Opus 5.5 * lance: follow a reloaded jwt.filer_signing.key The Lance namespace copied the cluster key at startup. A SIGHUP reload refreshes the gateway's filer guard but not that copy, so after a key rotation the restarted vector engine and worker were no longer trusted. Read the key from the guard on every check. Co-Authored-By: Claude Opus 5.5 * iceberg: keep a vector bucket's indexes read-only An index in a vector bucket is also a catalog table the Iceberg REST catalog loads, and LoadTable vended credentials that could write it to any caller granted s3tables:UpdateTable. A commit staged its metadata file in the index directory on the same grant. Apply the vector namespace rule to both, after the caller's own UpdateTable check. Co-Authored-By: Claude Opus 5.5 * iceberg: confine a created table's location to its namespace before writing Stage-create skips the registration that confines a location to the table's namespace, yet vends write credentials for that location. A caller allowed to create a table in any namespace of a bucket could name another namespace's prefix, a vector index included, and get write credentials to it. The non-staged create also wrote its first metadata file there before the registration refused the location. Co-Authored-By: Claude Opus 5.5 * lance: a declare that finds an existing table answers like describe Creating a name that already exists is idempotent, so a caller allowed only to create tables could declare someone else's table and get write credentials to it, and clear its deregistered marker. Describe would have vended that caller read-only credentials. A declare now looks the name up first; when it only finds the table, writing needs UpdateTable, as it does for describe. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- seaweed-vector/crates/server/src/main.rs | 11 +- seaweed-vector/crates/server/src/service.rs | 3 +- seaweed-worker/README.md | 5 + seaweed-worker/crates/core/src/token.rs | 19 ++- .../crates/lance/src/catalog/mod.rs | 2 +- .../crates/lance/src/catalog/namespace.rs | 121 +++++++++++++++- .../crates/lance/src/jobs/cleanup.rs | 13 +- .../crates/lance/src/jobs/compact.rs | 13 +- .../crates/lance/src/jobs/indices.rs | 13 +- seaweed-worker/crates/lance/src/jobs/mod.rs | 5 + seaweed-worker/crates/lance/src/jobs/sort.rs | 13 +- seaweed-worker/crates/lance/src/main.rs | 9 +- seaweed-worker/crates/lance/src/preview.rs | 13 +- weed/command/s3.go | 1 + weed/s3api/iceberg/commit_helpers.go | 4 + weed/s3api/iceberg/handlers_table.go | 9 ++ weed/s3api/iceberg/vector_namespace_test.go | 130 ++++++++++++++++++ weed/s3api/lance/cluster_token_test.go | 94 +++++++++++++ weed/s3api/lance/handlers_table.go | 49 +++++-- weed/s3api/lance/handlers_test.go | 43 ++++++ weed/s3api/lance/server.go | 57 ++++++++ weed/s3api/lance/storage.go | 17 +++ weed/s3api/lance/vector_namespace_test.go | 88 ++++++++++++ weed/s3api/s3api_server.go | 9 ++ weed/s3api/s3tables/handler_table.go | 5 +- weed/security/grpc_peer.go | 20 ++- 26 files changed, 724 insertions(+), 42 deletions(-) create mode 100644 weed/s3api/iceberg/vector_namespace_test.go create mode 100644 weed/s3api/lance/cluster_token_test.go diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index 10d46f372d..aa42defcc7 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -1,7 +1,7 @@ use anyhow::Result; use clap::Parser; use tonic::transport::Server; -use weed_lance_worker::catalog::NamespaceClient; +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; @@ -31,6 +31,10 @@ struct Args { /// Memory for caching reads of write-once table files, in MiB; 0 disables. #[arg(long, env = "WEED_VECTOR_READ_CACHE_MB", default_value_t = 512)] read_cache_mb: u64, + /// The `jwt.filer_signing.key` of the cluster. The Lance namespace vends + /// credentials that write index data only to a caller holding it. + #[arg(long, env = "WEED_FILER_SIGNING_KEY")] + filer_signing_key: Option, } fn parse_option(s: &str) -> Result<(String, String), String> { @@ -111,7 +115,10 @@ async fn main() -> Result<()> { )); } let fallback: FallbackOptions = args.storage_options.into_iter().collect(); - let client = NamespaceClient::new(args.namespace.clone()).with_fallback(&fallback); + let signing_key = SigningKey::new(args.filer_signing_key); + let client = NamespaceClient::new(args.namespace.clone()) + .with_fallback(&fallback) + .with_signing_key(signing_key.as_ref()); let mut service = VectorServer::new(client, fallback); if args.read_cache_mb > 0 { service = service.with_read_cache(ReadCache::new(args.read_cache_mb << 20)); diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index cb9f51b1c4..ea26f6c9bf 100644 --- a/seaweed-vector/crates/server/src/service.rs +++ b/seaweed-vector/crates/server/src/service.rs @@ -424,9 +424,10 @@ impl VectorService for VectorServer { if dimension == 0 || dimension > 4096 { return Err(Status::invalid_argument("dimension must be 1..=4096")); } + // The gateway has already created the table through the catalog. let description = self .client - .declare_table(&table.id, None) + .describe_table(&table.id) .await .map_err(internal)?; let mut options = self.fallback.clone(); diff --git a/seaweed-worker/README.md b/seaweed-worker/README.md index 9dde317711..02497807d9 100644 --- a/seaweed-worker/README.md +++ b/seaweed-worker/README.md @@ -95,6 +95,11 @@ gateway without STS configured vends no credentials at all, so `--access-key` and `--secret-key` supply a fallback; anything the namespace does vend wins over them. +In a vector bucket's `vector` namespace the gateway vends write credentials only +to a caller that proves it holds `jwt.filer_signing.key`. Pass that key with +`--filer-signing-key` so compaction, cleanup and index builds can rewrite +vector indexes. + ## State All three jobs are implemented and tested end to end against a live gateway: diff --git a/seaweed-worker/crates/core/src/token.rs b/seaweed-worker/crates/core/src/token.rs index 3df69398ef..21c01b9e08 100644 --- a/seaweed-worker/crates/core/src/token.rs +++ b/seaweed-worker/crates/core/src/token.rs @@ -33,9 +33,26 @@ pub fn admin_token(signing_key: &[u8], expires_after_secs: u64) -> Option Vec { + let key = hmac::Key::new(hmac::HMAC_SHA256, signing_key); + hmac::sign(&key, purpose.as_bytes()).as_ref().to_vec() +} + #[cfg(test)] mod tests { - use super::admin_token; + use super::{admin_token, derive_key}; + + #[test] + fn derive_key_matches_the_go_derivation() { + let derived = derive_key(b"cluster-key", "seaweedfs-lance-cluster-token"); + let hex: String = derived.iter().map(|b| format!("{b:02x}")).collect(); + assert_eq!( + hex, + "b38c0b66da41621f2a9b6de68e4239ddd0d2dd7b4ede01aa91bd34a17edc1021" + ); + } #[test] fn no_key_no_token() { diff --git a/seaweed-worker/crates/lance/src/catalog/mod.rs b/seaweed-worker/crates/lance/src/catalog/mod.rs index 142de4eb52..a3a91ad307 100644 --- a/seaweed-worker/crates/lance/src/catalog/mod.rs +++ b/seaweed-worker/crates/lance/src/catalog/mod.rs @@ -7,4 +7,4 @@ pub mod namespace; -pub use namespace::{NamespaceClient, TableDescription, parse_id}; +pub use namespace::{NamespaceClient, SigningKey, TableDescription, parse_id}; diff --git a/seaweed-worker/crates/lance/src/catalog/namespace.rs b/seaweed-worker/crates/lance/src/catalog/namespace.rs index 6aca480495..bbbfda6f0f 100644 --- a/seaweed-worker/crates/lance/src/catalog/namespace.rs +++ b/seaweed-worker/crates/lance/src/catalog/namespace.rs @@ -8,6 +8,39 @@ use serde::Deserialize; /// The delimiter the Lance namespace joins identifier parts with. const DELIMITER: &str = "$"; +/// Carries a token that proves the caller holds the cluster key, which marks +/// it as one of the cluster's own components. +const CLUSTER_TOKEN_HEADER: &str = "X-Seaweedfs-Cluster-Token"; + +/// A cluster token is minted for each request, so it only has to outlast the +/// clock skew between this host and the gateway. +const CLUSTER_TOKEN_SECS: u64 = 300; + +/// The key cluster tokens are signed with, derived from the cluster key +/// (`jwt.filer_signing.key`) the way the gateway derives it, so a token +/// captured on the wire does not authorize admin calls. The catalog vends +/// credentials that write a vector bucket's index data only to callers that +/// send one. Not Debug, so the key cannot reach a log. +#[derive(Clone)] +pub struct SigningKey { + key: Vec, +} + +impl SigningKey { + pub fn new(key: Option) -> Option { + key.filter(|key| !key.is_empty()).map(|key| Self { + key: seaweed_worker_core::token::derive_key( + key.as_bytes(), + "seaweedfs-lance-cluster-token", + ), + }) + } + + fn token(&self) -> Option { + seaweed_worker_core::token::admin_token(&self.key, CLUSTER_TOKEN_SECS) + } +} + #[derive(Debug, Deserialize)] pub struct TableDescription { pub location: String, @@ -35,6 +68,7 @@ pub struct NamespaceClient { base_url: String, http: reqwest::Client, api_key: Option, + signing_key: Option, } /// A namespace call that has not answered by now is not going to. Without this @@ -71,6 +105,7 @@ impl NamespaceClient { base_url: base_url.into().trim_end_matches('/').to_string(), http, api_key: None, + signing_key: None, } } @@ -92,10 +127,25 @@ impl NamespaceClient { self } + pub fn with_signing_key(mut self, signing_key: Option<&SigningKey>) -> Self { + if signing_key.is_some() && !protected_transport(&self.base_url) { + tracing::warn!( + "catalog url {} exposes the cluster token in cleartext", + self.base_url + ); + } + self.signing_key = signing_key.cloned(); + self + } + fn authed(&self, request: reqwest::RequestBuilder) -> reqwest::RequestBuilder { - match &self.api_key { + let request = match &self.api_key { Some(key) => request.header("x-api-key", key), None => request, + }; + match self.signing_key.as_ref().and_then(SigningKey::token) { + Some(token) => request.header(CLUSTER_TOKEN_HEADER, token), + None => request, } } @@ -224,7 +274,74 @@ pub fn parse_id(encoded: &str) -> Vec { #[cfg(test)] mod tests { - use super::{MAX_REDIRECT_HOPS, may_follow, protected_transport, same_origin}; + use super::{ + CLUSTER_TOKEN_SECS, MAX_REDIRECT_HOPS, NamespaceClient, SigningKey, may_follow, + protected_transport, same_origin, + }; + + /// A catalog that answers one request with a table location and hands back + /// the request it read. + async fn recording_catalog() -> (String, tokio::sync::oneshot::Receiver) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let (sender, receiver) = tokio::sync::oneshot::channel(); + tokio::spawn(async move { + let (mut conn, _) = listener.accept().await.unwrap(); + let mut request = vec![0u8; 8192]; + let n = conn.read(&mut request).await.unwrap(); + let _ = sender.send(String::from_utf8_lossy(&request[..n]).into_owned()); + let body = r#"{"location":"s3://b/vector/t"}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = conn.write_all(response.as_bytes()).await; + }); + (url, receiver) + } + + fn header<'a>(request: &'a str, name: &str) -> Option<&'a str> { + request.lines().find_map(|line| { + let (key, value) = line.split_once(':')?; + key.eq_ignore_ascii_case(name).then(|| value.trim()) + }) + } + + #[tokio::test] + async fn signing_key_sends_a_cluster_token() { + let id = ["b".to_string(), "vector".to_string(), "t".to_string()]; + let (url, request) = recording_catalog().await; + let key = SigningKey::new(Some("cluster-key".into())); + NamespaceClient::new(url) + .with_signing_key(key.as_ref()) + .describe_table(&id) + .await + .unwrap(); + let request = request.await.unwrap(); + let token = header(&request, "x-seaweedfs-cluster-token").expect("cluster token"); + let raw = + seaweed_worker_core::token::admin_token(b"cluster-key", CLUSTER_TOKEN_SECS).unwrap(); + assert_eq!(token.split('.').count(), 3); + assert_ne!( + token.rsplit('.').next(), + raw.rsplit('.').next(), + "signed with the cluster key itself" + ); + + let (url, request) = recording_catalog().await; + NamespaceClient::new(url).describe_table(&id).await.unwrap(); + assert_eq!( + header(&request.await.unwrap(), "x-seaweedfs-cluster-token"), + None + ); + } + + #[test] + fn an_empty_signing_key_is_no_key() { + assert!(SigningKey::new(None).is_none()); + assert!(SigningKey::new(Some(String::new())).is_none()); + } #[test] fn may_follow_bounds_hops_within_the_origin() { diff --git a/seaweed-worker/crates/lance/src/jobs/cleanup.rs b/seaweed-worker/crates/lance/src/jobs/cleanup.rs index c8c4a3ec36..028739d688 100644 --- a/seaweed-worker/crates/lance/src/jobs/cleanup.rs +++ b/seaweed-worker/crates/lance/src/jobs/cleanup.rs @@ -13,7 +13,7 @@ use seaweed_worker_core::pb::{ use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler}; use tracing::warn; -use crate::catalog::{NamespaceClient, parse_id}; +use crate::catalog::{NamespaceClient, SigningKey, parse_id}; use crate::dataset; use crate::jobs::{clamp, string_list, table_id}; @@ -32,6 +32,7 @@ const MAX_MIN_VERSIONS: i64 = 1000; pub struct CleanupVersionsHandler { namespace_url: String, fallback: dataset::FallbackOptions, + signing_key: Option, metrics: Option, } @@ -40,6 +41,7 @@ impl CleanupVersionsHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + signing_key: None, metrics: None, } } @@ -54,8 +56,15 @@ impl CleanupVersionsHandler { self } + pub fn with_signing_key(mut self, signing_key: Option) -> Self { + self.signing_key = signing_key; + self + } + fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) + NamespaceClient::new(self.namespace_url.clone()) + .with_fallback(&self.fallback) + .with_signing_key(self.signing_key.as_ref()) } } diff --git a/seaweed-worker/crates/lance/src/jobs/compact.rs b/seaweed-worker/crates/lance/src/jobs/compact.rs index 728437420d..a4ae5d7565 100644 --- a/seaweed-worker/crates/lance/src/jobs/compact.rs +++ b/seaweed-worker/crates/lance/src/jobs/compact.rs @@ -12,7 +12,7 @@ use seaweed_worker_core::pb::{ use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler}; use tracing::warn; -use crate::catalog::{NamespaceClient, parse_id}; +use crate::catalog::{NamespaceClient, SigningKey, parse_id}; use crate::dataset; use crate::jobs::{FORMAT, clamp, observation, string_list, table_id}; @@ -35,6 +35,7 @@ const MIN_FRAGMENTS_CEILING: i64 = 4096; pub struct CompactHandler { namespace_url: String, fallback: dataset::FallbackOptions, + signing_key: Option, metrics: Option, } @@ -43,6 +44,7 @@ impl CompactHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + signing_key: None, metrics: None, } } @@ -58,8 +60,15 @@ impl CompactHandler { self } + pub fn with_signing_key(mut self, signing_key: Option) -> Self { + self.signing_key = signing_key; + self + } + fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) + NamespaceClient::new(self.namespace_url.clone()) + .with_fallback(&self.fallback) + .with_signing_key(self.signing_key.as_ref()) } } diff --git a/seaweed-worker/crates/lance/src/jobs/indices.rs b/seaweed-worker/crates/lance/src/jobs/indices.rs index e619ff43d5..24f87b0c68 100644 --- a/seaweed-worker/crates/lance/src/jobs/indices.rs +++ b/seaweed-worker/crates/lance/src/jobs/indices.rs @@ -13,7 +13,7 @@ use seaweed_worker_core::pb::{ use seaweed_worker_core::{DetectionSender, ExecutionSender, JobHandler}; use tracing::warn; -use crate::catalog::{NamespaceClient, parse_id}; +use crate::catalog::{NamespaceClient, SigningKey, parse_id}; use crate::dataset::{self, OpenTable}; use crate::jobs::{clamp, string_list, table_id}; @@ -35,6 +35,7 @@ const MAX_UNINDEXED_CEILING: i64 = 100_000_000; pub struct OptimizeIndicesHandler { namespace_url: String, fallback: dataset::FallbackOptions, + signing_key: Option, metrics: Option, } @@ -43,6 +44,7 @@ impl OptimizeIndicesHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + signing_key: None, metrics: None, } } @@ -57,8 +59,15 @@ impl OptimizeIndicesHandler { self } + pub fn with_signing_key(mut self, signing_key: Option) -> Self { + self.signing_key = signing_key; + self + } + fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) + NamespaceClient::new(self.namespace_url.clone()) + .with_fallback(&self.fallback) + .with_signing_key(self.signing_key.as_ref()) } } diff --git a/seaweed-worker/crates/lance/src/jobs/mod.rs b/seaweed-worker/crates/lance/src/jobs/mod.rs index b5502a7747..a9c3bcfb51 100644 --- a/seaweed-worker/crates/lance/src/jobs/mod.rs +++ b/seaweed-worker/crates/lance/src/jobs/mod.rs @@ -42,27 +42,32 @@ pub(crate) fn table_id(parameters: &HashMap) -> Option, metrics: Option, ) -> Vec> { vec![ Arc::new( compact::CompactHandler::new(namespace_url.clone()) .with_fallback(fallback.clone()) + .with_signing_key(signing_key.clone()) .with_metrics(metrics.clone()), ), Arc::new( indices::OptimizeIndicesHandler::new(namespace_url.clone()) .with_fallback(fallback.clone()) + .with_signing_key(signing_key.clone()) .with_metrics(metrics.clone()), ), Arc::new( sort::SortHandler::new(namespace_url.clone()) .with_fallback(fallback.clone()) + .with_signing_key(signing_key.clone()) .with_metrics(metrics.clone()), ), Arc::new( cleanup::CleanupVersionsHandler::new(namespace_url) .with_fallback(fallback) + .with_signing_key(signing_key) .with_metrics(metrics), ), ] diff --git a/seaweed-worker/crates/lance/src/jobs/sort.rs b/seaweed-worker/crates/lance/src/jobs/sort.rs index ed8aeaa7ba..5ac6859f6a 100644 --- a/seaweed-worker/crates/lance/src/jobs/sort.rs +++ b/seaweed-worker/crates/lance/src/jobs/sort.rs @@ -23,7 +23,7 @@ use seaweed_worker_sort::{ }; use tracing::warn; -use crate::catalog::{NamespaceClient, parse_id}; +use crate::catalog::{NamespaceClient, SigningKey, parse_id}; use crate::dataset; use crate::jobs::{FORMAT, clamp, observation, string_list, table_id}; @@ -54,6 +54,7 @@ const BYTES_PER_MB: u64 = 1024 * 1024; pub struct SortHandler { namespace_url: String, fallback: dataset::FallbackOptions, + signing_key: Option, metrics: Option, } @@ -62,6 +63,7 @@ impl SortHandler { Self { namespace_url, fallback: dataset::FallbackOptions::new(), + signing_key: None, metrics: None, } } @@ -77,8 +79,15 @@ impl SortHandler { self } + pub fn with_signing_key(mut self, signing_key: Option) -> Self { + self.signing_key = signing_key; + self + } + fn client(&self) -> NamespaceClient { - NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback) + NamespaceClient::new(self.namespace_url.clone()) + .with_fallback(&self.fallback) + .with_signing_key(self.signing_key.as_ref()) } } diff --git a/seaweed-worker/crates/lance/src/main.rs b/seaweed-worker/crates/lance/src/main.rs index 8f688cf357..b47a20bf8a 100644 --- a/seaweed-worker/crates/lance/src/main.rs +++ b/seaweed-worker/crates/lance/src/main.rs @@ -63,7 +63,8 @@ struct Args { /// 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. + /// authenticate the worker stream with a signing key. The Lance namespace + /// takes it too: only a holder gets write credentials to vector indexes. #[arg(long, env = "WEED_FILER_SIGNING_KEY")] filer_signing_key: Option, @@ -136,6 +137,7 @@ async fn main() -> Result<()> { let args = Args::parse(); let tls = args.tls()?; + let signing_key = weed_lance_worker::catalog::SigningKey::new(args.filer_signing_key.clone()); let options = WorkerOptions { admin_address: args.admin, worker_id: args.id, @@ -164,9 +166,10 @@ async fn main() -> Result<()> { } let mut registry = Registry::new().with_preview(Arc::new( - weed_lance_worker::preview::LancePreview::new(args.namespace.clone(), fallback.clone()), + weed_lance_worker::preview::LancePreview::new(args.namespace.clone(), fallback.clone()) + .with_signing_key(signing_key.clone()), )); - for handler in handlers(args.namespace, fallback, Some(lance_metrics)) { + for handler in handlers(args.namespace, fallback, signing_key, Some(lance_metrics)) { registry = registry.register(handler); } diff --git a/seaweed-worker/crates/lance/src/preview.rs b/seaweed-worker/crates/lance/src/preview.rs index 596d8700a0..49dd6e2775 100644 --- a/seaweed-worker/crates/lance/src/preview.rs +++ b/seaweed-worker/crates/lance/src/preview.rs @@ -10,13 +10,14 @@ use async_trait::async_trait; use futures::TryStreamExt; use seaweed_worker_core::{Preview, PreviewProvider}; -use crate::catalog::NamespaceClient; +use crate::catalog::{NamespaceClient, SigningKey}; use crate::dataset::{self, FallbackOptions}; use crate::jobs::FORMAT; pub struct LancePreview { namespace_url: String, fallback: FallbackOptions, + signing_key: Option, } impl LancePreview { @@ -24,8 +25,14 @@ impl LancePreview { Self { namespace_url, fallback, + signing_key: None, } } + + pub fn with_signing_key(mut self, signing_key: Option) -> Self { + self.signing_key = signing_key; + self + } } #[async_trait] @@ -35,7 +42,9 @@ impl PreviewProvider for LancePreview { } async fn preview(&self, object_id: &[String], row_limit: usize) -> Result { - let client = NamespaceClient::new(self.namespace_url.clone()).with_fallback(&self.fallback); + let client = NamespaceClient::new(self.namespace_url.clone()) + .with_fallback(&self.fallback) + .with_signing_key(self.signing_key.as_ref()); let table = dataset::open(&client, object_id, &self.fallback).await?; let total_rows = table.dataset.count_rows(None).await.unwrap_or(0) as i64; diff --git a/weed/command/s3.go b/weed/command/s3.go index f74bcda38b..94ed45d9f0 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -675,6 +675,7 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer, listenO lanceServer := lance.NewServer(s3ApiServer, s3ApiServer) lanceServer.SetCredentialValidator(s3ApiServer) + lanceServer.SetSigningKey(s3ApiServer.FilerSigningKey) if s3opt.icebergCredentialRole != nil && *s3opt.icebergCredentialRole != "" { lanceServer.SetCredentialVendor(lanceCredentialVendor{s3ApiServer}) } diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index bd725d019d..8dd19818ec 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/weed/s3api/iceberg/commit_helpers.go @@ -282,6 +282,10 @@ func (s *Server) authorizeCommit(ctx context.Context, bucketARN string, namespac Name: tableName, }, identityName) }) + if err == nil { + bucketName, _ := s3tables.ParseBucketNameFromARN(bucketARN) + err = s.tablesManager.CheckVectorNamespaceWrite(ctx, s.filerClient, bucketName, namespace) + } if err == nil { return nil } diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index d99ba5a708..98debaeb12 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -308,6 +308,12 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) { writeManagerError(w, authErr) return } + // Registration confines the location to the table's namespace, but both + // branches write, and stage-create vends credentials, before it runs. + if err := s3tables.ValidateMetadataLocation(metadataLocation, bucketName, flattenNamespacePath(namespace)); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", err.Error()) + return + } // Stage-create persists metadata in the internal staged area and skips S3Tables registration. if req.StageCreate { @@ -786,6 +792,9 @@ func (s *Server) mayUpdateTable(r *http.Request, bucketName string, namespace [] err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return s.tablesManager.AuthorizeUpdateTable(r.Context(), s3tables.NewManagerClient(client), updateReq, identityName) }) + if err == nil { + err = s.tablesManager.CheckVectorNamespaceWrite(r.Context(), s.filerClient, bucketName, namespace) + } if err != nil { glog.V(1).Infof("Iceberg: UpdateTable check for %s/%s failed, vending read-only: %v", bucketName, tableName, err) } diff --git a/weed/s3api/iceberg/vector_namespace_test.go b/weed/s3api/iceberg/vector_namespace_test.go new file mode 100644 index 0000000000..dad8e4ecb6 --- /dev/null +++ b/weed/s3api/iceberg/vector_namespace_test.go @@ -0,0 +1,130 @@ +package iceberg + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +type openAuthenticator struct{} + +func (openAuthenticator) AuthenticateRequest(*http.Request) (string, interface{}, s3err.ErrorCode) { + return s3tables.DefaultAccountID, nil, s3err.ErrNone +} + +func (openAuthenticator) DefaultAllow() bool { return true } + +// An index in a vector bucket is a catalog table the Iceberg catalog can load, +// but a caller holding every s3tables grant gets neither write credentials to +// it nor a commit into it. +func TestVectorIndexIsReadOnlyOverIceberg(t *testing.T) { + filer := s3tablestest.Start(t) + vendor := &stubVendor{credentials: &VendedCredentials{AccessKeyID: "ASIAEXAMPLE"}} + s := NewServer(s3tables.NewManagerClient(filer.Client), openAuthenticator{}) + s.SetCredentialVendor(vendor) + s.s3Endpoint = "http://127.0.0.1:8333" + router := mux.NewRouter().SkipClean(true) + s.RegisterRoutes(router) + + admin := s3tables.NewManager() + admin.SetTrusted(true) + client := s3tables.NewManagerClient(filer.Client) + arn, err := s3tables.BuildBucketARN(s3tables.DefaultRegion, s3tables.DefaultAccountID, "vec") + if err != nil { + t.Fatal(err) + } + metadata, _ := json.Marshal(map[string]any{"dimension": 4, "distanceMetric": "cosine"}) + for _, op := range []struct { + name string + req interface{} + }{ + {"CreateTableBucket", &s3tables.CreateTableBucketRequest{Name: "vec", Format: s3tables.FormatLance}}, + {"CreateNamespace", &s3tables.CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{s3tables.VectorIndexNamespace}}}, + {"CreateTable", &s3tables.CreateTableRequest{TableBucketARN: arn, Namespace: []string{s3tables.VectorIndexNamespace}, + Name: "idx", Format: s3tables.FormatLance, Metadata: &s3tables.TableMetadata{FullMetadata: metadata}}}, + } { + if err := admin.Execute(t.Context(), client, op.name, op.req, nil, ""); err != nil { + t.Fatalf("%s: %v", op.name, err) + } + } + + req := httptest.NewRequest(http.MethodGet, "/v1/vec/namespaces/vector/tables/idx", nil) + req.Header.Set(accessDelegationHeader, "vended-credentials") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("load = %d (%s)", rec.Code, rec.Body.String()) + } + if vendor.gotPrefix != "vector/idx" || !vendor.gotReadOnly { + t.Errorf("vended prefix %q read-only %v, want vector/idx read-only", vendor.gotPrefix, vendor.gotReadOnly) + } + + reqErr := s.authorizeCommit(t.Context(), arn, []string{s3tables.VectorIndexNamespace}, "idx", s3tables.DefaultAccountID) + if reqErr == nil || reqErr.status != http.StatusForbidden { + t.Errorf("commit authorization = %+v, want 403", reqErr) + } +} + +// Stage-create registers nothing, so the catalog's own location and vector +// namespace checks never see it. It must not hand out write credentials to an +// index, whether the staged table names the index's location from another +// namespace or sits in the vector namespace itself. +func TestStageCreateCannotReachVectorIndex(t *testing.T) { + filer := s3tablestest.Start(t) + s := NewServer(s3tables.NewManagerClient(filer.Client), openAuthenticator{}) + s.s3Endpoint = "http://127.0.0.1:8333" + router := mux.NewRouter().SkipClean(true) + s.RegisterRoutes(router) + + admin := s3tables.NewManager() + admin.SetTrusted(true) + client := s3tables.NewManagerClient(filer.Client) + arn, err := s3tables.BuildBucketARN(s3tables.DefaultRegion, s3tables.DefaultAccountID, "vec") + if err != nil { + t.Fatal(err) + } + for _, op := range []struct { + name string + req interface{} + }{ + {"CreateTableBucket", &s3tables.CreateTableBucketRequest{Name: "vec", Format: s3tables.FormatLance}}, + {"CreateNamespace", &s3tables.CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{s3tables.VectorIndexNamespace}}}, + {"CreateNamespace", &s3tables.CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{"other"}}}, + {"CreateTable", &s3tables.CreateTableRequest{TableBucketARN: arn, Namespace: []string{s3tables.VectorIndexNamespace}, + Name: "idx", Format: s3tables.FormatLance}}, + } { + if err := admin.Execute(t.Context(), client, op.name, op.req, nil, ""); err != nil { + t.Fatalf("%s: %v", op.name, err) + } + } + + for name, tc := range map[string]struct { + namespace, body string + status int + }{ + "index location from another namespace": {"other", `{"name":"x","stage-create":true,"location":"s3://vec/vector/idx"}`, http.StatusBadRequest}, + "vector namespace": {s3tables.VectorIndexNamespace, `{"name":"y","stage-create":true}`, http.StatusForbidden}, + } { + t.Run(name, func(t *testing.T) { + vendor := &stubVendor{credentials: &VendedCredentials{AccessKeyID: "ASIAEXAMPLE"}} + s.SetCredentialVendor(vendor) + req := httptest.NewRequest(http.MethodPost, "/v1/vec/namespaces/"+tc.namespace+"/tables", strings.NewReader(tc.body)) + req.Header.Set(accessDelegationHeader, "vended-credentials") + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != tc.status { + t.Errorf("stage-create = %d (%s), want %d", rec.Code, rec.Body.String(), tc.status) + } + if vendor.gotPrefix != "" && !vendor.gotReadOnly { + t.Errorf("vended write credentials to %q", vendor.gotPrefix) + } + }) + } +} diff --git a/weed/s3api/lance/cluster_token_test.go b/weed/s3api/lance/cluster_token_test.go new file mode 100644 index 0000000000..15409a3aa3 --- /dev/null +++ b/weed/s3api/lance/cluster_token_test.go @@ -0,0 +1,94 @@ +package lance + +import ( + "encoding/hex" + "net/http" + "net/http/httptest" + "testing" + "time" + + jwt "github.com/golang-jwt/jwt/v5" + "github.com/seaweedfs/seaweedfs/weed/security" +) + +func signAdminClaims(t *testing.T, key security.SigningKey, claims security.SeaweedFilerAdminClaims) string { + t.Helper() + token, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(key)) + if err != nil { + t.Fatalf("sign: %v", err) + } + return token +} + +func trustedSeenByHandler(t *testing.T, s *Server, token string) bool { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/v1/namespace/$/list", nil) + if token != "" { + req.Header.Set(clusterTokenHeader, token) + } + trusted := false + rec := httptest.NewRecorder() + s.Auth(func(w http.ResponseWriter, r *http.Request) { + trusted = isTrustedCaller(r.Context()) + w.WriteHeader(http.StatusOK) + })(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d (%s), want 200", rec.Code, rec.Body.String()) + } + return trusted +} + +func TestClusterTokenMarksTheCallerTrusted(t *testing.T) { + key := security.SigningKey("cluster-key") + s := NewServer(nil, openAuthenticator{}) + s.SetSigningKey(func() security.SigningKey { return key }) + + derived := clusterTokenKey(key) + expired := security.SeaweedFilerAdminClaims{} + expired.ExpiresAt = jwt.NewNumericDate(time.Now().Add(-time.Minute)) + for _, tc := range []struct { + name string + token string + want bool + }{ + {"no token", "", false}, + {"signed with the derived key", string(security.GenJwtForFilerAdmin(derived, 60)), true}, + {"signed with the cluster key itself", string(security.GenJwtForFilerAdmin(key, 60)), false}, + {"signed with another key", string(security.GenJwtForFilerAdmin(clusterTokenKey(security.SigningKey("other-key")), 60)), false}, + {"expired", signAdminClaims(t, derived, expired), false}, + {"session token", signAdminClaims(t, derived, security.SeaweedFilerAdminClaims{SessionId: "session"}), false}, + {"garbage", "not-a-token", false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := trustedSeenByHandler(t, s, tc.token); got != tc.want { + t.Errorf("trusted = %v, want %v", got, tc.want) + } + }) + } + + token := string(security.GenJwtForFilerAdmin(derived, 60)) + key = security.SigningKey("rotated-key") + if trustedSeenByHandler(t, s, token) { + t.Error("a token from before a key rotation is still trusted") + } + if !trustedSeenByHandler(t, s, string(security.GenJwtForFilerAdmin(clusterTokenKey(key), 60))) { + t.Error("a token signed after a key rotation is not trusted") + } + + key = nil + if trustedSeenByHandler(t, s, token) { + t.Error("a server with no cluster key trusted a caller") + } + s.SetSigningKey(nil) + if trustedSeenByHandler(t, s, token) { + t.Error("a server with no key source trusted a caller") + } +} + +// The Rust catalog client derives the same key; its tests pin this value too. +func TestClusterTokenKeyDerivation(t *testing.T) { + got := hex.EncodeToString(clusterTokenKey(security.SigningKey("cluster-key"))) + if want := "b38c0b66da41621f2a9b6de68e4239ddd0d2dd7b4ede01aa91bd34a17edc1021"; got != want { + t.Errorf("clusterTokenKey = %s, want %s", got, want) + } +} diff --git a/weed/s3api/lance/handlers_table.go b/weed/s3api/lance/handlers_table.go index 65c0e1a11d..c2027d2d32 100644 --- a/weed/s3api/lance/handlers_table.go +++ b/weed/s3api/lance/handlers_table.go @@ -135,24 +135,35 @@ func (s *Server) handleDeclareTable(w http.ResponseWriter, r *http.Request) { return } - if err := s.createTable(r, bucket, ns, name, location); err != nil { + _, lookupErr := s.loadLanceTable(r, bucket, ns, name) + location, err = s.createTable(r, bucket, ns, name, location) + if err != nil { writeStorageError(w, r, err) return } - if err := s.writeMarker(r, location, reservedMarker); err != nil { - // The catalog entry is the authority; the marker only mirrors it for - // clients that read the storage prefix directly. - glog.V(1).Infof("lance: could not write %s for %s: %v", reservedMarker, location, err) + // A declare that creates the table may write it: a Lance commit is a direct + // S3 write that never revisits the catalog, so the create authorization is + // the write evidence. One that only finds the table answers like describe. + var writable bool + if isNotFound(lookupErr) { + writable = s.mayWriteNamespace(r, bucket, ns) + } else { + writable = s.mayUpdateTable(r, bucket, ns, name) } - // Declaring a name that was deregistered brings it back, the same way - // registering it does. - if err := s.removeMarker(r, location, deregisteredMarker); err != nil { - glog.V(2).Infof("lance: could not clear %s for %s: %v", deregisteredMarker, location, err) + if writable { + if err := s.writeMarker(r, location, reservedMarker); err != nil { + // The catalog entry is the authority; the marker only mirrors it for + // clients that read the storage prefix directly. + glog.V(1).Infof("lance: could not write %s for %s: %v", reservedMarker, location, err) + } + // Declaring a name that was deregistered brings it back, the same way + // registering it does. + if err := s.removeMarker(r, location, deregisteredMarker); err != nil { + glog.V(2).Infof("lance: could not clear %s for %s: %v", deregisteredMarker, location, err) + } } - // A Lance commit is a direct S3 write that never revisits the catalog, so - // the create authorization above is the write evidence for the dataset. - options, err := s.storageOptions(r, bucket, location, wants(r, "vend_credentials", req.VendCredentials), func() bool { return true }) + options, err := s.storageOptions(r, bucket, location, wants(r, "vend_credentials", req.VendCredentials), func() bool { return writable }) if err != nil { writeError(w, r, http.StatusInternalServerError, codeInternal, err.Error()) return @@ -353,7 +364,7 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { } } default: - if err := s.createTable(r, bucket, ns, name, location); err != nil { + if location, err = s.createTable(r, bucket, ns, name, location); err != nil { writeStorageError(w, r, err) return } @@ -545,7 +556,9 @@ func (s *Server) loadLanceTable(r *http.Request, bucket string, ns []string, nam return &lanceTable{location: location, versionToken: resp.VersionToken}, nil } -func (s *Server) createTable(r *http.Request, bucket string, ns []string, name, location string) error { +// createTable returns the location the catalog holds for the table. When the +// name already exists that is its own location, not the one asked for. +func (s *Server) createTable(r *http.Request, bucket string, ns []string, name, location string) (string, error) { req := &s3tables.CreateTableRequest{ TableBucketARN: bucketARN(bucket), Namespace: ns, @@ -556,7 +569,13 @@ func (s *Server) createTable(r *http.Request, bucket string, ns []string, name, MetadataLocation: location, } var resp s3tables.CreateTableResponse - return s.execute(r, "CreateTable", req, &resp) + if err := s.execute(r, "CreateTable", req, &resp); err != nil { + return "", err + } + if resp.MetadataLocation == "" { + return tableLocation(bucket, ns, name), nil + } + return resp.MetadataLocation, nil } // repointTable moves an existing entry to another dataset location without diff --git a/weed/s3api/lance/handlers_test.go b/weed/s3api/lance/handlers_test.go index d62bdc7089..72d9b84f68 100644 --- a/weed/s3api/lance/handlers_test.go +++ b/weed/s3api/lance/handlers_test.go @@ -415,3 +415,46 @@ func TestDataPlaneIsUnsupported(t *testing.T) { t.Fatalf("error code = %d, want %d", got.Code, codeUnsupported) } } + +// createOnlyAuthenticator signs in an identity that may create and read tables +// but not update them. +type createOnlyAuthenticator struct{} + +func (createOnlyAuthenticator) AuthenticateRequest(*http.Request) (string, interface{}, s3err.ErrorCode) { + return "creator", &struct { + Name string + Actions []string + }{Name: "creator", Actions: []string{"s3tables:CreateTable", "s3tables:GetTable"}}, s3err.ErrNone +} + +func (createOnlyAuthenticator) DefaultAllow() bool { return false } + +// The right to create a name is the write evidence for a table the declare +// creates, but not for one it only finds. +func TestDeclareVendsWriteOnlyForATableItCreates(t *testing.T) { + h := newTestHarness(t) + h.createBucket(t, "lnc", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/namespace/lnc$ns/create", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$theirs/declare", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$theirs/deregister", `{}`, http.StatusOK) + + server := NewServer(s3tables.NewManagerClient(h.filer.Client), createOnlyAuthenticator{}) + server.SetCredentialVendor(recordingVendor{}) + router := mux.NewRouter().SkipClean(true) + server.RegisterRoutes(router) + creator := &testHarness{router: router, filer: h.filer, admin: h.admin, server: server} + + for _, tc := range []struct{ name, want string }{ + {"theirs", "read-only"}, + {"mine", "read-write"}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := creator.vendedKey(t, "/v1/table/lnc$ns$"+tc.name+"/declare", ""); got != tc.want { + t.Errorf("vended %q, want %q", got, tc.want) + } + }) + } + if h.filer.Get("/buckets/lnc/ns/theirs", deregisteredMarker) == nil { + t.Error("a declare that may not write the table brought it back") + } +} diff --git a/weed/s3api/lance/server.go b/weed/s3api/lance/server.go index 53effee9e5..5d25e8c860 100644 --- a/weed/s3api/lance/server.go +++ b/weed/s3api/lance/server.go @@ -2,6 +2,8 @@ package lance import ( "context" + "crypto/hmac" + "crypto/sha256" "net/http" "strings" "time" @@ -12,6 +14,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/security" ) // FilerClient provides access to the filer for storage operations. @@ -61,6 +64,7 @@ type Server struct { credentialVendor CredentialVendor s3Endpoint string s3Region string + signingKey func() security.SigningKey } // NewServer creates a Lance namespace server over the given filer. @@ -89,6 +93,13 @@ func (s *Server) SetCredentialVendor(vendor CredentialVendor) { s.credentialVendor = vendor } +// SetSigningKey sets where to read the cluster key (jwt.filer_signing.key) +// that a trusted caller's clusterTokenHeader is checked against. It is read on +// every check, so a reloaded key takes effect at once. +func (s *Server) SetSigningKey(signingKey func() security.SigningKey) { + s.signingKey = signingKey +} + // SetS3Endpoint configures the S3 endpoint advertised in storage_options so a // client can reach the dataset without separately discovering the S3 address. func (s *Server) SetS3Endpoint(endpoint string) { @@ -171,6 +182,9 @@ func loggingMiddleware(next http.Handler) http.Handler { // header; the S3 authenticator stays for callers that can SigV4-sign. func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { + if s.hasClusterToken(r) { + r = r.WithContext(withTrustedCaller(r.Context())) + } // A request carrying a Bearer token is a Lance REST client. An invalid // or expired token gets 401 immediately; falling through to the S3 // authenticator would parse the header as SigV4 and fail with a @@ -250,3 +264,46 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc { handler(w, r) } } + +// clusterTokenHeader carries a token signed with clusterTokenKey. The vector +// engine and the Lance maintenance worker send it next to their identity's +// credentials, and only they may write index data in a vector bucket. +const clusterTokenHeader = "X-Seaweedfs-Cluster-Token" + +// clusterTokenKey derives the key cluster tokens are signed with from the +// cluster key, so a token captured on its way here does not authorize the +// admin calls that take the cluster key itself. +func clusterTokenKey(key security.SigningKey) security.SigningKey { + mac := hmac.New(sha256.New, key) + mac.Write([]byte("seaweedfs-lance-cluster-token")) + return mac.Sum(nil) +} + +type trustedCallerKey struct{} + +func withTrustedCaller(ctx context.Context) context.Context { + return context.WithValue(ctx, trustedCallerKey{}, true) +} + +func isTrustedCaller(ctx context.Context) bool { + trusted, _ := ctx.Value(trustedCallerKey{}).(bool) + return trusted +} + +// hasClusterToken reports whether the request proves it holds the cluster key. +// Without a configured key no caller can. +func (s *Server) hasClusterToken(r *http.Request) bool { + token := r.Header.Get(clusterTokenHeader) + if token == "" || s.signingKey == nil { + return false + } + key := s.signingKey() + if len(key) == 0 { + return false + } + if security.IsFilerAdminToken(clusterTokenKey(key), security.EncodedJwt(token)) { + return true + } + glog.V(1).Infof("lance: %s from %s is not signed with a key derived from jwt.filer_signing.key", clusterTokenHeader, r.RemoteAddr) + return false +} diff --git a/weed/s3api/lance/storage.go b/weed/s3api/lance/storage.go index c4ccd67385..52a35640b5 100644 --- a/weed/s3api/lance/storage.go +++ b/weed/s3api/lance/storage.go @@ -177,6 +177,9 @@ func (s *Server) storageOptions(r *http.Request, bucket, location string, vend b // credentials would cover, the same gate the Iceberg catalog applies so a // read-only caller never gets write-capable storage_options. func (s *Server) mayUpdateTable(r *http.Request, bucket string, ns []string, name string) bool { + if !s.mayWriteNamespace(r, bucket, ns) { + return false + } req := &s3tables.UpdateTableRequest{ TableBucketARN: bucketARN(bucket), Namespace: ns, @@ -192,6 +195,20 @@ func (s *Server) mayUpdateTable(r *http.Request, bucket string, ns []string, nam return err == nil } +// mayWriteNamespace keeps the vector namespace of a vector bucket out of reach +// of s3tables grants: its index data is written only by trusted callers, and +// the vector bucket policy gates everyone else through S3 Vectors. +func (s *Server) mayWriteNamespace(r *http.Request, bucket string, ns []string) bool { + if isTrustedCaller(r.Context()) { + return true + } + err := s.tablesManager.CheckVectorNamespaceWrite(r.Context(), s.filerClient, bucket, ns) + if err != nil { + glog.V(1).Infof("lance: %s/%s is read-only to this caller: %v", bucket, strings.Join(ns, "."), err) + } + return err == nil +} + // locationPrefix strips the s3://bucket/ part of a location, leaving the key // prefix a credential is scoped to. func locationPrefix(location string) string { diff --git a/weed/s3api/lance/vector_namespace_test.go b/weed/s3api/lance/vector_namespace_test.go index 01ec1770a8..3185684df3 100644 --- a/weed/s3api/lance/vector_namespace_test.go +++ b/weed/s3api/lance/vector_namespace_test.go @@ -1,10 +1,14 @@ package lance import ( + "context" "net/http" + "net/http/httptest" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/security" ) // seedVectorIndex creates bucket/vector/idx through the trusted path, the way @@ -75,3 +79,87 @@ func TestVectorNamespaceOfUndeclaredBucketIsWritableOverLance(t *testing.T) { h.mustDo(t, http.MethodPost, idx+"/register", `{"location":"s3://old/vector/idx"}`, http.StatusOK) h.mustDo(t, http.MethodPost, idx+"/drop", `{}`, http.StatusOK) } + +// recordingVendor vends a credential whose key says whether it may write and +// whose secret is the prefix it covers. +type recordingVendor struct{} + +func (recordingVendor) VendTableCredentials(_ context.Context, _, _, prefix string, readOnly bool) (*VendedCredentials, error) { + if readOnly { + return &VendedCredentials{AccessKeyID: "read-only", SecretAccessKey: prefix}, nil + } + return &VendedCredentials{AccessKeyID: "read-write", SecretAccessKey: prefix}, nil +} + +func (h *testHarness) vendedKey(t *testing.T, target, token string) string { + t.Helper() + req := httptest.NewRequest(http.MethodPost, target, strings.NewReader(`{"vend_credentials":true}`)) + if token != "" { + req.Header.Set(clusterTokenHeader, token) + } + rec := httptest.NewRecorder() + h.router.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("POST %s = %d (%s)", target, rec.Code, rec.Body.String()) + } + return decode[DescribeTableResponse](t, rec).StorageOptions["aws_access_key_id"] +} + +// A caller holding every s3tables grant still gets read-only credentials to a +// vector bucket's index data; only the cluster's own components can write it. +func TestVectorNamespaceVendsWriteCredentialsOnlyToTrustedCallers(t *testing.T) { + h := newTestHarness(t) + key := security.SigningKey("cluster-key") + h.server.SetSigningKey(func() security.SigningKey { return key }) + h.server.SetCredentialVendor(recordingVendor{}) + h.seedVectorIndex(t, "vec", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/table/vec$other$tbl/declare", `{}`, http.StatusOK) + h.seedVectorIndex(t, "old", "") + + h.vendedKey(t, "/v1/table/vec$vector$idx/declare", "") + if h.filer.Get("/buckets/vec/vector/idx", reservedMarker) != nil { + t.Error("an untrusted declare wrote into the index") + } + + trusted := string(security.GenJwtForFilerAdmin(clusterTokenKey(key), 60)) + forged := string(security.GenJwtForFilerAdmin(key, 60)) + for _, tc := range []struct { + name, target, token, want string + }{ + {"describe the index", "/v1/table/vec$vector$idx/describe", "", "read-only"}, + {"declare the index", "/v1/table/vec$vector$idx/declare", "", "read-only"}, + {"describe with an admin token", "/v1/table/vec$vector$idx/describe", forged, "read-only"}, + {"trusted describe", "/v1/table/vec$vector$idx/describe", trusted, "read-write"}, + {"trusted declare", "/v1/table/vec$vector$idx/declare", trusted, "read-write"}, + {"another namespace", "/v1/table/vec$other$tbl/describe", "", "read-write"}, + {"bucket from before formats", "/v1/table/old$vector$idx/describe", "", "read-write"}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := h.vendedKey(t, tc.target, tc.token); got != tc.want { + t.Errorf("vended %q, want %q", got, tc.want) + } + }) + } +} + +// Declaring a name that already exists answers for that table, so naming the +// index's location on another table gets no credentials to the index. +func TestDeclareOfAnExistingTableKeepsItsLocation(t *testing.T) { + h := newTestHarness(t) + h.server.SetCredentialVendor(recordingVendor{}) + h.seedVectorIndex(t, "vec", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/table/vec$other$tbl/declare", `{}`, http.StatusOK) + + rec := h.mustDo(t, http.MethodPost, "/v1/table/vec$other$tbl/declare", + `{"location":"s3://vec/vector/idx","vend_credentials":true}`, http.StatusOK) + resp := decode[DeclareTableResponse](t, rec) + if resp.Location != "s3://vec/other/tbl" { + t.Errorf("location = %q, want the table's own s3://vec/other/tbl", resp.Location) + } + if got := resp.StorageOptions["aws_secret_access_key"]; got != "other/tbl" { + t.Errorf("credentials cover %q, want other/tbl", got) + } + if h.filer.Get("/buckets/vec/vector/idx", reservedMarker) != nil { + t.Error("the declare wrote into the index") + } +} diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 8bd16cb1ec..281bbaa6ef 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -1500,6 +1500,15 @@ func (s3a *S3ApiServer) AuthenticateRequest(r *http.Request) (string, interface{ return "", nil, err } +// FilerSigningKey is the current jwt.filer_signing.key, which follows a +// security.toml reload. +func (s3a *S3ApiServer) FilerSigningKey() security.SigningKey { + if s3a.filerGuard == nil { + return nil + } + return s3a.filerGuard.SigningKey() +} + // DefaultAllow returns whether access is allowed by default when no policy is found func (s3a *S3ApiServer) DefaultAllow() bool { if s3a.iam == nil || s3a.iam.iamIntegration == nil { diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index a9df3373d8..10edb2e94d 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -188,8 +188,9 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName) resp := &CreateTableResponse{ - TableARN: tableARN, - VersionToken: versionToken, + TableARN: tableARN, + VersionToken: versionToken, + MetadataLocation: metadata.MetadataLocation, } h.writeJSON(w, http.StatusOK, resp) diff --git a/weed/security/grpc_peer.go b/weed/security/grpc_peer.go index a6b1fba95e..9a2844067d 100644 --- a/weed/security/grpc_peer.go +++ b/weed/security/grpc_peer.go @@ -38,16 +38,26 @@ func CheckGrpcAdminPeer(ctx context.Context, signingKey SigningKey) error { if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") || parts[1] == "" { return status.Error(codes.Unauthenticated, "authorization header must use Bearer scheme") } - parsed, err := DecodeJwt(signingKey, EncodedJwt(parts[1]), &SeaweedFilerAdminClaims{}) - if err != nil || parsed == nil || !parsed.Valid { - return status.Error(codes.Unauthenticated, "invalid admin token") - } - if claims, ok := parsed.Claims.(*SeaweedFilerAdminClaims); !ok || claims.SessionId != "" { + if !IsFilerAdminToken(signingKey, EncodedJwt(parts[1])) { return status.Error(codes.Unauthenticated, "invalid admin token") } return nil } +// IsFilerAdminToken reports whether token is an admin token signed by +// signingKey. An STS session token signed with the same key does not count. +func IsFilerAdminToken(signingKey SigningKey, token EncodedJwt) bool { + if len(signingKey) == 0 || token == "" { + return false + } + parsed, err := DecodeJwt(signingKey, token, &SeaweedFilerAdminClaims{}) + if err != nil || parsed == nil || !parsed.Valid { + return false + } + claims, ok := parsed.Claims.(*SeaweedFilerAdminClaims) + return ok && claims.SessionId == "" +} + // GrpcAdminPeerCredentials returns outgoing metadata carrying a Bearer token // for admin-plane gRPC handlers, or nil when no signing key is configured. func GrpcAdminPeerCredentials(signingKey SigningKey, expiresAfterSec int) metadata.MD {