diff --git a/backend/internal/app/app.go b/backend/internal/app/app.go index 5ea1838..a1c93d7 100644 --- a/backend/internal/app/app.go +++ b/backend/internal/app/app.go @@ -105,6 +105,24 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string taskRepo := repository.NewSQLTaskRepository(db.DB, runtimeDatabaseConfig.Driver) transactionRepo := repository.NewSQLDeploymentTransactionRepository(db.DB, runtimeDatabaseConfig.Driver) deploymentService := deployment.NewServiceWithJournal(deployment.SSHFactory{Secrets: secretStore}, transactionRepo) + remoteValidator := certificate.RemoteValidator{} + deploymentService.VerifyRecoveryTarget = func(ctx context.Context, server deployment.Server, baseline deployment.RecoveryBaseline, committed bool) error { + expected := baseline.PreviousFingerprint + if committed { + expected = baseline.ExpectedFingerprint + } + if expected == "" || baseline.Endpoint == "" || len(baseline.Domains) == 0 { + return fmt.Errorf("恢复站点基线不完整") + } + validation, err := remoteValidator.ValidateHTTPSAt(ctx, baseline.Endpoint, server.Address, baseline.Domains) + if err != nil { + return err + } + if validation.FingerprintSHA256 != expected { + return fmt.Errorf("站点证书指纹不符合事务基线") + } + return nil + } if err := deploymentService.RecoverPending(ctx, serverRepo.GetServerByID); err != nil { logger.Error("恢复中断部署事务失败，已保留现场等待人工核查", "error", err) } @@ -128,7 +146,6 @@ func NewWithConfigPath(ctx context.Context, cfg config.Config, configPath string certificates.SetDNSAuthorizationChecker(aliyunAuthChecker) notifier := notification.New(cfg.Notification, secretStore) - remoteValidator := certificate.RemoteValidator{} workflow := &CertificateWorkflow{ Certificates: certificates, Domains: repo, diff --git a/backend/internal/app/certificate_workflow.go b/backend/internal/app/certificate_workflow.go index 577240b..95ca591 100644 --- a/backend/internal/app/certificate_workflow.go +++ b/backend/internal/app/certificate_workflow.go @@ -290,6 +290,10 @@ func (w *CertificateWorkflow) deploy(ctx context.Context, result certificate.Iss server: target.Server, baselineFingerprint: baselineFingerprint, }) + targets[len(verificationTargets)-1].Baseline = deployment.RecoveryBaseline{ + Endpoint: endpoint, Domains: append([]string(nil), result.Domains...), + PreviousFingerprint: baselineFingerprint, ExpectedFingerprint: expectedFingerprint, + } } verification := deployment.Verification{ Verify: func(verificationContext context.Context) error { diff --git a/backend/internal/database/migrate.go b/backend/internal/database/migrate.go index 7cd511b..0e6e8e3 100644 --- a/backend/internal/database/migrate.go +++ b/backend/internal/database/migrate.go @@ -19,6 +19,7 @@ import ( //go:embed migrations/sqlite/0015_task_idempotency.sql migrations/mysql/0015_task_idempotency.sql migrations/postgres/0015_task_idempotency.sql //go:embed migrations/sqlite/0016_task_deployment_results.sql migrations/mysql/0016_task_deployment_results.sql migrations/postgres/0016_task_deployment_results.sql //go:embed migrations/sqlite/0017_deployment_transactions.sql migrations/mysql/0017_deployment_transactions.sql migrations/postgres/0017_deployment_transactions.sql +//go:embed migrations/sqlite/0018_recovery_evidence.sql migrations/mysql/0018_recovery_evidence.sql migrations/postgres/0018_recovery_evidence.sql var migrationFiles embed.FS func Migrate(ctx context.Context, db *sql.DB, driver string) error { @@ -43,7 +44,7 @@ func Migrate(ctx context.Context, db *sql.DB, driver string) error { return nil } -const latestMigrationVersion = 17 +const latestMigrationVersion = 18 // currentMigrationVersion 在执行业务迁移前拒绝新版数据库及不连续的迁移历史。 func currentMigrationVersion(ctx context.Context, db *sql.DB) (int, error) { @@ -229,6 +230,8 @@ func migrationName(version int) string { return "task_deployment_results" case 17: return "deployment_transactions" + case 18: + return "recovery_evidence" default: return "unknown" } diff --git a/backend/internal/database/migrations/mysql/0018_recovery_evidence.sql b/backend/internal/database/migrations/mysql/0018_recovery_evidence.sql new file mode 100644 index 0000000..f69baae --- /dev/null +++ b/backend/internal/database/migrations/mysql/0018_recovery_evidence.sql @@ -0,0 +1 @@ +ALTER TABLE deployment_transactions ADD COLUMN recovery_evidence_json LONGTEXT NULL; diff --git a/backend/internal/database/migrations/postgres/0018_recovery_evidence.sql b/backend/internal/database/migrations/postgres/0018_recovery_evidence.sql new file mode 100644 index 0000000..7b02b53 --- /dev/null +++ b/backend/internal/database/migrations/postgres/0018_recovery_evidence.sql @@ -0,0 +1 @@ +ALTER TABLE deployment_transactions ADD COLUMN recovery_evidence_json TEXT; diff --git a/backend/internal/database/migrations/sqlite/0018_recovery_evidence.sql b/backend/internal/database/migrations/sqlite/0018_recovery_evidence.sql new file mode 100644 index 0000000..7b02b53 --- /dev/null +++ b/backend/internal/database/migrations/sqlite/0018_recovery_evidence.sql @@ -0,0 +1 @@ +ALTER TABLE deployment_transactions ADD COLUMN recovery_evidence_json TEXT; diff --git a/backend/internal/deployment/recovery_evidence.go b/backend/internal/deployment/recovery_evidence.go new file mode 100644 index 0000000..07038a1 --- /dev/null +++ b/backend/internal/deployment/recovery_evidence.go @@ -0,0 +1,133 @@ +package deployment + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "reflect" +) + +// RecoveryBaseline 只保存站点身份，不保存证书内容、私钥或登录凭据。 +type RecoveryBaseline struct { + Endpoint string `json:"endpoint"` + Domains []string `json:"domains"` + PreviousFingerprint string `json:"previous_fingerprint"` + ExpectedFingerprint string `json:"expected_fingerprint"` +} + +// RecoveryEvidence 保存恢复所需的不可变身份与已核验决策。 +type RecoveryEvidence struct { + Version int `json:"version"` + LockToken string `json:"lock_token"` + Address string `json:"address"` + Port int `json:"port"` + HostKeyFingerprint string `json:"host_key_fingerprint"` + ValidationCommand []string `json:"validation_command"` + ReloadCommand []string `json:"reload_command"` + Files map[string]string `json:"files"` + Baseline RecoveryBaseline `json:"baseline"` +} + +// EvidenceClient 通过原子符号链接保存锁归属，文件摘要不会将私钥传回业务层。 +type EvidenceClient interface { + CreateOwnedLock(context.Context, string, string) error + ReadLockOwner(context.Context, string) (string, error) + FileDigest(context.Context, string) (string, error) +} + +func contentDigest(data []byte) string { + sum := sha256.Sum256(data) + return hex.EncodeToString(sum[:]) +} + +func validDigest(value string) bool { + decoded, err := hex.DecodeString(value) + return err == nil && len(decoded) == sha256.Size +} + +func checkOwnedLock(ctx context.Context, client Client, path, token string, allowMissing bool) error { + reader, ok := client.(EvidenceClient) + if !ok || !validDigest(token) { + return errors.New("缺少可验证的部署锁归属信息") + } + owner, err := reader.ReadLockOwner(ctx, path) + if err != nil { + return err + } + if owner == "" && allowMissing { + return nil + } + if owner != token { + return errors.New("部署锁不属于本次事务，已保留现场") + } + return nil +} + +func checkFileDigest(ctx context.Context, client Client, path, expected string) error { + reader, ok := client.(EvidenceClient) + if !ok || !validDigest(expected) { + return errors.New("缺少可信文件摘要") + } + actual, err := reader.FileDigest(ctx, path) + if err != nil { + return err + } + if actual != expected { + return fmt.Errorf("文件身份与事务记录不一致：%s", path) + } + return nil +} + +func validateRecoveryEvidence(record TransactionRecord, server Server) error { + e := record.Evidence + if e == nil || e.Version != 1 || !validDigest(e.LockToken) { + return errors.New("历史事务缺少可信恢复证据，需核查现场，不自动接管") + } + if e.Address != server.Address || e.Port != server.Port || e.HostKeyFingerprint != server.HostKeyFingerprint || + !reflect.DeepEqual(e.ValidationCommand, server.ValidationCommand) || !reflect.DeepEqual(e.ReloadCommand, server.ReloadCommand) { + return errors.New("服务器身份或恢复命令已变更，不自动接管历史事务") + } + if len(record.Paths) != 3 || len(record.LockPaths) != 3 || len(e.Files) != 3 { + return errors.New("部署事务目标集合不完整") + } + paths := make(map[string]bool) + for _, path := range record.Paths { + digest := e.Files[path] + if paths[path] || RemotePath(path) != path || len(path) < 2 || path[0] != '/' || + !validDigest(digest) { + return errors.New("部署事务路径或文件摘要无效") + } + paths[path] = true + } + locks := make(map[string]bool) + for _, lock := range record.LockPaths { + matched := false + for path := range paths { + matched = matched || lock == path+".scert-lock" + } + if !matched || locks[lock] { + return errors.New("部署事务锁集合无效") + } + locks[lock] = true + } + for _, temporary := range record.TemporaryPaths { + matched := false + for path := range paths { + matched = matched || temporary == path+".scert-tmp-"+record.ID + } + if !matched { + return errors.New("部署事务临时路径无效") + } + } + seen := make(map[string]bool) + for _, backup := range record.Backups { + if !paths[backup.Original] || seen[backup.Original] || backup.Saved != backup.Original+".scert-backup-"+record.ID || + (backup.OriginalExists && !validDigest(backup.OriginalDigest)) { + return errors.New("部署事务备份身份无效") + } + seen[backup.Original] = true + } + return nil +} diff --git a/backend/internal/deployment/service.go b/backend/internal/deployment/service.go index 2acd067..6b9d6c1 100644 --- a/backend/internal/deployment/service.go +++ b/backend/internal/deployment/service.go @@ -13,7 +13,8 @@ import ( ) type Target struct { - Server Server + Server Server + Baseline RecoveryBaseline } type Artifact struct { @@ -51,9 +52,10 @@ type Report struct { } type Service struct { - factory ClientFactory - journal TransactionJournal - now func() time.Time + factory ClientFactory + journal TransactionJournal + now func() time.Time + VerifyRecoveryTarget func(context.Context, Server, RecoveryBaseline, bool) error } func NewService(factory ClientFactory) *Service { @@ -91,7 +93,7 @@ func (s *Service) deploy(ctx context.Context, targets []Target, artifact Artifac return report } for index, target := range targets { - result := s.deployOne(ctx, target.Server, artifact, pending, index) + result := s.deployOne(ctx, target.Server, artifact, pending, index, target.Baseline) report.Results = append(report.Results, result) if result.Status == "success" { report.Success++ @@ -134,7 +136,7 @@ func loadArtifact(saved SavedFiles) (Artifact, error) { return Artifact{Certificate: certificateData, FullChain: fullChainData, PrivateKey: privateKeyData}, nil } -func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifact, pending *[]*pendingDeployment, index int) (result Result) { +func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifact, pending *[]*pendingDeployment, index int, baseline RecoveryBaseline) (result Result) { ctx, cancel := context.WithTimeout(ctx, 5*time.Minute) defer cancel() server = server.Normalize() @@ -181,6 +183,21 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac lockPaths = append(lockPaths, file.path+".scert-lock") } sort.Strings(lockPaths) + backupID := s.backupID() + result.BackupID = backupID + journalRecord, transactionEnabled := s.newTransactionRecord(ctx, server, paths, lockPaths, backupID, index, baseline) + if transactionEnabled { + if _, ok := client.(EvidenceClient); !ok { + result.FailureCode = FailureUnavailable + result.Message = "当前客户端不支持部署恢复证据" + return result + } + if err := s.journal.CreateDeploymentTransaction(ctx, *journalRecord); err != nil { + result.FailureCode = FailureUnavailable + result.Message = fmt.Sprintf("保存部署事务现场失败：%v", err) + return result + } + } heldLocks := make([]string, 0, len(lockPaths)) retainLocks := false releaseLocks := func() { @@ -197,6 +214,15 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac } released := true for index := len(heldLocks) - 1; index >= 0; index-- { + if transactionEnabled { + if err := checkOwnedLock(releaseCtx, client, heldLocks[index], journalRecord.Evidence.LockToken, true); err != nil { + released = false + result.Status = "failed" + result.FailureCode = FailureLockRelease + result.Message += fmt.Sprintf("；部署锁归属核验失败：%v", err) + continue + } + } if err := client.Unlock(releaseCtx, heldLocks[index]); err != nil { released = false result.Status = "failed" @@ -220,15 +246,12 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return result } heldLocks = append(heldLocks, lockPath) - } - backupID := s.backupID() - result.BackupID = backupID - journalRecord, transactionEnabled := s.newTransactionRecord(ctx, server, paths, lockPaths, backupID, index) - if transactionEnabled { - if err := s.journal.CreateDeploymentTransaction(ctx, *journalRecord); err != nil { - result.FailureCode = FailureUnavailable - result.Message = fmt.Sprintf("保存部署事务现场失败：%v", err) - return result + if transactionEnabled { + if err := client.(EvidenceClient).CreateOwnedLock(ctx, lockPath, journalRecord.Evidence.LockToken); err != nil { + result.FailureCode = FailureUnavailable + result.Message = fmt.Sprintf("保存部署锁归属失败：%v", err) + return result + } } } persistTransaction := func(persistContext context.Context, phase string) error { @@ -249,6 +272,11 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac result.Message += fmt.Sprintf("；保存部署事务最终状态失败：%v", err) } } + if err := persistTransaction(ctx, TransactionPhaseLocked); err != nil { + result.FailureCode = FailureUnavailable + result.Message = fmt.Sprintf("保存部署锁阶段失败：%v", err) + return result + } backups := make([]backupFile, 0, len(paths)) temporary := make([]string, 0, len(paths)) installed := make([]string, 0, len(paths)) @@ -258,6 +286,9 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac if err := reconnectDeploymentClient(rollbackContext, s.factory, server, &client); err != nil { return fmt.Errorf("重新连接服务器以回滚失败：%w", err) } + if transactionEnabled { + return restoreTransactionFiles(rollbackContext, client, *journalRecord) + } var rollbackErrors []error for _, path := range temporary { if err := client.Remove(rollbackContext, path); err != nil { @@ -300,10 +331,18 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac if !retainLocks { result.FailureCode = FailureRuntimeUnknown } + retainLocks = true // 命令可能已部分执行；文件回滚并不证明运行中的服务已经恢复。 message += "；服务运行状态未确认，请检查服务并在确认旧配置后重新加载" } result.Message = message + if !retainLocks { + if err := persistTransaction(context.WithoutCancel(ctx), TransactionPhaseRollbackVerified); err != nil { + retainLocks = true + result.FailureCode = FailureRecoveryUnknown + result.Message += fmt.Sprintf("；保存回滚核验阶段失败：%v", err) + } + } releaseLocks() if retainLocks || result.FailureCode == FailureLockRelease { finishTransaction(TransactionStatusRecoveryRequired, result.Message) @@ -317,9 +356,6 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac temporaryPath := file.path + ".scert-tmp-" + backupID // 上传失败也可能留下半份私钥，必须在开始上传前登记清理路径。 temporary = append(temporary, temporaryPath) - if transactionEnabled { - journalRecord.TemporaryPaths = append(journalRecord.TemporaryPaths, temporaryPath) - } if err := client.Upload(ctx, temporaryPath, file.data, file.mode); err != nil { var notCreated *uploadNotCreatedError if errors.As(err, ¬Created) { @@ -344,7 +380,14 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return failed(fmt.Sprintf("备份目标已存在或无法检查，已停止替换：%s：%v", backupPath, checkErr)) } if transactionEnabled { - journalRecord.Backups = append(journalRecord.Backups, TransactionBackup{Original: file.path, Saved: backupPath, OriginalExists: exists}) + originalDigest := "" + if exists { + originalDigest, err = client.(EvidenceClient).FileDigest(ctx, file.path) + if err != nil { + return failed(fmt.Sprintf("读取旧证书文件摘要失败：%v", err)) + } + } + journalRecord.Backups = append(journalRecord.Backups, TransactionBackup{Original: file.path, Saved: backupPath, OriginalExists: exists, OriginalDigest: originalDigest}) if err := persistTransaction(ctx, TransactionPhaseReplacing); err != nil { return failed(fmt.Sprintf("保存替换阶段失败：%v", err)) } @@ -430,7 +473,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return result } transaction.commit = func(commitContext context.Context) Result { - if err := persistTransaction(commitContext, "committing"); err != nil { + if err := persistTransaction(commitContext, TransactionPhaseCommitting); err != nil { result.Status = "failed" result.FailureCode = FailureCleanup result.Message = fmt.Sprintf("保存提交阶段失败：%v", err) @@ -443,6 +486,11 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return result } result.Message = "证书部署、配置校验并 reload 成功" + if err := persistTransaction(commitContext, TransactionPhaseCommitVerified); err != nil { + result.Status = "failed" + result.FailureCode = FailureCleanup + result.Message = fmt.Sprintf("保存提交核验阶段失败：%v", err) + } return result } transaction.finish = func(retain bool) Result { @@ -450,6 +498,13 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return result } finished = true + if !retain && result.Status == "failed" && result.FailureCode == FailureRecoveryPending { + if err := persistTransaction(context.WithoutCancel(ctx), TransactionPhaseRollbackVerified); err != nil { + retain = true + result.FailureCode = FailureRecoveryUnknown + result.Message = fmt.Sprintf("保存回滚核验阶段失败：%v", err) + } + } if !closed { retainLocks = retain releaseLocks() @@ -494,7 +549,7 @@ func (s *Service) deployOne(ctx context.Context, server Server, artifact Artifac return result } -func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths []remoteFile, lockPaths []string, id string, index int) (*TransactionRecord, bool) { +func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths []remoteFile, lockPaths []string, id string, index int, baseline RecoveryBaseline) (*TransactionRecord, bool) { transactionContext, ok := TransactionContextFromContext(ctx) if !ok || s == nil || s.journal == nil { return nil, false @@ -509,12 +564,22 @@ func (s *Service) newTransactionRecord(ctx context.Context, server Server, paths if s.now != nil { now = s.now().UTC() } + files := make(map[string]string, len(paths)) + for _, file := range paths { + files[file.path] = contentDigest(file.data) + } return &TransactionRecord{ ID: id, TaskID: transactionContext.TaskID, AttemptID: transactionContext.AttemptID, AttemptNumber: transactionContext.AttemptNumber, ResultIndex: index, ServerID: server.ID, ServerName: server.Name, Paths: pathValues, TemporaryPaths: temporaryPaths, LockPaths: append([]string(nil), lockPaths...), - Phase: TransactionPhaseLocked, Status: TransactionStatusPending, UpdatedAt: now, + Phase: TransactionPhaseAcquiring, Status: TransactionStatusPending, UpdatedAt: now, + Evidence: &RecoveryEvidence{ + Version: 1, LockToken: contentDigest([]byte("scert-lock:" + id)), + Address: server.Address, Port: server.Port, HostKeyFingerprint: server.HostKeyFingerprint, + ValidationCommand: append([]string(nil), server.ValidationCommand...), + ReloadCommand: append([]string(nil), server.ReloadCommand...), Files: files, Baseline: baseline, + }, }, true } @@ -532,24 +597,29 @@ func (s *Service) RecoverPending(ctx context.Context, resolve func(context.Conte } var recoveryErrors []error for _, record := range records { - if err := s.recoverTransaction(ctx, record, resolve); err != nil { + if err := s.recoverTransaction(ctx, &record, resolve); err != nil { recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 恢复失败：%w", record.ID, err)) record.Status = TransactionStatusRecoveryRequired - record.Phase = TransactionStatusRecoveryRequired record.Message = err.Error() if updateErr := s.journal.UpdateDeploymentTransaction(context.WithoutCancel(ctx), record); updateErr != nil { recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 保存人工核查状态失败：%w", record.ID, updateErr)) } continue } - if err := s.journal.FinishDeploymentTransaction(context.WithoutCancel(ctx), record.ID, TransactionStatusRolledBack, "服务启动已完成中断部署事务回滚"); err != nil { + status := TransactionStatusRolledBack + message := "服务启动已完成中断部署事务回滚" + if record.Phase == TransactionPhaseCommitting || record.Phase == TransactionPhaseCommitVerified { + status = TransactionStatusCommitted + message = "服务启动已完成中断部署事务提交清理" + } + if err := s.journal.FinishDeploymentTransaction(context.WithoutCancel(ctx), record.ID, status, message); err != nil { recoveryErrors = append(recoveryErrors, fmt.Errorf("事务 %s 保存恢复结果失败：%w", record.ID, err)) } } return errors.Join(recoveryErrors...) } -func (s *Service) recoverTransaction(ctx context.Context, record TransactionRecord, resolve func(context.Context, string) (Server, error)) error { +func (s *Service) recoverTransaction(ctx context.Context, record *TransactionRecord, resolve func(context.Context, string) (Server, error)) error { server, err := resolve(ctx, record.ServerID) if err != nil { return fmt.Errorf("读取服务器配置失败：%w", err) @@ -565,56 +635,17 @@ func (s *Service) recoverTransaction(ctx context.Context, record TransactionReco return fmt.Errorf("连接服务器失败：%w", err) } defer client.Close() - for _, lockPath := range record.LockPaths { - exists, checkErr := client.Exists(recoveryContext, lockPath) - if checkErr != nil || !exists { - if checkErr == nil { - checkErr = errors.New("部署锁不存在") - } - return fmt.Errorf("无法证明事务仍拥有部署锁 %s：%w", lockPath, checkErr) - } + if err := validateRecoveryEvidence(*record, server); err != nil { + return err } - for _, temporaryPath := range record.TemporaryPaths { - if exists, checkErr := client.Exists(recoveryContext, temporaryPath); checkErr != nil { - return fmt.Errorf("检查临时文件失败：%w", checkErr) - } else if exists { - if err := client.Remove(recoveryContext, temporaryPath); err != nil { - return fmt.Errorf("删除临时文件失败：%w", err) - } - } + if record.Phase == TransactionPhaseAcquiring || record.Phase == TransactionPhaseRollbackVerified || record.Phase == TransactionPhaseCommitVerified { + return releaseTransactionLocks(recoveryContext, client, *record, true) } - for index := len(record.Backups) - 1; index >= 0; index-- { - backup := record.Backups[index] - backupExists, checkErr := client.Exists(recoveryContext, backup.Saved) - if checkErr != nil { - return fmt.Errorf("检查备份文件失败：%w", checkErr) - } - targetExists, checkErr := client.Exists(recoveryContext, backup.Original) - if checkErr != nil { - return fmt.Errorf("检查目标文件失败：%w", checkErr) - } - if backup.OriginalExists { - if !backupExists { - if targetExists { - continue - } - return fmt.Errorf("备份文件和目标文件均不存在：%s", backup.Original) - } - if targetExists { - if err := client.Remove(recoveryContext, backup.Original); err != nil { - return fmt.Errorf("删除待恢复目标文件失败：%w", err) - } - } - if err := client.Rename(recoveryContext, backup.Saved, backup.Original); err != nil { - return fmt.Errorf("恢复备份文件失败：%w", err) - } - continue - } - if targetExists { - if err := client.Remove(recoveryContext, backup.Original); err != nil { - return fmt.Errorf("删除新增目标文件失败：%w", err) - } - } + if record.Phase == TransactionPhaseCommitting { + return s.recoverCommittedTransaction(recoveryContext, record, server, client) + } + if err := restoreTransactionFiles(recoveryContext, client, *record); err != nil { + return err } if record.ReloadAttempted { for _, command := range [][]string{server.ValidationCommand, server.ReloadCommand} { @@ -626,15 +657,132 @@ func (s *Service) recoverTransaction(ctx context.Context, record TransactionReco return fmt.Errorf("恢复服务失败：%w", runErr) } } + if s.VerifyRecoveryTarget == nil || record.Evidence.Baseline.PreviousFingerprint == "" { + return errors.New("缺少可信站点恢复基线，已保留现场等待人工核查") + } + if err := s.VerifyRecoveryTarget(recoveryContext, server, record.Evidence.Baseline, false); err != nil { + return fmt.Errorf("恢复站点核验失败：%w", err) + } } + if err := s.persistRecoveryPhase(recoveryContext, record, TransactionPhaseRollbackVerified); err != nil { + return err + } + return releaseTransactionLocks(recoveryContext, client, *record, true) +} + +func (s *Service) recoverCommittedTransaction(ctx context.Context, record *TransactionRecord, server Server, client Client) error { + for _, lockPath := range record.LockPaths { + if err := checkOwnedLock(ctx, client, lockPath, record.Evidence.LockToken, false); err != nil { + return fmt.Errorf("无法证明提交事务仍拥有部署锁 %s：%w", lockPath, err) + } + } + for _, path := range record.Paths { + if err := checkFileDigest(ctx, client, path, record.Evidence.Files[path]); err != nil { + return err + } + } + if s.VerifyRecoveryTarget == nil || record.Evidence.Baseline.ExpectedFingerprint == "" { + return errors.New("缺少可信提交站点基线，已保留现场等待人工核查") + } + if err := s.VerifyRecoveryTarget(ctx, server, record.Evidence.Baseline, true); err != nil { + return fmt.Errorf("提交站点核验失败：%w", err) + } + for _, temporary := range record.TemporaryPaths { + if err := client.Remove(ctx, temporary); err != nil { + return fmt.Errorf("清理提交临时文件失败：%w", err) + } + } + for _, backup := range record.Backups { + if exists, err := client.Exists(ctx, backup.Saved); err != nil { + return err + } else if exists { + if err := checkFileDigest(ctx, client, backup.Saved, backup.OriginalDigest); err != nil { + return err + } + } + if err := client.Remove(ctx, backup.Saved); err != nil { + return fmt.Errorf("清理提交备份失败：%w", err) + } + } + if err := s.persistRecoveryPhase(ctx, record, TransactionPhaseCommitVerified); err != nil { + return err + } + return releaseTransactionLocks(ctx, client, *record, true) +} + +func restoreTransactionFiles(ctx context.Context, client Client, record TransactionRecord) error { + for _, temporaryPath := range record.TemporaryPaths { + if exists, err := client.Exists(ctx, temporaryPath); err != nil { + return fmt.Errorf("检查临时文件失败：%w", err) + } else if exists { + if err := client.Remove(ctx, temporaryPath); err != nil { + return fmt.Errorf("删除临时文件失败：%w", err) + } + } + } + for index := len(record.Backups) - 1; index >= 0; index-- { + backup := record.Backups[index] + backupExists, err := client.Exists(ctx, backup.Saved) + if err != nil { + return fmt.Errorf("检查备份文件失败：%w", err) + } + targetExists, err := client.Exists(ctx, backup.Original) + if err != nil { + return fmt.Errorf("检查目标文件失败：%w", err) + } + if backup.OriginalExists { + if !backupExists { + if !targetExists { + return fmt.Errorf("备份文件和目标文件均不存在：%s", backup.Original) + } + if err := checkFileDigest(ctx, client, backup.Original, backup.OriginalDigest); err != nil { + return fmt.Errorf("无法确认目标文件已恢复：%w", err) + } + continue + } + if targetExists { + if err := client.Remove(ctx, backup.Original); err != nil { + return fmt.Errorf("删除待恢复目标文件失败：%w", err) + } + } + if err := client.Rename(ctx, backup.Saved, backup.Original); err != nil { + return fmt.Errorf("恢复备份文件失败：%w", err) + } + continue + } + if targetExists { + if err := client.Remove(ctx, backup.Original); err != nil { + return fmt.Errorf("删除新增目标文件失败：%w", err) + } + } + } + return nil +} + +func releaseTransactionLocks(ctx context.Context, client Client, record TransactionRecord, verifyOwner bool) error { for index := len(record.LockPaths) - 1; index >= 0; index-- { - if err := client.Unlock(recoveryContext, record.LockPaths[index]); err != nil { + lockPath := record.LockPaths[index] + if verifyOwner { + if err := checkOwnedLock(ctx, client, lockPath, record.Evidence.LockToken, false); err != nil { + return fmt.Errorf("无法证明事务仍拥有部署锁 %s：%w", lockPath, err) + } + } + if err := client.Unlock(ctx, lockPath); err != nil { return fmt.Errorf("释放部署锁失败：%w", err) } } return nil } +func (s *Service) persistRecoveryPhase(ctx context.Context, record *TransactionRecord, phase string) error { + record.Phase = phase + record.UpdatedAt = time.Now().UTC() + if s.now != nil { + record.UpdatedAt = s.now().UTC() + } + return s.journal.UpdateDeploymentTransaction(ctx, *record) +} + func cleanupBackups(ctx context.Context, client Client, backups []backupFile, paths []remoteFile) error { seen := make(map[string]struct{}, len(backups)) for _, item := range backups { diff --git a/backend/internal/deployment/service_test.go b/backend/internal/deployment/service_test.go index fbd667d..89291c5 100644 --- a/backend/internal/deployment/service_test.go +++ b/backend/internal/deployment/service_test.go @@ -63,10 +63,35 @@ func (c *memoryClient) Lock(_ context.Context, path string) error { } func (c *memoryClient) Unlock(_ context.Context, path string) error { + delete(c.files, path+"/owner") delete(c.files, path) return nil } +func (c *memoryClient) CreateOwnedLock(_ context.Context, path, token string) error { + if _, exists := c.files[path]; !exists { + return errors.New("模拟锁不存在") + } + c.files[path+"/owner"] = memoryFile{data: []byte(token), mode: 0o600} + return nil +} + +func (c *memoryClient) ReadLockOwner(_ context.Context, path string) (string, error) { + file, ok := c.files[path+"/owner"] + if !ok { + return "", nil + } + return string(file.data), nil +} + +func (c *memoryClient) FileDigest(_ context.Context, path string) (string, error) { + file, ok := c.files[path] + if !ok { + return "", errors.New("模拟文件不存在") + } + return contentDigest(file.data), nil +} + func (c *memoryClient) Exists(_ context.Context, path string) (bool, error) { _, ok := c.files[path] return ok, nil diff --git a/backend/internal/deployment/ssh.go b/backend/internal/deployment/ssh.go index 1d94e0c..16e057e 100644 --- a/backend/internal/deployment/ssh.go +++ b/backend/internal/deployment/ssh.go @@ -2,7 +2,9 @@ package deployment import ( "context" + "crypto/sha256" "encoding/base64" + "encoding/hex" "errors" "fmt" "io" @@ -302,12 +304,65 @@ func (c *sshClientAdapter) Lock(ctx context.Context, path string) error { } func (c *sshClientAdapter) Unlock(ctx context.Context, path string) error { - return c.fileOperation(ctx, func() error { return c.sftp.RemoveDirectory(path) }) + return c.fileOperation(ctx, func() error { + if err := c.sftp.Remove(path + "/owner"); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return c.sftp.RemoveDirectory(path) + }) +} + +func (c *sshClientAdapter) CreateOwnedLock(ctx context.Context, path, token string) error { + if !validDigest(token) { + return errors.New("部署锁令牌格式无效") + } + return c.Upload(ctx, path+"/owner", []byte(token), 0o600) +} + +func (c *sshClientAdapter) ReadLockOwner(ctx context.Context, path string) (string, error) { + return sshOperation(ctx, c, func() (string, error) { + file, err := c.sftp.Open(path + "/owner") + if errors.Is(err, os.ErrNotExist) { + return "", nil + } + if err != nil { + return "", err + } + defer file.Close() + data, err := io.ReadAll(file) + return strings.TrimSpace(string(data)), err + }) +} + +func (c *sshClientAdapter) FileDigest(ctx context.Context, path string) (string, error) { + return sshOperation(ctx, c, func() (string, error) { + info, err := c.sftp.Lstat(path) + if err != nil { + return "", err + } + if !info.Mode().IsRegular() || info.Size() > 16*1024*1024 { + return "", errors.New("证书文件类型或大小无效") + } + file, err := c.sftp.Open(path) + if err != nil { + return "", err + } + defer file.Close() + hash := sha256.New() + count, err := io.Copy(hash, io.LimitReader(file, 16*1024*1024+1)) + if err != nil { + return "", err + } + if count > 16*1024*1024 { + return "", errors.New("证书文件超过摘要读取上限") + } + return hex.EncodeToString(hash.Sum(nil)), nil + }) } func (c *sshClientAdapter) Exists(ctx context.Context, path string) (bool, error) { return sshOperation(ctx, c, func() (bool, error) { - _, err := c.sftp.Stat(path) + _, err := c.sftp.Lstat(path) if errors.Is(err, os.ErrNotExist) { return false, nil } diff --git a/backend/internal/deployment/transaction.go b/backend/internal/deployment/transaction.go index 708b2f2..b3ffc26 100644 --- a/backend/internal/deployment/transaction.go +++ b/backend/internal/deployment/transaction.go @@ -11,6 +11,7 @@ type TransactionBackup struct { Original string `json:"original"` Saved string `json:"saved"` OriginalExists bool `json:"original_exists"` + OriginalDigest string `json:"original_digest,omitempty"` } // TransactionRecord 保存远程部署事务的可恢复现场，不包含证书内容或凭据。 @@ -32,6 +33,7 @@ type TransactionRecord struct { Status string Message string UpdatedAt time.Time + Evidence *RecoveryEvidence } // TransactionJournal 持久化部署现场，保证进程重启后仍能识别未提交事务。 @@ -48,13 +50,17 @@ const ( TransactionStatusRolledBack = "rolled_back" TransactionStatusRecoveryRequired = "recovery_required" - TransactionPhaseLocked = "locked" - TransactionPhaseUploaded = "uploaded" - TransactionPhaseReplacing = "replacing" - TransactionPhaseInstalled = "installed" - TransactionPhaseReloadAttempted = "reload_attempted" - TransactionPhaseCommitted = "committed" - TransactionPhaseRolledBack = "rolled_back" + TransactionPhaseLocked = "locked" + TransactionPhaseAcquiring = "acquiring" + TransactionPhaseRollbackVerified = "rollback_verified" + TransactionPhaseCommitVerified = "commit_verified" + TransactionPhaseCommitting = "committing" + TransactionPhaseUploaded = "uploaded" + TransactionPhaseReplacing = "replacing" + TransactionPhaseInstalled = "installed" + TransactionPhaseReloadAttempted = "reload_attempted" + TransactionPhaseCommitted = "committed" + TransactionPhaseRolledBack = "rolled_back" ) type transactionContextKey struct{} diff --git a/backend/internal/deployment/transaction_recovery_test.go b/backend/internal/deployment/transaction_recovery_test.go index 75779e3..a814807 100644 --- a/backend/internal/deployment/transaction_recovery_test.go +++ b/backend/internal/deployment/transaction_recovery_test.go @@ -40,31 +40,37 @@ func (j *memoryTransactionJournal) ListPendingDeploymentTransactions(context.Con func TestRecoverPendingRollsBackOwnedDeployment(t *testing.T) { server := testServer("recovery") + server = server.Normalize() client := newMemoryClient() id := "recovery-transaction" paths := []string{server.CertificatePath, server.CertificateChainPath, server.PrivateKeyPath} locks := make([]string, 0, len(paths)) backups := make([]TransactionBackup, 0, len(paths)) temporary := make([]string, 0, len(paths)) + files := make(map[string]string, len(paths)) for _, path := range paths { lockPath := path + ".scert-lock" backupPath := path + ".scert-backup-" + id temporaryPath := path + ".scert-tmp-" + id locks = append(locks, lockPath) - backups = append(backups, TransactionBackup{Original: path, Saved: backupPath, OriginalExists: true}) + backups = append(backups, TransactionBackup{Original: path, Saved: backupPath, OriginalExists: true, OriginalDigest: contentDigest([]byte("old"))}) temporary = append(temporary, temporaryPath) client.files[lockPath] = memoryFile{} + client.files[lockPath+"/owner"] = memoryFile{data: []byte(contentDigest([]byte("scert-lock:" + id))), mode: 0o600} client.files[backupPath] = memoryFile{data: []byte("old"), mode: 0o600} client.files[path] = memoryFile{data: []byte("new"), mode: 0o600} client.files[temporaryPath] = memoryFile{data: []byte("partial"), mode: 0o600} + files[path] = contentDigest([]byte("old")) } journal := &memoryTransactionJournal{pending: []TransactionRecord{{ ID: id, TaskID: "task", AttemptID: "attempt", AttemptNumber: 1, ResultIndex: 0, ServerID: server.ID, ServerName: server.Name, Paths: paths, TemporaryPaths: temporary, InstalledPaths: paths, Backups: backups, LockPaths: locks, ReloadAttempted: true, Phase: TransactionPhaseReloadAttempted, Status: TransactionStatusPending, + Evidence: &RecoveryEvidence{Version: 1, LockToken: contentDigest([]byte("scert-lock:" + id)), Address: server.Address, Port: server.Port, HostKeyFingerprint: server.HostKeyFingerprint, ValidationCommand: append([]string(nil), server.ValidationCommand...), ReloadCommand: append([]string(nil), server.ReloadCommand...), Files: files, Baseline: RecoveryBaseline{PreviousFingerprint: "old", Endpoint: "https://example.com", Domains: []string{"example.com"}}}, }}} service := NewServiceWithJournal(memoryFactory{clients: map[string]*memoryClient{server.ID: client}}, journal) + service.VerifyRecoveryTarget = func(context.Context, Server, RecoveryBaseline, bool) error { return nil } if err := service.RecoverPending(t.Context(), func(context.Context, string) (Server, error) { return server, nil }); err != nil { t.Fatalf("事务恢复失败：%v", err) } diff --git a/backend/internal/repository/deployment_transaction_repository.go b/backend/internal/repository/deployment_transaction_repository.go index 9825980..59bf2b6 100644 --- a/backend/internal/repository/deployment_transaction_repository.go +++ b/backend/internal/repository/deployment_transaction_repository.go @@ -41,11 +41,11 @@ func (r *SQLDeploymentTransactionRepository) CreateDeploymentTransaction(ctx con (id, task_id, attempt_id, attempt_number, result_index, server_id, server_name, certificate_path, certificate_chain_path, private_key_path, temporary_paths_json, installed_paths_json, backups_json, lock_paths_json, reload_attempted, phase, status, - message, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), + message, updated_at, recovery_evidence_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`), item.ID, item.TaskID, item.AttemptID, item.AttemptNumber, item.ResultIndex, item.ServerID, item.ServerName, values.paths[0], values.paths[1], values.paths[2], values.temporary, values.installed, values.backups, - values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt) + values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, values.evidence) if err != nil { return fmt.Errorf("保存部署事务现场失败：%w", err) } @@ -66,9 +66,9 @@ func (r *SQLDeploymentTransactionRepository) UpdateDeploymentTransaction(ctx con result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET certificate_path = ?, certificate_chain_path = ?, private_key_path = ?, temporary_paths_json = ?, installed_paths_json = ?, backups_json = ?, lock_paths_json = ?, - reload_attempted = ?, phase = ?, status = ?, message = ?, updated_at = ? WHERE id = ?`), + reload_attempted = ?, phase = ?, status = ?, message = ?, updated_at = ?, recovery_evidence_json = ? WHERE id = ?`), values.paths[0], values.paths[1], values.paths[2], values.temporary, values.installed, values.backups, - values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, item.ID) + values.locks, boolInt(item.ReloadAttempted), item.Phase, item.Status, item.Message, values.updatedAt, values.evidence, item.ID) if err != nil { return fmt.Errorf("更新部署事务现场失败：%w", err) } @@ -91,8 +91,8 @@ func (r *SQLDeploymentTransactionRepository) FinishDeploymentTransaction(ctx con if id == "" || status == "" { return errors.New("部署事务状态参数不能为空") } - result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET status = ?, phase = ?, message = ?, updated_at = ? WHERE id = ?`), - status, status, message, time.Now().UTC(), id) + result, err := r.db.ExecContext(ctx, r.query(`UPDATE deployment_transactions SET status = ?, phase = CASE WHEN ? = 'recovery_required' THEN phase ELSE ? END, message = ?, updated_at = ? WHERE id = ?`), + status, status, status, message, time.Now().UTC(), id) if err != nil { return fmt.Errorf("完成部署事务记录失败：%w", err) } @@ -113,7 +113,7 @@ func (r *SQLDeploymentTransactionRepository) ListPendingDeploymentTransactions(c rows, err := r.db.QueryContext(ctx, r.query(`SELECT id, task_id, attempt_id, attempt_number, result_index, server_id, server_name, certificate_path, certificate_chain_path, private_key_path, temporary_paths_json, installed_paths_json, backups_json, lock_paths_json, - reload_attempted, phase, status, message, updated_at + reload_attempted, phase, status, message, updated_at, recovery_evidence_json FROM deployment_transactions WHERE status IN ('pending', 'recovery_required') ORDER BY updated_at, id`)) if err != nil { return nil, fmt.Errorf("查询待恢复部署事务失败：%w", err) @@ -140,6 +140,7 @@ type deploymentTransactionValues struct { backups string locks string updatedAt time.Time + evidence string } func validateDeploymentTransaction(item deployment.TransactionRecord) error { @@ -159,6 +160,10 @@ func validateDeploymentTransaction(item deployment.TransactionRecord) error { } func encodeDeploymentTransaction(item deployment.TransactionRecord) (deploymentTransactionValues, error) { + evidence, err := json.Marshal(item.Evidence) + if err != nil { + return deploymentTransactionValues{}, fmt.Errorf("编码部署恢复证据失败：%w", err) + } paths := [3]string{strings.TrimSpace(item.Paths[0]), strings.TrimSpace(item.Paths[1]), strings.TrimSpace(item.Paths[2])} temporary, err := json.Marshal(item.TemporaryPaths) if err != nil { @@ -183,7 +188,7 @@ func encodeDeploymentTransaction(item deployment.TransactionRecord) (deploymentT updatedAt = updatedAt.UTC() } return deploymentTransactionValues{ - paths: paths, temporary: string(temporary), installed: string(installed), backups: string(backups), locks: string(locks), updatedAt: updatedAt, + paths: paths, temporary: string(temporary), installed: string(installed), backups: string(backups), locks: string(locks), updatedAt: updatedAt, evidence: string(evidence), }, nil } @@ -196,12 +201,18 @@ func scanDeploymentTransaction(row deploymentTransactionScanner) (deployment.Tra var paths [3]string var temporary, installed, backups, locks string var reloadAttempted int + var evidence sql.NullString if err := row.Scan(&item.ID, &item.TaskID, &item.AttemptID, &item.AttemptNumber, &item.ResultIndex, &item.ServerID, &item.ServerName, &paths[0], &paths[1], &paths[2], &temporary, &installed, &backups, &locks, - &reloadAttempted, &item.Phase, &item.Status, &item.Message, &item.UpdatedAt); err != nil { + &reloadAttempted, &item.Phase, &item.Status, &item.Message, &item.UpdatedAt, &evidence); err != nil { return deployment.TransactionRecord{}, fmt.Errorf("读取部署事务失败：%w", err) } item.Paths = paths[:] + if evidence.Valid && evidence.String != "" { + if err := json.Unmarshal([]byte(evidence.String), &item.Evidence); err != nil { + return deployment.TransactionRecord{}, fmt.Errorf("解析部署恢复证据失败：%w", err) + } + } item.ReloadAttempted = reloadAttempted != 0 if err := json.Unmarshal([]byte(temporary), &item.TemporaryPaths); err != nil { return deployment.TransactionRecord{}, fmt.Errorf("解析部署临时路径失败：%w", err)