From 3dbb67e7db6e50df88a06b79110d176e4c094e1b Mon Sep 17 00:00:00 2001 From: Stuart Dallas Date: Mon, 20 Jul 2026 20:41:36 +0100 Subject: [PATCH] feat: Add ranges, listing pagination, MD5 ETags and metadata support Broadens S3 API compatibility beyond the operations the Docker registry strictly needs: - Range GET: single byte ranges including open-ended and suffix forms, with 206/Content-Range responses and 416 InvalidRange when unsatisfiable; malformed or multi-range headers fall back to a full 200 per RFC 9110 - Conditional GET/HEAD: If-None-Match and If-Modified-Since return 304 - ListObjectsV2 (list-type=2) with KeyCount, continuation tokens and start-after; real pagination for v1 via marker/NextMarker. Listings are now in lexicographic key order with IsTruncated set correctly (previously hard-coded false, so buckets over max-keys silently looked complete) - MD5 ETags: PutObject, CopyObject and multipart complete persist the content MD5 (multipart: md5-of-md5s-N) in a metadata sidecar under .metadata/, so PUT, GET, HEAD and listings all agree; pre-existing objects without a sidecar keep the mtime-based fallback - Content-Type and x-amz-meta-* metadata round-trip through PUT, multipart and copy (honouring x-amz-metadata-directive) - GetBucketLocation and bucket subresource handling: versioning/acl stubs, AWS error codes for absent lifecycle/cors/policy/tagging/ encryption/object-lock config, and no-op PUT/DELETE so clients can no longer create or delete a bucket through a subresource URL, or corrupt object data via PUT ?acl/?tagging - ListBuckets no longer exposes the internal .multipart and .metadata directories as buckets Also fixes a listing bug where a non-matching file visited before the prefix could SkipDir the rest of its directory, dropping matching keys. Verified end-to-end against registry:2 (push/pull with a 40MB layer, zero errors) plus wire-format spot checks of location, V2 pagination, ranges and versioning against a live server. --- pkg/s3/compat_test.go | 623 +++++++++++++++++++++++++++++++++++++++ pkg/s3/handler.go | 393 ++++++++++++++++++++++-- pkg/s3/types.go | 76 ++++- pkg/storage/metadata.go | 60 ++++ pkg/storage/multipart.go | 20 +- pkg/storage/storage.go | 274 +++++++++++++---- 6 files changed, 1363 insertions(+), 83 deletions(-) create mode 100644 pkg/s3/compat_test.go create mode 100644 pkg/storage/metadata.go diff --git a/pkg/s3/compat_test.go b/pkg/s3/compat_test.go new file mode 100644 index 0000000..9490b9d --- /dev/null +++ b/pkg/s3/compat_test.go @@ -0,0 +1,623 @@ +package s3 + +import ( + "crypto/md5" + "encoding/xml" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestPutObjectETagIsMD5(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + + content := "consistent etag content" + expectedETag := fmt.Sprintf("\"%x\"", md5.Sum([]byte(content))) + + req := httptest.NewRequest(http.MethodPut, "/test-bucket/test.txt", strings.NewReader(content)) + req.ContentLength = int64(len(content)) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("Expected status 200, got %d", w.Code) + } + if got := w.Header().Get("ETag"); got != expectedETag { + t.Errorf("PUT: expected ETag %s, got %s", expectedETag, got) + } + + // HEAD, GET and List must agree + req = httptest.NewRequest(http.MethodHead, "/test-bucket/test.txt", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if got := w.Header().Get("ETag"); got != expectedETag { + t.Errorf("HEAD: expected ETag %s, got %s", expectedETag, got) + } + + req = httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if got := w.Header().Get("ETag"); got != expectedETag { + t.Errorf("GET: expected ETag %s, got %s", expectedETag, got) + } + + req = httptest.NewRequest(http.MethodGet, "/test-bucket", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + var listing ListObjectsResponse + if err := xml.Unmarshal(w.Body.Bytes(), &listing); err != nil { + t.Fatalf("Failed to parse listing: %v", err) + } + if len(listing.Contents) != 1 || listing.Contents[0].ETag != expectedETag { + t.Errorf("List: expected ETag %s, got %+v", expectedETag, listing.Contents) + } +} + +func TestObjectMetadataRoundTrip(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + + req := httptest.NewRequest(http.MethodPut, "/test-bucket/doc.html", strings.NewReader("")) + req.ContentLength = 13 + req.Header.Set("Content-Type", "text/html") + req.Header.Set("x-amz-meta-owner", "stuart") + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("Expected status 200, got %d", w.Code) + } + + req = httptest.NewRequest(http.MethodHead, "/test-bucket/doc.html", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if got := w.Header().Get("Content-Type"); got != "text/html" { + t.Errorf("Expected Content-Type text/html, got %s", got) + } + if got := w.Header().Get("x-amz-meta-owner"); got != "stuart" { + t.Errorf("Expected x-amz-meta-owner stuart, got %s", got) + } +} + +func TestCopyObjectMetadata(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + + req := httptest.NewRequest(http.MethodPut, "/test-bucket/source.css", strings.NewReader("body{}")) + req.ContentLength = 6 + req.Header.Set("Content-Type", "text/css") + req.Header.Set("x-amz-meta-origin", "hand-written") + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + // Default COPY directive carries metadata over + req = httptest.NewRequest(http.MethodPut, "/test-bucket/copied.css", nil) + req.Header.Set("x-amz-copy-source", "/test-bucket/source.css") + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("Copy failed: %d %s", w.Code, w.Body.String()) + } + + req = httptest.NewRequest(http.MethodHead, "/test-bucket/copied.css", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if got := w.Header().Get("Content-Type"); got != "text/css" { + t.Errorf("COPY: expected Content-Type text/css, got %s", got) + } + if got := w.Header().Get("x-amz-meta-origin"); got != "hand-written" { + t.Errorf("COPY: expected x-amz-meta-origin hand-written, got %s", got) + } + + // REPLACE directive uses the request's metadata + req = httptest.NewRequest(http.MethodPut, "/test-bucket/replaced.css", nil) + req.Header.Set("x-amz-copy-source", "/test-bucket/source.css") + req.Header.Set("x-amz-metadata-directive", "REPLACE") + req.Header.Set("Content-Type", "text/plain") + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("Copy with REPLACE failed: %d %s", w.Code, w.Body.String()) + } + + req = httptest.NewRequest(http.MethodHead, "/test-bucket/replaced.css", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if got := w.Header().Get("Content-Type"); got != "text/plain" { + t.Errorf("REPLACE: expected Content-Type text/plain, got %s", got) + } + if got := w.Header().Get("x-amz-meta-origin"); got != "" { + t.Errorf("REPLACE: expected no x-amz-meta-origin, got %s", got) + } +} + +func TestMultipartUploadMetadata(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + + // Initiate with metadata + req := httptest.NewRequest(http.MethodPost, "/test-bucket/big.bin?uploads", nil) + req.Header.Set("Content-Type", "application/x-custom") + req.Header.Set("x-amz-meta-source", "multipart") + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("Initiate failed: %d", w.Code) + } + var initResult InitiateMultipartUploadResult + if err := xml.Unmarshal(w.Body.Bytes(), &initResult); err != nil { + t.Fatalf("Failed to parse initiate response: %v", err) + } + + // Upload one part and complete + content := "part content" + req = httptest.NewRequest(http.MethodPut, + fmt.Sprintf("/test-bucket/big.bin?partNumber=1&uploadId=%s", initResult.UploadID), + strings.NewReader(content)) + req.ContentLength = int64(len(content)) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("Upload part failed: %d", w.Code) + } + etag := w.Header().Get("ETag") + + completeXML := fmt.Sprintf( + "1%s", etag) + req = httptest.NewRequest(http.MethodPost, + fmt.Sprintf("/test-bucket/big.bin?uploadId=%s", initResult.UploadID), + strings.NewReader(completeXML)) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("Complete failed: %d %s", w.Code, w.Body.String()) + } + + req = httptest.NewRequest(http.MethodHead, "/test-bucket/big.bin", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if got := w.Header().Get("Content-Type"); got != "application/x-custom" { + t.Errorf("Expected Content-Type application/x-custom, got %s", got) + } + if got := w.Header().Get("x-amz-meta-source"); got != "multipart" { + t.Errorf("Expected x-amz-meta-source multipart, got %s", got) + } + // Multipart ETag has the -N suffix + if got := w.Header().Get("ETag"); !strings.HasSuffix(got, "-1\"") { + t.Errorf("Expected multipart ETag with -1 suffix, got %s", got) + } +} + +func TestGetObjectRange(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + content := "0123456789abcdefghij" + putTestObject(t, handler, "test-bucket", "data.bin", content) + + tests := []struct { + name string + rangeHeader string + expectedCode int + expectedBody string + expectedRange string + }{ + {"first five", "bytes=0-4", http.StatusPartialContent, "01234", "bytes 0-4/20"}, + {"middle", "bytes=5-9", http.StatusPartialContent, "56789", "bytes 5-9/20"}, + {"last byte", "bytes=19-19", http.StatusPartialContent, "j", "bytes 19-19/20"}, + {"open ended", "bytes=15-", http.StatusPartialContent, "fghij", "bytes 15-19/20"}, + {"suffix", "bytes=-5", http.StatusPartialContent, "fghij", "bytes 15-19/20"}, + {"suffix larger than object", "bytes=-100", http.StatusPartialContent, content, "bytes 0-19/20"}, + {"end clamped to size", "bytes=10-100", http.StatusPartialContent, "abcdefghij", "bytes 10-19/20"}, + {"start beyond size", "bytes=20-25", http.StatusRequestedRangeNotSatisfiable, "", ""}, + {"malformed ignored", "bytes=abc", http.StatusOK, content, ""}, + {"multi-range ignored", "bytes=0-1,5-6", http.StatusOK, content, ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/test-bucket/data.bin", nil) + req.Header.Set("Range", tt.rangeHeader) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Code != tt.expectedCode { + t.Fatalf("Expected status %d, got %d: %s", tt.expectedCode, w.Code, w.Body.String()) + } + if tt.expectedCode == http.StatusRequestedRangeNotSatisfiable { + if got := w.Header().Get("Content-Range"); got != "bytes */20" { + t.Errorf("Expected Content-Range bytes */20, got %s", got) + } + return + } + if w.Body.String() != tt.expectedBody { + t.Errorf("Expected body %q, got %q", tt.expectedBody, w.Body.String()) + } + if tt.expectedRange != "" { + if got := w.Header().Get("Content-Range"); got != tt.expectedRange { + t.Errorf("Expected Content-Range %s, got %s", tt.expectedRange, got) + } + } + }) + } +} + +func TestConditionalRequests(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + putTestObject(t, handler, "test-bucket", "test.txt", "conditional content") + + // Get the current ETag + req := httptest.NewRequest(http.MethodHead, "/test-bucket/test.txt", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + etag := w.Header().Get("ETag") + + // Matching If-None-Match returns 304 on GET and HEAD + for _, method := range []string{http.MethodGet, http.MethodHead} { + req = httptest.NewRequest(method, "/test-bucket/test.txt", nil) + req.Header.Set("If-None-Match", etag) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusNotModified { + t.Errorf("%s If-None-Match: expected 304, got %d", method, w.Code) + } + } + + // Non-matching If-None-Match returns the object + req = httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt", nil) + req.Header.Set("If-None-Match", "\"different\"") + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Errorf("Non-matching If-None-Match: expected 200, got %d", w.Code) + } + + // If-Modified-Since in the future returns 304 + req = httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt", nil) + req.Header.Set("If-Modified-Since", time.Now().Add(time.Hour).UTC().Format(http.TimeFormat)) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusNotModified { + t.Errorf("Future If-Modified-Since: expected 304, got %d", w.Code) + } + + // If-Modified-Since in the past returns the object + req = httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt", nil) + req.Header.Set("If-Modified-Since", time.Now().Add(-time.Hour).UTC().Format(http.TimeFormat)) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Errorf("Past If-Modified-Since: expected 200, got %d", w.Code) + } +} + +func listKeys(t *testing.T, handler *Handler, url string) ([]string, ListObjectsV2Response) { + t.Helper() + + req := httptest.NewRequest(http.MethodGet, url, nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("List %s failed: %d %s", url, w.Code, w.Body.String()) + } + + var response ListObjectsV2Response + if err := xml.Unmarshal(w.Body.Bytes(), &response); err != nil { + t.Fatalf("Failed to parse listing: %v", err) + } + + var keys []string + for _, obj := range response.Contents { + keys = append(keys, obj.Key) + } + return keys, response +} + +func TestListObjectsV1Pagination(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + for _, key := range []string{"a.txt", "b.txt", "c.txt", "d.txt", "e.txt"} { + putTestObject(t, handler, "test-bucket", key, "x") + } + + // First page + req := httptest.NewRequest(http.MethodGet, "/test-bucket?max-keys=2", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + var page1 ListObjectsResponse + if err := xml.Unmarshal(w.Body.Bytes(), &page1); err != nil { + t.Fatalf("Failed to parse page 1: %v", err) + } + if len(page1.Contents) != 2 || !page1.IsTruncated { + t.Fatalf("Expected 2 keys and truncation, got %d keys truncated=%v", len(page1.Contents), page1.IsTruncated) + } + if page1.NextMarker != "b.txt" { + t.Errorf("Expected NextMarker b.txt, got %s", page1.NextMarker) + } + + // Second page resumes after the marker + req = httptest.NewRequest(http.MethodGet, "/test-bucket?max-keys=2&marker="+page1.NextMarker, nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + var page2 ListObjectsResponse + if err := xml.Unmarshal(w.Body.Bytes(), &page2); err != nil { + t.Fatalf("Failed to parse page 2: %v", err) + } + if len(page2.Contents) != 2 || page2.Contents[0].Key != "c.txt" || page2.Contents[1].Key != "d.txt" { + t.Errorf("Expected [c.txt d.txt], got %+v", page2.Contents) + } + + // Final page + req = httptest.NewRequest(http.MethodGet, "/test-bucket?max-keys=2&marker="+page2.NextMarker, nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + var page3 ListObjectsResponse + if err := xml.Unmarshal(w.Body.Bytes(), &page3); err != nil { + t.Fatalf("Failed to parse page 3: %v", err) + } + if len(page3.Contents) != 1 || page3.Contents[0].Key != "e.txt" || page3.IsTruncated { + t.Errorf("Expected final page [e.txt] untruncated, got %+v truncated=%v", page3.Contents, page3.IsTruncated) + } +} + +func TestListObjectsV2(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + for _, key := range []string{"a.txt", "b.txt", "c.txt", "dir/nested.txt", "e.txt"} { + putTestObject(t, handler, "test-bucket", key, "x") + } + + // Full listing includes KeyCount + keys, response := listKeys(t, handler, "/test-bucket?list-type=2") + if len(keys) != 5 || response.KeyCount != 5 { + t.Errorf("Expected 5 keys with KeyCount 5, got %d keys KeyCount=%d", len(keys), response.KeyCount) + } + + // Paginate via continuation tokens, collecting everything + var collected []string + token := "" + for page := 0; page < 10; page++ { + url := "/test-bucket?list-type=2&max-keys=2" + if token != "" { + url += "&continuation-token=" + token + } + keys, response := listKeys(t, handler, url) + collected = append(collected, keys...) + if !response.IsTruncated { + break + } + if response.NextContinuationToken == "" { + t.Fatal("Truncated response missing NextContinuationToken") + } + token = response.NextContinuationToken + } + expected := []string{"a.txt", "b.txt", "c.txt", "dir/nested.txt", "e.txt"} + if strings.Join(collected, ",") != strings.Join(expected, ",") { + t.Errorf("Expected %v, got %v", expected, collected) + } + + // start-after + keys, _ = listKeys(t, handler, "/test-bucket?list-type=2&start-after=b.txt") + if strings.Join(keys, ",") != "c.txt,dir/nested.txt,e.txt" { + t.Errorf("Expected keys after b.txt, got %v", keys) + } + + // Delimiter rolls up common prefixes, which count toward KeyCount + keys, response = listKeys(t, handler, "/test-bucket?list-type=2&delimiter=/") + if len(keys) != 4 || len(response.CommonPrefixes) != 1 || response.CommonPrefixes[0].Prefix != "dir/" { + t.Errorf("Expected 4 keys + prefix dir/, got %v %+v", keys, response.CommonPrefixes) + } + if response.KeyCount != 5 { + t.Errorf("Expected KeyCount 5 (4 keys + 1 prefix), got %d", response.KeyCount) + } +} + +func TestListObjectsDelimiterPagination(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + for _, key := range []string{"a/1.txt", "a/2.txt", "b/1.txt", "c.txt", "d/1.txt"} { + putTestObject(t, handler, "test-bucket", key, "x") + } + + // Page through with delimiter; prefixes and keys interleave in key order + var entries []string + token := "" + for page := 0; page < 10; page++ { + url := "/test-bucket?list-type=2&delimiter=/&max-keys=2" + if token != "" { + url += "&continuation-token=" + token + } + keys, response := listKeys(t, handler, url) + entries = append(entries, keys...) + for _, cp := range response.CommonPrefixes { + entries = append(entries, cp.Prefix) + } + if !response.IsTruncated { + break + } + token = response.NextContinuationToken + } + + expected := map[string]bool{"a/": true, "b/": true, "c.txt": true, "d/": true} + if len(entries) != 4 { + t.Fatalf("Expected 4 entries, got %v", entries) + } + for _, e := range entries { + if !expected[e] { + t.Errorf("Unexpected entry %s in %v", e, entries) + } + } +} + +func TestGetBucketLocation(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + + req := httptest.NewRequest(http.MethodGet, "/test-bucket?location", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("Expected status 200, got %d: %s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "LocationConstraint") { + t.Errorf("Expected LocationConstraint response, got %s", w.Body.String()) + } + + // Missing bucket + req = httptest.NewRequest(http.MethodGet, "/no-such-bucket?location", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusNotFound { + t.Errorf("Expected 404 for missing bucket, got %d", w.Code) + } +} + +func TestBucketSubresources(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + + tests := []struct { + query string + expectedCode int + expectedBody string + }{ + {"versioning", http.StatusOK, "VersioningConfiguration"}, + {"acl", http.StatusOK, "AccessControlPolicy"}, + {"tagging", http.StatusNotFound, "NoSuchTagSet"}, + {"lifecycle", http.StatusNotFound, "NoSuchLifecycleConfiguration"}, + {"cors", http.StatusNotFound, "NoSuchCORSConfiguration"}, + {"policy", http.StatusNotFound, "NoSuchBucketPolicy"}, + {"encryption", http.StatusNotFound, "ServerSideEncryptionConfigurationNotFoundError"}, + {"object-lock", http.StatusNotFound, "ObjectLockConfigurationNotFoundError"}, + } + + for _, tt := range tests { + t.Run("GET "+tt.query, func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/test-bucket?"+tt.query, nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + if w.Code != tt.expectedCode { + t.Errorf("Expected status %d, got %d", tt.expectedCode, w.Code) + } + if !strings.Contains(w.Body.String(), tt.expectedBody) { + t.Errorf("Expected %s in response, got %s", tt.expectedBody, w.Body.String()) + } + }) + } + + // PUT of a subresource is a no-op, not a bucket creation + req := httptest.NewRequest(http.MethodPut, "/test-bucket?versioning", strings.NewReader("")) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Errorf("PUT ?versioning: expected 200, got %d: %s", w.Code, w.Body.String()) + } + + // DELETE of a subresource must NOT delete the bucket + req = httptest.NewRequest(http.MethodDelete, "/test-bucket?lifecycle", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusNoContent { + t.Errorf("DELETE ?lifecycle: expected 204, got %d", w.Code) + } + if err := store.HeadBucket("test-bucket"); err != nil { + t.Error("Bucket was deleted by a subresource DELETE") + } +} + +func TestObjectSubresources(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("test-bucket") + content := "protected content" + putTestObject(t, handler, "test-bucket", "test.txt", content) + + // GET ?acl returns a stub policy + req := httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt?acl", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), "AccessControlPolicy") { + t.Errorf("GET ?acl: expected 200 with policy, got %d: %s", w.Code, w.Body.String()) + } + + // GET ?tagging returns an empty tag set + req = httptest.NewRequest(http.MethodGet, "/test-bucket/test.txt?tagging", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), "Tagging") { + t.Errorf("GET ?tagging: expected 200 with Tagging, got %d: %s", w.Code, w.Body.String()) + } + + // PUT ?acl must not overwrite the object data + req = httptest.NewRequest(http.MethodPut, "/test-bucket/test.txt?acl", strings.NewReader("")) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Errorf("PUT ?acl: expected 200, got %d", w.Code) + } + code, body := getTestObject(t, handler, "test-bucket", "test.txt") + if code != http.StatusOK || body != content { + t.Errorf("Object corrupted by PUT ?acl: %d %q", code, body) + } + + // Missing object + req = httptest.NewRequest(http.MethodGet, "/test-bucket/no-such-key?acl", nil) + w = httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusNotFound { + t.Errorf("GET ?acl on missing key: expected 404, got %d", w.Code) + } +} + +func TestListBucketsHidesInternalDirs(t *testing.T) { + handler, store, cleanup := setupTestHandler(t) + defer cleanup() + + store.CreateBucket("visible-bucket") + // Object metadata and multipart state must not surface as buckets + putTestObject(t, handler, "visible-bucket", "file.txt", "content") + if _, err := store.InitiateMultipartUpload("visible-bucket", "pending.bin"); err != nil { + t.Fatalf("Failed to initiate upload: %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/", nil) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + + var response ListBucketsResponse + if err := xml.Unmarshal(w.Body.Bytes(), &response); err != nil { + t.Fatalf("Failed to parse response: %v", err) + } + if len(response.Buckets.Buckets) != 1 || response.Buckets.Buckets[0].Name != "visible-bucket" { + t.Errorf("Expected only visible-bucket, got %+v", response.Buckets.Buckets) + } +} diff --git a/pkg/s3/handler.go b/pkg/s3/handler.go index 06cbd5a..df38832 100644 --- a/pkg/s3/handler.go +++ b/pkg/s3/handler.go @@ -1,6 +1,7 @@ package s3 import ( + "encoding/base64" "encoding/xml" "fmt" "io" @@ -65,16 +66,40 @@ func (h *Handler) handleServiceOperation(w http.ResponseWriter, r *http.Request) } } +// bucketSubresources are the bucket configuration subresources s3dir +// recognises but does not implement. GETs receive a stub or the S3 error code +// a real bucket without that configuration would return; PUTs and DELETEs are +// accepted as no-ops so clients cannot accidentally create or delete the +// bucket itself through them +var bucketSubresources = []string{ + "accelerate", "acl", "cors", "encryption", "lifecycle", "location", + "logging", "notification", "object-lock", "policy", "replication", + "requestPayment", "tagging", "versioning", "website", +} + +// hasBucketSubresource reports whether the query addresses a recognised +// bucket configuration subresource +func hasBucketSubresource(query url.Values) bool { + for _, sub := range bucketSubresources { + if query.Has(sub) { + return true + } + } + return false +} + // handleBucketOperation handles bucket-level operations func (h *Handler) handleBucketOperation(w http.ResponseWriter, r *http.Request, bucket string) { + query := r.URL.Query() + // Check for multipart uploads listing - if r.Method == http.MethodGet && r.URL.Query().Has("uploads") { + if r.Method == http.MethodGet && query.Has("uploads") { h.listMultipartUploads(w, r, bucket) return } // Check for batch delete - if r.Method == http.MethodPost && r.URL.Query().Has("delete") { + if r.Method == http.MethodPost && query.Has("delete") { if h.readOnly { writeError(w, "AccessDenied", "Read-only mode", http.StatusForbidden) return @@ -83,6 +108,12 @@ func (h *Handler) handleBucketOperation(w http.ResponseWriter, r *http.Request, return } + // Bucket configuration subresources + if hasBucketSubresource(query) { + h.handleBucketSubresource(w, r, bucket) + return + } + switch r.Method { case http.MethodGet: h.listObjects(w, r, bucket) @@ -157,6 +188,13 @@ func (h *Handler) handleObjectOperation(w http.ResponseWriter, r *http.Request, return } + // Object subresources (?acl, ?tagging): served as stubs so clients don't + // corrupt object data through the plain PUT path + if query.Has("acl") || query.Has("tagging") { + h.handleObjectSubresource(w, r, bucket, key) + return + } + // Standard object operations switch r.Method { case http.MethodGet: @@ -184,6 +222,78 @@ func (h *Handler) handleObjectOperation(w http.ResponseWriter, r *http.Request, } } +// handleBucketSubresource handles requests addressing bucket configuration +// subresources (?location, ?versioning, ?acl, ?lifecycle, ...) +func (h *Handler) handleBucketSubresource(w http.ResponseWriter, r *http.Request, bucket string) { + if err := h.storage.HeadBucket(bucket); err != nil { + writeError(w, "NoSuchBucket", "The specified bucket does not exist", http.StatusNotFound) + return + } + + query := r.URL.Query() + + switch r.Method { + case http.MethodGet: + switch { + case query.Has("location"): + // Empty value means us-east-1, matching AWS + writeXML(w, LocationConstraint{}, http.StatusOK) + case query.Has("versioning"): + writeXML(w, VersioningConfiguration{}, http.StatusOK) + case query.Has("acl"): + writeXML(w, ownerFullControlACL(), http.StatusOK) + case query.Has("tagging"): + writeError(w, "NoSuchTagSet", "The TagSet does not exist", http.StatusNotFound) + case query.Has("lifecycle"): + writeError(w, "NoSuchLifecycleConfiguration", "The lifecycle configuration does not exist", http.StatusNotFound) + case query.Has("cors"): + writeError(w, "NoSuchCORSConfiguration", "The CORS configuration does not exist", http.StatusNotFound) + case query.Has("policy"): + writeError(w, "NoSuchBucketPolicy", "The bucket policy does not exist", http.StatusNotFound) + case query.Has("encryption"): + writeError(w, "ServerSideEncryptionConfigurationNotFoundError", "The server side encryption configuration was not found", http.StatusNotFound) + case query.Has("object-lock"): + writeError(w, "ObjectLockConfigurationNotFoundError", "Object Lock configuration does not exist for this bucket", http.StatusNotFound) + default: + writeError(w, "NotImplemented", "This bucket subresource is not implemented", http.StatusNotImplemented) + } + case http.MethodPut: + if h.readOnly { + writeError(w, "AccessDenied", "Read-only mode", http.StatusForbidden) + return + } + // Accept configuration writes as no-ops + w.WriteHeader(http.StatusOK) + case http.MethodDelete: + if h.readOnly { + writeError(w, "AccessDenied", "Read-only mode", http.StatusForbidden) + return + } + // Accept configuration deletes as no-ops (this must not delete the bucket) + w.WriteHeader(http.StatusNoContent) + default: + writeError(w, "MethodNotAllowed", "Method not allowed", http.StatusMethodNotAllowed) + } +} + +// ownerFullControlACL is the stub ACL granting the s3dir owner full control +func ownerFullControlACL() AccessControlPolicy { + return AccessControlPolicy{ + Owner: Owner{ID: "s3dir", DisplayName: "s3dir"}, + AccessControlList: AccessControlList{ + Grants: []Grant{{ + Grantee: Grantee{ + XMLNSXSI: "http://www.w3.org/2001/XMLSchema-instance", + Type: "CanonicalUser", + ID: "s3dir", + DisplayName: "s3dir", + }, + Permission: "FULL_CONTROL", + }}, + }, + } +} + // listBuckets lists all buckets func (h *Handler) listBuckets(w http.ResponseWriter, r *http.Request) { buckets, err := h.storage.ListBuckets() @@ -211,19 +321,39 @@ func (h *Handler) listBuckets(w http.ResponseWriter, r *http.Request) { writeXML(w, response, http.StatusOK) } -// listObjects lists objects in a bucket +// listObjects lists objects in a bucket, handling both ListObjects (v1) and +// ListObjectsV2 requests func (h *Handler) listObjects(w http.ResponseWriter, r *http.Request, bucket string) { query := r.URL.Query() prefix := query.Get("prefix") delimiter := query.Get("delimiter") maxKeys := 1000 if mk := query.Get("max-keys"); mk != "" { - if n, err := strconv.Atoi(mk); err == nil && n > 0 { + if n, err := strconv.Atoi(mk); err == nil && n >= 0 { maxKeys = n } } - objects, commonPrefixes, err := h.storage.ListObjects(bucket, prefix, delimiter, maxKeys) + // Determine where to resume from. V1 uses marker; V2 uses an opaque + // continuation token (base64 of the last returned entry), falling back to + // start-after on the first page + listV2 := query.Get("list-type") == "2" + marker := query.Get("marker") + continuationToken := query.Get("continuation-token") + startAfter := query.Get("start-after") + if listV2 { + marker = startAfter + if continuationToken != "" { + decoded, err := base64.StdEncoding.DecodeString(continuationToken) + if err != nil { + writeError(w, "InvalidArgument", "The continuation token provided is incorrect", http.StatusBadRequest) + return + } + marker = string(decoded) + } + } + + objects, commonPrefixes, truncated, nextMarker, err := h.storage.ListObjectsPage(bucket, prefix, delimiter, marker, maxKeys) if err != nil { if strings.Contains(err.Error(), "not found") { writeError(w, "NoSuchBucket", "The specified bucket does not exist", http.StatusNotFound) @@ -249,12 +379,34 @@ func (h *Handler) listObjects(w http.ResponseWriter, r *http.Request, bucket str prefixes = append(prefixes, CommonPrefix{Prefix: cp}) } + if listV2 { + response := ListObjectsV2Response{ + Name: bucket, + Prefix: prefix, + Delimiter: delimiter, + StartAfter: startAfter, + ContinuationToken: continuationToken, + KeyCount: len(contents) + len(prefixes), + MaxKeys: maxKeys, + IsTruncated: truncated, + Contents: contents, + CommonPrefixes: prefixes, + } + if truncated { + response.NextContinuationToken = base64.StdEncoding.EncodeToString([]byte(nextMarker)) + } + writeXML(w, response, http.StatusOK) + return + } + response := ListObjectsResponse{ Name: bucket, Prefix: prefix, Delimiter: delimiter, + Marker: marker, + NextMarker: nextMarker, MaxKeys: maxKeys, - IsTruncated: false, + IsTruncated: truncated, Contents: contents, CommonPrefixes: prefixes, } @@ -262,9 +414,105 @@ func (h *Handler) listObjects(w http.ResponseWriter, r *http.Request, bucket str writeXML(w, response, http.StatusOK) } -// getObject retrieves an object +// setObjectHeaders sets the standard response headers for an object +func setObjectHeaders(w http.ResponseWriter, info *storage.ObjectInfo) { + contentType := info.ContentType + if contentType == "" { + contentType = "application/octet-stream" + } + w.Header().Set("Content-Type", contentType) + w.Header().Set("ETag", info.ETag) + w.Header().Set("Last-Modified", info.LastModified.UTC().Format(http.TimeFormat)) + w.Header().Set("Accept-Ranges", "bytes") + for name, value := range info.UserMetadata { + w.Header().Set("x-amz-meta-"+name, value) + } +} + +// checkNotModified evaluates the If-None-Match and If-Modified-Since request +// headers against the object, reporting whether a 304 should be returned. +// If-None-Match takes precedence when present, per RFC 9110 +func checkNotModified(r *http.Request, info *storage.ObjectInfo) bool { + if inm := r.Header.Get("If-None-Match"); inm != "" { + if inm == "*" { + return true + } + for _, candidate := range strings.Split(inm, ",") { + candidate = strings.TrimPrefix(strings.TrimSpace(candidate), "W/") + if candidate == info.ETag { + return true + } + } + return false + } + + if ims := r.Header.Get("If-Modified-Since"); ims != "" { + if t, err := http.ParseTime(ims); err == nil { + // HTTP dates have second precision + return !info.LastModified.Truncate(time.Second).After(t) + } + } + + return false +} + +// parseRangeHeader parses a single Range request header against an object of +// the given size. valid is false for headers that should be ignored (malformed +// or multi-range); satisfiable is false when a valid range lies outside the +// object and a 416 must be returned +func parseRangeHeader(value string, size int64) (start, length int64, valid, satisfiable bool) { + spec, ok := strings.CutPrefix(value, "bytes=") + if !ok || strings.Contains(spec, ",") { + return 0, 0, false, false + } + + startStr, endStr, ok := strings.Cut(spec, "-") + if !ok { + return 0, 0, false, false + } + + if startStr == "" { + // Suffix range: last N bytes + suffix, err := strconv.ParseInt(endStr, 10, 64) + if err != nil || suffix < 0 { + return 0, 0, false, false + } + if suffix == 0 || size == 0 { + return 0, 0, true, false + } + if suffix > size { + suffix = size + } + return size - suffix, suffix, true, true + } + + first, err := strconv.ParseInt(startStr, 10, 64) + if err != nil || first < 0 { + return 0, 0, false, false + } + + last := size - 1 + if endStr != "" { + last, err = strconv.ParseInt(endStr, 10, 64) + if err != nil || last < first { + return 0, 0, false, false + } + if last > size-1 { + last = size - 1 + } + } + + if first >= size { + return 0, 0, true, false + } + + return first, last - first + 1, true, true +} + +// getObject retrieves an object, honouring Range and conditional request +// headers func (h *Handler) getObject(w http.ResponseWriter, r *http.Request, bucket, key string) { - reader, info, err := h.storage.GetObject(bucket, key) + info, err := h.storage.HeadObject(bucket, key) if err != nil { if strings.Contains(err.Error(), "not found") { writeError(w, "NoSuchKey", "The specified key does not exist", http.StatusNotFound) @@ -273,13 +521,54 @@ func (h *Handler) getObject(w http.ResponseWriter, r *http.Request, bucket, key } return } + + if checkNotModified(r, info) { + w.Header().Set("ETag", info.ETag) + w.Header().Set("Last-Modified", info.LastModified.UTC().Format(http.TimeFormat)) + w.WriteHeader(http.StatusNotModified) + return + } + + // Ranged read + if rangeHeader := r.Header.Get("Range"); rangeHeader != "" { + start, length, valid, satisfiable := parseRangeHeader(rangeHeader, info.Size) + if valid { + if !satisfiable { + w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", info.Size)) + writeError(w, "InvalidRange", "The requested range is not satisfiable", http.StatusRequestedRangeNotSatisfiable) + return + } + + reader, _, err := h.storage.GetObjectRange(bucket, key, start, length) + if err != nil { + writeError(w, "InternalError", err.Error(), http.StatusInternalServerError) + return + } + defer reader.Close() + + setObjectHeaders(w, info) + w.Header().Set("Content-Length", strconv.FormatInt(length, 10)) + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, info.Size)) + w.WriteHeader(http.StatusPartialContent) + + if _, err := io.Copy(w, reader); err != nil { + // Error writing to response - headers are already sent + return + } + return + } + // Malformed range headers are ignored and the full object returned + } + + reader, _, err := h.storage.GetObject(bucket, key) + if err != nil { + writeError(w, "InternalError", err.Error(), http.StatusInternalServerError) + return + } defer reader.Close() - // Set headers - w.Header().Set("Content-Type", "application/octet-stream") + setObjectHeaders(w, info) w.Header().Set("Content-Length", strconv.FormatInt(info.Size, 10)) - w.Header().Set("ETag", info.ETag) - w.Header().Set("Last-Modified", info.LastModified.UTC().Format(http.TimeFormat)) // Copy object data to response if _, err := io.Copy(w, reader); err != nil { @@ -301,14 +590,33 @@ func (h *Handler) headObject(w http.ResponseWriter, r *http.Request, bucket, key return } - // Set headers - w.Header().Set("Content-Type", "application/octet-stream") + if checkNotModified(r, info) { + w.Header().Set("ETag", info.ETag) + w.Header().Set("Last-Modified", info.LastModified.UTC().Format(http.TimeFormat)) + w.WriteHeader(http.StatusNotModified) + return + } + + setObjectHeaders(w, info) w.Header().Set("Content-Length", strconv.FormatInt(info.Size, 10)) - w.Header().Set("ETag", info.ETag) - w.Header().Set("Last-Modified", info.LastModified.UTC().Format(http.TimeFormat)) w.WriteHeader(http.StatusOK) } +// userMetadataFromHeader extracts x-amz-meta-* headers into a metadata map +// with lowercased names +func userMetadataFromHeader(header http.Header) map[string]string { + var meta map[string]string + for name, values := range header { + if strings.HasPrefix(name, "X-Amz-Meta-") && len(values) > 0 { + if meta == nil { + meta = make(map[string]string) + } + meta[strings.ToLower(strings.TrimPrefix(name, "X-Amz-Meta-"))] = values[0] + } + } + return meta +} + // putObject stores an object func (h *Handler) putObject(w http.ResponseWriter, r *http.Request, bucket, key string) { contentLength := r.ContentLength @@ -317,13 +625,13 @@ func (h *Handler) putObject(w http.ResponseWriter, r *http.Request, bucket, key return } - if err := h.storage.PutObject(bucket, key, r.Body, contentLength); err != nil { + etag, err := h.storage.PutObjectWithMetadata(bucket, key, r.Body, contentLength, + r.Header.Get("Content-Type"), userMetadataFromHeader(r.Header)) + if err != nil { writeError(w, "InternalError", err.Error(), http.StatusInternalServerError) return } - // Generate ETag (simplified) - etag := fmt.Sprintf("\"%x\"", time.Now().Unix()) w.Header().Set("ETag", etag) w.WriteHeader(http.StatusOK) } @@ -432,6 +740,46 @@ func writeError(w http.ResponseWriter, code, message string, statusCode int) { writeXML(w, errorResponse, statusCode) } +// handleObjectSubresource handles requests addressing object subresources +// (?acl, ?tagging) +func (h *Handler) handleObjectSubresource(w http.ResponseWriter, r *http.Request, bucket, key string) { + if _, err := h.storage.HeadObject(bucket, key); err != nil { + if strings.Contains(err.Error(), "not found") { + writeError(w, "NoSuchKey", "The specified key does not exist", http.StatusNotFound) + } else { + writeError(w, "InternalError", err.Error(), http.StatusInternalServerError) + } + return + } + + query := r.URL.Query() + + switch r.Method { + case http.MethodGet: + if query.Has("acl") { + writeXML(w, ownerFullControlACL(), http.StatusOK) + } else { + // Objects have no tags; an empty TagSet is the AWS response + writeXML(w, Tagging{}, http.StatusOK) + } + case http.MethodPut: + if h.readOnly { + writeError(w, "AccessDenied", "Read-only mode", http.StatusForbidden) + return + } + // Accept ACL and tagging writes as no-ops + w.WriteHeader(http.StatusOK) + case http.MethodDelete: + if h.readOnly { + writeError(w, "AccessDenied", "Read-only mode", http.StatusForbidden) + return + } + w.WriteHeader(http.StatusNoContent) + default: + writeError(w, "MethodNotAllowed", "Method not allowed", http.StatusMethodNotAllowed) + } +} + // parseCopySource parses an x-amz-copy-source header value into bucket and key. // The value is "/{bucket}/{key}" or "{bucket}/{key}" and may be URL-encoded func parseCopySource(source string) (bucket, key string, err error) { @@ -488,7 +836,9 @@ func (h *Handler) copyObject(w http.ResponseWriter, r *http.Request, bucket, key return } - info, err := h.storage.CopyObject(srcBucket, srcKey, bucket, key) + replaceMetadata := strings.EqualFold(r.Header.Get("x-amz-metadata-directive"), "REPLACE") + info, err := h.storage.CopyObjectWithMetadata(srcBucket, srcKey, bucket, key, + replaceMetadata, r.Header.Get("Content-Type"), userMetadataFromHeader(r.Header)) if err != nil { if strings.Contains(err.Error(), "not found") { writeError(w, "NoSuchKey", "The specified key does not exist", http.StatusNotFound) @@ -581,7 +931,8 @@ func (h *Handler) deleteObjects(w http.ResponseWriter, r *http.Request, bucket s // initiateMultipartUpload initiates a multipart upload func (h *Handler) initiateMultipartUpload(w http.ResponseWriter, r *http.Request, bucket, key string) { - uploadID, err := h.storage.InitiateMultipartUpload(bucket, key) + uploadID, err := h.storage.InitiateMultipartUploadWithMetadata(bucket, key, + r.Header.Get("Content-Type"), userMetadataFromHeader(r.Header)) if err != nil { writeError(w, "InternalError", err.Error(), http.StatusInternalServerError) return diff --git a/pkg/s3/types.go b/pkg/s3/types.go index 588aac4..5b901e9 100644 --- a/pkg/s3/types.go +++ b/pkg/s3/types.go @@ -26,18 +26,36 @@ type Owner struct { DisplayName string `xml:"DisplayName"` } -// ListObjectsResponse is the response for ListObjects operation +// ListObjectsResponse is the response for the ListObjects (v1) operation type ListObjectsResponse struct { XMLName xml.Name `xml:"ListBucketResult"` Name string `xml:"Name"` Prefix string `xml:"Prefix,omitempty"` Delimiter string `xml:"Delimiter,omitempty"` + Marker string `xml:"Marker"` + NextMarker string `xml:"NextMarker,omitempty"` MaxKeys int `xml:"MaxKeys"` IsTruncated bool `xml:"IsTruncated"` Contents []Object `xml:"Contents"` CommonPrefixes []CommonPrefix `xml:"CommonPrefixes,omitempty"` } +// ListObjectsV2Response is the response for the ListObjectsV2 operation +type ListObjectsV2Response struct { + XMLName xml.Name `xml:"ListBucketResult"` + Name string `xml:"Name"` + Prefix string `xml:"Prefix,omitempty"` + Delimiter string `xml:"Delimiter,omitempty"` + StartAfter string `xml:"StartAfter,omitempty"` + ContinuationToken string `xml:"ContinuationToken,omitempty"` + NextContinuationToken string `xml:"NextContinuationToken,omitempty"` + KeyCount int `xml:"KeyCount"` + MaxKeys int `xml:"MaxKeys"` + IsTruncated bool `xml:"IsTruncated"` + Contents []Object `xml:"Contents"` + CommonPrefixes []CommonPrefix `xml:"CommonPrefixes,omitempty"` +} + // Object represents an S3 object type Object struct { Key string `xml:"Key"` @@ -131,6 +149,62 @@ type ListMultipartUploadsResult struct { Uploads []Upload `xml:"Upload"` } +// LocationConstraint is the response for GetBucketLocation. An empty value +// means the default region (us-east-1), matching AWS behaviour +type LocationConstraint struct { + XMLName xml.Name `xml:"LocationConstraint"` + Value string `xml:",chardata"` +} + +// VersioningConfiguration is the response for GetBucketVersioning. An empty +// configuration means versioning has never been enabled +type VersioningConfiguration struct { + XMLName xml.Name `xml:"VersioningConfiguration"` +} + +// AccessControlPolicy is the response for GetBucketAcl and GetObjectAcl +type AccessControlPolicy struct { + XMLName xml.Name `xml:"AccessControlPolicy"` + Owner Owner `xml:"Owner"` + AccessControlList AccessControlList `xml:"AccessControlList"` +} + +// AccessControlList contains the grants of an access control policy +type AccessControlList struct { + Grants []Grant `xml:"Grant"` +} + +// Grant represents a single ACL grant +type Grant struct { + Grantee Grantee `xml:"Grantee"` + Permission string `xml:"Permission"` +} + +// Grantee identifies who a grant applies to +type Grantee struct { + XMLNSXSI string `xml:"xmlns:xsi,attr"` + Type string `xml:"xsi:type,attr"` + ID string `xml:"ID"` + DisplayName string `xml:"DisplayName"` +} + +// Tagging is the response for GetObjectTagging +type Tagging struct { + XMLName xml.Name `xml:"Tagging"` + TagSet TagSet `xml:"TagSet"` +} + +// TagSet contains the tags of a tagging configuration +type TagSet struct { + Tags []Tag `xml:"Tag"` +} + +// Tag is a single key/value tag +type Tag struct { + Key string `xml:"Key"` + Value string `xml:"Value"` +} + // CopyObjectResult is the response for CopyObject type CopyObjectResult struct { XMLName xml.Name `xml:"CopyObjectResult"` diff --git a/pkg/storage/metadata.go b/pkg/storage/metadata.go new file mode 100644 index 0000000..02d6a97 --- /dev/null +++ b/pkg/storage/metadata.go @@ -0,0 +1,60 @@ +package storage + +import ( + "encoding/json" + "os" + "path/filepath" +) + +// metadataDirName is the directory under baseDir holding object metadata +// sidecar files, mirroring the bucket/key layout +const metadataDirName = ".metadata" + +// objectMetadata is persisted as a JSON sidecar for each object. The ETag is +// stored without surrounding quotes +type objectMetadata struct { + ETag string `json:"etag,omitempty"` + ContentType string `json:"contentType,omitempty"` + UserMetadata map[string]string `json:"userMetadata,omitempty"` +} + +func objectMetadataPath(baseDir, bucket, key string) string { + return filepath.Join(baseDir, metadataDirName, bucket, filepath.FromSlash(key)+".json") +} + +// writeObjectMetadataFile persists the metadata sidecar for an object +func writeObjectMetadataFile(baseDir, bucket, key string, meta *objectMetadata) error { + path := objectMetadataPath(baseDir, bucket, key) + + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { + return err + } + + data, err := json.Marshal(meta) + if err != nil { + return err + } + + return os.WriteFile(path, data, 0644) +} + +// readObjectMetadataFile loads the metadata sidecar for an object, returning +// nil if no sidecar exists (e.g. objects created before metadata support) +func readObjectMetadataFile(baseDir, bucket, key string) *objectMetadata { + data, err := os.ReadFile(objectMetadataPath(baseDir, bucket, key)) + if err != nil { + return nil + } + + var meta objectMetadata + if err := json.Unmarshal(data, &meta); err != nil { + return nil + } + + return &meta +} + +// removeObjectMetadataFile deletes the metadata sidecar for an object, if any +func removeObjectMetadataFile(baseDir, bucket, key string) { + os.Remove(objectMetadataPath(baseDir, bucket, key)) +} diff --git a/pkg/storage/multipart.go b/pkg/storage/multipart.go index fef7d7f..375e993 100644 --- a/pkg/storage/multipart.go +++ b/pkg/storage/multipart.go @@ -19,6 +19,8 @@ type MultipartUpload struct { UploadID string Bucket string Key string + ContentType string + UserMetadata map[string]string Initiated time.Time LastActivity time.Time Parts map[int]*UploadPart @@ -62,7 +64,7 @@ func NewMultipartManager(baseDir string) *MultipartManager { } // InitiateUpload starts a new multipart upload -func (m *MultipartManager) InitiateUpload(bucket, key string) (string, error) { +func (m *MultipartManager) InitiateUpload(bucket, key, contentType string, userMetadata map[string]string) (string, error) { m.mu.Lock() defer m.mu.Unlock() @@ -74,6 +76,8 @@ func (m *MultipartManager) InitiateUpload(bucket, key string) (string, error) { UploadID: uploadID, Bucket: bucket, Key: key, + ContentType: contentType, + UserMetadata: userMetadata, Initiated: now, LastActivity: now, Parts: make(map[int]*UploadPart), @@ -242,6 +246,16 @@ func (m *MultipartManager) CompleteUpload(uploadID string, parts []CompletePart) // Generate ETag in S3 multipart format: MD5-of-MD5s + part count etag := fmt.Sprintf("\"%s-%d\"", hex.EncodeToString(hash.Sum(nil)), len(parts)) + // Persist the object's metadata sidecar + meta := &objectMetadata{ + ETag: strings.Trim(etag, "\""), + ContentType: upload.ContentType, + UserMetadata: upload.UserMetadata, + } + if err := writeObjectMetadataFile(m.baseDir, upload.Bucket, upload.Key, meta); err != nil { + return "", fmt.Errorf("failed to write metadata: %w", err) + } + // Cleanup - remove from uploads map and delete parts directory m.mu.Lock() delete(m.uploads, uploadID) @@ -351,6 +365,8 @@ func (m *MultipartManager) saveUploadMetadata(upload *MultipartUpload) error { UploadID string Bucket string Key string + ContentType string + UserMetadata map[string]string Initiated time.Time LastActivity time.Time Parts map[int]*UploadPart @@ -358,6 +374,8 @@ func (m *MultipartManager) saveUploadMetadata(upload *MultipartUpload) error { UploadID: upload.UploadID, Bucket: upload.Bucket, Key: upload.Key, + ContentType: upload.ContentType, + UserMetadata: upload.UserMetadata, Initiated: upload.Initiated, LastActivity: upload.LastActivity, Parts: partsCopy, diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index b5997b8..a346449 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -7,6 +7,7 @@ import ( "io" "os" "path/filepath" + "sort" "strings" "time" ) @@ -18,6 +19,7 @@ type ObjectInfo struct { LastModified time.Time ETag string ContentType string + UserMetadata map[string]string } // Storage provides filesystem-based storage for S3 objects @@ -45,39 +47,58 @@ func New(baseDir string) (*Storage, error) { // PutObject stores an object func (s *Storage) PutObject(bucket, key string, reader io.Reader, size int64) error { + _, err := s.PutObjectWithMetadata(bucket, key, reader, size, "", nil) + return err +} + +// PutObjectWithMetadata stores an object along with its content type and user +// metadata, returning the quoted MD5 ETag of the content +func (s *Storage) PutObjectWithMetadata(bucket, key string, reader io.Reader, size int64, contentType string, userMetadata map[string]string) (string, error) { objectPath := s.objectPath(bucket, key) // Create parent directories if err := os.MkdirAll(filepath.Dir(objectPath), 0755); err != nil { - return fmt.Errorf("failed to create directory: %w", err) + return "", fmt.Errorf("failed to create directory: %w", err) } // Create temporary file tmpFile, err := os.CreateTemp(filepath.Dir(objectPath), ".s3dir-tmp-*") if err != nil { - return fmt.Errorf("failed to create temporary file: %w", err) + return "", fmt.Errorf("failed to create temporary file: %w", err) } tmpPath := tmpFile.Name() defer os.Remove(tmpPath) - // Copy data to temporary file using a fixed-size buffer to limit memory usage + // Copy data to temporary file using a fixed-size buffer to limit memory usage, + // calculating the content MD5 in the same pass // This ensures we stream data in 32KB chunks rather than allocating large buffers + hash := md5.New() buffer := make([]byte, 32*1024) // 32KB buffer - _, err = io.CopyBuffer(tmpFile, reader, buffer) + _, err = io.CopyBuffer(io.MultiWriter(tmpFile, hash), reader, buffer) closeErr := tmpFile.Close() if err != nil { - return fmt.Errorf("failed to write object: %w", err) + return "", fmt.Errorf("failed to write object: %w", err) } if closeErr != nil { - return fmt.Errorf("failed to close temporary file: %w", closeErr) + return "", fmt.Errorf("failed to close temporary file: %w", closeErr) } // Move temporary file to final location if err := os.Rename(tmpPath, objectPath); err != nil { - return fmt.Errorf("failed to move object: %w", err) + return "", fmt.Errorf("failed to move object: %w", err) } - return nil + etag := hex.EncodeToString(hash.Sum(nil)) + meta := &objectMetadata{ + ETag: etag, + ContentType: contentType, + UserMetadata: userMetadata, + } + if err := writeObjectMetadataFile(s.baseDir, bucket, key, meta); err != nil { + return "", fmt.Errorf("failed to write metadata: %w", err) + } + + return fmt.Sprintf("\"%s\"", etag), nil } // GetObject retrieves an object @@ -101,20 +122,91 @@ func (s *Storage) GetObject(bucket, key string) (io.ReadCloser, *ObjectInfo, err return nil, nil, fmt.Errorf("failed to open object: %w", err) } + return file, s.objectInfo(bucket, key, stat), nil +} + +// GetObjectRange retrieves a byte range of an object. start is the first byte +// offset and length the number of bytes to read +func (s *Storage) GetObjectRange(bucket, key string, start, length int64) (io.ReadCloser, *ObjectInfo, error) { + objectPath := s.objectPath(bucket, key) + + stat, err := os.Stat(objectPath) + if err != nil { + if os.IsNotExist(err) { + return nil, nil, fmt.Errorf("object not found") + } + return nil, nil, fmt.Errorf("failed to stat object: %w", err) + } + + if stat.IsDir() { + return nil, nil, fmt.Errorf("cannot get directory as object") + } + + file, err := os.Open(objectPath) + if err != nil { + return nil, nil, fmt.Errorf("failed to open object: %w", err) + } + + if _, err := file.Seek(start, io.SeekStart); err != nil { + file.Close() + return nil, nil, fmt.Errorf("failed to seek object: %w", err) + } + + reader := &rangeReadCloser{ + Reader: io.LimitReader(file, length), + file: file, + } + + return reader, s.objectInfo(bucket, key, stat), nil +} + +// rangeReadCloser wraps a limited reader over an open file so the file is +// closed when the caller finishes reading the range +type rangeReadCloser struct { + io.Reader + file *os.File +} + +func (r *rangeReadCloser) Close() error { + return r.file.Close() +} + +// objectInfo builds an ObjectInfo from a file stat plus the metadata sidecar, +// falling back to a modification-time ETag for objects without a sidecar +func (s *Storage) objectInfo(bucket, key string, stat os.FileInfo) *ObjectInfo { info := &ObjectInfo{ Key: key, Size: stat.Size(), LastModified: stat.ModTime(), - ETag: fmt.Sprintf("\"%x\"", stat.ModTime().Unix()), } - return file, info, nil + if meta := readObjectMetadataFile(s.baseDir, bucket, key); meta != nil { + if meta.ETag != "" { + info.ETag = fmt.Sprintf("\"%s\"", meta.ETag) + } + info.ContentType = meta.ContentType + info.UserMetadata = meta.UserMetadata + } + + if info.ETag == "" { + info.ETag = fmt.Sprintf("\"%x\"", stat.ModTime().Unix()) + } + + return info } // CopyObject copies an object server-side, streaming the data through a -// fixed-size buffer while calculating the MD5 of the content +// fixed-size buffer while calculating the MD5 of the content. The source +// object's content type and user metadata are carried over func (s *Storage) CopyObject(srcBucket, srcKey, dstBucket, dstKey string) (*ObjectInfo, error) { - reader, _, err := s.GetObject(srcBucket, srcKey) + return s.CopyObjectWithMetadata(srcBucket, srcKey, dstBucket, dstKey, false, "", nil) +} + +// CopyObjectWithMetadata copies an object server-side. When replaceMetadata is +// true the given content type and user metadata are stored on the destination +// instead of the source object's metadata +func (s *Storage) CopyObjectWithMetadata(srcBucket, srcKey, dstBucket, dstKey string, replaceMetadata bool, contentType string, userMetadata map[string]string) (*ObjectInfo, error) { + reader, srcInfo, err := s.GetObject(srcBucket, srcKey) if err != nil { return nil, err } @@ -157,11 +249,28 @@ func (s *Storage) CopyObject(srcBucket, srcKey, dstBucket, dstKey string) (*Obje return nil, fmt.Errorf("failed to stat object: %w", err) } + etag := hex.EncodeToString(hash.Sum(nil)) + meta := &objectMetadata{ + ETag: etag, + ContentType: contentType, + UserMetadata: userMetadata, + } + if !replaceMetadata { + // Carry over the source object's metadata (S3 COPY directive) + meta.ContentType = srcInfo.ContentType + meta.UserMetadata = srcInfo.UserMetadata + } + if err := writeObjectMetadataFile(s.baseDir, dstBucket, dstKey, meta); err != nil { + return nil, fmt.Errorf("failed to write metadata: %w", err) + } + return &ObjectInfo{ Key: dstKey, Size: written, LastModified: stat.ModTime(), - ETag: fmt.Sprintf("\"%s\"", hex.EncodeToString(hash.Sum(nil))), + ETag: fmt.Sprintf("\"%s\"", etag), + ContentType: meta.ContentType, + UserMetadata: meta.UserMetadata, }, nil } @@ -174,8 +283,12 @@ func (s *Storage) DeleteObject(bucket, key string) error { return fmt.Errorf("failed to delete object: %w", err) } + removeObjectMetadataFile(s.baseDir, bucket, key) + // Clean up empty parent directories s.cleanupEmptyDirs(filepath.Dir(objectPath), s.bucketPath(bucket)) + metadataBucketDir := filepath.Join(s.baseDir, metadataDirName, bucket) + s.cleanupEmptyDirs(filepath.Dir(objectMetadataPath(s.baseDir, bucket, key)), metadataBucketDir) return nil } @@ -196,28 +309,36 @@ func (s *Storage) HeadObject(bucket, key string) (*ObjectInfo, error) { return nil, fmt.Errorf("cannot head directory as object") } - info := &ObjectInfo{ - Key: key, - Size: stat.Size(), - LastModified: stat.ModTime(), - ETag: fmt.Sprintf("\"%x\"", stat.ModTime().Unix()), - } - - return info, nil + return s.objectInfo(bucket, key, stat), nil } // ListObjects lists objects in a bucket with optional prefix and delimiter func (s *Storage) ListObjects(bucket, prefix, delimiter string, maxKeys int) ([]ObjectInfo, []string, error) { + objects, commonPrefixes, _, _, err := s.ListObjectsPage(bucket, prefix, delimiter, "", maxKeys) + return objects, commonPrefixes, err +} + +// listEntry is a single result of a listing: either an object or a rolled-up +// common prefix +type listEntry struct { + name string + isPrefix bool + stat os.FileInfo +} + +// ListObjectsPage lists objects in a bucket in lexicographic key order, +// returning entries strictly after marker, up to maxKeys objects and common +// prefixes combined (maxKeys <= 0 means unlimited). It reports whether the +// listing was truncated and the marker to resume from +func (s *Storage) ListObjectsPage(bucket, prefix, delimiter, marker string, maxKeys int) ([]ObjectInfo, []string, bool, string, error) { bucketPath := s.bucketPath(bucket) if _, err := os.Stat(bucketPath); os.IsNotExist(err) { - return nil, nil, fmt.Errorf("bucket not found") + return nil, nil, false, "", fmt.Errorf("bucket not found") } - var objects []ObjectInfo - var commonPrefixes []string - prefixMap := make(map[string]bool) - + // Collect all keys matching the prefix + var keys []listEntry err := filepath.Walk(bucketPath, func(path string, info os.FileInfo, err error) error { if err != nil { return nil // Skip files we can't access @@ -236,56 +357,79 @@ func (s *Storage) ListObjects(bucket, prefix, delimiter string, maxKeys int) ([] // Convert to S3-style key (forward slashes) key := filepath.ToSlash(relPath) - // Apply prefix filter - if prefix != "" && !strings.HasPrefix(key, prefix) { - if !strings.HasPrefix(prefix, key+"/") { + if info.IsDir() { + // Skip subtrees that cannot contain keys with the prefix + if prefix != "" && !strings.HasPrefix(key+"/", prefix) && !strings.HasPrefix(prefix, key+"/") { return filepath.SkipDir } return nil } - // Handle delimiter + if prefix != "" && !strings.HasPrefix(key, prefix) { + return nil + } + + keys = append(keys, listEntry{name: key, stat: info}) + return nil + }) + if err != nil { + return nil, nil, false, "", fmt.Errorf("failed to list objects: %w", err) + } + + // S3 listings are in lexicographic key order; filesystem walk order is not + sort.Slice(keys, func(i, j int) bool { + return keys[i].name < keys[j].name + }) + + // Roll up keys containing the delimiter into common prefixes. Keys sharing + // a common prefix are contiguous in sorted order, so deduplicating against + // the previous entry is sufficient + var entries []listEntry + for _, k := range keys { if delimiter != "" { - remainder := strings.TrimPrefix(key, prefix) + remainder := strings.TrimPrefix(k.name, prefix) if idx := strings.Index(remainder, delimiter); idx != -1 { commonPrefix := prefix + remainder[:idx+len(delimiter)] - if !prefixMap[commonPrefix] { - prefixMap[commonPrefix] = true - commonPrefixes = append(commonPrefixes, commonPrefix) - } - if info.IsDir() { - return filepath.SkipDir + if len(entries) > 0 && entries[len(entries)-1].isPrefix && entries[len(entries)-1].name == commonPrefix { + continue } - return nil + entries = append(entries, listEntry{name: commonPrefix, isPrefix: true}) + continue } } + entries = append(entries, k) + } - // Skip directories - if info.IsDir() { - return nil - } - - // Add object - objects = append(objects, ObjectInfo{ - Key: key, - Size: info.Size(), - LastModified: info.ModTime(), - ETag: fmt.Sprintf("\"%x\"", info.ModTime().Unix()), + // Resume strictly after the marker + if marker != "" { + start := sort.Search(len(entries), func(i int) bool { + return entries[i].name > marker }) + entries = entries[start:] + } - // Check max keys limit - if maxKeys > 0 && len(objects) >= maxKeys { - return filepath.SkipAll - } + truncated := maxKeys > 0 && len(entries) > maxKeys + if truncated { + entries = entries[:maxKeys] + } - return nil - }) + nextMarker := "" + if truncated { + nextMarker = entries[len(entries)-1].name + } - if err != nil && err != filepath.SkipAll { - return nil, nil, fmt.Errorf("failed to list objects: %w", err) + // Build results, reading metadata sidecars only for the returned page + var objects []ObjectInfo + var commonPrefixes []string + for _, e := range entries { + if e.isPrefix { + commonPrefixes = append(commonPrefixes, e.name) + } else { + objects = append(objects, *s.objectInfo(bucket, e.name, e.stat)) + } } - return objects, commonPrefixes, nil + return objects, commonPrefixes, truncated, nextMarker, nil } // CreateBucket creates a new bucket (directory) @@ -323,6 +467,9 @@ func (s *Storage) DeleteBucket(bucket string) error { return fmt.Errorf("failed to delete bucket: %w", err) } + // Remove any leftover metadata sidecars for the bucket + os.RemoveAll(filepath.Join(s.baseDir, metadataDirName, bucket)) + return nil } @@ -354,7 +501,8 @@ func (s *Storage) ListBuckets() ([]string, error) { var buckets []string for _, entry := range entries { - if entry.IsDir() { + // Skip internal directories (.multipart, .metadata) + if entry.IsDir() && !strings.HasPrefix(entry.Name(), ".") { buckets = append(buckets, entry.Name()) } } @@ -388,11 +536,17 @@ func (s *Storage) cleanupEmptyDirs(path, stopPath string) { // InitiateMultipartUpload starts a new multipart upload func (s *Storage) InitiateMultipartUpload(bucket, key string) (string, error) { + return s.InitiateMultipartUploadWithMetadata(bucket, key, "", nil) +} + +// InitiateMultipartUploadWithMetadata starts a new multipart upload, recording +// the content type and user metadata to store on the completed object +func (s *Storage) InitiateMultipartUploadWithMetadata(bucket, key, contentType string, userMetadata map[string]string) (string, error) { // Verify bucket exists if err := s.HeadBucket(bucket); err != nil { return "", err } - return s.multipart.InitiateUpload(bucket, key) + return s.multipart.InitiateUpload(bucket, key, contentType, userMetadata) } // UploadPart uploads a part of a multipart upload