diff --git a/go.mod b/go.mod index af03c4297..0f8fdddde 100644 --- a/go.mod +++ b/go.mod @@ -39,7 +39,7 @@ require ( github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-multierror v1.1.1 // indirect github.com/hashicorp/go-uuid v1.0.3 // indirect - github.com/jackc/pgx/v5 v5.9.0 + github.com/jackc/pgx/v5 v5.9.2 github.com/jcmturner/gofork v1.7.6 // indirect github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect github.com/jinzhu/copier v0.4.0 diff --git a/go.sum b/go.sum index 3b6f9b97d..1c30d438c 100644 --- a/go.sum +++ b/go.sum @@ -1403,8 +1403,8 @@ github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsI github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= -github.com/jackc/pgx/v5 v5.9.0 h1:T/dI+2TvmI2H8s/KH1/lXIbz1CUFk3gn5oTjr0/mBsE= -github.com/jackc/pgx/v5 v5.9.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw= +github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= github.com/jaegertracing/jaeger v1.47.0 h1:XXxTMO+GxX930gxKWsg90rFr6RswkCRIW0AgWFnTYsg= diff --git a/note/slides/Makefile b/note/slides/Makefile new file mode 100644 index 000000000..3b5c62df0 --- /dev/null +++ b/note/slides/Makefile @@ -0,0 +1,31 @@ +MARP ?= npx @marp-team/marp-cli +SRC := seaweedfs-production-setup.md +PDF := $(SRC:.md=.pdf) +HTML := $(SRC:.md=.html) +PPTX := $(SRC:.md=.pptx) +FLAGS := --allow-local-files + +.PHONY: all pdf html pptx watch clean + +all: pdf + +pdf: $(PDF) + +html: $(HTML) + +pptx: $(PPTX) + +$(PDF): $(SRC) + $(MARP) $(SRC) --pdf $(FLAGS) -o $@ + +$(HTML): $(SRC) + $(MARP) $(SRC) --html $(FLAGS) -o $@ + +$(PPTX): $(SRC) + $(MARP) $(SRC) --pptx $(FLAGS) -o $@ + +watch: + $(MARP) $(SRC) --watch --html $(FLAGS) + +clean: + rm -f $(PDF) $(HTML) $(PPTX) diff --git a/note/slides/object-store-layout.png b/note/slides/object-store-layout.png new file mode 100644 index 000000000..d02403b8a Binary files /dev/null and b/note/slides/object-store-layout.png differ diff --git a/note/slides/seaweedfs-production-setup.md b/note/slides/seaweedfs-production-setup.md new file mode 100644 index 000000000..0155120b2 --- /dev/null +++ b/note/slides/seaweedfs-production-setup.md @@ -0,0 +1,118 @@ +--- +marp: true +theme: default +paginate: true +--- + +# SeaweedFS Production Setup + +A walkthrough of the recommended deployment topology + +- Storage layer +- File access layer +- Backend operations +- Erasure coding for durability + +--- + +![bg contain](object-store-layout.png) + +--- + +# Storage Layer + +**3 Masters + Volume Servers** + +- **Masters (run 3)** + - Form a Raft quorum; tolerate 1 failure + - Track volume locations and cluster topology + - Lightweight; not on the data path for reads/writes after lookup + +- **Volume Servers (scale out)** + - Hold the actual needle data on local disks + - Add more servers for capacity and throughput + - Heartbeat to masters; clients talk to them directly for I/O + +Rule of thumb: start with 3 masters, then grow volume servers with data. + +--- + +# File Access Layer + +**Filer + DB for metadata** + +- Filer serves POSIX-like directory and file metadata +- Metadata store options: PostgreSQL, MySQL, Redis, Cassandra, TiKV, etc. +- Run **3 filers** for HA +- Filers are stateless relative to each other — state lives in the DB + +**S3 Servers** + +- Expose the S3 API on top of the filer +- Scale the count based on throughput requirements +- Can be co-located with filers or run independently + +--- + +# Backend Operations + +**1 Admin Server + a few Workers** + +- **Admin server (1 is enough)** + - Coordinates background jobs: balancing, EC encoding, vacuum, replication fixes + - **Not on the data path** — a restart does not affect object store reads/writes + - No HA required for normal operation + +- **Workers (a few)** + - Execute long-running tasks issued by the admin server + - Horizontally scalable based on maintenance workload + +Keeps housekeeping off the hot path. + +--- + +# Production Topology Summary + +| Component | Count | Role | +|----------------|----------------------|-----------------------------------| +| Master | 3 | Raft quorum, topology, volume map | +| Volume Server | N (scale out) | Stores data needles | +| Filer | 3 | Metadata gateway | +| Metadata DB | HA cluster | Persists filer metadata | +| S3 Server | N (by throughput) | S3 API frontend | +| Admin | 1 | Background job coordinator | +| Worker | A few | Executes admin-scheduled tasks | + +--- + +# Erasure Coding for Durability + +Prefer **many volume servers** so EC shards spread across failure domains. + +**Example: 8 volume servers, EC 5+3** + +- Each volume becomes 5 data shards + 3 parity shards +- Shards are placed on 8 distinct volume servers +- **Tolerates up to 3 simultaneous volume server failures** +- Storage overhead: 1.6x vs 3x for full replication + +``` +Volume -> [D1][D2][D3][D4][D5] + [P1][P2][P3] + | | | | | | | | + VS1 VS2 VS3 VS4 VS5 VS6 VS7 VS8 +``` + +More volume servers = more EC layouts available (e.g. 10+4, 6+3). + +--- + +# Getting Started Checklist + +1. Provision 3 master nodes (small boxes are fine) +2. Provision volume servers sized to your data footprint +3. Stand up an HA metadata DB, then 3 filers pointing at it +4. Add S3 servers sized to required throughput +5. Run 1 admin + a few workers for background maintenance +6. Once you have enough volume servers, enable EC (e.g. 5+3 on 8 servers) + +Scale each layer independently as usage grows. diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index c06ec5f6e..d152c9cb0 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1425,6 +1425,16 @@ impl VolumeServer for VolumeGrpcService { // Determine file path let path = if info.is_ec_volume { let store = self.state.store.read().unwrap(); + // std::fs::File::create truncates in place; a mounted + // EcVolume holds fds on the same inodes, so overwriting + // corrupts live readers. + if store.has_ec_volume(VolumeId(info.volume_id)) { + resp_error = Some(format!( + "ec volume {} is mounted; unmount before ReceiveFile", + info.volume_id + )); + break; + } // disk_id=0 means "unset" (protobuf default), so auto-select // mirrors VolumeEcShardsCopy: prefer a disk already holding // this volume's shards, then any HDD, then any disk. diff --git a/test/volume_server/grpc/ec_receive_truncates_mounted_test.go b/test/volume_server/grpc/ec_receive_truncates_mounted_test.go new file mode 100644 index 000000000..d207ae0c2 --- /dev/null +++ b/test/volume_server/grpc/ec_receive_truncates_mounted_test.go @@ -0,0 +1,186 @@ +package volume_server_grpc_test + +import ( + "context" + "fmt" + "net/http" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" +) + +func TestReceiveFileRejectsOverwriteOfMountedEcShard(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const volumeID = uint32(91841) + framework.AllocateVolume(t, grpcClient, volumeID, "") + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 918401, 0x9184CAFE) + uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, + []byte("ec-receive-overwrite-content-for-issue-9184-repro")) + _ = framework.ReadAllAndClose(t, uploadResp) + if uploadResp.StatusCode != http.StatusCreated { + t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + if _, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{ + VolumeId: volumeID, + Collection: "", + }); err != nil { + t.Fatalf("VolumeEcShardsGenerate: %v", err) + } + + if _, err := grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{ + VolumeId: volumeID, + Collection: "", + ShardIds: []uint32{0}, + }); err != nil { + t.Fatalf("VolumeEcShardsMount: %v", err) + } + + dataDir := filepath.Join(clusterHarness.BaseDir(), "volume") + shardPath := filepath.Join(dataDir, fmt.Sprintf("%d.ec00", volumeID)) + origInfo, err := os.Stat(shardPath) + if err != nil { + t.Fatalf("stat mounted shard %s: %v", shardPath, err) + } + origSize := origInfo.Size() + if origSize == 0 { + t.Fatalf("mounted shard %s unexpectedly empty", shardPath) + } + + readStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{ + VolumeId: volumeID, + ShardId: 0, + Offset: 0, + Size: 1, + }) + if err != nil { + t.Fatalf("VolumeEcShardRead (pre): %v", err) + } + if _, err := readStream.Recv(); err != nil { + t.Fatalf("VolumeEcShardRead Recv (pre): %v", err) + } + + overwritePayload := []byte("bug-9184-overwrite") + receiveStream, err := grpcClient.ReceiveFile(ctx) + if err != nil { + t.Fatalf("ReceiveFile stream create: %v", err) + } + if err = receiveStream.Send(&volume_server_pb.ReceiveFileRequest{ + Data: &volume_server_pb.ReceiveFileRequest_Info{ + Info: &volume_server_pb.ReceiveFileInfo{ + VolumeId: volumeID, + Ext: ".ec00", + Collection: "", + IsEcVolume: true, + ShardId: 0, + FileSize: uint64(len(overwritePayload)), + }, + }, + }); err != nil { + t.Fatalf("ReceiveFile send info: %v", err) + } + resp, err := receiveStream.CloseAndRecv() + if err != nil { + t.Logf("ReceiveFile rejected at stream level: %v", err) + } else { + if resp.GetError() == "" { + t.Fatalf("expected ReceiveFile to reject overwrite of mounted shard, got success: %+v", resp) + } + if !strings.Contains(resp.GetError(), "mounted") { + t.Fatalf("expected error to mention mounted; got: %s", resp.GetError()) + } + } + + afterInfo, err := os.Stat(shardPath) + if err != nil { + t.Fatalf("stat shard after rejected overwrite: %v", err) + } + if afterInfo.Size() != origSize { + t.Fatalf("shard %s was modified despite rejection: size was %d, now %d", + shardPath, origSize, afterInfo.Size()) + } + + postStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{ + VolumeId: volumeID, + ShardId: 0, + Offset: 0, + Size: 1, + }) + if err != nil { + t.Fatalf("VolumeEcShardRead (post): %v", err) + } + if _, err := postStream.Recv(); err != nil { + t.Fatalf("VolumeEcShardRead Recv (post): %v", err) + } + + t.Logf("ReceiveFile correctly refused overwrite; mounted shard intact at %d bytes", afterInfo.Size()) +} + +func TestReceiveFileAllowsEcShardWhenNoMount(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + const volumeID = uint32(91843) + const collection = "ec-receive-no-mount" + payload := []byte("ok-to-receive-not-mounted") + + stream, err := grpcClient.ReceiveFile(ctx) + if err != nil { + t.Fatalf("ReceiveFile stream create: %v", err) + } + if err = stream.Send(&volume_server_pb.ReceiveFileRequest{ + Data: &volume_server_pb.ReceiveFileRequest_Info{ + Info: &volume_server_pb.ReceiveFileInfo{ + VolumeId: volumeID, + Ext: ".ec00", + Collection: collection, + IsEcVolume: true, + ShardId: 0, + FileSize: uint64(len(payload)), + }, + }, + }); err != nil { + t.Fatalf("ReceiveFile send info: %v", err) + } + if err = stream.Send(&volume_server_pb.ReceiveFileRequest{ + Data: &volume_server_pb.ReceiveFileRequest_FileContent{FileContent: payload}, + }); err != nil { + t.Fatalf("ReceiveFile send content: %v", err) + } + resp, err := stream.CloseAndRecv() + if err != nil { + t.Fatalf("ReceiveFile close: %v", err) + } + if resp.GetError() != "" { + t.Fatalf("expected success on unmounted volume, got error: %s", resp.GetError()) + } + if resp.GetBytesWritten() != uint64(len(payload)) { + t.Fatalf("bytes_written mismatch: got %d want %d", resp.GetBytesWritten(), len(payload)) + } +} diff --git a/weed/admin/dash/user_management.go b/weed/admin/dash/user_management.go index 009f5cfca..d1b599acc 100644 --- a/weed/admin/dash/user_management.go +++ b/weed/admin/dash/user_management.go @@ -247,21 +247,29 @@ func (s *AdminServer) CreateAccessKey(username string, req *CreateAccessKeyReque req = &CreateAccessKeyRequest{} } - // Validate provided keys - if req.AccessKey != "" && (len(req.AccessKey) < 4 || len(req.AccessKey) > 128) { - return nil, fmt.Errorf("access key must be between 4 and 128 characters: %w", ErrInvalidInput) + // Validate provided keys (shared with the IAM API and embedded IAM paths). + if req.AccessKey != "" { + if err := iam.ValidateCallerSuppliedAccessKeyId(req.AccessKey); err != nil { + return nil, fmt.Errorf("%s: %w", err.Error(), ErrInvalidInput) + } } - if req.SecretKey != "" && (len(req.SecretKey) < 8 || len(req.SecretKey) > 128) { - return nil, fmt.Errorf("secret key must be between 8 and 128 characters: %w", ErrInvalidInput) + if req.SecretKey != "" { + if err := iam.ValidateCallerSuppliedSecretAccessKey(req.SecretKey); err != nil { + return nil, fmt.Errorf("%s: %w", err.Error(), ErrInvalidInput) + } + } + // Enforce the both-or-none rule to match the IAM API and embedded IAM + // paths — silently generating the missing half lets a caller end up + // with a credential they did not fully choose. + if (req.AccessKey != "") != (req.SecretKey != "") { + return nil, fmt.Errorf("access key and secret key must be supplied together: %w", ErrInvalidInput) } // Use provided keys or generate new ones accessKey := req.AccessKey + secretKey := req.SecretKey if accessKey == "" { accessKey = generateAccessKey() - } - secretKey := req.SecretKey - if secretKey == "" { secretKey = generateSecretKey() } diff --git a/weed/filer/redis3/kv_directory_children_test.go b/weed/filer/redis3/kv_directory_children_test.go index 3a42487db..e1a7e988e 100644 --- a/weed/filer/redis3/kv_directory_children_test.go +++ b/weed/filer/redis3/kv_directory_children_test.go @@ -2,109 +2,14 @@ package redis3 import ( "context" - "fmt" "strconv" "testing" - "time" "github.com/redis/go-redis/v9" "github.com/stvp/tempredis" ) -var names = []string{ - "cassandra.in.sh", - "cassandra", - "debug-cql.bat", - "nodetool", - "nodetool.bat", - "source-conf.ps1", - "sstableloader", - "sstableloader.bat", - "sstablescrub", - "sstablescrub.bat", - "sstableupgrade", - "sstableupgrade.bat", - "sstableutil", - "sstableutil.bat", - "sstableverify", - "sstableverify.bat", - "stop-server", - "stop-server.bat", - "stop-server.ps1", - "cassandra.in.bat", - "cqlsh.py", - "cqlsh", - "cassandra.ps1", - "cqlsh.bat", - "debug-cql", - "cassandra.bat", -} - -func yTestNameList(t *testing.T) { - server, err := tempredis.Start(tempredis.Config{}) - if err != nil { - panic(err) - } - defer server.Term() - - client := redis.NewClient(&redis.Options{ - Network: "unix", - Addr: server.Socket(), - }) - - store := newSkipListElementStore("/yyy/bin", client) - var data []byte - for _, name := range names { - nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit) - nameList.WriteName(name) - - nameList.ListNames("", func(name string) bool { - println(name) - return true - }) - - if nameList.HasChanges() { - data = nameList.ToBytes() - } - println() - } - - nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit) - nameList.ListNames("", func(name string) bool { - println(name) - return true - }) - -} - -func yBenchmarkNameList(b *testing.B) { - - server, err := tempredis.Start(tempredis.Config{}) - if err != nil { - panic(err) - } - defer server.Term() - - client := redis.NewClient(&redis.Options{ - Network: "unix", - Addr: server.Socket(), - }) - - store := newSkipListElementStore("/yyy/bin", client) - var data []byte - for i := 0; i < b.N; i++ { - nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit) - - nameList.WriteName(strconv.Itoa(i) + "namexxxxxxxxxxxxxxxxxxx") - - if nameList.HasChanges() { - data = nameList.ToBytes() - } - } -} - func BenchmarkRedis(b *testing.B) { - server, err := tempredis.Start(tempredis.Config{}) if err != nil { panic(err) @@ -120,92 +25,3 @@ func BenchmarkRedis(b *testing.B) { client.ZAddNX(context.Background(), "/yyy/bin", redis.Z{Score: 0, Member: strconv.Itoa(i) + "namexxxxxxxxxxxxxxxxxxx"}) } } - -func xTestNameListAdd(t *testing.T) { - - server, err := tempredis.Start(tempredis.Config{}) - if err != nil { - panic(err) - } - defer server.Term() - - client := redis.NewClient(&redis.Options{ - Addr: "localhost:6379", - Password: "", - DB: 0, - }) - - client.FlushAll(context.Background()) - - N := 364800 - - ts0 := time.Now() - store := newSkipListElementStore("/y", client) - var data []byte - nameList := LoadItemList(data, "/y", client, store, 100000) - for i := 0; i < N; i++ { - nameList.WriteName(fmt.Sprintf("%8d", i)) - } - - ts1 := time.Now() - - for i := 0; i < N; i++ { - client.ZAddNX(context.Background(), "/x", redis.Z{Score: 0, Member: fmt.Sprintf("name %8d", i)}) - } - ts2 := time.Now() - - fmt.Printf("%v %v", ts1.Sub(ts0), ts2.Sub(ts1)) - - /* - keys := client.Keys(context.Background(), "/*m").Val() - for _, k := range keys { - println("key", k) - for i, v := range client.ZRangeByLex(context.Background(), k, &redis.ZRangeBy{ - Min: "-", - Max: "+", - }).Val() { - println(" ", i, v) - } - } - */ -} - -func xBenchmarkNameList(b *testing.B) { - - server, err := tempredis.Start(tempredis.Config{}) - if err != nil { - panic(err) - } - defer server.Term() - - client := redis.NewClient(&redis.Options{ - Addr: "localhost:6379", - Password: "", - DB: 0, - }) - - store := newSkipListElementStore("/yyy/bin", client) - var data []byte - for i := 0; i < b.N; i++ { - nameList := LoadItemList(data, "/yyy/bin", client, store, maxNameBatchSizeLimit) - - nameList.WriteName(fmt.Sprintf("name %8d", i)) - - if nameList.HasChanges() { - data = nameList.ToBytes() - } - } -} - -func xBenchmarkRedis(b *testing.B) { - - client := redis.NewClient(&redis.Options{ - Addr: "localhost:6379", - Password: "", - DB: 0, - }) - - for i := 0; i < b.N; i++ { - client.ZAddNX(context.Background(), "/xxx/bin", redis.Z{Score: 0, Member: fmt.Sprintf("name %8d", i)}) - } -} diff --git a/weed/iam/redact.go b/weed/iam/redact.go new file mode 100644 index 000000000..7a27801c8 --- /dev/null +++ b/weed/iam/redact.go @@ -0,0 +1,32 @@ +package iam + +import "net/url" + +// sensitiveFormKeys is the set of IAM request form parameters whose values +// must never be written to logs. Matching is case-sensitive and uses the +// exact AWS IAM parameter name. Extend this when adding IAM actions that +// accept credentials, passwords, session tokens, or private keys. +var sensitiveFormKeys = map[string]struct{}{ + "SecretAccessKey": {}, + "Password": {}, + "NewPassword": {}, + "OldPassword": {}, + "PrivateKey": {}, + "SessionToken": {}, +} + +// RedactSensitiveFormValues returns a shallow copy of values with every +// sensitive key (see sensitiveFormKeys) replaced by "[REDACTED]". Intended +// for debug-level logging of IAM request forms so secrets do not leak into +// log sinks. +func RedactSensitiveFormValues(values url.Values) url.Values { + safe := make(url.Values, len(values)) + for k, v := range values { + if _, sensitive := sensitiveFormKeys[k]; sensitive { + safe[k] = []string{"[REDACTED]"} + } else { + safe[k] = v + } + } + return safe +} diff --git a/weed/iam/redact_test.go b/weed/iam/redact_test.go new file mode 100644 index 000000000..3f5bd7467 --- /dev/null +++ b/weed/iam/redact_test.go @@ -0,0 +1,33 @@ +package iam + +import ( + "net/url" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestRedactSensitiveFormValues(t *testing.T) { + in := url.Values{ + "Action": []string{"CreateAccessKey"}, + "UserName": []string{"alice"}, + "AccessKeyId": []string{"AKIAALICE1"}, + "SecretAccessKey": []string{"very-secret"}, + "Password": []string{"hunter2"}, + "NewPassword": []string{"new-hunter2"}, + "OldPassword": []string{"old-hunter2"}, + "PrivateKey": []string{"-----BEGIN-----"}, + "SessionToken": []string{"tok"}, + } + out := RedactSensitiveFormValues(in) + + assert.Equal(t, []string{"CreateAccessKey"}, out["Action"]) + assert.Equal(t, []string{"alice"}, out["UserName"]) + assert.Equal(t, []string{"AKIAALICE1"}, out["AccessKeyId"]) + for _, k := range []string{"SecretAccessKey", "Password", "NewPassword", "OldPassword", "PrivateKey", "SessionToken"} { + assert.Equal(t, []string{"[REDACTED]"}, out[k], "key %q should be redacted", k) + } + + // Input is not mutated. + assert.Equal(t, []string{"very-secret"}, in["SecretAccessKey"]) +} diff --git a/weed/iam/validation.go b/weed/iam/validation.go new file mode 100644 index 000000000..699d90ae9 --- /dev/null +++ b/weed/iam/validation.go @@ -0,0 +1,68 @@ +package iam + +import ( + "fmt" + + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" +) + +// ValidateCallerSuppliedAccessKeyId checks that a caller-supplied AccessKeyId +// is 4 to 128 ASCII alphanumeric characters. Returns nil if valid. +// +// The alphanumeric restriction avoids characters that would break SigV4 +// canonicalization (e.g. '/' and '=' appear as delimiters in Credential +// headers), so this is a stricter superset of the rule AWS enforces. +func ValidateCallerSuppliedAccessKeyId(accessKeyId string) error { + if len(accessKeyId) < 4 || len(accessKeyId) > 128 { + return fmt.Errorf("AccessKeyId must be 4 to 128 alphanumeric characters") + } + for _, r := range accessKeyId { + if !((r >= 'A' && r <= 'Z') || (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9')) { + return fmt.Errorf("AccessKeyId must be 4 to 128 alphanumeric characters") + } + } + return nil +} + +// ValidateCallerSuppliedSecretAccessKey checks that a caller-supplied +// SecretAccessKey is 8 to 128 characters. Returns nil if valid. +func ValidateCallerSuppliedSecretAccessKey(secretAccessKey string) error { + if len(secretAccessKey) < 8 || len(secretAccessKey) > 128 { + return fmt.Errorf("SecretAccessKey must be between 8 and 128 characters") + } + return nil +} + +// AccessKeyOwner identifies which entity in an S3ApiConfiguration already owns +// a given AccessKeyId. Returned by FindAccessKeyOwner for collision checks on +// caller-supplied credentials. +type AccessKeyOwner struct { + // Type is "user" or "service account". + Type string + // Name is the identity's Name (for users) or the service account's Id. + Name string +} + +// FindAccessKeyOwner scans s3cfg for an identity or service account whose +// credentials already contain accessKeyId. Returns nil if the key is free. +// +// Callers should log Name only at debug level — error responses returned to +// the caller should not include owner identity to avoid information leaks. +func FindAccessKeyOwner(s3cfg *iam_pb.S3ApiConfiguration, accessKeyId string) *AccessKeyOwner { + if s3cfg == nil || accessKeyId == "" { + return nil + } + for _, ident := range s3cfg.Identities { + for _, cred := range ident.Credentials { + if cred.AccessKey == accessKeyId { + return &AccessKeyOwner{Type: "user", Name: ident.Name} + } + } + } + for _, sa := range s3cfg.ServiceAccounts { + if sa.Credential != nil && sa.Credential.AccessKey == accessKeyId { + return &AccessKeyOwner{Type: "service account", Name: sa.Id} + } + } + return nil +} diff --git a/weed/iam/validation_test.go b/weed/iam/validation_test.go new file mode 100644 index 000000000..f5244c016 --- /dev/null +++ b/weed/iam/validation_test.go @@ -0,0 +1,104 @@ +package iam + +import ( + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/iam_pb" + "github.com/stretchr/testify/assert" +) + +func TestValidateCallerSuppliedAccessKeyId(t *testing.T) { + cases := []struct { + name string + input string + wantErr bool + }{ + {"empty", "", true}, + {"three chars", "abc", true}, + {"four chars ok", "abcd", false}, + {"mixed case alnum", "MyAppKey123", false}, + {"128 chars ok", strings.Repeat("a", 128), false}, + {"129 chars too long", strings.Repeat("a", 129), true}, + {"slash rejected", "foo/bar", true}, + {"equals rejected", "foo=bar", true}, + {"dash rejected", "foo-bar", true}, + {"underscore rejected", "foo_bar", true}, + {"unicode rejected", "fooö123", true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := ValidateCallerSuppliedAccessKeyId(tc.input) + if tc.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestValidateCallerSuppliedSecretAccessKey(t *testing.T) { + cases := []struct { + name string + input string + wantErr bool + }{ + {"empty", "", true}, + {"seven chars", "abcdefg", true}, + {"eight chars ok", "abcdefgh", false}, + {"128 chars ok", strings.Repeat("a", 128), false}, + {"129 chars too long", strings.Repeat("a", 129), true}, + {"non-alnum allowed in secret", "sec/ret=1", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := ValidateCallerSuppliedSecretAccessKey(tc.input) + if tc.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestFindAccessKeyOwner(t *testing.T) { + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + { + Name: "alice", + Credentials: []*iam_pb.Credential{ + {AccessKey: "AKIAALICE1", SecretKey: "s"}, + }, + }, + {Name: "bob"}, + }, + ServiceAccounts: []*iam_pb.ServiceAccount{ + {Id: "svc-1", Credential: &iam_pb.Credential{AccessKey: "SVCKEY1"}}, + {Id: "svc-2"}, + }, + } + + t.Run("matches user", func(t *testing.T) { + owner := FindAccessKeyOwner(s3cfg, "AKIAALICE1") + assert.NotNil(t, owner) + assert.Equal(t, "user", owner.Type) + assert.Equal(t, "alice", owner.Name) + }) + t.Run("matches service account", func(t *testing.T) { + owner := FindAccessKeyOwner(s3cfg, "SVCKEY1") + assert.NotNil(t, owner) + assert.Equal(t, "service account", owner.Type) + assert.Equal(t, "svc-1", owner.Name) + }) + t.Run("no match", func(t *testing.T) { + assert.Nil(t, FindAccessKeyOwner(s3cfg, "NOTTAKEN")) + }) + t.Run("empty key", func(t *testing.T) { + assert.Nil(t, FindAccessKeyOwner(s3cfg, "")) + }) + t.Run("nil config", func(t *testing.T) { + assert.Nil(t, FindAccessKeyOwner(nil, "anything")) + }) +} diff --git a/weed/iamapi/iamapi_management_handlers.go b/weed/iamapi/iamapi_management_handlers.go index 8411ebb5c..cee4cc048 100644 --- a/weed/iamapi/iamapi_management_handlers.go +++ b/weed/iamapi/iamapi_management_handlers.go @@ -929,13 +929,38 @@ func (iama *IamApiServer) CreateAccessKey(s3cfg *iam_pb.S3ApiConfiguration, valu userName := values.Get("UserName") status := iam.StatusTypeActive - accessKeyId, err := StringWithCharset(21, charsetUpper) - if err != nil { - return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)} + accessKeyId := values.Get("AccessKeyId") + secretAccessKey := values.Get("SecretAccessKey") + if accessKeyId != "" { + if err := iamlib.ValidateCallerSuppliedAccessKeyId(accessKeyId); err != nil { + return resp, &IamError{Code: iam.ErrCodeInvalidInputException, Error: err} + } } - secretAccessKey, err := StringWithCharset(42, charset) - if err != nil { - return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)} + if secretAccessKey != "" { + if err := iamlib.ValidateCallerSuppliedSecretAccessKey(secretAccessKey); err != nil { + return resp, &IamError{Code: iam.ErrCodeInvalidInputException, Error: err} + } + } + if (accessKeyId != "") != (secretAccessKey != "") { + return resp, &IamError{Code: iam.ErrCodeInvalidInputException, Error: fmt.Errorf("AccessKeyId and SecretAccessKey must be supplied together")} + } + if owner := iamlib.FindAccessKeyOwner(s3cfg, accessKeyId); owner != nil { + glog.V(4).Infof("CreateAccessKey: supplied AccessKeyId already in use by %s %s", owner.Type, owner.Name) + return resp, &IamError{Code: iam.ErrCodeEntityAlreadyExistsException, Error: fmt.Errorf("AccessKeyId is already in use")} + } + if accessKeyId == "" { + var err error + accessKeyId, err = StringWithCharset(21, charsetUpper) + if err != nil { + return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)} + } + } + if secretAccessKey == "" { + var err error + secretAccessKey, err = StringWithCharset(42, charset) + if err != nil { + return resp, &IamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)} + } } resp.CreateAccessKeyResult.AccessKey.AccessKeyId = &accessKeyId @@ -1085,7 +1110,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) { return } - glog.V(4).Infof("DoActions: %+v", values) + glog.V(4).Infof("DoActions: %+v", iamlib.RedactSensitiveFormValues(values)) var response iamlib.RequestIDSetter changed := true switch r.Form.Get("Action") { diff --git a/weed/iamapi/iamapi_management_handlers_test.go b/weed/iamapi/iamapi_management_handlers_test.go index 478bb41c5..9bffd4f85 100644 --- a/weed/iamapi/iamapi_management_handlers_test.go +++ b/weed/iamapi/iamapi_management_handlers_test.go @@ -3,6 +3,7 @@ package iamapi import ( "encoding/json" "net/url" + "strings" "testing" "github.com/aws/aws-sdk-go/service/iam" @@ -629,3 +630,266 @@ func TestListAttachedUserPolicies(t *testing.T) { assert.NotNil(t, iamErr) assert.Equal(t, iam.ErrCodeNoSuchEntityException, iamErr.Code) } + +func TestCreateAccessKeyWithCallerSuppliedKeys(t *testing.T) { + iama := newTestIamApiServer(Policies{}) + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{Name: "alice"}}, + } + values := url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"myappkey"}, + "SecretAccessKey": []string{"mysecret1234"}, + } + resp, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.Nil(t, iamErr) + assert.Equal(t, "myappkey", *resp.CreateAccessKeyResult.AccessKey.AccessKeyId) + assert.Equal(t, "mysecret1234", *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey) + assert.Equal(t, "alice", *resp.CreateAccessKeyResult.AccessKey.UserName) + assert.Equal(t, "myappkey", s3cfg.Identities[0].Credentials[0].AccessKey) + assert.Equal(t, "mysecret1234", s3cfg.Identities[0].Credentials[0].SecretKey) +} + +func TestCreateAccessKeyRandomGeneration(t *testing.T) { + iama := newTestIamApiServer(Policies{}) + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{Name: "alice"}}, + } + values := url.Values{ + "UserName": []string{"alice"}, + } + resp, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.Nil(t, iamErr) + assert.NotEmpty(t, *resp.CreateAccessKeyResult.AccessKey.AccessKeyId) + assert.NotEmpty(t, *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey) + assert.Len(t, s3cfg.Identities[0].Credentials, 1) +} + +func TestCreateAccessKeyRejectsWeakKeys(t *testing.T) { + iama := newTestIamApiServer(Policies{}) + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{Name: "alice"}}, + } + // Too short + values := url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"ab"}, + "SecretAccessKey": []string{"validsecret1"}, + } + _, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + + // Short secret + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"validkey"}, + "SecretAccessKey": []string{"short"}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + + // SigV4 delimiters + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"foo/bar=baz"}, + "SecretAccessKey": []string{"validsecret1"}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) +} + +func TestCreateAccessKeyRejectsCollision(t *testing.T) { + iama := newTestIamApiServer(Policies{}) + // Use a distinctive owner name ("ownerAlpha") that shares no substring + // with the expected error message so the leak assertion is meaningful. + const ownerName = "ownerAlpha" + + t.Run("identity credential", func(t *testing.T) { + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + { + Name: ownerName, + Credentials: []*iam_pb.Credential{ + {AccessKey: "takenkey", SecretKey: "existingsecret"}, + }, + }, + {Name: "newuser"}, + }, + } + values := url.Values{ + "UserName": []string{"newuser"}, + "AccessKeyId": []string{"takenkey"}, + "SecretAccessKey": []string{"newsecret123"}, + } + _, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeEntityAlreadyExistsException, iamErr.Code) + assert.NotContains(t, iamErr.Error.Error(), ownerName, "should not leak owner name") + assert.Len(t, s3cfg.Identities[1].Credentials, 0) + }) + + t.Run("service account credential", func(t *testing.T) { + const saId = "svcAlpha" + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + {Name: "newuser"}, + }, + ServiceAccounts: []*iam_pb.ServiceAccount{ + { + Id: saId, + Credential: &iam_pb.Credential{AccessKey: "takenkey", SecretKey: "existingsecret"}, + }, + }, + } + values := url.Values{ + "UserName": []string{"newuser"}, + "AccessKeyId": []string{"takenkey"}, + "SecretAccessKey": []string{"newsecret123"}, + } + _, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeEntityAlreadyExistsException, iamErr.Code) + assert.NotContains(t, iamErr.Error.Error(), saId, "should not leak owner id") + // The service account's existing credential must be untouched, and + // no new credential should be attached to the identity. + assert.Equal(t, "takenkey", s3cfg.ServiceAccounts[0].Credential.AccessKey) + assert.Equal(t, "existingsecret", s3cfg.ServiceAccounts[0].Credential.SecretKey) + assert.Len(t, s3cfg.Identities[0].Credentials, 0) + }) +} + +func TestCreateAccessKeyBoundary(t *testing.T) { + iama := newTestIamApiServer(Policies{}) + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{Name: "alice"}}, + } + // Exactly 4 chars - should pass + values := url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"abcd"}, + "SecretAccessKey": []string{"secretkey123"}, + } + resp, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.Nil(t, iamErr) + assert.Equal(t, "abcd", *resp.CreateAccessKeyResult.AccessKey.AccessKeyId) + + // Exactly 3 chars - should fail + s3cfg.Identities[0].Credentials = nil + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"abc"}, + "SecretAccessKey": []string{"secretkey123"}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + + // Exactly 128 chars - should pass + s3cfg.Identities[0].Credentials = nil + ak128 := strings.Repeat("a", 128) + sk128 := strings.Repeat("s", 128) + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{ak128}, + "SecretAccessKey": []string{sk128}, + } + resp, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.Nil(t, iamErr) + assert.Equal(t, ak128, *resp.CreateAccessKeyResult.AccessKey.AccessKeyId) + assert.Equal(t, sk128, *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey) + + // 129 chars AccessKeyId - should fail + s3cfg.Identities[0].Credentials = nil + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{strings.Repeat("a", 129)}, + "SecretAccessKey": []string{sk128}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + + // 7-char SecretAccessKey - should fail + s3cfg.Identities[0].Credentials = nil + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"validkey"}, + "SecretAccessKey": []string{"1234567"}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + + // Exactly 8-char SecretAccessKey - should pass (lower boundary) + s3cfg.Identities[0].Credentials = nil + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"validkey"}, + "SecretAccessKey": []string{"12345678"}, + } + resp, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.Nil(t, iamErr) + assert.Equal(t, "12345678", *resp.CreateAccessKeyResult.AccessKey.SecretAccessKey) + + // 129-char SecretAccessKey - should fail + s3cfg.Identities[0].Credentials = nil + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"validkey"}, + "SecretAccessKey": []string{strings.Repeat("s", 129)}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) +} + +func TestCreateAccessKeyRejectsPartialSupply(t *testing.T) { + iama := newTestIamApiServer(Policies{}) + s3cfg := &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{Name: "alice"}}, + } + // AccessKeyId supplied, SecretAccessKey omitted + values := url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"myappkey"}, + } + _, iamErr := iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + assert.Len(t, s3cfg.Identities[0].Credentials, 0) + + // SecretAccessKey supplied, AccessKeyId omitted + values = url.Values{ + "UserName": []string{"alice"}, + "SecretAccessKey": []string{"secretkey123"}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) + assert.Len(t, s3cfg.Identities[0].Credentials, 0) + + // Partial supply wins over collision: only AccessKeyId supplied, and + // it matches an existing credential. We must see InvalidInput, not + // EntityAlreadyExists — the both-or-none rule is more fundamental. + s3cfg = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + { + Name: "ownerAlpha", + Credentials: []*iam_pb.Credential{ + {AccessKey: "takenkey", SecretKey: "s"}, + }, + }, + {Name: "alice"}, + }, + } + values = url.Values{ + "UserName": []string{"alice"}, + "AccessKeyId": []string{"takenkey"}, + } + _, iamErr = iama.CreateAccessKey(s3cfg, values) + assert.NotNil(t, iamErr) + assert.Equal(t, iam.ErrCodeInvalidInputException, iamErr.Code) +} diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 722bd352a..1312911ba 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -455,6 +455,7 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction etag := getEtag(resp) if resp.StatusCode == http.StatusNoContent { ret.ETag = etag + ret.ContentMd5 = resp.Header.Get("Content-MD5") return &ret, nil } diff --git a/weed/s3api/s3api_embedded_iam.go b/weed/s3api/s3api_embedded_iam.go index 47f772520..1e581d545 100644 --- a/weed/s3api/s3api_embedded_iam.go +++ b/weed/s3api/s3api_embedded_iam.go @@ -410,32 +410,61 @@ func (e *EmbeddedIamApi) CreateAccessKey(s3cfg *iam_pb.S3ApiConfiguration, value userName := values.Get("UserName") status := iam.StatusTypeActive - // Generate AWS-standard access key: AKIA prefix + 16 random uppercase chars = 20 total - randomPart, err := iamStringWithCharset(AccessKeyLength-len(UserAccessKeyPrefix), iamCharsetUpper) - if err != nil { - return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)} + accessKeyId := values.Get("AccessKeyId") + secretAccessKey := values.Get("SecretAccessKey") + if accessKeyId != "" { + if err := iamlib.ValidateCallerSuppliedAccessKeyId(accessKeyId); err != nil { + return resp, &iamError{Code: iam.ErrCodeInvalidInputException, Error: err} + } } - accessKeyId := UserAccessKeyPrefix + randomPart - - secretAccessKey, err := iamStringWithCharset(SecretKeyLength, iamCharset) - if err != nil { - return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)} + if secretAccessKey != "" { + if err := iamlib.ValidateCallerSuppliedSecretAccessKey(secretAccessKey); err != nil { + return resp, &iamError{Code: iam.ErrCodeInvalidInputException, Error: err} + } + } + if (accessKeyId != "") != (secretAccessKey != "") { + return resp, &iamError{Code: iam.ErrCodeInvalidInputException, Error: fmt.Errorf("AccessKeyId and SecretAccessKey must be supplied together")} } + // Find the target user before touching the RNG or scanning for collisions, + // so a missing user fails fast without consuming entropy. + var target *iam_pb.Identity + for _, ident := range s3cfg.Identities { + if userName == ident.Name { + target = ident + break + } + } + if target == nil { + return resp, &iamError{Code: iam.ErrCodeNoSuchEntityException, Error: fmt.Errorf(iamUserDoesNotExist, userName)} + } + + if owner := iamlib.FindAccessKeyOwner(s3cfg, accessKeyId); owner != nil { + glog.V(4).Infof("CreateAccessKey: supplied AccessKeyId already in use by %s %s", owner.Type, owner.Name) + return resp, &iamError{Code: iam.ErrCodeEntityAlreadyExistsException, Error: fmt.Errorf("AccessKeyId is already in use")} + } + if accessKeyId == "" { + randomPart, err := iamStringWithCharset(AccessKeyLength-len(UserAccessKeyPrefix), iamCharsetUpper) + if err != nil { + return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate access key: %w", err)} + } + accessKeyId = UserAccessKeyPrefix + randomPart + } + if secretAccessKey == "" { + var err error + secretAccessKey, err = iamStringWithCharset(SecretKeyLength, iamCharset) + if err != nil { + return resp, &iamError{Code: iam.ErrCodeServiceFailureException, Error: fmt.Errorf("failed to generate secret key: %w", err)} + } + } resp.CreateAccessKeyResult.AccessKey.AccessKeyId = &accessKeyId resp.CreateAccessKeyResult.AccessKey.SecretAccessKey = &secretAccessKey resp.CreateAccessKeyResult.AccessKey.UserName = &userName resp.CreateAccessKeyResult.AccessKey.Status = &status - for _, ident := range s3cfg.Identities { - if userName == ident.Name { - ident.Credentials = append(ident.Credentials, - &iam_pb.Credential{AccessKey: accessKeyId, SecretKey: secretAccessKey, Status: iamAccessKeyStatusActive}) - return resp, nil - } - } - // User not found - return error instead of implicitly creating the user - return resp, &iamError{Code: iam.ErrCodeNoSuchEntityException, Error: fmt.Errorf(iamUserDoesNotExist, userName)} + target.Credentials = append(target.Credentials, + &iam_pb.Credential{AccessKey: accessKeyId, SecretKey: secretAccessKey, Status: iamAccessKeyStatusActive}) + return resp, nil } // DeleteAccessKey deletes an access key for a user. @@ -2102,7 +2131,7 @@ func (e *EmbeddedIamApi) ExecuteAction(ctx context.Context, values url.Values, s return nil, &iamError{Code: s3err.GetAPIError(s3err.ErrInternalError).Code, Error: fmt.Errorf("failed to get s3 api configuration: %v", err)} } - glog.V(4).Infof("IAM ExecuteAction: %+v", values) + glog.V(4).Infof("IAM ExecuteAction: %+v", iamlib.RedactSensitiveFormValues(values)) var response iamlib.RequestIDSetter changed := true switch values.Get("Action") { diff --git a/weed/s3api/s3api_embedded_iam_test.go b/weed/s3api/s3api_embedded_iam_test.go index 60f38b8fa..124831eb9 100644 --- a/weed/s3api/s3api_embedded_iam_test.go +++ b/weed/s3api/s3api_embedded_iam_test.go @@ -911,6 +911,356 @@ func TestEmbeddedIamCreateAccessKey(t *testing.T) { assert.Len(t, api.mockConfig.Identities[0].Credentials, 1) } +// TestEmbeddedIamCreateAccessKeyRejectsMissingUser verifies CreateAccessKey +// returns NoSuchEntity for an unknown user without mutating the config. +func TestEmbeddedIamCreateAccessKeyRejectsMissingUser(t *testing.T) { + api := NewEmbeddedIamApiForTest() + api.mockConfig = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{{Name: "ExistingUser"}}, + } + + form := url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "GhostUser") + + req, _ := http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr := httptest.NewRecorder() + apiRouter := mux.NewRouter().SkipClean(true) + apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions) + apiRouter.ServeHTTP(rr, req) + + assert.NotEqual(t, http.StatusOK, rr.Code) + // No new identity and no credential appended to the existing one. + assert.Len(t, api.mockConfig.Identities, 1) + assert.Len(t, api.mockConfig.Identities[0].Credentials, 0) +} + +// TestEmbeddedIamCreateAccessKeyWithCallerSuppliedKeys tests creating an access key with caller-supplied credentials +func TestEmbeddedIamCreateAccessKeyWithCallerSuppliedKeys(t *testing.T) { + api := NewEmbeddedIamApiForTest() + api.mockConfig = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + {Name: "TestUser"}, + }, + } + + form := url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "TestUser") + form.Set("AccessKeyId", "myapp") + form.Set("SecretAccessKey", "mysecret123") + + req, _ := http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr := httptest.NewRecorder() + apiRouter := mux.NewRouter().SkipClean(true) + apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions) + apiRouter.ServeHTTP(rr, req) + + assert.Equal(t, http.StatusOK, rr.Code) + + // Verify caller-supplied keys were used, not random ones + var out iamCreateAccessKeyResponse + err := xml.Unmarshal(rr.Body.Bytes(), &out) + require.NoError(t, err, "failed to unmarshal CreateAccessKey response") + require.NotNil(t, out.CreateAccessKeyResult.AccessKey.AccessKeyId) + require.NotNil(t, out.CreateAccessKeyResult.AccessKey.SecretAccessKey) + require.NotNil(t, out.CreateAccessKeyResult.AccessKey.UserName) + assert.Equal(t, "myapp", *out.CreateAccessKeyResult.AccessKey.AccessKeyId) + assert.Equal(t, "mysecret123", *out.CreateAccessKeyResult.AccessKey.SecretAccessKey) + assert.Equal(t, "TestUser", *out.CreateAccessKeyResult.AccessKey.UserName) + + // Verify credentials were persisted with caller-supplied keys + require.Len(t, api.mockConfig.Identities, 1) + require.Len(t, api.mockConfig.Identities[0].Credentials, 1) + assert.Equal(t, "myapp", api.mockConfig.Identities[0].Credentials[0].AccessKey) + assert.Equal(t, "mysecret123", api.mockConfig.Identities[0].Credentials[0].SecretKey) +} + +// TestEmbeddedIamCreateAccessKeyRejectsWeakKeys tests that weak caller-supplied keys are rejected +func TestEmbeddedIamCreateAccessKeyRejectsWeakKeys(t *testing.T) { + api := NewEmbeddedIamApiForTest() + api.mockConfig = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + {Name: "TestUser"}, + }, + } + + // AccessKeyId too short + form := url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "TestUser") + form.Set("AccessKeyId", "ab") + form.Set("SecretAccessKey", "validsecret123") + + req, _ := http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr := httptest.NewRecorder() + apiRouter := mux.NewRouter().SkipClean(true) + apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions) + apiRouter.ServeHTTP(rr, req) + + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "AccessKeyId must be 4 to 128 alphanumeric characters") + + // SecretAccessKey too short + form.Set("AccessKeyId", "validkey") + form.Set("SecretAccessKey", "short") + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "SecretAccessKey must be between 8 and 128 characters") + // AccessKeyId with SigV4 delimiters + form.Set("AccessKeyId", "foo/bar=baz") + form.Set("SecretAccessKey", "validsecret123") + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "AccessKeyId must be 4 to 128 alphanumeric characters") +} + +// TestEmbeddedIamCreateAccessKeyRejectsCollision tests that duplicate access keys are rejected +func TestEmbeddedIamCreateAccessKeyRejectsCollision(t *testing.T) { + api := NewEmbeddedIamApiForTest() + // Use a distinctive owner name ("ownerAlpha") so the leak assertion + // cannot accidentally match a word embedded in the error body. + const ownerName = "ownerAlpha" + api.mockConfig = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + { + Name: ownerName, + Credentials: []*iam_pb.Credential{ + {AccessKey: "takenkey", SecretKey: "existingsecret"}, + }, + }, + {Name: "NewUser"}, + }, + } + + form := url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "NewUser") + form.Set("AccessKeyId", "takenkey") + form.Set("SecretAccessKey", "newsecret123") + + req, _ := http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr := httptest.NewRecorder() + apiRouter := mux.NewRouter().SkipClean(true) + apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions) + apiRouter.ServeHTTP(rr, req) + + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "already in use") + assert.NotContains(t, rr.Body.String(), ownerName, "should not leak owner name") + + // Verify no credentials were added to NewUser. Look up by name because the + // memory store backs LoadConfiguration with a map, so Identities order is + // not stable across a save/load round trip. + var newUser *iam_pb.Identity + for _, ident := range api.mockConfig.Identities { + if ident.Name == "NewUser" { + newUser = ident + break + } + } + require.NotNil(t, newUser, "NewUser identity should still exist") + assert.Len(t, newUser.Credentials, 0) +} + +// TestEmbeddedIamCreateAccessKeyRejectsPartialSupply tests that supplying only +// one of AccessKeyId / SecretAccessKey is rejected. +func TestEmbeddedIamCreateAccessKeyRejectsPartialSupply(t *testing.T) { + api := NewEmbeddedIamApiForTest() + api.mockConfig = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + {Name: "TestUser"}, + }, + } + + apiRouter := mux.NewRouter().SkipClean(true) + apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions) + + // AccessKeyId supplied, SecretAccessKey omitted + form := url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "TestUser") + form.Set("AccessKeyId", "myappkey") + + req, _ := http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr := httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "AccessKeyId and SecretAccessKey must be supplied together") + assert.Len(t, api.mockConfig.Identities[0].Credentials, 0) + + // SecretAccessKey supplied, AccessKeyId omitted + form = url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "TestUser") + form.Set("SecretAccessKey", "validsecret1") + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "AccessKeyId and SecretAccessKey must be supplied together") + assert.Len(t, api.mockConfig.Identities[0].Credentials, 0) +} + +// TestEmbeddedIamCreateAccessKeyBoundary tests key length boundaries +func TestEmbeddedIamCreateAccessKeyBoundary(t *testing.T) { + api := NewEmbeddedIamApiForTest() + api.mockConfig = &iam_pb.S3ApiConfiguration{ + Identities: []*iam_pb.Identity{ + {Name: "TestUser"}, + }, + } + + apiRouter := mux.NewRouter().SkipClean(true) + apiRouter.Path("/").Methods(http.MethodPost).HandlerFunc(api.DoActions) + + // Exactly 4 chars — should pass + form := url.Values{} + form.Set("Action", "CreateAccessKey") + form.Set("UserName", "TestUser") + form.Set("AccessKeyId", "abcd") + form.Set("SecretAccessKey", "validsecret1") + + req, _ := http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr := httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.Equal(t, http.StatusOK, rr.Code) + + // Exactly 3 chars — should fail + api.mockConfig.Identities[0].Credentials = nil + form.Set("AccessKeyId", "abc") + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "alphanumeric") + + // Exactly 128 chars — should pass + api.mockConfig.Identities[0].Credentials = nil + ak128 := strings.Repeat("a", 128) + sk128 := strings.Repeat("s", 128) + form.Set("AccessKeyId", ak128) + form.Set("SecretAccessKey", sk128) + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.Equal(t, http.StatusOK, rr.Code) + + // 129 chars AccessKeyId — should fail + api.mockConfig.Identities[0].Credentials = nil + form.Set("AccessKeyId", strings.Repeat("a", 129)) + form.Set("SecretAccessKey", sk128) + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "alphanumeric") + + // 7-char SecretAccessKey — should fail + api.mockConfig.Identities[0].Credentials = nil + form.Set("AccessKeyId", "validkey") + form.Set("SecretAccessKey", "1234567") + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "SecretAccessKey must be between 8 and 128 characters") + + // Exactly 8-char SecretAccessKey — should pass (lower boundary) + api.mockConfig.Identities[0].Credentials = nil + form.Set("AccessKeyId", "validkey") + form.Set("SecretAccessKey", "12345678") + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.Equal(t, http.StatusOK, rr.Code) + + // 129-char SecretAccessKey — should fail + api.mockConfig.Identities[0].Credentials = nil + form.Set("AccessKeyId", "validkey") + form.Set("SecretAccessKey", strings.Repeat("s", 129)) + + req, _ = http.NewRequest("POST", "/", nil) + req.PostForm = form + req.Form = form + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + rr = httptest.NewRecorder() + apiRouter.ServeHTTP(rr, req) + assert.NotEqual(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "SecretAccessKey must be between 8 and 128 characters") +} + // TestEmbeddedIamDeleteAccessKey tests deleting an access key via direct form post func TestEmbeddedIamDeleteAccessKey(t *testing.T) { api := NewEmbeddedIamApiForTest() diff --git a/weed/server/filer_grpc_server_remote.go b/weed/server/filer_grpc_server_remote.go index eaaf0018a..e7fd0b79d 100644 --- a/weed/server/filer_grpc_server_remote.go +++ b/weed/server/filer_grpc_server_remote.go @@ -21,30 +21,44 @@ import ( ) func (fs *FilerServer) CacheRemoteObjectToLocalCluster(ctx context.Context, req *filer_pb.CacheRemoteObjectToLocalClusterRequest) (*filer_pb.CacheRemoteObjectToLocalClusterResponse, error) { - // Use singleflight to deduplicate concurrent caching requests for the same object + // Use singleflight to deduplicate concurrent caching requests for the same object. // This benefits all clients: S3 API, filer HTTP, Hadoop, etc. cacheKey := req.Directory + "/" + req.Name - result, err, shared := fs.remoteCacheGroup.Do(cacheKey, func() (interface{}, error) { - return fs.doCacheRemoteObjectToLocalCluster(ctx, req) + // Detach from caller ctx: on failure the error path deletes every chunk + // already written, so cancelling mid-download loses all progress. For + // blobs large enough that the download outlasts the caller's timeout + // the retry loop never converges. + bgCtx := context.WithoutCancel(ctx) + + // DoChan (vs Do) so the caller can bail out on ctx.Done() while the + // singleflight goroutine keeps caching on bgCtx; otherwise this handler + // goroutine stays blocked for the full download after the client is gone. + ch := fs.remoteCacheGroup.DoChan(cacheKey, func() (interface{}, error) { + return fs.doCacheRemoteObjectToLocalCluster(bgCtx, req) }) - if shared { - glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey) + select { + case <-ctx.Done(): + // Caller gave up; the detached cache keeps running and a later + // request will find the entry cached (or join the same singleflight). + return nil, ctx.Err() + case res := <-ch: + if res.Shared { + glog.V(2).Infof("CacheRemoteObjectToLocalCluster: shared result for %s", cacheKey) + } + if res.Err != nil { + return nil, res.Err + } + if res.Val == nil { + return nil, fmt.Errorf("unexpected nil result from singleflight") + } + resp, ok := res.Val.(*filer_pb.CacheRemoteObjectToLocalClusterResponse) + if !ok { + return nil, fmt.Errorf("unexpected result type from singleflight") + } + return resp, nil } - - if err != nil { - return nil, err - } - if result == nil { - return nil, fmt.Errorf("unexpected nil result from singleflight") - } - - resp, ok := result.(*filer_pb.CacheRemoteObjectToLocalClusterResponse) - if !ok { - return nil, fmt.Errorf("unexpected result type from singleflight") - } - return resp, nil } // doCacheRemoteObjectToLocalCluster performs the actual caching operation. diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 508bf26f3..e1aa7812c 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -561,6 +561,16 @@ func (vs *VolumeServer) ReceiveFile(stream volume_server_pb.VolumeServer_Receive fileInfo.VolumeId, fileInfo.Ext, fileInfo.Collection, fileInfo.ShardId, fileInfo.FileSize) if fileInfo.IsEcVolume { + // os.Create below truncates in place; a mounted EcVolume + // holds fds on the same inodes, so overwriting corrupts + // live readers. + if _, mounted := vs.store.FindEcVolume(needle.VolumeId(fileInfo.VolumeId)); mounted { + glog.Errorf("ReceiveFile: ec volume %d is mounted; refusing overwrite for %s", fileInfo.VolumeId, fileInfo.Ext) + return stream.SendAndClose(&volume_server_pb.ReceiveFileResponse{ + Error: fmt.Sprintf("ec volume %d is mounted; unmount before ReceiveFile", fileInfo.VolumeId), + }) + } + // disk_id=0 means "unset" (protobuf default), so auto-select // mirrors VolumeEcShardsCopy: prefer a disk already holding // this volume's shards, then any HDD, then any disk. diff --git a/weed/server/volume_grpc_read_all.go b/weed/server/volume_grpc_read_all.go index 0b1aff2ff..190082a6f 100644 --- a/weed/server/volume_grpc_read_all.go +++ b/weed/server/volume_grpc_read_all.go @@ -4,6 +4,7 @@ import ( "fmt" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) @@ -12,6 +13,7 @@ func (vs *VolumeServer) ReadAllNeedles(req *volume_server_pb.ReadAllNeedlesReque for _, vid := range req.VolumeIds { if err := vs.streamReadOneVolume(needle.VolumeId(vid), stream); err != nil { + stats.VolumeServerFileReadFailures.Inc() return err } } diff --git a/weed/server/volume_grpc_read_write.go b/weed/server/volume_grpc_read_write.go index 1398075b1..7879090d6 100644 --- a/weed/server/volume_grpc_read_write.go +++ b/weed/server/volume_grpc_read_write.go @@ -5,20 +5,28 @@ import ( "fmt" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" ) func (vs *VolumeServer) ReadNeedleBlob(ctx context.Context, req *volume_server_pb.ReadNeedleBlobRequest) (resp *volume_server_pb.ReadNeedleBlobResponse, err error) { - resp = &volume_server_pb.ReadNeedleBlobResponse{} - v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) - if v == nil { - return nil, fmt.Errorf("not found volume id %d", req.VolumeId) - } + if err := func() error { + resp = &volume_server_pb.ReadNeedleBlobResponse{} + v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) + if v == nil { + return fmt.Errorf("not found volume id %d", req.VolumeId) + } - resp.NeedleBlob, err = v.ReadNeedleBlob(req.Offset, types.Size(req.Size)) - if err != nil { - return nil, fmt.Errorf("read needle blob offset %d size %d: %v", req.Offset, req.Size, err) + resp.NeedleBlob, err = v.ReadNeedleBlob(req.Offset, types.Size(req.Size)) + if err != nil { + return fmt.Errorf("read needle blob offset %d size %d: %v", req.Offset, req.Size, err) + } + + return nil + }(); err != nil { + stats.VolumeServerFileReadFailures.Inc() + return nil, err } return resp, nil @@ -28,29 +36,37 @@ func (vs *VolumeServer) ReadNeedleMeta(ctx context.Context, req *volume_server_p resp = &volume_server_pb.ReadNeedleMetaResponse{} volumeId := needle.VolumeId(req.VolumeId) - n := &needle.Needle{ - Id: types.NeedleId(req.NeedleId), - Flags: 0x08, - } - size := req.Size - offset := req.Offset + if err := func() error { + n := &needle.Needle{ + Id: types.NeedleId(req.NeedleId), + Flags: 0x08, + } + size := req.Size + offset := req.Offset - hasVolume := vs.store.HasVolume(volumeId) - if !hasVolume { - return nil, fmt.Errorf("not found volume id %d and read needle metadata at ec shards is not supported", req.VolumeId) - } - err = vs.store.ReadVolumeNeedleMetaAt(volumeId, n, offset, size) - if err != nil { + hasVolume := vs.store.HasVolume(volumeId) + if !hasVolume { + return fmt.Errorf("not found volume id %d and read needle metadata at ec shards is not supported", req.VolumeId) + } + err = vs.store.ReadVolumeNeedleMetaAt(volumeId, n, offset, size) + if err != nil { + return err + } + + resp.Cookie = uint32(n.Cookie) + resp.LastModified = n.LastModified + resp.Crc = n.Checksum.Value() + if n.HasTtl() { + resp.Ttl = n.Ttl.String() + } + resp.AppendAtNs = n.AppendAtNs + + return nil + }(); err != nil { + stats.VolumeServerFileReadFailures.Inc() return nil, err } - resp.Cookie = uint32(n.Cookie) - resp.LastModified = n.LastModified - resp.Crc = n.Checksum.Value() - if n.HasTtl() { - resp.Ttl = n.Ttl.String() - } - resp.AppendAtNs = n.AppendAtNs return resp, nil } @@ -61,13 +77,19 @@ func (vs *VolumeServer) WriteNeedleBlob(ctx context.Context, req *volume_server_ resp = &volume_server_pb.WriteNeedleBlobResponse{} - v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) - if v == nil { - return nil, fmt.Errorf("not found volume id %d", req.VolumeId) - } + if err := func() error { + v := vs.store.GetVolume(needle.VolumeId(req.VolumeId)) + if v == nil { + return fmt.Errorf("not found volume id %d", req.VolumeId) + } + if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), req.NeedleBlob, types.Size(req.Size)); err != nil { + return fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err) + } - if err = v.WriteNeedleBlob(types.NeedleId(req.NeedleId), req.NeedleBlob, types.Size(req.Size)); err != nil { - return nil, fmt.Errorf("write blob needle %d size %d: %v", req.NeedleId, req.Size, err) + return nil + }(); err != nil { + stats.VolumeServerFileWriteFailures.Inc() + return nil, err } return resp, nil diff --git a/weed/server/volume_server_handlers_write.go b/weed/server/volume_server_handlers_write.go index 418f3c235..3477a411f 100644 --- a/weed/server/volume_server_handlers_write.go +++ b/weed/server/volume_server_handlers_write.go @@ -57,6 +57,7 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) { // http 204 status code does not allow body if writeError == nil && isUnchanged { SetEtag(w, reqNeedle.Etag()) + w.Header().Set("Content-MD5", contentMd5) w.WriteHeader(http.StatusNoContent) return } diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index b2c056153..b7468b7f5 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -370,6 +370,22 @@ var ( Help: "Number of master server disconnections.", }, []string{"address"}) + VolumeServerFileReadFailures = prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: "volumeServer", + Name: "file_read_failures", + Help: "Counter of overall failed file read requests from clients.", + }) + + VolumeServerFileWriteFailures = prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: "volumeServer", + Name: "file_write_failures", + Help: "Counter of overall failed file write requests from clients.", + }) + S3RequestCounter = prometheus.NewCounterVec( prometheus.CounterOpts{ Namespace: Namespace, @@ -537,6 +553,8 @@ func init() { Gather.MustRegister(VolumeServerInFlightDownloadSize) Gather.MustRegister(VolumeServerInFlightUploadSize) Gather.MustRegister(VolumeServerMasterDisconnections) + Gather.MustRegister(VolumeServerFileReadFailures) + Gather.MustRegister(VolumeServerFileWriteFailures) Gather.MustRegister(S3RequestCounter) Gather.MustRegister(S3HandlerCounter) diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index 7bdc4f741..0a49e3ac1 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -596,7 +596,6 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error { } } - // Report results if len(deleteErrors) > 0 { t.GetLogger().WithFields(map[string]interface{}{ "volume_id": t.volumeID, @@ -605,16 +604,19 @@ func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error { "total_replicas": len(replicas), "success_rate": float64(successCount) / float64(len(replicas)) * 100, "errors": deleteErrors, - }).Warning("Some volume deletions failed") - // Don't return error - EC task should still be considered successful if shards are mounted - } else { - t.GetLogger().WithFields(map[string]interface{}{ - "volume_id": t.volumeID, - "replica_count": len(replicas), - "replica_servers": replicas, - }).Info("Successfully deleted volume from all replica servers") + }).Error("Failed to delete some original volume replicas after EC encoding") + // A surviving source replica lets a later detection scan re-propose + // EC on the same volume, which retries over mounted shards. + return fmt.Errorf("failed to delete %d of %d original volume replicas for volume %d: %s", + len(deleteErrors), len(replicas), t.volumeID, strings.Join(deleteErrors, "; ")) } + t.GetLogger().WithFields(map[string]interface{}{ + "volume_id": t.volumeID, + "replica_count": len(replicas), + "replica_servers": replicas, + }).Info("Successfully deleted volume from all replica servers") + return nil } diff --git a/weed/worker/tasks/erasure_coding/ec_task_delete_swallow_test.go b/weed/worker/tasks/erasure_coding/ec_task_delete_swallow_test.go new file mode 100644 index 000000000..54c5f317a --- /dev/null +++ b/weed/worker/tasks/erasure_coding/ec_task_delete_swallow_test.go @@ -0,0 +1,113 @@ +package erasure_coding + +import ( + "context" + "net/http" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// One reachable replica + one unreachable: the reachable delete still +// succeeds, and the function surfaces an error naming the failure. +func TestDeleteOriginalVolumeSurfacesReplicaFailures(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const volumeID = uint32(91842) + framework.AllocateVolume(t, grpcClient, volumeID, "") + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 918420, 0x91842042) + uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, + []byte("delete-surface-content-for-issue-9184")) + _ = framework.ReadAllAndClose(t, uploadResp) + require.Equal(t, http.StatusCreated, uploadResp.StatusCode) + + task := NewErasureCodingTask( + "delete-surface-fix", + clusterHarness.VolumeServerAddress(), + volumeID, + "", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + + unreachable := "127.0.0.1:1" + task.sources = []*worker_pb.TaskSource{ + { + Node: clusterHarness.VolumeServerAddress(), + VolumeId: volumeID, + }, + { + Node: unreachable, + VolumeId: volumeID, + }, + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + err := task.deleteOriginalVolume(ctx) + require.Error(t, err, "deleteOriginalVolume must surface replica delete failures (#9184)") + require.Contains(t, err.Error(), unreachable, + "returned error should name the replica that failed: %v", err) + require.True(t, + strings.Contains(err.Error(), "failed to delete"), + "returned error should describe what failed: %v", err) + + _, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID}) + require.Error(t, statusErr, + "reachable replica %d should have been deleted before failure was surfaced", volumeID) +} + +func TestDeleteOriginalVolumeSucceedsWhenAllReplicasReachable(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const volumeID = uint32(91844) + framework.AllocateVolume(t, grpcClient, volumeID, "") + + httpClient := framework.NewHTTPClient() + fid := framework.NewFileID(volumeID, 918440, 0x91844042) + uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, + []byte("delete-happy-path-content-for-issue-9184")) + _ = framework.ReadAllAndClose(t, uploadResp) + require.Equal(t, http.StatusCreated, uploadResp.StatusCode) + + task := NewErasureCodingTask( + "delete-happy-path", + clusterHarness.VolumeServerAddress(), + volumeID, + "", + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + task.sources = []*worker_pb.TaskSource{ + {Node: clusterHarness.VolumeServerAddress(), VolumeId: volumeID}, + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + require.NoError(t, task.deleteOriginalVolume(ctx)) + + _, statusErr := grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID}) + require.Error(t, statusErr, "volume %d should be gone after successful delete", volumeID) +}