Merge pull request #2541 from versity/ben/skipdirs-fixes

fix: skip excluded directories and reserve their keys

backend/posix/list_objects_snapdir_test.go

@@ -16,6 +16,7 @@ package posix
 
 import (
 	"context"
+	"errors"
 	"os"
 	"path/filepath"
 	"sort"
@@ -25,7 +26,10 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/versity/versitygw/backend/meta"
+	"github.com/versity/versitygw/s3err"
+	"github.com/versity/versitygw/s3response"
 )
 
 // zfsSnapDir is the control directory a ZFS dataset exposes when it is mounted
@@ -73,12 +77,21 @@ func newSnapdirBackend(t *testing.T, storer meta.MetadataStorer, excludeDirs []s
 
 	createTestBucket(t, p, snapBucket)
 
-	bdir := p.BucketPath(snapBucket)
-	for name, data := range map[string]string{
+	writeBucketFiles(t, p, map[string]string{
 		"top.txt":                          "top",
 		"child/a.txt":                      "a",
 		"child/.zfs/snapshot/one/snap.txt": "snap",
-	} {
+	})
+	return p
+}
+
+// writeBucketFiles writes files, keyed by their slash-separated path in the
+// bucket, straight to the bucket directory.
+func writeBucketFiles(t *testing.T, p *Posix, files map[string]string) {
+	t.Helper()
+
+	bdir := p.BucketPath(snapBucket)
+	for name, data := range files {
 		full := filepath.Join(bdir, filepath.FromSlash(name))
 		if err := os.MkdirAll(filepath.Dir(full), 0755); err != nil {
 			t.Fatalf("mkdir %q: %v", name, err)
@@ -87,7 +100,6 @@ func newSnapdirBackend(t *testing.T, storer meta.MetadataStorer, excludeDirs []s
 			t.Fatalf("write %q: %v", name, err)
 		}
 	}
-	return p
 }
 
 func listObjectsKeys(p *Posix) ([]string, error) {
@@ -132,6 +144,28 @@ func listObjectVersionsKeys(p *Posix) ([]string, error) {
 	return keys, nil
 }
 
+type listOp struct {
+	name string
+	list func(*Posix) ([]string, error)
+}
+
+// listOps are the three listing operations that share the skip list.
+var listOps = []listOp{
+	{"ListObjects", listObjectsKeys},
+	{"ListObjectsV2", listObjectsV2Keys},
+	{"ListObjectVersions", listObjectVersionsKeys},
+}
+
+type metaMode struct {
+	name   string
+	storer func() meta.MetadataStorer
+}
+
+var snapdirMetaModes = []metaMode{
+	{"xattr", func() meta.MetadataStorer { return meta.XattrMeta{} }},
+	{"unsupported .zfs xattr", func() meta.MetadataStorer { return zfsSnapdirMeta{} }},
+}
+
 // TestListObjectsSkipsZFSSnapDir checks that a control directory named in
 // PosixOpts.ExcludeDirs is not listed as bucket content and is not walked.
 // The directory shows up in readdir and its metadata reads fail, which used to
@@ -139,25 +173,18 @@ func listObjectVersionsKeys(p *Posix) ([]string, error) {
 // not be walked either, or the listing would report the snapshot contents as
 // objects. All three listing operations share the same list.
 func TestListObjectsSkipsZFSSnapDir(t *testing.T) {
-	for mode, mkMeta := range map[string]func() meta.MetadataStorer{
-		"xattr":                  func() meta.MetadataStorer { return meta.XattrMeta{} },
-		"unsupported .zfs xattr": func() meta.MetadataStorer { return zfsSnapdirMeta{} },
-	} {
-		t.Run(mode, func(t *testing.T) {
-			p := newSnapdirBackend(t, mkMeta(), []string{zfsSnapDir})
+	for _, mode := range snapdirMetaModes {
+		t.Run(mode.name, func(t *testing.T) {
+			p := newSnapdirBackend(t, mode.storer(), []string{zfsSnapDir})
 
 			want := []string{"child/a.txt", "top.txt"}
-			for name, list := range map[string]func() ([]string, error){
-				"ListObjects":        func() ([]string, error) { return listObjectsKeys(p) },
-				"ListObjectsV2":      func() ([]string, error) { return listObjectsV2Keys(p) },
-				"ListObjectVersions": func() ([]string, error) { return listObjectVersionsKeys(p) },
-			} {
-				keys, err := list()
+			for _, op := range listOps {
+				keys, err := op.list(p)
 				if err != nil {
-					t.Fatalf("%s: %v", name, err)
+					t.Fatalf("%s: %v", op.name, err)
 				}
 				if !equalKeys(keys, want) {
-					t.Fatalf("%s: keys = %v, want %v", name, keys, want)
+					t.Fatalf("%s: keys = %v, want %v", op.name, keys, want)
 				}
 			}
 		})
@@ -172,43 +199,239 @@ func TestListObjectsExcludeDirsEmpty(t *testing.T) {
 	t.Run("xattr", func(t *testing.T) {
 		p := newSnapdirBackend(t, meta.XattrMeta{}, nil)
 
-		keys, err := listObjectsKeys(p)
-		if err != nil {
-			t.Fatalf("list objects: %v", err)
-		}
-		if !containsKey(keys, "child/.zfs/snapshot/one/snap.txt") {
-			t.Fatalf("keys = %v, want the .zfs snapshot object to be listed", keys)
+		for _, op := range listOps {
+			keys, err := op.list(p)
+			if err != nil {
+				t.Fatalf("%s: %v", op.name, err)
+			}
+			if !containsKey(keys, "child/.zfs/snapshot/one/snap.txt") {
+				t.Fatalf("%s: keys = %v, want the .zfs snapshot object to be listed", op.name, keys)
+			}
 		}
 	})
 
 	t.Run("unsupported .zfs xattr", func(t *testing.T) {
 		p := newSnapdirBackend(t, zfsSnapdirMeta{}, nil)
 
-		if _, err := listObjectsKeys(p); err == nil {
-			t.Fatal("list objects succeeded, want the .zfs metadata error")
+		for _, op := range listOps {
+			if _, err := op.list(p); err == nil {
+				t.Fatalf("%s succeeded, want the .zfs metadata error", op.name)
+			}
 		}
 	})
 }
 
-// TestListingSkipDirs checks the list the walks receive: the metadata
-// directory is always skipped, and an excluded directory that could never
-// match a directory entry name is rejected instead of silently doing nothing.
+// TestListObjectsExcludeDirsFileNotSkipped checks that only directories are
+// excluded. A regular file with an excluded name is an ordinary object: it is
+// listed, and it does not hide the objects that sort after it in the same
+// directory, which ListObjectVersions used to drop.
+func TestListObjectsExcludeDirsFileNotSkipped(t *testing.T) {
+	p := newSnapdirBackend(t, meta.XattrMeta{}, []string{zfsSnapDir})
+	writeBucketFiles(t, p, map[string]string{
+		"d/" + zfsSnapDir: "file",
+		"d/b.txt":         "b",
+		"d/c.txt":         "c",
+	})
+
+	want := []string{"child/a.txt", "d/" + zfsSnapDir, "d/b.txt", "d/c.txt", "top.txt"}
+	for _, op := range listOps {
+		keys, err := op.list(p)
+		if err != nil {
+			t.Fatalf("%s: %v", op.name, err)
+		}
+		if !equalKeys(keys, want) {
+			t.Errorf("%s: keys = %v, want %v", op.name, keys, want)
+		}
+	}
+}
+
+// TestListingSkipDirs checks the directories the walks receive: the metadata
+// directory is skipped at the bucket root only, the excluded directories at
+// any depth, and an excluded directory that could never match a directory
+// entry name is rejected instead of silently doing nothing.
 func TestListingSkipDirs(t *testing.T) {
 	got, err := listingSkipDirs([]string{zfsSnapDir})
 	if err != nil {
 		t.Fatalf("listing skip dirs: %v", err)
 	}
-	if want := []string{MetaTmpDir, zfsSnapDir}; !equalKeys(got, want) {
-		t.Fatalf("skip dirs = %v, want %v", got, want)
+	if want := []string{MetaTmpDir}; !equalKeys(got.Root, want) {
+		t.Errorf("root skip dirs = %v, want %v", got.Root, want)
+	}
+	if want := []string{zfsSnapDir}; !equalKeys(got.Any, want) {
+		t.Errorf("any-depth skip dirs = %v, want %v", got.Any, want)
 	}
 
-	for _, dir := range []string{"", "child/.zfs"} {
+	for _, dir := range []string{"", ".", "..", "child/.zfs"} {
 		if _, err := listingSkipDirs([]string{dir}); err == nil {
 			t.Errorf("listingSkipDirs(%q) succeeded, want an error", dir)
 		}
 	}
 }
 
+// TestListObjectsNestedMetaTmpDir checks that only the metadata directory at
+// the bucket root is hidden. The gateway never creates one further down, so a
+// nested .sgwtmp directory holds objects a client stored there, which object
+// requests accept and so the listings must report.
+func TestListObjectsNestedMetaTmpDir(t *testing.T) {
+	ctx := context.Background()
+	bucket := snapBucket
+	key := "d/" + MetaTmpDir + "/x"
+
+	p := newSnapdirBackend(t, meta.XattrMeta{}, []string{zfsSnapDir})
+	if err := putTestObject(t, p, bucket, key, "data"); err != nil {
+		t.Fatalf("PutObject %q: %v", key, err)
+	}
+	if _, err := p.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &key}); err != nil {
+		t.Fatalf("HeadObject %q: %v", key, err)
+	}
+	writeBucketFiles(t, p, map[string]string{MetaTmpDir + "/stray": "tmp"})
+
+	want := []string{"child/a.txt", key, "top.txt"}
+	for _, op := range listOps {
+		keys, err := op.list(p)
+		if err != nil {
+			t.Fatalf("%s: %v", op.name, err)
+		}
+		if !equalKeys(keys, want) {
+			t.Errorf("%s: keys = %v, want %v", op.name, keys, want)
+		}
+	}
+}
+
+// TestExcludeDirsCaseAgrees checks that the listings and object requests
+// match an excluded name the same way, ignoring case: a directory spelled in
+// another case is neither listed nor reachable, rather than listed with
+// objects that can't be read.
+func TestExcludeDirsCaseAgrees(t *testing.T) {
+	ctx := context.Background()
+	bucket := snapBucket
+	key := "e/.ZFS/x"
+
+	p := newSnapdirBackend(t, meta.XattrMeta{}, []string{zfsSnapDir})
+	writeBucketFiles(t, p, map[string]string{key: "x"})
+
+	want := []string{"child/a.txt", "top.txt"}
+	for _, op := range listOps {
+		keys, err := op.list(p)
+		if err != nil {
+			t.Fatalf("%s: %v", op.name, err)
+		}
+		if !equalKeys(keys, want) {
+			t.Errorf("%s: keys = %v, want %v", op.name, keys, want)
+		}
+	}
+
+	_, err := p.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &key})
+	if !errors.Is(err, s3err.GetAPIError(s3err.ErrExcludedObjectKey)) {
+		t.Errorf("HeadObject %q: got %v, want ReservedObjectKey", key, err)
+	}
+}
+
+// TestExcludedDirKeysReserved checks that object requests can't reach a path
+// under an excluded directory. The listings never show objects there, and on
+// ZFS the .zfs directory, reachable by name even with snapdir=hidden, would
+// serve snapshot contents under keys that policies on the live keys don't
+// cover.
+func TestExcludedDirKeysReserved(t *testing.T) {
+	ctx := context.Background()
+	reserved := s3err.GetAPIError(s3err.ErrExcludedObjectKey)
+	bucket := snapBucket
+	snapKey := "child/.zfs/snapshot/one/snap.txt"
+
+	p := newSnapdirBackend(t, meta.XattrMeta{}, []string{zfsSnapDir})
+	snapPath := filepath.Join(p.BucketPath(bucket), filepath.FromSlash(snapKey))
+
+	for _, key := range []string{
+		snapKey,
+		"child/.zfs/",
+		"child/.ZFS/snapshot/one/snap.txt",
+		"child/x/../.zfs/snapshot/one/snap.txt",
+		"child//.zfs/snapshot/one/snap.txt",
+		".zfs/a",
+		"child/a.txt/../.zfs/snapshot/one/snap.txt",
+	} {
+		_, err := p.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: aws.String(key)})
+		if !errors.Is(err, reserved) {
+			t.Errorf("HeadObject %q: got %v, want ReservedObjectKey", key, err)
+		}
+	}
+
+	if _, err := p.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &snapKey}); !errors.Is(err, reserved) {
+		t.Errorf("GetObject: got %v, want ReservedObjectKey", err)
+	}
+
+	for _, key := range []string{"child/.zfs/new.txt", "child/.zfs/snapshot/two/", "child/.zfs/"} {
+		if err := putTestObject(t, p, bucket, key, ""); !errors.Is(err, reserved) {
+			t.Errorf("PutObject %q: got %v, want ReservedObjectKey", key, err)
+		}
+	}
+	if _, err := os.Stat(filepath.Join(p.BucketPath(bucket), "child", zfsSnapDir, "new.txt")); !os.IsNotExist(err) {
+		t.Errorf("PutObject created an object under the excluded directory: %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(p.BucketPath(bucket), "child", zfsSnapDir, "snapshot", "two")); !os.IsNotExist(err) {
+		t.Errorf("PutObject created a directory under the excluded directory: %v", err)
+	}
+
+	if _, err := p.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: &bucket, Key: &snapKey}); !errors.Is(err, reserved) {
+		t.Errorf("DeleteObject: got %v, want ReservedObjectKey", err)
+	}
+	if _, err := os.Stat(snapPath); err != nil {
+		t.Errorf("DeleteObject removed the object under the excluded directory: %v", err)
+	}
+
+	_, err := p.CopyObject(ctx, s3response.CopyObjectInput{
+		Bucket:              &bucket,
+		Key:                 aws.String("copied.txt"),
+		CopySource:          aws.String(bucket + "/" + snapKey),
+		ExpectedBucketOwner: aws.String(""),
+	})
+	if !errors.Is(err, reserved) {
+		t.Errorf("CopyObject from excluded directory: got %v, want ReservedObjectKey", err)
+	}
+	_, err = p.CopyObject(ctx, s3response.CopyObjectInput{
+		Bucket:              &bucket,
+		Key:                 aws.String("child/.zfs/copied.txt"),
+		CopySource:          aws.String(bucket + "/top.txt"),
+		ExpectedBucketOwner: aws.String(""),
+	})
+	if !errors.Is(err, reserved) {
+		t.Errorf("CopyObject into excluded directory: got %v, want ReservedObjectKey", err)
+	}
+
+	_, err = p.CreateMultipartUpload(ctx, s3response.CreateMultipartUploadInput{
+		Bucket: &bucket, Key: aws.String("child/.zfs/mp.txt"),
+	})
+	if !errors.Is(err, reserved) {
+		t.Errorf("CreateMultipartUpload: got %v, want ReservedObjectKey", err)
+	}
+
+	// A file with an excluded name is an ordinary object, as the listings
+	// report it, and so are keys that only resemble the excluded name.
+	for _, key := range []string{"d/" + zfsSnapDir, "child/.zfsx/a", "child/x.zfs/a"} {
+		if err := putTestObject(t, p, bucket, key, "data"); err != nil {
+			t.Errorf("PutObject %q: %v", key, err)
+			continue
+		}
+		if _, err := p.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: aws.String(key)}); err != nil {
+			t.Errorf("HeadObject %q: %v", key, err)
+		}
+	}
+}
+
+// TestExcludeDirsEmptyKeysAllowed checks the default: without ExcludeDirs,
+// .zfs is an ordinary name and its objects can be read.
+func TestExcludeDirsEmptyKeysAllowed(t *testing.T) {
+	bucket := snapBucket
+	p := newSnapdirBackend(t, meta.XattrMeta{}, nil)
+
+	_, err := p.HeadObject(context.Background(), &s3.HeadObjectInput{
+		Bucket: &bucket, Key: aws.String("child/.zfs/snapshot/one/snap.txt"),
+	})
+	if err != nil {
+		t.Fatalf("HeadObject: %v", err)
+	}
+}
+
 func containsKey(keys []string, key string) bool {
 	for _, k := range keys {
 		if k == key {
@@ -233,3 +456,96 @@ func equalKeys(got, want []string) bool {
 	}
 	return true
 }
+
+// TestExcludedDirBucketRejected checks that an excluded name is not a bucket,
+// even with relaxed bucket names. On ZFS a bucket named .zfs at a dataset root
+// would otherwise serve snapshot contents through keys with no excluded
+// element of their own.
+func TestExcludedDirBucketRejected(t *testing.T) {
+	ctx := context.Background()
+	invalid := s3err.GetAPIError(s3err.ErrInvalidBucketName)
+
+	sidecar := t.TempDir()
+	sc, err := meta.NewSideCar(sidecar)
+	if err != nil {
+		t.Fatalf("new sidecar: %v", err)
+	}
+	root := t.TempDir()
+	p, err := New(root, sc, PosixOpts{
+		NewDirPerm:    0755,
+		AbsolutePaths: true,
+		SideCarDir:    sidecar,
+		ExcludeDirs:   []string{zfsSnapDir},
+	})
+	if err != nil {
+		t.Fatalf("new posix: %v", err)
+	}
+	t.Cleanup(p.Shutdown)
+
+	createTestBucket(t, p, "bkt")
+	snapKey := "snapshot/s1/bkt/private/secret.txt"
+	snapFile := filepath.Join(root, zfsSnapDir, filepath.FromSlash(snapKey))
+	if err := os.MkdirAll(filepath.Dir(snapFile), 0755); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+	if err := os.WriteFile(snapFile, []byte("secret"), 0644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	for _, bucket := range []string{zfsSnapDir, strings.ToUpper(zfsSnapDir)} {
+		if p.IsBucketValid(bucket) {
+			t.Errorf("IsBucketValid(%q) = true, want false", bucket)
+		}
+		if _, err := p.HeadBucket(ctx, &s3.HeadBucketInput{Bucket: &bucket}); !errors.Is(err, invalid) {
+			t.Errorf("HeadBucket %q: got %v, want InvalidBucketName", bucket, err)
+		}
+		// Rejected before the ACL is read, which fails on a ZFS control
+		// directory with xattr metadata.
+		if _, err := p.GetBucketAcl(ctx, &s3.GetBucketAclInput{Bucket: &bucket}); !errors.Is(err, invalid) {
+			t.Errorf("GetBucketAcl %q: got %v, want InvalidBucketName", bucket, err)
+		}
+		if _, err := p.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &snapKey}); !errors.Is(err, invalid) {
+			t.Errorf("HeadObject %q: got %v, want InvalidBucketName", bucket, err)
+		}
+		if _, err := p.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &snapKey}); !errors.Is(err, invalid) {
+			t.Errorf("GetObject %q: got %v, want InvalidBucketName", bucket, err)
+		}
+		if err := putTestObject(t, p, bucket, "new.txt", "x"); !errors.Is(err, invalid) {
+			t.Errorf("PutObject %q: got %v, want InvalidBucketName", bucket, err)
+		}
+		err := p.CreateBucket(ctx, &s3.CreateBucketInput{
+			Bucket:                    &bucket,
+			CreateBucketConfiguration: &types.CreateBucketConfiguration{},
+		}, []byte{})
+		if !errors.Is(err, invalid) {
+			t.Errorf("CreateBucket %q: got %v, want InvalidBucketName", bucket, err)
+		}
+	}
+	if _, err := os.Stat(filepath.Join(root, zfsSnapDir, "new.txt")); !os.IsNotExist(err) {
+		t.Errorf("PutObject created an object in the excluded directory: %v", err)
+	}
+
+	out, err := p.ListBuckets(ctx, s3response.ListBucketsInput{IsAdmin: true, MaxBuckets: 1000})
+	if err != nil {
+		t.Fatalf("ListBuckets: %v", err)
+	}
+	var names []string
+	for _, b := range out.Buckets.Bucket {
+		names = append(names, b.Name)
+	}
+	if !equalKeys(names, []string{"bkt"}) {
+		t.Errorf("ListBuckets: got %v, want [bkt]", names)
+	}
+
+	owners, err := p.ListBucketsAndOwners(ctx)
+	if err != nil {
+		t.Fatalf("ListBucketsAndOwners: %v", err)
+	}
+	names = nil
+	for _, b := range owners {
+		names = append(names, b.Name)
+	}
+	if !equalKeys(names, []string{"bkt"}) {
+		t.Errorf("ListBucketsAndOwners: got %v, want [bkt]", names)
+	}
+}

backend/posix/posix.go

@@ -150,10 +150,13 @@ type Posix struct {
 	//   multipart composite: "ALGO-<composite-checksum>-<part-count>"
 	dataIntegrityEtag bool
 
-	// skipDirs are the directory names the listing walks never descend
-	// into: the gateway's metadata directory plus PosixOpts.ExcludeDirs.
-	// It is built once in New so all three list operations share it.
-	skipDirs []string
+	// skipDirs are the directories the listing walks never descend into:
+	// the gateway's metadata directory at the bucket root, and the
+	// PosixOpts.ExcludeDirs entries at any depth. It is built once in New
+	// so all three list operations share it, and validateObjectName rejects
+	// keys through the excluded entries so objects can't be read or written
+	// where the listings never show them.
+	skipDirs backend.SkipDirs
 
 	// objLockSlots are process-local striped slots taken alongside the shared
 	// per-bucket lock files by lockObjectPublish. Channels make local lock
@@ -317,26 +320,38 @@ type PosixOpts struct {
 	// filesystem control directories that hold something other than object
 	// data, such as ".zfs" on a ZFS dataset mounted with snapdir=visible,
 	// where the metadata xattrs are unsupported and reading them would fail
-	// the whole listing. The gateway's own metadata directory is always
-	// skipped and does not need to be listed here.
+	// the whole listing. The gateway's own metadata directory at the bucket
+	// root is always skipped and does not need to be listed here.
+	//
+	// Only directories are excluded; a regular file with an excluded name
+	// is still listed. Names are matched ignoring case. Object keys with an
+	// excluded name as one of their directories, at any depth, are rejected
+	// with ReservedObjectKey, so objects there can't be read or written
+	// either. On ZFS that also keeps snapshot contents under .zfs from being
+	// read through the gateway. An excluded name is not a valid bucket name
+	// either, even with relaxed bucket name validation, and a directory with
+	// that name at the gateway root is not listed as a bucket.
 	ExcludeDirs []string
 }
 
-// listingSkipDirs returns the directory names the listing walks skip: the
-// gateway's own metadata directory, which is never bucket content, followed
-// by the configured filesystem-specific control directories. Each configured
-// entry has to be a single, non-empty path segment; anything else could never
+// listingSkipDirs returns the directories the listing walks skip: the
+// gateway's own metadata directory, which only exists at the bucket root and
+// is never bucket content, and the configured filesystem-specific control
+// directories, which are skipped at any depth. Each configured
+// entry has to be a single path segment naming a directory entry, so it can't
+// be empty, contain a separator, or be "." or ".."; anything else could never
 // match a directory entry name and would silently do nothing.
-func listingSkipDirs(excludeDirs []string) ([]string, error) {
-	skipDirs := make([]string, 0, len(excludeDirs)+1)
-	skipDirs = append(skipDirs, MetaTmpDir)
+func listingSkipDirs(excludeDirs []string) (backend.SkipDirs, error) {
 	for _, dir := range excludeDirs {
-		if dir == "" || strings.ContainsRune(dir, '/') || strings.ContainsRune(dir, filepath.Separator) {
-			return nil, fmt.Errorf("invalid excluded directory %q: must be a single, non-empty path segment", dir)
+		if dir == "" || dir == "." || dir == ".." ||
+			strings.ContainsRune(dir, '/') || strings.ContainsRune(dir, filepath.Separator) {
+			return backend.SkipDirs{}, fmt.Errorf("invalid excluded directory %q: must be a single path segment naming a directory entry", dir)
 		}
-		skipDirs = append(skipDirs, dir)
 	}
-	return skipDirs, nil
+	return backend.SkipDirs{
+		Root: []string{MetaTmpDir},
+		Any:  slices.Clone(excludeDirs),
+	}, nil
 }
 
 // New returns a Posix backend serving buckets from the directories under rootdir.
@@ -666,18 +681,51 @@ func (p *Posix) ValidateObjectKey(bucket, key string) error {
 // The path is resolved like ObjectPath resolves it, so that "./", "x/../" and
 // repeated slashes don't hide the directory, and its name is matched ignoring
 // case, like case-insensitive file systems match it.
+//
+// It also returns ReservedObjectKey if a directory in the path, at any depth,
+// is named in PosixOpts.ExcludeDirs. The listings never show objects there,
+// and on ZFS the .zfs directory would otherwise expose snapshot contents
+// under keys that bucket policies on the live keys don't cover. A file with
+// an excluded name is still a valid object, as the listings report it.
 func (p *Posix) validateObjectName(bucket, object string) error {
 	rel, err := filepath.Rel(p.BucketPath(bucket), p.ObjectPath(bucket, object))
 	if err != nil {
 		return fmt.Errorf("resolve object path: %w", err)
 	}
-	dir, _, _ := strings.Cut(filepath.ToSlash(rel), "/")
+	rel = filepath.ToSlash(rel)
+	dir, _, _ := strings.Cut(rel, "/")
 	if strings.EqualFold(dir, MetaTmpDir) {
 		return s3err.GetAPIError(s3err.ErrReservedObjectKey)
 	}
+	if p.inExcludedDir(rel, strings.HasSuffix(object, "/")) {
+		return s3err.GetAPIError(s3err.ErrExcludedObjectKey)
+	}
 	return nil
 }
 
+// inExcludedDir reports whether a directory in rel, a slash-separated path
+// relative to the bucket directory, is named in PosixOpts.ExcludeDirs. The
+// last element of rel is a directory only when isDir is set, as it is for
+// the key of a directory object.
+func (p *Posix) inExcludedDir(rel string, isDir bool) bool {
+	if len(p.skipDirs.Any) == 0 {
+		return false
+	}
+	dirs := strings.Split(rel, "/")
+	if !isDir {
+		dirs = dirs[:len(dirs)-1]
+	}
+	return slices.ContainsFunc(dirs, p.isExcludedName)
+}
+
+// isExcludedName reports whether name is named in PosixOpts.ExcludeDirs,
+// ignoring case.
+func (p *Posix) isExcludedName(name string) bool {
+	return slices.ContainsFunc(p.skipDirs.Any, func(excluded string) bool {
+		return strings.EqualFold(name, excluded)
+	})
+}
+
 // DoesBucketExist returns NoSuchBucket unless bucket names a bucket. Backends
 // built on Posix that add their own entry points must check buckets with it
 // rather than stat the bucket path themselves.
@@ -751,7 +799,7 @@ func (p *Posix) ListBuckets(ctx context.Context, input s3response.ListBucketsInp
 			continue
 		}
 
-		if !utils.IsValidBucketName(fi.Name()) {
+		if !utils.IsValidBucketName(fi.Name()) || p.isExcludedName(fi.Name()) {
 			continue
 		}
 
@@ -826,6 +874,14 @@ func (p *Posix) isBucketValid(bucket string) bool {
 		return false
 	}
 
+	// An excluded directory at the root is no bucket either, even with
+	// relaxed bucket names: on ZFS a bucket named .zfs at a dataset root
+	// would serve snapshot contents through keys that have no excluded
+	// element of their own.
+	if p.isExcludedName(bucket) {
+		return false
+	}
+
 	if !p.validateBucketName {
 		return true
 	}
@@ -8321,6 +8377,10 @@ func (p *Posix) ListBucketsAndOwners(ctx context.Context) (buckets []s3response.
 	}
 
 	for _, fi := range fis {
+		if p.isExcludedName(fi.Name()) {
+			continue
+		}
+
 		aclJSON, err := p.meta.RetrieveAttribute(nil, fi.Name(), "", aclkey)
 		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
 			return buckets, fmt.Errorf("get acl tag: %w", err)

backend/walk.go

@@ -48,7 +48,7 @@ type walkState struct {
 	marker     string
 	max        int32
 	getObj     GetObjFunc
-	skipdirs   []string
+	skipDirs   SkipDirs
 
 	// Mutable state
 	cpmap     cpMap
@@ -61,6 +61,27 @@ type walkState struct {
 
 type GetObjFunc func(path string, d fs.DirEntry) (s3response.Object, error)
 
+// SkipDirs names the directories a walk never descends into or reports as
+// objects. Regular files with these names are ordinary objects. Names are
+// matched ignoring case: a case-insensitive file system resolves a name in
+// any case to the same directory, and matching the same way on every file
+// system keeps the listings in agreement with backends that reject object
+// keys through these directories ignoring case.
+type SkipDirs struct {
+	// Root names directories skipped only at the top of the walked file
+	// system, such as the gateway's own metadata directory.
+	Root []string
+	// Any names directories skipped at every depth.
+	Any []string
+}
+
+// skips reports whether a directory named name is skipped. atRoot reports
+// whether the directory is at the top of the walked file system.
+func (s SkipDirs) skips(name string, atRoot bool) bool {
+	match := func(skip string) bool { return strings.EqualFold(name, skip) }
+	return slices.ContainsFunc(s.Any, match) || (atRoot && slices.ContainsFunc(s.Root, match))
+}
+
 var ErrSkipObj = errors.New("skip this object")
 
 // map to store object common prefixes
@@ -93,7 +114,7 @@ func (c cpMap) CpArray() []types.CommonPrefix {
 
 // Walk walks the supplied fs.FS and returns results compatible with list
 // objects responses
-func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker string, max int32, getObj GetObjFunc, skipdirs []string) (WalkResults, error) {
+func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker string, max int32, getObj GetObjFunc, skipDirs SkipDirs) (WalkResults, error) {
 	// if max is 0, it should return empty non-truncated result
 	if max == 0 {
 		return WalkResults{
@@ -109,7 +130,7 @@ func Walk(ctx context.Context, fileSystem fs.FS, prefix, delimiter, marker strin
 		marker:     marker,
 		max:        max,
 		getObj:     getObj,
-		skipdirs:   skipdirs,
+		skipDirs:   skipDirs,
 		cpmap:      cpMap{},
 	}
 
@@ -146,7 +167,7 @@ type GetVersionsFunc func(path, versionIdMarker string, pastVersionIdMarker *boo
 
 // WalkVersions walks the supplied fs.FS and returns results compatible with
 // ListObjectVersions action response
-func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyMarker, versionIdMarker string, max int, getObj GetVersionsFunc, skipdirs []string) (WalkVersioningResults, error) {
+func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyMarker, versionIdMarker string, max int, getObj GetVersionsFunc, skipDirs SkipDirs) (WalkVersioningResults, error) {
 	cpmap := cpMap{}
 	var objects []s3response.ObjectVersion
 	var delMarkers []s3response.DeleteMarkerEntry
@@ -172,7 +193,10 @@ func WalkVersions(ctx context.Context, fileSystem fs.FS, prefix, delimiter, keyM
 		if path == "." {
 			return nil
 		}
-		if slices.Contains(skipdirs, d.Name()) {
+		// Only directories are skipped by name. A file with a skipped
+		// name is an ordinary object, and returning SkipDir for a file
+		// would skip the rest of its directory.
+		if d.IsDir() && skipDirs.skips(d.Name(), !strings.Contains(path, pathSeparator)) {
 			return fs.SkipDir
 		}
 
@@ -381,7 +405,7 @@ func readDirEntries(path string, entries []fs.DirEntry, walkstate *walkState) {
 			return
 		}
 
-		if shouldSkip(entries[entriesIndex].Name(), walkstate) {
+		if shouldSkip(path, entries[entriesIndex], walkstate) {
 			entriesIndex++
 			continue
 		}
@@ -444,7 +468,7 @@ func processDir(
 			// if it is a directory, we
 			// need to call recursively to
 			// check for further preceding entries
-			if shouldSkip(entries[entriesIndex].Name(), walkstate) {
+			if shouldSkip(path, entries[entriesIndex], walkstate) {
 				entriesIndex++
 				continue
 			}
@@ -558,21 +582,22 @@ func checkAndMaybeAddCommonPrefix(objectName string, walkstate *walkState,
 	return false
 }
 
-func shouldSkip(name string, walkstate *walkState) bool {
-	return slices.Contains(walkstate.skipdirs, name)
+// shouldSkip reports whether entry, read from the directory path, is a
+// skipped directory. A file with a skipped name is an ordinary object and is
+// not skipped.
+func shouldSkip(path string, entry fs.DirEntry, walkstate *walkState) bool {
+	return entry.IsDir() && walkstate.skipDirs.skips(entry.Name(), path == pathDot)
 }
 
 // inSkipDir reports whether dir, a directory named by the listing prefix, is
-// a skipped directory or inside one. The prefix spells the directory names
-// as the client sent them, which a case-insensitive file system resolves to
-// a skipped directory in any case, so they are matched ignoring case.
+// a skipped directory or inside one.
 func inSkipDir(dir string, walkstate *walkState) bool {
+	i := 0
 	for name := range strings.SplitSeq(dir, pathSeparator) {
-		if slices.ContainsFunc(walkstate.skipdirs, func(skip string) bool {
-			return strings.EqualFold(name, skip)
-		}) {
+		if walkstate.skipDirs.skips(name, i == 0) {
 			return true
 		}
+		i++
 	}
 	return false
 }
@@ -636,7 +661,7 @@ func quickWalk(walkstate *walkState) error {
 	}
 
 	if useRootEntries {
-		// keep fast-path semantics but still honor skipdirs for the prefixed
+		// keep fast-path semantics but still honor skipDirs for the prefixed
 		// root and every directory above it
 		if !inSkipDir(rootDir, walkstate) {
 			// Mirror processDir: offer the root directory itself to getObj whenever

backend/walk_bench_test.go

@@ -284,7 +284,7 @@ func benchGetObj(path string, d fs.DirEntry) (s3response.Object, error) {
 // runWalk is a convenience wrapper so each benchmark loop is one line.
 func runWalk(b *testing.B, fsys fs.FS, prefix, delim, marker string, max int32) {
 	b.Helper()
-	_, err := backend.Walk(context.Background(), fsys, prefix, delim, marker, max, benchGetObj, nil)
+	_, err := backend.Walk(context.Background(), fsys, prefix, delim, marker, max, benchGetObj, backend.SkipDirs{})
 	if err != nil {
 		b.Fatalf("Walk: %v", err)
 	}
@@ -320,7 +320,7 @@ func BenchmarkWalkFlatPaged(b *testing.B) {
 		marker := ""
 		for {
 			res, err := backend.Walk(context.Background(),
-				fsys, "", "", marker, 1000, benchGetObj, nil)
+				fsys, "", "", marker, 1000, benchGetObj, backend.SkipDirs{})
 			if err != nil {
 				b.Fatalf("Walk: %v", err)
 			}
@@ -427,7 +427,7 @@ func BenchmarkWalkLargeFlatPaged(b *testing.B) {
 		marker := ""
 		for {
 			res, err := backend.Walk(context.Background(),
-				fsys, "", "", marker, 1000, benchGetObj, nil)
+				fsys, "", "", marker, 1000, benchGetObj, backend.SkipDirs{})
 			if err != nil {
 				b.Fatalf("Walk: %v", err)
 			}

backend/walk_test.go

@@ -432,7 +432,7 @@ func TestWalk(t *testing.T) {
 		for _, tc := range tt.cases {
 			res, err := backend.Walk(context.Background(),
 				tt.fsys, tc.prefix, tc.delimiter, tc.marker, tc.maxObjs,
-				tt.getobj, []string{})
+				tt.getobj, backend.SkipDirs{})
 			if err != nil {
 				t.Errorf("%v: walk: %v", tc.name, err)
 			}
@@ -575,7 +575,7 @@ func TestWalkStop(t *testing.T) {
 		_, err = backend.Walk(ctx, s, "", "", "", 1000,
 			func(path string, d fs.DirEntry) (s3response.Object, error) {
 				return s3response.Object{}, nil
-			}, []string{})
+			}, backend.SkipDirs{})
 	})
 
 	select {
@@ -609,7 +609,7 @@ func TestWalkPrefixInSkipDir(t *testing.T) {
 	} {
 		for _, delimiter := range []string{"", "/"} {
 			res, err := backend.Walk(context.Background(), fsys, prefix, delimiter, "", 1000,
-				getObj, []string{".sgwtmp"})
+				getObj, backend.SkipDirs{Root: []string{".sgwtmp"}})
 			if err != nil {
 				t.Errorf("prefix %q delimiter %q: walk: %v", prefix, delimiter, err)
 				continue
@@ -622,6 +622,166 @@ func TestWalkPrefixInSkipDir(t *testing.T) {
 	}
 }
 
+// skipNameFS has skipped directories, which are never walked, and regular
+// files with skipped names, which are ordinary objects. ".zfs" sorts before
+// the siblings of d/.zfs, so skipping that file the way a directory is skipped
+// would drop the siblings. "a.b" and "c.d" sort after "a" and "c" in readdir
+// order but before "a/" and "c/" as keys, so processDir reaches them through
+// its lookahead: the file a.b is listed and the directory c.d is skipped.
+var skipNameFS = fstest.MapFS{
+	".zfs/snapshot/one/snap.txt": {},
+	"a.b":                        {},
+	"a/x":                        {},
+	"c.d/y":                      {},
+	"c/x":                        {},
+	"d/.zfs":                     {},
+	"d/b.txt":                    {},
+	"d/c.txt":                    {},
+	"top.txt":                    {},
+}
+
+var skipNameDirs = backend.SkipDirs{Any: []string{".zfs", "a.b", "c.d"}}
+
+var skipNameKeys = []string{"a.b", "a/x", "c/x", "d/.zfs", "d/b.txt", "d/c.txt", "top.txt"}
+
+// TestWalkSkipDirsFileNotSkipped checks that Walk skips only directories
+// named in skipdirs, not regular files with the same name.
+func TestWalkSkipDirsFileNotSkipped(t *testing.T) {
+	res, err := backend.Walk(context.Background(), skipNameFS, "", "", "", 1000,
+		getObjSkipDirs, skipNameDirs)
+	if err != nil {
+		t.Fatalf("walk: %v", err)
+	}
+	var want []s3response.Object
+	for _, key := range skipNameKeys {
+		want = append(want, s3response.Object{Key: backend.GetPtrFromString(key)})
+	}
+	compareResultsOrdered("files named like skipped dirs", res, backend.WalkResults{Objects: want}, t)
+
+	// A prefix inside the directory starts the walk there, and the file
+	// with the skipped name is still listed.
+	res, err = backend.Walk(context.Background(), skipNameFS, "d/", "", "", 1000,
+		getObjSkipDirs, skipNameDirs)
+	if err != nil {
+		t.Fatalf("walk prefix d/: %v", err)
+	}
+	compareResultsOrdered("file named like skipped dir under prefix", res, backend.WalkResults{
+		Objects: []s3response.Object{
+			{Key: backend.GetPtrFromString("d/.zfs")},
+			{Key: backend.GetPtrFromString("d/b.txt")},
+			{Key: backend.GetPtrFromString("d/c.txt")},
+		},
+	}, t)
+}
+
+// TestWalkVersionsSkipDirsFileNotSkipped checks that WalkVersions skips only
+// directories named in skipdirs. It used to return fs.SkipDir for a regular
+// file with a skipped name too, which skips the rest of the file's directory.
+func TestWalkVersionsSkipDirsFileNotSkipped(t *testing.T) {
+	getFileVersions := func(path, versionIdMarker string, pastVersionIdMarker *bool, availableObjCount int, d fs.DirEntry) (*backend.ObjVersionFuncResult, error) {
+		if d.IsDir() {
+			return nil, backend.ErrSkipObj
+		}
+		return getVersionsTestFunc(path, versionIdMarker, pastVersionIdMarker, availableObjCount, d)
+	}
+
+	res, err := backend.WalkVersions(context.Background(), skipNameFS, "", "", "", "", 1000,
+		getFileVersions, skipNameDirs)
+	if err != nil {
+		t.Fatalf("WalkVersions: %v", err)
+	}
+	if !compareObjectVersionsOrdered(res.ObjectVersions, makeObjectVersions(skipNameKeys)) {
+		t.Errorf("unexpected ObjectVersions: got %v, want %v",
+			printVersionObjects(res.ObjectVersions), skipNameKeys)
+	}
+}
+
+// rootSkipFS has the gateway's metadata directory at the root, where it is
+// skipped, and directories of the same name further down, where they are
+// ordinary bucket content. It also has a skipped name spelled in another
+// case, which is skipped too, like object requests reject keys through it.
+var rootSkipFS = fstest.MapFS{
+	".sgwtmp/multipart/x": {},
+	".SGWTMP/y":           {},
+	"d/.sgwtmp/x":         {},
+	"e/.SGWTMP/z":         {},
+	"e/.ZFS/snap":         {},
+	"top.txt":             {},
+}
+
+var rootSkipDirs = backend.SkipDirs{Root: []string{".sgwtmp"}, Any: []string{".zfs"}}
+
+var rootSkipKeys = []string{"d/.sgwtmp/x", "e/.SGWTMP/z", "top.txt"}
+
+// TestWalkSkipDirsRootAndCase checks that Walk skips SkipDirs.Root names only
+// at the root, SkipDirs.Any names at any depth, and both ignoring case.
+func TestWalkSkipDirsRootAndCase(t *testing.T) {
+	ctx := context.Background()
+
+	res, err := backend.Walk(ctx, rootSkipFS, "", "", "", 1000, getObjSkipDirs, rootSkipDirs)
+	if err != nil {
+		t.Fatalf("walk: %v", err)
+	}
+	var want []s3response.Object
+	for _, key := range rootSkipKeys {
+		want = append(want, s3response.Object{Key: backend.GetPtrFromString(key)})
+	}
+	compareResultsOrdered("root and any-depth skip dirs", res, backend.WalkResults{Objects: want}, t)
+
+	res, err = backend.Walk(ctx, rootSkipFS, "", "/", "", 1000, getObjSkipDirs, rootSkipDirs)
+	if err != nil {
+		t.Fatalf("walk with delimiter: %v", err)
+	}
+	compareResultsOrdered("root and any-depth skip dirs with delimiter", res, backend.WalkResults{
+		CommonPrefixes: []types.CommonPrefix{
+			{Prefix: backend.GetPtrFromString("d/")},
+			{Prefix: backend.GetPtrFromString("e/")},
+		},
+		Objects: []s3response.Object{{Key: backend.GetPtrFromString("top.txt")}},
+	}, t)
+
+	for _, tc := range []struct {
+		prefix string
+		keys   []string
+	}{
+		{"d/.sgwtmp/", []string{"d/.sgwtmp/x"}},
+		{"e/.SGWTMP/", []string{"e/.SGWTMP/z"}},
+		{"e/.ZFS/", nil},
+		{".SGWTMP/", nil},
+	} {
+		res, err := backend.Walk(ctx, rootSkipFS, tc.prefix, "", "", 1000, getObjSkipDirs, rootSkipDirs)
+		if err != nil {
+			t.Fatalf("walk prefix %q: %v", tc.prefix, err)
+		}
+		var want []s3response.Object
+		for _, key := range tc.keys {
+			want = append(want, s3response.Object{Key: backend.GetPtrFromString(key)})
+		}
+		compareResultsOrdered("prefix "+tc.prefix, res, backend.WalkResults{Objects: want}, t)
+	}
+}
+
+// TestWalkVersionsSkipDirsRootAndCase checks that WalkVersions applies the
+// same root-only and case-insensitive rules as Walk.
+func TestWalkVersionsSkipDirsRootAndCase(t *testing.T) {
+	getFileVersions := func(path, versionIdMarker string, pastVersionIdMarker *bool, availableObjCount int, d fs.DirEntry) (*backend.ObjVersionFuncResult, error) {
+		if d.IsDir() {
+			return nil, backend.ErrSkipObj
+		}
+		return getVersionsTestFunc(path, versionIdMarker, pastVersionIdMarker, availableObjCount, d)
+	}
+
+	res, err := backend.WalkVersions(context.Background(), rootSkipFS, "", "", "", "", 1000,
+		getFileVersions, rootSkipDirs)
+	if err != nil {
+		t.Fatalf("WalkVersions: %v", err)
+	}
+	if !compareObjectVersionsOrdered(res.ObjectVersions, makeObjectVersions(rootSkipKeys)) {
+		t.Errorf("unexpected ObjectVersions: got %v, want %v",
+			printVersionObjects(res.ObjectVersions), rootSkipKeys)
+	}
+}
+
 // caseFoldFS opens the files of a MapFS with lowercase names by names in any
 // case, like a case-insensitive file system.
 type caseFoldFS struct {
@@ -651,7 +811,7 @@ func TestWalkDirectoryErrorPropagates(t *testing.T) {
 				return s3response.Object{}, fmt.Errorf("boom")
 			}
 			return getObj(path, d)
-		}, []string{})
+		}, backend.SkipDirs{})
 	if err == nil {
 		t.Fatal("expected error, got nil")
 	}
@@ -672,7 +832,7 @@ func TestWalkDirectoryErrSkipObjContinues(t *testing.T) {
 				return s3response.Object{}, backend.ErrSkipObj
 			}
 			return getObj(path, d)
-		}, []string{})
+		}, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("walk: %v", err)
 	}
@@ -854,7 +1014,7 @@ func TestOrderWalk(t *testing.T) {
 		for _, tc := range tt.cases {
 			res, err := backend.Walk(context.Background(),
 				tt.fsys, tc.prefix, tc.delimiter, tc.marker, tc.maxObjs,
-				tt.getobj, []string{})
+				tt.getobj, backend.SkipDirs{})
 			if err != nil {
 				t.Errorf("%v: walk: %v", tc.name, err)
 			}
@@ -1071,7 +1231,7 @@ func TestMarker(t *testing.T) {
 			for i, page := range tc.expected {
 				res, err := backend.Walk(context.Background(),
 					tt.fsys, tc.prefix, tc.delimiter, marker, tc.maxObjs,
-					tt.getobj, []string{})
+					tt.getobj, backend.SkipDirs{})
 				if err != nil {
 					t.Errorf("%v: walk: %v", tc.name, err)
 				}
@@ -1181,7 +1341,7 @@ func TestWalkVersions(t *testing.T) {
 		},
 	}
 
-	res, err := backend.WalkVersions(context.Background(), fsys, "", "", "", "", 1000, getVersionsTestFunc, []string{})
+	res, err := backend.WalkVersions(context.Background(), fsys, "", "", "", "", 1000, getVersionsTestFunc, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("walk versions: %v", err)
 	}
@@ -1213,7 +1373,7 @@ func TestOrderWalkVersions(t *testing.T) {
 		},
 	}
 
-	res, err := backend.WalkVersions(context.Background(), fsys, "dir1/", "", "", "", 1000, getVersionsTestFunc, []string{})
+	res, err := backend.WalkVersions(context.Background(), fsys, "dir1/", "", "", "", 1000, getVersionsTestFunc, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("order walk versions: %v", err)
 	}
@@ -1367,7 +1527,7 @@ func TestWalkVersionsTruncated(t *testing.T) {
 
 	// Test first page with limit of 5 (should be truncated)
 	maxKeys := 5
-	res1, err := backend.WalkVersions(context.Background(), fsys, "", "", "", "", maxKeys, getVersionsFunc, []string{})
+	res1, err := backend.WalkVersions(context.Background(), fsys, "", "", "", "", maxKeys, getVersionsFunc, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("walk versions first page: %v", err)
 	}
@@ -1402,7 +1562,7 @@ func TestWalkVersionsTruncated(t *testing.T) {
 	}
 
 	// Test second page using markers
-	res2, err := backend.WalkVersions(context.Background(), fsys, "", "", res1.NextMarker, res1.NextVersionIdMarker, maxKeys, getVersionsFunc, []string{})
+	res2, err := backend.WalkVersions(context.Background(), fsys, "", "", res1.NextMarker, res1.NextVersionIdMarker, maxKeys, getVersionsFunc, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("walk versions second page: %v", err)
 	}
@@ -1420,7 +1580,7 @@ func TestWalkVersionsTruncated(t *testing.T) {
 	// Test third page if needed
 	var res3 backend.WalkVersioningResults
 	if res2.Truncated {
-		res3, err = backend.WalkVersions(context.Background(), fsys, "", "", res2.NextMarker, res2.NextVersionIdMarker, maxKeys, getVersionsFunc, []string{})
+		res3, err = backend.WalkVersions(context.Background(), fsys, "", "", res2.NextMarker, res2.NextVersionIdMarker, maxKeys, getVersionsFunc, backend.SkipDirs{})
 		if err != nil {
 			t.Fatalf("walk versions third page: %v", err)
 		}
@@ -1460,7 +1620,7 @@ func TestWalkVersionsPrefixSkipsParents(t *testing.T) {
 	prefix := "vendor/Backup/vendor/Clients/"
 	delimiter := "/"
 
-	res, err := backend.WalkVersions(context.Background(), fsys, prefix, delimiter, "", "", 1000, getVersionsTestFunc, []string{})
+	res, err := backend.WalkVersions(context.Background(), fsys, prefix, delimiter, "", "", 1000, getVersionsTestFunc, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("WalkVersions: %v", err)
 	}
@@ -1493,7 +1653,7 @@ func TestWalkVersionsPrefixNoDelimiterSkipsParents(t *testing.T) {
 
 	prefix := "vendor/Backup/vendor/Clients/"
 
-	res, err := backend.WalkVersions(context.Background(), fsys, prefix, "", "", "", 1000, getVersionsTestFunc, []string{})
+	res, err := backend.WalkVersions(context.Background(), fsys, prefix, "", "", "", 1000, getVersionsTestFunc, backend.SkipDirs{})
 	if err != nil {
 		t.Fatalf("WalkVersions: %v", err)
 	}

cmd/internal/gwcli/posix.go

@@ -184,7 +184,7 @@ will be translated into the file /mnt/fs/gwroot/mybucket/a/b/c/myobject`,
 			},
 			&cli.StringSliceFlag{
 				Name:    "exclude-dirs",
-				Usage:   "directory names the object listings do not descend into or report as objects (can be specified multiple times); use it for a filesystem control directory that holds something other than object data, such as .zfs on a ZFS dataset mounted with snapdir=visible",
+				Usage:   "directory name, not a path, to hide from object listings and reserve at any depth in a bucket; object keys through it are rejected, and it can't be used as a bucket name (can be specified multiple times)",
 				EnvVars: []string{"VGW_EXCLUDE_DIRS"},
 			},
 		},

extra/example.conf

@@ -781,6 +781,24 @@ ROOT_SECRET_ACCESS_KEY=
 # as an opaque identifier instead of a content-MD5 signal.
 #VGW_DATA_INTEGRITY_ETAG=false
 
+# The VGW_EXCLUDE_DIRS option lists directory names, separated by commas, that
+# ListObjects, ListObjectsV2 and ListObjectVersions do not descend into or
+# report as objects. Use it for filesystem control directories that hold
+# something other than object data, such as .zfs on a ZFS dataset mounted with
+# snapdir=visible, where reading object metadata from the control directory
+# fails the whole listing. Each entry is a single directory name, not a path,
+# and matches a directory with that name, ignoring case, at any depth in a
+# bucket. Regular files with an excluded name are still listed. Object keys
+# with an excluded directory in their path are reserved: requests for them fail
+# with ReservedObjectKey, so objects already stored under an excluded directory
+# become unreachable through the gateway. On ZFS this also keeps snapshot
+# contents under .zfs, which is reachable by name even with snapdir=hidden,
+# from being read through the gateway. An excluded name is also rejected as a
+# bucket name, even with relaxed bucket name validation, and a directory with
+# that name at the gateway root is not listed as a bucket. The gateway's own
+# .sgwtmp directory at the bucket root is always excluded.
+#VGW_EXCLUDE_DIRS=
+
 ###########
 # scoutfs #
 ###########

s3err/s3err.go

@@ -212,6 +212,7 @@ const (
 	ErrACLsDisabled
 	ErrNoSpaceLeftOnDevice
 	ErrReservedObjectKey
+	ErrExcludedObjectKey
 
 	// Admin api errors
 	ErrAdminAccessDenied
@@ -816,6 +817,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "Object keys in the .sgwtmp directory are reserved for the gateway's internal use.",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrExcludedObjectKey: {
+		Code:           "ReservedObjectKey",
+		Description:    "Object keys in a directory excluded from object listings are reserved.",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 
 	// Admin api errors
 	ErrAdminAccessDenied: {