fix: the Stat call in s3 storage drivers should not rely on lexographical sort only (cd8513b7) · Commits · GitLab.org / container-registry

registry/storage/driver/s3-aws/v1/s3.go

+64 −17

Original line number	Diff line number	Diff line
		@@ -369,16 +369,18 @@ func (d *driver) Writer(ctx context.Context, path string, appendParam bool) (sto
		// in bytes and the creation time.
		func (d *driver) Stat(ctx context.Context, path string) (storagedriver.FileInfo, error) {
		s3Path := d.s3Path(path)
		resp, err := d.S3.ListObjectsV2WithContext(
		ctx,
		&s3.ListObjectsV2Input{
		listInput := &s3.ListObjectsV2Input{
		Bucket: aws.String(d.Bucket),
		Prefix: aws.String(s3Path),
		// NOTE(prozlach): Yes, AWS returns objects in lexicographical
		// order based on their key names for general purpose buckets.
		Delimiter: aws.String("/"),
		// NOTE(prozlach): AWS returns objects in lexicographical order
		// based on their key names for general purpose buckets, but there
		// is a catch - chars like `.` go before `/` so we need to go
		// through the list and do exact matching.
		// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
		MaxKeys: aws.Int64(1),
		})
		MaxKeys: aws.Int64(listMax),
		}
		resp, err := d.S3.ListObjectsV2WithContext(ctx, listInput)
		if err != nil {
		return nil, err
		}
		@@ -387,23 +389,68 @@ func (d *driver) Stat(ctx context.Context, path string) (storagedriver.FileInfo,
		Path: path,
		}

		if len(resp.Contents) == 1 {
		entry := resp.Contents[0]
		if *entry.Key != s3Path {
		if len(entry.Key) > len(s3Path) && (entry.Key)[len(s3Path)] != '/' {
		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: common.V1DriverName}
		}
		fi.IsDir = true
		} else {
		main:
		for {
		// Files matching prefix:
		noMoreFiles := false
		noMoreDirs := false
		loop_files:
		for _, entry := range resp.Contents {
		switch strings.Compare(*entry.Key, s3Path) {
		// NOTE(prozlach): The -1 case will never happen here as the List
		// call only gives us objects matching the prefix, and we can have
		// only two cases - file matches the prefix exactly (`0` case) or
		// file is longer (`1` case). Still - putting it here for
		// completeness.
		case -1:
		log.WithFields(log.Fields{
		"key": *entry.Key,
		}).Debugln("skipping predecessor entry as it does not match")
		continue loop_files
		case 0:
		fi.IsDir = false
		fi.Size = *entry.Size
		fi.ModTime = *entry.LastModified
		return storagedriver.FileInfoInternal{FileInfoFields: fi}, nil
		case 1:
		noMoreFiles = true
		break loop_files
		}
		} else {
		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: common.V1DriverName}
		}
		if len(resp.Contents) == 0 {
		noMoreFiles = true
		}

		// Prefix has subdirectories:
		loop_dirs:
		for _, commonPrefix := range resp.CommonPrefixes {
		switch strings.Compare(*commonPrefix.Prefix, s3Path+"/") {
		case -1:
		continue loop_dirs
		case 0:
		fi.IsDir = true
		return storagedriver.FileInfoInternal{FileInfoFields: fi}, nil
		case 1:
		noMoreDirs = true
		break loop_dirs
		}
		}
		if len(resp.CommonPrefixes) == 0 {
		noMoreDirs = true
		}

		if resp.IsTruncated == nil \|\| !*resp.IsTruncated \|\| (noMoreFiles && noMoreDirs) {
		break main
		}

		listInput.ContinuationToken = resp.NextContinuationToken
		resp, err = d.S3.ListObjectsV2WithContext(ctx, listInput)
		if err != nil {
		return nil, err
		}
		}

		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: common.V1DriverName}
		}

		// List returns a list of the objects that are direct descendants of the given path.

registry/storage/driver/s3-aws/v2/s3.go

+65 −18

Original line number	Diff line number	Diff line
		@@ -43,6 +43,7 @@ import (
		"github.com/docker/distribution/registry/storage/driver/s3-aws/common"
		"github.com/docker/distribution/version"
		"github.com/hashicorp/go-multierror"
		log "github.com/sirupsen/logrus"
		"gitlab.com/gitlab-org/labkit/fips"
		"golang.org/x/sync/errgroup"
		)
		@@ -456,41 +457,87 @@ func (d driver) statHead(ctx context.Context, path string) (storagedriver.File

		func (d driver) statList(ctx context.Context, path string) (storagedriver.FileInfoFields, error) {
		s3Path := d.s3Path(path)
		resp, err := d.S3.ListObjectsV2(
		ctx,
		&s3.ListObjectsV2Input{
		listInput := &s3.ListObjectsV2Input{
		Bucket: ptr.String(d.Bucket),
		Prefix: ptr.String(s3Path),
		// NOTE(prozlach): Yes, AWS returns objects in lexicographical
		// order based on their key names for general purpose buckets.
		Delimiter: ptr.String("/"),
		// NOTE(prozlach): AWS returns objects in lexicographical order
		// based on their key names for general purpose buckets, but there
		// is a catch - chars like `.` go before `/` so we need to go
		// through the list and do exact matching.
		// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
		MaxKeys: ptr.Int32(1),
		})
		MaxKeys: ptr.Int32(listMax),
		}
		resp, err := d.S3.ListObjectsV2(ctx, listInput)
		if err != nil {
		return nil, err
		}

		if len(resp.Contents) != 1 {
		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: common.V2DriverName}
		}

		entry := resp.Contents[0]
		fi := &storagedriver.FileInfoFields{
		Path: path,
		}

		if *entry.Key != s3Path {
		if len(entry.Key) > len(s3Path) && (entry.Key)[len(s3Path)] != '/' {
		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: common.V2DriverName}
		}
		fi.IsDir = true
		} else {
		main:
		for {
		// Files matching prefix:
		noMoreFiles := false
		noMoreDirs := false
		loop_files:
		for _, entry := range resp.Contents {
		// NOTE(prozlach): These two conditions will never be reached, as
		// the HEAD version of the stat call makes sure that the exact call
		// will be handled earlier, and the shorter string will always go
		// before longer strings (List does here a prefix-matching), so all
		// other objects will be at least longer than the s3Path one, falling
		// into `case 1` bucket. Still - I am putting it here for
		// completeness just in case something changes in the future.
		switch strings.Compare(*entry.Key, s3Path) {
		case -1:
		log.WithFields(log.Fields{
		"key": *entry.Key,
		}).Debugln("skipping predecessor entry as it does not match")
		continue loop_files
		case 0:
		log.WithFields(log.Fields{
		"key": *entry.Key,
		}).Debugln("entry matched, object found")
		fi.IsDir = false
		fi.Size = *entry.Size
		fi.ModTime = *entry.LastModified
		return fi, nil
		case 1:
		noMoreFiles = true
		break loop_files
		}
		}

		// Prefix has subdirectories:
		loop_dirs:
		for _, commonPrefix := range resp.CommonPrefixes {
		switch strings.Compare(*commonPrefix.Prefix, s3Path+"/") {
		case -1:
		continue loop_dirs
		case 0:
		fi.IsDir = true
		return fi, nil
		case 1:
		noMoreDirs = true
		break loop_dirs
		}
		}

		if resp.IsTruncated == nil \|\| !*resp.IsTruncated \|\| (noMoreFiles && noMoreDirs) {
		break main
		}

		listInput.ContinuationToken = resp.NextContinuationToken
		resp, err = d.S3.ListObjectsV2(ctx, listInput)
		if err != nil {
		return nil, err
		}
		}

		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: common.V2DriverName}
		}

		// Stat retrieves the FileInfo for the given path, including the current size

registry/storage/driver/testsuites/testsuites.go

+134 −38

Original line number	Diff line number	Diff line
		@@ -1989,41 +1989,71 @@ func (s *DriverSuite) TestDeleteOnlyDeletesSubpaths() {
		func (s *DriverSuite) TestStatCall() {
		// NOTE(prozlach): We explicitly need a different blob here to confirm that
		// in-place overwrite was indeed successful.
		// NOTE(prozlach): The idea is to create a hierarchy in the s3 bucket
		// where:
		// * there is one common directory for all the blobs (dirPathBase)
		// * there are two files in two different "subdirectories" under the same
		// common directory (dirA, dirB and filePath, filePathAux)
		// * both subdirectories should share the same prefix so that we could test
		// a stat on inexistant dir which is actually a common prefix for existing
		// subdirectories (DirPartialPrefix)
		contentAB := s.blobberFactory.GetBlobber(4096 * 2).GetAllBytes()
		contentA := contentAB[:4096]
		contentB := contentAB[4096:]
		dirPathBase := dtestutil.RandomPath(1, 24)
		dirA := "foo" + dtestutil.RandomFilename(13)
		dirB := "foo" + dtestutil.RandomFilename(13)
		partialPath := path.Join(dirPathBase, "foo")
		dirPath := path.Join(dirPathBase, dirA)
		dirPathAux := path.Join(dirPathBase, dirB)
		fileName := dtestutil.RandomFilename(32)
		filePath := path.Join(dirPath, fileName)

		fileNameBase := dtestutil.RandomFilename(32)
		dirPathBase := "/" + dtestutil.RandomFilename(24)

		dirPathPrefix := path.Join(dirPathBase, "fo")
		dirPathZero := path.Join(dirPathBase, "foo")
		dirPathLong := path.Join(dirPathBase, "fooabc")
		dirPathDot := path.Join(dirPathBase, "foo.")
		fileDirPathZero := path.Join(dirPathBase, "foo", fileNameBase)
		fileDirPathLong := path.Join(dirPathLong, fileNameBase)
		// Trigger a case where for given prefix there is more than one object
		filePathAux := path.Join(dirPathAux, fileName)
		s.T().Logf("directory: %s, filename: %s, filename aux: %s", dirPath, fileName, filePathAux)
		fileDirPathDot := path.Join(dirPathDot, fileNameBase)

		filePrefix := path.Join(dirPathBase, "bar")
		fileNotExists := path.Join(dirPathBase, "barb")
		fileZero := path.Join(dirPathBase, "bara")
		fileLong := path.Join(dirPathBase, "barabc")
		fileDot := path.Join(dirPathBase, "bar.a") // Azure shortens `bar.` to just `bar` :\|

		s.T().Logf(
		"fileNameBase: %s, dirPathBase: %s",
		fileNameBase, dirPathBase,
		)

		s.T().Logf(
		"fileZero: %s, fileLong: %s, fileDot: %s, file prefix: %s",
		fileZero, fileLong, fileDot, filePrefix,
		)

		err := s.StorageDriver.PutContent(s.ctx, filePath, contentA)
		s.T().Logf(
		"dir zero: %s, dir long: %s, dir dot: %s, dir prefix: %s",
		dirPathZero, dirPathLong, dirPathDot, dirPathPrefix,
		)

		err := s.StorageDriver.PutContent(s.ctx, fileDirPathLong, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriver.PutContent(s.ctx, fileDirPathDot, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriver.PutContent(s.ctx, fileDirPathZero, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriver.PutContent(s.ctx, fileZero, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriver.PutContent(s.ctx, filePathAux, contentA)
		err = s.StorageDriver.PutContent(s.ctx, fileLong, contentA)
		require.NoError(s.T(), err)
		defer s.deletePath(s.StorageDriver, firstPart(dirPath))
		err = s.StorageDriver.PutContent(s.ctx, fileDot, contentA)
		require.NoError(s.T(), err)
		defer s.deletePath(s.StorageDriver, dirPathBase)

		if s.StorageDriver.Name() != "filesystem" {
		err = s.StorageDriverRootless.PutContent(s.ctx, filePath, contentA)
		err = s.StorageDriverRootless.PutContent(s.ctx, fileDirPathLong, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriverRootless.PutContent(s.ctx, fileDirPathDot, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriverRootless.PutContent(s.ctx, fileDirPathZero, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriverRootless.PutContent(s.ctx, filePathAux, contentA)
		err = s.StorageDriverRootless.PutContent(s.ctx, fileZero, contentA)
		require.NoError(s.T(), err)
		defer s.deletePath(s.StorageDriverRootless, firstPart(dirPath))
		err = s.StorageDriverRootless.PutContent(s.ctx, fileLong, contentA)
		require.NoError(s.T(), err)
		err = s.StorageDriverRootless.PutContent(s.ctx, fileDot, contentA)
		require.NoError(s.T(), err)
		defer s.deletePath(s.StorageDriverRootless, dirPathBase)
		}

		// Call to stat on root directory. The storage healthcheck performs this
		@@ -2060,11 +2090,11 @@ func (s *DriverSuite) TestStatCall() {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, dirPath+"foo")
		fi, err := drv.Stat(s.ctx, dirPathLong+"bar")
		require.Error(s.T(), err)
		assert.ErrorIs(s.T(), err, storagedriver.PathNotFoundError{ // nolint: testifylint
		DriverName: drv.Name(),
		Path: dirPath + "foo",
		Path: dirPathLong + "bar",
		})
		assert.Nil(s.T(), fi)
		})
		@@ -2074,11 +2104,11 @@ func (s *DriverSuite) TestStatCall() {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, filePath+"bar")
		fi, err := drv.Stat(s.ctx, fileDirPathLong+"bar")
		require.Error(s.T(), err)
		assert.ErrorIs(s.T(), err, storagedriver.PathNotFoundError{ // nolint: testifylint
		DriverName: drv.Name(),
		Path: filePath + "bar",
		Path: fileDirPathLong + "bar",
		})
		assert.Nil(s.T(), fi)
		})
		@@ -2088,10 +2118,10 @@ func (s *DriverSuite) TestStatCall() {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, filePath)
		fi, err := drv.Stat(s.ctx, fileDirPathLong)
		require.NoError(s.T(), err)
		require.NotNil(s.T(), fi)
		assert.Equal(s.T(), filePath, fi.Path())
		assert.Equal(s.T(), fileDirPathLong, fi.Path())
		assert.Equal(s.T(), int64(len(contentA)), fi.Size())
		assert.False(s.T(), fi.IsDir())
		})
		@@ -2101,17 +2131,17 @@ func (s *DriverSuite) TestStatCall() {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, filePath)
		fi, err := drv.Stat(s.ctx, fileDirPathLong)
		require.NoError(s.T(), err)
		assert.NotNil(s.T(), fi)
		createdTime := fi.ModTime()

		// Sleep and modify the file
		time.Sleep(time.Second * 10)
		err = drv.PutContent(s.ctx, filePath, contentB)
		err = drv.PutContent(s.ctx, fileDirPathLong, contentB)
		require.NoError(s.T(), err)

		fi, err = drv.Stat(s.ctx, filePath)
		fi, err = drv.Stat(s.ctx, fileDirPathLong)
		require.NoError(s.T(), err)
		require.NotNil(s.T(), fi)
		modTime := fi.ModTime()
		@@ -2135,10 +2165,10 @@ func (s *DriverSuite) TestStatCall() {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, dirPath)
		fi, err := drv.Stat(s.ctx, dirPathLong)
		require.NoError(s.T(), err)
		require.NotNil(s.T(), fi)
		assert.Equal(s.T(), dirPath, fi.Path())
		assert.Equal(s.T(), dirPathLong, fi.Path())
		assert.Zero(s.T(), fi.Size())
		assert.True(s.T(), fi.IsDir())
		})
		@@ -2157,18 +2187,84 @@ func (s *DriverSuite) TestStatCall() {
		assert.True(s.T(), fi.IsDir())
		})

		// Call on a directory where there are other directories for which this
		// directory is a prefix. We include `.` as this test a special
		// case/issue with lexographic sorting where `.` goes before `/`.
		// The result should be a directory object/no error.
		s.Run(fmt.Sprintf("DirPartialPrefix Found - %s", name), func() {
		if s.StorageDriver.Name() == "filesystem" && name == "unprefixed" {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, dirPathZero)
		require.NoError(s.T(), err)
		require.NotNil(s.T(), fi)
		assert.Equal(s.T(), dirPathZero, fi.Path())
		assert.Zero(s.T(), fi.Size())
		assert.True(s.T(), fi.IsDir())
		})

		// Call on a partial name of the directory. This should result in
		// not-found, as partial match is still not a match for a directory.
		s.Run(fmt.Sprintf("DirPartialPrefix - %s", name), func() {
		s.Run(fmt.Sprintf("DirPartialPrefix NotFound - %s", name), func() {
		if s.StorageDriver.Name() == "filesystem" && name == "unprefixed" {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, dirPathPrefix)
		require.Error(s.T(), err)
		assert.ErrorIs(s.T(), err, storagedriver.PathNotFoundError{ // nolint: testifylint
		DriverName: drv.Name(),
		Path: dirPathPrefix,
		})
		assert.Nil(s.T(), fi)
		})

		// Call on a file where there are other files for which this
		// file is a prefix. We include `.` as this test a special
		// case/issue with lexographic sorting where `.` goes before `/`.
		// The result should be a directory object/no error.
		s.Run(fmt.Sprintf("FilePartialPrefix Found - %s", name), func() {
		if s.StorageDriver.Name() == "filesystem" && name == "unprefixed" {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, fileZero)
		require.NoError(s.T(), err)
		require.NotNil(s.T(), fi)
		assert.Equal(s.T(), fileZero, fi.Path())
		assert.EqualValues(s.T(), len(contentA), fi.Size())
		assert.False(s.T(), fi.IsDir())
		})

		// Call on a partial name of the file. This should result in
		// not-found, as partial match is still not a match.
		s.Run(fmt.Sprintf("FilePartialPrefix NotFound - %s", name), func() {
		if s.StorageDriver.Name() == "filesystem" && name == "unprefixed" {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, filePrefix)
		require.Error(s.T(), err)
		assert.ErrorIs(s.T(), err, storagedriver.PathNotFoundError{ // nolint: testifylint
		DriverName: drv.Name(),
		Path: filePrefix,
		})
		assert.Nil(s.T(), fi)
		})

		// Call on a file that is only a partial match. This should result in
		// not-found, as partial match is still not a match for a directory.
		s.Run(fmt.Sprintf("FilePartialPrefixNotTruncated NotFound - %s", name), func() {
		if s.StorageDriver.Name() == "filesystem" && name == "unprefixed" {
		s.T().Skip("filesystem driver does not support prefix-less operation")
		}

		fi, err := drv.Stat(s.ctx, partialPath)
		fi, err := drv.Stat(s.ctx, fileNotExists)
		require.Error(s.T(), err)
		assert.ErrorIs(s.T(), err, storagedriver.PathNotFoundError{ // nolint: testifylint
		DriverName: drv.Name(),
		Path: partialPath,
		Path: fileNotExists,
		})
		assert.Nil(s.T(), fi)
		})