Skip to content

Commit 2b3ee67

Browse files
terion-labsclaude
terion-labs
and
claude
committed
Add multipart upload and default AWS client upload compatibility
Make the S3-compatible API work with stock AWS CLI/SDK settings and support large-file uploads, all behind the existing MUTATIONS_ENABLED gates on edge and connector: - Decode aws-chunked bodies (STREAMING-UNSIGNED-PAYLOAD-TRAILER), the upload mode modern clients default to. X-Amz-Decoded-Content-Length is required; checksum trailers are discarded, not verified. - Accept signed SHA-256 payload hashes on uploads. The hash is covered by the verified SigV4 signature but the streamed body is not re-hashed against it; signed-chunk streaming stays rejected. - Add CreateMultipartUpload, UploadPart, CompleteMultipartUpload, and AbortMultipartUpload — the set `aws s3 cp` needs. Tickets carry a bounded multipart envelope; part bodies and the Complete part-list XML (capped at 1 MiB) stream through the upload-source channel. - Sign backend PutObject/UploadPart with UNSIGNED-PAYLOAD: upload bodies stream from the edge and are not seekable, which the SDK otherwise rejects on non-TLS backends. - Reflect the mutation gate in the 405 Allow header. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent 2454b54 commit 2b3ee67

19 files changed

Lines changed: 2048 additions & 123 deletions

‎cmd/edge-gateway/main.go‎

Lines changed: 132 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -276,9 +276,24 @@ type s3EdgeRequest struct {
276276
rangeHeader string
277277
operation tickets.Operation
278278
list *tickets.ListRequest
279+
multipart *tickets.MultipartRequest
279280
headBucket bool
280281
contentLength *int64
281282
contentType string
283+
// body is the effective upload body for operations that stream one:
284+
// the raw request body, or an aws-chunked decoder wrapped around it.
285+
body io.ReadCloser
286+
}
287+
288+
// operationStreamsUploadBody reports whether op carries a client request body
289+
// that streams through the edge (upload source or direct backend call).
290+
func operationStreamsUploadBody(op tickets.Operation) bool {
291+
switch op {
292+
case tickets.OperationPutObject, tickets.OperationUploadPart, tickets.OperationCompleteMultipartUpload:
293+
return true
294+
default:
295+
return false
296+
}
282297
}
283298

284299
func edgeS3APIMode(cfg config.EdgeConfig) s3api.RoutingMode {
@@ -324,8 +339,11 @@ func (s *edgeServer) serveS3API(w http.ResponseWriter, r *http.Request) {
324339
return
325340
}
326341

327-
ticket := tickets.Ticket{Version: tickets.Version, RequestID: reqID, Bucket: req.bucket, Key: req.key, Method: r.Method, Operation: req.operation, Range: req.rangeHeader, List: req.list, Server: req.server, DeadlineUnixMS: deadline.UnixMilli(), IngestURL: ingestURL, IngestToken: ingestToken, TraceID: reqID}
328-
if req.operation == tickets.OperationPutObject {
342+
ticket := tickets.Ticket{Version: tickets.Version, RequestID: reqID, Bucket: req.bucket, Key: req.key, Method: r.Method, Operation: req.operation, Range: req.rangeHeader, List: req.list, Multipart: req.multipart, Server: req.server, DeadlineUnixMS: deadline.UnixMilli(), IngestURL: ingestURL, IngestToken: ingestToken, TraceID: reqID}
343+
if req.operation == tickets.OperationCreateMultipartUpload {
344+
ticket.ContentType = req.contentType
345+
}
346+
if operationStreamsUploadBody(req.operation) {
329347
uploadToken, err := s.newToken()
330348
if err != nil {
331349
writeS3Error(w, r, s3HTTPError{status: http.StatusInternalServerError, code: "InternalError", message: "Request setup failed"})
@@ -351,13 +369,13 @@ func (s *edgeServer) serveS3API(w http.ResponseWriter, r *http.Request) {
351369
defer s.registry.Cancel(reqID, pending.ErrCanceled)
352370

353371
uploadRegistered := false
354-
if req.operation == tickets.OperationPutObject {
372+
if operationStreamsUploadBody(req.operation) {
355373
if s.uploadSources == nil {
356374
s.registry.Cancel(reqID, uploadsource.ErrInvalidSource)
357375
writeS3Error(w, r, s3HTTPError{status: http.StatusInternalServerError, code: "InternalError", message: "Request setup failed"})
358376
return
359377
}
360-
source := uploadsource.Source{RequestID: reqID, Token: ticket.UploadToken, Body: r.Body, ContentLength: req.contentLength, ContentType: req.contentType, Deadline: deadline, Context: r.Context()}
378+
source := uploadsource.Source{RequestID: reqID, Token: ticket.UploadToken, Body: req.body, ContentLength: req.contentLength, ContentType: req.contentType, Deadline: deadline, Context: r.Context()}
361379
if err := s.uploadSources.Register(source); err != nil {
362380
s.registry.Cancel(reqID, err)
363381
writeS3Error(w, r, s3ErrorForUploadSetup(err))
@@ -438,7 +456,7 @@ func (s *edgeServer) resolveS3Request(r *http.Request) (s3EdgeRequest, error) {
438456
ValidateServer: publicpath.ValidateAlias,
439457
})
440458
if err != nil {
441-
return s3EdgeRequest{}, s3ErrorForClassify(err)
459+
return s3EdgeRequest{}, s3ErrorForClassify(err, s.cfg.MutationsEnabled)
442460
}
443461

444462
req := s3EdgeRequest{server: mapping.Server, bucket: mapping.BackendBucket, key: mapping.BackendKey}
@@ -458,22 +476,67 @@ func (s *edgeServer) resolveS3Request(r *http.Request) (s3EdgeRequest, error) {
458476
case s3api.OperationPutObject:
459477
req.operation = tickets.OperationPutObject
460478
req.contentType = r.Header.Get("Content-Type")
461-
if r.ContentLength >= 0 {
462-
contentLength := r.ContentLength
463-
req.contentLength = &contentLength
464-
}
465479
if err := validateS3ObjectRequest(req); err != nil {
466480
return s3EdgeRequest{}, err
467481
}
468-
if err := s.validateS3MutationRequest(r, mapping.Operation, authCtx, req); err != nil {
482+
if err := s.validateS3MutationRequest(r, req.operation, authCtx); err != nil {
483+
return s3EdgeRequest{}, err
484+
}
485+
if err := resolveS3UploadBody(r, authCtx, &req); err != nil {
469486
return s3EdgeRequest{}, err
470487
}
471488
case s3api.OperationDeleteObject:
472489
req.operation = tickets.OperationDeleteObject
473490
if err := validateS3ObjectRequest(req); err != nil {
474491
return s3EdgeRequest{}, err
475492
}
476-
if err := s.validateS3MutationRequest(r, mapping.Operation, authCtx, req); err != nil {
493+
if err := s.validateS3MutationRequest(r, req.operation, authCtx); err != nil {
494+
return s3EdgeRequest{}, err
495+
}
496+
case s3api.OperationCreateMultipartUpload:
497+
req.operation = tickets.OperationCreateMultipartUpload
498+
req.contentType = r.Header.Get("Content-Type")
499+
req.multipart = multipartFromMapping(mapping)
500+
if err := validateS3ObjectRequest(req); err != nil {
501+
return s3EdgeRequest{}, err
502+
}
503+
if err := s.validateS3MutationRequest(r, req.operation, authCtx); err != nil {
504+
return s3EdgeRequest{}, err
505+
}
506+
case s3api.OperationUploadPart:
507+
req.operation = tickets.OperationUploadPart
508+
req.multipart = multipartFromMapping(mapping)
509+
if err := validateS3ObjectRequest(req); err != nil {
510+
return s3EdgeRequest{}, err
511+
}
512+
if err := s.validateS3MutationRequest(r, req.operation, authCtx); err != nil {
513+
return s3EdgeRequest{}, err
514+
}
515+
if err := resolveS3UploadBody(r, authCtx, &req); err != nil {
516+
return s3EdgeRequest{}, err
517+
}
518+
case s3api.OperationCompleteMultipartUpload:
519+
req.operation = tickets.OperationCompleteMultipartUpload
520+
req.multipart = multipartFromMapping(mapping)
521+
if err := validateS3ObjectRequest(req); err != nil {
522+
return s3EdgeRequest{}, err
523+
}
524+
if err := s.validateS3MutationRequest(r, req.operation, authCtx); err != nil {
525+
return s3EdgeRequest{}, err
526+
}
527+
if err := resolveS3UploadBody(r, authCtx, &req); err != nil {
528+
return s3EdgeRequest{}, err
529+
}
530+
if *req.contentLength > tickets.MaxCompleteMultipartBodyBytes {
531+
return s3EdgeRequest{}, s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
532+
}
533+
case s3api.OperationAbortMultipartUpload:
534+
req.operation = tickets.OperationAbortMultipartUpload
535+
req.multipart = multipartFromMapping(mapping)
536+
if err := validateS3ObjectRequest(req); err != nil {
537+
return s3EdgeRequest{}, err
538+
}
539+
if err := s.validateS3MutationRequest(r, req.operation, authCtx); err != nil {
477540
return s3EdgeRequest{}, err
478541
}
479542
case s3api.OperationListObjectsV2:
@@ -496,31 +559,64 @@ func (s *edgeServer) resolveS3Request(r *http.Request) (s3EdgeRequest, error) {
496559
return req, nil
497560
}
498561

499-
func (s *edgeServer) validateS3MutationRequest(r *http.Request, operation s3api.Operation, authCtx s3api.AuthContext, req s3EdgeRequest) error {
562+
func (s *edgeServer) validateS3MutationRequest(r *http.Request, operation tickets.Operation, authCtx s3api.AuthContext) error {
500563
if !s.cfg.MutationsEnabled {
501564
return s3HTTPError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method not allowed"}
502565
}
503566
if strings.TrimSpace(r.Header.Get("Range")) != "" {
504567
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
505568
}
506-
if err := s3api.ValidatePayloadHashForOperation(operation, authCtx); err != nil {
569+
if err := s3api.ValidatePayloadHashForOperation(s3api.Operation(operation), authCtx); err != nil {
507570
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
508571
}
509-
if operation == s3api.OperationPutObject {
510-
if req.contentLength == nil || *req.contentLength < 0 {
572+
if operationStreamsUploadBody(operation) {
573+
// The aws-chunked markers belong to the unsigned-trailer mode, which
574+
// resolveS3UploadBody decodes; anything else must be a plain body.
575+
if authCtx.PayloadHashMode() != s3api.PayloadHashModeStreamingUnsignedTrailer && hasAWSChunkedMarkers(r, authCtx) {
511576
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
512577
}
513-
if hasAWSChunkedMarkers(r, authCtx) {
578+
return nil
579+
}
580+
if deleteHasBodyOrChunkedMarkers(r, authCtx) {
581+
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
582+
}
583+
return nil
584+
}
585+
586+
// resolveS3UploadBody sets req.body and req.contentLength to the effective
587+
// (decoded) upload body: aws-chunked framing is unwrapped for the unsigned
588+
// trailer mode, everything else streams the raw body and needs an explicit
589+
// Content-Length.
590+
func resolveS3UploadBody(r *http.Request, authCtx s3api.AuthContext, req *s3EdgeRequest) error {
591+
if authCtx.PayloadHashMode() == s3api.PayloadHashModeStreamingUnsignedTrailer {
592+
decoded, err := strconv.ParseInt(strings.TrimSpace(r.Header.Get("X-Amz-Decoded-Content-Length")), 10, 64)
593+
if err != nil || decoded < 0 {
514594
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
515595
}
596+
req.contentLength = &decoded
597+
req.body = struct {
598+
io.Reader
599+
io.Closer
600+
}{s3api.NewAWSChunkedReader(r.Body, decoded), r.Body}
516601
return nil
517602
}
518-
if deleteHasBodyOrChunkedMarkers(r, authCtx) {
603+
if r.ContentLength < 0 {
519604
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
520605
}
606+
contentLength := r.ContentLength
607+
req.contentLength = &contentLength
608+
req.body = r.Body
521609
return nil
522610
}
523611

612+
func multipartFromMapping(mapping s3api.RequestMapping) *tickets.MultipartRequest {
613+
return &tickets.MultipartRequest{
614+
UploadID: mapping.Multipart.UploadID,
615+
PartNumber: int32(mapping.Multipart.PartNumber),
616+
Rewrite: tickets.MultipartRewrite{Bucket: mapping.S3Bucket, Key: mapping.S3Key},
617+
}
618+
}
619+
524620
func hasAWSChunkedMarkers(r *http.Request, authCtx s3api.AuthContext) bool {
525621
return authCtx.PayloadHashMode() == s3api.PayloadHashModeStreaming ||
526622
headerHasToken(r.Header, "Content-Encoding", "aws-chunked") ||
@@ -632,10 +728,12 @@ func (s *edgeServer) serveS3Direct(w http.ResponseWriter, r *http.Request, req s
632728
return
633729
}
634730

635-
fetchReq := s3fetch.Request{Method: r.Method, Operation: req.operation, Bucket: req.bucket, Key: req.key, Range: req.rangeHeader, List: req.list}
636-
if req.operation == tickets.OperationPutObject {
637-
fetchReq.Body = r.Body
731+
fetchReq := s3fetch.Request{Method: r.Method, Operation: req.operation, Bucket: req.bucket, Key: req.key, Range: req.rangeHeader, List: req.list, Multipart: req.multipart}
732+
if operationStreamsUploadBody(req.operation) {
733+
fetchReq.Body = req.body
638734
fetchReq.ContentLength = req.contentLength
735+
}
736+
if req.operation == tickets.OperationPutObject || req.operation == tickets.OperationCreateMultipartUpload {
639737
fetchReq.ContentType = req.contentType
640738
}
641739
fetched, err := fetcher.Fetch(r.Context(), fetchReq)
@@ -651,12 +749,12 @@ func (s *edgeServer) serveS3Direct(w http.ResponseWriter, r *http.Request, req s
651749
if fetched.Body != nil {
652750
defer fetched.Body.Close()
653751
}
654-
if req.operation == tickets.OperationPutObject {
752+
if req.operation == tickets.OperationPutObject || req.operation == tickets.OperationUploadPart {
655753
setTrimmedHeader(w.Header(), "ETag", fetched.ETag)
656754
w.WriteHeader(http.StatusOK)
657755
return
658756
}
659-
if req.operation == tickets.OperationDeleteObject {
757+
if req.operation == tickets.OperationDeleteObject || req.operation == tickets.OperationAbortMultipartUpload {
660758
w.WriteHeader(http.StatusNoContent)
661759
return
662760
}
@@ -688,6 +786,7 @@ type s3HTTPError struct {
688786
status int
689787
code string
690788
message string
789+
allow string
691790
}
692791

693792
func (e s3HTTPError) Error() string {
@@ -697,7 +796,11 @@ func (e s3HTTPError) Error() string {
697796
func writeS3Error(w http.ResponseWriter, r *http.Request, err error) {
698797
s3err := s3ErrorFor(err)
699798
if s3err.status == http.StatusMethodNotAllowed {
700-
w.Header().Set("Allow", "GET, HEAD")
799+
allow := s3err.allow
800+
if allow == "" {
801+
allow = "GET, HEAD"
802+
}
803+
w.Header().Set("Allow", allow)
701804
}
702805
w.Header().Set("Content-Type", "application/xml")
703806
w.WriteHeader(s3err.status)
@@ -731,10 +834,14 @@ func s3ErrorForAuth(err error) s3HTTPError {
731834
}
732835
}
733836

734-
func s3ErrorForClassify(err error) s3HTTPError {
837+
func s3ErrorForClassify(err error, mutationsEnabled bool) s3HTTPError {
735838
message := err.Error()
736839
if strings.Contains(message, "unsupported method") {
737-
return s3HTTPError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method not allowed"}
840+
allow := "GET, HEAD"
841+
if mutationsEnabled {
842+
allow = "GET, HEAD, PUT, POST, DELETE"
843+
}
844+
return s3HTTPError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method not allowed", allow: allow}
738845
}
739846
return s3HTTPError{status: http.StatusBadRequest, code: "InvalidRequest", message: "Invalid request"}
740847
}

‎cmd/edge-gateway/main_test.go‎

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -642,8 +642,39 @@ func TestS3APIMutationsDisabledBeforeSideEffects(t *testing.T) {
642642
}
643643
}
644644

645+
func TestS3APIUnsupportedMethodAllowHeaderReflectsMutationGate(t *testing.T) {
646+
const emptyPayloadSHA = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
647+
tests := []struct {
648+
name string
649+
mutationsEnabled bool
650+
wantAllow string
651+
}{
652+
{name: "mutations disabled", wantAllow: "GET, HEAD"},
653+
{name: "mutations enabled", mutationsEnabled: true, wantAllow: "GET, HEAD, PUT, POST, DELETE"},
654+
}
655+
for _, tt := range tests {
656+
t.Run(tt.name, func(t *testing.T) {
657+
pub := &fakePublisher{}
658+
edge, _ := testS3Edge(pub, config.EdgeConfig{AllowedBuckets: []string{"demo-bucket"}, MutationsEnabled: tt.mutationsEnabled}, nil)
659+
req := s3SignHeaderRequestWithBody(t, http.MethodPost, "/demo-bucket/file.txt", http.NoBody, 0, map[string]string{"x-amz-content-sha256": emptyPayloadSHA})
660+
661+
resp := httptest.NewRecorder()
662+
edge.ServeHTTP(resp, req)
663+
664+
if got := resp.Result().StatusCode; got != http.StatusMethodNotAllowed {
665+
t.Fatalf("status = %d, want %d; body=%q", got, http.StatusMethodNotAllowed, resp.Body.String())
666+
}
667+
if got := resp.Result().Header.Get("Allow"); got != tt.wantAllow {
668+
t.Fatalf("Allow = %q, want %q", got, tt.wantAllow)
669+
}
670+
if pub.count() != 0 {
671+
t.Fatalf("published %d tickets, want 0", pub.count())
672+
}
673+
})
674+
}
675+
}
676+
645677
func TestS3APIMutationPolicyRejectsBeforeSideEffects(t *testing.T) {
646-
signedPayload := sha256.Sum256([]byte("hello"))
647678
tests := []struct {
648679
name string
649680
method string
@@ -652,7 +683,8 @@ func TestS3APIMutationPolicyRejectsBeforeSideEffects(t *testing.T) {
652683
headers map[string]string
653684
mutate func(*http.Request)
654685
}{
655-
{name: "signed sha256 put", method: http.MethodPut, body: "hello", length: 5, headers: map[string]string{"x-amz-content-sha256": hex.EncodeToString(signedPayload[:])}},
686+
{name: "signed streaming put", method: http.MethodPut, body: "hello", length: 5, headers: map[string]string{"x-amz-content-sha256": "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"}},
687+
{name: "streaming put without decoded length", method: http.MethodPut, body: "hello", length: 5, headers: map[string]string{"x-amz-content-sha256": "STREAMING-UNSIGNED-PAYLOAD-TRAILER"}},
656688
{name: "aws chunked put", method: http.MethodPut, body: "hello", length: 5, headers: map[string]string{"x-amz-content-sha256": "UNSIGNED-PAYLOAD", "Content-Encoding": "aws-chunked"}},
657689
{name: "unknown length put", method: http.MethodPut, body: "hello", length: -1, headers: map[string]string{"x-amz-content-sha256": "UNSIGNED-PAYLOAD"}},
658690
{name: "delete body", method: http.MethodDelete, body: "x", length: 1, headers: map[string]string{"x-amz-content-sha256": "UNSIGNED-PAYLOAD"}},

0 commit comments

Comments
 (0)