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 {