Compare commits
2
Commits
d035e89631
...
87b1c16205
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
87b1c16205 | ||
|
|
d276d891da |
@@ -22,6 +22,23 @@ the tag exists and is exercised; what is left is merging `next` to
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- 2026-10-06: Made restore return an error instead of panicking on a
|
||||
malformed snapshot database
|
||||
([issue #231](https://git.eeqj.de/sneak/vaultik/issues/231)). A chunk
|
||||
hash shorter than 16 characters crashed the error message naming it,
|
||||
and `--verify` dereferenced a missing `chunks` row and allocated
|
||||
whatever chunk size the database gave. Those messages now go through
|
||||
`shortHash`, a missing row is an error, and `--verify` rejects a
|
||||
negative size and hashes each chunk as a stream.
|
||||
|
||||
- 2026-10-06: Made `s3://bucket/prefix` and `s3://bucket/prefix/` the same
|
||||
destination ([issue #222](https://git.eeqj.de/sneak/vaultik/issues/222)).
|
||||
The S3 client put the prefix directly in front of each key, so a prefix
|
||||
without a trailing slash stored `prefixblobs/...`. A non-empty prefix is
|
||||
now joined to every key with one `/`, giving the README's
|
||||
`<bucket>/<prefix>/blobs/...` layout. The `s3.prefix` config setting goes
|
||||
through the same client and gets the same join.
|
||||
|
||||
- 2026-10-06: Made restore apply owners, modes and times in an order
|
||||
that keeps them
|
||||
([issue #219](https://git.eeqj.de/sneak/vaultik/issues/219)). A
|
||||
|
||||
+11
-1
@@ -6,6 +6,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
@@ -30,6 +31,8 @@ type Client struct {
|
||||
|
||||
// Config contains S3 client configuration.
|
||||
// All fields are required except Prefix, which defaults to an empty string.
|
||||
// A non-empty Prefix is joined to every key with one "/", whether or not
|
||||
// it ends with one.
|
||||
// The Endpoint field should include the protocol (http:// or https://).
|
||||
type Config struct {
|
||||
Endpoint string
|
||||
@@ -75,10 +78,17 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
|
||||
|
||||
s3Client := s3.NewFromConfig(awsCfg, s3Opts)
|
||||
|
||||
// Every method below builds a key as prefix + key, so the prefix
|
||||
// must carry its own trailing "/".
|
||||
prefix := strings.TrimRight(cfg.Prefix, "/")
|
||||
if prefix != "" {
|
||||
prefix += "/"
|
||||
}
|
||||
|
||||
return &Client{
|
||||
s3Client: s3Client,
|
||||
bucket: cfg.Bucket,
|
||||
prefix: cfg.Prefix,
|
||||
prefix: prefix,
|
||||
endpoint: cfg.Endpoint,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -4,11 +4,14 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/johannesboyne/gofakes3"
|
||||
"github.com/johannesboyne/gofakes3/backend/s3mem"
|
||||
|
||||
"sneak.berlin/go/vaultik/internal/config"
|
||||
"sneak.berlin/go/vaultik/internal/s3"
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
)
|
||||
@@ -79,3 +82,109 @@ func TestS3StorerMissingKeyMapsToErrNotFound(t *testing.T) {
|
||||
t.Errorf("Stat on missing key: got %v, want ErrNotFound", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
|
||||
// writes: the README's remote storage layout, with the prefix joined to
|
||||
// each key by one "/". s3://b/p and s3://b/p/ must be the same
|
||||
// destination, or a host that writes the URL the other way finds no
|
||||
// snapshots. The listed object is put straight into the bucket, as
|
||||
// another host would have written it. Both List and ListStream are
|
||||
// checked: ListStream is what every snapshot listing goes through.
|
||||
func TestS3URLPrefixKeyLayout(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const (
|
||||
blobKey = "blobs/aa/bb/aabbccdd"
|
||||
listPrefix = "metadata/"
|
||||
manifestKey = listPrefix + "snap/manifest.json.zst"
|
||||
manifestBody = "manifest"
|
||||
)
|
||||
|
||||
cases := []struct {
|
||||
urlPath string // URL path after the bucket name
|
||||
keyPrefix string // what every key in the bucket must start with
|
||||
}{
|
||||
{urlPath: "/p", keyPrefix: "p/"},
|
||||
{urlPath: "/p/", keyPrefix: "p/"},
|
||||
{urlPath: "", keyPrefix: ""},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
storageURL := "s3://" + s3TestBucket + tc.urlPath
|
||||
|
||||
t.Run(storageURL, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
backend := s3mem.New()
|
||||
|
||||
err := backend.CreateBucket(s3TestBucket)
|
||||
if err != nil {
|
||||
t.Fatalf("create bucket: %v", err)
|
||||
}
|
||||
|
||||
srv := httptest.NewServer(gofakes3.New(backend).Server())
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
storer, err := storage.NewStorer(&config.Config{
|
||||
StorageURL: storageURL + "?endpoint=" + srv.URL,
|
||||
S3: config.S3Config{
|
||||
AccessKeyID: "key",
|
||||
SecretAccessKey: "secret",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("NewStorer: %v", err)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
err = storer.Put(ctx, blobKey, strings.NewReader("blob"))
|
||||
if err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
|
||||
_, err = backend.HeadObject(s3TestBucket, tc.keyPrefix+blobKey)
|
||||
if err != nil {
|
||||
t.Errorf("blob not stored at %q: %v", tc.keyPrefix+blobKey, err)
|
||||
}
|
||||
|
||||
_, err = backend.PutObject(s3TestBucket, tc.keyPrefix+manifestKey,
|
||||
nil, strings.NewReader(manifestBody), int64(len(manifestBody)))
|
||||
if err != nil {
|
||||
t.Fatalf("seed manifest: %v", err)
|
||||
}
|
||||
|
||||
keys, err := storer.List(ctx, listPrefix)
|
||||
if err != nil {
|
||||
t.Fatalf("List: %v", err)
|
||||
}
|
||||
|
||||
if !slices.Equal(keys, []string{manifestKey}) {
|
||||
t.Errorf("List(%q) = %q, want [%q]", listPrefix, keys, manifestKey)
|
||||
}
|
||||
|
||||
streamed := listStreamKeys(t, storer, listPrefix)
|
||||
if !slices.Equal(streamed, []string{manifestKey}) {
|
||||
t.Errorf("ListStream(%q) = %q, want [%q]", listPrefix, streamed, manifestKey)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// listStreamKeys returns the keys ListStream yields under a prefix, and
|
||||
// fails the test on a listing error.
|
||||
func listStreamKeys(t *testing.T, s storage.Storer, prefix string) []string {
|
||||
t.Helper()
|
||||
|
||||
var keys []string
|
||||
|
||||
for obj := range s.ListStream(context.Background(), prefix) {
|
||||
if obj.Err != nil {
|
||||
t.Fatalf("ListStream %q: %v", prefix, obj.Err)
|
||||
}
|
||||
|
||||
keys = append(keys, obj.Key)
|
||||
}
|
||||
|
||||
return keys
|
||||
}
|
||||
|
||||
+27
-15
@@ -42,6 +42,7 @@ var (
|
||||
errChunkNotInAnyBlob = errors.New("chunk not found in any blob")
|
||||
errBlobIDNotInHashIndex = errors.New("blob id missing from hash index")
|
||||
errShortChunkRead = errors.New("short read")
|
||||
errChunkRowMissing = errors.New("chunk has no row in the chunks table")
|
||||
errRestorePathEscapesTarget = errors.New(
|
||||
"refusing to restore path outside the target directory")
|
||||
errTrailingRestoreData = errors.New(
|
||||
@@ -1267,7 +1268,7 @@ func (s *restoreSession) writeFileChunks(
|
||||
blobChunk, ok := s.chunkToBlobMap[chunkHashStr]
|
||||
if !ok {
|
||||
return bytesWritten, timings, fmt.Errorf(
|
||||
"%w: %s", errChunkNotInAnyBlob, chunkHashStr[:16])
|
||||
"%w: %s", errChunkNotInAnyBlob, shortHash(chunkHashStr))
|
||||
}
|
||||
|
||||
blobHash, ok := s.blobIDToHash[blobChunk.BlobID.String()]
|
||||
@@ -1284,7 +1285,7 @@ func (s *restoreSession) writeFileChunks(
|
||||
if err != nil {
|
||||
return bytesWritten, timings, fmt.Errorf(
|
||||
"reading chunk %s from cached blob %s: %w",
|
||||
fc.ChunkHash[:16], blobHash[:16], err)
|
||||
shortHash(chunkHashStr), shortHash(blobHash), err)
|
||||
}
|
||||
|
||||
t0 = time.Now()
|
||||
@@ -1482,33 +1483,44 @@ func (v *Vaultik) verifyFile(
|
||||
chunk, err := repos.Chunks.GetByHash(ctx, fc.ChunkHash.String())
|
||||
if err != nil {
|
||||
return bytesVerified, fmt.Errorf("getting chunk %s: %w",
|
||||
fc.ChunkHash.String()[:16], err)
|
||||
shortHash(fc.ChunkHash.String()), err)
|
||||
}
|
||||
|
||||
// Read chunk data from file
|
||||
chunkData := make([]byte, chunk.Size)
|
||||
|
||||
n, err := io.ReadFull(f, chunkData)
|
||||
if err != nil {
|
||||
return bytesVerified, fmt.Errorf("reading chunk data: %w", err)
|
||||
if chunk == nil {
|
||||
return bytesVerified, fmt.Errorf("%w: %s",
|
||||
errChunkRowMissing, shortHash(fc.ChunkHash.String()))
|
||||
}
|
||||
|
||||
if int64(n) != chunk.Size {
|
||||
// chunk.Size comes from the snapshot database, which is not
|
||||
// trusted: reject a negative size, and hash the chunk by
|
||||
// streaming it rather than allocating that many bytes.
|
||||
if chunk.Size < 0 {
|
||||
return bytesVerified, fmt.Errorf("%w: chunk %d size %d",
|
||||
errNegativeChunkLength, fc.Idx, chunk.Size)
|
||||
}
|
||||
|
||||
hasher := sha256.New()
|
||||
|
||||
n, err := io.CopyN(hasher, f, chunk.Size)
|
||||
if errors.Is(err, io.EOF) {
|
||||
return bytesVerified, fmt.Errorf("%w: expected %d bytes, got %d",
|
||||
errShortChunkRead, chunk.Size, n)
|
||||
}
|
||||
|
||||
// Calculate hash and compare
|
||||
hash := sha256.Sum256(chunkData)
|
||||
actualHash := hex.EncodeToString(hash[:])
|
||||
if err != nil {
|
||||
return bytesVerified, fmt.Errorf("reading chunk data: %w", err)
|
||||
}
|
||||
|
||||
actualHash := hex.EncodeToString(hasher.Sum(nil))
|
||||
expectedHash := fc.ChunkHash.String()
|
||||
|
||||
if actualHash != expectedHash {
|
||||
return bytesVerified, fmt.Errorf("%w: chunk %d: expected %s, got %s",
|
||||
errChunkHashMismatch, fc.Idx, expectedHash[:16], actualHash[:16])
|
||||
errChunkHashMismatch, fc.Idx,
|
||||
shortHash(expectedHash), shortHash(actualHash))
|
||||
}
|
||||
|
||||
bytesVerified += int64(n)
|
||||
bytesVerified += n
|
||||
}
|
||||
|
||||
// The stored chunks account for the whole file, so the reader must
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
package vaultik //nolint:testpackage // drives unexported restore and verify steps
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/types"
|
||||
)
|
||||
|
||||
// These tests feed restore and --verify a snapshot database written by
|
||||
// hand, as a damaged or hostile store could serve one. Each malformed row
|
||||
// must end in an error, not a panic.
|
||||
|
||||
// shortChunkHash is shorter than the hash prefix that error messages print.
|
||||
const shortChunkHash = "abc"
|
||||
|
||||
// restoredFileContent is the content of the restored file under verify.
|
||||
const restoredFileContent = "xyz"
|
||||
|
||||
// craftedSnapshotDB opens an empty snapshot database in a temp directory.
|
||||
func craftedSnapshotDB(t *testing.T) (*database.DB, *database.Repositories) {
|
||||
t.Helper()
|
||||
|
||||
db, err := database.New(context.Background(),
|
||||
filepath.Join(t.TempDir(), "snapshot.db"))
|
||||
require.NoError(t, err)
|
||||
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
return db, database.NewRepositories(db)
|
||||
}
|
||||
|
||||
// craftedFile adds a regular file whose only chunk has the given hash.
|
||||
// Adding the chunks row, if any, is left to the caller.
|
||||
func craftedFile(
|
||||
t *testing.T, repos *database.Repositories, chunkHash string,
|
||||
) *database.File {
|
||||
t.Helper()
|
||||
|
||||
ctx := context.Background()
|
||||
file := &database.File{
|
||||
Path: "/src/f",
|
||||
MTime: time.Now().UTC(),
|
||||
Size: int64(len(restoredFileContent)),
|
||||
Mode: 0o644,
|
||||
}
|
||||
require.NoError(t, repos.Files.Create(ctx, nil, file))
|
||||
require.NoError(t, repos.FileChunks.Create(ctx, nil, &database.FileChunk{
|
||||
FileID: file.ID,
|
||||
ChunkHash: types.ChunkHash(chunkHash),
|
||||
}))
|
||||
|
||||
return file
|
||||
}
|
||||
|
||||
// TestRestoreShortChunkHashInNoBlob proves a file whose short chunk hash
|
||||
// has no blob_chunks row fails restore planning and the chunk write with
|
||||
// an error.
|
||||
func TestRestoreShortChunkHashInNoBlob(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
_, repos := craftedSnapshotDB(t)
|
||||
require.NoError(t, repos.Chunks.Create(ctx, nil,
|
||||
&database.Chunk{ChunkHash: shortChunkHash, Size: 3}))
|
||||
file := craftedFile(t, repos, shortChunkHash)
|
||||
|
||||
v := NewForTesting(nil)
|
||||
|
||||
chunkToBlobMap, err := v.buildChunkToBlobMap(ctx, repos)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = newRestorePlan(ctx, repos, []*database.File{file},
|
||||
chunkToBlobMap, map[string]string{})
|
||||
require.ErrorIs(t, err, errPlanChunkMissing)
|
||||
|
||||
fileChunks, err := repos.FileChunks.GetByFileID(ctx, file.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
out, err := afero.NewMemMapFs().Create("out")
|
||||
require.NoError(t, err)
|
||||
|
||||
session := &restoreSession{
|
||||
v: v.Vaultik, ctx: ctx, chunkToBlobMap: chunkToBlobMap,
|
||||
}
|
||||
|
||||
_, _, err = session.writeFileChunks(out, fileChunks)
|
||||
require.ErrorIs(t, err, errChunkNotInAnyBlob)
|
||||
}
|
||||
|
||||
// TestRestoreShortChunkHashReadPastBlobEnd proves a short chunk hash
|
||||
// whose blob_chunks row reads past the end of its blob fails the chunk
|
||||
// write with an error.
|
||||
func TestRestoreShortChunkHashReadPastBlobEnd(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
_, repos := craftedSnapshotDB(t)
|
||||
|
||||
blobHash := strings.Repeat("b", blobHashHexLen)
|
||||
blob := &database.Blob{
|
||||
ID: types.NewBlobID(),
|
||||
Hash: types.BlobHash(blobHash),
|
||||
CreatedTS: time.Now().UTC(),
|
||||
}
|
||||
require.NoError(t, repos.Blobs.Create(ctx, nil, blob))
|
||||
require.NoError(t, repos.Chunks.Create(ctx, nil,
|
||||
&database.Chunk{ChunkHash: shortChunkHash, Size: 3}))
|
||||
require.NoError(t, repos.BlobChunks.Create(ctx, nil, &database.BlobChunk{
|
||||
BlobID: blob.ID,
|
||||
ChunkHash: shortChunkHash,
|
||||
Length: 100,
|
||||
}))
|
||||
file := craftedFile(t, repos, shortChunkHash)
|
||||
|
||||
cache, err := newBlobDiskCache(1 << 20)
|
||||
require.NoError(t, err)
|
||||
|
||||
t.Cleanup(func() { _ = cache.Close() })
|
||||
|
||||
require.NoError(t, cache.Put(blobHash, []byte("abc")))
|
||||
|
||||
v := NewForTesting(nil)
|
||||
|
||||
chunkToBlobMap, err := v.buildChunkToBlobMap(ctx, repos)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, blobIDToHash, err := v.buildBlobIndexes(repos)
|
||||
require.NoError(t, err)
|
||||
|
||||
fileChunks, err := repos.FileChunks.GetByFileID(ctx, file.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
out, err := afero.NewMemMapFs().Create("out")
|
||||
require.NoError(t, err)
|
||||
|
||||
session := &restoreSession{
|
||||
v: v.Vaultik,
|
||||
ctx: ctx,
|
||||
chunkToBlobMap: chunkToBlobMap,
|
||||
blobIDToHash: blobIDToHash,
|
||||
blobCache: cache,
|
||||
}
|
||||
|
||||
_, _, err = session.writeFileChunks(out, fileChunks)
|
||||
require.ErrorIs(t, err, errCacheReadBeyondBlob)
|
||||
}
|
||||
|
||||
// TestVerifyFileMalformedChunkRow proves --verify returns an error for a
|
||||
// chunk with no chunks row, a short chunk hash, and a chunk size from the
|
||||
// database that is negative or larger than the restored file.
|
||||
func TestVerifyFileMalformedChunkRow(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
fullHash := types.ChunkHash(strings.Repeat("c", blobHashHexLen))
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
hash types.ChunkHash
|
||||
chunk *database.Chunk // nil adds no chunks row
|
||||
want error
|
||||
}{
|
||||
{
|
||||
name: "missing chunk row",
|
||||
hash: fullHash,
|
||||
want: errChunkRowMissing,
|
||||
},
|
||||
{
|
||||
name: "short hash",
|
||||
hash: shortChunkHash,
|
||||
chunk: &database.Chunk{ChunkHash: shortChunkHash, Size: 3},
|
||||
want: errChunkHashMismatch,
|
||||
},
|
||||
{
|
||||
name: "size larger than the file",
|
||||
hash: fullHash,
|
||||
chunk: &database.Chunk{ChunkHash: fullHash, Size: math.MaxInt64},
|
||||
want: errShortChunkRead,
|
||||
},
|
||||
{
|
||||
name: "negative size",
|
||||
hash: fullHash,
|
||||
chunk: &database.Chunk{ChunkHash: fullHash, Size: -1},
|
||||
want: errNegativeChunkLength,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := context.Background()
|
||||
db, repos := craftedSnapshotDB(t)
|
||||
|
||||
if tt.chunk == nil {
|
||||
// A crafted database need not satisfy its foreign keys.
|
||||
_, err := db.Conn().ExecContext(ctx, "PRAGMA foreign_keys = OFF")
|
||||
require.NoError(t, err)
|
||||
} else {
|
||||
require.NoError(t, repos.Chunks.Create(ctx, nil, tt.chunk))
|
||||
}
|
||||
|
||||
file := craftedFile(t, repos, tt.hash.String())
|
||||
|
||||
v := NewForTesting(nil)
|
||||
v.Fs = afero.NewMemMapFs()
|
||||
require.NoError(t, afero.WriteFile(v.Fs, "/restore/f",
|
||||
[]byte(restoredFileContent), 0o600))
|
||||
|
||||
_, err := v.verifyFile(ctx, repos, file, "/restore/f")
|
||||
require.ErrorIs(t, err, tt.want)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -75,7 +75,7 @@ func newRestorePlan(
|
||||
bc, ok := chunkToBlobMap[fc.ChunkHash.String()]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("planning %s: %w: %s",
|
||||
f.Path, errPlanChunkMissing, fc.ChunkHash.String()[:16])
|
||||
f.Path, errPlanChunkMissing, shortHash(fc.ChunkHash.String()))
|
||||
}
|
||||
|
||||
hash, ok := blobIDToHash[bc.BlobID.String()]
|
||||
|
||||
Reference in New Issue
Block a user