Objects fetched from the store are untrusted; several decode paths let one expand or print without limit. - blobgen.LimitReader errors past a byte cap (not io.LimitReader silent EOF). DecodeManifest reads through caps on both compressed input and decompressed output, far above any real manifest, so json.Decode cannot buffer a compressible bomb. FetchAndDecryptBlob bounds decompression to the blob recorded uncompressed_size (not the restoring host blob_size_limit). - downloadSnapshotDB streams straight from storage to its temp file with io.Copy, replacing two ReadAll calls that held the whole database twice. - FetchBlob drops the per-blob Stat round-trip, its expectedSize parameter and returned size, all of which only fed a debug log. - TTYHandler and ui.Writer escape control characters in messages, attribute keys/values, and rendered identifiers/paths before colour codes are applied, so a crafted value cannot drive the terminal. Model: opus-4-8
771 lines
22 KiB
Go
771 lines
22 KiB
Go
package vaultik
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"filippo.io/age"
|
|
|
|
"sneak.berlin/go/vaultik/internal/blobgen"
|
|
"sneak.berlin/go/vaultik/internal/database"
|
|
"sneak.berlin/go/vaultik/internal/log"
|
|
"sneak.berlin/go/vaultik/internal/snapshot"
|
|
)
|
|
|
|
// Sentinel errors for snapshot verification failures.
|
|
var (
|
|
errVerificationFailed = errors.New("verification failed")
|
|
errSecretKeyRequired = errors.New(
|
|
"VAULTIK_AGE_SECRET_KEY not set; required for deep verification")
|
|
errChunksOutOfOrder = errors.New("chunks out of order")
|
|
errChunkHashMismatch = errors.New("chunk hash mismatch")
|
|
errNegativeChunkLength = errors.New("chunk length is negative")
|
|
errTrailingBlobData = errors.New(
|
|
"blob has unexpected trailing bytes not covered by chunk list")
|
|
errManifestExtraBlob = errors.New("manifest contains blob not in database")
|
|
errManifestMissingBlob = errors.New(
|
|
"manifest omits blob present in database")
|
|
errBlobSizeMismatch = errors.New("blob size mismatch")
|
|
)
|
|
|
|
// verifyStatusFailed is the JSON status value for a failed verification.
|
|
const verifyStatusFailed = "failed"
|
|
|
|
// VerifyOptions contains options for the verify command
|
|
type VerifyOptions struct {
|
|
Deep bool
|
|
JSON bool
|
|
}
|
|
|
|
// VerifyResult contains the result of a snapshot verification
|
|
//
|
|
//nolint:tagliatelle // snake_case is the established JSON output format
|
|
type VerifyResult struct {
|
|
SnapshotID string `json:"snapshot_id"`
|
|
Status string `json:"status"` // "ok" or "failed"
|
|
Mode string `json:"mode"` // "shallow" or "deep"
|
|
BlobCount int `json:"blob_count"`
|
|
TotalSize int64 `json:"total_size"`
|
|
Verified int `json:"verified"`
|
|
Missing int `json:"missing"`
|
|
MissingSize int64 `json:"missing_size,omitempty"`
|
|
Mismatched int `json:"mismatched,omitempty"`
|
|
// DatabaseMissing is set by shallow verify when the snapshot's
|
|
// encrypted database (metadata/<key>/db.zst.age) is absent, which
|
|
// makes the snapshot unrestorable regardless of the blobs.
|
|
DatabaseMissing bool `json:"database_missing,omitempty"`
|
|
ErrorMessage string `json:"error,omitempty"`
|
|
}
|
|
|
|
// deepVerifyFailure records a failure in the result and returns it appropriately
|
|
func (v *Vaultik) deepVerifyFailure(
|
|
result *VerifyResult, opts *VerifyOptions, msg string, err error,
|
|
) error {
|
|
result.Status = verifyStatusFailed
|
|
|
|
result.ErrorMessage = msg
|
|
if opts.JSON {
|
|
return v.outputVerifyJSON(result)
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return fmt.Errorf("%w: %s", errVerificationFailed, msg)
|
|
}
|
|
|
|
// RunDeepVerify executes deep verification operation
|
|
func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
|
|
result := &VerifyResult{
|
|
SnapshotID: snapshotID,
|
|
Mode: "deep",
|
|
}
|
|
|
|
if !v.CanDecrypt() {
|
|
return v.deepVerifyFailure(result, opts,
|
|
errSecretKeyRequired.Error(), errSecretKeyRequired)
|
|
}
|
|
|
|
// Parse the age secret key once, the same way restore does, and reuse
|
|
// the identities for the database and every blob.
|
|
identities, err := v.restoreIdentities()
|
|
if err != nil {
|
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
|
}
|
|
|
|
log.Info("Starting snapshot verification", "snapshot_id", snapshotID, "mode", "deep")
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf("Deep verification of snapshot: %s\n\n", snapshotID)
|
|
}
|
|
|
|
manifest, tempDB, dbBlobs, err := v.loadVerificationData(
|
|
snapshotID, opts, result, identities)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
defer func() {
|
|
if tempDB != nil {
|
|
_ = tempDB.Close()
|
|
}
|
|
}()
|
|
|
|
result.BlobCount = len(dbBlobs)
|
|
|
|
var totalSize int64
|
|
for _, blob := range dbBlobs {
|
|
totalSize += blob.CompressedSize
|
|
}
|
|
|
|
result.TotalSize = totalSize
|
|
|
|
err = v.runVerificationSteps(
|
|
manifest, dbBlobs, tempDB, opts, result, totalSize, identities)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
result.Status = "ok"
|
|
result.Verified = len(dbBlobs)
|
|
|
|
if opts.JSON {
|
|
return v.outputVerifyJSON(result)
|
|
}
|
|
|
|
log.Info("✓ Verification completed successfully",
|
|
"snapshot_id", snapshotID, "mode", "deep", "blobs_verified", len(dbBlobs))
|
|
v.stdoutf("\n✓ Verification completed successfully\n")
|
|
v.stdoutf(" Snapshot: %s\n", snapshotID)
|
|
v.stdoutf(" Blobs verified: %d\n", len(dbBlobs))
|
|
v.stdoutf(" Total size: %s\n", ubytes(totalSize))
|
|
|
|
return nil
|
|
}
|
|
|
|
// loadVerificationData downloads manifest, database, and blob list for verification
|
|
func (v *Vaultik) loadVerificationData(
|
|
snapshotID string, opts *VerifyOptions, result *VerifyResult,
|
|
identities []age.Identity,
|
|
) (*snapshot.Manifest, *tempDB, []snapshot.BlobInfo, error) {
|
|
// Resolve the identifier to the snapshot's remote key. A human ID is
|
|
// hashed; a remote key (or its abbreviation, as printed for a
|
|
// remote-only snapshot) is used as-is, so a host with no local index
|
|
// can verify a snapshot it can only see on the store.
|
|
remoteKey, err := v.resolveSnapshotRemoteKey(snapshotID)
|
|
if err != nil {
|
|
return nil, nil, nil, v.deepVerifyFailure(result, opts,
|
|
fmt.Sprintf("resolving snapshot identifier: %v", err), err)
|
|
}
|
|
|
|
// Download manifest. downloadManifestByKey is the single reader for
|
|
// remote manifests; see its doc comment.
|
|
log.Info("Downloading manifest", "remote_key", remoteKey)
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf("Downloading manifest...\n")
|
|
}
|
|
|
|
manifest, err := v.downloadManifestByKey(remoteKey)
|
|
if err != nil {
|
|
return nil, nil, nil, v.deepVerifyFailure(result, opts,
|
|
fmt.Sprintf("failed to download manifest: %v", err),
|
|
fmt.Errorf("failed to download manifest: %w", err))
|
|
}
|
|
|
|
log.Info("Manifest loaded",
|
|
"manifest_blob_count", manifest.BlobCount,
|
|
"manifest_total_size", ubytes(manifest.TotalCompressedSize))
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf("Manifest loaded: %d blobs (%s)\n",
|
|
manifest.BlobCount, ubytes(manifest.TotalCompressedSize))
|
|
v.stdoutf("Downloading and decrypting database...\n")
|
|
}
|
|
|
|
tdb, err := v.downloadVerifiedSnapshotDB(
|
|
snapshotID, remoteKey, opts, result, identities)
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
|
|
dbBlobs, err := v.getBlobsFromDatabase(tdb.db.Conn())
|
|
if err != nil {
|
|
_ = tdb.Close()
|
|
|
|
return nil, nil, nil, v.deepVerifyFailure(result, opts,
|
|
fmt.Sprintf("failed to get blobs from database: %v", err),
|
|
fmt.Errorf("failed to get blobs from database: %w", err))
|
|
}
|
|
|
|
var dbTotalSize int64
|
|
for _, b := range dbBlobs {
|
|
dbTotalSize += b.CompressedSize
|
|
}
|
|
|
|
log.Info("Database loaded",
|
|
"db_blob_count", len(dbBlobs),
|
|
"db_total_size", ubytes(dbTotalSize))
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf("Database loaded: %d blobs (%s)\n",
|
|
len(dbBlobs), ubytes(dbTotalSize))
|
|
}
|
|
|
|
return manifest, tdb, dbBlobs, nil
|
|
}
|
|
|
|
// downloadVerifiedSnapshotDB downloads and decrypts the snapshot metadata
|
|
// database and confirms it really is the snapshot named by remoteKey
|
|
// before any of its rows are trusted (see verifySnapshotDBIdentity). On
|
|
// any failure it records the failure in result and returns the error the
|
|
// caller should propagate; the temp database is closed on a rejected
|
|
// identity so nothing is left on disk.
|
|
func (v *Vaultik) downloadVerifiedSnapshotDB(
|
|
snapshotID, remoteKey string, opts *VerifyOptions, result *VerifyResult,
|
|
identities []age.Identity,
|
|
) (*tempDB, error) {
|
|
dbPath := fmt.Sprintf("metadata/%s/db.zst.age", remoteKey)
|
|
log.Info("Downloading encrypted database", "path", dbPath)
|
|
|
|
dbReader, err := v.Storage.Get(v.ctx, dbPath)
|
|
if err != nil {
|
|
return nil, v.deepVerifyFailure(result, opts,
|
|
fmt.Sprintf("failed to download database: %v", err),
|
|
fmt.Errorf("failed to download database: %w", err))
|
|
}
|
|
|
|
defer func() { _ = dbReader.Close() }()
|
|
|
|
tdb, err := v.decryptAndLoadDatabase(dbReader, identities)
|
|
if err != nil {
|
|
return nil, v.deepVerifyFailure(result, opts,
|
|
fmt.Sprintf("failed to decrypt database: %v", err),
|
|
fmt.Errorf("failed to decrypt database: %w", err))
|
|
}
|
|
|
|
err = v.verifySnapshotDBIdentity(tdb.db, snapshotID, remoteKey)
|
|
if err != nil {
|
|
_ = tdb.Close()
|
|
|
|
return nil, v.deepVerifyFailure(result, opts, err.Error(), err)
|
|
}
|
|
|
|
return tdb, nil
|
|
}
|
|
|
|
// runVerificationSteps executes manifest verification, blob existence
|
|
// check, and deep content verification.
|
|
func (v *Vaultik) runVerificationSteps(
|
|
manifest *snapshot.Manifest,
|
|
dbBlobs []snapshot.BlobInfo,
|
|
tdb *tempDB,
|
|
opts *VerifyOptions,
|
|
result *VerifyResult,
|
|
totalSize int64,
|
|
identities []age.Identity,
|
|
) error {
|
|
if !opts.JSON {
|
|
v.stdoutf("Verifying manifest against database...\n")
|
|
}
|
|
|
|
err := v.verifyManifestAgainstDatabase(manifest, dbBlobs)
|
|
if err != nil {
|
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
|
}
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf("Manifest verified.\n")
|
|
v.stdoutf("Checking blob existence in remote storage...\n")
|
|
}
|
|
|
|
err = v.verifyBlobExistenceFromDB(dbBlobs)
|
|
if err != nil {
|
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
|
}
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf("All blobs exist.\n")
|
|
v.stdoutf("Downloading and verifying blob contents (%d blobs, %s)...\n",
|
|
len(dbBlobs), ubytes(totalSize))
|
|
}
|
|
|
|
err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
|
|
if err != nil {
|
|
return v.deepVerifyFailure(result, opts, err.Error(), err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// tempDB is the downloaded snapshot database opened read-only for deep
|
|
// verify, held in a private temp directory removed in full on Close.
|
|
type tempDB struct {
|
|
db *database.DB
|
|
tempDir string
|
|
}
|
|
|
|
func (t *tempDB) Close() error {
|
|
err := t.db.Close()
|
|
// Remove the whole private directory so the decrypted database and
|
|
// any SQLite side files are gone on every path.
|
|
_ = os.RemoveAll(t.tempDir)
|
|
|
|
return err
|
|
}
|
|
|
|
// decryptAndLoadDatabase decrypts and loads the binary SQLite database
|
|
// from the encrypted stream. It reads through the same blobgen reader restore
|
|
// uses, streaming the decrypted, decompressed database to a temp file.
|
|
func (v *Vaultik) decryptAndLoadDatabase(
|
|
reader io.ReadCloser, identities []age.Identity,
|
|
) (*tempDB, error) {
|
|
// Decrypt and decompress through the shared blobgen reader.
|
|
blobReader, err := blobgen.NewReader(reader, identities...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create decryption reader: %w", err)
|
|
}
|
|
|
|
defer func() { _ = blobReader.Close() }()
|
|
|
|
// Materialize the decrypted database inside a private (0700) temp
|
|
// directory so it is never world-readable, and remove the whole
|
|
// directory on any failure below.
|
|
tempDir, err := os.MkdirTemp("", "vaultik-verify-")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create temp directory: %w", err)
|
|
}
|
|
|
|
success := false
|
|
|
|
defer func() {
|
|
if !success {
|
|
_ = os.RemoveAll(tempDir)
|
|
}
|
|
}()
|
|
|
|
dbPath := filepath.Join(tempDir, snapshotDBFilename)
|
|
|
|
//nolint:gosec // G304: dbPath is our MkdirTemp dir plus a constant filename
|
|
tempFile, err := os.OpenFile(
|
|
dbPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, restoreFileMode)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create temp file: %w", err)
|
|
}
|
|
|
|
// Stream decompress directly to file
|
|
log.Info("Decompressing database...")
|
|
|
|
written, err := io.Copy(tempFile, blobReader)
|
|
if err != nil {
|
|
_ = tempFile.Close()
|
|
|
|
return nil, fmt.Errorf("failed to decompress database: %w", err)
|
|
}
|
|
|
|
err = tempFile.Close()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to close temp database file: %w", err)
|
|
}
|
|
|
|
log.Info("Database decompressed", "size", ubytes(written))
|
|
|
|
db, err := database.OpenReadOnly(v.ctx, dbPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to open database: %w", err)
|
|
}
|
|
|
|
success = true
|
|
|
|
return &tempDB{db: db, tempDir: tempDir}, nil
|
|
}
|
|
|
|
// verifyBlob downloads and verifies a single blob
|
|
func (v *Vaultik) verifyBlob(
|
|
blobInfo snapshot.BlobInfo, db *sql.DB, identities []age.Identity,
|
|
) error {
|
|
// Download blob using shared fetch method
|
|
reader, err := v.FetchBlob(v.ctx, blobInfo.Hash)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to download: %w", err)
|
|
}
|
|
|
|
defer func() { _ = reader.Close() }()
|
|
|
|
// Decrypt and decompress through the shared blobgen reader, which hashes
|
|
// the plaintext as it is read. A blob's hash — its remote name — is the
|
|
// double SHA-256 of that plaintext (see blobgen.DoubleSHA256), not of the
|
|
// encrypted bytes.
|
|
blobReader, err := blobgen.NewReader(reader, identities...)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to create blob reader: %w", err)
|
|
}
|
|
|
|
defer func() { _ = blobReader.Close() }()
|
|
|
|
chunkCount, err := v.verifyBlobChunks(db, blobInfo.Hash, blobReader)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = v.verifyBlobFinalIntegrity(blobReader, blobInfo.Hash)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
log.Info("Blob verified",
|
|
"hash", shortHash(blobInfo.Hash)+"...",
|
|
"chunks", chunkCount,
|
|
"size", ubytes(blobInfo.CompressedSize),
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
// verifyBlobChunks queries blob chunks from the database and verifies
|
|
// each chunk's hash against the decompressed blob stream.
|
|
func (v *Vaultik) verifyBlobChunks(
|
|
db *sql.DB, blobHash string, decompressor io.Reader,
|
|
) (int, error) {
|
|
query := `
|
|
SELECT bc.chunk_hash, bc.offset, bc.length
|
|
FROM blob_chunks bc
|
|
JOIN blobs b ON bc.blob_id = b.id
|
|
WHERE b.blob_hash = ?
|
|
ORDER BY bc.offset
|
|
`
|
|
|
|
rows, err := db.QueryContext(v.ctx, query, blobHash)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to query blob chunks: %w", err)
|
|
}
|
|
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var lastOffset int64 = -1
|
|
|
|
chunkCount := 0
|
|
totalRead := int64(0)
|
|
|
|
// Verify each chunk in the blob
|
|
for rows.Next() {
|
|
var (
|
|
chunkHash string
|
|
offset, length int64
|
|
)
|
|
|
|
err := rows.Scan(&chunkHash, &offset, &length)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to scan chunk row: %w", err)
|
|
}
|
|
|
|
// Verify chunk ordering
|
|
if offset <= lastOffset {
|
|
return 0, fmt.Errorf("%w: offset %d after %d",
|
|
errChunksOutOfOrder, offset, lastOffset)
|
|
}
|
|
|
|
lastOffset = offset
|
|
|
|
// Read chunk data from decompressed stream
|
|
if offset > totalRead {
|
|
// Skip to the correct offset
|
|
skipBytes := offset - totalRead
|
|
|
|
_, err = io.CopyN(io.Discard, decompressor, skipBytes)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to skip to offset %d: %w", offset, err)
|
|
}
|
|
|
|
totalRead = offset
|
|
}
|
|
|
|
// length comes from an untrusted blob_chunks row: reject a
|
|
// negative value, and hash by streaming exactly length bytes
|
|
// rather than allocating a database-supplied size up front.
|
|
if length < 0 {
|
|
return 0, fmt.Errorf("%w: offset %d length %d",
|
|
errNegativeChunkLength, offset, length)
|
|
}
|
|
|
|
hasher := sha256.New()
|
|
|
|
n, err := io.CopyN(hasher, decompressor, length)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to read chunk at offset %d: %w", offset, err)
|
|
}
|
|
|
|
totalRead += n
|
|
|
|
calculatedHash := hex.EncodeToString(hasher.Sum(nil))
|
|
if calculatedHash != chunkHash {
|
|
return 0, fmt.Errorf("%w at offset %d: calculated %s, expected %s",
|
|
errChunkHashMismatch, offset, calculatedHash, chunkHash)
|
|
}
|
|
|
|
chunkCount++
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("error iterating blob chunks: %w", err)
|
|
}
|
|
|
|
return chunkCount, nil
|
|
}
|
|
|
|
// verifyBlobFinalIntegrity checks that no trailing data exists in the
|
|
// decompressed stream and that the blob hash matches the expected value.
|
|
func (v *Vaultik) verifyBlobFinalIntegrity(
|
|
blobReader *blobgen.Reader, expectedHash string,
|
|
) error {
|
|
// Verify no remaining data in blob - if the chunk list is accurate,
|
|
// the blob should be fully consumed. Draining to EOF also completes the
|
|
// reader's plaintext hash.
|
|
remaining, err := io.Copy(io.Discard, blobReader)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to check for remaining blob data: %w", err)
|
|
}
|
|
|
|
if remaining > 0 {
|
|
return fmt.Errorf("%w: %d bytes", errTrailingBlobData, remaining)
|
|
}
|
|
|
|
// The blob hash is the double SHA-256 of its plaintext content.
|
|
calculatedBlobHash := hex.EncodeToString(
|
|
blobgen.DoubleSHA256(blobReader.Sum256()))
|
|
|
|
if calculatedBlobHash != expectedHash {
|
|
return fmt.Errorf("%w: calculated %s, expected %s",
|
|
errBlobHashMismatch, calculatedBlobHash, expectedHash)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// getBlobsFromDatabase gets all blobs for the snapshot from the database.
|
|
//
|
|
// The exported per-snapshot database holds exactly one snapshot's data
|
|
// (see cleanSnapshotDB), so every row in snapshot_blobs belongs to it.
|
|
// We select them directly rather than filtering by the human snapshot ID,
|
|
// which a host restoring from the store alone does not have.
|
|
func (v *Vaultik) getBlobsFromDatabase(db *sql.DB) ([]snapshot.BlobInfo, error) {
|
|
query := `
|
|
SELECT b.blob_hash, b.compressed_size
|
|
FROM snapshot_blobs sb
|
|
JOIN blobs b ON sb.blob_hash = b.blob_hash
|
|
ORDER BY b.blob_hash
|
|
`
|
|
|
|
rows, err := db.QueryContext(v.ctx, query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to query snapshot blobs: %w", err)
|
|
}
|
|
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var blobs []snapshot.BlobInfo
|
|
|
|
for rows.Next() {
|
|
var (
|
|
hash string
|
|
size int64
|
|
)
|
|
|
|
err := rows.Scan(&hash, &size)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to scan blob row: %w", err)
|
|
}
|
|
|
|
blobs = append(blobs, snapshot.BlobInfo{
|
|
Hash: hash,
|
|
CompressedSize: size,
|
|
})
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error iterating blobs: %w", err)
|
|
}
|
|
|
|
return blobs, nil
|
|
}
|
|
|
|
// verifyManifestAgainstDatabase verifies the manifest matches the
|
|
// authoritative database.
|
|
func (v *Vaultik) verifyManifestAgainstDatabase(
|
|
manifest *snapshot.Manifest, dbBlobs []snapshot.BlobInfo,
|
|
) error {
|
|
log.Info("Verifying manifest against database")
|
|
|
|
// Build map of database blobs
|
|
dbBlobMap := make(map[string]int64)
|
|
for _, blob := range dbBlobs {
|
|
dbBlobMap[blob.Hash] = blob.CompressedSize
|
|
}
|
|
|
|
// Build map of manifest blobs
|
|
manifestBlobMap := make(map[string]int64)
|
|
for _, blob := range manifest.Blobs {
|
|
manifestBlobMap[blob.Hash] = blob.CompressedSize
|
|
}
|
|
|
|
// The manifest is the only blob list prune consults, so it must match
|
|
// the database exactly. A blob in the manifest but not the database
|
|
// points at a corrupt manifest; a blob in the database but omitted
|
|
// from the manifest would be pruned away while this snapshot still
|
|
// needs it. Either divergence fails verification.
|
|
for hash, manifestSize := range manifestBlobMap {
|
|
dbSize, exists := dbBlobMap[hash]
|
|
if !exists {
|
|
return fmt.Errorf("%w: %s", errManifestExtraBlob, hash)
|
|
}
|
|
|
|
if dbSize != manifestSize {
|
|
return fmt.Errorf(
|
|
"%w: blob %s: database has %d bytes, manifest has %d bytes",
|
|
errBlobSizeMismatch, hash, dbSize, manifestSize)
|
|
}
|
|
}
|
|
|
|
for hash := range dbBlobMap {
|
|
if _, exists := manifestBlobMap[hash]; !exists {
|
|
return fmt.Errorf("%w: %s", errManifestMissingBlob, hash)
|
|
}
|
|
}
|
|
|
|
log.Info("✓ Manifest verified against database",
|
|
"manifest_blobs", len(manifestBlobMap),
|
|
"database_blobs", len(dbBlobMap),
|
|
)
|
|
|
|
return nil
|
|
}
|
|
|
|
// verifyBlobExistenceFromDB checks that all blobs from database exist in S3
|
|
func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
|
|
log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
|
|
|
|
for i, blob := range blobs {
|
|
// The hash is read from the snapshot database, which is not
|
|
// trusted; check it before it is spliced into a storage path.
|
|
if !isBlobHash(blob.Hash) {
|
|
return fmt.Errorf("%w: %s", errInvalidBlobHash, shortHash(blob.Hash))
|
|
}
|
|
|
|
// Construct blob path
|
|
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash)
|
|
|
|
// Check blob exists
|
|
stat, err := v.Storage.Stat(v.ctx, blobPath)
|
|
if err != nil {
|
|
return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
|
|
}
|
|
|
|
// Verify size matches
|
|
if stat.Size != blob.CompressedSize {
|
|
return fmt.Errorf(
|
|
"%w: blob %s: S3 has %d bytes, database has %d bytes",
|
|
errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
|
|
}
|
|
|
|
// Progress update every 100 blobs
|
|
if (i+1)%progressLogEvery == 0 || i == len(blobs)-1 {
|
|
log.Info("Blob existence check progress",
|
|
"checked", i+1,
|
|
"total", len(blobs),
|
|
"percent", fmt.Sprintf("%.1f%%",
|
|
float64(i+1)/float64(len(blobs))*percentScale),
|
|
)
|
|
}
|
|
}
|
|
|
|
log.Info("✓ All blobs exist in storage")
|
|
|
|
return nil
|
|
}
|
|
|
|
// performDeepVerificationFromDB downloads and verifies the content of
|
|
// each blob using the database as source.
|
|
func (v *Vaultik) performDeepVerificationFromDB(
|
|
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
|
|
identities []age.Identity,
|
|
) error {
|
|
// Calculate total bytes for ETA
|
|
var totalBytesExpected int64
|
|
for _, b := range blobs {
|
|
totalBytesExpected += b.CompressedSize
|
|
}
|
|
|
|
log.Info("Starting deep verification - downloading and verifying all blobs",
|
|
"blob_count", len(blobs),
|
|
"total_size", ubytes(totalBytesExpected),
|
|
)
|
|
|
|
startTime := time.Now()
|
|
bytesProcessed := int64(0)
|
|
|
|
for i, blobInfo := range blobs {
|
|
// Verify individual blob
|
|
err := v.verifyBlob(blobInfo, db, identities)
|
|
if err != nil {
|
|
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
|
|
}
|
|
|
|
bytesProcessed += blobInfo.CompressedSize
|
|
elapsed := time.Since(startTime)
|
|
remaining := len(blobs) - (i + 1)
|
|
|
|
// Calculate ETA based on bytes processed
|
|
var eta time.Duration
|
|
|
|
if bytesProcessed > 0 {
|
|
bytesPerSec := float64(bytesProcessed) / elapsed.Seconds()
|
|
|
|
bytesRemaining := totalBytesExpected - bytesProcessed
|
|
if bytesPerSec > 0 {
|
|
eta = time.Duration(float64(bytesRemaining)/bytesPerSec) * time.Second
|
|
}
|
|
}
|
|
|
|
log.Info("Verification progress",
|
|
"blobs_done", i+1,
|
|
"blobs_total", len(blobs),
|
|
"blobs_remaining", remaining,
|
|
"bytes_done", bytesProcessed,
|
|
"bytes_done_human", ubytes(bytesProcessed),
|
|
"bytes_total", totalBytesExpected,
|
|
"bytes_total_human", ubytes(totalBytesExpected),
|
|
"elapsed", elapsed.Round(time.Second),
|
|
"eta", eta.Round(time.Second),
|
|
)
|
|
|
|
if !opts.JSON {
|
|
v.stdoutf(" Verified %d/%d blobs (%d remaining) - %s/%s - elapsed %s, eta %s\n",
|
|
i+1, len(blobs), remaining,
|
|
ubytes(bytesProcessed),
|
|
ubytes(totalBytesExpected),
|
|
elapsed.Round(time.Second),
|
|
eta.Round(time.Second))
|
|
}
|
|
}
|
|
|
|
totalElapsed := time.Since(startTime)
|
|
log.Info("✓ Deep verification completed successfully",
|
|
"blobs_verified", len(blobs),
|
|
"total_bytes", bytesProcessed,
|
|
"total_bytes_human", ubytes(bytesProcessed),
|
|
"duration", totalElapsed.Round(time.Second),
|
|
)
|
|
|
|
return nil
|
|
}
|