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//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 }