YGoetschel
3 weeks ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
50 changed files with 1737 additions and 441 deletions
-
30.github/workflows/go.yml
-
2test/kafka/kafka-client-loadtest/tools/pom.xml
-
213test/s3tables/sts_integration/sts_integration_test.go
-
18test/volume_server/grpc/scrub_query_test.go
-
5weed/admin/dash/admin_data.go
-
5weed/admin/dash/service_account_management.go
-
4weed/admin/view/app/service_accounts.templ
-
4weed/admin/view/app/service_accounts_templ.go
-
126weed/iam/integration/iam_integration_test.go
-
168weed/iam/integration/iam_manager.go
-
62weed/iam/policy/policy_engine.go
-
9weed/iam/sts/session_claims.go
-
3weed/iam/sts/session_claims_test.go
-
35weed/iam/sts/session_policy.go
-
228weed/iam/sts/session_policy_test.go
-
29weed/iam/sts/sts_service.go
-
2weed/iam/sts/token_utils.go
-
33weed/iam/utils/arn_utils.go
-
2weed/iamapi/iamapi_handlers.go
-
63weed/iamapi/iamapi_management_handlers.go
-
1weed/iamapi/iamapi_response.go
-
52weed/iamapi/iamapi_test.go
-
8weed/mount/weedfs_rename.go
-
93weed/mount/weedfs_rename_test.go
-
7weed/pb/server_address.go
-
4weed/pb/volume_server_pb/volume_server.pb.go
-
102weed/pb/volume_server_pb/volume_server_grpc.pb.go
-
17weed/s3api/auth_credentials.go
-
45weed/s3api/auth_credentials_subscribe.go
-
124weed/s3api/auth_credentials_subscribe_test.go
-
14weed/s3api/auth_credentials_test.go
-
8weed/s3api/auth_signature_v4.go
-
3weed/s3api/s3api_object_handlers_list.go
-
86weed/s3api/s3api_object_handlers_list_test.go
-
11weed/s3api/s3api_server.go
-
141weed/s3api/s3api_sts.go
-
153weed/s3api/s3api_sts_assume_role_test.go
-
6weed/server/master_grpc_server.go
-
16weed/server/volume_grpc_scrub.go
-
2weed/shell/command_s3_configure.go
-
3weed/shell/command_volume_scrub.go
-
19weed/storage/erasure_coding/ec_volume_scrub.go
-
2weed/storage/needle/needle_read.go
-
BINweed/storage/test_files/bitrot_volume.dat
-
BINweed/storage/test_files/bitrot_volume.idx
-
BINweed/storage/test_files/healthy_volume.dat
-
BINweed/storage/test_files/healthy_volume.idx
-
124weed/storage/volume_checking.go
-
80weed/storage/volume_checking_test.go
-
16weed/topology/topology.go
@ -0,0 +1,35 @@ |
|||
package sts |
|||
|
|||
import ( |
|||
"encoding/json" |
|||
"fmt" |
|||
"strings" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/iam/policy" |
|||
) |
|||
|
|||
// NormalizeSessionPolicy validates and normalizes inline session policy JSON.
|
|||
// It returns an empty string if the input is empty or whitespace.
|
|||
func NormalizeSessionPolicy(policyJSON string) (string, error) { |
|||
trimmed := strings.TrimSpace(policyJSON) |
|||
if trimmed == "" { |
|||
return "", nil |
|||
} |
|||
const maxSessionPolicySize = 2048 |
|||
if len(trimmed) > maxSessionPolicySize { |
|||
return "", fmt.Errorf("session policy exceeds maximum size of %d characters", maxSessionPolicySize) |
|||
} |
|||
|
|||
var policyDoc policy.PolicyDocument |
|||
if err := json.Unmarshal([]byte(trimmed), &policyDoc); err != nil { |
|||
return "", fmt.Errorf("invalid session policy JSON: %w", err) |
|||
} |
|||
if err := policy.ValidatePolicyDocument(&policyDoc); err != nil { |
|||
return "", fmt.Errorf("invalid session policy document: %w", err) |
|||
} |
|||
normalized, err := json.Marshal(&policyDoc) |
|||
if err != nil { |
|||
return "", fmt.Errorf("failed to normalize session policy: %w", err) |
|||
} |
|||
return string(normalized), nil |
|||
} |
|||
@ -0,0 +1,93 @@ |
|||
package mount |
|||
|
|||
import ( |
|||
"context" |
|||
"path/filepath" |
|||
"testing" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/util" |
|||
) |
|||
|
|||
func TestHandleRenameResponseCachesTargetForUncachedDirectory(t *testing.T) { |
|||
uidGidMapper, err := meta_cache.NewUidGidMapper("", "") |
|||
if err != nil { |
|||
t.Fatalf("create uid/gid mapper: %v", err) |
|||
} |
|||
|
|||
root := util.FullPath("/") |
|||
inodeToPath := NewInodeToPath(root, 1) |
|||
|
|||
mc := meta_cache.NewMetaCache( |
|||
filepath.Join(t.TempDir(), "meta"), |
|||
uidGidMapper, |
|||
root, |
|||
func(path util.FullPath) { |
|||
inodeToPath.MarkChildrenCached(path) |
|||
}, |
|||
func(path util.FullPath) bool { |
|||
return inodeToPath.IsChildrenCached(path) |
|||
}, |
|||
func(util.FullPath, *filer_pb.Entry) {}, |
|||
nil, |
|||
) |
|||
defer mc.Shutdown() |
|||
|
|||
parentPath := util.FullPath("/repo/.git") |
|||
sourcePath := parentPath.Child("config.lock") |
|||
targetPath := parentPath.Child("config") |
|||
|
|||
inodeToPath.Lookup(parentPath, 1, true, false, 0, true) |
|||
sourceInode := inodeToPath.Lookup(sourcePath, 1, false, false, 0, true) |
|||
inodeToPath.Lookup(targetPath, 1, false, false, 0, true) |
|||
|
|||
wfs := &WFS{ |
|||
metaCache: mc, |
|||
inodeToPath: inodeToPath, |
|||
fhMap: NewFileHandleToInode(), |
|||
} |
|||
|
|||
resp := &filer_pb.StreamRenameEntryResponse{ |
|||
Directory: string(parentPath), |
|||
EventNotification: &filer_pb.EventNotification{ |
|||
OldEntry: &filer_pb.Entry{ |
|||
Name: "config.lock", |
|||
}, |
|||
NewEntry: &filer_pb.Entry{ |
|||
Name: "config", |
|||
Attributes: &filer_pb.FuseAttributes{ |
|||
Crtime: 1, |
|||
Mtime: 1, |
|||
FileMode: 0100644, |
|||
FileSize: 53, |
|||
Inode: sourceInode, |
|||
}, |
|||
}, |
|||
NewParentPath: string(parentPath), |
|||
}, |
|||
} |
|||
|
|||
if err := wfs.handleRenameResponse(context.Background(), resp); err != nil { |
|||
t.Fatalf("handle rename response: %v", err) |
|||
} |
|||
|
|||
entry, findErr := mc.FindEntry(context.Background(), targetPath) |
|||
if findErr != nil { |
|||
t.Fatalf("find target entry: %v", findErr) |
|||
} |
|||
if entry == nil { |
|||
t.Fatalf("target entry %s not cached", targetPath) |
|||
} |
|||
if entry.FileSize != 53 { |
|||
t.Fatalf("cached file size = %d, want 53", entry.FileSize) |
|||
} |
|||
|
|||
updatedInode, found := inodeToPath.GetInode(targetPath) |
|||
if !found { |
|||
t.Fatalf("target path %s missing inode mapping", targetPath) |
|||
} |
|||
if updatedInode != sourceInode { |
|||
t.Fatalf("target inode = %d, want %d", updatedInode, sourceInode) |
|||
} |
|||
} |
|||
@ -0,0 +1,124 @@ |
|||
package s3api |
|||
|
|||
import ( |
|||
"context" |
|||
"sync" |
|||
"testing" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/credential" |
|||
_ "github.com/seaweedfs/seaweedfs/weed/credential/memory" |
|||
"github.com/seaweedfs/seaweedfs/weed/filer" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" |
|||
) |
|||
|
|||
func TestOnIamConfigChangeLegacyIdentityDeletionReloadsConfiguration(t *testing.T) { |
|||
s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{ |
|||
{ |
|||
Name: "anonymous", |
|||
Actions: []string{ |
|||
"Read:test", |
|||
}, |
|||
}, |
|||
}) |
|||
|
|||
err := s3a.onIamConfigChange( |
|||
filer.IamConfigDirectory, |
|||
&filer_pb.Entry{Name: filer.IamIdentityFile}, |
|||
nil, |
|||
) |
|||
if err != nil { |
|||
t.Fatalf("onIamConfigChange returned error for legacy identity deletion: %v", err) |
|||
} |
|||
|
|||
if !hasIdentity(s3a.iam, "anonymous") { |
|||
t.Fatalf("expected anonymous identity to remain loaded after legacy identity deletion event") |
|||
} |
|||
} |
|||
|
|||
func TestOnIamConfigChangeReloadsOnIamIdentityDirectoryChanges(t *testing.T) { |
|||
s3a := newTestS3ApiServerWithMemoryIAM(t, []*iam_pb.Identity{ |
|||
{Name: "anonymous"}, |
|||
}) |
|||
|
|||
// Seed initial in-memory IAM state.
|
|||
if err := s3a.iam.LoadS3ApiConfigurationFromCredentialManager(); err != nil { |
|||
t.Fatalf("failed to load initial IAM configuration: %v", err) |
|||
} |
|||
if hasIdentity(s3a.iam, "alice") { |
|||
t.Fatalf("did not expect alice identity before creating user") |
|||
} |
|||
|
|||
if err := s3a.iam.credentialManager.CreateUser(context.Background(), &iam_pb.Identity{Name: "alice"}); err != nil { |
|||
t.Fatalf("failed to create alice in memory credential manager: %v", err) |
|||
} |
|||
|
|||
err := s3a.onIamConfigChange( |
|||
filer.IamConfigDirectory+"/identities", |
|||
nil, |
|||
&filer_pb.Entry{Name: "alice.json"}, |
|||
) |
|||
if err != nil { |
|||
t.Fatalf("onIamConfigChange returned error for identities directory update: %v", err) |
|||
} |
|||
|
|||
if !hasIdentity(s3a.iam, "alice") { |
|||
t.Fatalf("expected alice identity to be loaded after /etc/iam/identities update") |
|||
} |
|||
} |
|||
|
|||
func newTestS3ApiServerWithMemoryIAM(t *testing.T, identities []*iam_pb.Identity) *S3ApiServer { |
|||
t.Helper() |
|||
|
|||
// Create S3ApiConfiguration for test with provided identities
|
|||
config := &iam_pb.S3ApiConfiguration{ |
|||
Identities: identities, |
|||
Accounts: []*iam_pb.Account{}, |
|||
ServiceAccounts: []*iam_pb.ServiceAccount{}, |
|||
} |
|||
|
|||
// Create memory credential manager
|
|||
cm, err := credential.NewCredentialManager(credential.StoreTypeMemory, nil, "") |
|||
if err != nil { |
|||
t.Fatalf("failed to create memory credential manager: %v", err) |
|||
} |
|||
|
|||
// Save test configuration
|
|||
if err := cm.SaveConfiguration(context.Background(), config); err != nil { |
|||
t.Fatalf("failed to save test configuration: %v", err) |
|||
} |
|||
|
|||
// Create a test IAM instance
|
|||
iam := &IdentityAccessManagement{ |
|||
m: sync.RWMutex{}, |
|||
nameToIdentity: make(map[string]*Identity), |
|||
accessKeyIdent: make(map[string]*Identity), |
|||
identities: []*Identity{}, |
|||
policies: make(map[string]*iam_pb.Policy), |
|||
accounts: make(map[string]*Account), |
|||
emailAccount: make(map[string]*Account), |
|||
hashes: make(map[string]*sync.Pool), |
|||
hashCounters: make(map[string]*int32), |
|||
isAuthEnabled: false, |
|||
stopChan: make(chan struct{}), |
|||
useStaticConfig: false, |
|||
credentialManager: cm, |
|||
} |
|||
|
|||
// Load test configuration
|
|||
if err := iam.ReplaceS3ApiConfiguration(config); err != nil { |
|||
t.Fatalf("failed to load test configuration: %v", err) |
|||
} |
|||
|
|||
return &S3ApiServer{ |
|||
iam: iam, |
|||
} |
|||
} |
|||
|
|||
func hasIdentity(iam *IdentityAccessManagement, identityName string) bool { |
|||
iam.m.RLock() |
|||
defer iam.m.RUnlock() |
|||
|
|||
_, ok := iam.nameToIdentity[identityName] |
|||
return ok |
|||
} |
|||
@ -0,0 +1,153 @@ |
|||
package s3api |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"net/http" |
|||
"net/url" |
|||
"testing" |
|||
|
|||
"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/assert" |
|||
"github.com/stretchr/testify/require" |
|||
) |
|||
|
|||
// TestAssumeRole_CallerIdentityFallback tests the fallback logic when RoleArn is missing
|
|||
func TestAssumeRole_CallerIdentityFallback(t *testing.T) { |
|||
// Setup STS service
|
|||
stsService, _ := setupTestSTSService(t) |
|||
|
|||
// Create IAM integration mock
|
|||
iamMock := &MockIAMIntegration{ |
|||
authorizeFunc: func(ctx context.Context, identity *IAMIdentity, action Action, bucket, object string, r *http.Request) s3err.ErrorCode { |
|||
// Allow global sts:AssumeRole
|
|||
if action == "sts:AssumeRole" { |
|||
return s3err.ErrNone |
|||
} |
|||
return s3err.ErrAccessDenied |
|||
}, |
|||
validateTrustPolicyFunc: func(ctx context.Context, roleArn, principalArn string) error { |
|||
// Allow all trust policies for this test
|
|||
return nil |
|||
}, |
|||
} |
|||
|
|||
// Create IAM service with the mock integration
|
|||
iam := &IdentityAccessManagement{ |
|||
iamIntegration: iamMock, |
|||
} |
|||
|
|||
// Create STS handlers
|
|||
stsHandlers := NewSTSHandlers(stsService, iam) |
|||
|
|||
// Test case 1: Caller is an IAM User, RoleArn is missing
|
|||
t.Run("Caller is IAM User, No RoleArn", func(t *testing.T) { |
|||
// Mock request
|
|||
req, err := http.NewRequest("POST", "/", nil) |
|||
require.NoError(t, err) |
|||
req.Form = url.Values{} |
|||
req.Form.Set("Action", "AssumeRole") |
|||
req.Form.Set("RoleSessionName", "test-session") |
|||
req.Form.Set("Version", "2011-06-15") |
|||
|
|||
// Mock the authenticated identity (IAM User)
|
|||
callerIdentity := &Identity{ |
|||
Name: "alice", |
|||
Account: &AccountAdmin, |
|||
PrincipalArn: fmt.Sprintf("arn:aws:iam::%s:user/alice", defaultAccountID), |
|||
Actions: []Action{s3_constants.ACTION_ADMIN}, |
|||
} |
|||
|
|||
// 1. Test prepareSTSCredentials with NO RoleArn (simulating the fallback logic having passed PrincipalArn)
|
|||
// expected RoleArn passed to prepareSTSCredentials would be the caller's PrincipalArn
|
|||
fallbackRoleArn := callerIdentity.PrincipalArn |
|||
|
|||
// Prepare custom claims for the session (mimicking handleAssumeRole logic)
|
|||
var modifyClaims func(claims *sts.STSSessionClaims) |
|||
if callerIdentity.isAdmin() { |
|||
modifyClaims = func(claims *sts.STSSessionClaims) { |
|||
if claims.RequestContext == nil { |
|||
claims.RequestContext = make(map[string]interface{}) |
|||
} |
|||
claims.RequestContext["is_admin"] = true |
|||
} |
|||
} |
|||
|
|||
stsCreds, assumedUser, err := stsHandlers.prepareSTSCredentials(fallbackRoleArn, "test-session", nil, "", modifyClaims) |
|||
require.NoError(t, err) |
|||
|
|||
// Assertions
|
|||
// The role name should be extracted from the user ARN ("alice")
|
|||
assert.Contains(t, assumedUser.Arn, fmt.Sprintf("assumed-role/alice/test-session")) |
|||
assert.Contains(t, assumedUser.AssumedRoleId, "alice:test-session") |
|||
|
|||
// Verify token claims using ValidateSessionToken
|
|||
sessionInfo, err := stsService.ValidateSessionToken(context.Background(), stsCreds.SessionToken) |
|||
require.NoError(t, err) |
|||
|
|||
// The RoleArn in session info should match the fallback ARN (user ARN)
|
|||
assert.Equal(t, fallbackRoleArn, sessionInfo.RoleArn) |
|||
|
|||
// Verify is_admin claim is present
|
|||
isAdmin, ok := sessionInfo.RequestContext["is_admin"].(bool) |
|||
assert.True(t, ok, "is_admin claim should be present") |
|||
assert.True(t, isAdmin, "is_admin claim should be true") |
|||
}) |
|||
|
|||
// Test case 2: Caller is an STS Assumed Role, No RoleArn
|
|||
t.Run("Caller is STS Assumed Role, No RoleArn", func(t *testing.T) { |
|||
// Mock identity
|
|||
callerIdentity := &Identity{ |
|||
Name: "arn:aws:sts::111122223333:assumed-role/admin/session1", |
|||
Account: &AccountAdmin, |
|||
PrincipalArn: "arn:aws:sts::111122223333:assumed-role/admin/session1", |
|||
} |
|||
|
|||
fallbackRoleArn := callerIdentity.PrincipalArn |
|||
|
|||
stsCreds, assumedUser, err := stsHandlers.prepareSTSCredentials(fallbackRoleArn, "nested-session", nil, "", nil) |
|||
require.NoError(t, err) |
|||
|
|||
// The role name should be extracted from the assumed role ARN ("admin")
|
|||
assert.Contains(t, assumedUser.Arn, "assumed-role/admin/nested-session") |
|||
assert.Contains(t, assumedUser.AssumedRoleId, "admin:nested-session") |
|||
|
|||
// Check claims
|
|||
sessionInfo, err := stsService.ValidateSessionToken(context.Background(), stsCreds.SessionToken) |
|||
require.NoError(t, err) |
|||
assert.Equal(t, fallbackRoleArn, sessionInfo.RoleArn) |
|||
}) |
|||
|
|||
// Test case 3: Explicit RoleArn provided (Standard AssumeRole)
|
|||
t.Run("Explicit RoleArn Provided", func(t *testing.T) { |
|||
explicitRoleArn := "arn:aws:iam::111122223333:role/TargetRole" |
|||
|
|||
stsCreds, assumedUser, err := stsHandlers.prepareSTSCredentials(explicitRoleArn, "explicit-session", nil, "", nil) |
|||
require.NoError(t, err) |
|||
|
|||
// Role name should be "TargetRole"
|
|||
assert.Contains(t, assumedUser.Arn, "assumed-role/TargetRole/explicit-session") |
|||
|
|||
// Check claims
|
|||
sessionInfo, err := stsService.ValidateSessionToken(context.Background(), stsCreds.SessionToken) |
|||
require.NoError(t, err) |
|||
assert.Equal(t, explicitRoleArn, sessionInfo.RoleArn) |
|||
}) |
|||
|
|||
// Test case 4: Malformed ARN (Edge case)
|
|||
t.Run("Malformed ARN", func(t *testing.T) { |
|||
malformedArn := "invalid-arn" |
|||
|
|||
stsCreds, assumedUser, err := stsHandlers.prepareSTSCredentials(malformedArn, "bad-session", nil, "", nil) |
|||
require.NoError(t, err) |
|||
|
|||
// Fallback behavior: use full string as role name if extraction fails
|
|||
assert.Contains(t, assumedUser.Arn, "assumed-role/invalid-arn/bad-session") |
|||
|
|||
sessionInfo, err := stsService.ValidateSessionToken(context.Background(), stsCreds.SessionToken) |
|||
require.NoError(t, err) |
|||
assert.Equal(t, malformedArn, sessionInfo.RoleArn) |
|||
}) |
|||
} |
|||
@ -0,0 +1,80 @@ |
|||
package storage |
|||
|
|||
import ( |
|||
"fmt" |
|||
"os" |
|||
"reflect" |
|||
"testing" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/storage/backend" |
|||
"github.com/seaweedfs/seaweedfs/weed/storage/needle" |
|||
) |
|||
|
|||
func TestScrubVolumeData(t *testing.T) { |
|||
testCases := []struct { |
|||
name string |
|||
dataPath string |
|||
indexPath string |
|||
version needle.Version |
|||
want int64 |
|||
wantErrs []error |
|||
}{ |
|||
{ |
|||
name: "healthy volume", |
|||
dataPath: "./test_files/healthy_volume.dat", |
|||
indexPath: "./test_files/healthy_volume.idx", |
|||
version: needle.Version3, |
|||
want: 27, |
|||
wantErrs: []error{}, |
|||
}, |
|||
{ |
|||
name: "bitrot volume", |
|||
dataPath: "./test_files/bitrot_volume.dat", |
|||
indexPath: "./test_files/bitrot_volume.idx", |
|||
version: needle.Version3, |
|||
want: 27, |
|||
wantErrs: []error{ |
|||
fmt.Errorf("needle 3 on volume 0: invalid CRC for needle 3 (got 0b243a0d, want 4af853fb), data on disk corrupted"), |
|||
fmt.Errorf("needle 48 on volume 0: invalid CRC for needle 30 (got 3c40e8d5, want 5077fea1), data on disk corrupted"), |
|||
fmt.Errorf("data file size for volume 0 (942864) doesn't match the size for 27 needles read (942856)"), |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
for _, tc := range testCases { |
|||
t.Run(tc.name, func(t *testing.T) { |
|||
datFile, err := os.OpenFile(tc.dataPath, os.O_RDONLY, 0) |
|||
if err != nil { |
|||
t.Fatalf("failed to open data file: %v", err) |
|||
} |
|||
defer datFile.Close() |
|||
|
|||
idxFile, err := os.OpenFile(tc.indexPath, os.O_RDONLY, 0) |
|||
if err != nil { |
|||
t.Fatalf("failed to open index file: %v", err) |
|||
} |
|||
defer idxFile.Close() |
|||
|
|||
idxStat, err := idxFile.Stat() |
|||
if err != nil { |
|||
t.Fatalf("failed to stat index file: %v", err) |
|||
} |
|||
|
|||
v := Volume{ |
|||
volumeInfo: &volume_server_pb.VolumeInfo{ |
|||
Version: uint32(tc.version), |
|||
}, |
|||
} |
|||
|
|||
got, gotErrs := v.scrubVolumeData(backend.NewDiskFile(datFile), idxFile, idxStat.Size()) |
|||
|
|||
if got != tc.want { |
|||
t.Errorf("expected %d files processed, got %d", tc.want, got) |
|||
} |
|||
if !reflect.DeepEqual(gotErrs, tc.wantErrs) { |
|||
t.Errorf("expected errors %v, got %v", tc.wantErrs, gotErrs) |
|||
} |
|||
}) |
|||
} |
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue