diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index aa42defcc7..42bbc9ca3b 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -1,6 +1,7 @@ -use anyhow::Result; +use anyhow::{Context, Result}; use clap::Parser; use tonic::transport::Server; +use tonic::transport::server::TcpIncoming; use weed_lance_worker::catalog::{NamespaceClient, SigningKey}; use weed_lance_worker::dataset::FallbackOptions; use weed_vector_server::pb::vector::vector_service_server::VectorServiceServer; @@ -10,15 +11,15 @@ use weed_vector_server::service::VectorServer; /// The VectorService the S3 gateway fans vector calls out to. #[derive(Parser)] struct Args { - /// Lance Namespace base URL, e.g. http://localhost:8888/lance. + /// Lance Namespace base URL: the S3 gateway's Lance port (-s3.port.lance). #[arg( long, env = "WEED_VECTOR_NAMESPACE", - default_value = "http://localhost:8888/lance" + default_value = "http://localhost:9101" )] namespace: String, /// gRPC listen address. - #[arg(long, env = "WEED_VECTOR_LISTEN", default_value = "127.0.0.1:8080")] + #[arg(long, env = "WEED_VECTOR_LISTEN", default_value = "127.0.0.1:8090")] listen: String, /// object_store option fallback for deployments that vend no credentials: /// key=value, repeatable. @@ -104,6 +105,11 @@ fn advertised_addr(listen: &str, masters: &[String]) -> String { async fn main() -> Result<()> { tracing_subscriber::fmt::init(); let args = Args::parse_from(normalize_args(std::env::args().collect())); + // Bind before registering, so the master never lists an engine that is + // about to exit because its port is taken. + let incoming = TcpIncoming::bind(args.listen.parse()?) + .with_context(|| format!("listen on {}", args.listen))? + .with_nodelay(Some(true)); if let Some(masters) = &args.masters { let masters: Vec = masters.split(',').map(|s| s.trim().to_string()).collect(); for master in &masters { @@ -132,7 +138,7 @@ async fn main() -> Result<()> { .max_decoding_message_size(1 << 30) .max_encoding_message_size(1 << 30), ) - .serve(args.listen.parse()?) + .serve_with_incoming(incoming) .await?; Ok(()) } diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index ea26f6c9bf..8ccf041c2b 100644 --- a/seaweed-vector/crates/server/src/service.rs +++ b/seaweed-vector/crates/server/src/service.rs @@ -430,8 +430,7 @@ impl VectorService for VectorServer { .describe_table(&table.id) .await .map_err(internal)?; - let mut options = self.fallback.clone(); - options.extend(description.storage_options.clone()); + let options = dataset::table_options(&description, &self.fallback); let schema = schema_for(dimension, metric, &req.non_filterable_metadata_keys); let empty = RecordBatch::new_empty(schema.clone()); let reader = arrow_array::RecordBatchIterator::new(vec![Ok(empty)].into_iter(), schema); diff --git a/seaweed-worker/crates/lance/src/dataset.rs b/seaweed-worker/crates/lance/src/dataset.rs index d28bcf2d4c..cdca46d0bd 100644 --- a/seaweed-worker/crates/lance/src/dataset.rs +++ b/seaweed-worker/crates/lance/src/dataset.rs @@ -65,21 +65,38 @@ pub async fn open_with( }) } -/// The namespace vends object_store's own option names, so they pass straight -/// through. What it vends always wins: the fallback exists because a deployment -/// without STS vends no credentials at all, and then the worker has no other way -/// to reach the data. +/// The object_store options for a table. The namespace vends object_store's own +/// option names, so they pass straight through. What it vends always wins: the +/// fallback exists because a deployment without STS vends no credentials at +/// all, and then the worker has no other way to reach the data. +pub fn table_options( + description: &TableDescription, + fallback: &FallbackOptions, +) -> HashMap { + let mut options: HashMap = fallback.clone(); + options.extend(description.storage_options.clone()); + // A store that takes unsigned requests needs no credentials, but without + // any lance builds the AWS default chain, and that probes instance + // metadata on every open. + if options + .get("aws_skip_signature") + .is_some_and(|v| v == "true") + { + options.entry("aws_access_key_id".into()).or_default(); + options.entry("aws_secret_access_key".into()).or_default(); + } + options +} + async fn open_at( description: &TableDescription, fallback: &FallbackOptions, wrapper: Option>, ) -> Result { - let mut options: HashMap = fallback.clone(); - options.extend(description.storage_options.clone()); DatasetBuilder::from_uri(&description.location) .with_store_params(ObjectStoreParams { storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( - options, + table_options(description, fallback), ))), object_store_wrapper: wrapper, ..Default::default() @@ -102,14 +119,12 @@ pub async fn drop( Err(err) => return Err(err), }; check_drop_location(id, &description.location)?; - let mut options: HashMap = fallback.clone(); - options.extend(description.storage_options.clone()); let (store, path) = ObjectStore::from_uri_and_params( Arc::new(ObjectStoreRegistry::default()), &description.location, &ObjectStoreParams { storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( - options, + table_options(&description, fallback), ))), ..Default::default() }, @@ -203,12 +218,45 @@ fn schema_json(dataset: &Dataset) -> Option { #[cfg(test)] mod tests { - use super::{check_drop_location, table_location}; + use super::{FallbackOptions, check_drop_location, table_location, table_options}; + use crate::catalog::TableDescription; fn id(parts: &[&str]) -> Vec { parts.iter().map(|part| part.to_string()).collect() } + fn options(pairs: &[(&str, &str)]) -> FallbackOptions { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + } + + #[test] + fn table_options_keep_an_unsigned_store_off_the_aws_default_chain() { + let described = |vended: &[(&str, &str)]| TableDescription { + location: "s3://bkt/vector/x".into(), + storage_options: options(vended), + managed_versioning: false, + }; + let unsigned = described(&[("aws_skip_signature", "true")]); + + let got = table_options(&unsigned, &FallbackOptions::new()); + assert_eq!(got.get("aws_access_key_id").map(String::as_str), Some("")); + assert_eq!( + got.get("aws_secret_access_key").map(String::as_str), + Some("") + ); + + let fallback = options(&[("aws_access_key_id", "k"), ("aws_secret_access_key", "s")]); + let got = table_options(&unsigned, &fallback); + assert_eq!(got.get("aws_access_key_id").map(String::as_str), Some("k")); + + let signed = described(&[("aws_endpoint", "http://s3:8333")]); + let got = table_options(&signed, &FallbackOptions::new()); + assert!(!got.contains_key("aws_access_key_id"), "{got:?}"); + } + #[test] fn table_location_matches_the_gateway_layout() { assert_eq!( diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index dfb5ab116d..69f4831bba 100644 --- a/test/s3vectors/conftest.py +++ b/test/s3vectors/conftest.py @@ -70,8 +70,18 @@ def stop(process): @pytest.fixture(scope="module", params=combinations(), ids=lambda p: "-".join(p)) def cluster(request, tmp_path_factory): - volume, mode = request.param - directory = tmp_path_factory.mktemp(f"vectors-{volume}-{mode}") + yield from run_cluster(request.param, tmp_path_factory, identities=True) + + +@pytest.fixture(scope="module", params=combinations(), ids=lambda p: "-".join(p)) +def open_cluster(request, tmp_path_factory): + """A cluster with no S3 identities, and an engine given no credentials.""" + yield from run_cluster(request.param, tmp_path_factory, identities=False) + + +def run_cluster(param, tmp_path_factory, identities): + volume, mode = param + directory = tmp_path_factory.mktemp(f"vectors-{volume}-{mode}" + ("" if identities else "-open")) print(f"\nVector integration logs: {directory}") weed = binary("WEED_BINARY", "weed") engine = binary("VECTOR_BINARY", "weed-vector") @@ -89,10 +99,12 @@ def cluster(request, tmp_path_factory): master_addr = f"127.0.0.1:{master}.{master_grpc}" filer_addr = f"127.0.0.1:{filer}.{filer_grpc}" - def start(name, args): + def start(name, args, unset=()): log = stack.enter_context((directory / f"{name}.log").open("w")) env = dict(os.environ, AWS_ACCESS_KEY_ID="test", AWS_SECRET_ACCESS_KEY="test", AWS_DEFAULT_REGION="us-east-1", AWS_EC2_METADATA_DISABLED="true") + for key in unset: + env.pop(key) process = subprocess.Popen(args, cwd=directory, stdout=log, stderr=log, env=env) processes[name] = process stack.callback(stop, process) @@ -117,14 +129,20 @@ def cluster(request, tmp_path_factory): f"-defaultStoreDir={directory / 'filer'}"]) wait_for(lambda: listening(filer_grpc), processes, "filer not ready") endpoint = f"http://127.0.0.1:{s3}" - start("s3", [weed, "-v=1", "s3", f"-filer={filer_addr}", f"-config={HERE / 's3.json'}", + # weed s3 turns AWS_ACCESS_KEY_ID into an admin identity, and the engine + # would find it as a credential. + if identities: + config, unset = [f"-config={HERE / 's3.json'}"], () + credentials = ["--storage-option", "aws_access_key_id=test", + "--storage-option", "aws_secret_access_key=test"] + else: + config, unset, credentials = [], ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"), [] + start("s3", [weed, "-v=1", "s3", f"-filer={filer_addr}", *config, "-ip=127.0.0.1", "-ip.bind=127.0.0.1", f"-port={s3}", - f"-port.grpc={s3_grpc}", "-port.iceberg=0", f"-port.lance={lance}"]) + f"-port.grpc={s3_grpc}", "-port.iceberg=0", f"-port.lance={lance}"], unset) wait_for(lambda: http_ready(f"http://127.0.0.1:{lance}/v1/namespace/root/list"), processes, "catalog not ready") start("vector", [engine, "--namespace", f"http://127.0.0.1:{lance}", - "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", - "--storage-option", "aws_access_key_id=test", - "--storage-option", "aws_secret_access_key=test"]) + "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", *credentials], unset) wait_for(lambda: listening(vector), processes, "vector engine not ready") wait_for( diff --git a/test/s3vectors/s3vectors_test.py b/test/s3vectors/s3vectors_test.py index 53ce40c4dd..adbeb1c246 100644 --- a/test/s3vectors/s3vectors_test.py +++ b/test/s3vectors/s3vectors_test.py @@ -5,6 +5,7 @@ import time import uuid import boto3 +from botocore.config import Config from botocore.exceptions import ClientError import pytest @@ -415,3 +416,17 @@ def test_query_without_writer(vectors, index, cluster): raise time.sleep(0.2) assert [v["key"] for v in result["vectors"]] == ["survivor"] + + +def test_engine_without_credentials(open_cluster): + """With no S3 identities, an engine given no credentials still serves an index.""" + client = boto3.client("s3vectors", endpoint_url=open_cluster["endpoint"], aws_access_key_id="any", + aws_secret_access_key="any", region_name="us-east-1", + config=Config(connect_timeout=3, read_timeout=20, retries={"max_attempts": 0})) + index = {"vectorBucketName": BUCKET, "indexName": INDEX} + client.create_vector_bucket(vectorBucketName=BUCKET) + client.create_index(**index, dataType="float32", dimension=2, distanceMetric="cosine") + client.put_vectors(**index, vectors=[{"key": "a", "data": {"float32": [1, 0]}}, + {"key": "b", "data": {"float32": [0, 1]}}]) + hits = client.query_vectors(**index, queryVector={"float32": [1, 0]}, topK=1)["vectors"] + assert [h["key"] for h in hits] == ["a"] diff --git a/weed/s3api/lance/handlers_oauth_test.go b/weed/s3api/lance/handlers_oauth_test.go index 236b101f32..c16e47ce0e 100644 --- a/weed/s3api/lance/handlers_oauth_test.go +++ b/weed/s3api/lance/handlers_oauth_test.go @@ -49,8 +49,9 @@ func newTestServerWithOAuth() *Server { } type mockS3Authenticator struct { - called bool - errCode s3err.ErrorCode + called bool + errCode s3err.ErrorCode + authDisabled bool } func (m *mockS3Authenticator) AuthenticateRequest(r *http.Request) (string, interface{}, s3err.ErrorCode) { @@ -63,6 +64,8 @@ func (m *mockS3Authenticator) AuthenticateRequest(r *http.Request) (string, inte func (m *mockS3Authenticator) DefaultAllow() bool { return false } +func (m *mockS3Authenticator) AuthEnabled() bool { return !m.authDisabled } + func mintTestToken(t *testing.T, accessKey, secret string, issuedAt, expiresAt time.Time) string { t.Helper() key := deriveSigningKey(accessKey, secret) @@ -316,6 +319,32 @@ func TestAuthApiKeyInvalid(t *testing.T) { } } +// Without identities the S3 port takes any signature, so the catalog takes any +// key or token and leaves the caller to the S3 authenticator. +func TestAuthTakesAnyKeyWhenS3AuthIsOff(t *testing.T) { + s := newTestServerWithOAuth() + s.authenticator = &mockS3Authenticator{authDisabled: true} + + for header, value := range map[string]string{ + "x-api-key": "unknown:key", + "Authorization": "Bearer not-a-token", + } { + var gotIdentity string + handler := s.Auth(func(w http.ResponseWriter, r *http.Request) { + gotIdentity = s3_constants.GetIdentityNameFromContext(r) + w.WriteHeader(http.StatusOK) + }) + req := httptest.NewRequest(http.MethodGet, "/v1/namespace/$/list", nil) + req.Header.Set(header, value) + rec := httptest.NewRecorder() + handler(rec, req) + + if rec.Code != http.StatusOK || gotIdentity != "s3user" { + t.Fatalf("%s: status = %d, identity = %q, want 200 as s3user", header, rec.Code, gotIdentity) + } + } +} + func TestAuthNoBearerStillUsesS3Authenticator(t *testing.T) { s := newTestServerWithOAuth() auth := &mockS3Authenticator{errCode: s3err.ErrNone} diff --git a/weed/s3api/lance/handlers_test.go b/weed/s3api/lance/handlers_test.go index 72d9b84f68..037f66ff83 100644 --- a/weed/s3api/lance/handlers_test.go +++ b/weed/s3api/lance/handlers_test.go @@ -24,6 +24,14 @@ func (openAuthenticator) AuthenticateRequest(*http.Request) (string, interface{} func (openAuthenticator) DefaultAllow() bool { return true } +func (openAuthenticator) AuthEnabled() bool { return false } + +// signedAuthenticator admits the same caller on a gateway that checks +// credentials. +type signedAuthenticator struct{ openAuthenticator } + +func (signedAuthenticator) AuthEnabled() bool { return true } + type testHarness struct { router *mux.Router filer *s3tablestest.MemFiler @@ -134,6 +142,28 @@ func decode[T any](t *testing.T, recorder *httptest.ResponseRecorder) T { return out } +// While the S3 port takes unsigned requests the catalog says so, and a client +// with no credentials of its own still reaches the data. Once auth is on it +// has to sign again. +func TestStorageOptionsSkipSignatureOnlyWhileS3AuthIsOff(t *testing.T) { + h := newTestHarness(t) + h.createBucket(t, "open", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/namespace/open$ns/create", `{}`, http.StatusOK) + + declared := decode[DeclareTableResponse](t, + h.mustDo(t, http.MethodPost, "/v1/table/open$ns$t/declare", `{}`, http.StatusOK)) + if declared.StorageOptions["aws_skip_signature"] != "true" { + t.Fatalf("auth off: storage options = %v, want aws_skip_signature", declared.StorageOptions) + } + + h.server.authenticator = signedAuthenticator{} + described := decode[DescribeTableResponse](t, + h.mustDo(t, http.MethodPost, "/v1/table/open$ns$t/describe", `{}`, http.StatusOK)) + if _, ok := described.StorageOptions["aws_skip_signature"]; ok { + t.Fatalf("auth on: storage options = %v, want no aws_skip_signature", described.StorageOptions) + } +} + // The lifecycle a Lance client drives: create the namespace, declare the table // before any data exists, resolve it to a location, then deregister and bring it // back by registering the same location. @@ -429,6 +459,8 @@ func (createOnlyAuthenticator) AuthenticateRequest(*http.Request) (string, inter func (createOnlyAuthenticator) DefaultAllow() bool { return false } +func (createOnlyAuthenticator) AuthEnabled() bool { return true } + // 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) { diff --git a/weed/s3api/lance/server.go b/weed/s3api/lance/server.go index 5d25e8c860..286205a70c 100644 --- a/weed/s3api/lance/server.go +++ b/weed/s3api/lance/server.go @@ -25,6 +25,8 @@ type FilerClient interface { type S3Authenticator interface { AuthenticateRequest(r *http.Request) (string, interface{}, s3err.ErrorCode) DefaultAllow() bool + // AuthEnabled reports whether the S3 server checks credentials at all. + AuthEnabled() bool } // CredentialValidator validates S3 access key / secret key pairs @@ -177,6 +179,13 @@ func loggingMiddleware(next http.Handler) http.Handler { }) } +// authEnabled is false when the S3 server has no identities and takes any +// signature. The catalog then takes any key too, and leaves the caller to the +// S3 authenticator. +func (s *Server) authEnabled() bool { + return s.authenticator == nil || s.authenticator.AuthEnabled() +} + // Auth authenticates the caller and puts the identity in the request context. // Lance clients authenticate the catalog with a Bearer token or an x-api-key // header; the S3 authenticator stays for callers that can SigV4-sign. @@ -190,7 +199,7 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc { // authenticator would parse the header as SigV4 and fail with a // different error that clients cannot act on. // The auth scheme is case-insensitive (RFC 7235). - if strings.HasPrefix(strings.ToLower(r.Header.Get("Authorization")), "bearer ") { + if s.authEnabled() && strings.HasPrefix(strings.ToLower(r.Header.Get("Authorization")), "bearer ") { if identityName, identity, accessKey, secretKey, ok := s.authenticateBearerKey(r); ok { ctx := r.Context() ctx = s3_constants.SetIdentityNameInContext(ctx, identityName) @@ -208,7 +217,7 @@ func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc { return } - if apiKey := r.Header.Get("x-api-key"); apiKey != "" { + if apiKey := r.Header.Get("x-api-key"); apiKey != "" && s.authEnabled() { if identityName, identity, ok := s.authenticateApiKey(apiKey); ok { accessKey, secretKey, _ := strings.Cut(apiKey, ":") ctx := r.Context() diff --git a/weed/s3api/lance/storage.go b/weed/s3api/lance/storage.go index 52a35640b5..e719680118 100644 --- a/weed/s3api/lance/storage.go +++ b/weed/s3api/lance/storage.go @@ -149,6 +149,11 @@ func (s *Server) storageOptions(r *http.Request, bucket, location string, vend b if s.s3Region != "" { options["aws_region"] = s.s3Region } + if !s.authEnabled() { + // The S3 port takes unsigned requests, and a client with no + // credentials of its own would otherwise fail looking for some. + options["aws_skip_signature"] = "true" + } if !vend || s.credentialVendor == nil { return options, nil diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 281bbaa6ef..b6f8c1486a 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -1509,6 +1509,12 @@ func (s3a *S3ApiServer) FilerSigningKey() security.SigningKey { return s3a.filerGuard.SigningKey() } +// AuthEnabled reports whether requests must carry valid credentials. With no +// identities configured, every request is allowed as the admin. +func (s3a *S3ApiServer) AuthEnabled() bool { + return s3a.iam == nil || s3a.iam.isEnabled() +} + // 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/s3api_tables_auth_test.go b/weed/s3api/s3api_tables_auth_test.go index 6ab18d59d1..85080f5a69 100644 --- a/weed/s3api/s3api_tables_auth_test.go +++ b/weed/s3api/s3api_tables_auth_test.go @@ -159,6 +159,34 @@ func TestLanceAuthRejectsFailedSignature(t *testing.T) { assert.NotEqual(t, http.StatusOK, rr.Code) } +// The Lance catalog takes any x-api-key while the S3 port runs without +// identities, as the admin, and refuses an unknown key once auth is on. +func TestLanceAuthTakesAnyKeyOnlyWhileS3AuthIsOff(t *testing.T) { + s3a := &S3ApiServer{iam: &IdentityAccessManagement{}} + server := lance.NewServer(nil, s3a) + server.SetCredentialValidator(s3a) + + call := func() (int, string) { + req := httptest.NewRequest(http.MethodGet, "/v1/namespace/list", nil) + req.Header.Set("x-api-key", "unknown:key") + identity := "" + rr := httptest.NewRecorder() + server.Auth(func(w http.ResponseWriter, r *http.Request) { + identity = s3_constants.GetIdentityNameFromContext(r) + w.WriteHeader(http.StatusOK) + })(rr, req) + return rr.Code, identity + } + + code, identity := call() + assert.Equal(t, http.StatusOK, code) + assert.Equal(t, s3_constants.AccountAdminId, identity) + + s3a.iam.isAuthEnabled = true + code, _ = call() + assert.Equal(t, http.StatusUnauthorized, code) +} + func TestAuthSignatureOnlyScrubsAccountHeader(t *testing.T) { s3a := setupDefaultAllowAuthServer(t)