From 671b1e9ef6cba3c7b859ae8e0373fd2026eb5336 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hans-J=C3=BCrgen=20Sch=C3=B6nig?= Date: Sat, 18 Jul 2026 21:06:07 +0300 Subject: [PATCH] fix: harden repository mutations and CLI safety --- internal/agent/restore_executor.go | 6 + internal/backup/gc_commit_fence_test.go | 80 ++++++ internal/backup/manifest_store.go | 145 +++++++++- internal/backup/post_publish_failure_test.go | 72 +++++ internal/backup/rotate.go | 28 ++ internal/backup/runner/dek_reuse.go | 30 +- internal/backup/runner/runner.go | 11 + internal/capacity/preflight.go | 4 + internal/capacity/preflight_test.go | 13 + internal/cli/anomaly.go | 4 +- internal/cli/backup.go | 5 + internal/cli/backup_controlplane.go | 33 ++- internal/cli/backup_controlplane_test.go | 35 +++ internal/cli/capacity_preflight.go | 4 + internal/cli/cost.go | 4 + internal/cli/dispatch.go | 15 + internal/cli/forecast.go | 4 +- internal/cli/insider.go | 7 + internal/cli/integrity.go | 46 +-- internal/cli/numeric.go | 7 + internal/cli/rare_flags_regression_test.go | 98 +++++++ internal/cli/rare_options_contract_test.go | 26 ++ internal/cli/repair.go | 24 +- internal/cli/repo_bundle.go | 3 + internal/cli/repo_gc.go | 8 + internal/cli/repo_replicate.go | 4 +- internal/cli/restore.go | 13 +- internal/cli/restore_controlplane.go | 13 + internal/cli/restore_controlplane_test.go | 69 +++++ internal/cli/rotate.go | 9 + internal/cli/rotate_test.go | 14 + internal/cli/server.go | 5 +- internal/cli/threshold.go | 6 + internal/cli/wal.go | 29 +- internal/cli/wal_encrypt_test.go | 33 +-- internal/pg/walsink/push.go | 8 + internal/pg/walsink/walsink.go | 17 ++ internal/pg/walsink/worm_propagation_test.go | 9 + internal/repo/bundle/bundle.go | 33 ++- internal/repo/bundle/bundle_caps_test.go | 6 +- internal/repo/bundle/bundle_test.go | 4 +- internal/repo/bundle/review79_bundle_test.go | 8 +- internal/repo/mutationlock.go | 140 ++++++++++ internal/repo/replicate.go | 10 + internal/repo/sharedkey/sharedkey.go | 280 ++++++++++++++++++- internal/repo/sharedkey/sharedkey_test.go | 52 ++++ internal/server/server.go | 4 +- 47 files changed, 1331 insertions(+), 147 deletions(-) create mode 100644 internal/backup/gc_commit_fence_test.go create mode 100644 internal/backup/post_publish_failure_test.go create mode 100644 internal/cli/numeric.go create mode 100644 internal/cli/rare_flags_regression_test.go create mode 100644 internal/cli/rare_options_contract_test.go create mode 100644 internal/repo/mutationlock.go diff --git a/internal/agent/restore_executor.go b/internal/agent/restore_executor.go index e893930..d140404 100644 --- a/internal/agent/restore_executor.go +++ b/internal/agent/restore_executor.go @@ -17,6 +17,7 @@ import ( "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" "github.com/cybertec-postgresql/pg_hardstorage/internal/restore" "github.com/cybertec-postgresql/pg_hardstorage/internal/restore/naturaltime" + "github.com/cybertec-postgresql/pg_hardstorage/internal/restore/postverify" "github.com/cybertec-postgresql/pg_hardstorage/internal/restore/walfetchcmd" ) @@ -164,6 +165,10 @@ func (e *RestoreExecutor) Execute(ctx context.Context, job *ControlPlaneJob, pro } allowOverwrite, _ := job.Args["allow_overwrite"].(bool) + verifyRestoreMode, _ := job.Args["verify_restore"].(string) + if _, err := postverify.ParseMode(verifyRestoreMode); err != nil { + return nil, fmt.Errorf("restore-executor: verify_restore: %w", err) + } rec, err := buildRecoveryFromArgs(job.Args, targetTime) if err != nil { @@ -220,6 +225,7 @@ func (e *RestoreExecutor) Execute(ctx context.Context, job *ControlPlaneJob, pro TargetDir: targetDir, Verifier: e.verifier, AllowOverwrite: allowOverwrite, + VerifyMode: verifyRestoreMode, Recovery: rec, TablespaceRemap: tsRemap, KEKForRef: kekFor, diff --git a/internal/backup/gc_commit_fence_test.go b/internal/backup/gc_commit_fence_test.go new file mode 100644 index 0000000..929c987 --- /dev/null +++ b/internal/backup/gc_commit_fence_test.go @@ -0,0 +1,80 @@ +package backup_test + +import ( + "bytes" + "context" + "errors" + "testing" + + "github.com/cybertec-postgresql/pg_hardstorage/internal/backup" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" + "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" +) + +// A GC snapshot can select an old orphan immediately before a backup reuses +// it. The mutation fence must prevent publication during sweep, and the final +// existence check must prevent publication after GC removed the chunk. +func TestCommit_GCFencePreventsStaleSnapshotCorruption(t *testing.T) { + store, sp, signer, _ := newStore(t) + ctx := context.Background() + body := []byte("old orphan reused by a new backup") + h := repo.HashOf(body) + if _, err := sp.Put(ctx, repo.ChunkKey(h), bytes.NewReader(body), storage.PutOptions{IfNotExists: true}); err != nil { + t.Fatal(err) + } + + m := sampleManifest() + m.Files = []backup.FileEntry{{ + Path: "base/1/1", Size: int64(len(body)), + Chunks: []backup.ChunkRef{{Hash: h, Offset: 0, Len: int64(len(body))}}, + }} + + gcLock, err := repo.AcquireMutationLock(ctx, sp, "test GC sweep") + if err != nil { + t.Fatal(err) + } + if err := store.Commit(ctx, m, signer, backup.CommitOptions{RequireChunksPresent: true}); !errors.Is(err, repo.ErrMutationLocked) { + t.Fatalf("commit during GC = %v, want ErrMutationLocked", err) + } + if err := sp.Delete(ctx, repo.ChunkKey(h)); err != nil { + t.Fatal(err) + } + if err := gcLock.Release(ctx); err != nil { + t.Fatal(err) + } + + if err := store.Commit(ctx, m, signer, backup.CommitOptions{RequireChunksPresent: true}); err == nil { + t.Fatal("commit after GC deleted a referenced chunk must fail") + } + if _, err := sp.Stat(ctx, backup.PrimaryPath(m.Deployment, m.BackupID)); !errors.Is(err, storage.ErrNotFound) { + t.Fatalf("primary manifest should not exist after fenced failure; stat=%v", err) + } +} + +func TestUndelete_GCFencePreventsReferenceResurrectionDuringSweep(t *testing.T) { + store, sp, signer, _ := newStore(t) + ctx := context.Background() + m := sampleManifest() + if err := store.Commit(ctx, m, signer, backup.CommitOptions{}); err != nil { + t.Fatal(err) + } + if err := store.SoftDelete(ctx, m.Deployment, m.BackupID, "test", "gc race"); err != nil { + t.Fatal(err) + } + + gcLock, err := repo.AcquireMutationLock(ctx, sp, "test GC sweep") + if err != nil { + t.Fatal(err) + } + defer func() { _ = gcLock.Release(context.Background()) }() + + if _, err := store.Undelete(ctx, m.Deployment, m.BackupID); !errors.Is(err, repo.ErrMutationLocked) { + t.Fatalf("undelete during GC = %v, want ErrMutationLocked", err) + } + if _, err := store.UndeleteForce(ctx, m.Deployment, m.BackupID); !errors.Is(err, repo.ErrMutationLocked) { + t.Fatalf("force undelete during GC = %v, want ErrMutationLocked", err) + } + if dead, err := store.IsTombstoned(ctx, m.Deployment, m.BackupID); err != nil || !dead { + t.Fatalf("tombstone changed while GC held fence: dead=%v err=%v", dead, err) + } +} diff --git a/internal/backup/manifest_store.go b/internal/backup/manifest_store.go index 2ab946f..bb98f31 100644 --- a/internal/backup/manifest_store.go +++ b/internal/backup/manifest_store.go @@ -18,6 +18,7 @@ import ( "time" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" + "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" ) // ManifestStore writes and reads signed manifests against any StoragePlugin. @@ -178,6 +179,12 @@ type CommitOptions struct { // Governance) when RetainUntil is set. Empty implies // Compliance (regulatory-grade default). RetentionMode storage.WORMMode + + // RequireChunksPresent performs a final fenced existence check before + // publishing the manifest. Production backup writers enable it; tests and + // forensic importers that intentionally construct manifest-only fixtures + // may leave it false. + RequireChunksPresent bool } // Commit signs m (if not already signed), writes it atomically to the @@ -228,9 +235,29 @@ func (ms *ManifestStore) Commit(ctx context.Context, m *Manifest, signer *Signer if err != nil { return fmt.Errorf("backup: marshal manifest %s: %w", m.BackupID, err) } - primaryKey := PrimaryPath(m.Deployment, m.BackupID) - if err := ms.commitAtomic(ctx, primaryKey, body, opts); err != nil { + purpose := "backup manifest " + m.Deployment + "/" + m.BackupID + lock, err := ms.acquireCommitMutationLock(ctx, purpose) + if err != nil { + if errors.Is(err, ErrAlreadyCommitted) { + return ErrAlreadyCommitted + } + return fmt.Errorf("backup: commit mutation lock: %w", err) + } + defer func() { _ = lock.Release(context.Background()) }() + + // GC can run while this backup uploads/deduplicates chunks. Holding the + // mutation lock prevents a new GC snapshot, and this final existence gate + // catches chunks a preceding GC deleted before we acquired the lock. + if opts.RequireChunksPresent { + if check, err := CheckChunkExistence(ctx, ms.sp, m); err != nil { + return fmt.Errorf("backup: pre-commit chunk check: %w", err) + } else if !check.AllPresent() { + return fmt.Errorf("backup: refusing to commit %s: %d referenced chunks are missing after repository mutation fencing", m.BackupID, len(check.Missing)) + } + } + + if err := ms.commitAtomic(ctx, primaryKey, body); err != nil { if errors.Is(err, storage.ErrAlreadyExists) { return ErrAlreadyCommitted } @@ -262,6 +289,10 @@ func (ms *ManifestStore) Commit(ctx context.Context, m *Manifest, signer *Signer }) } } + if err := ms.applyRetention(ctx, primaryKey, opts); err != nil { + return ms.rollbackPrimaryCommit(ctx, primaryKey, + fmt.Errorf("backup: commit primary %q: %w", primaryKey, err)) + } // Record the deployment in the index so Deployments() can enumerate // the fleet without scanning every manifest object. Best-effort and @@ -300,6 +331,42 @@ func (ms *ManifestStore) Commit(ctx context.Context, m *Manifest, signer *Signer return nil } +const ( + commitLockRetryInterval = 10 * time.Millisecond + commitLockMaxWait = 5 * time.Second +) + +// acquireCommitMutationLock preserves Commit's documented duplicate-writer +// semantics while the repository-wide GC fence is held. A competing commit +// for this exact primary is allowed a bounded wait to publish; unrelated +// mutations still fail immediately, and abandoned locks remain fail-closed. +func (ms *ManifestStore) acquireCommitMutationLock(ctx context.Context, purpose string) (*repo.MutationLock, error) { + timer := time.NewTimer(commitLockMaxWait) + defer timer.Stop() + + for { + lock, err := repo.AcquireMutationLock(ctx, ms.sp, purpose) + if err == nil { + return lock, nil + } + var held *repo.MutationLockedError + if !errors.As(err, &held) || held.Purpose != purpose { + return nil, err + } + + retry := time.NewTimer(commitLockRetryInterval) + select { + case <-ctx.Done(): + retry.Stop() + return nil, ctx.Err() + case <-timer.C: + retry.Stop() + return nil, err + case <-retry.C: + } + } +} + // rollbackPrimaryCommit removes the just-written primary manifest after // a post-write chain check failed, returning cause. If the rollback // Delete ITSELF fails, the manifest is left orphaned at its primary key @@ -339,7 +406,7 @@ func (ms *ManifestStore) rollbackPrimaryCommit(ctx context.Context, primaryKey s // stays an ordinary, deletable staging object (reaped by the rename's // source-delete on success, or by `repo gc`'s stale-staging sweep if a // crash orphaned it). -func (ms *ManifestStore) commitAtomic(ctx context.Context, key string, body []byte, opts CommitOptions) error { +func (ms *ManifestStore) commitAtomic(ctx context.Context, key string, body []byte) error { tmp := key + ".tmp." + randSuffix() _, err := ms.sp.Put(ctx, tmp, bytes.NewReader(body), storage.PutOptions{ ContentLength: int64(len(body)), @@ -348,16 +415,42 @@ func (ms *ManifestStore) commitAtomic(ctx context.Context, key string, body []by return fmt.Errorf("write tmp: %w", err) } if err := ms.sp.RenameIfNotExists(ctx, tmp, key); err != nil { - // Best-effort cleanup of the tmp; never propagate a tmp-cleanup - // failure (the primary error is what matters). + if errors.Is(err, storage.ErrAlreadyExists) { + _ = ms.sp.Delete(ctx, tmp) + return err + } + // Copy-based backends can publish dst successfully and then report + // an error while deleting src. Confirm the canonical bytes before + // treating that ambiguous outcome as a failed publication; otherwise + // Commit would return before its parent-liveness checks even though a + // visible authoritative manifest already existed. + if same, _ := ms.keyEquals(ctx, key, body); same { + _ = ms.sp.Delete(ctx, tmp) + return nil + } _ = ms.sp.Delete(ctx, tmp) return err } - // Apply WORM retention to the committed manifest itself. A backend - // without WORM returns ErrUnsupported, which is expected and - // ignored ("backends without WORM ignore this"); any other failure - // means the operator-requested lock did NOT apply, which must - // surface. + return nil +} + +func (ms *ManifestStore) keyEquals(ctx context.Context, key string, want []byte) (bool, error) { + rc, err := ms.sp.Get(ctx, key) + if err != nil { + return false, err + } + defer rc.Close() + got, err := stdio.ReadAll(stdio.LimitReader(rc, int64(len(want))+1)) + if err != nil { + return false, err + } + return bytes.Equal(got, want), nil +} + +// applyRetention runs only after the primary's chain-safety checks. If it +// fails, Commit rolls the primary back, so a returned error cannot leave an +// unlocked authoritative manifest behind unnoticed. +func (ms *ManifestStore) applyRetention(ctx context.Context, key string, opts CommitOptions) error { if !opts.RetainUntil.IsZero() { mode := opts.RetentionMode if mode == "" { @@ -401,9 +494,13 @@ func (ms *ManifestStore) commitReplica(ctx context.Context, replicaKey string, b case <-time.After(replicaCommitBackoff * time.Duration(attempt)): } } - err := ms.commitAtomic(ctx, replicaKey, body, opts) + err := ms.commitAtomic(ctx, replicaKey, body) if err == nil || errors.Is(err, storage.ErrAlreadyExists) { - return nil + if rerr := ms.applyRetention(ctx, replicaKey, opts); rerr == nil { + return nil + } else { + err = rerr + } } lastErr = err } @@ -1522,6 +1619,17 @@ func (ms *ManifestStore) Undelete(ctx context.Context, deployment, backupID stri // no restorability check needed. return false, nil } + lock, err := repo.AcquireMutationLock(ctx, ms.sp, "backup undelete "+deployment+"/"+backupID) + if err != nil { + return false, fmt.Errorf("backup: Undelete: mutation lock: %w", err) + } + defer func() { _ = lock.Release(context.Background()) }() + // The marker may have been removed while we waited for the fence. + if dead, err = ms.IsTombstoned(ctx, deployment, backupID); err != nil { + return false, fmt.Errorf("backup: Undelete: tombstone re-check: %w", err) + } else if !dead { + return false, nil + } // Read the still-tombstoned body without signature verification // (we only need Files/Chunks, mirroring findLiveDescendants) and // confirm every referenced chunk is still addressable in the repo. @@ -1556,6 +1664,9 @@ func (ms *ManifestStore) Undelete(ctx context.Context, deployment, backupID stri // chunks). The resurrected backup may be un-restorable — that is the // caller's explicit, eyes-open choice. Prefer Undelete. func (ms *ManifestStore) UndeleteForce(ctx context.Context, deployment, backupID string) (bool, error) { + if err := validateRef(deployment, backupID); err != nil { + return false, err + } if deployment == "" || backupID == "" { return false, errors.New("backup: UndeleteForce requires deployment and backupID") } @@ -1566,6 +1677,16 @@ func (ms *ManifestStore) UndeleteForce(ctx context.Context, deployment, backupID if !dead { return false, nil } + lock, err := repo.AcquireMutationLock(ctx, ms.sp, "backup force-undelete "+deployment+"/"+backupID) + if err != nil { + return false, fmt.Errorf("backup: UndeleteForce: mutation lock: %w", err) + } + defer func() { _ = lock.Release(context.Background()) }() + if dead, err = ms.IsTombstoned(ctx, deployment, backupID); err != nil { + return false, fmt.Errorf("backup: UndeleteForce: tombstone re-check: %w", err) + } else if !dead { + return false, nil + } return ms.removeTombstone(ctx, deployment, backupID) } diff --git a/internal/backup/post_publish_failure_test.go b/internal/backup/post_publish_failure_test.go new file mode 100644 index 0000000..c3fedc5 --- /dev/null +++ b/internal/backup/post_publish_failure_test.go @@ -0,0 +1,72 @@ +package backup_test + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/cybertec-postgresql/pg_hardstorage/internal/backup" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" +) + +type renameLandedErrorSP struct{ storage.StoragePlugin } + +func (s *renameLandedErrorSP) RenameIfNotExists(ctx context.Context, src, dst string) error { + if err := s.StoragePlugin.RenameIfNotExists(ctx, src, dst); err != nil { + return err + } + if strings.HasSuffix(dst, "/manifest.json") { + return errors.New("source cleanup failed after destination landed") + } + return nil +} + +func TestCommit_AmbiguousRenameStillRunsParentSafety(t *testing.T) { + _, raw, signer, _ := newStore(t) + sp := &renameLandedErrorSP{StoragePlugin: raw} + store := backup.NewManifestStore(sp) + m := sampleManifest() + m.Type = backup.BackupTypeIncremental + m.ParentBackupID = "missing-parent" + + err := store.Commit(context.Background(), m, signer, backup.CommitOptions{}) + if err == nil { + t.Fatal("orphaned incremental must be rejected even when rename reports a post-publish error") + } + if _, statErr := raw.Stat(context.Background(), backup.PrimaryPath(m.Deployment, m.BackupID)); !errors.Is(statErr, storage.ErrNotFound) { + t.Fatalf("orphaned primary remained visible: %v", statErr) + } +} + +func TestCommit_ExistingIdenticalManifestStillReturnsAlreadyCommitted(t *testing.T) { + store, _, signer, _ := newStore(t) + if err := store.Commit(context.Background(), sampleManifest(), signer, backup.CommitOptions{}); err != nil { + t.Fatal(err) + } + if err := store.Commit(context.Background(), sampleManifest(), signer, backup.CommitOptions{}); !errors.Is(err, backup.ErrAlreadyCommitted) { + t.Fatalf("second commit = %v, want ErrAlreadyCommitted", err) + } +} + +type retentionFailSP struct{ storage.StoragePlugin } + +func (s *retentionFailSP) SetRetention(context.Context, string, time.Time, storage.WORMMode) error { + return errors.New("retention service unavailable") +} + +func TestCommit_RetentionFailureRollsBackPrimary(t *testing.T) { + _, raw, signer, _ := newStore(t) + store := backup.NewManifestStore(&retentionFailSP{StoragePlugin: raw}) + m := sampleManifest() + err := store.Commit(context.Background(), m, signer, backup.CommitOptions{ + RetainUntil: time.Now().Add(time.Hour), RetentionMode: storage.WORMCompliance, + }) + if err == nil { + t.Fatal("retention failure must fail commit") + } + if _, statErr := raw.Stat(context.Background(), backup.PrimaryPath(m.Deployment, m.BackupID)); !errors.Is(statErr, storage.ErrNotFound) { + t.Fatalf("unlocked primary remained visible: %v", statErr) + } +} diff --git a/internal/backup/rotate.go b/internal/backup/rotate.go index cdbb9d3..9812a23 100644 --- a/internal/backup/rotate.go +++ b/internal/backup/rotate.go @@ -13,6 +13,8 @@ import ( "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/encryption" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" + "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" + "github.com/cybertec-postgresql/pg_hardstorage/internal/repo/sharedkey" ) // RotateKEKSchema is the on-disk version tag for RotateKEKResult bodies. @@ -176,6 +178,16 @@ func RotateKEK(ctx context.Context, sp storage.StoragePlugin, opts RotateKEKOpti res.StoppedAt = time.Now().UTC() res.DurationMS = res.StoppedAt.Sub(res.StartedAt).Milliseconds() } + var mutationLock *repo.MutationLock + if !opts.DryRun { + var lockErr error + mutationLock, lockErr = repo.AcquireMutationLock(ctx, sp, "KEK rotation") + if lockErr != nil { + finish() + return res, fmt.Errorf("backup rotate-kek: mutation lock: %w", lockErr) + } + defer func() { _ = mutationLock.Release(context.Background()) }() + } store := NewManifestStore(sp) deployments, err := store.Deployments(ctx) @@ -244,6 +256,22 @@ func RotateKEK(ctx context.Context, sp storage.StoragePlugin, opts RotateKEKOpti } } } + if !opts.DryRun && res.Failed == 0 { + unwrapOld := func(wrapped []byte) ([]byte, error) { + dek, err := encryption.Unwrap(opts.OldKEK, wrapped) + if err != nil { + return nil, err + } + return dek[:], nil + } + wrapNew := func(dek [encryption.KeyLen]byte) ([]byte, error) { + return encryption.Wrap(opts.NewKEK, dek) + } + if err := sharedkey.RotateDomain(ctx, sp, opts.OldKEKRef, opts.NewKEKRef, unwrapOld, wrapNew); err != nil { + finish() + return res, fmt.Errorf("backup rotate-kek: rotate repository encryption domain: %w", err) + } + } finish() return res, nil diff --git a/internal/backup/runner/dek_reuse.go b/internal/backup/runner/dek_reuse.go index 94a9550..e553527 100644 --- a/internal/backup/runner/dek_reuse.go +++ b/internal/backup/runner/dek_reuse.go @@ -33,32 +33,20 @@ import ( // backup: generating a fresh DEK there yields a backup whose deduped chunks // are unrestorable, discovered only at restore time (issue #28). func selectDEK(ctx context.Context, sp storage.StoragePlugin, kekRef string, cfg *EncryptionConfig) ([encryption.KeyLen]byte, error) { - var dek [encryption.KeyLen]byte - - res, err := sharedkey.Resolve(ctx, sp, kekRef, func(wrapped []byte) ([]byte, error) { + unwrap := func(wrapped []byte) ([]byte, error) { return unwrapDEKForReuse(ctx, cfg, wrapped) - }) - if err != nil { - return dek, fmt.Errorf("backup: cannot determine whether a DEK already exists for KEK %q; refusing a fresh DEK that the CAS's plaintext-hash dedup would leave unrestorable against existing chunks: %w", kekRef, err) } - if res.DEK != nil { - if len(res.DEK) != encryption.KeyLen { - return dek, fmt.Errorf("backup: reused DEK for KEK %q has wrong length %d (want %d)", kekRef, len(res.DEK), encryption.KeyLen) + wrap := func(dek [encryption.KeyLen]byte) ([]byte, error) { + if cfg.Provider != nil { + return cfg.Provider.WrapDEK(ctx, dek[:]) } - copy(dek[:], res.DEK) - return dek, nil + return encryption.Wrap(cfg.KEK, dek) } - if res.SawCandidate { - return dek, fmt.Errorf("backup: a prior DEK for KEK %q exists but none of its recorded wrapped form(s) could be unwrapped for reuse; refusing a fresh DEK that would leave deduped chunks unrestorable (verify the KEK material matches this ref)", kekRef) - } - - // Positively no prior DEK for this KEK in any backup or WAL manifest — - // the first encrypted artifact in this repo. A fresh DEK is correct. - fresh, gerr := encryption.GenerateDEK() - if gerr != nil { - return dek, fmt.Errorf("backup: generate DEK: %w", gerr) + dek, err := sharedkey.ResolveOrCreate(ctx, sp, kekRef, unwrap, wrap) + if err != nil { + return [encryption.KeyLen]byte{}, fmt.Errorf("backup: establish repository encryption domain for KEK %q: %w", kekRef, err) } - return fresh, nil + return dek, nil } // looksLikePrimaryManifest returns true for keys of the shape diff --git a/internal/backup/runner/runner.go b/internal/backup/runner/runner.go index 4c41599..d47d857 100644 --- a/internal/backup/runner/runner.go +++ b/internal/backup/runner/runner.go @@ -37,6 +37,7 @@ import ( "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo/casdefault" + "github.com/cybertec-postgresql/pg_hardstorage/internal/repo/sharedkey" "github.com/cybertec-postgresql/pg_hardstorage/internal/wal/gapstate" ) @@ -477,6 +478,15 @@ func Take(ctx context.Context, opts TakeOptions) (*Result, error) { // fleet-wide audit story coherent (every byte from this backup // retired at the same moment). wormNow := time.Now().UTC() + if opts.Encryption == nil { + // The CAS namespace is keyed by plaintext hash. Once ciphertext has + // won a key, a plaintext writer cannot safely deduplicate against it + // because its manifest carries no DEK. Atomically claim/check the + // repository's plaintext domain before writing any chunks. + if err := sharedkey.EnsurePlaintextAllowed(ctx, sp); err != nil { + return nil, fmt.Errorf("backup: encryption posture: %w", err) + } + } // Dedup hints: seed the CAS with the chunk hashes from this // deployment's most recent prior backup. A re-backup of a mostly- @@ -787,6 +797,7 @@ func Take(ctx context.Context, opts TakeOptions) (*Result, error) { } commitOpts := backup.CommitOptions{ + RequireChunksPresent: true, OnReplicaError: func(rerr error) { // Primary is committed; redundancy copy failed. Surface as // a warning so the operator knows the cross-prefix-survival diff --git a/internal/capacity/preflight.go b/internal/capacity/preflight.go index 3800761..e0461d8 100644 --- a/internal/capacity/preflight.go +++ b/internal/capacity/preflight.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "math" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" ) @@ -91,6 +92,9 @@ func Preflight(ctx context.Context, sp storage.StoragePlugin, opts PreflightOpti return nil, errors.New("capacity: Preflight requires ProjectedBytes > 0") } safety := opts.SafetyFactor + if math.IsNaN(safety) || math.IsInf(safety, 0) { + return nil, errors.New("capacity: Preflight requires a finite SafetyFactor") + } if safety <= 0 { safety = DefaultSafetyFactor } diff --git a/internal/capacity/preflight_test.go b/internal/capacity/preflight_test.go index 038e5ce..1677a3e 100644 --- a/internal/capacity/preflight_test.go +++ b/internal/capacity/preflight_test.go @@ -3,6 +3,7 @@ package capacity_test import ( "context" "errors" + "math" "testing" "github.com/cybertec-postgresql/pg_hardstorage/internal/capacity" @@ -181,3 +182,15 @@ func TestPreflight_CustomSafetyFactor(t *testing.T) { t.Errorf("RequiredBytes = %d, want 150", res.RequiredBytes) } } + +func TestPreflightRejectsNonFiniteSafetyFactor(t *testing.T) { + sp := &fakeFreeSpaceSP{info: storage.FreeSpaceInfo{AvailableBytes: 1 << 20}} + for _, factor := range []float64{math.NaN(), math.Inf(1), math.Inf(-1)} { + if _, err := capacity.Preflight(context.Background(), sp, capacity.PreflightOptions{ + ProjectedBytes: 100, + SafetyFactor: factor, + }); err == nil { + t.Fatalf("non-finite SafetyFactor %v was accepted", factor) + } + } +} diff --git a/internal/cli/anomaly.go b/internal/cli/anomaly.go index 7d61ac7..d8f9ce4 100644 --- a/internal/cli/anomaly.go +++ b/internal/cli/anomaly.go @@ -104,9 +104,9 @@ func newAnomalyCheckCmd() *cobra.Command { func runAnomalyCheck(cmd *cobra.Command, deployment, repoURL string, threshold float64, window, minSamples int, all bool) error { d := DispatcherFrom(cmd) - if threshold <= 0 { + if !finiteFloat(threshold) || threshold <= 0 { return output.NewError("usage.bad_flag", - fmt.Sprintf("anomaly check: --threshold must be > 0; got %v", threshold)).Wrap(output.ErrUsage) + fmt.Sprintf("anomaly check: --threshold must be a finite value > 0; got %v", threshold)).Wrap(output.ErrUsage) } if minSamples < 2 { return output.NewError("usage.bad_flag", diff --git a/internal/cli/backup.go b/internal/cli/backup.go index fd4d50b..5671cb9 100644 --- a/internal/cli/backup.go +++ b/internal/cli/backup.go @@ -169,6 +169,11 @@ func runBackup(cmd *cobra.Command, opts runOptions) error { if opts.dispatch.controlPlane != "" { return runBackupControlPlane(cmd, opts) } + if !finiteFloat(opts.capacitySafetyFactor) || opts.capacitySafetyFactor <= 0 { + return output.NewError("usage.bad_flag", + fmt.Sprintf("backup: --capacity-safety-factor must be a finite value > 0; got %v", opts.capacitySafetyFactor)). + Wrap(output.ErrUsage) + } // `backup ` resolves --pg-connection / --repo from the // named deployment in pg_hardstorage.yaml when the operator didn't diff --git a/internal/cli/backup_controlplane.go b/internal/cli/backup_controlplane.go index 17545c6..bda6f92 100644 --- a/internal/cli/backup_controlplane.go +++ b/internal/cli/backup_controlplane.go @@ -23,6 +23,22 @@ import ( // wherever they are. func runBackupControlPlane(cmd *cobra.Command, opts runOptions) error { d := DispatcherFrom(cmd) + if err := rejectChangedDispatchFlags(cmd, "backup", + "pg-connection", + "tenant", + "include-wal", + "encrypt", + "no-encrypt", + "kek", + "kms-config", + "incremental-from", + "ignore-capacity", + "capacity-safety-factor", + "allow-concurrent", + "verbose", + ); err != nil { + return err + } cli, err := newDispatchClient(&opts.dispatch) if err != nil { @@ -47,21 +63,8 @@ func runBackupControlPlane(cmd *cobra.Command, opts runOptions) error { if opts.repoURL != "" { body["repo"] = opts.repoURL } - // tenant + encrypt/no-encrypt are deployment-config concerns on - // the agent side (the agent's local config picks the tenant; the - // keyring picks the encryption posture). Passing them through the - // API is a+ enhancement once we extend EnqueueOptions; for - // now we surface a clear refusal so the operator isn't surprised - // when their flag is silently ignored. - if opts.tenant != "" && opts.tenant != "default" { - return output.NewError("usage.unsupported_flag", - "backup --control-plane: --tenant is set by the agent's local config, not the CLI; remove the flag or take the backup locally"). - Wrap(output.ErrUsage) - } - if opts.encrypt || opts.noEncrypt { - return output.NewError("usage.unsupported_flag", - "backup --control-plane: encryption posture is set by the agent's keyring, not --encrypt/--no-encrypt; remove the flag or take the backup locally"). - Wrap(output.ErrUsage) + if opts.stallTimeout != 0 { + body["inactivity_timeout"] = opts.stallTimeout.String() } id, err := cli.EnqueueBackup(cmd.Context(), opts.deployment, body) diff --git a/internal/cli/backup_controlplane_test.go b/internal/cli/backup_controlplane_test.go index 269b1a7..b757c98 100644 --- a/internal/cli/backup_controlplane_test.go +++ b/internal/cli/backup_controlplane_test.go @@ -149,6 +149,7 @@ func TestBackup_ControlPlane_FastFlowsThroughArgs(t *testing.T) { "backup", "db1", "--fast", "--label", "before-migration", + "--stall-timeout", "45s", "-o", "json", } onJob := func(s *server.Server, jobID string) { @@ -165,6 +166,9 @@ func TestBackup_ControlPlane_FastFlowsThroughArgs(t *testing.T) { if j.Args["label"] != "before-migration" { t.Errorf("Args.label = %v", j.Args["label"]) } + if j.Args["inactivity_timeout"] != "45s" { + t.Errorf("Args.inactivity_timeout = %v, want 45s", j.Args["inactivity_timeout"]) + } // Then drive the job to completion so the test exits. if _, err := s.Jobs().Claim(server.ClaimOptions{ AgentID: "a", @@ -185,3 +189,34 @@ func TestBackup_ControlPlane_FastFlowsThroughArgs(t *testing.T) { t.Fatalf("exit=%d\nstdout=%s\nstderr=%s", exit, stdout, stderr) } } + +func TestBackup_ControlPlane_RefusesUnrepresentableFlags(t *testing.T) { + tests := [][]string{ + {"--pg-connection", "postgres://ignored"}, + {"--include-wal"}, + {"--kek", "local:archive"}, + {"--kms-config", "region=eu-central-1"}, + {"--incremental-from", "parent-id"}, + {"--ignore-capacity"}, + {"--capacity-safety-factor", "1.5"}, + {"--allow-concurrent"}, + {"--verbose"}, + } + for _, extra := range tests { + t.Run(strings.Join(extra, "_"), func(t *testing.T) { + root := cli.NewRoot() + var out, errb bytes.Buffer + root.SetOut(&out) + root.SetErr(&errb) + args := []string{"backup", "db1", "--control-plane", "http://does-not-matter:8443", "-o", "json"} + root.SetArgs(append(args, extra...)) + exit := cli.Run(root) + if exit != int(output.ExitMisuse) { + t.Fatalf("exit=%d, want %d; stderr=%s", exit, output.ExitMisuse, errb.String()) + } + if !strings.Contains(errb.String(), "usage.unsupported_flag") { + t.Fatalf("missing unsupported-flag error: %s", errb.String()) + } + }) + } +} diff --git a/internal/cli/capacity_preflight.go b/internal/cli/capacity_preflight.go index 2c82d48..969ee98 100644 --- a/internal/cli/capacity_preflight.go +++ b/internal/cli/capacity_preflight.go @@ -106,6 +106,10 @@ type capacityPreflightOptions struct { func runCapacityPreflight(cmd *cobra.Command, opts capacityPreflightOptions) error { d := DispatcherFrom(cmd) + if !finiteFloat(opts.safetyFactor) || opts.safetyFactor <= 0 { + return output.NewError("usage.bad_flag", + fmt.Sprintf("capacity preflight: --safety-factor must be a finite value > 0; got %v", opts.safetyFactor)).Wrap(output.ErrUsage) + } // Positional-or-flag: guard the resolved value, not the flag. if opts.repoURL == "" { return missingFlagErr(cmd, "--repo (or a positional )") diff --git a/internal/cli/cost.go b/internal/cli/cost.go index b5b76b0..934a9ea 100644 --- a/internal/cli/cost.go +++ b/internal/cli/cost.go @@ -75,6 +75,10 @@ type costReportOptions struct { func runCostReport(cmd *cobra.Command, opts costReportOptions) error { d := DispatcherFrom(cmd) + if !finiteFloat(opts.price) || opts.price < 0 { + return output.NewError("usage.bad_flag", + "cost report: --price-per-gb-month must be finite and >= 0").Wrap(output.ErrUsage) + } _, sp, err := openRepo(cmd.Context(), opts.repoURL) if err != nil { return err diff --git a/internal/cli/dispatch.go b/internal/cli/dispatch.go index 9846e28..546c6c4 100644 --- a/internal/cli/dispatch.go +++ b/internal/cli/dispatch.go @@ -318,6 +318,21 @@ func registerDispatchFlags(c *cobra.Command, f *dispatchAuthFlags) { "poll cadence (seconds) for /v1/jobs/") } +// rejectChangedDispatchFlags prevents a control-plane invocation from +// accepting command flags that the remote job protocol cannot represent. +// Silently dropping one of these flags can make an agent execute different +// work from what the operator requested. +func rejectChangedDispatchFlags(cmd *cobra.Command, operation string, names ...string) error { + for _, name := range names { + if cmd.Flags().Changed(name) { + return output.NewError("usage.unsupported_flag", + fmt.Sprintf("%s --control-plane: --%s is not supported by remote dispatch; remove the flag or run the command locally", operation, name)). + Wrap(output.ErrUsage) + } + } + return nil +} + // newDispatchClient builds a *DispatchClient from the parsed flags. // Returns nil + structured error if the control-plane URL is set but // some required piece (e.g. token file unreadable) can't be loaded. diff --git a/internal/cli/forecast.go b/internal/cli/forecast.go index b1b74a1..6eef742 100644 --- a/internal/cli/forecast.go +++ b/internal/cli/forecast.go @@ -166,9 +166,9 @@ func runForecast(cmd *cobra.Command, f forecastFlags) error { } horizons = append(horizons, dur) } - if f.pricePerGBMonth < 0 { + if !finiteFloat(f.pricePerGBMonth) || f.pricePerGBMonth < 0 { return output.NewError("usage.bad_flag", - "forecast: --price-per-gb-month must be >= 0").Wrap(output.ErrUsage) + "forecast: --price-per-gb-month must be finite and >= 0").Wrap(output.ErrUsage) } switch f.format { case "", "json", "markdown": diff --git a/internal/cli/insider.go b/internal/cli/insider.go index 2df64ba..3bb5351 100644 --- a/internal/cli/insider.go +++ b/internal/cli/insider.go @@ -112,6 +112,10 @@ type insiderScanFlags struct { func runInsiderScan(cmd *cobra.Command, f insiderScanFlags) error { d := DispatcherFrom(cmd) + if !finiteFloat(f.factor) || f.factor <= 0 { + return output.NewError("usage.bad_flag", + fmt.Sprintf("insider scan: --spike-factor must be a finite value > 0; got %v", f.factor)).Wrap(output.ErrUsage) + } failOn, err := parseFailOnSeverity(f.failOn) if err != nil { return output.NewError("usage.bad_flag", @@ -122,6 +126,9 @@ func runInsiderScan(cmd *cobra.Command, f insiderScanFlags) error { return err } defer sp.Close() + if err := assertRepoWritable(cmd.Context(), sp, "insider scan"); err != nil { + return err + } auditStore := audit.NewStoreWithRetention(sp, repoMeta.WORM) det := insider.NewDetector(auditStore) scan, err := det.Run(cmd.Context(), insider.Options{ diff --git a/internal/cli/integrity.go b/internal/cli/integrity.go index f7d76dc..efcd5bc 100644 --- a/internal/cli/integrity.go +++ b/internal/cli/integrity.go @@ -132,6 +132,11 @@ func runIntegrityRun(cmd *cobra.Command, f integrityRunFlags) error { return err } defer sp.Close() + if !f.skipSign { + if err := assertRepoWritable(cmd.Context(), sp, "integrity run"); err != nil { + return err + } + } signer, verifier, err := loadSignerForJIT() // shared keystore loader if err != nil { return err @@ -163,25 +168,28 @@ func runIntegrityRun(cmd *cobra.Command, f integrityRunFlags) error { fmt.Sprintf("integrity run: persist: %v", err)).Wrap(err) } } - // Best-effort audit append (failure is not fatal). - auditStore := audit.NewStoreWithRetention(sp, repoMeta.WORM) - _ = auditStore.Append(cmd.Context(), &audit.Event{ - Action: "integrity.run", - Timestamp: time.Now().UTC(), - Body: map[string]any{ - "run_id": run.ID, - "status": string(run.Status), - "strategy": strategy.Mode, - "deployment": f.deployment, - "manifests_total": run.Manifests.Total, - "signatures_ok": run.Manifests.SignaturesOK, - "signatures_fail": run.Manifests.SignaturesFail, - "chunks_referenced": run.Chunks.DistinctReferenced, - "chunks_missing": run.Chunks.Missing, - "chunks_mismatched": run.Chunks.Mismatched, - "chunks_verified": run.Chunks.Verified, - }, - }) + // Best-effort audit append (failure is not fatal). --skip-sign is a + // genuinely read-only testing mode, so it must not append either. + if !f.skipSign { + auditStore := audit.NewStoreWithRetention(sp, repoMeta.WORM) + _ = auditStore.Append(cmd.Context(), &audit.Event{ + Action: "integrity.run", + Timestamp: time.Now().UTC(), + Body: map[string]any{ + "run_id": run.ID, + "status": string(run.Status), + "strategy": strategy.Mode, + "deployment": f.deployment, + "manifests_total": run.Manifests.Total, + "signatures_ok": run.Manifests.SignaturesOK, + "signatures_fail": run.Manifests.SignaturesFail, + "chunks_referenced": run.Chunks.DistinctReferenced, + "chunks_missing": run.Chunks.Missing, + "chunks_mismatched": run.Chunks.Mismatched, + "chunks_verified": run.Chunks.Verified, + }, + }) + } body := integrityRunBody{Run: run} // Dual-stream: emit body regardless; on issues, return a verify.* // error to flip the exit code. diff --git a/internal/cli/numeric.go b/internal/cli/numeric.go new file mode 100644 index 0000000..1954ea6 --- /dev/null +++ b/internal/cli/numeric.go @@ -0,0 +1,7 @@ +package cli + +import "math" + +func finiteFloat(v float64) bool { + return !math.IsNaN(v) && !math.IsInf(v, 0) +} diff --git a/internal/cli/rare_flags_regression_test.go b/internal/cli/rare_flags_regression_test.go new file mode 100644 index 0000000..df1c430 --- /dev/null +++ b/internal/cli/rare_flags_regression_test.go @@ -0,0 +1,98 @@ +package cli_test + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/cybertec-postgresql/pg_hardstorage/internal/output" + "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" +) + +func TestRareFloatFlagsRejectNonFiniteValues(t *testing.T) { + commands := [][]string{ + {"backup", "db1", "--pg-connection", "postgres://unused", "--repo", "file:///missing", "--capacity-safety-factor"}, + {"capacity", "preflight", "file:///missing", "--projected-bytes", "1", "--safety-factor"}, + {"repo", "replicate", "--from", "file:///from", "--to", "file:///to", "--max-mbps"}, + {"anomaly", "check", "db1", "--repo", "file:///missing", "--threshold"}, + {"insider", "scan", "--repo", "file:///missing", "--spike-factor"}, + {"forecast", "file:///missing", "--price-per-gb-month"}, + {"cost", "report", "--repo", "file:///missing", "--price-per-gb-month"}, + } + for _, base := range commands { + for _, value := range []string{"NaN", "+Inf", "-Inf"} { + name := strings.Join(base[:min(2, len(base))], "_") + "_" + value + t.Run(name, func(t *testing.T) { + args := append(append([]string(nil), base...), value) + stdout, stderr, exit := runCmd(t, args...) + if exit != int(output.ExitMisuse) { + t.Fatalf("exit=%d, want %d\nstdout=%s\nstderr=%s", exit, output.ExitMisuse, stdout, stderr) + } + if !strings.Contains(stdout+stderr, "finite") { + t.Fatalf("expected finite-value error\nstdout=%s\nstderr=%s", stdout, stderr) + } + }) + } + } +} + +func TestRepairScrubRejectsNegativeLimitBeforeRepoAccess(t *testing.T) { + stdout, stderr, exit := runCmd(t, + "repair", "scrub", "--repo", "file:///missing", "--limit", "-1") + if exit != int(output.ExitMisuse) { + t.Fatalf("exit=%d, want %d\nstdout=%s\nstderr=%s", exit, output.ExitMisuse, stdout, stderr) + } + if !strings.Contains(stdout+stderr, "--limit must be >= 0") { + t.Fatalf("unexpected error\nstdout=%s\nstderr=%s", stdout, stderr) + } +} + +func TestMutatingRareCommandsRespectReadOnlyMode(t *testing.T) { + tests := []struct { + name string + args func(t *testing.T, repoURL string) []string + }{ + {"rotate_apply", func(_ *testing.T, u string) []string { + return []string{"rotate", "--repo", u, "--apply"} + }}, + {"repair_chunks_apply", func(_ *testing.T, u string) []string { + return []string{"repair", "chunks", "--orphans", "--apply", "--repo", u} + }}, + {"repair_scrub_heal", func(_ *testing.T, u string) []string { + return []string{"repair", "scrub", "--heal", "--replica", "file:///different", "--repo", u} + }}, + {"integrity_run", func(_ *testing.T, u string) []string { + return []string{"integrity", "run", "--repo", u} + }}, + {"insider_scan", func(_ *testing.T, u string) []string { + return []string{"insider", "scan", "--repo", u} + }}, + {"bundle_import", func(t *testing.T, u string) []string { + in := filepath.Join(t.TempDir(), "bundle.tar") + if err := os.WriteFile(in, nil, 0o600); err != nil { + t.Fatal(err) + } + return []string{"repo", "bundle", "import", "--to", u, "--in", in} + }}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + repoURL := initRepoForTest(t) + if _, err := repo.SetMode(context.Background(), repo.SetModeOptions{ + URL: repoURL, Mode: repo.ModeReadOnly, + }); err != nil { + t.Fatalf("set read-only: %v", err) + } + stdout, stderr, exit := runCmd(t, tc.args(t, repoURL)...) + if exit != int(output.ExitConflict) { + t.Fatalf("exit=%d, want %d\nstdout=%s\nstderr=%s", exit, output.ExitConflict, stdout, stderr) + } + if !strings.Contains(stdout+stderr, "repo_read_only") { + t.Fatalf("missing read-only error\nstdout=%s\nstderr=%s", stdout, stderr) + } + }) + } +} diff --git a/internal/cli/rare_options_contract_test.go b/internal/cli/rare_options_contract_test.go new file mode 100644 index 0000000..9633a69 --- /dev/null +++ b/internal/cli/rare_options_contract_test.go @@ -0,0 +1,26 @@ +package cli + +import ( + "strings" + "testing" +) + +func TestRepairScrubDoesNotAdvertisePhantomDryRunFlag(t *testing.T) { + cmd := newRepairScrubCmd() + if cmd.Flags().Lookup("dry-run-heal") != nil { + t.Fatal("unexpected --dry-run-heal flag") + } + if strings.Contains(cmd.Long, "--dry-run-heal") { + t.Fatal("help advertises nonexistent --dry-run-heal") + } +} + +func TestServerHelpDescribesFlagOnlyRuntimeConfiguration(t *testing.T) { + long := newRealServerCmd().Long + if strings.Contains(long, "flags below override the config file's server") { + t.Fatal("server help still promises an unsupported server: config block") + } + if !strings.Contains(long, "flag-only") { + t.Fatal("server help does not explain the current flag-only contract") + } +} diff --git a/internal/cli/repair.go b/internal/cli/repair.go index 4ea919a..e530df6 100644 --- a/internal/cli/repair.go +++ b/internal/cli/repair.go @@ -839,6 +839,17 @@ func runRepairChunks(cmd *cobra.Command, repoURL string, orphans, apply bool, mi return mapRepoOpenErr(repoURL, err) } defer sp.Close() + var mutationLock *repo.MutationLock + if orphans && apply { + if err := assertRepoWritable(cmd.Context(), sp, "repair chunks --orphans --apply"); err != nil { + return err + } + mutationLock, err = repo.AcquireMutationLock(cmd.Context(), sp, "repair chunks --orphans --apply") + if err != nil { + return output.NewError("repair.mutation_locked", fmt.Sprintf("repair chunks: %v", err)).Wrap(err) + } + defer func() { _ = mutationLock.Release(context.Background()) }() + } refs, err := repo.CollectReferences(cmd.Context(), sp) if err != nil { @@ -947,8 +958,8 @@ successful heal restores the local copy to a state where the CAS's plaintext-SHA round-trip passes again. Heal is best-effort: a chunk missing at the replica is reported as -NotAtReplica and the run continues with the next mismatch. Use ---dry-run-heal to see what *would* heal without writing.`, +NotAtReplica and the run continues with the next mismatch. Omit +--heal to run the read-only diagnostic scrub.`, RunE: func(cmd *cobra.Command, _ []string) error { return runRepairScrub(cmd, repoURL, limit, heal, replicaURL) }, @@ -967,6 +978,10 @@ NotAtReplica and the run continues with the next mismatch. Use func runRepairScrub(cmd *cobra.Command, repoURL string, limit int, heal bool, replicaURL string) error { d := DispatcherFrom(cmd) + if limit < 0 { + return output.NewError("usage.bad_flag", + "repair scrub: --limit must be >= 0").Wrap(output.ErrUsage) + } if heal && replicaURL == "" { return output.NewError("usage.missing_flag", "repair scrub: --heal requires --replica ").Wrap(output.ErrUsage) @@ -980,6 +995,11 @@ func runRepairScrub(cmd *cobra.Command, repoURL string, limit int, heal bool, re return mapRepoOpenErr(repoURL, err) } defer sp.Close() + if heal { + if err := assertRepoWritable(cmd.Context(), sp, "repair scrub --heal"); err != nil { + return err + } + } // Per-manifest scrub: every backup manifest carries its own // encryption block (KEK ref + wrapped DEK), so the CAS that can diff --git a/internal/cli/repo_bundle.go b/internal/cli/repo_bundle.go index 6fb26c0..6f34e57 100644 --- a/internal/cli/repo_bundle.go +++ b/internal/cli/repo_bundle.go @@ -145,6 +145,9 @@ func newRepoBundleImportCmd() *cobra.Command { return err } defer sp.Close() + if err := assertRepoWritable(cmd.Context(), sp, "repo bundle import"); err != nil { + return err + } bm, err := bundle.Import(cmd.Context(), f, sp, bundle.ImportOptions{}) if err != nil { diff --git a/internal/cli/repo_gc.go b/internal/cli/repo_gc.go index d732de4..445a3fb 100644 --- a/internal/cli/repo_gc.go +++ b/internal/cli/repo_gc.go @@ -145,6 +145,14 @@ func runRepoGC(cmd *cobra.Command, repoURL string, apply bool, approvalID string "hint": "the gate fires only on --apply; this dry-run does not consult the approval", })) } + var mutationLock *repo.MutationLock + if apply { + mutationLock, err = repo.AcquireMutationLock(cmd.Context(), sp, "repo gc --apply") + if err != nil { + return output.NewError("repo.gc.mutation_locked", fmt.Sprintf("repo gc: %v", err)).Wrap(err) + } + defer func() { _ = mutationLock.Release(context.Background()) }() + } // Tombstone-grace: a 0 flag value means "disable" (caller // explicitly opts out — historical aggressive behaviour); a diff --git a/internal/cli/repo_replicate.go b/internal/cli/repo_replicate.go index 74c8917..13d8000 100644 --- a/internal/cli/repo_replicate.go +++ b/internal/cli/repo_replicate.go @@ -115,9 +115,9 @@ func runRepoReplicate(cmd *cobra.Command, from, to string, includeWAL, dryRun bo return output.NewError("usage.bad_flag", "repo replicate: --from and --to must differ").Wrap(output.ErrUsage) } - if maxMbps < 0 { + if !finiteFloat(maxMbps) || maxMbps < 0 { return output.NewError("usage.bad_flag", - fmt.Sprintf("repo replicate: --max-mbps must be >= 0; got %v", maxMbps)).Wrap(output.ErrUsage) + fmt.Sprintf("repo replicate: --max-mbps must be finite and >= 0; got %v", maxMbps)).Wrap(output.ErrUsage) } if maxMbps > 0 && scheduleExpr != "" { return output.NewError("usage.bad_flag", diff --git a/internal/cli/restore.go b/internal/cli/restore.go index 1ba1a9c..4446791 100644 --- a/internal/cli/restore.go +++ b/internal/cli/restore.go @@ -24,6 +24,7 @@ import ( "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" "github.com/cybertec-postgresql/pg_hardstorage/internal/restore" "github.com/cybertec-postgresql/pg_hardstorage/internal/restore/naturaltime" + "github.com/cybertec-postgresql/pg_hardstorage/internal/restore/postverify" "github.com/cybertec-postgresql/pg_hardstorage/internal/restore/walfetchcmd" ) @@ -203,6 +204,13 @@ func runRestore(cmd *cobra.Command, opts restoreOpts) error { if err := validateRestoreTargets(&opts); err != nil { return err } + verifyMode, err := restore.ParseVerifyMode(opts.verifyMode) + if err != nil { + return err + } + if _, err := postverify.ParseMode(opts.verifyRestoreMode); err != nil { + return output.NewError("usage.bad_verify_restore_mode", err.Error()).Wrap(output.ErrUsage) + } // Control-plane mode short-circuits the local execution path. // The repo / keyring are the agent's concern; the operator only @@ -226,11 +234,6 @@ func runRestore(cmd *cobra.Command, opts restoreOpts) error { return missingFlagErr(cmd, missing...) } - verifyMode, err := restore.ParseVerifyMode(opts.verifyMode) - if err != nil { - return err - } - // Resolve the keyring path the same way the backup command does. p, err := paths.Resolve(paths.DefaultOptions()) if err != nil { diff --git a/internal/cli/restore_controlplane.go b/internal/cli/restore_controlplane.go index 6e09138..ec5f0cc 100644 --- a/internal/cli/restore_controlplane.go +++ b/internal/cli/restore_controlplane.go @@ -39,6 +39,17 @@ import ( // control-plane restore. func runRestoreControlPlane(cmd *cobra.Command, opts restoreOpts) error { d := DispatcherFrom(cmd) + if err := rejectChangedDispatchFlags(cmd, "restore", + "preview", + "force-foreign", + "chain-staging-root", + "reset-chain-staging", + "kms-config", + "skip-gap-check", + "require-threshold-attestation", + ); err != nil { + return err + } // Required-field validation. We mirror the server-side checks at // the CLI boundary too so the operator gets a clear local error @@ -78,6 +89,8 @@ func runRestoreControlPlane(cmd *cobra.Command, opts restoreOpts) error { "backup_id": opts.backupID, // "latest" is resolved server-side "target_dir": opts.targetDir, "allow_overwrite": opts.force, + "verify_after": opts.verifyMode, + "verify_restore": opts.verifyRestoreMode, } if opts.repoURL != "" { body["repo"] = opts.repoURL diff --git a/internal/cli/restore_controlplane_test.go b/internal/cli/restore_controlplane_test.go index baf08c0..e1556c3 100644 --- a/internal/cli/restore_controlplane_test.go +++ b/internal/cli/restore_controlplane_test.go @@ -63,9 +63,22 @@ func TestRestore_ControlPlane_HappyPath(t *testing.T) { args := []string{ "restore", "db1", "latest", "--target", "/tmp/restored", + "--verify", "require", + "--verify-restore", "required", "-o", "json", } onJob := func(s *server.Server, jobID string) { + j, err := s.Jobs().Get(jobID) + if err != nil { + t.Errorf("get job: %v", err) + return + } + if j.Args["verify_after"] != "require" { + t.Errorf("verify_after=%v, want require", j.Args["verify_after"]) + } + if j.Args["verify_restore"] != "required" { + t.Errorf("verify_restore=%v, want required", j.Args["verify_restore"]) + } // Claim — emulates what an agent would do. if _, err := s.Jobs().Claim(server.ClaimOptions{ AgentID: "fake-agent", @@ -192,3 +205,59 @@ func TestRestore_ControlPlane_ConflictingTargets(t *testing.T) { t.Errorf("expected conflicting_targets code; stderr=%s", errb.String()) } } + +func TestRestore_ControlPlane_RefusesLocalOnlySafetyFlags(t *testing.T) { + tests := [][]string{ + {"--preview"}, + {"--require-threshold-attestation", "prod-admins"}, + {"--skip-gap-check"}, + {"--kms-config", "region=eu-central-1"}, + {"--chain-staging-root", "/tmp/staging"}, + {"--reset-chain-staging"}, + {"--force-foreign"}, + } + for _, extra := range tests { + t.Run(strings.Join(extra, "_"), func(t *testing.T) { + root := cli.NewRoot() + var out, errb bytes.Buffer + root.SetOut(&out) + root.SetErr(&errb) + args := []string{ + "restore", "db1", "latest", + "--target", "/tmp/restored", + "--control-plane", "http://does-not-matter:8443", + "-o", "json", + } + root.SetArgs(append(args, extra...)) + exit := cli.Run(root) + if exit != int(output.ExitMisuse) { + t.Fatalf("exit=%d, want %d; stderr=%s", exit, output.ExitMisuse, errb.String()) + } + if !strings.Contains(errb.String(), "usage.unsupported_flag") { + t.Fatalf("missing unsupported-flag error: %s", errb.String()) + } + }) + } +} + +func TestRestore_ControlPlane_RejectsBadVerificationModesBeforeDispatch(t *testing.T) { + for _, extra := range [][]string{ + {"--verify", "typo"}, + {"--verify-restore", "typo"}, + } { + t.Run(strings.Join(extra, "_"), func(t *testing.T) { + root := cli.NewRoot() + var out, errb bytes.Buffer + root.SetOut(&out) + root.SetErr(&errb) + args := []string{ + "restore", "db1", "latest", "--target", "/tmp/restored", + "--control-plane", "http://does-not-matter:8443", "-o", "json", + } + root.SetArgs(append(args, extra...)) + if exit := cli.Run(root); exit != int(output.ExitMisuse) { + t.Fatalf("exit=%d, want misuse; stderr=%s", exit, errb.String()) + } + }) + } +} diff --git a/internal/cli/rotate.go b/internal/cli/rotate.go index e200c77..8b1853c 100644 --- a/internal/cli/rotate.go +++ b/internal/cli/rotate.go @@ -128,6 +128,11 @@ func runRotate(cmd *cobra.Command, opts rotateOpts) error { return mapRepoOpenErr(opts.repoURL, err) } defer sp.Close() + if opts.apply { + if err := assertRepoWritable(cmd.Context(), sp, "rotate --apply"); err != nil { + return err + } + } auditStore := audit.NewStoreWithRetention(sp, repoMeta.WORM) store := backup.NewManifestStore(sp) @@ -227,6 +232,10 @@ func runRotate(cmd *cobra.Command, opts rotateOpts) error { func buildPolicy(opts rotateOpts) (retention.Policy, error) { switch strings.ToLower(opts.policy) { case "gfs": + if opts.keepDaily < 0 || opts.keepWeekly < 0 || opts.keepMonthly < 0 || opts.keepYearly < 0 { + return nil, output.NewError("usage.bad_flag", + "rotate: GFS --keep-daily, --keep-weekly, --keep-monthly, and --keep-yearly must all be >= 0").Wrap(output.ErrUsage) + } return retention.GFSPolicy{ KeepDaily: opts.keepDaily, KeepWeekly: opts.keepWeekly, diff --git a/internal/cli/rotate_test.go b/internal/cli/rotate_test.go index fc82499..3ccde2c 100644 --- a/internal/cli/rotate_test.go +++ b/internal/cli/rotate_test.go @@ -69,6 +69,20 @@ func TestBuildPolicy_CountRejectsNegativeKeepFulls(t *testing.T) { } } +func TestBuildPolicy_GFSRejectsNegativeCounts(t *testing.T) { + tests := []rotateOpts{ + {policy: "gfs", keepDaily: -1}, + {policy: "gfs", keepWeekly: -1}, + {policy: "gfs", keepMonthly: -1}, + {policy: "gfs", keepYearly: -1}, + } + for _, opts := range tests { + if _, err := buildPolicy(opts); err == nil { + t.Fatalf("negative GFS count accepted: %+v", opts) + } + } +} + func TestShapeDecisions_KeepBeforeDeleteWhenSameStoppedAt(t *testing.T) { now := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC) keep := &backup.Manifest{BackupID: "k1", StoppedAt: now} diff --git a/internal/cli/server.go b/internal/cli/server.go index fbba53a..9393ae9 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -68,8 +68,9 @@ Endpoints (all under /v1/): GET /agents [?include_inactive=true] POST /agents/heartbeat -The flags below override the config file's server: block. Use ---config to point at a YAML; flag overrides win.`, +Server runtime settings are currently flag-only. The global --config file +continues to configure deployments, repositories, sinks, and paths, but it +does not define a server: block.`, SilenceUsage: true, RunE: func(cmd *cobra.Command, _ []string) error { cfg := server.Config{ diff --git a/internal/cli/threshold.go b/internal/cli/threshold.go index 36c59aa..0a6de13 100644 --- a/internal/cli/threshold.go +++ b/internal/cli/threshold.go @@ -237,6 +237,9 @@ func runThresholdRosterCreate(cmd *cobra.Command, id string, f thresholdRosterCr return err } defer sp.Close() + if err := assertRepoWritable(cmd.Context(), sp, "threshold roster create"); err != nil { + return err + } // WORM-lock the roster on a compliance repo (it gates restore / // kms-shred quorum, so it must be as immutable as the audit chain), // and anchor its creator to the local key that just signed it. @@ -420,6 +423,9 @@ func runThresholdAttestSign(cmd *cobra.Command, kind, id string, f thresholdAtte return err } defer sp.Close() + if err := assertRepoWritable(cmd.Context(), sp, "threshold attest sign"); err != nil { + return err + } // Anchor the roster to the local operator key before vouching for a // backup under it — don't lend a signature to a roster this operator // didn't create (a forged roster's attestation is inert at the restore diff --git a/internal/cli/wal.go b/internal/cli/wal.go index bbd0b30..219386c 100644 --- a/internal/cli/wal.go +++ b/internal/cli/wal.go @@ -929,10 +929,10 @@ func deploymentBackupKEKRef(ctx context.Context, sp storage.StoragePlugin, deplo // read-side mitigation still covers any cross-posture chunk collision. func buildWALEncryption(ctx context.Context, sp storage.StoragePlugin, deployment string, cfg *runner.EncryptionConfig) (encryption.Encryptor, *walsink.EncryptionInfo, error) { if cfg == nil { - return nil, nil, nil // no encryption configured → plaintext WAL - } - if established, ok := deploymentBackupKEKRef(ctx, sp, deployment); ok && established != cfg.KEKRef { - return nil, nil, nil // mismatched posture → stay plaintext (avoid divergence) + if err := sharedkey.EnsurePlaintextAllowed(ctx, sp); err != nil { + return nil, nil, fmt.Errorf("wal: encryption posture: %w", err) + } + return nil, nil, nil } // Custody-specific wrap/unwrap — the only thing that differs between the @@ -954,26 +954,9 @@ func buildWALEncryption(ctx context.Context, sp storage.StoragePlugin, deploymen wrap = func(dek [encryption.KeyLen]byte) ([]byte, error) { return encryption.Wrap(kek, dek) } } - res, rerr := sharedkey.Resolve(ctx, sp, cfg.KEKRef, unwrap) + dek, rerr := sharedkey.ResolveOrCreate(ctx, sp, cfg.KEKRef, unwrap, wrap) if rerr != nil { - return nil, nil, fmt.Errorf("wal: cannot determine the shared DEK; refusing to write WAL that the CAS's plaintext-hash dedup would leave unrestorable: %w", rerr) - } - - var dek [encryption.KeyLen]byte - switch { - case res.DEK != nil: - if len(res.DEK) != encryption.KeyLen { - return nil, nil, fmt.Errorf("wal: reused shared DEK has wrong length %d", len(res.DEK)) - } - copy(dek[:], res.DEK) - case res.SawCandidate: - return nil, nil, fmt.Errorf("wal: a prior DEK for KEK %q exists but could not be unwrapped; refusing to write WAL under a fresh DEK that would leave deduped chunks unrestorable (verify the KEK material matches this repo)", cfg.KEKRef) - default: - fresh, gerr := encryption.GenerateDEK() - if gerr != nil { - return nil, nil, fmt.Errorf("wal: generate DEK: %w", gerr) - } - dek = fresh + return nil, nil, fmt.Errorf("wal: establish repository encryption domain: %w", rerr) } wrapped, werr := wrap(dek) diff --git a/internal/cli/wal_encrypt_test.go b/internal/cli/wal_encrypt_test.go index a761a16..391cec1 100644 --- a/internal/cli/wal_encrypt_test.go +++ b/internal/cli/wal_encrypt_test.go @@ -7,6 +7,7 @@ import ( "errors" "os" "path/filepath" + "strings" "testing" "github.com/cybertec-postgresql/pg_hardstorage/internal/backup" @@ -208,13 +209,13 @@ func TestWalFetch_EncryptedWithoutKeyringFails(t *testing.T) { } } -// TestWalPush_MixedCloudPostureStaysPlaintext guards the divergence hazard: a +// TestWalPush_MixedCloudPostureIsRefused guards the divergence hazard: a // local kek.bin is present, but the deployment's backups are encrypted under a -// CLOUD KEKRef. Encrypting WAL under the local KEK would diverge from the -// backup DEK and leave deduped chunks unrestorable, so WAL must stay plaintext -// (cloud-KMS WAL encryption is tracked in #108). The read-side mitigation -// still covers any chunk collision. -func TestWalPush_MixedCloudPostureStaysPlaintext(t *testing.T) { +// CLOUD KEKRef. Neither encrypting WAL under the local KEK nor falling back to +// plaintext is safe in a global plaintext-addressed CAS: either posture can +// deduplicate against bytes the other writer cannot restore. Refuse before a +// segment manifest is committed. +func TestWalPush_MixedCloudPostureIsRefused(t *testing.T) { keyringDir := t.TempDir() t.Setenv("PG_HARDSTORAGE_KEYRING_DIR", keyringDir) if _, _, err := keystore.LoadOrGenerateKEK(keyringDir); err != nil { @@ -244,7 +245,7 @@ func TestWalPush_MixedCloudPostureStaysPlaintext(t *testing.T) { t.Fatal(err) } - // Push a WAL segment — it must NOT be encrypted (would diverge). + // Push a WAL segment — it must be refused before committing a manifest. segmentName := "000000010000000000000005" segPath := filepath.Join(t.TempDir(), segmentName) body := make([]byte, walsink.SegmentSize) @@ -255,21 +256,15 @@ func TestWalPush_MixedCloudPostureStaysPlaintext(t *testing.T) { t.Fatal(err) } if _, errb, exit := runCLI(t, "wal", "push", "db1", segPath, - "--repo", repoURL, "--system-identifier", "7000000000000000001", "-o", "json"); exit != int(output.ExitOK) { - t.Fatalf("wal push exit=%d\n%s", exit, errb) + "--repo", repoURL, "--system-identifier", "7000000000000000001", "-o", "json"); exit == int(output.ExitOK) { + t.Fatalf("wal push unexpectedly accepted a conflicting encryption posture\n%s", errb) + } else if !strings.Contains(errb, "undecryptable dedup collision") { + t.Fatalf("wal push returned the wrong refusal\n%s", errb) } segManifest := filepath.Join(repoRoot, "wal", "db1", "00000001", segmentName+".json") - sraw, err := os.ReadFile(segManifest) - if err != nil { - t.Fatalf("read segment manifest: %v", err) - } - var sm walsink.SegmentManifest - if err := json.Unmarshal(sraw, &sm); err != nil { - t.Fatalf("parse segment manifest: %v", err) - } - if sm.Encryption != nil { - t.Errorf("WAL must stay plaintext when backups use a cloud KEKRef (divergence guard); got envelope %+v", sm.Encryption) + if _, err := os.Stat(segManifest); !os.IsNotExist(err) { + t.Fatalf("conflicting WAL push committed a segment manifest: %v", err) } } diff --git a/internal/pg/walsink/push.go b/internal/pg/walsink/push.go index 6daf1cc..ea98110 100644 --- a/internal/pg/walsink/push.go +++ b/internal/pg/walsink/push.go @@ -506,6 +506,14 @@ func ParseSegmentName(name string, segmentSize int64) (tli uint32, segNum uint64 // the push path stays self-contained. The optional WORM policy // propagates a per-Put retention deadline. func commitManifestStandalone(ctx context.Context, sp storage.StoragePlugin, m *SegmentManifest, worm *repo.WORMPolicy) error { + lock, err := repo.AcquireMutationLock(ctx, sp, "WAL push manifest "+m.SegmentName) + if err != nil { + return fmt.Errorf("walsink push: commit mutation lock: %w", err) + } + defer func() { _ = lock.Release(context.Background()) }() + if err := ensureSegmentChunksPresent(ctx, sp, m); err != nil { + return err + } body, err := m.MarshalToBytes() if err != nil { return err diff --git a/internal/pg/walsink/walsink.go b/internal/pg/walsink/walsink.go index 2fcfd8e..fbd5c62 100644 --- a/internal/pg/walsink/walsink.go +++ b/internal/pg/walsink/walsink.go @@ -854,6 +854,14 @@ func (s *Sink) procErrLoad() error { // renames it to the canonical key. RenameIfNotExists makes a re-commit // of an existing segment a no-op (idempotent). func (s *Sink) commitManifest(ctx context.Context, m *SegmentManifest) error { + lock, err := repo.AcquireMutationLock(ctx, s.sp, "WAL manifest "+m.SegmentName) + if err != nil { + return fmt.Errorf("walsink: commit mutation lock: %w", err) + } + defer func() { _ = lock.Release(context.Background()) }() + if err := ensureSegmentChunksPresent(ctx, s.sp, m); err != nil { + return err + } body, err := m.MarshalToBytes() if err != nil { return err @@ -890,6 +898,15 @@ func (s *Sink) commitManifest(ctx context.Context, m *SegmentManifest) error { return nil } +func ensureSegmentChunksPresent(ctx context.Context, sp storage.StoragePlugin, m *SegmentManifest) error { + for _, ref := range m.Chunks { + if _, err := sp.Stat(ctx, repo.ChunkKey(ref.Hash)); err != nil { + return fmt.Errorf("walsink: refusing to commit segment %s: referenced chunk %s is unavailable after repository mutation fencing: %w", m.SegmentName, ref.Hash, err) + } + } + return nil +} + // randSuffix returns a short hex suffix for tmp-file names so // concurrent commits (rare here, but possible if multiple agents ever // stream the same slot) don't collide on the staging slot. diff --git a/internal/pg/walsink/worm_propagation_test.go b/internal/pg/walsink/worm_propagation_test.go index d0e9d18..e4829cb 100644 --- a/internal/pg/walsink/worm_propagation_test.go +++ b/internal/pg/walsink/worm_propagation_test.go @@ -120,6 +120,9 @@ func TestPushSegmentFile_PropagatesWORMRetention(t *testing.T) { t.Fatal("expected at least one Put") } for _, p := range rec.puts { + if p.Key == repo.MutationLockKey { + continue // ephemeral coordination record, never retained + } if p.Opts.RetainUntil.IsZero() { t.Errorf("Put %q has zero RetainUntil despite WORM policy", p.Key) } @@ -132,6 +135,9 @@ func TestPushSegmentFile_PropagatesWORMRetention(t *testing.T) { // know the manifest path (not just chunks) carries WORM too. var sawManifestTmp bool for _, p := range rec.puts { + if p.Key == repo.MutationLockKey { + continue // ephemeral coordination record, never retained + } if strings.Contains(p.Key, "wal/db1/") && strings.Contains(p.Key, ".json.tmp.") { sawManifestTmp = true break @@ -230,6 +236,9 @@ func TestSink_PropagatesWORMRetentionToManifest(t *testing.T) { } var sawManifest bool for _, p := range rec.puts { + if p.Key == repo.MutationLockKey { + continue + } if p.Opts.RetainUntil.IsZero() { t.Errorf("streamed Put %q has zero RetainUntil despite WORM policy", p.Key) } diff --git a/internal/repo/bundle/bundle.go b/internal/repo/bundle/bundle.go index 1bf7c25..42f03bf 100644 --- a/internal/repo/bundle/bundle.go +++ b/internal/repo/bundle/bundle.go @@ -41,6 +41,9 @@ import ( "time" "github.com/cybertec-postgresql/pg_hardstorage/internal/backup" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/compression" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/compression/none" + compressionzstd "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/compression/zstd" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" ) @@ -291,6 +294,11 @@ const ( // already present are tolerated (Commit returns ErrExists which is // downgraded to a no-op here). Returns the bundle's Manifest. func Import(ctx context.Context, r io.Reader, sp storage.StoragePlugin, opts ImportOptions) (*Manifest, error) { + mutationLock, err := repo.AcquireMutationLock(ctx, sp, "repository bundle import") + if err != nil { + return nil, fmt.Errorf("bundle: mutation lock: %w", err) + } + defer func() { _ = mutationLock.Release(context.Background()) }() tr := tar.NewReader(r) maxEntries := opts.MaxEntries @@ -589,7 +597,30 @@ func verifyChunkPayload(key string, body []byte) error { if err != nil { return fmt.Errorf("bundle: reject chunk %s: not a valid chunk key: %w", key, err) } - got := repo.HashOf(body) + algo, enc, payload, err := compression.ReadEnvelope(body) + if err != nil { + return fmt.Errorf("bundle: reject chunk %s: invalid storage envelope: %w", key, err) + } + // The key addresses plaintext, while bundle entries preserve the raw + // compression/encryption envelope. Encrypted payloads cannot be checked + // against the plaintext address without the manifest's KEK; AES-GCM will + // authenticate them on restore. We can still reject malformed envelopes. + if enc.IsEncrypted() { + return nil + } + var plaintext []byte + switch algo { + case compression.AlgoNone: + plaintext, err = (none.Compressor{}).Decompress(payload) + case compression.AlgoZstd: + plaintext, err = compressionzstd.NewDefault().Decompress(payload) + default: + err = fmt.Errorf("unsupported compression algorithm %d", algo) + } + if err != nil { + return fmt.Errorf("bundle: reject chunk %s: decode storage envelope: %w", key, err) + } + got := repo.HashOf(plaintext) if got != want { return fmt.Errorf("bundle: reject chunk %s: payload SHA-256 %s does not match key hash %s", key, got.String(), want.String()) diff --git a/internal/repo/bundle/bundle_caps_test.go b/internal/repo/bundle/bundle_caps_test.go index 540c581..8b1db54 100644 --- a/internal/repo/bundle/bundle_caps_test.go +++ b/internal/repo/bundle/bundle_caps_test.go @@ -7,6 +7,7 @@ import ( "strings" "testing" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/compression" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo/bundle" ) @@ -28,15 +29,16 @@ func tarWithChunkEntries(t *testing.T, n, size int) []byte { body = nil } key := repo.ChunkKey(repo.HashOf(body)) + envelope := compression.WriteEnvelope(compression.AlgoNone, compression.EncryptionFields{}, body) if err := tw.WriteHeader(&tar.Header{ Name: key, Typeflag: tar.TypeReg, - Size: int64(len(body)), + Size: int64(len(envelope)), Mode: 0o644, }); err != nil { t.Fatal(err) } - if _, err := tw.Write(body); err != nil { + if _, err := tw.Write(envelope); err != nil { t.Fatal(err) } } diff --git a/internal/repo/bundle/bundle_test.go b/internal/repo/bundle/bundle_test.go index e783297..c3da157 100644 --- a/internal/repo/bundle/bundle_test.go +++ b/internal/repo/bundle/bundle_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/cybertec-postgresql/pg_hardstorage/internal/backup" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/compression" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage/fs" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" @@ -58,7 +59,8 @@ func putChunk(t *testing.T, sp storage.StoragePlugin, body []byte) (repo.Hash, i t.Helper() hash := repo.HashOf(body) key := repo.ChunkKey(hash) - _, err := sp.Put(context.Background(), key, bytes.NewReader(body), storage.PutOptions{IfNotExists: true}) + envelope := compression.WriteEnvelope(compression.AlgoNone, compression.EncryptionFields{}, body) + _, err := sp.Put(context.Background(), key, bytes.NewReader(envelope), storage.PutOptions{IfNotExists: true}) if err != nil { t.Fatalf("put chunk: %v", err) } diff --git a/internal/repo/bundle/review79_bundle_test.go b/internal/repo/bundle/review79_bundle_test.go index 11c3f0d..c7db0b8 100644 --- a/internal/repo/bundle/review79_bundle_test.go +++ b/internal/repo/bundle/review79_bundle_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/compression" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo" "github.com/cybertec-postgresql/pg_hardstorage/internal/repo/bundle" @@ -47,12 +48,12 @@ func TestImport_RejectsChunkHashMismatch(t *testing.T) { honest := []byte("honest-payload") key := repo.ChunkKey(repo.HashOf(honest)) // ... but the bundle ships DIFFERENT bytes at that key. - forged := "totally-different-bytes" + forged := compression.WriteEnvelope(compression.AlgoNone, compression.EncryptionFields{}, []byte("totally-different-bytes")) bundleJSON := `{"schema":"pg_hardstorage.repobundle.v1"}` data := writeTar(t, []struct{ name, body string }{ {"bundle.json", bundleJSON}, - {key, forged}, + {key, string(forged)}, }) dst := newRepo(t) @@ -76,10 +77,11 @@ func TestImport_RejectsChunkHashMismatch(t *testing.T) { func TestImport_AcceptsMatchingChunkHash(t *testing.T) { honest := "honest-payload" key := repo.ChunkKey(repo.HashOf([]byte(honest))) + envelope := compression.WriteEnvelope(compression.AlgoNone, compression.EncryptionFields{}, []byte(honest)) bundleJSON := `{"schema":"pg_hardstorage.repobundle.v1"}` data := writeTar(t, []struct{ name, body string }{ {"bundle.json", bundleJSON}, - {key, honest}, + {key, string(envelope)}, }) dst := newRepo(t) diff --git a/internal/repo/mutationlock.go b/internal/repo/mutationlock.go new file mode 100644 index 0000000..65e9c25 --- /dev/null +++ b/internal/repo/mutationlock.go @@ -0,0 +1,140 @@ +package repo + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "time" + + "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/storage" +) + +// MutationLockKey serializes manifest publication with destructive GC. Chunk +// uploads may proceed without the lock, but every manifest publisher must hold +// it and re-check its chunks before publication. GC holds it from reference +// collection through deletion, so it cannot delete from a stale snapshot and +// then race a manifest into existence. +const MutationLockKey = "_locks/repository-mutation.json" + +var ErrMutationLocked = errors.New("repo: another repository mutation is in progress") + +// MutationLockedError describes the current holder when its lock record can +// be read. Callers may use Purpose to distinguish a duplicate operation they +// can briefly wait for from an unrelated mutation that should fail fast. +type MutationLockedError struct { + Purpose string +} + +func (e *MutationLockedError) Error() string { + if e.Purpose == "" { + return fmt.Sprintf("%v (lock key %s)", ErrMutationLocked, MutationLockKey) + } + return fmt.Sprintf("%v: %s (lock key %s)", ErrMutationLocked, e.Purpose, MutationLockKey) +} + +func (e *MutationLockedError) Unwrap() error { return ErrMutationLocked } + +type mutationLockBody struct { + Owner string `json:"owner"` + Purpose string `json:"purpose"` + CreatedAt time.Time `json:"created_at"` +} + +type MutationLock struct { + sp storage.StoragePlugin + owner string +} + +// AcquireMutationLock atomically acquires the repository mutation lock. Locks +// deliberately do not expire automatically: breaking a lock that still has a +// live holder re-opens the GC/commit race. A lock left by a crashed process is +// fail-closed and can be removed only after an operator confirms no writer or +// GC process is active. +func AcquireMutationLock(ctx context.Context, sp storage.StoragePlugin, purpose string) (*MutationLock, error) { + if sp == nil { + return nil, errors.New("repo: mutation lock requires storage") + } + var raw [16]byte + if _, err := rand.Read(raw[:]); err != nil { + return nil, fmt.Errorf("repo: mutation lock owner: %w", err) + } + owner := hex.EncodeToString(raw[:]) + body, err := json.Marshal(mutationLockBody{Owner: owner, Purpose: purpose, CreatedAt: time.Now().UTC()}) + if err != nil { + return nil, err + } + for { + _, err = sp.Put(ctx, MutationLockKey, bytes.NewReader(body), storage.PutOptions{ + IfNotExists: true, ContentLength: int64(len(body)), + }) + if errors.Is(err, storage.ErrAlreadyExists) { + held, readErr := readMutationLockBody(ctx, sp) + if errors.Is(readErr, storage.ErrNotFound) { + // The holder released between our conditional Put and Get. + // Retry the acquisition instead of surfacing a false conflict. + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + continue + } + if readErr != nil { + return nil, fmt.Errorf("%w (lock key %s; holder metadata unreadable: %v)", ErrMutationLocked, MutationLockKey, readErr) + } + return nil, &MutationLockedError{Purpose: held.Purpose} + } + if err != nil { + return nil, fmt.Errorf("repo: acquire mutation lock: %w", err) + } + return &MutationLock{sp: sp, owner: owner}, nil + } +} + +// Release removes the lock only when the stored owner still matches. +func (l *MutationLock) Release(ctx context.Context) error { + if l == nil || l.sp == nil { + return nil + } + held, err := readMutationLockBody(ctx, l.sp) + if err != nil { + if errors.Is(err, storage.ErrNotFound) { + return nil + } + return fmt.Errorf("repo: read mutation lock for release: %w", err) + } + if held.Owner != l.owner { + return errors.New("repo: mutation lock ownership changed; refusing to release another holder's lock") + } + if err := l.sp.Delete(ctx, MutationLockKey); err != nil { + return fmt.Errorf("repo: release mutation lock: %w", err) + } + l.sp = nil + return nil +} + +func readMutationLockBody(ctx context.Context, sp storage.StoragePlugin) (mutationLockBody, error) { + rc, err := sp.Get(ctx, MutationLockKey) + if err != nil { + return mutationLockBody{}, err + } + body, readErr := io.ReadAll(io.LimitReader(rc, 1<<20)) + closeErr := rc.Close() + if readErr != nil { + return mutationLockBody{}, fmt.Errorf("repo: read mutation lock body: %w", readErr) + } + if closeErr != nil { + return mutationLockBody{}, fmt.Errorf("repo: close mutation lock body: %w", closeErr) + } + var held mutationLockBody + if err := json.Unmarshal(body, &held); err != nil { + return mutationLockBody{}, fmt.Errorf("repo: decode mutation lock: %w", err) + } + if held.Owner == "" { + return mutationLockBody{}, errors.New("repo: mutation lock has no owner") + } + return held, nil +} diff --git a/internal/repo/replicate.go b/internal/repo/replicate.go index e899b0c..8b3046f 100644 --- a/internal/repo/replicate.go +++ b/internal/repo/replicate.go @@ -212,6 +212,16 @@ func Replicate(ctx context.Context, src, dst storage.StoragePlugin, opts Replica return res, fmt.Errorf("%w: destination repo is WORM-configured (%s) but backend %q does not enforce retention; the replica would be freely deletable. Use a WORM-capable destination or pass AllowUnenforcedWORM to accept the gap", ErrRetentionUnenforceable, opts.DstWORM.Mode, dst.Name()) } + var mutationLock *MutationLock + if !opts.DryRun { + var lockErr error + mutationLock, lockErr = AcquireMutationLock(ctx, dst, "repository replication") + if lockErr != nil { + finish() + return res, fmt.Errorf("repo replicate: mutation lock: %w", lockErr) + } + defer func() { _ = mutationLock.Release(context.Background()) }() + } // Pass 1: collect tombstoned backup IDs at src. A separate pass // (rather than interleaving with the main walk) keeps the logic diff --git a/internal/repo/sharedkey/sharedkey.go b/internal/repo/sharedkey/sharedkey.go index 3a8e646..2f975f4 100644 --- a/internal/repo/sharedkey/sharedkey.go +++ b/internal/repo/sharedkey/sharedkey.go @@ -15,11 +15,14 @@ package sharedkey import ( + "bytes" "context" "encoding/base64" "encoding/json" + "errors" "fmt" "io" + "sort" "strings" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/encryption" @@ -29,6 +32,21 @@ import ( // Scheme is the only envelope scheme DEK reuse applies to. const Scheme = "aes-256-gcm" +// DomainKey is the repository-wide encryption-domain record. The CAS key +// space is global and keyed by plaintext hash, so every encrypted writer in a +// repository must use the same plaintext DEK. A conditional create of this +// record is the serialization point for the first backup/WAL writer. +const DomainKey = "_encryption/domain.json" + +const domainSchema = "pg_hardstorage.encryption_domain.v1" + +type domainRecord struct { + Schema string `json:"schema"` + Scheme string `json:"scheme"` + KEKRef string `json:"kek_ref"` + WrappedDEK string `json:"wrapped_dek"` +} + // Unwrapper turns a wrapped-DEK blob into the plaintext DEK. It encapsulates // the KEK custody model — local AES-256-GCM unwrap or a cloud-KMS round-trip — // so this package stays free of keystore / kms dependencies. Returns a non-nil @@ -84,6 +102,7 @@ type manifestEnvelope struct { // candidate. This matches the backup-side dek-reuse posture. func Resolve(ctx context.Context, sp storage.StoragePlugin, wantKEKRef string, unwrap Unwrapper) (Result, error) { var res Result + var resolved []byte for _, prefix := range []string{"manifests/", "wal/"} { for info, err := range sp.List(ctx, prefix) { if err != nil { @@ -92,20 +111,219 @@ func Resolve(ctx context.Context, sp storage.StoragePlugin, wantKEKRef string, u if !isManifestKey(prefix, info.Key) { continue } - wrapped, ok := readWrappedDEK(ctx, sp, info.Key, wantKEKRef) - if !ok { + env, err := readEnvelope(ctx, sp, info.Key) + if err != nil { + return Result{}, fmt.Errorf("sharedkey: read %s: %w", info.Key, err) + } + if env == nil || env.Scheme != Scheme || env.KEKRef != wantKEKRef { continue } + wrapped, err := base64.StdEncoding.DecodeString(env.WrappedDEK) + if err != nil { + return Result{}, fmt.Errorf("sharedkey: decode wrapped DEK in %s: %w", info.Key, err) + } + ok := len(wrapped) > 0 + if !ok { + return Result{}, fmt.Errorf("sharedkey: empty wrapped DEK in %s", info.Key) + } res.SawCandidate = true - if dek, uerr := unwrap(wrapped); uerr == nil && len(dek) == encryption.KeyLen { - res.DEK = dek - return res, nil + dek, uerr := unwrap(wrapped) + if uerr != nil || len(dek) != encryption.KeyLen { + continue + } + if resolved == nil { + resolved = append([]byte(nil), dek...) + continue + } + if !bytes.Equal(resolved, dek) { + return Result{}, fmt.Errorf("sharedkey: manifests under KEK %q contain divergent DEKs; refusing to choose one for a plaintext-addressed CAS", wantKEKRef) } } } + res.DEK = resolved return res, nil } +// ResolveOrCreate returns the one repository-wide plaintext DEK. The first +// writer publishes DomainKey with IfNotExists; a concurrent loser reads and +// adopts the winner's record. Repositories bootstrapped before DomainKey was +// introduced are migrated from their manifests, but only when all encrypted +// manifests use the requested KEKRef and resolve to one DEK. +func ResolveOrCreate( + ctx context.Context, + sp storage.StoragePlugin, + kekRef string, + unwrap Unwrapper, + wrap func([encryption.KeyLen]byte) ([]byte, error), +) ([encryption.KeyLen]byte, error) { + if sp == nil || unwrap == nil || wrap == nil || kekRef == "" { + return [encryption.KeyLen]byte{}, errors.New("sharedkey: storage, kekRef, wrap and unwrap are required") + } + if rec, found, err := readDomain(ctx, sp); err != nil { + return [encryption.KeyLen]byte{}, err + } else if found { + return unwrapDomain(rec, kekRef, unwrap) + } + + refs, err := DiscoverKEKRefs(ctx, sp) + if err != nil { + return [encryption.KeyLen]byte{}, err + } + if len(refs) > 1 || len(refs) == 1 && refs[0] != kekRef { + return [encryption.KeyLen]byte{}, fmt.Errorf("sharedkey: repository CAS already contains encrypted artifacts under KEKRef(s) %v; requested %q would create an undecryptable dedup collision", refs, kekRef) + } + + res, err := Resolve(ctx, sp, kekRef, unwrap) + if err != nil { + return [encryption.KeyLen]byte{}, err + } + var dek [encryption.KeyLen]byte + switch { + case res.DEK != nil: + copy(dek[:], res.DEK) + case res.SawCandidate: + return dek, fmt.Errorf("sharedkey: prior DEK for KEK %q cannot be unwrapped", kekRef) + default: + fresh, err := encryption.GenerateDEK() + if err != nil { + return dek, fmt.Errorf("sharedkey: generate DEK: %w", err) + } + dek = fresh + } + + w, err := wrap(dek) + if err != nil { + return [encryption.KeyLen]byte{}, fmt.Errorf("sharedkey: wrap domain DEK: %w", err) + } + rec := domainRecord{Schema: domainSchema, Scheme: Scheme, KEKRef: kekRef, WrappedDEK: base64.StdEncoding.EncodeToString(w)} + body, err := json.Marshal(rec) + if err != nil { + return [encryption.KeyLen]byte{}, fmt.Errorf("sharedkey: marshal domain: %w", err) + } + _, err = sp.Put(ctx, DomainKey, bytes.NewReader(body), storage.PutOptions{ + IfNotExists: true, ContentLength: int64(len(body)), + }) + if err == nil { + return dek, nil + } + if !errors.Is(err, storage.ErrAlreadyExists) { + return [encryption.KeyLen]byte{}, fmt.Errorf("sharedkey: publish domain: %w", err) + } + winner, found, err := readDomain(ctx, sp) + if err != nil { + return [encryption.KeyLen]byte{}, err + } + if !found { + return [encryption.KeyLen]byte{}, errors.New("sharedkey: encryption-domain create lost race but winner is absent") + } + return unwrapDomain(winner, kekRef, unwrap) +} + +// EnsurePlaintextAllowed refuses a plaintext writer once any encrypted +// artifact/domain exists. An encryption-aware CAS can read legacy plaintext +// envelopes, but a plaintext CAS cannot read an encrypted dedup winner. +func EnsurePlaintextAllowed(ctx context.Context, sp storage.StoragePlugin) error { + if rec, found, err := readDomain(ctx, sp); err != nil { + return err + } else if found { + if rec.Scheme == "none" { + return nil + } + return errors.New("sharedkey: repository encryption domain is established; refusing a plaintext writer that could deduplicate against ciphertext it cannot restore") + } + refs, err := DiscoverKEKRefs(ctx, sp) + if err != nil { + return err + } + if len(refs) > 0 { + return fmt.Errorf("sharedkey: repository contains encrypted artifacts under KEKRef(s) %v; refusing plaintext writes", refs) + } + rec := domainRecord{Schema: domainSchema, Scheme: "none"} + body, _ := json.Marshal(rec) + _, err = sp.Put(ctx, DomainKey, bytes.NewReader(body), storage.PutOptions{ + IfNotExists: true, ContentLength: int64(len(body)), + }) + if err == nil { + return nil + } + if !errors.Is(err, storage.ErrAlreadyExists) { + return fmt.Errorf("sharedkey: publish plaintext domain: %w", err) + } + winner, found, err := readDomain(ctx, sp) + if err != nil { + return err + } + if !found || winner.Scheme != "none" { + return errors.New("sharedkey: concurrent encrypted writer established the repository domain; refusing plaintext writes") + } + return nil +} + +// RotateDomain re-wraps the repository DEK after every manifest has been +// rotated. The caller must hold the repository mutation lock so no writer can +// observe a half-rotated key posture. Legacy repositories without DomainKey +// are left alone; their next writer bootstraps the record from rotated +// manifests. +func RotateDomain( + ctx context.Context, + sp storage.StoragePlugin, + oldKEKRef, newKEKRef string, + unwrapOld Unwrapper, + wrapNew func([encryption.KeyLen]byte) ([]byte, error), +) error { + rec, found, err := readDomain(ctx, sp) + if err != nil || !found { + return err + } + dek, err := unwrapDomain(rec, oldKEKRef, unwrapOld) + if err != nil { + return err + } + w, err := wrapNew(dek) + if err != nil { + return fmt.Errorf("sharedkey: wrap rotated domain DEK: %w", err) + } + next := domainRecord{Schema: domainSchema, Scheme: Scheme, KEKRef: newKEKRef, WrappedDEK: base64.StdEncoding.EncodeToString(w)} + body, err := json.Marshal(next) + if err != nil { + return err + } + if _, err := sp.Put(ctx, DomainKey, bytes.NewReader(body), storage.PutOptions{ContentLength: int64(len(body))}); err != nil { + return fmt.Errorf("sharedkey: rewrite encryption domain: %w", err) + } + return nil +} + +// DiscoverKEKRefs strictly reads every committed manifest and returns the +// distinct encryption KEKRefs. Read/parse failures are fatal: treating an +// unreadable encrypted manifest as "no key exists" can mint a divergent DEK. +func DiscoverKEKRefs(ctx context.Context, sp storage.StoragePlugin) ([]string, error) { + set := map[string]struct{}{} + for _, prefix := range []string{"manifests/", "wal/"} { + for info, err := range sp.List(ctx, prefix) { + if err != nil { + return nil, fmt.Errorf("sharedkey: list %s: %w", prefix, err) + } + if !isManifestKey(prefix, info.Key) { + continue + } + env, err := readEnvelope(ctx, sp, info.Key) + if err != nil { + return nil, fmt.Errorf("sharedkey: read %s: %w", info.Key, err) + } + if env != nil && env.Scheme == Scheme && env.KEKRef != "" { + set[env.KEKRef] = struct{}{} + } + } + } + refs := make([]string, 0, len(set)) + for ref := range set { + refs = append(refs, ref) + } + sort.Strings(refs) + return refs, nil +} + // isManifestKey reports whether key under prefix is a manifest worth reading // for an envelope: a primary base-backup manifest, or a committed WAL segment // manifest (never a staging .tmp or a replica copy). @@ -127,26 +345,60 @@ func isManifestKey(prefix, key string) bool { // decoded wrapped-DEK bytes when the envelope is an aes-256-gcm wrap under // wantKEKRef. Any read / parse / mismatch yields ok=false so one bad or // unrelated object never aborts the scan. -func readWrappedDEK(ctx context.Context, sp storage.StoragePlugin, key, wantKEKRef string) ([]byte, bool) { +func readEnvelope(ctx context.Context, sp storage.StoragePlugin, key string) (*envelope, error) { rc, err := sp.Get(ctx, key) if err != nil { - return nil, false + return nil, err } defer rc.Close() body, err := io.ReadAll(rc) if err != nil { - return nil, false + return nil, err } var m manifestEnvelope if err := json.Unmarshal(body, &m); err != nil { - return nil, false + return nil, err } - if m.Encryption == nil || m.Encryption.Scheme != Scheme || m.Encryption.KEKRef != wantKEKRef { - return nil, false + return m.Encryption, nil +} + +func readDomain(ctx context.Context, sp storage.StoragePlugin) (domainRecord, bool, error) { + rc, err := sp.Get(ctx, DomainKey) + if errors.Is(err, storage.ErrNotFound) { + return domainRecord{}, false, nil } - wrapped, err := base64.StdEncoding.DecodeString(m.Encryption.WrappedDEK) if err != nil { - return nil, false + return domainRecord{}, false, fmt.Errorf("sharedkey: read encryption domain: %w", err) + } + defer rc.Close() + var rec domainRecord + if err := json.NewDecoder(io.LimitReader(rc, 1<<20)).Decode(&rec); err != nil { + return domainRecord{}, false, fmt.Errorf("sharedkey: decode encryption domain: %w", err) + } + validPlain := rec.Scheme == "none" && rec.KEKRef == "" && rec.WrappedDEK == "" + validEncrypted := rec.Scheme == Scheme && rec.KEKRef != "" && rec.WrappedDEK != "" + if rec.Schema != domainSchema || !validPlain && !validEncrypted { + return domainRecord{}, false, fmt.Errorf("sharedkey: invalid encryption domain record") + } + return rec, true, nil +} + +func unwrapDomain(rec domainRecord, wantKEKRef string, unwrap Unwrapper) ([encryption.KeyLen]byte, error) { + var out [encryption.KeyLen]byte + if rec.KEKRef != wantKEKRef { + return out, fmt.Errorf("sharedkey: repository encryption domain uses KEKRef %q; requested %q would create an undecryptable dedup collision", rec.KEKRef, wantKEKRef) + } + w, err := base64.StdEncoding.DecodeString(rec.WrappedDEK) + if err != nil { + return out, fmt.Errorf("sharedkey: decode domain wrapped DEK: %w", err) + } + dek, err := unwrap(w) + if err != nil { + return out, fmt.Errorf("sharedkey: unwrap domain DEK: %w", err) + } + if len(dek) != encryption.KeyLen { + return out, fmt.Errorf("sharedkey: domain DEK length %d, want %d", len(dek), encryption.KeyLen) } - return wrapped, true + copy(out[:], dek) + return out, nil } diff --git a/internal/repo/sharedkey/sharedkey_test.go b/internal/repo/sharedkey/sharedkey_test.go index 3a45340..691a740 100644 --- a/internal/repo/sharedkey/sharedkey_test.go +++ b/internal/repo/sharedkey/sharedkey_test.go @@ -8,6 +8,7 @@ import ( "fmt" "iter" "net/url" + "sync" "testing" "github.com/cybertec-postgresql/pg_hardstorage/internal/plugin/encryption" @@ -72,6 +73,57 @@ func mustWrap(t *testing.T, kek, dek [encryption.KeyLen]byte) []byte { return w } +func wrapperFor(kek [encryption.KeyLen]byte) func([encryption.KeyLen]byte) ([]byte, error) { + return func(dek [encryption.KeyLen]byte) ([]byte, error) { return encryption.Wrap(kek, dek) } +} + +func TestResolveOrCreate_ConcurrentFirstWritersConverge(t *testing.T) { + sp := newSP(t) + kek := testKEK(4) + var wg sync.WaitGroup + deks := make([][encryption.KeyLen]byte, 2) + errs := make([]error, 2) + for i := range deks { + wg.Add(1) + go func(i int) { + defer wg.Done() + deks[i], errs[i] = sharedkey.ResolveOrCreate(context.Background(), sp, + "local:default", unwrapperFor(kek), wrapperFor(kek)) + }(i) + } + wg.Wait() + for _, err := range errs { + if err != nil { + t.Fatalf("ResolveOrCreate: %v", err) + } + } + if deks[0] != deks[1] { + t.Fatalf("concurrent first writers diverged: %x != %x", deks[0], deks[1]) + } +} + +func TestResolveOrCreate_RejectsDifferentKEKDomain(t *testing.T) { + sp := newSP(t) + kekA, kekB := testKEK(1), testKEK(9) + if _, err := sharedkey.ResolveOrCreate(context.Background(), sp, "tenant:a", unwrapperFor(kekA), wrapperFor(kekA)); err != nil { + t.Fatal(err) + } + if _, err := sharedkey.ResolveOrCreate(context.Background(), sp, "tenant:b", unwrapperFor(kekB), wrapperFor(kekB)); err == nil { + t.Fatal("different KEKRef must not share the plaintext-addressed CAS namespace") + } +} + +func TestEncryptionDomain_RejectsPlaintextEncryptedMix(t *testing.T) { + sp := newSP(t) + if err := sharedkey.EnsurePlaintextAllowed(context.Background(), sp); err != nil { + t.Fatal(err) + } + kek := testKEK(3) + if _, err := sharedkey.ResolveOrCreate(context.Background(), sp, "local:default", unwrapperFor(kek), wrapperFor(kek)); err == nil { + t.Fatal("encrypted writer must not enter a repository claimed by plaintext writers") + } +} + // TestResolve_FindsInBackupManifest: the classic issue-#28 path — a wrapped // DEK on a base-backup manifest is found and unwrapped. func TestResolve_FindsInBackupManifest(t *testing.T) { diff --git a/internal/server/server.go b/internal/server/server.go index 918ce49..d990150 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -56,8 +56,8 @@ import ( // contract. const SchemaResult = "pg_hardstorage.server.v1" -// Config configures one server. Loaded from the agent config file's -// top-level `server:` block. +// Config configures one server. The CLI currently constructs it from +// explicit server flags; there is no top-level `server:` config-file block. type Config struct { // Listen is the bind address. Default: 127.0.0.1:8443. Set to // 0.0.0.0:8443 to expose externally.