diff --git a/test/s3tables/table-buckets/verify_object_test.go b/test/s3tables/table-buckets/verify_object_test.go new file mode 100644 index 0000000000..cf8bd2394d --- /dev/null +++ b/test/s3tables/table-buckets/verify_object_test.go @@ -0,0 +1,139 @@ +package s3tables + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/smithy-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func newObjectClient(t *testing.T) *s3.Client { + t.Helper() + cfg, err := config.LoadDefaultConfig(context.Background(), + config.WithRegion(testRegion), + config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(testAccessKey, testSecretKey, "")), + ) + require.NoError(t, err) + return s3.NewFromConfig(cfg, func(o *s3.Options) { + o.BaseEndpoint = aws.String(sharedCluster.s3Endpoint) + o.UsePathStyle = true + }) +} + +func apiErrorCode(err error) string { + var apiErr smithy.APIError + if errors.As(err, &apiErr) { + return apiErr.ErrorCode() + } + return "" +} + +// The external-volume verification object round-trips at a table bucket's +// root the way the check uses it: PUT, HEAD, GET, list, DELETE. Any other key +// at the root is still refused. +func TestTableBucketExternalVolumeVerificationRoundTrip(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + ctx := context.Background() + tables := NewS3TablesClient(sharedCluster.s3Endpoint, testRegion, testAccessKey, testSecretKey) + bucket := "verify-" + randomString(8) + created, err := tables.CreateTableBucket(bucket, nil) + require.NoError(t, err) + defer tables.DeleteTableBucket(created.ARN) + client := newObjectClient(t) + + body := []byte("verification payload!") // 21 bytes + key := fmt.Sprintf("verify_%d_%d", time.Now().UnixMilli(), 1081846483) + + // the control: an ordinary key at the root is refused + _, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String("notes.parquet"), Body: bytes.NewReader(body)}) + require.Error(t, err) + assert.Equal(t, "AccessDenied", apiErrorCode(err)) + + _, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(key), Body: bytes.NewReader(body), ContentLength: aws.Int64(int64(len(body)))}) + require.NoError(t, err) + + head, err := client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + require.NoError(t, err) + assert.Equal(t, int64(len(body)), aws.ToInt64(head.ContentLength)) + + got, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + require.NoError(t, err) + data, err := io.ReadAll(got.Body) + got.Body.Close() + require.NoError(t, err) + assert.Equal(t, body, data) + + list, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket), Prefix: aws.String("verify_")}) + require.NoError(t, err) + var keys []string + for _, o := range list.Contents { + keys = append(keys, aws.ToString(o.Key)) + } + assert.Equal(t, []string{key}, keys) + + _, err = client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + require.NoError(t, err) + _, err = client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(bucket), Key: aws.String(key)}) + require.Error(t, err, "the verification object must be gone after DELETE") +} + +// A namespace can carry a name of the verification key's shape. The key +// then names the namespace's directory: a PUT or DELETE through it must not +// replace or remove the namespace. +func TestTableBucketVerificationKeyLeavesNamespaceAlone(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + ctx := context.Background() + tables := NewS3TablesClient(sharedCluster.s3Endpoint, testRegion, testAccessKey, testSecretKey) + bucket := "verify-ns-" + randomString(8) + created, err := tables.CreateTableBucket(bucket, nil) + require.NoError(t, err) + defer tables.DeleteTableBucket(created.ARN) + client := newObjectClient(t) + body := []byte("verification payload!") + + listed := func(name string) bool { + resp, err := tables.ListNamespaces(created.ARN, "", "", 0) + require.NoError(t, err) + for _, ns := range resp.Namespaces { + if len(ns.Namespace) == 1 && ns.Namespace[0] == name { + return true + } + } + return false + } + for i, op := range []string{"PUT", "DELETE"} { + t.Run(op, func(t *testing.T) { + name := fmt.Sprintf("verify_%d_%d", time.Now().UnixMilli(), 7+i) + _, err := tables.CreateNamespace(created.ARN, []string{name}) + require.NoError(t, err) + defer tables.DeleteNamespace(created.ARN, []string{name}) + require.True(t, listed(name), "setup: the namespace is listed") + + if op == "PUT" { + _, err = client.PutObject(ctx, &s3.PutObjectInput{Bucket: aws.String(bucket), Key: aws.String(name), Body: bytes.NewReader(body), ContentLength: aws.Int64(int64(len(body)))}) + assert.Error(t, err, "a PUT over a namespace must be refused") + } else { + _, err = client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(bucket), Key: aws.String(name)}) + assert.Error(t, err, "a DELETE of a namespace must be refused") + } + _, err = tables.GetNamespace(created.ARN, []string{name}) + assert.NoError(t, err, "the namespace survives the %s", op) + assert.True(t, listed(name), "the namespace is still listed after the %s", op) + }) + } +} diff --git a/weed/s3api/bucket_paths.go b/weed/s3api/bucket_paths.go index c143e213ea..2d70a03826 100644 --- a/weed/s3api/bucket_paths.go +++ b/weed/s3api/bucket_paths.go @@ -2,8 +2,12 @@ package s3api import ( "errors" + "net/http" "path" + "regexp" + "strconv" "strings" + "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -83,6 +87,117 @@ func (s3a *S3ApiServer) validateTableBucketObjectPath(bucket, object string) err return nil } +// volumeVerifyKey matches the object Snowflake writes, reads, lists and +// deletes at the base of an S3-compatible external volume before it creates +// or writes Iceberg tables there (SYSTEM$VERIFY_EXTERNAL_VOLUME, also run +// implicitly): /verify__, a 21-byte body. +// A table bucket refuses every key outside the Iceberg layout, so without +// this exception Snowflake cannot use a table bucket as an external volume. +// +// The exception is narrow so that it cannot be used to store data: +// - only at the bucket root: nothing can be encoded in a prefix, written +// into a hidden tree or take a namespace or table name; +// - one shape: 13 digits of epoch milliseconds, then 1-10 digits; +// - never a directory: a namespace may carry a name of that shape, and the +// key then names its directory (volumeVerifyKeyIsDirectory); +// - a PUT must be dated within volumeVerifyWindow of the server's clock, +// declare a Content-Length of at most volumeVerifyMaxBytes (no chunked or +// undeclared body), and carry no user metadata, tagging, stored content +// headers or extraction request (volumeVerifyPutAllowed). Reads and +// deletes take a key of any date, so an object a check left behind can +// still be removed. +// +// Only the calls the check makes (PUT, GET, HEAD, DELETE, DeleteObjects) +// accept the key; multipart, copy, rename, ACL, tagging and POST policy keep +// the plain layout rule. +var volumeVerifyKey = regexp.MustCompile(`^verify_([0-9]{13})_[0-9]{1,10}$`) + +const ( + volumeVerifyMaxBytes = 1024 + volumeVerifyWindow = time.Hour +) + +// volumeVerifyNow is the clock a PUT's key is checked against. +var volumeVerifyNow = time.Now + +// volumeVerifyKeyTime is the time a verification key is dated, and whether +// the object has the key's shape. +func volumeVerifyKeyTime(object string) (time.Time, bool) { + m := volumeVerifyKey.FindStringSubmatch(strings.TrimPrefix(object, "/")) + if m == nil { + return time.Time{}, false + } + ms, err := strconv.ParseInt(m[1], 10, 64) + if err != nil { + return time.Time{}, false + } + return time.UnixMilli(ms), true +} + +func isVolumeVerifyKey(object string) bool { + _, ok := volumeVerifyKeyTime(object) + return ok +} + +// volumeVerifyLookup reads the entry a verification key names. +var volumeVerifyLookup = func(s3a *S3ApiServer, dir, name string) (*filer_pb.Entry, error) { + return s3a.getEntry(dir, name) +} + +// volumeVerifyKeyIsDirectory reports whether the key names a directory (a +// namespace), or whether that cannot be known. +func (s3a *S3ApiServer) volumeVerifyKeyIsDirectory(bucket, object string) bool { + entry, err := volumeVerifyLookup(s3a, s3a.bucketDir(bucket), strings.TrimPrefix(object, "/")) + if err != nil { + return !errors.Is(err, filer_pb.ErrNotFound) + } + return entry != nil && entry.IsDirectory +} + +// validateTableBucketObjectPathOrVerify is validateTableBucketObjectPath that +// also accepts the external-volume verification key, and reports it so the +// PUT handler can apply its limits. +func (s3a *S3ApiServer) validateTableBucketObjectPathOrVerify(bucket, object string) (isVerify bool, err error) { + if !s3a.isTableBucket(bucket) { + return false, nil + } + if isVolumeVerifyKey(object) && !s3a.volumeVerifyKeyIsDirectory(bucket, object) { + return true, nil + } + return false, s3a.validateTableBucketObjectPath(bucket, object) +} + +// volumeVerifyPutAllowed reports whether a PUT to the verification key is +// dated now, declares a small body and stores nothing beside it. +func volumeVerifyPutAllowed(r *http.Request, object string) bool { + at, ok := volumeVerifyKeyTime(object) + if !ok { + return false + } + if d := volumeVerifyNow().Sub(at); d > volumeVerifyWindow || d < -volumeVerifyWindow { + return false + } + if r.ContentLength < 0 || r.ContentLength > volumeVerifyMaxBytes || len(r.TransferEncoding) > 0 { + return false + } + if isExtractZip(r) { + return false + } + for name, values := range r.Header { + n := strings.ToLower(name) + if strings.HasPrefix(n, "x-amz-meta-") || n == "x-amz-tagging" || n == "content-disposition" || + n == "response-content-disposition" || n == "content-language" || + n == "cache-control" || n == "expires" || n == "x-amz-website-redirect-location" { + return false + } + // aws-chunked only frames the body; any other coding is stored. + if n == "content-encoding" && storedContentEncoding(values) != "" { + return false + } + } + return true +} + func (s3a *S3ApiServer) bucketPrefix(bucket string) string { return s3a.bucketDir(bucket) + "/" } diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 96d3dcade8..8e747550b7 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -804,7 +804,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) bucket, object := s3_constants.GetBucketAndObject(r) glog.V(3).Infof("GetObjectHandler %s %s", bucket, object) - if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil { + if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } @@ -2499,7 +2499,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request bucket, object := s3_constants.GetBucketAndObject(r) glog.V(3).Infof("HeadObjectHandler %s %s", bucket, object) - if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil { + if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index 4e28d9c5c5..3817a28adc 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/weed/s3api/s3api_object_handlers_delete.go @@ -210,7 +210,7 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque // The filer ops below each retry, and a failover walk runs the whole set // once per filer, so the backoff comes out of one allowance held here. r = r.WithContext(withFilerRetryBudget(r.Context(), filerRetryRequestBudget)) - if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil { + if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil { s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } @@ -479,7 +479,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h deleteErrors = append(deleteErrors, deleteErrorFromCode(validationCode, object.Key, object.VersionId)) continue } - if err := s3a.validateTableBucketObjectPath(bucket, object.Key); err != nil { + if _, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object.Key); err != nil { deleteErrors = append(deleteErrors, deleteErrorFromCode(s3err.ErrAccessDenied, object.Key, object.VersionId)) continue } diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 3b895f9260..1b18f0c560 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -112,7 +112,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) s3err.WriteErrorResponse(w, r, s3err.ErrKeyTooLongError) return } - if err := s3a.validateTableBucketObjectPath(bucket, object); err != nil { + if isVerify, err := s3a.validateTableBucketObjectPathOrVerify(bucket, object); err != nil || (isVerify && !volumeVerifyPutAllowed(r, object)) { s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } @@ -991,7 +991,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } glog.V(3).Infof("putToFiler: Calling CreateEntry for %s", filePath) err := filer_pb.CreateEntry(context.Background(), client, req) - if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) { + // A verification key in a table bucket never becomes a prefix + // object: the directory it meets is a namespace, created after the + // handler looked. + if errors.Is(err, filer_pb.ErrExistingIsDirectory) && !isReservedDirectoryName(entry.Name) && + !(isVolumeVerifyKey(object) && s3a.isTableBucket(bucket)) { // Other keys are nested under this one. S3 keys are flat, so the key is // stored on the directory they live under rather than refused. entry.MarkPrefixObject() diff --git a/weed/s3api/table_bucket_verify_object_test.go b/weed/s3api/table_bucket_verify_object_test.go new file mode 100644 index 0000000000..789d51f377 --- /dev/null +++ b/weed/s3api/table_bucket_verify_object_test.go @@ -0,0 +1,163 @@ +package s3api + +import ( + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/stretchr/testify/assert" +) + +const verifyTestBucket = "tables" + +// verifyTestNow is the server clock the tests pin; keys are dated around it. +const verifyTestNow = 1790000000000 + +func newVerifyTestServer(t *testing.T) *S3ApiServer { + t.Helper() + saved := volumeVerifyNow + volumeVerifyNow = func() time.Time { return time.UnixMilli(verifyTestNow) } + savedLookup := volumeVerifyLookup + volumeVerifyLookup = func(*S3ApiServer, string, string) (*filer_pb.Entry, error) { return nil, filer_pb.ErrNotFound } + t.Cleanup(func() { volumeVerifyNow, volumeVerifyLookup = saved, savedLookup }) + s3a := &S3ApiServer{option: &S3ApiServerOption{BucketsPath: "/buckets"}} + s3a.bucketRegistry = NewBucketRegistry(s3a) + s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: verifyTestBucket, IsTableBucket: true}) + s3a.bucketRegistry.setMetadataCache(&BucketMetaData{Name: "plain", IsTableBucket: false}) + return s3a +} + +func TestTableBucketAcceptsExternalVolumeVerifyKey(t *testing.T) { + s3a := newVerifyTestServer(t) + for _, key := range []string{ + "verify_1790000000000_1081846483", + "/verify_1790000000000_1", + "verify_1790003300000_42", // 55 minutes later: inside the window + "verify_1789996700000_42", // 55 minutes earlier + // reads and deletes take a key of any date (PUT limits it) + "verify_1789992000000_1", + "verify_1790008000000_1", + } { + isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key) + assert.True(t, isVerify, key) + assert.NoError(t, err, key) + // the handlers that keep the plain rule still refuse it + assert.Error(t, s3a.validateTableBucketObjectPath(verifyTestBucket, key), key) + } + // the Iceberg layout is unchanged: the positive control + isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, "ns/t/data/part-00000.parquet") + assert.False(t, isVerify) + assert.NoError(t, err) +} + +func TestTableBucketRefusesVerifyKeyLookalikes(t *testing.T) { + s3a := newVerifyTestServer(t) + for _, key := range []string{ + // the bucket root only + "base/verify_1790000000000_1", + "ns/verify_1790000000000_1", + ".uploads/verify_1790000000000_1", + ".uploads/abc/verify_1790000000000_1", + // the exact shape only + "verify_1790000000000_1.txt", + "verify_1790000000000", + "verify_179000000000_1", // 12-digit timestamp + "verify_1790000000000_12345678901", // 11-digit number + "verify_abc_123", + "verify_1790000000000_1/x", + "xverify_1790000000000_1", + // ordinary non-Iceberg keys + "notes.txt", + "ns/t/data/notes.txt", + } { + isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key) + assert.False(t, isVerify, key) + assert.Error(t, err, key) + } + // a bucket that is not a table bucket is unaffected + isVerify, err := s3a.validateTableBucketObjectPathOrVerify("plain", "verify_1790000000000_1") + assert.False(t, isVerify) + assert.NoError(t, err) +} + +func verifyPutRequest(key string) *http.Request { + req := httptest.NewRequest(http.MethodPut, "/"+verifyTestBucket+"/"+key, strings.NewReader("x")) + return mux.SetURLVars(req, map[string]string{"bucket": verifyTestBucket, "object": key}) +} + +// A PUT to the verification key cannot store data: a small declared body and +// nothing stored beside it. +func TestTableBucketVerifyKeyPutLimits(t *testing.T) { + s3a := newVerifyTestServer(t) + const key = "verify_1790000000000_1081846483" + for name, mutate := range map[string]func(*http.Request){ + "too large": func(r *http.Request) { r.ContentLength = volumeVerifyMaxBytes + 1 }, + "content-language": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Content-Language", "fr") }, + "content-encoding": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Content-Encoding", "gzip") }, + "response-content-disposition": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Response-Content-Disposition", "attachment") }, + "extract query": func(r *http.Request) { r.ContentLength = 0; r.URL.RawQuery = "extract=zip" }, + "extract header": func(r *http.Request) { r.ContentLength = 0; r.Header.Set(s3_constants.SeaweedFSExtract, "zip") }, + "undeclared length": func(r *http.Request) { r.ContentLength = -1 }, + "chunked": func(r *http.Request) { r.ContentLength = 0; r.TransferEncoding = []string{"chunked"} }, + "user metadata": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("X-Amz-Meta-Note", "x") }, + "tagging": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("X-Amz-Tagging", "a=b") }, + "content-disposition": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("Content-Disposition", "attachment") }, + "website redirect": func(r *http.Request) { r.ContentLength = 0; r.Header.Set("X-Amz-Website-Redirect-Location", "/x") }, + } { + req := verifyPutRequest(key) + mutate(req) + assert.False(t, volumeVerifyPutAllowed(req, key), name) + rec := httptest.NewRecorder() + s3a.PutObjectHandler(rec, req) + assert.Equal(t, http.StatusForbidden, rec.Code, name) + } + ok := verifyPutRequest(key) + ok.ContentLength = 21 + ok.Header.Set("Content-Type", "application/octet-stream") + ok.Header.Set("Content-Encoding", "aws-chunked") + assert.True(t, volumeVerifyPutAllowed(ok, key), "the verification object itself") + + // a PUT is dated within the window of the server's clock + for _, old := range []string{"verify_1789992000000_1", "verify_1790008000000_1"} { + req := verifyPutRequest(old) + req.ContentLength = 21 + assert.False(t, volumeVerifyPutAllowed(req, old), old) + rec := httptest.NewRecorder() + s3a.PutObjectHandler(rec, req) + assert.Equal(t, http.StatusForbidden, rec.Code, old) + } +} + +// A namespace may carry a name of the key's shape; the key then names its +// directory and gets the plain layout rule, as does a key whose entry cannot +// be read. +func TestTableBucketVerifyKeyNamingDirectoryIsRefused(t *testing.T) { + s3a := newVerifyTestServer(t) + const key = "verify_1790000000000_7" + for name, lookup := range map[string]func(*S3ApiServer, string, string) (*filer_pb.Entry, error){ + "namespace directory": func(*S3ApiServer, string, string) (*filer_pb.Entry, error) { + return &filer_pb.Entry{Name: key, IsDirectory: true}, nil + }, + "lookup failure": func(*S3ApiServer, string, string) (*filer_pb.Entry, error) { + return nil, errors.New("filer unavailable") + }, + } { + volumeVerifyLookup = lookup + isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key) + assert.False(t, isVerify, name) + assert.Error(t, err, name) + } + // the control: an existing verification object is a file + volumeVerifyLookup = func(*S3ApiServer, string, string) (*filer_pb.Entry, error) { + return &filer_pb.Entry{Name: key}, nil + } + isVerify, err := s3a.validateTableBucketObjectPathOrVerify(verifyTestBucket, key) + assert.True(t, isVerify) + assert.NoError(t, err) +}