Author SHA1 Message Date
sneak 8585bc87bb Refuse credentials in a storage URL's endpoint or rclone remote name (closes #322)
check / check (push) Canceled after 0s
storage_url refused only the outer URL's userinfo and unknown query
parameters. Userinfo inside the s3 endpoint= value, and rclone's
remote,param=value parameters in the remote name, were accepted and
then printed by info, remote info and snapshot create. Both are now
refused without echoing the value: a host name never contains "@", and
an rclone remote name never contains a comma. The comma is checked
before url.Parse, because url.Parse refuses a name such as
":s3,secret_access_key=..." or "x,secret=a:b", and its error prints
the name from the colon on. The rclone case has its own error pointing
at rclone config, since the existing credentials error names the s3.*
keys.

Model: opus-5-5
2026-10-13 18:38:25 +00:00
14 changed files with 142 additions and 504 deletions
+3 -24
View File
@@ -5,16 +5,7 @@
FROM golangci/golangci-lint:v2.14.0@sha256:ad862ba6b3798cbe0fd9fd7408d498fd74fbd2623a92406b2fd3898faf0bf98f AS lint FROM golangci/golangci-lint:v2.14.0@sha256:ad862ba6b3798cbe0fd9fd7408d498fd74fbd2623a92406b2fd3898faf0bf98f AS lint
WORKDIR /src WORKDIR /src
COPY go.mod go.sum ./ COPY go.mod go.sum ./
# Without the time limit a stalled connection to the module proxy holds RUN go mod download
# this step, and every build on the host sharing it, until the kernel
# drops the connection, which can take an hour. The limit is well above
# the three minutes a working download can take on a busy build host.
RUN for attempt in 1 2 3; do \
timeout 300 go mod download && exit 0; \
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
done; \
echo "module download failed: giving up after 3 attempts" >&2; \
exit 1
COPY . . COPY . .
# `golangci-lint run` silently ignores an unknown top-level key in # `golangci-lint run` silently ignores an unknown top-level key in
# .golangci.yml, such as a misspelt `linters:`; `config verify` fails on it. # .golangci.yml, such as a misspelt `linters:`; `config verify` fails on it.
@@ -27,13 +18,7 @@ RUN golangci-lint run --config .golangci.yml ./...
FROM golang:1.26.1@sha256:cd78d88e00afadbedd272f977d375a6247455f3a4b1178f8ae8bbcb201743a8a AS test FROM golang:1.26.1@sha256:cd78d88e00afadbedd272f977d375a6247455f3a4b1178f8ae8bbcb201743a8a AS test
WORKDIR /src WORKDIR /src
COPY go.mod go.sum ./ COPY go.mod go.sum ./
# Limited and retried as in the lint phase above. RUN go mod download
RUN for attempt in 1 2 3; do \
timeout 300 go mod download && exit 0; \
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
done; \
echo "module download failed: giving up after 3 attempts" >&2; \
exit 1
COPY . . COPY . .
RUN go test -timeout 90s -race -cover ./... || \ RUN go test -timeout 90s -race -cover ./... || \
{ echo "--- Rerunning with -v for details ---"; \ { echo "--- Rerunning with -v for details ---"; \
@@ -51,13 +36,7 @@ RUN apk add --no-cache git
RUN git config --system --add safe.directory /src RUN git config --system --add safe.directory /src
WORKDIR /src WORKDIR /src
COPY go.mod go.sum ./ COPY go.mod go.sum ./
# Limited and retried as in the lint phase above. RUN go mod download
RUN for attempt in 1 2 3; do \
timeout 300 go mod download && exit 0; \
echo "go mod download attempt $attempt failed or ran over 300s" >&2; \
done; \
echo "module download failed: giving up after 3 attempts" >&2; \
exit 1
COPY . . COPY . .
# The VERSION build arg when one is given, otherwise # The VERSION build arg when one is given, otherwise
+2 -5
View File
@@ -294,8 +294,7 @@ needed.
of the local index and the backup destination store — with timestamps of the local index and the backup destination store — with timestamps
and three sizes per snapshot (compressed remote size; total and three sizes per snapshot (compressed remote size; total
uncompressed chunk size; size of the chunks that snapshot stored that uncompressed chunk size; size of the chunks that snapshot stored that
were not stored before). Both chunk sizes are recorded when the were not stored before, recorded when its backup ran).
snapshot's backup runs.
Listing the destination store does **not** require the age secret key, Listing the destination store does **not** require the age secret key,
so it works in vaultik's intended configuration, where the backed-up so it works in vaultik's intended configuration, where the backed-up
@@ -339,9 +338,7 @@ local index alone, and still exits zero.
**`snapshot purge`**: Remove old snapshots based on criteria. Retention is **`snapshot purge`**: Remove old snapshots based on criteria. Retention is
per-snapshot-name (`--keep-latest` keeps the latest of each name, not the per-snapshot-name (`--keep-latest` keeps the latest of each name, not the
latest globally). If a snapshot cannot be removed from the local index, the latest globally).
command stops there, leaves that snapshot's metadata on the destination store
and exits non-zero; `snapshot create --prune` fails the same way.
* `--keep-latest`: Keep only the most recent snapshot of each name * `--keep-latest`: Keep only the most recent snapshot of each name
* `--older-than <duration>`: Remove snapshots older than duration (e.g. `30d`, * `--older-than <duration>`: Remove snapshots older than duration (e.g. `30d`,
`4w`, `6mo`, `1y`; `m` is minutes, `mo` is months) `4w`, `6mo`, `1y`; `m` is minutes, `mo` is months)
-45
View File
@@ -1,45 +0,0 @@
package main_test
import (
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestDockerfileBoundsEveryModuleDownload fails unless every
// `go mod download` in the Dockerfile runs under a time limit, is tried
// a fixed number of times, and then fails the build with a message that
// names the module download (issue #351). Like the version tests, this
// parses the Dockerfile; that a stalled download really is cut off and
// retried was checked by hand.
func TestDockerfileBoundsEveryModuleDownload(t *testing.T) {
t.Parallel()
var downloads []string
for _, instruction := range instructions(t, productDockerfile) {
if strings.Contains(instruction, "go mod download") {
downloads = append(downloads, instruction)
}
}
require.NotEmpty(t, downloads, "%s must download modules",
productDockerfile)
for _, download := range downloads {
assert.Contains(t, download, "timeout 300 go mod download",
"%s: a module download must have a time limit: %s",
productDockerfile, download)
assert.Contains(t, download, "for attempt in 1 2 3",
"%s: a module download must be retried: %s",
productDockerfile, download)
assert.Contains(t, download, "module download failed",
"%s: a module download that never succeeds must say so: %s",
productDockerfile, download)
assert.True(t, strings.HasSuffix(download, "exit 1"),
"%s: a module download that never succeeds must fail the"+
" build: %s", productDockerfile, download)
}
}
-1
View File
@@ -122,7 +122,6 @@ Tracks backup snapshots.
- `new_chunk_size` (INTEGER) - Total size of those chunks before compression; `snapshot list` shows it as NEW CHUNK SIZE - `new_chunk_size` (INTEGER) - Total size of those chunks before compression; `snapshot list` shows it as NEW CHUNK SIZE
- `blob_count` (INTEGER) - Number of blobs this snapshot created - `blob_count` (INTEGER) - Number of blobs this snapshot created
- `total_size` (INTEGER) - Total size of all files - `total_size` (INTEGER) - Total size of all files
- `uncompressed_chunk_size` (INTEGER) - Total size before compression of the distinct chunks the snapshot's files reference, recorded when its backup ran; `snapshot list` shows it as UNCOMPRESSED SIZE
- `blob_size` (INTEGER) - Total compressed size of all referenced blobs - `blob_size` (INTEGER) - Total compressed size of all referenced blobs
- `blob_uncompressed_size` (INTEGER) - Total uncompressed size of all referenced blobs - `blob_uncompressed_size` (INTEGER) - Total uncompressed size of all referenced blobs
- `compression_ratio` (REAL) - Compression ratio achieved - `compression_ratio` (REAL) - Compression ratio achieved
-90
View File
@@ -1,90 +0,0 @@
package cli //nolint:testpackage // shares runEntry and the hermetic config helpers
import (
"context"
"database/sql"
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sneak.berlin/go/vaultik/internal/database"
"sneak.berlin/go/vaultik/internal/snapshot"
"sneak.berlin/go/vaultik/internal/types"
)
// The `snapshot purge` command word and flags used below.
const (
cmdPurge = "purge"
flagKeepLatest = "--keep-latest"
flagForce = "--force"
)
// TestEntrySnapshotPurgeKeepsMetadataWhenLocalIndexDeleteFails makes
// removing a snapshot from the local index fail and checks that
// `snapshot purge` exits non-zero and leaves that snapshot's metadata on
// the destination store. It used to log the error, delete the metadata
// anyway and exit 0.
//
//nolint:paralleltest // replaces os.Args, os.Stdout, os.Stderr and the xdg globals
func TestEntrySnapshotPurgeKeepsMetadataWhenLocalIndexDeleteFails(t *testing.T) {
ctx := context.Background()
configPath := writeHermeticPruneConfig(t, false)
// writeHermeticPruneConfig puts the index and the store directory
// beside the config file.
dir := filepath.Dir(configPath)
// Two snapshots of the same name: --keep-latest purges the older.
olderStart := time.Date(2026, 5, 1, 9, 0, 0, 0, time.UTC)
newerStart := olderStart.Add(time.Hour)
olderID := "test-host_test_" + olderStart.Format(time.RFC3339)
newerID := "test-host_test_" + newerStart.Format(time.RFC3339)
db, err := database.New(ctx, filepath.Join(dir, "index.sqlite"))
require.NoError(t, err)
repos := database.NewRepositories(db)
manifestPaths := make(map[string]string)
for id, startedAt := range map[string]time.Time{
olderID: olderStart,
newerID: newerStart,
} {
completedAt := startedAt.Add(time.Minute)
snap := &database.Snapshot{
ID: types.SnapshotID(id),
Hostname: "test-host",
VaultikVersion: "test",
StartedAt: startedAt,
CompletedAt: &completedAt,
}
err = repos.WithTx(ctx, func(ctx context.Context, tx *sql.Tx) error {
return repos.Snapshots.Create(ctx, tx, snap)
})
require.NoError(t, err)
manifestPaths[id] = filepath.Join(dir, "store", "metadata",
snapshot.RemoteSnapshotKey(id), "manifest.json.zst")
require.NoError(t, os.MkdirAll(filepath.Dir(manifestPaths[id]), 0o750))
require.NoError(t, os.WriteFile(manifestPaths[id], []byte("stub"), 0o600))
}
_, err = db.Conn().ExecContext(ctx, `
CREATE TRIGGER refuse_snapshot_delete BEFORE DELETE ON snapshots
BEGIN SELECT RAISE(ABORT, 'simulated index error'); END`)
require.NoError(t, err)
require.NoError(t, db.Close())
code, _, stderr := runEntry(t,
flagConfig, configPath, cmdSnapshot, cmdPurge, flagKeepLatest, flagForce)
assert.Equal(t, 1, code)
assert.Contains(t, stderr, "simulated index error")
assert.FileExists(t, manifestPaths[olderID],
"a snapshot still in the local index keeps its metadata")
}
-1
View File
@@ -89,7 +89,6 @@ CREATE TABLE IF NOT EXISTS snapshots (
new_chunk_size INTEGER NOT NULL DEFAULT 0, new_chunk_size INTEGER NOT NULL DEFAULT 0,
blob_count INTEGER NOT NULL DEFAULT 0, blob_count INTEGER NOT NULL DEFAULT 0,
total_size INTEGER NOT NULL DEFAULT 0, total_size INTEGER NOT NULL DEFAULT 0,
uncompressed_chunk_size INTEGER NOT NULL DEFAULT 0,
blob_size INTEGER NOT NULL DEFAULT 0, blob_size INTEGER NOT NULL DEFAULT 0,
blob_uncompressed_size INTEGER NOT NULL DEFAULT 0, blob_uncompressed_size INTEGER NOT NULL DEFAULT 0,
compression_ratio REAL NOT NULL DEFAULT 1.0, compression_ratio REAL NOT NULL DEFAULT 1.0,
+17 -23
View File
@@ -120,11 +120,7 @@ func (r *SnapshotRepository) UpdateCounts(
return nil return nil
} }
// UpdateExtendedStats updates extended statistics for a snapshot. It also // UpdateExtendedStats updates extended statistics for a snapshot
// records the total plaintext size of the distinct chunks the snapshot's
// files reference. Only the snapshot's own backup can take that total: a
// file's row is shared by every snapshot that lists it, and a later
// backup of a changed file rewrites the row with the new chunks.
func (r *SnapshotRepository) UpdateExtendedStats( func (r *SnapshotRepository) UpdateExtendedStats(
ctx context.Context, ctx context.Context,
tx *sql.Tx, tx *sql.Tx,
@@ -149,17 +145,7 @@ func (r *SnapshotRepository) UpdateExtendedStats(
compression_level = ?, compression_level = ?,
upload_bytes = ?, upload_bytes = ?,
upload_duration_ms = ?, upload_duration_ms = ?,
new_chunk_size = ?, new_chunk_size = ?
uncompressed_chunk_size = (
SELECT COALESCE(SUM(c.size), 0)
FROM (
SELECT DISTINCT fc.chunk_hash
FROM snapshot_files sf
JOIN file_chunks fc ON sf.file_id = fc.file_id
WHERE sf.snapshot_id = snapshots.id
) sc
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
)
WHERE id = ? WHERE id = ?
` `
@@ -569,22 +555,30 @@ func (r *SnapshotRepository) GetSnapshotBlobSizes(
return compressed, uncompressed, nil return compressed, uncompressed, nil
} }
// GetSnapshotUncompressedChunkSize returns the total plaintext size of the // GetSnapshotUncompressedChunkSize returns the sum of plaintext sizes of all unique
// distinct chunks the snapshot's files reference, as the snapshot's backup // chunks referenced by a snapshot (via snapshot_files → file_chunks → chunks).
// recorded it (see UpdateExtendedStats).
func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize( func (r *SnapshotRepository) GetSnapshotUncompressedChunkSize(
ctx context.Context, snapshotID string, ctx context.Context, snapshotID string,
) (int64, error) { ) (int64, error) {
query := `SELECT uncompressed_chunk_size FROM snapshots WHERE id = ?` query := `
SELECT COALESCE(SUM(c.size), 0)
FROM (
SELECT DISTINCT fc.chunk_hash
FROM snapshot_files sf
JOIN file_chunks fc ON sf.file_id = fc.file_id
WHERE sf.snapshot_id = ?
) sc
JOIN chunks c ON sc.chunk_hash = c.chunk_hash
`
var uncompressedChunkSize int64 var totalSize int64
err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&uncompressedChunkSize) err := r.db.conn.QueryRowContext(ctx, query, snapshotID).Scan(&totalSize)
if err != nil { if err != nil {
return 0, fmt.Errorf("querying uncompressed chunk size: %w", err) return 0, fmt.Errorf("querying uncompressed chunk size: %w", err)
} }
return uncompressedChunkSize, nil return totalSize, nil
} }
// GetSnapshotNewChunkSize returns the total plaintext size of the chunks the // GetSnapshotNewChunkSize returns the total plaintext size of the chunks the
+32 -6
View File
@@ -29,7 +29,10 @@ var (
ErrURLCredentials = errors.New( ErrURLCredentials = errors.New(
"storage URL must not carry credentials; " + "storage URL must not carry credentials; " +
"set s3.access_key_id and s3.secret_access_key in the config instead") "set s3.access_key_id and s3.secret_access_key in the config instead")
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL") ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
ErrRcloneRemoteParams = errors.New(
"rclone URL must not carry parameters in the remote name; " +
"set them on the remote with rclone config instead")
) )
// URL represents a parsed storage URL. // URL represents a parsed storage URL.
@@ -90,8 +93,9 @@ func parseFileURL(rawPath string) (*URL, error) {
} }
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in // parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
// the userinfo and any query parameter other than endpoint, region and // the userinfo or in the endpoint parameter's userinfo, and any query
// ssl, so a credential-bearing URL is never stored or echoed. // parameter other than endpoint, region and ssl, so a credential-bearing
// URL is never stored or echoed.
func parseS3URL(rawURL string) (*URL, error) { func parseS3URL(rawURL string) (*URL, error) {
u, err := url.Parse(rawURL) u, err := url.Parse(rawURL)
if err != nil { if err != nil {
@@ -114,6 +118,14 @@ func parseS3URL(rawURL string) (*URL, error) {
return nil, err return nil, err
} }
// A host name never contains "@", so one in an endpoint, with or
// without a scheme, can only introduce userinfo.
for _, endpoint := range query["endpoint"] {
if strings.Contains(endpoint, "@") {
return nil, ErrURLCredentials
}
}
return &URL{ return &URL{
Scheme: schemeS3, Scheme: schemeS3,
Bucket: bucket, Bucket: bucket,
@@ -125,9 +137,23 @@ func parseS3URL(rawURL string) (*URL, error) {
} }
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no // parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
// query parameters, so credentials in the userinfo and any parameter at // query parameters, so credentials in the userinfo, parameters in the
// all are rejected rather than silently ignored. // remote name and any query parameter at all are rejected rather than
// silently ignored.
func parseRcloneURL(rawURL string) (*URL, error) { func parseRcloneURL(rawURL string) (*URL, error) {
// The remote name runs up to the first "/", "?" or "#". It never
// contains a comma; one starts rclone's remote,param=value form,
// whose values can be credentials. This is checked before url.Parse,
// whose error for such a name can print part of it.
remote := strings.TrimPrefix(rawURL, "rclone://")
if end := strings.IndexAny(remote, "/?#"); end >= 0 {
remote = remote[:end]
}
if strings.Contains(remote, ",") {
return nil, ErrRcloneRemoteParams
}
u, err := url.Parse(rawURL) u, err := url.Parse(rawURL)
if err != nil { if err != nil {
return nil, wrapParseError(err) return nil, wrapParseError(err)
@@ -137,7 +163,7 @@ func parseRcloneURL(rawURL string) (*URL, error) {
return nil, ErrURLCredentials return nil, ErrURLCredentials
} }
remote := u.Host remote = u.Host
if remote == "" { if remote == "" {
return nil, ErrMissingRemote return nil, ErrMissingRemote
} }
+61
View File
@@ -150,6 +150,19 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
wantErr: storage.ErrURLCredentials, wantErr: storage.ErrURLCredentials,
secrets: []string{key, secret, "mybucket"}, secrets: []string{key, secret, "mybucket"},
}, },
{
name: "s3 userinfo in endpoint",
raw: "s3://mybucket?endpoint=https://" +
key + ":" + secret + "@minio.example.com",
wantErr: storage.ErrURLCredentials,
secrets: []string{key, secret},
},
{
name: "s3 userinfo in endpoint without scheme",
raw: "s3://mybucket?endpoint=" + key + ":" + secret + "@minio.example.com",
wantErr: storage.ErrURLCredentials,
secrets: []string{key, secret},
},
{ {
name: "s3 unknown query param", name: "s3 unknown query param",
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret, raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
@@ -203,6 +216,54 @@ func TestParseStorageURLRejectsCredentials(t *testing.T) {
} }
} }
// TestParseStorageURLRejectsRcloneRemoteParams checks that an rclone
// remote name carrying parameters (rclone's "remote,param=value" form,
// which can hold a credential) is rejected without echoing the value,
// including the URLs that url.Parse itself refuses.
func TestParseStorageURLRejectsRcloneRemoteParams(t *testing.T) {
t.Parallel()
cases := []struct {
name string
raw string
secrets []string // must not appear in the error message
}{
{
name: "configured remote",
raw: "rclone://gdrive,client_secret=topsecret/backups",
secrets: []string{"topsecret"},
},
{
name: "remote defined inline",
raw: "rclone://:s3,access_key_id=KEY,secret_access_key=SECRET/bucket",
secrets: []string{"KEY", "SECRET"},
},
{
name: "colon in a parameter value",
raw: "rclone://gdrive,client_secret=ab:cd/backups",
secrets: []string{"cd"},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
_, err := storage.ParseStorageURL(tc.raw)
if !errors.Is(err, storage.ErrRcloneRemoteParams) {
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
tc.raw, err, storage.ErrRcloneRemoteParams)
}
for _, secret := range tc.secrets {
if strings.Contains(err.Error(), secret) {
t.Errorf("error message leaked %q: %v", secret, err.Error())
}
}
})
}
}
// TestParseStorageURLParseFailureHidesURL checks that when url.Parse // TestParseStorageURLParseFailureHidesURL checks that when url.Parse
// itself fails, the wrapped error carries only the inner cause, not the // itself fails, the wrapped error carries only the inner cause, not the
// *url.Error whose text embeds the raw (possibly credential-bearing) URL. // *url.Error whose text embeds the raw (possibly credential-bearing) URL.
-172
View File
@@ -1,9 +1,7 @@
package vaultik_test package vaultik_test
import ( import (
"bytes"
"context" "context"
"crypto/rand"
"io" "io"
"os" "os"
"path/filepath" "path/filepath"
@@ -11,9 +9,7 @@ import (
"github.com/spf13/afero" "github.com/spf13/afero"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"sneak.berlin/go/vaultik/internal/config"
"sneak.berlin/go/vaultik/internal/log" "sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/storage"
"sneak.berlin/go/vaultik/internal/ui" "sneak.berlin/go/vaultik/internal/ui"
"sneak.berlin/go/vaultik/internal/vaultik" "sneak.berlin/go/vaultik/internal/vaultik"
) )
@@ -81,174 +77,6 @@ func TestDeepVerifyAcceptsHealthyAndRejectsCorruptBlob(t *testing.T) {
"deep verify should fail on a corrupted blob") "deep verify should fail on a corrupted blob")
} }
// TestDeepVerifyJSONCountsBlobs checks the verified and missing counts
// that deep verify reports under --json: on a healthy snapshot, with a
// damaged blob, and with missing blobs. Deep verify stops at the first
// damaged blob, so the blobs read before it are the ones verified; when
// blobs are missing it reads no blob contents, but counts every missing
// blob.
func TestDeepVerifyJSONCountsBlobs(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
fs := afero.NewOsFs()
ctx := context.Background()
storeDir := filepath.Join(t.TempDir(), "remote")
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
healthy, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.NoError(t, err)
require.Equal(t, "ok", healthy.Status)
require.Len(t, keys, healthy.BlobCount)
require.Equal(t, healthy.BlobCount, healthy.Verified)
require.Zero(t, healthy.Missing)
// Damage the last blob read, keeping its length, so every blob before
// it is read and passes first.
lastPath := filepath.Join(storeDir, keys[len(keys)-1])
blob, err := afero.ReadFile(fs, lastPath)
require.NoError(t, err)
blob[len(blob)/2] ^= 0xff
require.NoError(t, afero.WriteFile(fs, lastPath, blob, 0o600))
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.Error(t, err)
require.Equal(t, "failed", got.Status)
require.Equal(t, healthy.BlobCount-1, got.Verified)
require.Zero(t, got.Missing)
// Delete the first blob and the damaged last one: both must be
// counted as missing, not only the first one found.
var deletedSize int64
for _, key := range []string{keys[0], keys[len(keys)-1]} {
info, statErr := storer.Stat(ctx, key)
require.NoError(t, statErr)
deletedSize += info.Size
require.NoError(t, storer.Delete(ctx, key))
}
got, err = deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.Error(t, err)
require.Equal(t, "failed", got.Status)
require.Equal(t, 2, got.Missing)
require.Equal(t, deletedSize, got.MissingSize)
require.Zero(t, got.Verified)
}
// TestDeepVerifyJSONCountsWrongSizeBlob checks that under --deep --json a
// blob of the wrong size is counted in mismatched and fails the run, and
// that the check goes on past it to count a missing blob after it.
func TestDeepVerifyJSONCountsWrongSizeBlob(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
fs := afero.NewOsFs()
ctx := context.Background()
storeDir := filepath.Join(t.TempDir(), "remote")
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
// Make the first blob checked longer, and delete the last one.
firstPath := filepath.Join(storeDir, keys[0])
blob, err := afero.ReadFile(fs, firstPath)
require.NoError(t, err)
blob = append(blob, []byte("extra")...)
require.NoError(t, afero.WriteFile(fs, firstPath, blob, 0o600))
require.NoError(t, storer.Delete(ctx, keys[len(keys)-1]))
got, err := deepVerifyJSON(ctx, t, cfg, storer, snapshotID)
require.Error(t, err)
require.Equal(t, "failed", got.Status)
require.Equal(t, 1, got.Mismatched)
require.Equal(t, 1, got.Missing)
require.Zero(t, got.Verified)
}
// TestDeepVerifyStopsOnBlobThatCannotBeChecked checks that a blob that is
// present in storage but cannot be checked is not counted as missing:
// deep verify stops and fails with the storage error.
func TestDeepVerifyStopsOnBlobThatCannotBeChecked(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
ctx := context.Background()
storeDir := filepath.Join(t.TempDir(), "remote")
cfg, storer, snapshotID, keys := backUpSeveralBlobs(ctx, t, storeDir)
failing := statFailingStorer{Storer: storer, failKeys: []string{keys[0]}}
got, err := deepVerifyJSON(ctx, t, cfg, failing, snapshotID)
require.Error(t, err)
require.Equal(t, "failed", got.Status)
require.Zero(t, got.Missing)
require.Contains(t, got.ErrorMessage, errStatDenied.Error())
}
// backUpSeveralBlobs backs up random data, which does not compress or
// deduplicate, so it fills at least three blobs in a store on disk at
// storeDir. It returns the blobs' storage keys sorted, which is hash
// order, the order deep verify reads them in.
func backUpSeveralBlobs(
ctx context.Context, t *testing.T, storeDir string,
) (*config.Config, *storage.FileStorer, string, []string) {
t.Helper()
fs := afero.NewOsFs()
tempDir := t.TempDir()
dataDir := filepath.Join(tempDir, "source")
dbPath := filepath.Join(tempDir, "index.sqlite")
chunkSize := int64(32 * 1024)
maxBlobSize := int64(128 * 1024)
data := make([]byte, maxBlobSize*4)
_, err := rand.Read(data)
require.NoError(t, err)
require.NoError(t, fs.MkdirAll(dataDir, 0o755))
require.NoError(t, afero.WriteFile(fs,
filepath.Join(dataDir, "data.bin"), data, 0o644))
cfg, storer, snapshotID := runFileStorageBackup(
ctx, t, fs, dataDir, storeDir, dbPath, chunkSize, maxBlobSize)
keys := storedBlobKeys(t, storeDir)
require.GreaterOrEqual(t, len(keys), 3)
return cfg, storer, snapshotID, keys
}
// deepVerifyJSON runs deep verify with --json on snapshotID and returns
// the decoded result and the error RunDeepVerify returned.
func deepVerifyJSON(
ctx context.Context, t *testing.T,
cfg *config.Config, storer storage.Storer, snapshotID string,
) (vaultik.VerifyResult, error) {
t.Helper()
var out bytes.Buffer
v := &vaultik.Vaultik{
Config: cfg,
Storage: storer,
Fs: afero.NewOsFs(),
Stdout: &out,
Stderr: io.Discard,
UI: ui.NewWithColor(io.Discard, false),
}
v.SetContext(ctx)
err := v.RunDeepVerify(snapshotID,
&vaultik.VerifyOptions{Deep: true, JSON: true})
return decodeVerifyResult(t, out.Bytes()), err
}
// corruptOneBlob flips a middle byte of the first blob file found under // corruptOneBlob flips a middle byte of the first blob file found under
// blobsDir, leaving the file length unchanged. // blobsDir, leaving the file length unchanged.
func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) { func corruptOneBlob(t *testing.T, fs afero.Fs, blobsDir string) {
+3 -5
View File
@@ -627,8 +627,7 @@ func selectSnapshotsToPurge(
} }
// confirmAndExecutePurge shows deletion candidates, confirms with the // confirmAndExecutePurge shows deletion candidates, confirms with the
// user, and deletes snapshots. It stops with an error at the first // user, and deletes snapshots.
// snapshot it cannot delete.
func (v *Vaultik) confirmAndExecutePurge( func (v *Vaultik) confirmAndExecutePurge(
toDelete []SnapshotInfo, force, quiet bool, toDelete []SnapshotInfo, force, quiet bool,
) error { ) error {
@@ -670,11 +669,10 @@ func (v *Vaultik) confirmAndExecutePurge(
snapshotID := snap.ID.String() snapshotID := snap.ID.String()
log.Info("Deleting snapshot", "id", snapshotID) log.Info("Deleting snapshot", "id", snapshotID)
// Stop before the remote delete: a snapshot still in the local
// index must keep its metadata on the destination store.
err := v.deleteSnapshotFromLocalDB(snapshotID) err := v.deleteSnapshotFromLocalDB(snapshotID)
if err != nil { if err != nil {
return fmt.Errorf("deleting snapshot %s from local index: %w", snapshotID, err) log.Error("Failed to delete from local database",
"snapshot_id", snapshotID, "error", err)
} }
_, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID)) _, err = v.deleteRemoteSnapshotByKey(snapshot.RemoteSnapshotKey(snapshotID))
+7 -8
View File
@@ -483,14 +483,13 @@ func TestListSnapshots_UnreadableManifestDoesNotHideOthers(t *testing.T) {
// //
//nolint:tagliatelle // snake_case is the established output format //nolint:tagliatelle // snake_case is the established output format
type listJSONRow struct { type listJSONRow struct {
ID string `json:"id"` ID string `json:"id"`
RemoteKey string `json:"remote_key"` RemoteKey string `json:"remote_key"`
Timestamp string `json:"timestamp"` Timestamp string `json:"timestamp"`
CompressedSize int64 `json:"compressed_size"` CompressedSize int64 `json:"compressed_size"`
UncompressedSize int64 `json:"uncompressed_size"` NewChunkSize int64 `json:"new_chunk_size"`
NewChunkSize int64 `json:"new_chunk_size"` LocallyTracked bool `json:"locally_tracked"`
LocallyTracked bool `json:"locally_tracked"` RemotePresent *bool `json:"remote_present"`
RemotePresent *bool `json:"remote_present"`
} }
// decodeListJSON parses the command's stdout, which must contain // decodeListJSON parses the command's stdout, which must contain
@@ -1,85 +0,0 @@
package vaultik_test
import (
"bytes"
"context"
"path/filepath"
"testing"
"github.com/spf13/afero"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sneak.berlin/go/vaultik/internal/database"
"sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/storage"
"sneak.berlin/go/vaultik/internal/vaultik"
)
// TestUncompressedSizeOfChangedFileStaysFixed covers
// https://git.eeqj.de/sneak/vaultik/issues/360. A backup that changes a
// file leaves the earlier snapshot's UNCOMPRESSED SIZE as it was when that
// snapshot's backup completed. The file's row in the local index is shared
// by both snapshots and rewritten by the second backup.
func TestUncompressedSizeOfChangedFileStaysFixed(t *testing.T) {
log.Initialize(log.Config{})
t.Parallel()
fs := afero.NewOsFs()
tempDir := t.TempDir()
dataDir := filepath.Join(tempDir, "src")
dbPath := filepath.Join(tempDir, "index.sqlite")
changedPath := filepath.Join(dataDir, "changed.bin")
ctx := context.Background()
files := writeFaultSourceTree(t, fs, dataDir)
cfg := changedFileConfig(dataDir, dbPath)
appendRandomBytes(t, fs, files, changedPath, appendedFileSize)
store, err := storage.NewFileStorer(filepath.Join(tempDir, "remote"))
require.NoError(t, err)
db, err := database.New(ctx, dbPath)
require.NoError(t, err)
t.Cleanup(func() { _ = db.Close() })
repos := database.NewRepositories(db)
v := newBackupVaultik(ctx, cfg, store, repos, db, fs)
require.NoError(t, backUp(v, "first"))
// With nothing to compare against, sizeOfNewChunks totals every
// distinct chunk of the files.
firstID := localSnapshotID(ctx, t, repos, "first")
firstSize := sizeOfNewChunks(t, files, nil)
require.Equal(t, firstSize, listedUncompressedSizes(t, v)[firstID],
"the first snapshot must count the chunks of every file it holds")
appendRandomBytes(t, fs, files, changedPath, faultChunkSize)
require.NoError(t, backUp(v, "second"))
secondID := localSnapshotID(ctx, t, repos, "second")
sizes := listedUncompressedSizes(t, v)
assert.Equal(t, firstSize, sizes[firstID],
"a later backup must not change the first snapshot's figure")
assert.Equal(t, sizeOfNewChunks(t, files, nil), sizes[secondID],
"the second snapshot must count the changed file's chunks")
}
// listedUncompressedSizes runs `snapshot list --json` and returns each
// listed snapshot's uncompressed_size by snapshot ID.
func listedUncompressedSizes(t *testing.T, v *vaultik.Vaultik) map[string]int64 {
t.Helper()
var out bytes.Buffer
v.Stdout = &out
require.NoError(t, v.ListSnapshots(true))
sizes := make(map[string]int64)
for _, row := range decodeListJSON(t, out.String()) {
sizes[row.ID] = row.UncompressedSize
}
return sizes
}
+17 -39
View File
@@ -17,7 +17,6 @@ import (
"sneak.berlin/go/vaultik/internal/database" "sneak.berlin/go/vaultik/internal/database"
"sneak.berlin/go/vaultik/internal/log" "sneak.berlin/go/vaultik/internal/log"
"sneak.berlin/go/vaultik/internal/snapshot" "sneak.berlin/go/vaultik/internal/snapshot"
"sneak.berlin/go/vaultik/internal/storage"
) )
// Sentinel errors for snapshot verification failures. // Sentinel errors for snapshot verification failures.
@@ -33,9 +32,7 @@ var (
errManifestExtraBlob = errors.New("manifest contains blob not in database") errManifestExtraBlob = errors.New("manifest contains blob not in database")
errManifestMissingBlob = errors.New( errManifestMissingBlob = errors.New(
"manifest omits blob present in database") "manifest omits blob present in database")
errBlobSizeMismatch = errors.New("blob size mismatch") errBlobSizeMismatch = errors.New("blob size mismatch")
errBlobsNotInStorage = errors.New(
"blobs missing from storage or stored with the wrong size")
) )
// verifyStatusFailed is the JSON status value for a failed verification. // verifyStatusFailed is the JSON status value for a failed verification.
@@ -152,6 +149,7 @@ func (v *Vaultik) RunDeepVerify(snapshotID string, opts *VerifyOptions) error {
} }
result.Status = "ok" result.Status = "ok"
result.Verified = len(dbBlobs)
if opts.JSON { if opts.JSON {
return v.outputVerifyJSON(result) return v.outputVerifyJSON(result)
@@ -308,7 +306,7 @@ func (v *Vaultik) runVerificationSteps(
v.stdoutf("Checking blob existence in remote storage...\n") v.stdoutf("Checking blob existence in remote storage...\n")
} }
err = v.verifyBlobExistenceFromDB(dbBlobs, result) err = v.verifyBlobExistenceFromDB(dbBlobs)
if err != nil { if err != nil {
return v.deepVerifyFailure(result, opts, err.Error(), err) return v.deepVerifyFailure(result, opts, err.Error(), err)
} }
@@ -319,8 +317,7 @@ func (v *Vaultik) runVerificationSteps(
len(dbBlobs), ubytes(totalSize)) len(dbBlobs), ubytes(totalSize))
} }
err = v.performDeepVerificationFromDB( err = v.performDeepVerificationFromDB(dbBlobs, tdb.db.Conn(), opts, identities)
dbBlobs, tdb.db.Conn(), opts, result, identities)
if err != nil { if err != nil {
return v.deepVerifyFailure(result, opts, err.Error(), err) return v.deepVerifyFailure(result, opts, err.Error(), err)
} }
@@ -672,14 +669,8 @@ func (v *Vaultik) verifyManifestAgainstDatabase(
return nil return nil
} }
// verifyBlobExistenceFromDB checks that every blob from the database is // verifyBlobExistenceFromDB checks that all blobs from database exist in S3
// in storage with the size the database records. It checks them all, func (v *Vaultik) verifyBlobExistenceFromDB(blobs []snapshot.BlobInfo) error {
// counting the missing and wrong-size blobs in result, and fails if it
// found any. Any storage error other than not found stops the check and
// fails with that error.
func (v *Vaultik) verifyBlobExistenceFromDB(
blobs []snapshot.BlobInfo, result *VerifyResult,
) error {
log.Info("Verifying blob existence in S3", "blob_count", len(blobs)) log.Info("Verifying blob existence in S3", "blob_count", len(blobs))
for i, blob := range blobs { for i, blob := range blobs {
@@ -692,23 +683,17 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
// Construct blob path // Construct blob path
blobPath := fmt.Sprintf("blobs/%s/%s/%s", blob.Hash[:2], blob.Hash[2:4], blob.Hash) 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) stat, err := v.Storage.Stat(v.ctx, blobPath)
switch { if err != nil {
case errors.Is(err, storage.ErrNotFound): return fmt.Errorf("blob %s missing from storage: %w", blob.Hash, err)
log.Warn("Blob missing from storage", "hash", blob.Hash) }
result.Missing++ // Verify size matches
result.MissingSize += blob.CompressedSize if stat.Size != blob.CompressedSize {
case err != nil: return fmt.Errorf(
return fmt.Errorf("checking blob %s in storage: %w", "%w: blob %s: S3 has %d bytes, database has %d bytes",
blob.Hash, err) errBlobSizeMismatch, blob.Hash, stat.Size, blob.CompressedSize)
case stat.Size != blob.CompressedSize:
log.Warn("Blob has the wrong size in storage",
"hash", blob.Hash,
"storage_size", stat.Size,
"database_size", blob.CompressedSize)
result.Mismatched++
} }
// Progress update every 100 blobs // Progress update every 100 blobs
@@ -722,22 +707,16 @@ func (v *Vaultik) verifyBlobExistenceFromDB(
} }
} }
if result.Missing > 0 || result.Mismatched > 0 {
return fmt.Errorf("%w: %d missing, %d with the wrong size",
errBlobsNotInStorage, result.Missing, result.Mismatched)
}
log.Info("✓ All blobs exist in storage") log.Info("✓ All blobs exist in storage")
return nil return nil
} }
// performDeepVerificationFromDB downloads and verifies the content of // performDeepVerificationFromDB downloads and verifies the content of
// each blob using the database as source, stopping at the first blob that // each blob using the database as source.
// fails. Each blob that passes is counted in result.Verified.
func (v *Vaultik) performDeepVerificationFromDB( func (v *Vaultik) performDeepVerificationFromDB(
blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions, blobs []snapshot.BlobInfo, db *sql.DB, opts *VerifyOptions,
result *VerifyResult, identities []age.Identity, identities []age.Identity,
) error { ) error {
// Calculate total bytes for ETA // Calculate total bytes for ETA
var totalBytesExpected int64 var totalBytesExpected int64
@@ -760,7 +739,6 @@ func (v *Vaultik) performDeepVerificationFromDB(
return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err) return fmt.Errorf("blob %s verification failed: %w", blobInfo.Hash, err)
} }
result.Verified++
bytesProcessed += blobInfo.CompressedSize bytesProcessed += blobInfo.CompressedSize
elapsed := time.Since(startTime) elapsed := time.Since(startTime)
remaining := len(blobs) - (i + 1) remaining := len(blobs) - (i + 1)