mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-02 18:22:40 +01:00
s3vectors: accept bodies up to the 32 MiB limit, report larger ones as a size error, log engine failures (#12320)
* s3api: hash an s3vectors body up to the Vectors API limit SigV4 verification of a non-S3 request hashed at most 10 MiB of the body and silently dropped the rest, so any S3 Vectors request over 10 MiB failed its signature check with 403, although the API accepts 32 MiB. The limit now follows the signing service, and a body over it is reported as ErrEntityTooLarge instead of being truncated; the embedded IAM answers that with InvalidInput, as it already does for a body over its own limit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * s3vectors: answer an oversized request with a ValidationException A body over 32 MiB was refused as "403 AccessDenied: authentication failed". A declared Content-Length over the limit is now refused before authentication, and a streamed body over it, whether the signature check or the handler reads it, gets the same "request body exceeds 32 MiB" ValidationException. The limit is the one SigV4 verification uses. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * s3vectors: log engine and catalog failures at the default level An engine error reached the client as a bare "internal error", and the cause was logged only with -v=1, so a misconfigured engine left nothing in a default log. Engine and catalog failures are now logged as errors, and a reported unavailability as a warning, with the operation, bucket and index. The response still leaves the engine's message out, since it can name internal hosts and paths, and points at the gateway log instead. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * s3api: hash the signed body through readRequestBody streamHashRequestBody repeated readRequestBody's read-past-the-limit check; it now reads the body with readRequestBody and hashes the bytes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
2eb219df61
commit
02d22800aa
10 files changed
+197
-52
No files matched your search
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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"
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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{}}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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).
|
||||
|
||||
Reference in new issue
Block a user