diff --git a/weed/s3api/auth_signature_v4.go b/weed/s3api/auth_signature_v4.go index 661bd7002f..2332d36a3a 100644 --- a/weed/s3api/auth_signature_v4.go +++ b/weed/s3api/auth_signature_v4.go @@ -25,6 +25,7 @@ import ( "crypto/subtle" "encoding/base64" "encoding/hex" + "errors" "io" "net" "net/http" @@ -74,22 +75,19 @@ func streamHashRequestBody(r *http.Request, sizeLimit int64) (string, error) { return emptySHA256, nil } - limitedReader := io.LimitReader(r.Body, sizeLimit) - hasher := sha256.New() - var bodyBuffer bytes.Buffer - - // Use io.Copy with an io.MultiWriter to hash and buffer the body simultaneously. - if _, err := io.Copy(io.MultiWriter(hasher, &bodyBuffer), limitedReader); err != nil { + body, err := readRequestBody(r, sizeLimit) + if err != nil { return "", err } - r.Body = io.NopCloser(bytes.NewReader(bodyBuffer.Bytes())) + r.Body = io.NopCloser(bytes.NewReader(body)) - if bodyBuffer.Len() == 0 { + if len(body) == 0 { return emptySHA256, nil } - return hex.EncodeToString(hasher.Sum(nil)), nil + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]), nil } // getContentSha256Cksum retrieves the "x-amz-content-sha256" header value. @@ -655,8 +653,15 @@ func extractV4AuthInfoFromHeader(r *http.Request) (*v4AuthInfo, s3err.ErrorCode) hashedPayload := getContentSha256Cksum(r) if signV4Values.Credential.scope.service != "s3" && hashedPayload == emptySHA256 && r.Body != nil { + sizeLimit := int64(iamRequestBodyLimit) + if signV4Values.Credential.scope.service == "s3vectors" { + sizeLimit = s3_constants.S3VectorsMaxRequestBodyBytes + } var hashErr error - hashedPayload, hashErr = streamHashRequestBody(r, iamRequestBodyLimit) + hashedPayload, hashErr = streamHashRequestBody(r, sizeLimit) + if errors.Is(hashErr, errRequestBodyTooLarge) { + return nil, s3err.ErrEntityTooLarge + } if hashErr != nil { return nil, s3err.ErrInternalError } diff --git a/weed/s3api/auto_signature_v4_test.go b/weed/s3api/auto_signature_v4_test.go index e7747515d6..34874ae8dc 100644 --- a/weed/s3api/auto_signature_v4_test.go +++ b/weed/s3api/auto_signature_v4_test.go @@ -16,6 +16,8 @@ import ( "time" "unicode/utf8" + "github.com/aws/aws-sdk-go/aws/credentials" + v4 "github.com/aws/aws-sdk-go/aws/signer/v4" "github.com/gorilla/mux" "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" @@ -1820,18 +1822,10 @@ func TestIAMLargeBodySecurityLimit(t *testing.T) { ", SignedHeaders=content-type;host;x-amz-date, Signature=dummysignature" req.Header.Set("Authorization", authHeader) - // The function should complete successfully but limit the body to 10 MiB identity, _, errCode := iam.doesSignatureMatch(req) - // Should get signature mismatch (dummy signature) but not internal error - assert.Equal(t, s3err.ErrSignatureDoesNotMatch, errCode) + assert.Equal(t, s3err.ErrEntityTooLarge, errCode) assert.Nil(t, identity) - - // Verify the body was truncated to the limit (10 MiB) - bodyBytes, err := io.ReadAll(req.Body) - assert.NoError(t, err) - assert.Equal(t, 10*(1<<20), len(bodyBytes)) // Should be exactly 10 MiB - assert.Equal(t, strings.Repeat("A", 10*(1<<20)), string(bodyBytes)) // All As, but truncated } // Test the streaming hash implementation directly @@ -1922,30 +1916,58 @@ func TestStreamingVsNonStreamingConsistency(t *testing.T) { } } -// Test streaming with size limit enforcement func TestStreamingWithSizeLimit(t *testing.T) { - // Create a payload larger than the limit - largePayload := strings.Repeat("A", 11*(1<<20)) // 11 MiB - - req, err := http.NewRequest("POST", "http://localhost:8111/", strings.NewReader(largePayload)) + exact := strings.Repeat("A", 1<<20) + req, err := http.NewRequest("POST", "http://localhost:8111/", strings.NewReader(exact)) assert.NoError(t, err) - - // Stream with the limit - hash, err := streamHashRequestBody(req, iamRequestBodyLimit) + hash, err := streamHashRequestBody(req, 1<<20) assert.NoError(t, err) + assert.Equal(t, getSHA256Hash([]byte(exact)), hash) - // Verify the hash is computed for the truncated content (10 MiB) - truncatedPayload := strings.Repeat("A", 10*(1<<20)) - expectedHash := sha256.Sum256([]byte(truncatedPayload)) - expectedHashStr := hex.EncodeToString(expectedHash[:]) - - assert.Equal(t, expectedHashStr, hash) - - // Verify the body was truncated - bodyBytes, err := io.ReadAll(req.Body) + req, err = http.NewRequest("POST", "http://localhost:8111/", strings.NewReader(exact+"A")) assert.NoError(t, err) - assert.Equal(t, 10*(1<<20), len(bodyBytes)) - assert.Equal(t, truncatedPayload, string(bodyBytes)) + _, err = streamHashRequestBody(req, 1<<20) + assert.ErrorIs(t, err, errRequestBodyTooLarge) +} + +func TestNonS3BodyHashLimitFollowsService(t *testing.T) { + iam := &IdentityAccessManagement{} + assert.NoError(t, iam.loadS3ApiConfiguration(&iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{ + Name: "vec", + Credentials: []*iam_pb.Credential{{AccessKey: "vec_key", SecretKey: "vec_secret"}}, + Actions: []string{"Admin"}, + }}, + })) + signer := v4.NewSigner(credentials.NewStaticCredentials("vec_key", "vec_secret", "")) + + for _, tc := range []struct { + service string + size int + want s3err.ErrorCode + }{ + {"s3vectors", 20 << 20, s3err.ErrNone}, + {"s3vectors", s3_constants.S3VectorsMaxRequestBodyBytes, s3err.ErrNone}, + {"s3vectors", s3_constants.S3VectorsMaxRequestBodyBytes + 1, s3err.ErrEntityTooLarge}, + {"iam", iamRequestBodyLimit + 1, s3err.ErrEntityTooLarge}, + } { + t.Run(fmt.Sprintf("%s_%d", tc.service, tc.size), func(t *testing.T) { + body := strings.Repeat("v", tc.size) + req, err := http.NewRequest("POST", "http://localhost:8333/PutVectors", strings.NewReader(body)) + assert.NoError(t, err) + _, err = signer.Sign(req, strings.NewReader(body), tc.service, "us-east-1", time.Now()) + assert.NoError(t, err) + + identity, _, errCode := iam.doesSignatureMatch(req) + assert.Equal(t, tc.want, errCode) + if tc.want == s3err.ErrNone { + assert.Equal(t, "vec", identity.Name) + got, err := io.ReadAll(req.Body) + assert.NoError(t, err) + assert.Equal(t, len(body), len(got)) + } + }) + } } // Benchmark streaming vs non-streaming memory usage diff --git a/weed/s3api/s3_constants/s3_config.go b/weed/s3api/s3_constants/s3_config.go index d2d2c257aa..ab34ce39af 100644 --- a/weed/s3api/s3_constants/s3_config.go +++ b/weed/s3api/s3_constants/s3_config.go @@ -13,6 +13,9 @@ var ( Separator = ":" ) +// S3VectorsMaxRequestBodyBytes caps the body of an S3 Vectors API request. +const S3VectorsMaxRequestBodyBytes = 32 << 20 + func Concat(elements ...string) string { return strings.Join(elements, Separator) } diff --git a/weed/s3api/s3api_embedded_iam.go b/weed/s3api/s3api_embedded_iam.go index 1c6aa1bec0..6b5e244469 100644 --- a/weed/s3api/s3api_embedded_iam.go +++ b/weed/s3api/s3api_embedded_iam.go @@ -342,6 +342,8 @@ func iamAuthErrorCode(errCode s3err.ErrorCode) string { return "RequestTimeTooSkewed" case s3err.ErrInternalError: return iam.ErrCodeServiceFailureException + case s3err.ErrEntityTooLarge: + return "InvalidInput" default: return "IncompleteSignature" } diff --git a/weed/s3api/s3api_embedded_iam_test.go b/weed/s3api/s3api_embedded_iam_test.go index 4269958de9..a7043b7114 100644 --- a/weed/s3api/s3api_embedded_iam_test.go +++ b/weed/s3api/s3api_embedded_iam_test.go @@ -3041,5 +3041,6 @@ func TestIamAuthErrorCode(t *testing.T) { assert.Equal(t, "SignatureDoesNotMatch", iamAuthErrorCode(s3err.ErrSignatureDoesNotMatch)) assert.Equal(t, "AccessDenied", iamAuthErrorCode(s3err.ErrAccessDenied)) assert.Equal(t, "RequestTimeTooSkewed", iamAuthErrorCode(s3err.ErrRequestTimeTooSkewed)) + assert.Equal(t, "InvalidInput", iamAuthErrorCode(s3err.ErrEntityTooLarge)) assert.Equal(t, "IncompleteSignature", iamAuthErrorCode(s3err.ErrMissingFields)) } diff --git a/weed/s3api/s3vectors/auth_test.go b/weed/s3api/s3vectors/auth_test.go index a333265d0d..56e326e5a9 100644 --- a/weed/s3api/s3vectors/auth_test.go +++ b/weed/s3api/s3vectors/auth_test.go @@ -1,8 +1,11 @@ package s3vectors import ( + "errors" + "io" "net/http" "net/http/httptest" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/s3api" @@ -12,9 +15,15 @@ import ( type fakeAuthenticator struct { identity *s3api.Identity + errCode s3err.ErrorCode + calls int } func (f *fakeAuthenticator) AuthenticateRequest(r *http.Request) (string, interface{}, s3err.ErrorCode) { + f.calls++ + if f.errCode != s3err.ErrNone { + return "", nil, f.errCode + } return f.identity.Name, f.identity, s3err.ErrNone } @@ -35,3 +44,52 @@ func TestAuthKeepsIdentityObject(t *testing.T) { t.Fatalf("identity object missing from context: %v", got) } } + +func TestAuthAnswersOversizedBodyWithValidationException(t *testing.T) { + alice := &s3api.Identity{Name: "alice", Actions: []s3api.Action{"Admin"}} + for _, tc := range []struct { + name string + contentLength int64 + errCode s3err.ErrorCode + wantStatus int + wantCode string + wantAuthCalls int + }{ + {"declared length over the limit", maxRequestBodyBytes + 1, s3err.ErrNone, http.StatusBadRequest, "ValidationException", 0}, + {"streamed body over the limit", -1, s3err.ErrEntityTooLarge, http.StatusBadRequest, "ValidationException", 1}, + {"bad signature", 10, s3err.ErrSignatureDoesNotMatch, http.StatusForbidden, "AccessDenied", 1}, + {"declared length at the limit", maxRequestBodyBytes, s3err.ErrNone, http.StatusOK, "", 1}, + } { + t.Run(tc.name, func(t *testing.T) { + auth := &fakeAuthenticator{identity: alice, errCode: tc.errCode} + s := &Server{authenticator: auth} + req := httptest.NewRequest(http.MethodPost, "/PutVectors", strings.NewReader("{}")) + req.ContentLength = tc.contentLength + rec := httptest.NewRecorder() + s.Auth(func(w http.ResponseWriter, r *http.Request) {})(rec, req) + + if rec.Code != tc.wantStatus || auth.calls != tc.wantAuthCalls { + t.Fatalf("status %d after %d auth calls, want %d after %d: %s", rec.Code, auth.calls, tc.wantStatus, tc.wantAuthCalls, rec.Body) + } + if tc.wantCode != "" && !strings.Contains(rec.Body.String(), `"code":"`+tc.wantCode+`"`) { + t.Fatalf("body %s, want code %s", rec.Body, tc.wantCode) + } + if tc.wantCode == "ValidationException" && !strings.Contains(rec.Body.String(), "request body exceeds 32 MiB") { + t.Fatalf("body %s does not name the limit", rec.Body) + } + }) + } +} + +func TestReadJSONRefusesBodyOverTheLimit(t *testing.T) { + var req PutVectorsRequest + over := httptest.NewRequest(http.MethodPost, "/PutVectors", io.MultiReader( + strings.NewReader(`{"indexName":"`), strings.NewReader(strings.Repeat("a", maxRequestBodyBytes)), strings.NewReader(`"}`))) + if err := readJSON(over, &req); !errors.Is(err, errRequestBodyTooLarge) { + t.Fatalf("readJSON over the limit = %v, want %v", err, errRequestBodyTooLarge) + } + within := httptest.NewRequest(http.MethodPost, "/PutVectors", strings.NewReader(`{"indexName":"idx"}`)) + if err := readJSON(within, &req); err != nil || req.IndexName != "idx" { + t.Fatalf("readJSON within the limit = %v, indexName %q", err, req.IndexName) + } +} diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 593e4237ce..7e7bd12e83 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -183,7 +183,8 @@ func catalogErrCode(err error) string { func catalogErr(w http.ResponseWriter, err error) { var apiErr *s3tables.S3TablesError if !errors.As(err, &apiErr) { - writeError(w, http.StatusInternalServerError, "InternalFailure", "internal error") + glog.Errorf("S3 Vectors: catalog: %v", err) + writeError(w, http.StatusInternalServerError, "InternalFailure", internalErrorMessage) return } switch { @@ -201,10 +202,12 @@ func catalogErr(w http.ResponseWriter, err error) { } } +const internalErrorMessage = "internal error, see the S3 gateway log" + // vectorErr maps a VectorService gRPC status onto the API's error shape. // Engine messages reach clients only for codes that describe the request // itself; transport and internal details stay in the server log. -func vectorErr(w http.ResponseWriter, err error) { +func vectorErr(w http.ResponseWriter, r *http.Request, arns vectorBucketARNs, err error) { msg := status.Convert(err).Message() switch status.Code(err) { case codes.NotFound: @@ -214,10 +217,11 @@ func vectorErr(w http.ResponseWriter, err error) { case codes.PermissionDenied, codes.Unauthenticated: writeError(w, http.StatusForbidden, "AccessDeniedException", "access denied") case codes.Unavailable: + glog.Warningf("S3 Vectors: %s %s/%s: %v", operationFromRequest(r), arns.bucket, arns.index, err) writeError(w, http.StatusServiceUnavailable, "ServiceUnavailableException", "vector service unavailable") default: - glog.V(1).Infof("S3 Vectors: engine error: %v", err) - writeError(w, http.StatusInternalServerError, "InternalFailure", "internal error") + glog.Errorf("S3 Vectors: %s %s/%s: %v", operationFromRequest(r), arns.bucket, arns.index, err) + writeError(w, http.StatusInternalServerError, "InternalFailure", internalErrorMessage) } } @@ -1026,7 +1030,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { } else if gerr != nil { glog.Errorf("S3 Vectors: rollback lookup for %s/%s failed: %v", arns.bucket, arns.index, gerr) } - vectorErr(w, err) + vectorErr(w, r, arns, err) return } writeJSON(w, http.StatusOK, &CreateIndexResponse{IndexArn: s.indexArn(arns.bucket, arns.index)}) @@ -1279,7 +1283,7 @@ func (s *Server) handlePutVectors(w http.ResponseWriter, r *http.Request) { Vectors: vectors, }) if err != nil { - vectorErr(w, err) + vectorErr(w, r, arns, err) return } writeJSON(w, http.StatusOK, &PutVectorsResponse{}) @@ -1360,7 +1364,7 @@ func (s *Server) handleGetVectors(w http.ResponseWriter, r *http.Request) { ReturnMetadata: req.ReturnMetadata, }) if err != nil { - vectorErr(w, err) + vectorErr(w, r, arns, err) return } writeJSON(w, http.StatusOK, &GetVectorsResponse{ @@ -1419,7 +1423,7 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) { } resp, err := s.vectors.ListVectors(r.Context(), listReq) if err != nil { - vectorErr(w, err) + vectorErr(w, r, arns, err) return } // An engine that predates segment_count lists every row; page through @@ -1429,7 +1433,7 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) { out, nextToken, err := s.listSegmentVectors(r.Context(), listReq.Table, offset, maxResults, uint64(*req.SegmentCount), uint64(*req.SegmentIndex), req.ReturnData, req.ReturnMetadata, resp) if err != nil { - vectorErr(w, err) + vectorErr(w, r, arns, err) return } writeJSON(w, http.StatusOK, &ListVectorsResponse{ @@ -1559,7 +1563,7 @@ func (s *Server) handleDeleteVectors(w http.ResponseWriter, r *http.Request) { Keys: req.Keys, }) if err != nil { - vectorErr(w, err) + vectorErr(w, r, arns, err) return } writeJSON(w, http.StatusOK, &DeleteVectorsResponse{}) @@ -1932,7 +1936,7 @@ func (s *Server) handleQueryVectors(w http.ResponseWriter, r *http.Request) { ReturnDistance: req.ReturnDistance, }) if err != nil { - vectorErr(w, err) + vectorErr(w, r, arns, err) return } out := &QueryVectorsResponse{DistanceMetric: metric, Vectors: []*QueryOutputVector{}} diff --git a/weed/s3api/s3vectors/handlers_test.go b/weed/s3api/s3vectors/handlers_test.go index bb31b644a7..6ed043d8bc 100644 --- a/weed/s3api/s3vectors/handlers_test.go +++ b/weed/s3api/s3vectors/handlers_test.go @@ -19,6 +19,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // fakeVectors serves a fixed row set with the engine's offset-token paging @@ -1073,3 +1075,31 @@ func TestCreateIndexSurvivesAConcurrentNamespaceCreate(t *testing.T) { t.Fatalf("first create index: %d %s", w.Code, w.Body.String()) } } + +func TestVectorErrKeepsEngineDetailsOutOfTheResponse(t *testing.T) { + arns := vectorBucketARNs{bucket: "media", index: "movies"} + for _, tc := range []struct { + err error + wantStatus int + wantCode string + wantMessage string + }{ + {status.Error(codes.Internal, "declare: 500 from http://10.0.0.5:8888/lance/v1/table/media$vector$movies/declare"), + http.StatusInternalServerError, "InternalFailure", internalErrorMessage}, + {status.Error(codes.Unavailable, "no weed-vector registered on master"), + http.StatusServiceUnavailable, "ServiceUnavailableException", "vector service unavailable"}, + {status.Error(codes.InvalidArgument, "vector has dimension 2, index has 3"), + http.StatusBadRequest, "ValidationException", "vector has dimension 2, index has 3"}, + } { + rec := httptest.NewRecorder() + vectorErr(rec, httptest.NewRequest(http.MethodPost, "/PutVectors", nil), arns, tc.err) + var body map[string]string + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if rec.Code != tc.wantStatus || body["code"] != tc.wantCode || body["message"] != tc.wantMessage { + t.Errorf("%v: got %d %s %q, want %d %s %q", status.Code(tc.err), rec.Code, body["code"], body["message"], + tc.wantStatus, tc.wantCode, tc.wantMessage) + } + } +} diff --git a/weed/s3api/s3vectors/server.go b/weed/s3api/s3vectors/server.go index c9be580589..ceb7389643 100644 --- a/weed/s3api/s3vectors/server.go +++ b/weed/s3api/s3vectors/server.go @@ -129,8 +129,17 @@ 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 { return func(w http.ResponseWriter, r *http.Request) { + if r.ContentLength > maxRequestBodyBytes { + writeError(w, http.StatusBadRequest, "ValidationException", errRequestBodyTooLarge.Error()) + return + } identity, identityObj, errCode := s.authenticator.AuthenticateRequest(r) - if errCode != s3err.ErrNone { + switch errCode { + case s3err.ErrNone: + case s3err.ErrEntityTooLarge: + writeError(w, http.StatusBadRequest, "ValidationException", errRequestBodyTooLarge.Error()) + return + default: writeError(w, http.StatusForbidden, "AccessDenied", "authentication failed") return } diff --git a/weed/s3api/s3vectors/types.go b/weed/s3api/s3vectors/types.go index a9c5b96c51..0bbbaff6f6 100644 --- a/weed/s3api/s3vectors/types.go +++ b/weed/s3api/s3vectors/types.go @@ -1,8 +1,10 @@ package s3vectors import ( + "bytes" "context" "encoding/json" + "fmt" "io" "net/http" "time" @@ -35,10 +37,19 @@ func writeError(w http.ResponseWriter, status int, code, message string) { }) } -const maxRequestBodyBytes = 32 << 20 +const maxRequestBodyBytes = s3_constants.S3VectorsMaxRequestBodyBytes + +var errRequestBodyTooLarge = fmt.Errorf("request body exceeds %d MiB", maxRequestBodyBytes>>20) func readJSON(r *http.Request, req interface{}) error { - return json.NewDecoder(io.LimitReader(r.Body, maxRequestBodyBytes)).Decode(req) + body, err := io.ReadAll(io.LimitReader(r.Body, maxRequestBodyBytes+1)) + if err != nil { + return err + } + if len(body) > maxRequestBodyBytes { + return errRequestBodyTooLarge + } + return json.NewDecoder(bytes.NewReader(body)).Decode(req) } // Request/response shapes for the AWS S3 Vectors API (awsJson1_1).