s3: give account-less identities a distinct owner instead of admin (#9962)

* s3: stop collapsing account-less identities into the admin account

Identities configured without an account block all defaulted to the
shared admin account, so distinct users got the same owner id and
ownership checks could not tell them apart. checkAccessByOwnership also
treated that id as an admin bypass, so any account-less caller passed
ownership for any bucket. Give such identities a distinct account id from
their name, and decide the ownership admin bypass by Admin capability
rather than by the account id. isUserAdmin is now nil-safe.

* s3: use the context identity in isUserAdmin before re-authenticating

The Auth middleware already verifies and stores the identity in the
request context. Read it there first so the ownership/admin checks don't
re-run signature verification, which is redundant and fails once the
request body has been consumed.

* s3: nil-guard the context identity in isUserAdmin

A non-nil interface wrapping a typed-nil *Identity passes the type
assertion; guard against it before calling isAdmin().

* s3: trim verbose comments

weed/s3api/auth_account_collapse_test.go

@@ -0,0 +1,72 @@
+package s3api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAccountForUnscopedIdentity(t *testing.T) {
+	assert.Equal(t, "alice", accountForUnscopedIdentity("alice").Id, "a named identity gets its own account id")
+	assert.NotEqual(t, AccountAdmin.Id, accountForUnscopedIdentity("alice").Id, "a named identity must not inherit the admin account")
+	assert.Same(t, &AccountAdmin, accountForUnscopedIdentity(AccountAdmin.Id), "the conventional admin keeps the admin account")
+	assert.Same(t, &AccountAdmin, accountForUnscopedIdentity(""), "an empty name falls back to the admin account")
+}
+
+func TestUnscopedIdentitiesGetDistinctAccounts(t *testing.T) {
+	resetMemoryStore()
+
+	config := `{
+  "identities": [
+    {"name": "alice", "credentials": [{"accessKey": "alice_ak", "secretKey": "alice_sk"}], "actions": ["Read"]},
+    {"name": "admin", "credentials": [{"accessKey": "admin_ak", "secretKey": "admin_sk"}], "actions": ["Admin"]}
+  ]
+}`
+	tmp, err := os.CreateTemp("", "s3-config-*.json")
+	require.NoError(t, err)
+	defer os.Remove(tmp.Name())
+	_, err = tmp.WriteString(config)
+	require.NoError(t, err)
+	require.NoError(t, tmp.Close())
+
+	iam := NewIdentityAccessManagementWithStore(&S3ApiServerOption{Config: tmp.Name()}, nil, "memory")
+
+	alice, _, found := iam.LookupByAccessKey("alice_ak")
+	require.True(t, found)
+	require.NotNil(t, alice.Account)
+	assert.Equal(t, "alice", alice.Account.Id, "a non-admin account-less identity owns resources as itself, not as admin")
+
+	admin, _, found := iam.LookupByAccessKey("admin_ak")
+	require.True(t, found)
+	require.NotNil(t, admin.Account)
+	assert.Equal(t, AccountAdmin.Id, admin.Account.Id, "the admin identity keeps the admin account")
+}
+
+// A distinct non-owner is denied an admin-owned bucket (iam nil => isUserAdmin
+// false, so only real ownership grants access).
+func TestCheckAccessByOwnershipDeniesNonOwner(t *testing.T) {
+	adminOwner := AccountAdmin.Id
+	s3a := &S3ApiServer{
+		bucketRegistry: &BucketRegistry{
+			metadataCache: map[string]*BucketMetaData{
+				"b": {Name: "b", Owner: &s3.Owner{ID: &adminOwner}},
+			},
+			notFound: map[string]struct{}{},
+		},
+	}
+
+	nonOwner := httptest.NewRequest(http.MethodGet, "/b?ownershipControls=", nil)
+	nonOwner.Header.Set(s3_constants.AmzAccountId, "alice")
+	assert.Equal(t, s3err.ErrAccessDenied, s3a.checkAccessByOwnership(nonOwner, "b"), "a distinct non-owner is denied the admin-owned bucket")
+
+	owner := httptest.NewRequest(http.MethodGet, "/b?ownershipControls=", nil)
+	owner.Header.Set(s3_constants.AmzAccountId, AccountAdmin.Id)
+	assert.Equal(t, s3err.ErrNone, s3a.checkAccessByOwnership(owner, "b"), "the actual owner is still allowed")
+}

weed/s3api/auth_credentials.go

@@ -137,6 +137,19 @@ var (
 	}
 )
 
+// accountForUnscopedIdentity gives an identity with no configured account a
+// distinct account id from its name, so account-less identities are not all
+// collapsed into the shared admin account for ownership/ACL checks.
+func accountForUnscopedIdentity(name string) *Account {
+	if name == "" || name == AccountAdmin.Id {
+		return &AccountAdmin
+	}
+	return &Account{
+		Id:          name,
+		DisplayName: name,
+	}
+}
+
 type Credential struct {
 	AccessKey  string
 	SecretKey  string
@@ -614,7 +627,7 @@ func (iam *IdentityAccessManagement) ReplaceS3ApiConfiguration(config *iam_pb.S3
 			t.Account = &AccountAnonymous
 			identityAnonymous = t
 		case ident.Account == nil:
-			t.Account = &AccountAdmin
+			t.Account = accountForUnscopedIdentity(t.Name)
 		default:
 			if account, ok := accounts[ident.Account.Id]; ok {
 				t.Account = account
@@ -833,7 +846,7 @@ func (iam *IdentityAccessManagement) MergeS3ApiConfiguration(config *iam_pb.S3Ap
 			t.Account = &AccountAnonymous
 			identityAnonymous = t
 		case ident.Account == nil:
-			t.Account = &AccountAdmin
+			t.Account = accountForUnscopedIdentity(t.Name)
 		default:
 			if account, ok := accounts[ident.Account.Id]; ok {
 				t.Account = account

weed/s3api/s3api_acp.go

@@ -21,8 +21,12 @@ func (s3a *S3ApiServer) checkAccessByOwnership(r *http.Request, bucket string) s
 	if errCode != s3err.ErrNone {
 		return errCode
 	}
+	// Admin by capability, not account id: account-less identities share the "admin" id.
+	if s3a.isUserAdmin(r) {
+		return s3err.ErrNone
+	}
 	accountId := getAccountId(r)
-	if accountId == AccountAdmin.Id || accountId == *metadata.Owner.ID {
+	if metadata.Owner != nil && metadata.Owner.ID != nil && accountId == *metadata.Owner.ID {
 		return s3err.ErrNone
 	}
 	return s3err.ErrAccessDenied

weed/s3api/s3api_bucket_handlers.go

@@ -672,6 +672,16 @@ func (s3a *S3ApiServer) hasAccess(r *http.Request, entry *filer_pb.Entry) bool {
 // isUserAdmin securely checks if the authenticated user is an admin
 // This validates admin status through proper IAM authentication, not spoofable headers
 func (s3a *S3ApiServer) isUserAdmin(r *http.Request) bool {
+	// Reuse the identity the Auth middleware stored; re-authenticating here would
+	// fail once the request body has been read.
+	if identityObj := s3_constants.GetIdentityFromContext(r); identityObj != nil {
+		if identity, ok := identityObj.(*Identity); ok {
+			return identity != nil && identity.isAdmin()
+		}
+	}
+	if s3a.iam == nil {
+		return false
+	}
 	// Use a minimal admin action to authenticate and check admin status
 	adminAction := Action("Admin")
 	identity, errCode := s3a.iam.authRequest(r, adminAction)