Browse Source
Enforce IAM for S3 Tables bucket creation (#8388)
Enforce IAM for S3 Tables bucket creation (#8388)
* Enforce IAM for s3tables bucket creation * Prefer IAM path when policies exist * Ensure IAM enforcement honors default allow * address comments * Reused the precomputed principal when setting tableBucketMetadata.OwnerAccountID, avoiding the redundant getAccountID call. * get identity * fix * dedup * fix * comments * fix tests * update iam config * go fmt * fix ports * fix flags * mini clean shutdown * Revert "update iam config" This reverts commitpull/7183/mergeca48fdbb0a. Revert "mini clean shutdown" This reverts commit9e17f6baff. Revert "fix flags" This reverts commite9e7b29d2f. Revert "go fmt" This reverts commitbd3241960b. * test/s3tables: share single weed mini per test package via TestMain Previously each top-level test function in the catalog and s3tables package started and stopped its own weed mini instance. This caused failures when a prior instance wasn't cleanly stopped before the next one started (port conflicts, leaked global state). Changes: - catalog/iceberg_catalog_test.go: introduce TestMain that starts one shared TestEnvironment (external weed binary) before all tests and tears it down after. All individual test functions now use sharedEnv. Added randomSuffix() for unique resource names across tests. - catalog/pyiceberg_test.go: updated to use sharedEnv instead of per-test environments. - catalog/pyiceberg_test_helpers.go -> pyiceberg_test_helpers_test.go: renamed to a _test.go file so it can access TestEnvironment which is defined in a test file. - table-buckets/setup.go: add package-level sharedCluster variable. - table-buckets/s3tables_integration_test.go: introduce TestMain that starts one shared TestCluster before all tests. TestS3TablesIntegration now uses sharedCluster. Extract startMiniClusterInDir (no *testing.T) for TestMain use. TestS3TablesCreateBucketIAMPolicy keeps its own cluster (different IAM config). Remove miniClusterMutex (no longer needed). Fix Stop() to not panic when t is nil." * delete * parse * default allow should work with anonymous * fix port * iceberg route The failures are from Iceberg REST using the default bucket warehouse when no prefix is provided. Your tests create random buckets, so /v1/namespaces was looking in warehouse and failing. I updated the tests to use the prefixed Iceberg routes (/v1/{bucket}/...) via a small helper. * test(s3tables): fix port conflicts and IAM ARN matching in integration tests - Pass -master.dir explicitly to prevent filer store directory collision between shared cluster and per-test clusters running in the same process - Pass -volume.port.public and -volume.publicUrl to prevent the global publicPort flag (mutated from 0 → concrete port by first cluster) from being reused by a second cluster, causing 'address already in use' - Remove the flag-reset loop in Stop() that reset global flag values while other goroutines were reading them (race → panic) - Fix IAM policy Resource ARN in TestS3TablesCreateBucketIAMPolicy to use wildcards (arn:aws:s3tables:*:*:bucket/<name>) because the handler generates ARNs with its own DefaultRegion (us-east-1) and principal name ('admin'), not the test constants testRegion/testAccountID
Chris Lu
15 hours ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 671 additions and 172 deletions
-
210test/s3tables/catalog/iceberg_catalog_test.go
-
15test/s3tables/catalog/pyiceberg_test.go
-
0test/s3tables/catalog/pyiceberg_test_helpers_test.go
-
225test/s3tables/table-buckets/s3tables_integration_test.go
-
14test/s3tables/table-buckets/setup.go
-
39weed/s3api/identity_reflection_test.go
-
8weed/s3api/s3api_tables.go
-
21weed/s3api/s3tables/handler.go
-
45weed/s3api/s3tables/handler_bucket_create.go
-
266weed/s3api/s3tables/iam.go
@ -0,0 +1,39 @@ |
|||
package s3api |
|||
|
|||
import ( |
|||
"reflect" |
|||
"testing" |
|||
) |
|||
|
|||
// TestIdentityFieldsForS3TablesReflection ensures the identity struct keeps the
|
|||
// fields relied on by s3tables.getIdentityPrincipalArn, getIdentityPolicyNames,
|
|||
// and getIdentityClaims via reflection.
|
|||
func TestIdentityFieldsForS3TablesReflection(t *testing.T) { |
|||
typ := reflect.TypeOf(Identity{}) |
|||
checkField(t, typ, "PrincipalArn", reflect.String) |
|||
field, ok := typ.FieldByName("PolicyNames") |
|||
if !ok { |
|||
t.Fatalf("Identity.PolicyNames missing") |
|||
} |
|||
if field.Type.Kind() != reflect.Slice { |
|||
t.Fatalf("Identity.PolicyNames must be a slice, got %s", field.Type.Kind()) |
|||
} |
|||
field, ok = typ.FieldByName("Claims") |
|||
if !ok { |
|||
t.Fatalf("Identity.Claims missing") |
|||
} |
|||
if field.Type.Kind() != reflect.Map || field.Type.Key().Kind() != reflect.String { |
|||
t.Fatalf("Identity.Claims must be map[string]..., got %s/%s", field.Type.Kind(), field.Type.Key().Kind()) |
|||
} |
|||
} |
|||
|
|||
func checkField(t *testing.T, typ reflect.Type, name string, kind reflect.Kind) { |
|||
t.Helper() |
|||
field, ok := typ.FieldByName(name) |
|||
if !ok { |
|||
t.Fatalf("Identity.%s missing", name) |
|||
} |
|||
if field.Type.Kind() != kind { |
|||
t.Fatalf("Identity.%s must be %s, got %s", name, kind, field.Type.Kind()) |
|||
} |
|||
} |
|||
@ -0,0 +1,266 @@ |
|||
package s3tables |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"net/http" |
|||
"reflect" |
|||
"strings" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/glog" |
|||
"github.com/seaweedfs/seaweedfs/weed/iam/integration" |
|||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" |
|||
) |
|||
|
|||
// IAMAuthorizer allows s3tables handlers to evaluate IAM policies without importing s3api.
|
|||
type IAMAuthorizer interface { |
|||
IsActionAllowed(ctx context.Context, request *integration.ActionRequest) (bool, error) |
|||
} |
|||
|
|||
// SetIAMAuthorizer injects the IAM authorizer for policy-based access checks.
|
|||
func (h *S3TablesHandler) SetIAMAuthorizer(authorizer IAMAuthorizer) { |
|||
h.iamAuthorizer = authorizer |
|||
} |
|||
|
|||
func (h *S3TablesHandler) shouldUseIAM(r *http.Request, identityActions, identityPolicyNames []string) bool { |
|||
if h.iamAuthorizer == nil || r == nil { |
|||
return false |
|||
} |
|||
if s3_constants.GetIdentityFromContext(r) == nil { |
|||
return false |
|||
} |
|||
// When default-allow is enabled, keep anonymous requests on the legacy path
|
|||
// to preserve zero-config behavior (IAM policies are not available for anonymous).
|
|||
if h.defaultAllow && isAnonymousIdentity(r) { |
|||
return false |
|||
} |
|||
// An empty inline `identityActions` slice doesn't mean the identity has no
|
|||
// permissions—it just means authorization lives in IAM policies or session
|
|||
// tokens instead of static action lists. We therefore prefer the IAM path
|
|||
// whenever inline actions are absent and fall back to default policy names
|
|||
// or session tokens.
|
|||
if hasSessionToken(r) { |
|||
return true |
|||
} |
|||
if len(identityActions) == 0 { |
|||
return true |
|||
} |
|||
return len(identityPolicyNames) > 0 |
|||
} |
|||
|
|||
func isAnonymousIdentity(r *http.Request) bool { |
|||
val, ok := getIdentityStructValue(r) |
|||
if !ok { |
|||
return false |
|||
} |
|||
if nameField := val.FieldByName("Name"); nameField.IsValid() && nameField.Kind() == reflect.String { |
|||
if nameField.String() == s3_constants.AccountAnonymousId { |
|||
return true |
|||
} |
|||
} |
|||
accountField := val.FieldByName("Account") |
|||
if accountField.IsValid() && !accountField.IsNil() { |
|||
if accountField.Kind() == reflect.Ptr { |
|||
accountField = accountField.Elem() |
|||
} |
|||
if accountField.Kind() == reflect.Struct { |
|||
if idField := accountField.FieldByName("Id"); idField.IsValid() && idField.Kind() == reflect.String { |
|||
if idField.String() == s3_constants.AccountAnonymousId { |
|||
return true |
|||
} |
|||
} |
|||
} |
|||
} |
|||
return false |
|||
} |
|||
|
|||
func hasSessionToken(r *http.Request) bool { |
|||
return extractSessionToken(r) != "" |
|||
} |
|||
|
|||
func extractSessionToken(r *http.Request) string { |
|||
if token := r.Header.Get("X-SeaweedFS-Session-Token"); token != "" { |
|||
return token |
|||
} |
|||
if token := r.Header.Get("X-Amz-Security-Token"); token != "" { |
|||
return token |
|||
} |
|||
return r.URL.Query().Get("X-Amz-Security-Token") |
|||
} |
|||
|
|||
func (h *S3TablesHandler) authorizeIAMAction(r *http.Request, identityPolicyNames []string, action string, resources ...string) (bool, error) { |
|||
if h.iamAuthorizer == nil { |
|||
err := fmt.Errorf("nil iamAuthorizer in authorizeIAMAction") |
|||
glog.V(2).Infof("S3Tables: %v", err) |
|||
return false, err |
|||
} |
|||
principal := r.Header.Get("X-SeaweedFS-Principal") |
|||
if principal == "" { |
|||
principal = getIdentityPrincipalArn(r) |
|||
} |
|||
if principal == "" { |
|||
return false, fmt.Errorf("missing principal for IAM authorization") |
|||
} |
|||
|
|||
if !strings.Contains(action, ":") { |
|||
action = "s3tables:" + action |
|||
} |
|||
|
|||
sessionToken := extractSessionToken(r) |
|||
|
|||
requestContext := buildIAMRequestContext(r, getIdentityClaims(r)) |
|||
policyNames := identityPolicyNames |
|||
if len(policyNames) == 0 { |
|||
policyNames = getIdentityPolicyNames(r) |
|||
} |
|||
|
|||
if len(resources) == 0 { |
|||
return false, fmt.Errorf("no resources provided to authorizeIAMAction") |
|||
} |
|||
checkedResource := false |
|||
for _, resource := range resources { |
|||
if resource == "" { |
|||
continue |
|||
} |
|||
checkedResource = true |
|||
allowed, err := h.iamAuthorizer.IsActionAllowed(r.Context(), &integration.ActionRequest{ |
|||
Principal: principal, |
|||
Action: action, |
|||
Resource: resource, |
|||
SessionToken: sessionToken, |
|||
RequestContext: requestContext, |
|||
PolicyNames: policyNames, |
|||
}) |
|||
if err != nil { |
|||
glog.V(2).Infof("S3Tables: IAM authorization error action=%s resource=%s principal=%s: %v", action, resource, principal, err) |
|||
return false, err |
|||
} |
|||
if !allowed { |
|||
err := fmt.Errorf("access denied by IAM for resource %s", resource) |
|||
return false, err |
|||
} |
|||
} |
|||
if !checkedResource { |
|||
return false, fmt.Errorf("no non-empty resources provided to authorizeIAMAction") |
|||
} |
|||
return true, nil |
|||
} |
|||
|
|||
func getIdentityPrincipalArn(r *http.Request) string { |
|||
val, ok := getIdentityStructValue(r) |
|||
if !ok { |
|||
return "" |
|||
} |
|||
field := val.FieldByName("PrincipalArn") |
|||
if field.IsValid() && field.Kind() == reflect.String { |
|||
return field.String() |
|||
} |
|||
return "" |
|||
} |
|||
|
|||
func getIdentityPolicyNames(r *http.Request) []string { |
|||
val, ok := getIdentityStructValue(r) |
|||
if !ok { |
|||
return nil |
|||
} |
|||
field := val.FieldByName("PolicyNames") |
|||
if !field.IsValid() || field.Kind() != reflect.Slice { |
|||
return nil |
|||
} |
|||
policies := make([]string, 0, field.Len()) |
|||
for i := 0; i < field.Len(); i++ { |
|||
item := field.Index(i) |
|||
if item.Kind() == reflect.String { |
|||
policies = append(policies, item.String()) |
|||
} else if item.CanInterface() { |
|||
policies = append(policies, fmt.Sprint(item.Interface())) |
|||
} |
|||
} |
|||
if len(policies) == 0 { |
|||
return nil |
|||
} |
|||
return policies |
|||
} |
|||
|
|||
func getIdentityClaims(r *http.Request) map[string]interface{} { |
|||
val, ok := getIdentityStructValue(r) |
|||
if !ok { |
|||
return nil |
|||
} |
|||
field := val.FieldByName("Claims") |
|||
if !field.IsValid() || field.Kind() != reflect.Map || field.IsNil() { |
|||
return nil |
|||
} |
|||
if field.Type().Key().Kind() != reflect.String { |
|||
return nil |
|||
} |
|||
claims := make(map[string]interface{}, field.Len()) |
|||
for _, key := range field.MapKeys() { |
|||
if key.Kind() != reflect.String { |
|||
continue |
|||
} |
|||
val := field.MapIndex(key) |
|||
if !val.IsValid() { |
|||
continue |
|||
} |
|||
claims[key.String()] = val.Interface() |
|||
} |
|||
if len(claims) == 0 { |
|||
return nil |
|||
} |
|||
return claims |
|||
} |
|||
|
|||
func buildIAMRequestContext(r *http.Request, claims map[string]interface{}) map[string]interface{} { |
|||
ctx := make(map[string]interface{}) |
|||
if ua := r.Header.Get("User-Agent"); ua != "" { |
|||
ctx["userAgent"] = ua |
|||
} |
|||
if referer := r.Header.Get("Referer"); referer != "" { |
|||
ctx["referer"] = referer |
|||
} |
|||
for k, v := range claims { |
|||
if strings.HasPrefix(k, "jwt:") { |
|||
if _, exists := ctx[k]; !exists { |
|||
ctx[k] = v |
|||
} |
|||
} |
|||
} |
|||
for k, v := range claims { |
|||
if strings.HasPrefix(k, "jwt:") { |
|||
continue |
|||
} |
|||
if _, exists := ctx[k]; !exists { |
|||
ctx[k] = v |
|||
} |
|||
jwtKey := "jwt:" + k |
|||
if _, exists := ctx[jwtKey]; !exists { |
|||
ctx[jwtKey] = v |
|||
} |
|||
} |
|||
if len(ctx) == 0 { |
|||
return nil |
|||
} |
|||
return ctx |
|||
} |
|||
|
|||
// getIdentityStructValue fetches the identity struct held in the request context.
|
|||
// The identity is expected to be a pointer to a struct with the fields used by
|
|||
// the reflection helpers (PrincipalArn string, PolicyNames []string,
|
|||
// Claims map[string]interface{}).
|
|||
// This helper centralizes the nil-check and ptr-deref logic so callers focus on
|
|||
// reading the specific fields they need.
|
|||
func getIdentityStructValue(r *http.Request) (reflect.Value, bool) { |
|||
identityRaw := s3_constants.GetIdentityFromContext(r) |
|||
if identityRaw == nil { |
|||
return reflect.Value{}, false |
|||
} |
|||
val := reflect.ValueOf(identityRaw) |
|||
if val.Kind() == reflect.Ptr { |
|||
val = val.Elem() |
|||
} |
|||
if val.Kind() != reflect.Struct { |
|||
return reflect.Value{}, false |
|||
} |
|||
return val, true |
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue