Skip to content

Commit d8e7892

Browse files
terion-labsterion-labs
committed
Add S3 mutation support to s3fetch
1 parent 7862170 commit d8e7892

2 files changed

Lines changed: 275 additions & 6 deletions

File tree

‎internal/s3fetch/fetcher.go‎

Lines changed: 112 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"io"
1010
"net/http"
1111
"time"
12+
"unicode"
1213

1314
"github.com/aws/aws-sdk-go-v2/aws"
1415
awsconfig "github.com/aws/aws-sdk-go-v2/config"
@@ -28,12 +29,15 @@ var (
2829
)
2930

3031
type Request struct {
31-
Method string
32-
Operation tickets.Operation
33-
Bucket string
34-
Key string
35-
Range string
36-
List *tickets.ListRequest
32+
Method string
33+
Operation tickets.Operation
34+
Bucket string
35+
Key string
36+
Range string
37+
List *tickets.ListRequest
38+
Body io.Reader
39+
ContentLength *int64
40+
ContentType string
3741
}
3842

3943
type Object struct {
@@ -103,6 +107,10 @@ func (f *Fetcher) Fetch(ctx context.Context, req Request) (*Object, error) {
103107
return f.head(ctx, req)
104108
case tickets.OperationListObjectsV2:
105109
return f.list(ctx, req)
110+
case tickets.OperationPutObject:
111+
return f.put(ctx, req)
112+
case tickets.OperationDeleteObject:
113+
return f.delete(ctx, req)
106114
default:
107115
return f.get(ctx, req)
108116
}
@@ -133,6 +141,38 @@ func (f *Fetcher) get(ctx context.Context, req Request) (*Object, error) {
133141
}, nil
134142
}
135143

144+
func (f *Fetcher) put(ctx context.Context, req Request) (*Object, error) {
145+
input := &s3.PutObjectInput{
146+
Bucket: aws.String(req.Bucket),
147+
Key: aws.String(req.Key),
148+
Body: req.Body,
149+
ContentLength: req.ContentLength,
150+
}
151+
if req.ContentType != "" {
152+
input.ContentType = aws.String(req.ContentType)
153+
}
154+
out, err := f.client.PutObject(ctx, input)
155+
if err != nil {
156+
return nil, mapError(err)
157+
}
158+
return &Object{
159+
StatusCode: http.StatusOK,
160+
ETag: value(out.ETag),
161+
Body: http.NoBody,
162+
}, nil
163+
}
164+
165+
func (f *Fetcher) delete(ctx context.Context, req Request) (*Object, error) {
166+
_, err := f.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(req.Bucket), Key: aws.String(req.Key)})
167+
if err != nil {
168+
return nil, mapError(err)
169+
}
170+
return &Object{
171+
StatusCode: http.StatusNoContent,
172+
Body: http.NoBody,
173+
}, nil
174+
}
175+
136176
func (f *Fetcher) head(ctx context.Context, req Request) (*Object, error) {
137177
input := &s3.HeadObjectInput{Bucket: aws.String(req.Bucket), Key: aws.String(req.Key)}
138178
if req.Range != "" {
@@ -241,6 +281,9 @@ func validateRequest(req Request) (tickets.Operation, error) {
241281

242282
switch operation {
243283
case tickets.OperationGetObject, tickets.OperationHeadObject:
284+
if err := validateMutationFieldsOmitted(req); err != nil {
285+
return "", err
286+
}
244287
if req.List != nil {
245288
return "", fmt.Errorf("%w: list metadata must be omitted for object requests", ErrInvalidRequest)
246289
}
@@ -251,6 +294,9 @@ func validateRequest(req Request) (tickets.Operation, error) {
251294
return "", fmt.Errorf("%w: %v", ErrInvalidRequest, err)
252295
}
253296
case tickets.OperationListObjectsV2:
297+
if err := validateMutationFieldsOmitted(req); err != nil {
298+
return "", err
299+
}
254300
if req.Key != "" {
255301
return "", fmt.Errorf("%w: key must be empty for ListObjectsV2", ErrInvalidRequest)
256302
}
@@ -260,12 +306,72 @@ func validateRequest(req Request) (tickets.Operation, error) {
260306
if err := tickets.ValidateListRequest(req.List); err != nil {
261307
return "", fmt.Errorf("%w: %v", ErrInvalidRequest, err)
262308
}
309+
case tickets.OperationPutObject:
310+
if req.List != nil {
311+
return "", fmt.Errorf("%w: list metadata must be omitted for PutObject", ErrInvalidRequest)
312+
}
313+
if req.Range != "" {
314+
return "", fmt.Errorf("%w: range must be omitted for PutObject", ErrInvalidRequest)
315+
}
316+
if err := tickets.ValidateKey(req.Key); err != nil {
317+
return "", fmt.Errorf("%w: %v", ErrInvalidRequest, err)
318+
}
319+
if req.Body == nil {
320+
return "", fmt.Errorf("%w: body is required for PutObject", ErrInvalidRequest)
321+
}
322+
if req.ContentLength == nil {
323+
return "", fmt.Errorf("%w: content length is required for PutObject", ErrInvalidRequest)
324+
}
325+
if *req.ContentLength < 0 {
326+
return "", fmt.Errorf("%w: content length must be non-negative", ErrInvalidRequest)
327+
}
328+
if err := validateContentType(req.ContentType); err != nil {
329+
return "", err
330+
}
331+
case tickets.OperationDeleteObject:
332+
if err := validateMutationFieldsOmitted(req); err != nil {
333+
return "", err
334+
}
335+
if req.List != nil {
336+
return "", fmt.Errorf("%w: list metadata must be omitted for DeleteObject", ErrInvalidRequest)
337+
}
338+
if req.Range != "" {
339+
return "", fmt.Errorf("%w: range must be omitted for DeleteObject", ErrInvalidRequest)
340+
}
341+
if err := tickets.ValidateKey(req.Key); err != nil {
342+
return "", fmt.Errorf("%w: %v", ErrInvalidRequest, err)
343+
}
263344
default:
264345
return "", fmt.Errorf("%w: unsupported operation", ErrInvalidRequest)
265346
}
266347
return operation, nil
267348
}
268349

350+
func validateMutationFieldsOmitted(req Request) error {
351+
if req.Body != nil {
352+
return fmt.Errorf("%w: body must be omitted", ErrInvalidRequest)
353+
}
354+
if req.ContentLength != nil {
355+
return fmt.Errorf("%w: content length must be omitted", ErrInvalidRequest)
356+
}
357+
if req.ContentType != "" {
358+
return fmt.Errorf("%w: content type must be omitted", ErrInvalidRequest)
359+
}
360+
return nil
361+
}
362+
363+
func validateContentType(contentType string) error {
364+
if len(contentType) > 255 {
365+
return fmt.Errorf("%w: content type is too long", ErrInvalidRequest)
366+
}
367+
for _, r := range contentType {
368+
if r == 0 || unicode.IsControl(r) {
369+
return fmt.Errorf("%w: content type must not contain control characters", ErrInvalidRequest)
370+
}
371+
}
372+
return nil
373+
}
374+
269375
func mapError(err error) error {
270376
var noSuchKey *types.NoSuchKey
271377
if errors.As(err, &noSuchKey) {

‎internal/s3fetch/fetcher_test.go‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,75 @@ func TestFetcherHeadObjectUsesHEADAndNoBody(t *testing.T) {
7676
}
7777
}
7878

79+
func TestFetcherPutObjectStreamsBodyAndMapsETag(t *testing.T) {
80+
var gotMethod, gotPath, gotContentType, gotBody string
81+
var gotContentLength int64
82+
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
83+
gotMethod = r.Method
84+
gotPath = r.URL.Path
85+
gotContentType = r.Header.Get("Content-Type")
86+
gotContentLength = r.ContentLength
87+
body, err := io.ReadAll(r.Body)
88+
if err != nil {
89+
t.Fatalf("ReadAll(request body) error = %v", err)
90+
}
91+
gotBody = string(body)
92+
w.Header().Set("ETag", `"put-etag"`)
93+
w.WriteHeader(http.StatusOK)
94+
}))
95+
defer ts.Close()
96+
97+
fetcher, err := New(context.Background(), testConfig(ts.URL))
98+
if err != nil {
99+
t.Fatalf("New() error = %v", err)
100+
}
101+
contentLength := int64(len("hello mutation"))
102+
obj, err := fetcher.Fetch(context.Background(), Request{
103+
Method: http.MethodPut,
104+
Operation: tickets.OperationPutObject,
105+
Bucket: "demo-bucket",
106+
Key: "objects/file.txt",
107+
Body: strings.NewReader("hello mutation"),
108+
ContentLength: &contentLength,
109+
ContentType: "text/plain",
110+
})
111+
if err != nil {
112+
t.Fatalf("Fetch() error = %v", err)
113+
}
114+
115+
if gotMethod != http.MethodPut || gotPath != "/demo-bucket/objects/file.txt" || gotContentType != "text/plain" || gotContentLength != contentLength || gotBody != "hello mutation" {
116+
t.Fatalf("request method=%q path=%q contentType=%q contentLength=%d body=%q", gotMethod, gotPath, gotContentType, gotContentLength, gotBody)
117+
}
118+
if obj.StatusCode != http.StatusOK || obj.ETag != `"put-etag"` || obj.Body != http.NoBody {
119+
t.Fatalf("object = %#v", obj)
120+
}
121+
}
122+
123+
func TestFetcherDeleteObjectUsesDeleteAndNoBody(t *testing.T) {
124+
var gotMethod, gotPath string
125+
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
126+
gotMethod = r.Method
127+
gotPath = r.URL.Path
128+
w.WriteHeader(http.StatusNoContent)
129+
}))
130+
defer ts.Close()
131+
132+
fetcher, err := New(context.Background(), testConfig(ts.URL))
133+
if err != nil {
134+
t.Fatalf("New() error = %v", err)
135+
}
136+
obj, err := fetcher.Fetch(context.Background(), Request{Method: http.MethodDelete, Operation: tickets.OperationDeleteObject, Bucket: "demo-bucket", Key: "objects/file.txt"})
137+
if err != nil {
138+
t.Fatalf("Fetch() error = %v", err)
139+
}
140+
if gotMethod != http.MethodDelete || gotPath != "/demo-bucket/objects/file.txt" {
141+
t.Fatalf("request method=%q path=%q", gotMethod, gotPath)
142+
}
143+
if obj.StatusCode != http.StatusNoContent || obj.Body != http.NoBody {
144+
t.Fatalf("object = %#v", obj)
145+
}
146+
}
147+
79148
func TestFetcherListObjectsV2RequestShapeAndXMLMetadata(t *testing.T) {
80149
var gotMethod, gotPath string
81150
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -325,6 +394,100 @@ func TestFetcherRejectsInvalidListRequestBeforeS3(t *testing.T) {
325394
}
326395
}
327396

397+
func TestFetcherRejectsInvalidMutationRequestBeforeS3(t *testing.T) {
398+
called := false
399+
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
400+
called = true
401+
}))
402+
defer ts.Close()
403+
404+
fetcher, err := New(context.Background(), testConfig(ts.URL))
405+
if err != nil {
406+
t.Fatalf("New() error = %v", err)
407+
}
408+
validLength := int64(5)
409+
negativeLength := int64(-1)
410+
tests := []struct {
411+
name string
412+
req Request
413+
}{
414+
{
415+
name: "put missing body",
416+
req: Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", ContentLength: &validLength},
417+
},
418+
{
419+
name: "put missing content length",
420+
req: Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", Body: strings.NewReader("hello")},
421+
},
422+
{
423+
name: "put negative content length",
424+
req: Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", Body: strings.NewReader("hello"), ContentLength: &negativeLength},
425+
},
426+
{
427+
name: "put range",
428+
req: Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", Range: "bytes=0-1", Body: strings.NewReader("hello"), ContentLength: &validLength},
429+
},
430+
{
431+
name: "put list metadata",
432+
req: Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", List: &tickets.ListRequest{MaxKeys: 10, Rewrite: tickets.ListRewrite{Bucket: "public-bucket"}}, Body: strings.NewReader("hello"), ContentLength: &validLength},
433+
},
434+
{
435+
name: "put unsafe content type",
436+
req: Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", Body: strings.NewReader("hello"), ContentLength: &validLength, ContentType: "text/plain\n"},
437+
},
438+
{
439+
name: "delete body",
440+
req: Request{Method: http.MethodDelete, Operation: tickets.OperationDeleteObject, Bucket: "demo-bucket", Key: "objects/file.txt", Body: strings.NewReader("hello")},
441+
},
442+
{
443+
name: "delete content length",
444+
req: Request{Method: http.MethodDelete, Operation: tickets.OperationDeleteObject, Bucket: "demo-bucket", Key: "objects/file.txt", ContentLength: &validLength},
445+
},
446+
{
447+
name: "delete range",
448+
req: Request{Method: http.MethodDelete, Operation: tickets.OperationDeleteObject, Bucket: "demo-bucket", Key: "objects/file.txt", Range: "bytes=0-1"},
449+
},
450+
{
451+
name: "delete list metadata",
452+
req: Request{Method: http.MethodDelete, Operation: tickets.OperationDeleteObject, Bucket: "demo-bucket", Key: "objects/file.txt", List: &tickets.ListRequest{MaxKeys: 10, Rewrite: tickets.ListRewrite{Bucket: "public-bucket"}}},
453+
},
454+
}
455+
for _, tc := range tests {
456+
t.Run(tc.name, func(t *testing.T) {
457+
_, err := fetcher.Fetch(context.Background(), tc.req)
458+
if !errors.Is(err, ErrInvalidRequest) {
459+
t.Fatalf("Fetch() error = %v, want ErrInvalidRequest", err)
460+
}
461+
})
462+
}
463+
if called {
464+
t.Fatal("S3 server was called for invalid mutation request")
465+
}
466+
}
467+
468+
func TestFetcherMapsMutationBackendErrors(t *testing.T) {
469+
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
470+
w.Header().Set("Content-Type", "application/xml")
471+
w.WriteHeader(http.StatusNotFound)
472+
_, _ = w.Write([]byte(`<Error><Code>NoSuchBucket</Code><Message>missing bucket</Message></Error>`))
473+
}))
474+
defer ts.Close()
475+
476+
fetcher, err := New(context.Background(), testConfig(ts.URL))
477+
if err != nil {
478+
t.Fatalf("New() error = %v", err)
479+
}
480+
contentLength := int64(5)
481+
_, err = fetcher.Fetch(context.Background(), Request{Method: http.MethodPut, Operation: tickets.OperationPutObject, Bucket: "demo-bucket", Key: "objects/file.txt", Body: strings.NewReader("hello"), ContentLength: &contentLength})
482+
if !errors.Is(err, ErrNotFound) {
483+
t.Fatalf("PutObject Fetch() error = %v, want ErrNotFound", err)
484+
}
485+
_, err = fetcher.Fetch(context.Background(), Request{Method: http.MethodDelete, Operation: tickets.OperationDeleteObject, Bucket: "demo-bucket", Key: "objects/file.txt"})
486+
if !errors.Is(err, ErrNotFound) {
487+
t.Fatalf("DeleteObject Fetch() error = %v, want ErrNotFound", err)
488+
}
489+
}
490+
328491
func TestFetcherMapsMissingBucketForListObjectsV2(t *testing.T) {
329492
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
330493
w.Header().Set("Content-Type", "application/xml")

0 commit comments

Comments
 (0)