GithubBackup/BackupX
1
0
Fork 0
mirror of https://github.com/Awuqing/BackupX.git synced 2026-08-28 19:47:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(cluster): 绑定 Agent 中转产物到备份记录

Browse Source
This commit is contained in:
Awuqing
2026-08-07 20:54:57 +08:00
parent c2805da6df
commit 755632e19b
5 changed files with 203 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/agent/executor.go
+1 -1
View File
@@ -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)
server/internal/agent/executor_test.go
+3 -2
View File
@@ -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"})
server/internal/backup/temp_files.go
+11
View File
@@ -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, " ", "-")
server/internal/service/agent_service.go
+87 -25
View File
@@ -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
server/internal/service/agent_service_test.go
+101 -7
View File
@@ -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")
}