保存人工处置准备绑定与重新核验开发检查点

This commit is contained in:
xuans 2026-10-03 19:23:14 +08:00
parent 6b77730114
commit ed0c97b957
4 changed files with 490 additions and 0 deletions

View File

@ -44,6 +44,10 @@ type RecoveryLockObservation struct {
// InspectPending 只读取待恢复记录与远端文件,绝不运行命令、建立锁或修改事务。
// 缺失的历史证据保持缺失,当前观测摘要不能补写为当时的恢复证据。
func (s *Service) InspectPending(ctx context.Context, id string, resolve func(context.Context, string) (Server, error)) (RecoveryInspection, error) {
return s.inspectPending(ctx, id, resolve, nil)
}
func (s *Service) inspectPending(ctx context.Context, id string, resolve func(context.Context, string) (Server, error), bind func(TransactionRecord, Server) error) (RecoveryInspection, error) {
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if err := ctx.Err(); err != nil {
@ -85,6 +89,11 @@ func (s *Service) InspectPending(ctx context.Context, id string, resolve func(co
if record.Evidence != nil && validateRecoveryEvidence(*record, server) != nil {
return RecoveryInspection{}, ErrRecoveryInspectionIdentity
}
if bind != nil {
if err := bind(*record, server); err != nil {
return RecoveryInspection{}, err
}
}
if err := ctx.Err(); err != nil {
return RecoveryInspection{}, err
}

View File

@ -0,0 +1,204 @@
package deployment
import (
"context"
"crypto/sha256"
"encoding/json"
"errors"
"strings"
"time"
"unicode"
)
type RecoveryAction string
const (
RecoveryActionRecord RecoveryAction = "record_inspection"
RecoveryActionCommit RecoveryAction = "finish_commit"
RecoveryActionRollback RecoveryAction = "restore_baseline"
RecoveryReviewLifetime = time.Minute
)
var (
ErrRecoveryReviewInvalid = errors.New("人工处置准备参数无效")
ErrRecoveryReviewExpired = errors.New("人工处置准备已过期,请重新核查")
ErrRecoveryReviewChanged = errors.New("事务、服务器或现场已变化,请重新准备处置")
ErrRecoveryReviewEvidence = errors.New("历史恢复证据不足,只能记录核查结论,不能提交或回滚")
)
type RecoveryReviewSummary struct {
TransactionID string `json:"transaction_id"`
ServerID string `json:"server_id"`
Phase string `json:"phase"`
Action RecoveryAction `json:"action"`
HistoricalEvidence bool `json:"historical_evidence"`
ObservedAt time.Time `json:"observed_at"`
ExpiresAt time.Time `json:"expires_at"`
}
// RecoveryReview 只保存只读准备摘要和不可公开的绑定值,不保存凭据或原始恢复证据。
// 它不是执行授权;调用者仍须取得持久化占用、验证动作条件并原子写入审计。
type RecoveryReview struct {
summary RecoveryReviewSummary
operator string
recordDigest [sha256.Size]byte
serverDigest [sha256.Size]byte
observationDigest [sha256.Size]byte
service *Service
}
func (r RecoveryReview) Summary() RecoveryReviewSummary { return r.summary }
func validReviewOperator(operator string) bool {
return strings.TrimSpace(operator) == operator && len(operator) > 0 && len(operator) <= 160 && !strings.ContainsFunc(operator, unicode.IsControl)
}
func validRecoveryAction(action RecoveryAction) bool {
return action == RecoveryActionRecord || action == RecoveryActionCommit || action == RecoveryActionRollback
}
// PrepareRecoveryReview 在同一操作上限内读取前后记录,拒绝采样期间发生的配置或事务变化。
// 当前采样摘要只绑定本次准备,不补写到历史 Evidence,也不改变事务终态。
func (s *Service) PrepareRecoveryReview(ctx context.Context, id, operator string, action RecoveryAction, resolve func(context.Context, string) (Server, error)) (RecoveryReview, error) {
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
if !validReviewOperator(operator) || !validRecoveryAction(action) || !validInspectionID(id) {
return RecoveryReview{}, ErrRecoveryReviewInvalid
}
if s == nil || s.now == nil || s.journal == nil || s.factory == nil || resolve == nil {
return RecoveryReview{}, ErrRecoveryInspectionUnavailable
}
started := s.now()
record, recordDigest, serverDigest, err := s.reviewRecord(ctx, id, resolve)
if err != nil {
return RecoveryReview{}, err
}
if action != RecoveryActionRecord && record.Evidence == nil {
return RecoveryReview{}, ErrRecoveryReviewEvidence
}
inspection, err := s.inspectPending(ctx, id, resolve, func(sampleRecord TransactionRecord, sampleServer Server) error {
sampleRecordDigest, err := reviewDigest(sampleRecord)
if err != nil {
return err
}
sampleServerDigest, err := reviewServerDigest(sampleServer)
if err != nil {
return err
}
if recordDigest != sampleRecordDigest || serverDigest != sampleServerDigest {
return ErrRecoveryReviewChanged
}
return nil
})
if err != nil {
return RecoveryReview{}, err
}
_, currentRecord, currentServer, err := s.reviewRecord(ctx, id, resolve)
if err != nil {
return RecoveryReview{}, err
}
if recordDigest != currentRecord || serverDigest != currentServer {
return RecoveryReview{}, ErrRecoveryReviewChanged
}
observedAt := inspection.ObservedAt
inspection.ObservedAt = time.Time{}
observationDigest, err := reviewDigest(inspection)
if err != nil {
return RecoveryReview{}, err
}
if err := ctx.Err(); err != nil {
return RecoveryReview{}, err
}
expiresAt := started.Add(RecoveryReviewLifetime)
if !s.now().Before(expiresAt) {
return RecoveryReview{}, ErrRecoveryReviewExpired
}
return RecoveryReview{
summary: RecoveryReviewSummary{TransactionID: id, ServerID: record.ServerID, Phase: record.Phase,
Action: action, HistoricalEvidence: record.Evidence != nil, ObservedAt: observedAt, ExpiresAt: expiresAt},
operator: operator, recordDigest: recordDigest, serverDigest: serverDigest, observationDigest: observationDigest, service: s,
}, nil
}
// RecheckRecoveryReview 仅验证重新读取的状态与原准备一致,不消费确认凭据或执行远端变更。
// 后续执行器必须在持久化互斥范围内调用,并再次执行动作专属的锁、文件和严格 HTTPS 检查。
func (s *Service) RecheckRecoveryReview(ctx context.Context, review RecoveryReview, operator string, resolve func(context.Context, string) (Server, error)) error {
if s == nil || review.service != s || review.operator != operator || !validReviewOperator(operator) {
return ErrRecoveryReviewInvalid
}
if s.now == nil || !s.now().Before(review.summary.ExpiresAt) {
return ErrRecoveryReviewExpired
}
current, err := s.PrepareRecoveryReview(ctx, review.summary.TransactionID, operator, review.summary.Action, resolve)
if err != nil {
return err
}
if !s.now().Before(review.summary.ExpiresAt) {
return ErrRecoveryReviewExpired
}
if current.recordDigest != review.recordDigest || current.serverDigest != review.serverDigest || current.observationDigest != review.observationDigest {
return ErrRecoveryReviewChanged
}
return ctx.Err()
}
func reviewDigest(value any) ([sha256.Size]byte, error) {
data, err := json.Marshal(value)
if err != nil {
return [sha256.Size]byte{}, ErrRecoveryReviewInvalid
}
return sha256.Sum256(data), nil
}
func (s *Service) reviewRecord(ctx context.Context, id string, resolve func(context.Context, string) (Server, error)) (TransactionRecord, [sha256.Size]byte, [sha256.Size]byte, error) {
fail := func(err error) (TransactionRecord, [sha256.Size]byte, [sha256.Size]byte, error) {
return TransactionRecord{}, [sha256.Size]byte{}, [sha256.Size]byte{}, err
}
if err := ctx.Err(); err != nil {
return fail(err)
}
records, err := s.journal.ListPendingDeploymentTransactions(ctx)
if err != nil {
return fail(inspectionError(ctx, ErrRecoveryInspectionUnavailable))
}
var found *TransactionRecord
for i := range records {
if records[i].ID == id {
if found != nil {
return fail(ErrRecoveryInspectionRecord)
}
found = &records[i]
}
}
if found == nil || validateInspectionRecord(*found) != nil || !validRecoverySummary(found.ServerID) ||
(found.Phase != "" && !validRecoverySummary(found.Phase)) {
return fail(ErrRecoveryInspectionRecord)
}
server, err := resolve(ctx, found.ServerID)
if err != nil {
return fail(inspectionError(ctx, ErrRecoveryInspectionUnavailable))
}
server = server.Normalize()
if server.ID != found.ServerID || server.Validate() != nil || (found.Evidence != nil && validateRecoveryEvidence(*found, server) != nil) {
return fail(ErrRecoveryInspectionIdentity)
}
recordDigest, err := reviewDigest(found)
if err != nil {
return fail(err)
}
serverDigest, err := reviewServerDigest(server)
if err != nil {
return fail(err)
}
if err := ctx.Err(); err != nil {
return fail(err)
}
return *found, recordDigest, serverDigest, nil
}
func reviewServerDigest(server Server) ([sha256.Size]byte, error) {
return reviewDigest(struct {
Server Server
CredentialReference string
}{Server: server, CredentialReference: server.PasswordRef})
}

View File

@ -0,0 +1,266 @@
package deployment
import (
"context"
"encoding/json"
"errors"
"maps"
"reflect"
"strings"
"testing"
"time"
)
func TestRecoveryReviewReadOnlyBindings(t *testing.T) {
for _, action := range []RecoveryAction{RecoveryActionRecord, RecoveryActionCommit, RecoveryActionRollback} {
t.Run(string(action), func(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
before := maps.Clone(memory.files)
encodedBefore, _ := json.Marshal(record)
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
client := &inspectionClient{memoryClient: memory}
service := NewServiceWithJournal(fixedClientFactory{client}, journal)
resolve := func(context.Context, string) (Server, error) { return server, nil }
review, err := service.PrepareRecoveryReview(t.Context(), record.ID, "unix:0", action, resolve)
if err != nil {
t.Fatal(err)
}
summary := review.Summary()
if summary.TransactionID != record.ID || summary.ServerID != record.ServerID || summary.Action != action ||
!summary.HistoricalEvidence || summary.ObservedAt.IsZero() || !summary.ExpiresAt.After(summary.ObservedAt) {
t.Fatal("invalid preparation summary")
}
summary.TransactionID = "tampered"
if err := service.RecheckRecoveryReview(t.Context(), review, "unix:0", resolve); err != nil {
t.Fatal(err)
}
encodedAfter, _ := json.Marshal(journal.pending[0])
if client.mutations != 0 || !reflect.DeepEqual(before, memory.files) || string(encodedAfter) != string(encodedBefore) {
t.Fatal("review changed original transaction or remote state")
}
encoded, _ := json.Marshal(review)
if string(encoded) != "{}" {
t.Fatal("private bindings serialized")
}
encoded, _ = json.Marshal(review.Summary())
if strings.Contains(string(encoded), record.Evidence.LockToken) || strings.Contains(string(encoded), "credential") {
t.Fatal("public summary contains secret evidence")
}
})
}
}
func TestRecoveryReviewLegacyCannotPrepareWrites(t *testing.T) {
for _, action := range []RecoveryAction{RecoveryActionRecord, RecoveryActionCommit, RecoveryActionRollback} {
t.Run(string(action), func(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
record.Evidence = nil
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
client := &inspectionClient{memoryClient: memory}
factory := &inspectionFactory{client: client}
service := NewServiceWithJournal(factory, journal)
review, err := service.PrepareRecoveryReview(t.Context(), record.ID, "unix:0", action, func(context.Context, string) (Server, error) { return server, nil })
if action == RecoveryActionRecord {
if err != nil || review.Summary().HistoricalEvidence {
t.Fatalf("legacy read failed: %v", err)
}
} else if !errors.Is(err, ErrRecoveryReviewEvidence) || factory.calls != 0 || review.Summary().TransactionID != "" {
t.Fatal("legacy write preparation accepted or connected remotely")
}
if journal.pending[0].Evidence != nil || client.mutations != 0 {
t.Fatal("legacy evidence manufactured")
}
})
}
}
func TestRecoveryReviewRejectsChanges(t *testing.T) {
for _, scenario := range []string{"phase", "message", "time", "target", "backup", "temporary", "lock", "server-user", "server-password-reference", "server-fingerprint", "server-command", "server-path", "record-missing"} {
t.Run(scenario, func(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
client := &inspectionClient{memoryClient: memory}
service := NewServiceWithJournal(fixedClientFactory{client}, journal)
resolve := func(context.Context, string) (Server, error) { return server, nil }
review, err := service.PrepareRecoveryReview(t.Context(), record.ID, "unix:0", RecoveryActionRollback, resolve)
if err != nil {
t.Fatal(err)
}
switch scenario {
case "phase":
journal.pending[0].Phase = TransactionPhaseCommitting
case "message":
journal.pending[0].Message = "changed"
case "time":
journal.pending[0].UpdatedAt = time.Now().UTC()
case "target":
memory.files[record.Paths[0]] = memoryFile{data: []byte("changed")}
case "backup":
memory.files[record.Backups[0].Saved] = memoryFile{data: []byte("changed")}
case "temporary":
delete(memory.files, record.TemporaryPaths[0])
case "lock":
memory.files[record.LockPaths[0]+"/owner"] = memoryFile{data: []byte(contentDigest([]byte("foreign")))}
case "server-user":
server.Username = "other"
case "server-password-reference":
server.PasswordRef = "changed-reference"
case "server-fingerprint":
server.HostKeyFingerprint = "SHA256:" + strings.Repeat("B", 43)
case "server-command":
server.ReloadCommand = []string{"systemctl", "reload", "other"}
case "server-path":
server.CertificatePath = "/different/cert.pem"
case "record-missing":
journal.pending = nil
}
if err := service.RecheckRecoveryReview(t.Context(), review, "unix:0", resolve); err == nil {
t.Fatal("changed state reused old preparation")
}
if client.mutations != 0 {
t.Fatal("recheck modified remote state")
}
})
}
}
func TestRecoveryReviewRejectsChangesDuringSampling(t *testing.T) {
for _, scenario := range []string{"record", "server", "cancel"} {
t.Run(scenario, func(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
client := &inspectionClient{memoryClient: memory}
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
client.afterDigest = func() {
switch scenario {
case "record":
journal.pending[0].Message = "changed"
case "server":
server.PasswordRef = "changed-reference"
case "cancel":
cancel()
}
}
service := NewServiceWithJournal(fixedClientFactory{client}, journal)
review, err := service.PrepareRecoveryReview(ctx, record.ID, "unix:0", RecoveryActionRecord, func(context.Context, string) (Server, error) { return server, nil })
want := ErrRecoveryReviewChanged
if scenario == "cancel" {
want = context.Canceled
}
if !errors.Is(err, want) || review.Summary().TransactionID != "" || client.mutations != 0 {
t.Fatalf("sampling change was not rejected: %v", err)
}
})
}
}
func TestRecoveryReviewExpiryAndIdentity(t *testing.T) {
for _, scenario := range []string{"expired", "expires-during-recheck", "operator", "other-service", "zero", "canceled"} {
t.Run(scenario, func(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
client := &inspectionClient{memoryClient: memory}
factory := &inspectionFactory{client: client}
service := NewServiceWithJournal(factory, journal)
now := time.Now()
service.now = func() time.Time { return now }
resolve := func(context.Context, string) (Server, error) { return server, nil }
review, err := service.PrepareRecoveryReview(t.Context(), record.ID, "unix:0", RecoveryActionRecord, resolve)
if err != nil {
t.Fatal(err)
}
calls := factory.calls
operator := "unix:0"
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
want := ErrRecoveryReviewInvalid
switch scenario {
case "expired":
now = review.Summary().ExpiresAt
want = ErrRecoveryReviewExpired
case "expires-during-recheck":
client.afterDigest = func() { now = review.Summary().ExpiresAt }
want = ErrRecoveryReviewExpired
case "operator":
operator = "unix:1"
case "other-service":
service = NewServiceWithJournal(factory, journal)
case "zero":
review = RecoveryReview{}
case "canceled":
cancel()
want = context.Canceled
}
if err := service.RecheckRecoveryReview(ctx, review, operator, resolve); !errors.Is(err, want) {
t.Fatalf("wrong expiry or identity result: %v", err)
}
if scenario != "expires-during-recheck" && factory.calls != calls {
t.Fatal("rejected identity or expiry read remote state")
}
if client.mutations != 0 {
t.Fatal("rejected review mutated remote state")
}
})
}
}
func TestRecoveryReviewBindsActualSampleConfiguration(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
client := &inspectionClient{memoryClient: memory}
factory := &inspectionFactory{client: client}
service := NewServiceWithJournal(factory, journal)
reads := 0
resolve := func(context.Context, string) (Server, error) {
reads++
value := server
if reads == 2 {
value.PasswordRef = "transient-configuration"
}
return value, nil
}
review, err := service.PrepareRecoveryReview(t.Context(), record.ID, "unix:0", RecoveryActionRecord, resolve)
if !errors.Is(err, ErrRecoveryReviewChanged) || review.Summary().TransactionID != "" || factory.calls != 0 {
t.Fatal("sample used a different configuration from its binding")
}
}
func TestRecoveryReviewRejectsInvalidPreparation(t *testing.T) {
for _, scenario := range []string{"operator-empty", "operator-space", "operator-control", "operator-large", "action", "id", "nil-resolver", "nil-service", "canceled"} {
t.Run(scenario, func(t *testing.T) {
server, memory, record := recoveryPreflightFixture()
journal := &memoryTransactionJournal{pending: []TransactionRecord{record}}
factory := &inspectionFactory{client: &inspectionClient{memoryClient: memory}}
service := NewServiceWithJournal(factory, journal)
resolve := func(context.Context, string) (Server, error) { return server, nil }
operator, id, action := "unix:0", record.ID, RecoveryActionRecord
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
switch scenario {
case "operator-empty":
operator = ""
case "operator-space":
operator = " unix:0"
case "operator-control":
operator = "unix:0\n"
case "operator-large":
operator = strings.Repeat("x", 161)
case "action":
action = "unknown"
case "id":
id = "../bad"
case "nil-resolver":
resolve = nil
case "nil-service":
service = nil
case "canceled":
cancel()
}
review, err := service.PrepareRecoveryReview(ctx, id, operator, action, resolve)
if err == nil || review.Summary().TransactionID != "" || factory.calls != 0 {
t.Fatal("invalid preparation reached remote state or returned a review")
}
})
}
}

View File

@ -12,6 +12,17 @@
- 不允许删除无归属、归属不符的锁,不允许覆盖与新旧摘要均不匹配的目标文件;出现不一致即保留现场。
- 远端文件动作不能与数据库组成原子事务。必须先持久化处置意图与审计,再执行有界动作;最终状态与完成审计原子落库。进程崩溃必须留下可识别的未完成处置,而不是假成功。
## 2026年10月3日19:07 准备与重新核验实现任务
修改前检查点 `6b77730` 已推送。当前先实现 R3-B 的只读绑定基础,不把它当作执行授权或完整处置功能。
- [~] 绑定受信任操作者、明确动作、事务记录、服务器配置和完整现场采样;公开摘要不携带历史证据、凭据或锁归属令牌。
- [ ] 一分钟有效期,准备及重新核验共用三十秒操作上限;取消、过期、配置变化、事务变化和现场变化均拒绝。
- [ ] 缺可信历史证据的事务只允许准备记录核查结论,不允许准备提交或回滚,不改变历史状态或现场。
- [ ] 完成单元及竞态验证;再接入一次性本机确认凭据、数据库处置占用和原子审计。
准备对象及重新核验结果只是执行器的前置输入,不代表已经写入处置意图、取得互斥、完成动作资格核验或获得远端写入授权;在持久化和完整执行链路接入前,不对用户开放写操作入口。
## 需求和任务清单
- [~] R3-A:确定动作模型、独立审计以及失败边界。本设计已落文档,代码尚未开始。