Browse Source
s3api: fix static IAM policy enforcement after reload (#8532)
s3api: fix static IAM policy enforcement after reload (#8532)
* s3api: honor attached IAM policies over legacy actions * s3api: hydrate IAM policy docs during config reload * s3api: use policy-aware auth when listing buckets * credential: propagate context through filer_etc policy reads * credential: make legacy policy deletes durable * s3api: exercise managed policy runtime loader * s3api: allow static IAM users without session tokens * iam: deny unmatched attached policies under default allow * iam: load embedded policy files from filer store * s3api: require session tokens for IAM presigning * s3api: sync runtime policies into zero-config IAM * credential: respect context in policy file loads * credential: serialize legacy policy deletes * iam: align filer policy store naming * s3api: use authenticated principals for presigning * iam: deep copy policy conditions * s3api: require request creation in policy tests * filer: keep ReadInsideFiler as the context-aware API * iam: harden filer policy store writes * credential: strengthen legacy policy serialization test * credential: forward runtime policy loaders through wrapper * s3api: harden runtime policy merging * iam: require typed already-exists errorspull/8522/merge
Chris Lu
18 hours ago
committed by
GitHub
GitHub
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
30 changed files with 1900 additions and 165 deletions
-
2weed/admin/dash/mq_management.go
-
12weed/credential/filer_etc/filer_etc_identity.go
-
181weed/credential/filer_etc/filer_etc_policy.go
-
369weed/credential/filer_etc/filer_etc_policy_test.go
-
6weed/credential/filer_etc/filer_etc_service_account.go
-
1weed/credential/filer_etc/filer_etc_store.go
-
40weed/credential/propagating_store.go
-
2weed/filer/filer_conf.go
-
4weed/filer/read_write.go
-
7weed/filer/remote_mapping.go
-
2weed/filer/remote_storage.go
-
64weed/iam/integration/iam_manager.go
-
21weed/iam/policy/policy_engine.go
-
260weed/iam/policy/policy_store.go
-
314weed/iam/policy/policy_store_test.go
-
3weed/mount/filer_conf.go
-
2weed/mq/kafka/gateway/coordinator_registry.go
-
2weed/mq/offset/consumer_group_storage.go
-
3weed/mq/offset/filer_storage.go
-
2weed/mq/topic/topic.go
-
2weed/query/engine/broker_client.go
-
190weed/s3api/auth_credentials.go
-
359weed/s3api/auth_credentials_test.go
-
11weed/s3api/s3_iam_middleware.go
-
93weed/s3api/s3_iam_simple_test.go
-
26weed/s3api/s3_presigned_url_iam.go
-
29weed/s3api/s3_presigned_url_iam_test.go
-
15weed/s3api/s3api_bucket_handlers.go
-
40weed/s3api/s3api_bucket_handlers_test.go
-
3weed/s3api/s3api_circuit_breaker.go
@ -0,0 +1,369 @@ |
|||
package filer_etc |
|||
|
|||
import ( |
|||
"context" |
|||
"net" |
|||
"sort" |
|||
"strconv" |
|||
"sync" |
|||
"testing" |
|||
"time" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/filer" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" |
|||
"github.com/stretchr/testify/assert" |
|||
"github.com/stretchr/testify/require" |
|||
"google.golang.org/grpc" |
|||
"google.golang.org/grpc/codes" |
|||
"google.golang.org/grpc/credentials/insecure" |
|||
"google.golang.org/grpc/status" |
|||
"google.golang.org/protobuf/proto" |
|||
) |
|||
|
|||
type policyTestFilerServer struct { |
|||
filer_pb.UnimplementedSeaweedFilerServer |
|||
mu sync.RWMutex |
|||
entries map[string]*filer_pb.Entry |
|||
contentlessListEntry map[string]struct{} |
|||
beforeLookup func(context.Context, string, string) error |
|||
afterListEntry func(string, string) |
|||
beforeDelete func(string, string) error |
|||
beforeUpdate func(string, string) error |
|||
} |
|||
|
|||
func newPolicyTestFilerServer() *policyTestFilerServer { |
|||
return &policyTestFilerServer{ |
|||
entries: make(map[string]*filer_pb.Entry), |
|||
contentlessListEntry: make(map[string]struct{}), |
|||
} |
|||
} |
|||
|
|||
func (s *policyTestFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { |
|||
s.mu.RLock() |
|||
beforeLookup := s.beforeLookup |
|||
s.mu.RUnlock() |
|||
if beforeLookup != nil { |
|||
if err := beforeLookup(ctx, req.Directory, req.Name); err != nil { |
|||
return nil, err |
|||
} |
|||
} |
|||
|
|||
s.mu.RLock() |
|||
defer s.mu.RUnlock() |
|||
|
|||
entry, found := s.entries[filerEntryKey(req.Directory, req.Name)] |
|||
if !found { |
|||
return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) |
|||
} |
|||
|
|||
return &filer_pb.LookupDirectoryEntryResponse{Entry: cloneEntry(entry)}, nil |
|||
} |
|||
|
|||
func (s *policyTestFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error { |
|||
s.mu.RLock() |
|||
defer s.mu.RUnlock() |
|||
|
|||
names := make([]string, 0) |
|||
for key := range s.entries { |
|||
dir, name := splitFilerEntryKey(key) |
|||
if dir != req.Directory { |
|||
continue |
|||
} |
|||
names = append(names, name) |
|||
} |
|||
sort.Strings(names) |
|||
|
|||
for _, name := range names { |
|||
entry := cloneEntry(s.entries[filerEntryKey(req.Directory, name)]) |
|||
if _, found := s.contentlessListEntry[filerEntryKey(req.Directory, name)]; found { |
|||
entry.Content = nil |
|||
} |
|||
if err := stream.Send(&filer_pb.ListEntriesResponse{Entry: entry}); err != nil { |
|||
return err |
|||
} |
|||
if s.afterListEntry != nil { |
|||
s.afterListEntry(req.Directory, name) |
|||
} |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (s *policyTestFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { |
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
|
|||
s.entries[filerEntryKey(req.Directory, req.Entry.Name)] = cloneEntry(req.Entry) |
|||
return &filer_pb.CreateEntryResponse{}, nil |
|||
} |
|||
|
|||
func (s *policyTestFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { |
|||
s.mu.RLock() |
|||
beforeUpdate := s.beforeUpdate |
|||
s.mu.RUnlock() |
|||
if beforeUpdate != nil { |
|||
if err := beforeUpdate(req.Directory, req.Entry.Name); err != nil { |
|||
return nil, err |
|||
} |
|||
} |
|||
|
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
|
|||
s.entries[filerEntryKey(req.Directory, req.Entry.Name)] = cloneEntry(req.Entry) |
|||
return &filer_pb.UpdateEntryResponse{}, nil |
|||
} |
|||
|
|||
func (s *policyTestFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { |
|||
s.mu.RLock() |
|||
beforeDelete := s.beforeDelete |
|||
s.mu.RUnlock() |
|||
if beforeDelete != nil { |
|||
if err := beforeDelete(req.Directory, req.Name); err != nil { |
|||
return nil, err |
|||
} |
|||
} |
|||
|
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
|
|||
key := filerEntryKey(req.Directory, req.Name) |
|||
if _, found := s.entries[key]; !found { |
|||
return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) |
|||
} |
|||
|
|||
delete(s.entries, key) |
|||
return &filer_pb.DeleteEntryResponse{}, nil |
|||
} |
|||
|
|||
func newPolicyTestStore(t *testing.T) *FilerEtcStore { |
|||
store, _ := newPolicyTestStoreWithServer(t) |
|||
return store |
|||
} |
|||
|
|||
func newPolicyTestStoreWithServer(t *testing.T) (*FilerEtcStore, *policyTestFilerServer) { |
|||
t.Helper() |
|||
|
|||
lis, err := net.Listen("tcp", "127.0.0.1:0") |
|||
require.NoError(t, err) |
|||
|
|||
server := newPolicyTestFilerServer() |
|||
grpcServer := pb.NewGrpcServer() |
|||
filer_pb.RegisterSeaweedFilerServer(grpcServer, server) |
|||
go func() { |
|||
_ = grpcServer.Serve(lis) |
|||
}() |
|||
|
|||
t.Cleanup(func() { |
|||
grpcServer.Stop() |
|||
_ = lis.Close() |
|||
}) |
|||
|
|||
store := &FilerEtcStore{} |
|||
host, portString, err := net.SplitHostPort(lis.Addr().String()) |
|||
require.NoError(t, err) |
|||
grpcPort, err := strconv.Atoi(portString) |
|||
require.NoError(t, err) |
|||
store.SetFilerAddressFunc(func() pb.ServerAddress { |
|||
return pb.NewServerAddress(host, 1, grpcPort) |
|||
}, grpc.WithTransportCredentials(insecure.NewCredentials())) |
|||
|
|||
return store, server |
|||
} |
|||
|
|||
func TestFilerEtcStoreListPolicyNamesIncludesLegacyPolicies(t *testing.T) { |
|||
ctx := context.Background() |
|||
store := newPolicyTestStore(t) |
|||
|
|||
legacyPolicies := newPoliciesCollection() |
|||
legacyPolicies.Policies["legacy-only"] = testPolicyDocument("s3:GetObject", "arn:aws:s3:::legacy-only/*") |
|||
legacyPolicies.Policies["shared"] = testPolicyDocument("s3:GetObject", "arn:aws:s3:::shared/*") |
|||
require.NoError(t, store.saveLegacyPoliciesCollection(ctx, legacyPolicies)) |
|||
|
|||
require.NoError(t, store.savePolicy(ctx, "multi-file-only", testPolicyDocument("s3:PutObject", "arn:aws:s3:::multi-file-only/*"))) |
|||
require.NoError(t, store.savePolicy(ctx, "shared", testPolicyDocument("s3:DeleteObject", "arn:aws:s3:::shared/*"))) |
|||
|
|||
names, err := store.ListPolicyNames(ctx) |
|||
require.NoError(t, err) |
|||
|
|||
assert.ElementsMatch(t, []string{"legacy-only", "multi-file-only", "shared"}, names) |
|||
} |
|||
|
|||
func TestFilerEtcStoreDeletePolicyRemovesLegacyManagedCopy(t *testing.T) { |
|||
ctx := context.Background() |
|||
store := newPolicyTestStore(t) |
|||
|
|||
inlinePolicy := testPolicyDocument("s3:PutObject", "arn:aws:s3:::inline-user/*") |
|||
legacyPolicies := newPoliciesCollection() |
|||
legacyPolicies.Policies["legacy-only"] = testPolicyDocument("s3:GetObject", "arn:aws:s3:::legacy-only/*") |
|||
legacyPolicies.InlinePolicies["inline-user"] = map[string]policy_engine.PolicyDocument{ |
|||
"PutOnly": inlinePolicy, |
|||
} |
|||
require.NoError(t, store.saveLegacyPoliciesCollection(ctx, legacyPolicies)) |
|||
|
|||
managedPolicies, err := store.LoadManagedPolicies(ctx) |
|||
require.NoError(t, err) |
|||
assert.Equal(t, []string{"legacy-only"}, managedPolicyNames(managedPolicies)) |
|||
|
|||
require.NoError(t, store.DeletePolicy(ctx, "legacy-only")) |
|||
|
|||
managedPolicies, err = store.LoadManagedPolicies(ctx) |
|||
require.NoError(t, err) |
|||
assert.Empty(t, managedPolicies) |
|||
|
|||
inlinePolicies, err := store.LoadInlinePolicies(ctx) |
|||
require.NoError(t, err) |
|||
assertInlinePolicyPreserved(t, inlinePolicies, "inline-user", "PutOnly") |
|||
|
|||
loadedLegacyPolicies, foundLegacy, err := store.loadLegacyPoliciesCollection(ctx) |
|||
require.NoError(t, err) |
|||
require.True(t, foundLegacy) |
|||
assert.Empty(t, loadedLegacyPolicies.Policies) |
|||
assertInlinePolicyPreserved(t, loadedLegacyPolicies.InlinePolicies, "inline-user", "PutOnly") |
|||
} |
|||
|
|||
func TestFilerEtcStoreDeletePolicySerializesLegacyUpdates(t *testing.T) { |
|||
ctx := context.Background() |
|||
store, server := newPolicyTestStoreWithServer(t) |
|||
|
|||
legacyPolicies := newPoliciesCollection() |
|||
legacyPolicies.Policies["first"] = testPolicyDocument("s3:GetObject", "arn:aws:s3:::first/*") |
|||
legacyPolicies.Policies["second"] = testPolicyDocument("s3:GetObject", "arn:aws:s3:::second/*") |
|||
require.NoError(t, store.saveLegacyPoliciesCollection(ctx, legacyPolicies)) |
|||
require.NoError(t, store.savePolicy(ctx, "first", testPolicyDocument("s3:GetObject", "arn:aws:s3:::first/*"))) |
|||
require.NoError(t, store.savePolicy(ctx, "second", testPolicyDocument("s3:GetObject", "arn:aws:s3:::second/*"))) |
|||
|
|||
firstSaveStarted := make(chan struct{}) |
|||
releaseFirstSave := make(chan struct{}) |
|||
secondReachedDelete := make(chan struct{}, 1) |
|||
var blockOnce sync.Once |
|||
|
|||
server.mu.Lock() |
|||
server.beforeUpdate = func(dir string, name string) error { |
|||
if dir == filer.IamConfigDirectory && name == filer.IamPoliciesFile { |
|||
blockOnce.Do(func() { |
|||
close(firstSaveStarted) |
|||
<-releaseFirstSave |
|||
}) |
|||
} |
|||
return nil |
|||
} |
|||
server.beforeDelete = func(dir string, name string) error { |
|||
if dir == filer.IamConfigDirectory+"/"+IamPoliciesDirectory && name == "second.json" { |
|||
select { |
|||
case secondReachedDelete <- struct{}{}: |
|||
default: |
|||
} |
|||
} |
|||
return nil |
|||
} |
|||
server.mu.Unlock() |
|||
|
|||
firstDeleteErr := make(chan error, 1) |
|||
go func() { |
|||
firstDeleteErr <- store.DeletePolicy(ctx, "first") |
|||
}() |
|||
|
|||
<-firstSaveStarted |
|||
|
|||
secondDeleteErr := make(chan error, 1) |
|||
go func() { |
|||
secondDeleteErr <- store.DeletePolicy(ctx, "second") |
|||
}() |
|||
|
|||
select { |
|||
case <-secondReachedDelete: |
|||
t.Fatal("second delete reached filer mutation while first delete was still blocked") |
|||
case <-time.After(300 * time.Millisecond): |
|||
} |
|||
|
|||
close(releaseFirstSave) |
|||
|
|||
require.NoError(t, <-firstDeleteErr) |
|||
require.NoError(t, <-secondDeleteErr) |
|||
|
|||
loadedLegacyPolicies, foundLegacy, err := store.loadLegacyPoliciesCollection(ctx) |
|||
require.NoError(t, err) |
|||
require.True(t, foundLegacy) |
|||
assert.Empty(t, loadedLegacyPolicies.Policies) |
|||
} |
|||
|
|||
func TestFilerEtcStoreLoadManagedPoliciesRespectsReadContext(t *testing.T) { |
|||
ctx, cancel := context.WithCancel(context.Background()) |
|||
store, server := newPolicyTestStoreWithServer(t) |
|||
|
|||
require.NoError(t, store.savePolicy(context.Background(), "cancel-me", testPolicyDocument("s3:GetObject", "arn:aws:s3:::cancel-me/*"))) |
|||
|
|||
server.mu.Lock() |
|||
server.contentlessListEntry[filerEntryKey(filer.IamConfigDirectory+"/"+IamPoliciesDirectory, "cancel-me.json")] = struct{}{} |
|||
server.beforeLookup = func(ctx context.Context, dir string, name string) error { |
|||
if dir == filer.IamConfigDirectory+"/"+IamPoliciesDirectory && name == "cancel-me.json" { |
|||
cancel() |
|||
return status.Error(codes.Canceled, context.Canceled.Error()) |
|||
} |
|||
return nil |
|||
} |
|||
server.mu.Unlock() |
|||
|
|||
managedPolicies, err := store.LoadManagedPolicies(ctx) |
|||
require.NoError(t, err) |
|||
assert.Empty(t, managedPolicies) |
|||
} |
|||
|
|||
func testPolicyDocument(action string, resource string) policy_engine.PolicyDocument { |
|||
return policy_engine.PolicyDocument{ |
|||
Version: policy_engine.PolicyVersion2012_10_17, |
|||
Statement: []policy_engine.PolicyStatement{ |
|||
{ |
|||
Effect: policy_engine.PolicyEffectAllow, |
|||
Action: policy_engine.NewStringOrStringSlice(action), |
|||
Resource: policy_engine.NewStringOrStringSlice(resource), |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
|
|||
func managedPolicyNames(policies []*iam_pb.Policy) []string { |
|||
names := make([]string, 0, len(policies)) |
|||
for _, policy := range policies { |
|||
names = append(names, policy.Name) |
|||
} |
|||
sort.Strings(names) |
|||
return names |
|||
} |
|||
|
|||
func assertInlinePolicyPreserved(t *testing.T, inlinePolicies map[string]map[string]policy_engine.PolicyDocument, userName string, policyName string) { |
|||
t.Helper() |
|||
|
|||
userPolicies, found := inlinePolicies[userName] |
|||
require.True(t, found) |
|||
|
|||
policy, found := userPolicies[policyName] |
|||
require.True(t, found) |
|||
assert.Equal(t, policy_engine.PolicyVersion2012_10_17, policy.Version) |
|||
require.Len(t, policy.Statement, 1) |
|||
assert.Equal(t, policy_engine.PolicyEffectAllow, policy.Statement[0].Effect) |
|||
} |
|||
|
|||
func cloneEntry(entry *filer_pb.Entry) *filer_pb.Entry { |
|||
if entry == nil { |
|||
return nil |
|||
} |
|||
return proto.Clone(entry).(*filer_pb.Entry) |
|||
} |
|||
|
|||
func filerEntryKey(dir string, name string) string { |
|||
return dir + "\x00" + name |
|||
} |
|||
|
|||
func splitFilerEntryKey(key string) (dir string, name string) { |
|||
for idx := 0; idx < len(key); idx++ { |
|||
if key[idx] == '\x00' { |
|||
return key[:idx], key[idx+1:] |
|||
} |
|||
} |
|||
return key, "" |
|||
} |
|||
@ -0,0 +1,314 @@ |
|||
package policy |
|||
|
|||
import ( |
|||
"context" |
|||
"encoding/json" |
|||
"fmt" |
|||
"net" |
|||
"sort" |
|||
"strconv" |
|||
"sync" |
|||
"testing" |
|||
|
|||
"github.com/seaweedfs/seaweedfs/weed/pb" |
|||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" |
|||
"github.com/stretchr/testify/assert" |
|||
"github.com/stretchr/testify/require" |
|||
"google.golang.org/grpc" |
|||
"google.golang.org/grpc/codes" |
|||
"google.golang.org/grpc/credentials/insecure" |
|||
"google.golang.org/grpc/status" |
|||
"google.golang.org/protobuf/proto" |
|||
) |
|||
|
|||
type policyStoreTestFilerServer struct { |
|||
filer_pb.UnimplementedSeaweedFilerServer |
|||
mu sync.RWMutex |
|||
entries map[string]*filer_pb.Entry |
|||
} |
|||
|
|||
func newPolicyStoreTestFilerServer() *policyStoreTestFilerServer { |
|||
return &policyStoreTestFilerServer{ |
|||
entries: make(map[string]*filer_pb.Entry), |
|||
} |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { |
|||
s.mu.RLock() |
|||
defer s.mu.RUnlock() |
|||
|
|||
entry, found := s.entries[policyStoreTestEntryKey(req.Directory, req.Name)] |
|||
if !found { |
|||
return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) |
|||
} |
|||
|
|||
return &filer_pb.LookupDirectoryEntryResponse{Entry: clonePolicyStoreEntry(entry)}, nil |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { |
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
|
|||
key := policyStoreTestEntryKey(req.Directory, req.Entry.Name) |
|||
if _, found := s.entries[key]; found { |
|||
return nil, status.Error(codes.AlreadyExists, "entry already exists") |
|||
} |
|||
|
|||
s.entries[key] = clonePolicyStoreEntry(req.Entry) |
|||
return &filer_pb.CreateEntryResponse{}, nil |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { |
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
|
|||
key := policyStoreTestEntryKey(req.Directory, req.Entry.Name) |
|||
if _, found := s.entries[key]; !found { |
|||
return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) |
|||
} |
|||
|
|||
s.entries[key] = clonePolicyStoreEntry(req.Entry) |
|||
return &filer_pb.UpdateEntryResponse{}, nil |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream grpc.ServerStreamingServer[filer_pb.ListEntriesResponse]) error { |
|||
s.mu.RLock() |
|||
defer s.mu.RUnlock() |
|||
|
|||
names := make([]string, 0) |
|||
for key := range s.entries { |
|||
dir, name := splitPolicyStoreEntryKey(key) |
|||
if dir != req.Directory { |
|||
continue |
|||
} |
|||
if req.Prefix != "" && len(name) >= len(req.Prefix) && name[:len(req.Prefix)] != req.Prefix { |
|||
continue |
|||
} |
|||
if req.Prefix != "" && len(name) < len(req.Prefix) { |
|||
continue |
|||
} |
|||
names = append(names, name) |
|||
} |
|||
sort.Strings(names) |
|||
|
|||
for _, name := range names { |
|||
if err := stream.Send(&filer_pb.ListEntriesResponse{ |
|||
Entry: clonePolicyStoreEntry(s.entries[policyStoreTestEntryKey(req.Directory, name)]), |
|||
}); err != nil { |
|||
return err |
|||
} |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) { |
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
|
|||
key := policyStoreTestEntryKey(req.Directory, req.Name) |
|||
if _, found := s.entries[key]; !found { |
|||
return nil, status.Error(codes.NotFound, filer_pb.ErrNotFound.Error()) |
|||
} |
|||
|
|||
delete(s.entries, key) |
|||
return &filer_pb.DeleteEntryResponse{}, nil |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) putPolicyFile(t *testing.T, dir string, name string, document *PolicyDocument) { |
|||
t.Helper() |
|||
|
|||
content, err := json.Marshal(document) |
|||
require.NoError(t, err) |
|||
|
|||
s.mu.Lock() |
|||
defer s.mu.Unlock() |
|||
s.entries[policyStoreTestEntryKey(dir, name)] = &filer_pb.Entry{ |
|||
Name: name, |
|||
Content: content, |
|||
} |
|||
} |
|||
|
|||
func (s *policyStoreTestFilerServer) hasEntry(dir string, name string) bool { |
|||
s.mu.RLock() |
|||
defer s.mu.RUnlock() |
|||
_, found := s.entries[policyStoreTestEntryKey(dir, name)] |
|||
return found |
|||
} |
|||
|
|||
func newTestFilerPolicyStore(t *testing.T) (*FilerPolicyStore, *policyStoreTestFilerServer) { |
|||
t.Helper() |
|||
|
|||
lis, err := net.Listen("tcp", "127.0.0.1:0") |
|||
require.NoError(t, err) |
|||
|
|||
server := newPolicyStoreTestFilerServer() |
|||
grpcServer := pb.NewGrpcServer() |
|||
filer_pb.RegisterSeaweedFilerServer(grpcServer, server) |
|||
go func() { |
|||
_ = grpcServer.Serve(lis) |
|||
}() |
|||
|
|||
t.Cleanup(func() { |
|||
grpcServer.Stop() |
|||
_ = lis.Close() |
|||
}) |
|||
|
|||
host, portString, err := net.SplitHostPort(lis.Addr().String()) |
|||
require.NoError(t, err) |
|||
grpcPort, err := strconv.Atoi(portString) |
|||
require.NoError(t, err) |
|||
|
|||
store, err := NewFilerPolicyStore(nil, func() string { |
|||
return string(pb.NewServerAddress(host, 1, grpcPort)) |
|||
}) |
|||
require.NoError(t, err) |
|||
store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) |
|||
|
|||
return store, server |
|||
} |
|||
|
|||
func TestFilerPolicyStoreGetPolicyPrefersCanonicalFiles(t *testing.T) { |
|||
ctx := context.Background() |
|||
store, server := newTestFilerPolicyStore(t) |
|||
|
|||
server.putPolicyFile(t, store.basePath, "cli-bucket-access-policy.json", testPolicyDocument("s3:ListBucket", "arn:aws:s3:::cli-allowed-bucket")) |
|||
server.putPolicyFile(t, store.basePath, "policy_cli-bucket-access-policy.json", testPolicyDocument("s3:PutObject", "arn:aws:s3:::cli-forbidden-bucket/*")) |
|||
|
|||
document, err := store.GetPolicy(ctx, "", "cli-bucket-access-policy") |
|||
require.NoError(t, err) |
|||
require.Len(t, document.Statement, 1) |
|||
assert.Equal(t, "s3:ListBucket", document.Statement[0].Action[0]) |
|||
assert.Equal(t, "arn:aws:s3:::cli-allowed-bucket", document.Statement[0].Resource[0]) |
|||
} |
|||
|
|||
func TestFilerPolicyStoreListPoliciesIncludesCanonicalAndLegacyFiles(t *testing.T) { |
|||
ctx := context.Background() |
|||
store, server := newTestFilerPolicyStore(t) |
|||
|
|||
server.putPolicyFile(t, store.basePath, "canonical-only.json", testPolicyDocument("s3:GetObject", "arn:aws:s3:::canonical-only/*")) |
|||
server.putPolicyFile(t, store.basePath, "policy_legacy-only.json", testPolicyDocument("s3:PutObject", "arn:aws:s3:::legacy-only/*")) |
|||
server.putPolicyFile(t, store.basePath, "shared.json", testPolicyDocument("s3:DeleteObject", "arn:aws:s3:::shared/*")) |
|||
server.putPolicyFile(t, store.basePath, "policy_shared.json", testPolicyDocument("s3:ListBucket", "arn:aws:s3:::shared")) |
|||
server.putPolicyFile(t, store.basePath, "policy_invalid:name.json", testPolicyDocument("s3:GetObject", "arn:aws:s3:::ignored/*")) |
|||
server.putPolicyFile(t, store.basePath, "bucket-policy:bucket-a.json", testPolicyDocument("s3:ListBucket", "arn:aws:s3:::bucket-a")) |
|||
|
|||
names, err := store.ListPolicies(ctx, "") |
|||
require.NoError(t, err) |
|||
|
|||
assert.ElementsMatch(t, []string{"canonical-only", "legacy-only", "shared", "bucket-policy:bucket-a"}, names) |
|||
} |
|||
|
|||
func TestFilerPolicyStoreDeletePolicyRemovesCanonicalAndLegacyFiles(t *testing.T) { |
|||
ctx := context.Background() |
|||
store, server := newTestFilerPolicyStore(t) |
|||
|
|||
server.putPolicyFile(t, store.basePath, "dual-format.json", testPolicyDocument("s3:GetObject", "arn:aws:s3:::dual-format/*")) |
|||
server.putPolicyFile(t, store.basePath, "policy_dual-format.json", testPolicyDocument("s3:PutObject", "arn:aws:s3:::dual-format/*")) |
|||
|
|||
require.NoError(t, store.DeletePolicy(ctx, "", "dual-format")) |
|||
assert.False(t, server.hasEntry(store.basePath, "dual-format.json")) |
|||
assert.False(t, server.hasEntry(store.basePath, "policy_dual-format.json")) |
|||
} |
|||
|
|||
func TestFilerPolicyStoreStorePolicyWritesCanonicalFileAndRemovesLegacyTwin(t *testing.T) { |
|||
ctx := context.Background() |
|||
store, server := newTestFilerPolicyStore(t) |
|||
|
|||
server.putPolicyFile(t, store.basePath, "policy_dual-format.json", testPolicyDocument("s3:PutObject", "arn:aws:s3:::dual-format/*")) |
|||
|
|||
require.NoError(t, store.StorePolicy(ctx, "", "dual-format", testPolicyDocument("s3:GetObject", "arn:aws:s3:::dual-format/*"))) |
|||
|
|||
assert.True(t, server.hasEntry(store.basePath, "dual-format.json")) |
|||
assert.False(t, server.hasEntry(store.basePath, "policy_dual-format.json")) |
|||
|
|||
document, err := store.GetPolicy(ctx, "", "dual-format") |
|||
require.NoError(t, err) |
|||
require.Len(t, document.Statement, 1) |
|||
assert.Equal(t, "s3:GetObject", document.Statement[0].Action[0]) |
|||
} |
|||
|
|||
func TestFilerPolicyStoreStorePolicyUpdatesExistingCanonicalFile(t *testing.T) { |
|||
ctx := context.Background() |
|||
store, server := newTestFilerPolicyStore(t) |
|||
|
|||
server.putPolicyFile(t, store.basePath, "existing.json", testPolicyDocument("s3:PutObject", "arn:aws:s3:::existing/*")) |
|||
|
|||
require.NoError(t, store.StorePolicy(ctx, "", "existing", testPolicyDocument("s3:GetObject", "arn:aws:s3:::existing/*"))) |
|||
|
|||
document, err := store.GetPolicy(ctx, "", "existing") |
|||
require.NoError(t, err) |
|||
require.Len(t, document.Statement, 1) |
|||
assert.Equal(t, "s3:GetObject", document.Statement[0].Action[0]) |
|||
assert.Equal(t, "arn:aws:s3:::existing/*", document.Statement[0].Resource[0]) |
|||
} |
|||
|
|||
func TestCopyPolicyDocumentClonesConditionState(t *testing.T) { |
|||
original := &PolicyDocument{ |
|||
Version: "2012-10-17", |
|||
Statement: []Statement{ |
|||
{ |
|||
Effect: "Allow", |
|||
Action: []string{"s3:GetObject"}, |
|||
Resource: []string{ |
|||
"arn:aws:s3:::test-bucket/*", |
|||
}, |
|||
Condition: map[string]map[string]interface{}{ |
|||
"StringEquals": { |
|||
"s3:prefix": []string{"public/", "private/"}, |
|||
}, |
|||
"Null": { |
|||
"aws:PrincipalArn": "false", |
|||
}, |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
copied := copyPolicyDocument(original) |
|||
require.NotNil(t, copied) |
|||
|
|||
original.Statement[0].Condition["StringEquals"]["s3:prefix"] = []string{"mutated/"} |
|||
original.Statement[0].Condition["Null"]["aws:PrincipalArn"] = "true" |
|||
|
|||
assert.Equal(t, []string{"public/", "private/"}, copied.Statement[0].Condition["StringEquals"]["s3:prefix"]) |
|||
assert.Equal(t, "false", copied.Statement[0].Condition["Null"]["aws:PrincipalArn"]) |
|||
} |
|||
|
|||
func TestIsAlreadyExistsPolicyStoreErrorUsesStatusCode(t *testing.T) { |
|||
assert.True(t, isAlreadyExistsPolicyStoreError(status.Error(codes.AlreadyExists, "entry already exists"))) |
|||
assert.False(t, isAlreadyExistsPolicyStoreError(fmt.Errorf("entry already exists"))) |
|||
} |
|||
|
|||
func testPolicyDocument(action string, resource string) *PolicyDocument { |
|||
return &PolicyDocument{ |
|||
Version: "2012-10-17", |
|||
Statement: []Statement{ |
|||
{ |
|||
Effect: "Allow", |
|||
Action: []string{action}, |
|||
Resource: []string{resource}, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
|
|||
func clonePolicyStoreEntry(entry *filer_pb.Entry) *filer_pb.Entry { |
|||
if entry == nil { |
|||
return nil |
|||
} |
|||
return proto.Clone(entry).(*filer_pb.Entry) |
|||
} |
|||
|
|||
func policyStoreTestEntryKey(dir string, name string) string { |
|||
return dir + "\x00" + name |
|||
} |
|||
|
|||
func splitPolicyStoreEntryKey(key string) (string, string) { |
|||
for i := 0; i < len(key); i++ { |
|||
if key[i] == '\x00' { |
|||
return key[:i], key[i+1:] |
|||
} |
|||
} |
|||
return key, "" |
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue