From 6c473433c595efd1fcc4f42c31bfdaa9023830a1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 10 Nov 2026 04:21:56 -0800 Subject: [PATCH] vector: bound request buffering before authentication and document installation (#12537) * s3vectors: bound body buffering before authentication * docs: include vector engine installation and source builds * s3vectors: prevent MIME sniffing of JSON error responses * docs: load vector Cargo configuration when building from source * docs: note the native protoc requirement for Windows vector builds * docs: make vector source checkout and working directory explicit * ci: handle deb822 sources during Ubuntu mirror failover * ci: bound Ubuntu mirror attempts and reject partial index updates --- .github/workflows/e2e.yml | 3 + README.md | 20 ++++- docker/apt-install | 18 ++--- docker/test_apt_install.py | 110 ++++++++++++++++++++++++++ weed/command/s3.go | 1 + weed/s3api/s3vectors/limits.go | 13 ++++ weed/s3api/s3vectors/limits_test.go | 117 ++++++++++++++++++++++++++++ weed/s3api/s3vectors/server.go | 45 ++++++++--- weed/s3api/s3vectors/types.go | 1 + 9 files changed, 308 insertions(+), 20 deletions(-) create mode 100644 docker/test_apt_install.py diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index ecc3756edc..07e38dc481 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -61,6 +61,9 @@ jobs: restore-keys: | ${{ runner.os }}-buildx-e2e- + - name: Test APT mirror fallback + run: python3 test_apt_install.py + - name: Install dependencies run: | sudo rm -f /etc/apt/sources.list.d/azure-cli.list /etc/apt/sources.list.d/microsoft-prod.list diff --git a/README.md b/README.md index dee4858bdf..aa8f58f83a 100644 --- a/README.md +++ b/README.md @@ -244,7 +244,25 @@ Keep embeddings alongside your objects and tables in the same SeaweedFS cluster. * Find the nearest vectors with cosine or Euclidean distance, filter by metadata, and return distances and metadata with the results. * Control access with `s3vectors:*` IAM actions and vector bucket policies; tag buckets and indexes for organization and access control. -Run the separate `weed-vector` binary alongside the cluster. For the local cluster started with `weed mini` above: +Install the separate vector engine; the default installer installs only `weed`: + +```bash +curl -fsSL https://raw.githubusercontent.com/seaweedfs/seaweedfs/master/install.sh | bash -s -- --component vector +``` + +Release 4.53 provides vector binaries for Linux amd64 and arm64. If your release has no binary for your platform, build from [source](seaweed-vector). From a directory where you want a new checkout: + +```bash +git clone https://github.com/seaweedfs/seaweedfs.git +cd seaweedfs +(cd seaweed-vector && cargo build --release -p weed-vector-server) +``` + +If you already have a checkout, start at its root and skip the clone and `cd seaweedfs` steps. + +For Windows source builds, set `PROTOC` to a native `protoc.exe` before running Cargo; the [desktop build workflow](.github/workflows/rust-vector-platforms.yml) shows how to select the vendored executable. + +Run the installed `weed-vector` binary alongside the local cluster started with `weed mini` above (or use `seaweed-vector/target/release/weed-vector` after building from source): ```bash weed-vector --masters=localhost:9333 diff --git a/docker/apt-install b/docker/apt-install index 8dfd999d39..69133e151d 100755 --- a/docker/apt-install +++ b/docker/apt-install @@ -3,17 +3,15 @@ # unreachable. See docker/Dockerfile.e2e. set -e -# Every rewrite starts from the pristine list, so a mirror that just failed does -# not become the pattern the next rewrite has to match. -[ -f /etc/apt/sources.list.orig ] || cp /etc/apt/sources.list /etc/apt/sources.list.orig - for mirror in azure.archive.ubuntu.com archive.ubuntu.com; do - # Any archive host, so this works whether the pristine list came from the - # base image (archive.ubuntu.com) or a CI runner (azure.archive.ubuntu.com). - sed "s|http://[a-z0-9.]*archive\.ubuntu\.com/ubuntu|http://$mirror/ubuntu|g; s|http://security\.ubuntu\.com/ubuntu|http://$mirror/ubuntu|g" \ - /etc/apt/sources.list.orig > /etc/apt/sources.list - if apt-get -o Acquire::Retries=3 -o Acquire::http::Timeout=15 update && \ - DEBIAN_FRONTEND=noninteractive apt-get -o Acquire::Retries=3 -o Acquire::http::Timeout=15 install -y \ + for apt_source in /etc/apt/sources.list /etc/apt/sources.list.d/*.list /etc/apt/sources.list.d/*.sources; do + [ -f "$apt_source" ] || continue + [ -f "$apt_source.orig" ] || cp "$apt_source" "$apt_source.orig" + sed -E "s|(https?://)[a-z0-9.]*archive\.ubuntu\.com/ubuntu|\1$mirror/ubuntu|g; s|(https?://)security\.ubuntu\.com/ubuntu|\1$mirror/ubuntu|g" \ + "$apt_source.orig" > "$apt_source" + done + if timeout --kill-after=5s 120s apt-get -o Acquire::Retries=1 -o Acquire::http::Timeout=15 -o Acquire::https::Timeout=15 -o APT::Update::Error-Mode=any update && \ + DEBIAN_FRONTEND=noninteractive timeout --kill-after=5s 180s apt-get -o Acquire::Retries=1 -o Acquire::http::Timeout=15 -o Acquire::https::Timeout=15 install -y \ --no-install-recommends --no-install-suggests "$@"; then apt-get clean rm -rf /var/lib/apt/lists/* diff --git a/docker/test_apt_install.py b/docker/test_apt_install.py new file mode 100644 index 0000000000..08f775bc33 --- /dev/null +++ b/docker/test_apt_install.py @@ -0,0 +1,110 @@ +import json +import os +from pathlib import Path +import subprocess +import tempfile +import unittest + + +class AptInstallTests(unittest.TestCase): + def run_helper(self, sources, fail_action="update", timeout_azure=False): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + apt = root / "apt" + (apt / "sources.list.d").mkdir(parents=True) + (apt / "sources.list").write_text("") + for name, content in sources.items(): + (apt / name).write_text(content) + helper = root / "apt-install" + helper.write_text(Path(__file__).with_name("apt-install").read_text() + .replace("/etc/apt", str(apt)) + .replace("/var/lib/apt", str(root / "state"))) + commands = root / "bin" + commands.mkdir() + fake_apt = commands / "apt-get" + fake_apt.write_text(r"""#!/usr/bin/env python3 +import json, os, pathlib, sys +root = pathlib.Path(os.environ["VECTOR_APT_TEST_ROOT"]) +apt = root / "apt" +sources = [apt / "sources.list", *apt.glob("sources.list.d/*.list"), *apt.glob("sources.list.d/*.sources")] +content = "".join(p.read_text() for p in sources) +action = next(a for a in sys.argv[1:] if a in ("update", "install", "clean")) +with (root / "calls").open("a") as output: + output.write(json.dumps({"action": action, "sources": content, "args": sys.argv[1:]}) + "\n") +if "azure.archive.ubuntu.com" in content and action == "update" and os.environ["VECTOR_APT_FAIL_ACTION"] == "partial": + sys.exit(100 if "APT::Update::Error-Mode=any" in sys.argv else 0) +if "azure.archive.ubuntu.com" in content and action == os.environ["VECTOR_APT_FAIL_ACTION"]: + sys.exit(100) +""") + fake_apt.chmod(0o755) + fake_timeout = commands / "timeout" + fake_timeout.write_text(r"""#!/usr/bin/env python3 +import json, os, pathlib, sys +root = pathlib.Path(os.environ["VECTOR_APT_TEST_ROOT"]) +apt = root / "apt" +sources = [apt / "sources.list", *apt.glob("sources.list.d/*.list"), *apt.glob("sources.list.d/*.sources")] +content = "".join(p.read_text() for p in sources) +with (root / "calls").open("a") as output: + output.write(json.dumps({"action": "timeout", "args": sys.argv[1:]}) + "\n") +if os.environ["VECTOR_APT_TIMEOUT_AZURE"] and "azure.archive.ubuntu.com" in content: + sys.exit(124) +os.execvp(sys.argv[3], sys.argv[3:]) +""") + fake_timeout.chmod(0o755) + result = subprocess.run(["/bin/sh", str(helper), "fuse"], + env={**os.environ, "PATH": str(commands) + os.pathsep + os.environ["PATH"], + "VECTOR_APT_TEST_ROOT": str(root), "VECTOR_APT_FAIL_ACTION": fail_action, + "VECTOR_APT_TIMEOUT_AZURE": "1" if timeout_azure else ""}, + text=True, capture_output=True, timeout=5) + calls = [json.loads(line) for line in (root / "calls").read_text().splitlines()] if (root / "calls").exists() else [] + return result, calls + + def test_fallback_rewrites_all_source_formats(self): + for protocol in ("http", "https"): + for layout in ("legacy", "deb822", "mixed"): + with self.subTest(protocol=protocol, layout=layout): + sources = {} + if layout in ("legacy", "mixed"): + sources["sources.list"] = f"deb {protocol}://azure.archive.ubuntu.com/ubuntu noble main\n" + if layout in ("deb822", "mixed"): + sources["sources.list.d/ubuntu.sources"] = f"Types: deb\nURIs: {protocol}://azure.archive.ubuntu.com/ubuntu\nSuites: noble\nComponents: main\n" + result, calls = self.run_helper(sources) + self.assertEqual(result.returncode, 0, result.stderr) + installs = [call for call in calls if call["action"] == "install"] + self.assertEqual(len(installs), 1) + self.assertNotIn("azure.archive.ubuntu.com", installs[0]["sources"]) + self.assertIn(protocol + "://archive.ubuntu.com", installs[0]["sources"]) + + def test_install_failure_also_tries_the_next_mirror(self): + result, calls = self.run_helper({"sources.list.d/ubuntu.sources": + "Types: deb\nURIs: http://security.ubuntu.com/ubuntu\nSuites: noble-security\nComponents: main\n"}, + fail_action="install") + self.assertEqual(result.returncode, 0, result.stderr) + installs = [call for call in calls if call["action"] == "install"] + self.assertEqual(len(installs), 2) + self.assertIn("azure.archive.ubuntu.com", installs[0]["sources"]) + self.assertNotIn("azure.archive.ubuntu.com", installs[1]["sources"]) + + def test_partial_update_does_not_install_stale_indexes(self): + result, calls = self.run_helper({"sources.list": "deb http://azure.archive.ubuntu.com/ubuntu noble main\n"}, + fail_action="partial") + self.assertEqual(result.returncode, 0, result.stderr) + installs = [call for call in calls if call["action"] == "install"] + self.assertEqual(len(installs), 1) + self.assertNotIn("azure.archive.ubuntu.com", installs[0]["sources"]) + + def test_timeout_advances_to_the_next_mirror(self): + result, calls = self.run_helper({"sources.list": "deb http://azure.archive.ubuntu.com/ubuntu noble main\n"}, + fail_action="none", timeout_azure=True) + self.assertEqual(result.returncode, 0, result.stderr) + timeouts = [call for call in calls if call["action"] == "timeout"] + self.assertEqual(len(timeouts), 3) + for call in timeouts: + self.assertEqual(call["args"][0], "--kill-after=5s") + self.assertLessEqual(int(call["args"][1].removesuffix("s")), 180) + installs = [call for call in calls if call["action"] == "install"] + self.assertNotIn("azure.archive.ubuntu.com", installs[0]["sources"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/weed/command/s3.go b/weed/command/s3.go index c41d049650..8b65653a5d 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -760,6 +760,7 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp vectorsServer.SetEncryptionChecker(s3api.GetSSES3KeyManager().IsConfigured) vectorsServer.SetDescribe(describe) vectorsServer.SetLimiter(s3ApiServer.Limit) + vectorsServer.SetRequestLimits(int64(*s3opt.concurrentFileUploadLimit), int64(*s3opt.concurrentUploadLimitMB)*1024*1024) vectorsServer.RegisterRoutes(vectorsRouter) return vectorsRouter } diff --git a/weed/s3api/s3vectors/limits.go b/weed/s3api/s3vectors/limits.go index d88a356a7d..b143d899b8 100644 --- a/weed/s3api/s3vectors/limits.go +++ b/weed/s3api/s3vectors/limits.go @@ -10,12 +10,25 @@ import ( "github.com/gorilla/mux" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "golang.org/x/sync/semaphore" ) func (s *Server) SetLimiter(limit func(http.HandlerFunc, string) http.HandlerFunc) { s.limit = limit } +func (s *Server) SetRequestLimits(maxRequests, maxBytes int64) { + if maxRequests <= 0 { + maxRequests = 16 + } + if maxBytes <= 0 { + maxBytes = 8 * maxRequestBodyBytes + } + s.requestSlots = semaphore.NewWeighted(maxRequests) + s.requestBytes = semaphore.NewWeighted(maxBytes) + s.requestByteLimit = maxBytes +} + func (s *Server) limitRequest(handler http.HandlerFunc, w http.ResponseWriter, r *http.Request) { op := operationFromRequest(r) diff --git a/weed/s3api/s3vectors/limits_test.go b/weed/s3api/s3vectors/limits_test.go index 2f86755b43..7faf1d27b5 100644 --- a/weed/s3api/s3vectors/limits_test.go +++ b/weed/s3api/s3vectors/limits_test.go @@ -2,10 +2,16 @@ package s3vectors import ( "bytes" + "context" "encoding/json" + "encoding/xml" + "io" "net/http" "net/http/httptest" + "strings" + "sync/atomic" "testing" + "time" "github.com/gorilla/mux" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -123,3 +129,114 @@ func TestVectorLimitsCountChunkedBodyBytes(t *testing.T) { t.Fatal("mutated original request") } } + +type admissionAuth struct { + started, release chan struct{} + calls atomic.Int32 +} + +func (a *admissionAuth) DefaultAllow() bool { return true } +func (a *admissionAuth) AuthenticateRequest(r *http.Request) (string, interface{}, s3err.ErrorCode) { + a.calls.Add(1) + _, _ = io.ReadAll(r.Body) + if r.Header.Get("Block") != "" { + close(a.started) + select { + case <-a.release: + case <-r.Context().Done(): + } + } + if r.Header.Get("Deny") != "" { + return "", nil, s3err.ErrSignatureDoesNotMatch + } + return "tester", nil, s3err.ErrNone +} + +func TestVectorAdmissionBoundsBufferingBeforeAuthentication(t *testing.T) { + for _, tc := range []struct { + name string + requests, bytes int64 + chunked bool + }{ + {"request count", 1, 10, false}, + {"byte count", 2, 2, false}, + {"chunked body", 2, 2, true}, + } { + t.Run(tc.name, func(t *testing.T) { + auth := &admissionAuth{started: make(chan struct{}), release: make(chan struct{})} + server := NewServer(nil, auth, nil) + server.SetRequestLimits(tc.requests, tc.bytes) + handler := server.Auth(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }) + request := func() *http.Request { + r := httptest.NewRequest(http.MethodPost, "/PutVectors", strings.NewReader("{}")) + if tc.chunked { + r.ContentLength = -1 + } + return r + } + first := request() + first.Header.Set("Block", "true") + firstCtx, cancelFirst := context.WithTimeout(t.Context(), 5*time.Second) + defer cancelFirst() + done := make(chan struct{}) + go func() { defer close(done); handler(httptest.NewRecorder(), first.WithContext(firstCtx)) }() + select { + case <-auth.started: + case <-firstCtx.Done(): + t.Fatal("first request did not enter authentication") + } + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + waiting := httptest.NewRecorder() + handler(waiting, request().WithContext(ctx)) + if waiting.Code != http.StatusRequestTimeout || auth.calls.Load() != 1 { + t.Fatalf("waiting request buffered/authenticated: status %d, calls %d", waiting.Code, auth.calls.Load()) + } + close(auth.release) + <-done + denied := request() + denied.Header.Set("Deny", "true") + handler(httptest.NewRecorder(), denied.WithContext(firstCtx)) + after := httptest.NewRecorder() + handler(after, request().WithContext(firstCtx)) + if after.Code != http.StatusNoContent || auth.calls.Load() != 3 { + t.Fatalf("capacity leaked after success/failure: status %d, calls %d", after.Code, auth.calls.Load()) + } + }) + } +} + +func TestVectorErrorResponsesEscapeHTML(t *testing.T) { + payload := "" + for _, limited := range []bool{false, true} { + server := NewServer(nil, headerAuth{}, nil) + operation := payload + body := `{}` + if limited { + operation = "QueryVectors" + body = `{"vectorBucketName":"embeddings","indexName":"documents"}` + server.SetLimiter(func(http.HandlerFunc, string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTooManyRequests) + _ = xml.NewEncoder(w).Encode(struct { + XMLName xml.Name `xml:"Error"` + Code, Message string + }{Code: "SlowDown", Message: payload}) + } + }) + } + router := mux.NewRouter() + server.RegisterRoutes(router) + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body)) + request.Header.Set("X-Amz-Target", operation) + response := httptest.NewRecorder() + router.ServeHTTP(response, request) + var failure Error + if response.Code < 400 || json.Unmarshal(response.Body.Bytes(), &failure) != nil || !strings.Contains(failure.Message, payload) { + t.Fatalf("error response lost message: %d %s", response.Code, response.Body) + } + if strings.Contains(response.Body.String(), "<") || response.Header().Get("X-Content-Type-Options") != "nosniff" || response.Header().Get("Content-Type") != "application/x-amz-json-1.1" { + t.Fatalf("unsafe error response: %v %s", response.Header(), response.Body) + } + } +} diff --git a/weed/s3api/s3vectors/server.go b/weed/s3api/s3vectors/server.go index 65ad351748..55aa54403e 100644 --- a/weed/s3api/s3vectors/server.go +++ b/weed/s3api/s3vectors/server.go @@ -17,6 +17,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" stats_collect "github.com/seaweedfs/seaweedfs/weed/stats" + "golang.org/x/sync/semaphore" ) // FilerClient provides access to the filer for storage operations. @@ -32,14 +33,17 @@ type S3Authenticator interface { // Server implements the AWS S3 Vectors API. type Server struct { - filerClient FilerClient - tables *s3tables.Manager - internal *s3tables.Manager - vectors vector_pb.VectorServiceClient - authenticator S3Authenticator - region string - accountID string - limit func(http.HandlerFunc, string) http.HandlerFunc + filerClient FilerClient + tables *s3tables.Manager + internal *s3tables.Manager + vectors vector_pb.VectorServiceClient + authenticator S3Authenticator + region string + accountID string + limit func(http.HandlerFunc, string) http.HandlerFunc + requestSlots *semaphore.Weighted + requestBytes *semaphore.Weighted + requestByteLimit int64 } // NewServer creates an S3 Vectors server over the given filer and catalog. @@ -53,13 +57,15 @@ func NewServer(filerClient FilerClient, authenticator S3Authenticator, vectors v // demand read grants the API does not require. internal := s3tables.NewManager() internal.SetTrusted(true) - return &Server{ + server := &Server{ filerClient: filerClient, tables: manager, internal: internal, vectors: vectors, authenticator: authenticator, } + server.SetRequestLimits(0, 0) + return server } // SetRegion configures the region embedded in the ARNs this server mints. @@ -158,11 +164,32 @@ func (s *Server) dispatch(w http.ResponseWriter, r *http.Request) { // Auth authenticates the caller and puts the identity in the request context. func (s *Server) Auth(handler http.HandlerFunc) http.HandlerFunc { + if s.requestSlots == nil { + s.SetRequestLimits(0, 0) + } return func(w http.ResponseWriter, r *http.Request) { if r.ContentLength > maxRequestBodyBytes { writeError(w, http.StatusBadRequest, "ValidationException", errRequestBodyTooLarge.Error()) return } + if err := s.requestSlots.Acquire(r.Context(), 1); err != nil { + writeError(w, http.StatusRequestTimeout, "RequestTimeout", err.Error()) + return + } + defer s.requestSlots.Release(1) + var size int64 + if r.Body != nil && r.Body != http.NoBody { + size = r.ContentLength + if size <= 0 { + size = maxRequestBodyBytes + } + } + size = min(size, s.requestByteLimit) + if err := s.requestBytes.Acquire(r.Context(), size); err != nil { + writeError(w, http.StatusRequestTimeout, "RequestTimeout", err.Error()) + return + } + defer s.requestBytes.Release(size) r = r.WithContext(s3_constants.SetRequestBodyLimitInContext(r.Context(), maxRequestBodyBytes)) identity, identityObj, errCode := s.authenticator.AuthenticateRequest(r) switch errCode { diff --git a/weed/s3api/s3vectors/types.go b/weed/s3api/s3vectors/types.go index 9139683a7e..c2082c1e46 100644 --- a/weed/s3api/s3vectors/types.go +++ b/weed/s3api/s3vectors/types.go @@ -22,6 +22,7 @@ func identityOf(r *http.Request) string { func writeJSON(w http.ResponseWriter, status int, payload interface{}) { w.Header().Set("Content-Type", "application/x-amz-json-1.1") + w.Header().Set("X-Content-Type-Options", "nosniff") w.WriteHeader(status) if payload != nil { _ = json.NewEncoder(w).Encode(payload)