diff --git a/weed/credential/propagating_store.go b/weed/credential/propagating_store.go index 8d42feb47..f7ce1a7be 100644 --- a/weed/credential/propagating_store.go +++ b/weed/credential/propagating_store.go @@ -13,8 +13,11 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/wdclient" "google.golang.org/grpc" + "google.golang.org/grpc/metadata" ) var _ CredentialStore = &PropagatingCredentialStore{} @@ -48,6 +51,24 @@ func (s *PropagatingCredentialStore) SetFilerAddressFunc(getFiler func() pb.Serv } } +// withIamCacheAdminAuth attaches a Bearer token signed with jwt.filer_signing.key +// to the outgoing context so the S3 gateway's IAM-cache gRPC handlers accept the +// propagation. With no key configured it is a no-op, matching the S3 handler's +// checkAdminAuth. Returns the token's lifetime (0 = no expiry) so callers can +// cap any downstream timeout below it. +func withIamCacheAdminAuth(ctx context.Context) (context.Context, time.Duration) { + signingKey := util.GetViper().GetString("jwt.filer_signing.key") + if signingKey == "" { + return ctx, 0 + } + expiresAfterSec := util.GetViper().GetInt("jwt.filer_signing.expires_after_seconds") + token := security.GenJwtForFilerAdmin(security.SigningKey(signingKey), expiresAfterSec) + if token == "" { + return ctx, 0 + } + return metadata.AppendToOutgoingContext(ctx, "authorization", security.BearerPrefix+string(token)), time.Duration(expiresAfterSec) * time.Second +} + func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn func(context.Context, s3_pb.SeaweedS3IamCacheClient) error) { if s.masterClient == nil { return @@ -77,8 +98,16 @@ func (s *PropagatingCredentialStore) propagateChange(ctx context.Context, fn fun } glog.V(1).Infof("IAM: propagating change to %d S3 servers: %v", len(s3Servers), s3Servers) - // Create context with timeout for the propagation process - propagateCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + // Mint the admin token after master discovery so master retries can't burn + // through the token lifetime before the peer fan-out begins. Cap the + // propagation deadline below the token's expiry so slower peers don't see + // an expired token. + authedCtx, tokenTTL := withIamCacheAdminAuth(ctx) + propagateTimeout := 10 * time.Second + if tokenTTL > 0 && tokenTTL < propagateTimeout { + propagateTimeout = tokenTTL + } + propagateCtx, cancel := context.WithTimeout(authedCtx, propagateTimeout) defer cancel() var wg sync.WaitGroup diff --git a/weed/credential/propagating_store_auth_test.go b/weed/credential/propagating_store_auth_test.go new file mode 100644 index 000000000..f69e5a618 --- /dev/null +++ b/weed/credential/propagating_store_auth_test.go @@ -0,0 +1,64 @@ +package credential + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/metadata" +) + +func TestWithIamCacheAdminAuth_NoKey_NoOp(t *testing.T) { + util.GetViper().Set("jwt.filer_signing.key", "") + ctx, ttl := withIamCacheAdminAuth(context.Background()) + if ttl != 0 { + t.Fatalf("expected zero TTL without key, got %v", ttl) + } + md, ok := metadata.FromOutgoingContext(ctx) + if ok && len(md.Get("authorization")) > 0 { + t.Fatalf("expected no authorization metadata without key, got %v", md.Get("authorization")) + } +} + +func TestWithIamCacheAdminAuth_WithKey_AttachesBearer(t *testing.T) { + const k = "propagation-test-signing-key" + util.GetViper().Set("jwt.filer_signing.key", k) + defer util.GetViper().Set("jwt.filer_signing.key", "") + util.GetViper().Set("jwt.filer_signing.expires_after_seconds", 60) + + ctx, ttl := withIamCacheAdminAuth(context.Background()) + if ttl != 60*time.Second { + t.Fatalf("expected 60s TTL, got %v", ttl) + } + md, ok := metadata.FromOutgoingContext(ctx) + if !ok { + t.Fatal("expected outgoing metadata to be set") + } + headers := md.Get("authorization") + if len(headers) != 1 { + t.Fatalf("expected one authorization header, got %v", headers) + } + if !strings.HasPrefix(headers[0], security.BearerPrefix) { + t.Fatalf("expected Bearer scheme, got %q", headers[0]) + } + token := strings.TrimPrefix(headers[0], security.BearerPrefix) + parsed, err := security.DecodeJwt(security.SigningKey(k), security.EncodedJwt(token), &security.SeaweedFilerAdminClaims{}) + if err != nil || parsed == nil || !parsed.Valid { + t.Fatalf("attached token failed signature validation: %v", err) + } +} + +func TestWithIamCacheAdminAuth_ZeroExpiry_ZeroTTL(t *testing.T) { + const k = "propagation-test-signing-key" + util.GetViper().Set("jwt.filer_signing.key", k) + defer util.GetViper().Set("jwt.filer_signing.key", "") + util.GetViper().Set("jwt.filer_signing.expires_after_seconds", 0) + + _, ttl := withIamCacheAdminAuth(context.Background()) + if ttl != 0 { + t.Fatalf("expected zero TTL for no-expiry token, got %v", ttl) + } +} diff --git a/weed/s3api/s3api_server_grpc.go b/weed/s3api/s3api_server_grpc.go index 2aae60b74..65a7b60d5 100644 --- a/weed/s3api/s3api_server_grpc.go +++ b/weed/s3api/s3api_server_grpc.go @@ -2,10 +2,13 @@ package s3api import ( "context" + "strings" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "github.com/seaweedfs/seaweedfs/weed/security" "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" ) @@ -13,11 +16,46 @@ import ( // This interface is dedicated to UNIDIRECTIONAL updates from Filer to S3 Server. // S3 Server acts purely as a cache. +// checkAdminAuth verifies the caller presented a Bearer token signed by the +// filer write-signing key (jwt.filer_signing.key). It mirrors the filer's +// IamGrpcServer.checkAdminAuth so the same operator knob that locks down the +// filer IAM gRPC service also locks down this cache. With no key configured the +// check is a no-op, matching the rest of SeaweedFS's gRPC surface. +func (s3a *S3ApiServer) checkAdminAuth(ctx context.Context) error { + if s3a.filerGuard == nil { + return nil + } + signingKey := s3a.filerGuard.SigningKey() + if len(signingKey) == 0 { + return nil + } + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return status.Error(codes.Unauthenticated, "missing metadata") + } + authHeaders := md.Get("authorization") + if len(authHeaders) == 0 { + return status.Error(codes.Unauthenticated, "missing authorization metadata") + } + raw := strings.TrimSpace(authHeaders[0]) + parts := strings.Fields(raw) + if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") || parts[1] == "" { + return status.Error(codes.Unauthenticated, "authorization header must use Bearer scheme") + } + parsed, err := security.DecodeJwt(signingKey, security.EncodedJwt(parts[1]), &security.SeaweedFilerAdminClaims{}) + if err != nil || parsed == nil || !parsed.Valid { + return status.Error(codes.Unauthenticated, "invalid admin token") + } + return nil +} + func (s3a *S3ApiServer) PutIdentity(ctx context.Context, req *iam_pb.PutIdentityRequest) (*iam_pb.PutIdentityResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.Identity == nil { return nil, status.Errorf(codes.InvalidArgument, "identity is required") } - // Direct in-memory cache update glog.V(1).Infof("IAM: received identity update for %s", req.Identity.Name) if err := s3a.iam.UpsertIdentity(req.Identity); err != nil { glog.Errorf("failed to update identity cache for %s: %v", req.Identity.Name, err) @@ -27,16 +65,21 @@ func (s3a *S3ApiServer) PutIdentity(ctx context.Context, req *iam_pb.PutIdentity } func (s3a *S3ApiServer) RemoveIdentity(ctx context.Context, req *iam_pb.RemoveIdentityRequest) (*iam_pb.RemoveIdentityResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.Username == "" { return nil, status.Errorf(codes.InvalidArgument, "username is required") } - // Direct in-memory cache update glog.V(1).Infof("IAM: received identity removal for %s", req.Username) s3a.iam.RemoveIdentity(req.Username) return &iam_pb.RemoveIdentityResponse{}, nil } func (s3a *S3ApiServer) PutPolicy(ctx context.Context, req *iam_pb.PutPolicyRequest) (*iam_pb.PutPolicyResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.Name == "" { return nil, status.Errorf(codes.InvalidArgument, "policy name is required") } @@ -44,7 +87,6 @@ func (s3a *S3ApiServer) PutPolicy(ctx context.Context, req *iam_pb.PutPolicyRequ return nil, status.Errorf(codes.InvalidArgument, "policy content is required") } - // Update IAM policy cache glog.V(1).Infof("IAM: received policy update for %s", req.Name) if s3a.iam == nil { return nil, status.Errorf(codes.Internal, "IAM not initialized") @@ -58,11 +100,13 @@ func (s3a *S3ApiServer) PutPolicy(ctx context.Context, req *iam_pb.PutPolicyRequ } func (s3a *S3ApiServer) DeletePolicy(ctx context.Context, req *iam_pb.DeletePolicyRequest) (*iam_pb.DeletePolicyResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.Name == "" { return nil, status.Errorf(codes.InvalidArgument, "policy name is required") } - // Delete from IAM policy cache glog.V(1).Infof("IAM: received policy removal for %s", req.Name) if s3a.iam == nil { return nil, status.Errorf(codes.Internal, "IAM not initialized") @@ -76,6 +120,9 @@ func (s3a *S3ApiServer) DeletePolicy(ctx context.Context, req *iam_pb.DeletePoli } func (s3a *S3ApiServer) GetPolicy(ctx context.Context, req *iam_pb.GetPolicyRequest) (*iam_pb.GetPolicyResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.Name == "" { return nil, status.Errorf(codes.InvalidArgument, "policy name is required") } @@ -93,6 +140,9 @@ func (s3a *S3ApiServer) GetPolicy(ctx context.Context, req *iam_pb.GetPolicyRequ } func (s3a *S3ApiServer) PutGroup(ctx context.Context, req *iam_pb.PutGroupRequest) (*iam_pb.PutGroupResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.Group == nil { return nil, status.Errorf(codes.InvalidArgument, "group is required") } @@ -108,6 +158,9 @@ func (s3a *S3ApiServer) PutGroup(ctx context.Context, req *iam_pb.PutGroupReques } func (s3a *S3ApiServer) RemoveGroup(ctx context.Context, req *iam_pb.RemoveGroupRequest) (*iam_pb.RemoveGroupResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } if req.GroupName == "" { return nil, status.Errorf(codes.InvalidArgument, "group name is required") } @@ -120,6 +173,9 @@ func (s3a *S3ApiServer) RemoveGroup(ctx context.Context, req *iam_pb.RemoveGroup } func (s3a *S3ApiServer) ListPolicies(ctx context.Context, req *iam_pb.ListPoliciesRequest) (*iam_pb.ListPoliciesResponse, error) { + if err := s3a.checkAdminAuth(ctx); err != nil { + return nil, err + } resp := &iam_pb.ListPoliciesResponse{} if s3a.iam == nil { return nil, status.Errorf(codes.Internal, "IAM not initialized") diff --git a/weed/s3api/s3api_server_grpc_test.go b/weed/s3api/s3api_server_grpc_test.go new file mode 100644 index 000000000..f40971ee4 --- /dev/null +++ b/weed/s3api/s3api_server_grpc_test.go @@ -0,0 +1,206 @@ +package s3api + +import ( + "context" + "testing" + "time" + + jwt "github.com/golang-jwt/jwt/v5" + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +const testS3IamCacheSigningKey = "s3-iam-cache-test-key-do-not-use-in-prod" + +func newTestS3IamCacheServer(t *testing.T, signingKey string) *S3ApiServer { + t.Helper() + s3a := newTestS3ApiServerWithMemoryIAM(t, nil) + s3a.filerGuard = security.NewGuard(nil, signingKey, 10, "", 60) + return s3a +} + +func s3IamCacheBearerCtx(token string) context.Context { + md := metadata.New(map[string]string{"authorization": "Bearer " + token}) + return metadata.NewIncomingContext(context.Background(), md) +} + +func TestS3IamCache_NoMetadata_Unauthenticated(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.PutIdentity(context.Background(), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pwn", Actions: []string{"Admin"}}, + }) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutIdentity without metadata: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_MissingAuthorizationHeader_Unauthenticated(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + ctx := metadata.NewIncomingContext(context.Background(), metadata.New(map[string]string{"other": "value"})) + _, err := s.PutIdentity(ctx, &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pwn", Actions: []string{"Admin"}}, + }) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutIdentity with no authorization header: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_NonBearerAuthorization_Unauthenticated(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + md := metadata.New(map[string]string{"authorization": "Basic dXNlcjpwYXNz"}) + ctx := metadata.NewIncomingContext(context.Background(), md) + _, err := s.PutIdentity(ctx, &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pwn", Actions: []string{"Admin"}}, + }) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutIdentity with non-Bearer scheme: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_InvalidToken_Unauthenticated(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + bad := security.GenJwtForFilerAdmin(security.SigningKey("a-different-key"), 60) + if bad == "" { + t.Fatal("GenJwtForFilerAdmin returned empty") + } + _, err := s.PutIdentity(s3IamCacheBearerCtx(string(bad)), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pwn", Actions: []string{"Admin"}}, + }) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutIdentity with mis-signed token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_GarbageToken_Unauthenticated(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.PutIdentity(s3IamCacheBearerCtx("not.a.jwt"), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pwn", Actions: []string{"Admin"}}, + }) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutIdentity with garbage token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_ExpiredToken_Unauthenticated(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + claims := security.SeaweedFilerAdminClaims{ + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(time.Now().Add(-time.Hour)), + }, + } + tok := jwt.NewWithClaims(jwt.SigningMethodHS256, claims) + encoded, err := tok.SignedString([]byte(testS3IamCacheSigningKey)) + if err != nil { + t.Fatalf("SignedString: %v", err) + } + _, err = s.PutIdentity(s3IamCacheBearerCtx(encoded), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pwn", Actions: []string{"Admin"}}, + }) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutIdentity with expired token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_ValidToken_WritesIdentity(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + good := security.GenJwtForFilerAdmin(security.SigningKey(testS3IamCacheSigningKey), 60) + if good == "" { + t.Fatal("GenJwtForFilerAdmin returned empty") + } + _, err := s.PutIdentity(s3IamCacheBearerCtx(string(good)), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{ + Name: "pushed", + Actions: []string{"Admin"}, + Credentials: []*iam_pb.Credential{ + {AccessKey: "AKIAPUSHED", SecretKey: "s3cr3t"}, + }, + }, + }) + if err != nil { + t.Fatalf("PutIdentity with valid token: unexpected error %v", err) + } + s.iam.m.RLock() + id := s.iam.accessKeyIdent["AKIAPUSHED"] + s.iam.m.RUnlock() + if id == nil { + t.Fatalf("expected pushed identity to land in accessKeyIdent map") + } + if !id.isAdmin() { + t.Fatalf("expected pushed identity to be admin, actions=%v", id.Actions) + } +} + +func TestS3IamCache_NoSigningKey_Unauthenticated_Allowed(t *testing.T) { + s := newTestS3IamCacheServer(t, "") + _, err := s.PutIdentity(context.Background(), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pushed", Actions: []string{"Read"}}, + }) + if err != nil { + t.Fatalf("PutIdentity without key: unexpected error %v", err) + } + good := security.GenJwtForFilerAdmin(security.SigningKey(testS3IamCacheSigningKey), 60) + if _, err := s.PutIdentity(s3IamCacheBearerCtx(string(good)), &iam_pb.PutIdentityRequest{ + Identity: &iam_pb.Identity{Name: "pushed2", Actions: []string{"Read"}}, + }); err != nil { + t.Fatalf("PutIdentity with stray token but no server key: unexpected error %v", err) + } +} + +func TestS3IamCache_RemoveIdentity_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.RemoveIdentity(context.Background(), &iam_pb.RemoveIdentityRequest{Username: "anyone"}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("RemoveIdentity without token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_PutPolicy_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.PutPolicy(context.Background(), &iam_pb.PutPolicyRequest{Name: "p", Content: "{}"}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutPolicy without token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_DeletePolicy_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.DeletePolicy(context.Background(), &iam_pb.DeletePolicyRequest{Name: "p"}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("DeletePolicy without token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_GetPolicy_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.GetPolicy(context.Background(), &iam_pb.GetPolicyRequest{Name: "p"}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("GetPolicy without token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_ListPolicies_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.ListPolicies(context.Background(), &iam_pb.ListPoliciesRequest{}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("ListPolicies without token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_PutGroup_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.PutGroup(context.Background(), &iam_pb.PutGroupRequest{Group: &iam_pb.Group{Name: "g"}}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("PutGroup without token: got code %v, want %v (err=%v)", got, want, err) + } +} + +func TestS3IamCache_RemoveGroup_RequiresAuth(t *testing.T) { + s := newTestS3IamCacheServer(t, testS3IamCacheSigningKey) + _, err := s.RemoveGroup(context.Background(), &iam_pb.RemoveGroupRequest{GroupName: "g"}) + if got, want := status.Code(err), codes.Unauthenticated; got != want { + t.Fatalf("RemoveGroup without token: got code %v, want %v (err=%v)", got, want, err) + } +}