From 755632e19b769bc53dbab52e91f7cea18c405e9f Mon Sep 17 00:00:00 2001 From: Awuqing <3184394176@qq.com> Date: Fri, 7 Aug 2026 20:54:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(cluster):=20=E7=BB=91=E5=AE=9A=20Agent=20?= =?UTF-8?q?=E4=B8=AD=E8=BD=AC=E4=BA=A7=E7=89=A9=E5=88=B0=E5=A4=87=E4=BB=BD?= =?UTF-8?q?=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- server/internal/agent/executor.go | 2 +- server/internal/agent/executor_test.go | 5 +- server/internal/backup/temp_files.go | 11 ++ server/internal/service/agent_service.go | 112 ++++++++++++++---- server/internal/service/agent_service_test.go | 108 +++++++++++++++-- 5 files changed, 203 insertions(+), 35 deletions(-) diff --git a/server/internal/agent/executor.go b/server/internal/agent/executor.go index f80732a..e58ac7d 100644 --- a/server/internal/agent/executor.go +++ b/server/internal/agent/executor.go @@ -124,7 +124,7 @@ func (e *Executor) ExecuteRunTask(ctx context.Context, taskID, recordID uint) er } fileName := filepath.Base(finalPath) fileSize := info.Size() - storagePath := backup.BuildStorageKey(spec.Type, startedAt, fileName) + storagePath := backup.BuildRecordStorageKey(spec.Type, startedAt, recordID, fileName) // 5) 计算 checksum（一次读一次）并上传到所有目标 checksum, err := computeFileSHA256(finalPath) diff --git a/server/internal/agent/executor_test.go b/server/internal/agent/executor_test.go index 72189a1..b197418 100644 --- a/server/internal/agent/executor_test.go +++ b/server/internal/agent/executor_test.go @@ -141,8 +141,9 @@ func TestExecuteRunTaskRelaysMasterLocalDiskTarget(t *testing.T) { if got := r.Header.Get("X-BackupX-SHA256"); got != fmt.Sprintf("%x", digest[:]) { t.Fatalf("relay checksum header = %q", got) } - if r.Header.Get("X-BackupX-Object-Key") == "" || r.ContentLength != int64(len(body)) { - t.Fatalf("invalid relay metadata: key=%q length=%d body=%d", r.Header.Get("X-BackupX-Object-Key"), r.ContentLength, len(body)) + objectKey := r.Header.Get("X-BackupX-Object-Key") + if !strings.Contains(objectKey, "/records/99/") || r.ContentLength != int64(len(body)) { + t.Fatalf("invalid relay metadata: key=%q length=%d body=%d", objectKey, r.ContentLength, len(body)) } relayed = append([]byte(nil), body...) writeAgentEnvelope(t, w, map[string]string{"status": "ok"}) diff --git a/server/internal/backup/temp_files.go b/server/internal/backup/temp_files.go index db77fc2..cce68c5 100644 --- a/server/internal/backup/temp_files.go +++ b/server/internal/backup/temp_files.go @@ -55,6 +55,17 @@ func BuildStorageKey(backupType string, startedAt time.Time, fileName string) st return filepath.ToSlash(filepath.Join("BackupX", typeName, startedAt.UTC().Format("060102"), fileName)) } +// BuildRecordStorageKey gives remote-Agent artifacts a record-owned namespace. +// The Master validates this namespace before accepting a relayed upload, so one +// Agent cannot overwrite another record's object on centrally mounted storage. +func BuildRecordStorageKey(backupType string, startedAt time.Time, recordID uint, fileName string) string { + typeName := strings.TrimSpace(strings.ToLower(backupType)) + if typeName == "" { + typeName = "file" + } + return filepath.ToSlash(filepath.Join("BackupX", typeName, startedAt.UTC().Format("060102"), "records", fmt.Sprintf("%d", recordID), fileName)) +} + func sanitizeTaskName(value string) string { trimmed := strings.TrimSpace(strings.ToLower(value)) trimmed = strings.ReplaceAll(trimmed, " ", "-") diff --git a/server/internal/service/agent_service.go b/server/internal/service/agent_service.go index d4c97b9..9a32e14 100644 --- a/server/internal/service/agent_service.go +++ b/server/internal/service/agent_service.go @@ -2,16 +2,19 @@ package service import ( "context" + "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" + "math" "path" "strings" "time" "backupx/server/internal/apperror" + "backupx/server/internal/backup" "backupx/server/internal/model" "backupx/server/internal/repository" "backupx/server/internal/storage" @@ -282,12 +285,13 @@ func (s *AgentService) UploadArtifact(ctx context.Context, node *model.Node, rec if !masterRelay { return apperror.BadRequest("AGENT_ARTIFACT_RELAY_UNSUPPORTED", "该本地磁盘目标配置为 Agent 直接写入", nil) } - cleanKey := path.Clean(strings.TrimSpace(objectKey)) - if cleanKey == "." || path.IsAbs(cleanKey) || strings.HasPrefix(cleanKey, "../") || cleanKey != objectKey || strings.Contains(objectKey, "\\") { - return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象路径不安全", nil) + cleanKey, keyErr := s.validateArtifactKey(record, task, objectKey, true) + if keyErr != nil { + return keyErr } - checksumBytes, checksumErr := hex.DecodeString(strings.TrimSpace(checksum)) - if size < 0 || checksumErr != nil || len(checksumBytes) != 32 { + checksum = strings.TrimSpace(checksum) + checksumBytes, checksumErr := hex.DecodeString(checksum) + if size < 0 || size == math.MaxInt64 || checksumErr != nil || len(checksumBytes) != sha256.Size { return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "中转上传需要有效的大小和 SHA-256", checksumErr) } if target.QuotaBytes > 0 { @@ -302,8 +306,8 @@ func (s *AgentService) UploadArtifact(ctx context.Context, node *model.Node, rec break } } - if currentUsed+size > target.QuotaBytes { - return apperror.BadRequest("BACKUP_STORAGE_QUOTA_EXCEEDED", fmt.Sprintf("超出存储目标配额（%d + %d > %d）", currentUsed, size, target.QuotaBytes), nil) + if currentUsed >= target.QuotaBytes || size > target.QuotaBytes-currentUsed { + return apperror.BadRequest("BACKUP_STORAGE_QUOTA_EXCEEDED", fmt.Sprintf("超出存储目标配额（当前 %d，新增 %d，配额 %d）", currentUsed, size, target.QuotaBytes), nil) } } provider, err := s.registry.Create(ctx, target.Type, configMap) @@ -380,22 +384,19 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor allowedTargets[targetID] = struct{}{} } targetCache := make(map[uint]*model.StorageTarget) - validateTransferMode := func(targetID uint, transferMode string) error { + validateTransferMode := func(targetID uint, transferMode string) (string, error) { if _, ok := allowedTargets[targetID]; !ok { - return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil) - } - if transferMode == "" { - return nil + return "", apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil) } target := targetCache[targetID] if target == nil { var findErr error target, findErr = s.storageRepo.FindByID(ctx, targetID) if findErr != nil { - return findErr + return "", findErr } if target == nil { - return apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil) + return "", apperror.BadRequest("BACKUP_STORAGE_TARGET_INVALID", "存储目标不存在", nil) } targetCache[targetID] = target } @@ -403,31 +404,73 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor if strings.EqualFold(target.Type, storage.TypeLocalDisk) { var localConfig storage.LocalDiskConfig if err := s.cipher.DecryptJSON(target.ConfigCiphertext, &localConfig); err != nil { - return fmt.Errorf("decrypt storage config: %w", err) + return "", fmt.Errorf("decrypt storage config: %w", err) } if localConfig.MasterRelay { expectedMode = storage.TransferModeMasterRelay } } - if transferMode != expectedMode { - return apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "Agent 上报的存储传输模式与目标配置不一致", nil) + if transferMode != "" && transferMode != expectedMode { + return "", apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "Agent 上报的存储传输模式与目标配置不一致", nil) } - return nil + return expectedMode, nil } + selectedTransferMode := "" if update.StorageTargetID > 0 { if _, ok := allowedTargets[update.StorageTargetID]; !ok { return apperror.Unauthorized("BACKUP_STORAGE_TARGET_FORBIDDEN", "存储目标不属于该任务", nil) } - if err := validateTransferMode(update.StorageTargetID, update.StorageTransferMode); err != nil { - return err + var modeErr error + selectedTransferMode, modeErr = validateTransferMode(update.StorageTargetID, update.StorageTransferMode) + if modeErr != nil { + return modeErr } } else if update.StorageTransferMode != "" { return apperror.BadRequest("AGENT_STORAGE_TRANSFER_MODE_INVALID", "传输模式缺少对应的存储目标", nil) } - for _, result := range update.StorageUploadResults { - if err := validateTransferMode(result.StorageTargetID, result.TransferMode); err != nil { + for index := range update.StorageUploadResults { + result := &update.StorageUploadResults[index] + expectedMode, err := validateTransferMode(result.StorageTargetID, result.TransferMode) + if err != nil { return err } + if result.FileSize < 0 || (result.Status != "" && result.Status != "success" && result.Status != "failed") { + return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的存储结果无效", nil) + } + if result.StoragePath != "" { + normalizedPath, err := s.validateArtifactKey(record, task, result.StoragePath, expectedMode == storage.TransferModeMasterRelay) + if err != nil { + return err + } + result.StoragePath = normalizedPath + } + result.TransferMode = expectedMode + } + if update.StoragePath != "" { + if update.StorageTargetID == 0 { + return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "存储路径缺少对应的存储目标", nil) + } + cleanStoragePath, err := s.validateArtifactKey(record, task, update.StoragePath, selectedTransferMode == storage.TransferModeMasterRelay) + if err != nil { + return err + } + if update.FileName != "" && path.Base(cleanStoragePath) != update.FileName { + return apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "Agent 上报的文件名与存储路径不一致", nil) + } + update.StoragePath = cleanStoragePath + } + if update.Status != "" && update.Status != model.BackupRecordStatusRunning && update.Status != model.BackupRecordStatusSuccess && update.Status != model.BackupRecordStatusFailed { + return apperror.BadRequest("BACKUP_RECORD_STATUS_INVALID", "Agent 上报的备份状态无效", nil) + } + if update.FileSize < 0 || (update.FileName != "" && (path.Base(update.FileName) != update.FileName || strings.Contains(update.FileName, "\\"))) { + return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的备份文件信息无效", nil) + } + if update.Checksum != "" { + checksumBytes, checksumErr := hex.DecodeString(strings.TrimSpace(update.Checksum)) + if checksumErr != nil || len(checksumBytes) != sha256.Size { + return apperror.BadRequest("AGENT_ARTIFACT_INVALID", "Agent 上报的 SHA-256 无效", checksumErr) + } + update.Checksum = strings.ToLower(strings.TrimSpace(update.Checksum)) } if update.Status != "" { record.Status = update.Status @@ -446,9 +489,7 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor } if update.StorageTargetID > 0 { record.StorageTargetID = update.StorageTargetID - } - if update.StorageTransferMode != "" { - record.StorageTransferMode = update.StorageTransferMode + record.StorageTransferMode = selectedTransferMode } if len(update.StorageUploadResults) > 0 { if resultsJSON, marshalErr := json.Marshal(update.StorageUploadResults); marshalErr == nil { @@ -484,6 +525,27 @@ func (s *AgentService) UpdateRecord(ctx context.Context, node *model.Node, recor return nil } +func (s *AgentService) validateArtifactKey(record *model.BackupRecord, task *model.BackupTask, objectKey string, requireRecordNamespace bool) (string, error) { + if record == nil || task == nil { + return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "无法确认中转对象归属", nil) + } + rawKey := objectKey + cleanKey := path.Clean(rawKey) + fileName := path.Base(cleanKey) + if rawKey == "" || strings.TrimSpace(rawKey) != rawKey || cleanKey == "." || path.IsAbs(cleanKey) || strings.HasPrefix(cleanKey, "../") || cleanKey != rawKey || strings.Contains(rawKey, "\\") || fileName == "." || fileName == "/" { + return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象路径不安全", nil) + } + expectedKey := backup.BuildRecordStorageKey(task.Type, record.StartedAt, record.ID, fileName) + if requireRecordNamespace { + legacyKey := backup.BuildStorageKey(task.Type, record.StartedAt, fileName) + if cleanKey != expectedKey && cleanKey != legacyKey { + return "", apperror.BadRequest("AGENT_ARTIFACT_INVALID_PATH", "中转上传对象不属于当前备份记录", nil) + } + return expectedKey, nil + } + return cleanKey, nil +} + func recordBelongsToNode(record *model.BackupRecord, task *model.BackupTask, nodeID uint) bool { if record.NodeID != 0 { return record.NodeID == nodeID diff --git a/server/internal/service/agent_service_test.go b/server/internal/service/agent_service_test.go index 96b9102..4a66dfd 100644 --- a/server/internal/service/agent_service_test.go +++ b/server/internal/service/agent_service_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "backupx/server/internal/backup" "backupx/server/internal/config" "backupx/server/internal/database" "backupx/server/internal/logger" @@ -103,16 +104,21 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing. if _, err := svc.GetTaskSpec(ctx, other, 1); err == nil { t.Fatal("expected non-owner node to be forbidden from pooled task spec") } + record, err := records.FindByID(ctx, 1) + if err != nil { + t.Fatalf("FindByID record returned error: %v", err) + } + storagePath := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "backup.tar.gz") if err := svc.UpdateRecord(ctx, owner, 1, AgentRecordUpdate{ Status: model.BackupRecordStatusSuccess, FileName: "backup.tar.gz", FileSize: 123, - StoragePath: "tasks/1/backup.tar.gz", + StoragePath: storagePath, StorageTargetID: 1, StorageTransferMode: storage.TransferModeMasterRelay, StorageUploadResults: []StorageUploadResultItem{ - {StorageTargetID: 1, StorageTargetName: "local", Status: "success", StoragePath: "tasks/1/backup.tar.gz", FileSize: 123, TransferMode: storage.TransferModeMasterRelay}, + {StorageTargetID: 1, StorageTargetName: "local", Status: "success", StoragePath: storagePath, FileSize: 123, TransferMode: storage.TransferModeMasterRelay}, }, }); err != nil { t.Fatalf("owner UpdateRecord returned error: %v", err) @@ -139,12 +145,16 @@ func TestAgentServicePooledTaskUsesRecordNodeForSpecAndRecordUpdates(t *testing. } func TestAgentServiceRelaysRemoteArtifactToMasterLocalDisk(t *testing.T) { - svc, _, _, _, owner, other := newAgentServicePoolTestHarness(t) + svc, _, records, _, owner, other := newAgentServicePoolTestHarness(t) ctx := context.Background() payload := []byte("artifact from remote source server") digest := sha256.Sum256(payload) checksum := fmt.Sprintf("%x", digest[:]) - objectKey := "file/2026/08/06/remote-source.tar" + record, err := records.FindByID(ctx, 1) + if err != nil { + t.Fatalf("FindByID record returned error: %v", err) + } + objectKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "remote-source.tar") if err := svc.UploadArtifact(ctx, owner, 1, 1, objectKey, int64(len(payload)), checksum, bytes.NewReader(payload)); err != nil { t.Fatalf("UploadArtifact returned error: %v", err) @@ -165,13 +175,92 @@ func TestAgentServiceRelaysRemoteArtifactToMasterLocalDisk(t *testing.T) { if !bytes.Equal(stored, payload) { t.Fatalf("relayed artifact differs: got %q", stored) } - if err := svc.UploadArtifact(ctx, other, 1, 1, "file/forbidden.tar", int64(len(payload)), checksum, bytes.NewReader(payload)); err == nil { + if err := svc.UploadArtifact(ctx, other, 1, 1, objectKey, int64(len(payload)), checksum, bytes.NewReader(payload)); err == nil { t.Fatal("expected a different node to be forbidden from relaying the artifact") } + + legacyPayload := []byte("artifact from an older Agent") + legacyDigest := sha256.Sum256(legacyPayload) + legacyKey := backup.BuildStorageKey("file", record.StartedAt, "legacy-agent.tar") + canonicalKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "legacy-agent.tar") + if err := svc.UploadArtifact(ctx, owner, record.ID, target.ID, legacyKey, int64(len(legacyPayload)), fmt.Sprintf("%x", legacyDigest[:]), bytes.NewReader(legacyPayload)); err != nil { + t.Fatalf("UploadArtifact legacy key returned error: %v", err) + } + stored, err = os.ReadFile(filepath.Join(basePath, filepath.FromSlash(canonicalKey))) + if err != nil || !bytes.Equal(stored, legacyPayload) { + t.Fatalf("legacy Agent artifact was not normalized: data=%q err=%v", stored, err) + } + if _, err := os.Stat(filepath.Join(basePath, filepath.FromSlash(legacyKey))); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("legacy object key should not be written directly: %v", err) + } + if err := svc.UpdateRecord(ctx, owner, record.ID, AgentRecordUpdate{ + Status: model.BackupRecordStatusSuccess, + FileName: "legacy-agent.tar", + FileSize: int64(len(legacyPayload)), + Checksum: fmt.Sprintf("%x", legacyDigest[:]), + StoragePath: legacyKey, + StorageTargetID: target.ID, + StorageUploadResults: []StorageUploadResultItem{{ + StorageTargetID: target.ID, + Status: "success", + StoragePath: legacyKey, + FileSize: int64(len(legacyPayload)), + }}, + }); err != nil { + t.Fatalf("UpdateRecord legacy key returned error: %v", err) + } + updated, err := records.FindByID(ctx, record.ID) + if err != nil { + t.Fatalf("FindByID updated record returned error: %v", err) + } + if updated.StoragePath != canonicalKey || updated.StorageTransferMode != storage.TransferModeMasterRelay || !strings.Contains(updated.StorageUploadResults, canonicalKey) { + t.Fatalf("legacy Agent record was not normalized: %#v", updated) + } +} + +func TestAgentServiceRejectsArtifactOutsideRecordNamespace(t *testing.T) { + svc, _, records, _, owner, _ := newAgentServicePoolTestHarness(t) + ctx := context.Background() + record, err := records.FindByID(ctx, 1) + if err != nil { + t.Fatalf("FindByID record returned error: %v", err) + } + target, err := svc.storageRepo.FindByID(ctx, 1) + if err != nil || target == nil { + t.Fatalf("FindByID target: target=%#v err=%v", target, err) + } + config := map[string]any{} + if err := svc.cipher.DecryptJSON(target.ConfigCiphertext, &config); err != nil { + t.Fatalf("DecryptJSON target config: %v", err) + } + basePath, _ := config["basePath"].(string) + victimKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID+1, "victim.tar") + victimPath := filepath.Join(basePath, filepath.FromSlash(victimKey)) + if err := os.MkdirAll(filepath.Dir(victimPath), 0o755); err != nil { + t.Fatalf("MkdirAll victim parent: %v", err) + } + if err := os.WriteFile(victimPath, []byte("keep me"), 0o600); err != nil { + t.Fatalf("WriteFile victim: %v", err) + } + payload := []byte("overwrite") + digest := sha256.Sum256(payload) + if err := svc.UploadArtifact(ctx, owner, record.ID, target.ID, victimKey, int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload)); err == nil { + t.Fatal("expected another record namespace to be rejected") + } + stored, err := os.ReadFile(victimPath) + if err != nil { + t.Fatalf("ReadFile victim: %v", err) + } + if string(stored) != "keep me" { + t.Fatalf("victim object changed: %q", stored) + } + if err := svc.UpdateRecord(ctx, owner, record.ID, AgentRecordUpdate{StoragePath: victimKey, StorageTargetID: target.ID, StorageTransferMode: storage.TransferModeMasterRelay}); err == nil { + t.Fatal("expected another record namespace in status update to be rejected") + } } func TestAgentServiceKeepsExistingLocalDiskTargetsAgentLocal(t *testing.T) { - svc, _, _, _, owner, _ := newAgentServicePoolTestHarness(t) + svc, _, records, _, owner, _ := newAgentServicePoolTestHarness(t) ctx := context.Background() target, err := svc.storageRepo.FindByID(ctx, 1) if err != nil || target == nil { @@ -195,7 +284,12 @@ func TestAgentServiceKeepsExistingLocalDiskTargetsAgentLocal(t *testing.T) { } payload := []byte("must not be relayed") digest := sha256.Sum256(payload) - err = svc.UploadArtifact(ctx, owner, 1, 1, "file/legacy.tar", int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload)) + record, findErr := records.FindByID(ctx, 1) + if findErr != nil { + t.Fatalf("FindByID record returned error: %v", findErr) + } + objectKey := backup.BuildRecordStorageKey("file", record.StartedAt, record.ID, "legacy.tar") + err = svc.UploadArtifact(ctx, owner, 1, 1, objectKey, int64(len(payload)), fmt.Sprintf("%x", digest[:]), bytes.NewReader(payload)) if err == nil { t.Fatal("expected relay upload to be rejected for an Agent-local target") }