s3: keep a listing's start position inside the requested prefix (#11493)

* s3: a list marker that sorts past the prefix leaves nothing to list

AWS scopes a listing to keys under Prefix; StartAfter, Marker and
continuation tokens only reposition inside that range. A marker that
diverges from the prefix at a larger byte is after every key the prefix
can match, so the page is empty. normalizePrefixMarker used to keep such
a marker as the walk cutoff at the bucket root, where the walk descends
into the marker's own directory and returns keys the prefix never names.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: keep the listing variant's action when a prefix is promoted to object

authRequestWithAuthType promotes ?prefix= into the object argument for
the legacy CanDo path. ResolveS3Action treats a non-empty object as
object-level, so a bucket-level ?versions or ?uploads request carrying a
prefix missed its specific action and fell back to the base List action:
an s3:ListBucket grant then covered s3:ListBucketVersions, and an
explicit Deny on the specific action was skipped on the same path.

Resolve the action against the same bucket-level object the resource
ARN already uses.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* s3: treat GET ?uploads as a bucket listing for authorization

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* Update weed/s3api/auth_credentials.go

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

weed/s3api/auth_credentials.go

@@ -1753,6 +1753,23 @@ func (iam *IdentityAccessManagement) authenticateRequestInternal(r *http.Request
 	return identity, s3Err, reqAuthType
 }
 
+// isBucketListingRequest reports whether the request lists bucket contents:
+// the ACTION_LIST routes, and ListMultipartUploads (GET ?uploads), which the
+// router registers under ACTION_READ without an object-path constraint.
+// GET /bucket/key?uploads therefore also reaches the listing handler. Route
+// vars are used so a prefix promoted into the authorization object does not
+// hide the bucket-level shape.
+func isBucketListingRequest(r *http.Request, action Action) bool {
+	if action == s3_constants.ACTION_LIST {
+		return true
+	}
+	if r.Method != http.MethodGet || r.URL == nil || !r.URL.Query().Has("uploads") {
+		return false
+	}
+	_, object := s3_constants.GetBucketAndObject(r)
+	return object == "" || object == "/"
+}
+
 // authRequestWithAuthType authenticates and then authorizes a request for a given action.
 func (iam *IdentityAccessManagement) authRequestWithAuthType(r *http.Request, action Action) (*Identity, s3err.ErrorCode, authType) {
 	identity, s3Err, reqAuthType := iam.authenticateRequestInternal(r)
@@ -1810,11 +1827,11 @@ func (iam *IdentityAccessManagement) authRequestWithAuthType(r *http.Request, ac
 			if identity != nil {
 				claims = identity.Claims
 			}
-			// List is bucket-level; the prefix promoted into object (for the
-			// legacy CanDo path) must not scope the resource ARN. Prefix is
-			// matched via the s3:prefix Condition.
+			// Bucket listings are bucket-level; the prefix promoted into
+			// object (for the legacy CanDo path) must not scope the resource
+			// ARN. Prefix is matched via the s3:prefix Condition.
 			policyObject := object
-			if action == s3_constants.ACTION_LIST {
+			if isBucketListingRequest(r, action) {
 				policyObject = ""
 			}
 			allowed, evaluated, err := iam.policyEngine.EvaluatePolicy(bucket, policyObject, string(action), principal, r, claims, nil)
@@ -2517,11 +2534,12 @@ func (iam *IdentityAccessManagement) evaluateAttachedIAMPolicies(r *http.Request
 		return attachedIAMPolicyNoMatch
 	}
 
-	// List is bucket-level; the prefix promoted into object (for the legacy
-	// CanDo path) must not scope the resource ARN or the resolved action
-	// (e.g. ListBucketVersions on ?versions). Prefix is matched via s3:prefix.
+	// Bucket listings are bucket-level; the prefix promoted into object (for
+	// the legacy CanDo path) must not scope the resource ARN or the resolved
+	// action (e.g. ListBucketVersions on ?versions). Prefix is matched via
+	// s3:prefix.
 	resourceObject := object
-	if action == s3_constants.ACTION_LIST {
+	if isBucketListingRequest(r, action) {
 		resourceObject = ""
 	}
 	resource := buildResourceARN(bucket, resourceObject)
@@ -3021,11 +3039,11 @@ func (iam *IdentityAccessManagement) authorizeWithIAM(r *http.Request, identity
 // check evaluates the same action and resource ARN as the policy engine.
 func resolveS3AuthTarget(action Action, bucket, object string, r *http.Request) (s3Action, resourceArn string) {
 	resourceObjectKey := object
-	if action == s3_constants.ACTION_LIST {
+	if isBucketListingRequest(r, action) {
 		resourceObjectKey = ""
 	}
 	resourceArn = buildS3ResourceArn(bucket, resourceObjectKey)
-	s3Action = ResolveS3Action(r, string(action), bucket, object)
+	s3Action = ResolveS3Action(r, string(action), bucket, resourceObjectKey)
 	return
 }
 

weed/s3api/iam_list_prefix_regression_test.go

@@ -1,11 +1,17 @@
 package s3api
 
 import (
+	"context"
 	"encoding/json"
 	"net/http/httptest"
 	"testing"
+	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/iam/integration"
+	"github.com/seaweedfs/seaweedfs/weed/iam/policy"
+	"github.com/seaweedfs/seaweedfs/weed/iam/sts"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
 	"github.com/stretchr/testify/require"
 )
 
@@ -106,6 +112,152 @@ func TestEvaluateIAMPolicies_ListBucketVersionsWithPrefix(t *testing.T) {
 		"s3:ListBucketVersions must still resolve when listing with a prefix")
 }
 
+// GET ?uploads routes under ACTION_READ, not ACTION_LIST, but it is a bucket
+// listing: a promoted prefix must not resolve it to s3:GetObject, and only
+// s3:ListBucketMultipartUploads on the bucket ARN may allow it.
+func TestEvaluateIAMPolicies_ListMultipartUploadsWithPrefix(t *testing.T) {
+	const bucket = "test-bucket"
+
+	iam := &IdentityAccessManagement{}
+	require.NoError(t, iam.PutPolicy("list-uploads", mustPolicy(t, map[string]any{
+		"Version": "2012-10-17",
+		"Statement": []map[string]any{{
+			"Effect":   "Allow",
+			"Action":   "s3:ListBucketMultipartUploads",
+			"Resource": "arn:aws:s3:::" + bucket,
+		}},
+	})))
+	require.NoError(t, iam.PutPolicy("get-object", mustPolicy(t, map[string]any{
+		"Version": "2012-10-17",
+		"Statement": []map[string]any{{
+			"Effect":   "Allow",
+			"Action":   "s3:GetObject",
+			"Resource": "arn:aws:s3:::" + bucket + "/*",
+		}},
+	})))
+
+	uploadsReader := &Identity{
+		Name:        "dave",
+		Account:     &AccountAdmin,
+		PolicyNames: []string{"list-uploads"},
+		Credentials: []*Credential{{AccessKey: "AKIAEXAMPLE", SecretKey: "secret"}},
+	}
+	getReader := &Identity{
+		Name:        "erin",
+		Account:     &AccountAdmin,
+		PolicyNames: []string{"get-object"},
+		Credentials: []*Credential{{AccessKey: "AKIAEXAMPLE", SecretKey: "secret"}},
+	}
+
+	r := httptest.NewRequest("GET", "/"+bucket+"?uploads&prefix=uploads/", nil)
+	require.True(t, iam.evaluateIAMPolicies(r, uploadsReader, s3_constants.ACTION_READ, bucket, "uploads/"),
+		"s3:ListBucketMultipartUploads on the bucket must allow the uploads listing")
+	require.False(t, iam.evaluateIAMPolicies(r, getReader, s3_constants.ACTION_READ, bucket, "uploads/"),
+		"s3:GetObject must not satisfy the uploads listing")
+}
+
+// The IAM-integration authorizer must resolve the listing variant the same way
+// evaluateIAMPolicies does: a prefix promoted into the object argument is not
+// part of the request URL, so ?versions&prefix=... still resolves to
+// s3:ListBucketVersions and an s3:ListBucket grant must not cover it.
+func TestAuthorizeAction_ListVersionsWithPromotedPrefix(t *testing.T) {
+	const bucket = "test-bucket"
+
+	ctx := context.Background()
+	iamManager := integration.NewIAMManager()
+	require.NoError(t, iamManager.Initialize(&integration.IAMConfig{
+		STS: &sts.STSConfig{
+			TokenDuration:    sts.FlexibleDuration{Duration: time.Hour},
+			MaxSessionLength: sts.FlexibleDuration{Duration: 12 * time.Hour},
+			Issuer:           "test-sts",
+			SigningKey:       []byte("test-signing-key-32-characters-long"),
+		},
+		Policy: &policy.PolicyEngineConfig{DefaultEffect: "Deny", StoreType: "memory"},
+		Roles:  &integration.RoleStoreConfig{StoreType: "memory"},
+	}, func() string { return "localhost:8888" }))
+
+	require.NoError(t, iamManager.CreatePolicy(ctx, "", "ListBucketOnly", &policy.PolicyDocument{
+		Version: "2012-10-17",
+		Statement: []policy.Statement{{
+			Effect:    "Allow",
+			Action:    []string{"s3:ListBucket"},
+			Resource:  []string{"arn:aws:s3:::" + bucket},
+			Condition: map[string]map[string]interface{}{"StringLike": {"s3:prefix": "b/*"}},
+		}},
+	}))
+	require.NoError(t, iamManager.CreatePolicy(ctx, "", "ListVersionsOnly", &policy.PolicyDocument{
+		Version: "2012-10-17",
+		Statement: []policy.Statement{{
+			Effect:   "Allow",
+			Action:   []string{"s3:ListBucketVersions"},
+			Resource: []string{"arn:aws:s3:::" + bucket},
+		}},
+	}))
+	require.NoError(t, iamManager.CreatePolicy(ctx, "", "ListUploadsOnly", &policy.PolicyDocument{
+		Version: "2012-10-17",
+		Statement: []policy.Statement{{
+			Effect:   "Allow",
+			Action:   []string{"s3:ListBucketMultipartUploads"},
+			Resource: []string{"arn:aws:s3:::" + bucket},
+		}},
+	}))
+	require.NoError(t, iamManager.CreatePolicy(ctx, "", "GetObjectOnly", &policy.PolicyDocument{
+		Version: "2012-10-17",
+		Statement: []policy.Statement{{
+			Effect:   "Allow",
+			Action:   []string{"s3:GetObject"},
+			Resource: []string{"arn:aws:s3:::" + bucket + "/*"},
+		}},
+	}))
+
+	s3iam := NewS3IAMIntegration(iamManager, "localhost:8888")
+	reader := &IAMIdentity{
+		Name:        "reader",
+		Principal:   "arn:aws:iam::000000000000:user/reader",
+		PolicyNames: []string{"ListBucketOnly"},
+	}
+
+	versionsReq := httptest.NewRequest("GET", "/"+bucket+"?versions&prefix=b/", nil)
+	// object carries the promoted prefix, matching authRequestWithAuthType
+	require.Equal(t, s3err.ErrAccessDenied,
+		s3iam.AuthorizeAction(ctx, reader, s3_constants.ACTION_LIST, bucket, "b/", versionsReq),
+		"an s3:ListBucket grant must not cover ?versions listing")
+
+	listReq := httptest.NewRequest("GET", "/"+bucket+"?list-type=2&prefix=b/", nil)
+	require.Equal(t, s3err.ErrNone,
+		s3iam.AuthorizeAction(ctx, reader, s3_constants.ACTION_LIST, bucket, "b/", listReq),
+		"s3:ListBucket with a matching s3:prefix still lists")
+
+	versionsReader := &IAMIdentity{
+		Name:        "versions-reader",
+		Principal:   "arn:aws:iam::000000000000:user/versions-reader",
+		PolicyNames: []string{"ListVersionsOnly"},
+	}
+	require.Equal(t, s3err.ErrNone,
+		s3iam.AuthorizeAction(ctx, versionsReader, s3_constants.ACTION_LIST, bucket, "b/", versionsReq),
+		"s3:ListBucketVersions allows the versions listing")
+
+	// GET ?uploads routes under ACTION_READ; a promoted prefix must still
+	// resolve it to s3:ListBucketMultipartUploads, not s3:GetObject.
+	uploadsReq := httptest.NewRequest("GET", "/"+bucket+"?uploads&prefix=b/", nil)
+	uploadsReader := &IAMIdentity{
+		Name:        "uploads-reader",
+		Principal:   "arn:aws:iam::000000000000:user/uploads-reader",
+		PolicyNames: []string{"ListUploadsOnly"},
+	}
+	require.Equal(t, s3err.ErrNone,
+		s3iam.AuthorizeAction(ctx, uploadsReader, s3_constants.ACTION_READ, bucket, "b/", uploadsReq),
+		"s3:ListBucketMultipartUploads allows the uploads listing")
+	getReader := &IAMIdentity{
+		Name:        "get-reader",
+		Principal:   "arn:aws:iam::000000000000:user/get-reader",
+		PolicyNames: []string{"GetObjectOnly"},
+	}
+	require.Equal(t, s3err.ErrAccessDenied,
+		s3iam.AuthorizeAction(ctx, getReader, s3_constants.ACTION_READ, bucket, "b/", uploadsReq),
+		"s3:GetObject must not satisfy the uploads listing")
+}
+
 func mustPolicy(t *testing.T, doc map[string]any) string {
 	t.Helper()
 	b, err := json.Marshal(doc)

weed/s3api/s3_iam_middleware.go

@@ -222,7 +222,7 @@ func (s3iam *S3IAMIntegration) AuthorizeAction(ctx context.Context, identity *IA
 	// resource ARN stays at bucket level (matching AWS ListBucket semantics).
 	// See https://github.com/seaweedfs/seaweedfs/issues/8969
 	resourceObjectKey := objectKey
-	if action == "List" {
+	if isBucketListingRequest(r, action) {
 		listPrefix := r.URL.Query().Get("prefix")
 		if listPrefix != "" {
 			requestContext["s3:prefix"] = listPrefix
@@ -258,8 +258,10 @@ func (s3iam *S3IAMIntegration) AuthorizeAction(ctx context.Context, identity *IA
 		}
 	}
 
-	// Determine the specific S3 action based on the HTTP request details
-	specificAction := ResolveS3Action(r, string(action), bucket, objectKey)
+	// Determine the specific S3 action based on the HTTP request details. The
+	// prefix promoted into objectKey is not part of the URL; resolve against
+	// the bucket-level object so ?versions keeps its own action.
+	specificAction := ResolveS3Action(r, string(action), bucket, resourceObjectKey)
 
 	// Create action request
 	actionRequest := &integration.ActionRequest{

weed/s3api/s3api_object_handlers_list.go

@@ -813,6 +813,19 @@ func markerSortsBeforePrefix(prefix, marker string) bool {
 	return !strings.HasPrefix(marker, prefix) && marker < prefix
 }
 
+// markerSortsPastPrefix reports whether marker lies beyond the last key the
+// prefix can match. A marker that diverges from the prefix at a larger byte
+// is after every key under it, so the listing is empty no matter where the
+// walk would resume.
+func markerSortsPastPrefix(prefix, marker string) bool {
+	prefix = strings.TrimLeft(prefix, "/")
+	marker = strings.TrimLeft(marker, "/")
+	if marker == "" || prefix == "" {
+		return false
+	}
+	return !strings.HasPrefix(marker, prefix) && marker > prefix
+}
+
 // the prefix and marker may be in different directories
 // normalizePrefixMarker ensures the prefix and marker both starts from the same directory.
 // prefixEndsOnDelimiter tells the walk that the prefix names one directory, whose own key
@@ -912,6 +925,13 @@ func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.
 		return // Don't set isTruncated here - let caller decide based on whether more entries exist
 	}
 
+	// A marker past the prefix's range leaves nothing under the prefix to
+	// resume at, and descending into the marker's own directory below would
+	// drop the prefix filter entirely.
+	if markerSortsPastPrefix(prefix, marker) {
+		return
+	}
+
 	if strings.Contains(marker, "/") {
 		subDir, subMarker := toParentAndDescendants(marker)
 		// println("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker)

weed/s3api/s3api_object_handlers_list_marker_before_prefix_test.go

@@ -36,6 +36,75 @@ func Test_markerSortsBeforePrefix(t *testing.T) {
 	}
 }
 
+// A marker that sorts past the prefix leaves no key under the prefix to resume
+// at, so the listing is empty instead of continuing into later prefixes.
+func Test_markerSortsPastPrefix(t *testing.T) {
+	tests := []struct {
+		name   string
+		prefix string
+		marker string
+		want   bool
+	}{
+		{"marker is a later directory", "b/", "c/", true},
+		{"marker is a later key without a slash", "b/", "c", true},
+		{"marker shares the parent dir but sorts past", "data/a", "data/b", true},
+		{"marker diverges after the prefix", "b/", "b0/x", true},
+		{"leading slashes are ignored on both sides", "/b/", "/c/", true},
+		{"empty marker is in range", "b/", "", false},
+		{"empty prefix: every key is in scope", "", "c/", false},
+		{"marker under the prefix resumes inside it", "b/", "b/allowed-1", false},
+		{"marker equal to the prefix stands", "b/", "b/", false},
+		{"partial name prefix: marker under the match set stands", "parent", "parentDir/data/0e", false},
+		{"marker before the prefix is not past it", "b/", "a/", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, markerSortsPastPrefix(tt.prefix, tt.marker), "markerSortsPastPrefix(%q, %q)", tt.prefix, tt.marker)
+		})
+	}
+}
+
+// TestListWithMarkerPastPrefix walks the listing the way listFilerEntries does
+// for a start position past the requested prefix. normalizePrefixMarker hands
+// such a marker through at the bucket root, where the walk's marker descent
+// would otherwise drop the prefix filter and continue into a later prefix.
+func TestListWithMarkerPastPrefix(t *testing.T) {
+	client := &testFilerClient{
+		entriesByDir: map[string][]*filer_pb.Entry{
+			"/buckets/test":     {newDir("a"), newDir("b"), newDir("c")},
+			"/buckets/test/b":   {{Name: "allowed-1", Attributes: &filer_pb.FuseAttributes{}}},
+			"/buckets/test/c":   {{Name: "other-2", Attributes: &filer_pb.FuseAttributes{}}},
+			"/buckets/test/c/z": {{Name: "deep", Attributes: &filer_pb.FuseAttributes{}}},
+		},
+	}
+
+	for _, tt := range []struct {
+		name   string
+		marker string
+		want   []string
+	}{
+		{"marker is a later directory", "c/", nil},
+		{"marker is a deeper path in a later directory", "c/z", nil},
+		{"marker is a later bare key", "c", nil},
+		{"marker inside the prefix resumes normally", "b/", []string{"allowed-1"}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			requestDir, entryPrefix, entryMarker, prefixEndsOnDelimiter := normalizePrefixMarker("b/", tt.marker)
+			dir := "/buckets/test"
+			if requestDir != "" {
+				dir += "/" + requestDir
+			}
+			seen := listedNames(t, client, listDirectoryRequest{
+				dir:    dir,
+				prefix: entryPrefix,
+				marker: entryMarker,
+				bucket: "test",
+			}, &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: prefixEndsOnDelimiter})
+			assert.Equal(t, tt.want, seen)
+		})
+	}
+}
+
 // A marker ending on the delimiter is trimmed to a shorter cutoff for the walk, which no
 // longer excludes the key the client named, so that key is skipped as it streams.
 func Test_excludedMarkerKey(t *testing.T) {