Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8e335008cb |
@@ -167,7 +167,7 @@ vaultik version
|
||||
* `--verbose`, `-v`: Enable verbose output (on stderr — see below)
|
||||
* `--debug`: Enable debug output (on stderr — see below)
|
||||
* `--quiet`, `-q`: Suppress non-error output (also suppresses startup banner)
|
||||
* `--skip-errors`: Skip files that cannot be read when creating a snapshot, or that cannot be restored when restoring, instead of aborting. Packing and storage errors (which would leave a chunk recorded but not stored) still abort the run.
|
||||
* `--skip-errors`: Continue past per-file errors instead of aborting (applies to `snapshot create` and `restore`)
|
||||
|
||||
### locking
|
||||
|
||||
|
||||
+12
-26
@@ -4,7 +4,6 @@ import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
@@ -380,23 +379,12 @@ Examples:
|
||||
return err
|
||||
}
|
||||
|
||||
return writeConfigSet(os.Stdout, path, args[0], args[1])
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// writeConfigSet applies key=value to the config at path, writes it back
|
||||
// owner-only, and confirms the write by printing just the key name to w.
|
||||
// The value is never echoed: it may be a secret such as
|
||||
// s3.secret_access_key, and captured stdout or a pasted terminal would
|
||||
// then leak it.
|
||||
func writeConfigSet(w io.Writer, path, key, value string) error {
|
||||
root, err := loadYAMLFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = yamlPathSet(root, strings.Split(key, "."), value)
|
||||
err = yamlPathSet(root, strings.Split(args[0], "."), args[1])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -406,25 +394,23 @@ func writeConfigSet(w io.Writer, path, key, value string) error {
|
||||
return fmt.Errorf("marshaling config: %w", err)
|
||||
}
|
||||
|
||||
err = os.WriteFile(path, out, configFileMode)
|
||||
mode := os.FileMode(configFileMode)
|
||||
|
||||
info, statErr := os.Stat(path)
|
||||
if statErr == nil {
|
||||
mode = info.Mode().Perm()
|
||||
}
|
||||
|
||||
err = os.WriteFile(path, out, mode)
|
||||
if err != nil {
|
||||
return fmt.Errorf("writing config file: %w", err)
|
||||
}
|
||||
|
||||
// os.WriteFile does not change the mode of a file that already exists,
|
||||
// so a config that was group- or world-readable stays that way. As it
|
||||
// may hold S3 credentials, tighten it to owner-only after writing.
|
||||
info, statErr := os.Stat(path)
|
||||
if statErr == nil && info.Mode().Perm()&0o044 != 0 {
|
||||
err = os.Chmod(path, configFileMode)
|
||||
if err != nil {
|
||||
return fmt.Errorf("tightening config file permissions: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
_, _ = fmt.Fprintln(w, key)
|
||||
_, _ = fmt.Fprintf(os.Stdout, "%s = %s\n", args[0], args[1])
|
||||
|
||||
return nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// marshalConfigYAML renders a config document tree with 2-space indentation,
|
||||
|
||||
@@ -1,9 +1,6 @@
|
||||
package cli //nolint:testpackage // exercises unexported yamlPathGet/yamlPathSet
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -232,68 +229,6 @@ func TestConfigSetPreservesFormatting(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestWriteConfigSetHidesSecret checks that setting a secret key prints
|
||||
// only the key name, never the value, to the confirmation output.
|
||||
func TestWriteConfigSetHidesSecret(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const secret = "SUPERSECRETVALUE"
|
||||
|
||||
path := filepath.Join(t.TempDir(), "config.yaml")
|
||||
|
||||
err := os.WriteFile(path, []byte("version: 1\n"), 0o600)
|
||||
if err != nil {
|
||||
t.Fatalf("seed config: %v", err)
|
||||
}
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
err = writeConfigSet(&out, path, "s3.secret_access_key", secret)
|
||||
if err != nil {
|
||||
t.Fatalf("writeConfigSet: %v", err)
|
||||
}
|
||||
|
||||
if strings.Contains(out.String(), secret) {
|
||||
t.Errorf("output echoed the secret value: %q", out.String())
|
||||
}
|
||||
|
||||
if !strings.Contains(out.String(), "s3.secret_access_key") {
|
||||
t.Errorf("output did not confirm the key name: %q", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
// TestWriteConfigSetTightensMode checks that a pre-existing group- or
|
||||
// world-readable config is tightened to owner-only after a set, since
|
||||
// os.WriteFile leaves an existing file's mode untouched.
|
||||
func TestWriteConfigSetTightensMode(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
path := filepath.Join(t.TempDir(), "config.yaml")
|
||||
|
||||
// Seed a world-readable config; the loose mode is the condition under
|
||||
// test, so gosec's G306 is expected here.
|
||||
err := os.WriteFile(path, []byte("version: 1\n"), 0o644) //nolint:gosec // G306
|
||||
if err != nil {
|
||||
t.Fatalf("seed config: %v", err)
|
||||
}
|
||||
|
||||
var out bytes.Buffer
|
||||
|
||||
err = writeConfigSet(&out, path, "compression_level", "9")
|
||||
if err != nil {
|
||||
t.Fatalf("writeConfigSet: %v", err)
|
||||
}
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatalf("stat config: %v", err)
|
||||
}
|
||||
|
||||
if info.Mode().Perm() != 0o600 {
|
||||
t.Errorf("config mode = %04o, want 0600", info.Mode().Perm())
|
||||
}
|
||||
}
|
||||
|
||||
func splitPath(s string) []string {
|
||||
return strings.Split(s, ".")
|
||||
}
|
||||
|
||||
@@ -57,9 +57,8 @@ on the source system.`,
|
||||
cmd.PersistentFlags().BoolVarP(&rootFlags.Quiet, "quiet", "q", false,
|
||||
"Suppress non-error output")
|
||||
cmd.PersistentFlags().BoolVar(&rootFlags.SkipErrors, "skip-errors", false,
|
||||
"Skip files that cannot be read when creating a snapshot, or "+
|
||||
"that cannot be restored when restoring, instead of aborting "+
|
||||
"(packing and storage errors still abort)")
|
||||
"Continue past per-file errors instead of aborting "+
|
||||
"(applies to snapshot create and restore)")
|
||||
|
||||
// Add subcommands
|
||||
cmd.AddCommand(
|
||||
|
||||
@@ -63,9 +63,7 @@ type Scanner struct {
|
||||
exclude []string // Glob patterns for files/directories to exclude
|
||||
compiledExclude []compiledPattern // Compiled glob patterns
|
||||
progress *ProgressReporter
|
||||
// skipErrors skips files that cannot be opened or read (logged loudly);
|
||||
// packer, database, encryption, and upload errors still abort the run.
|
||||
skipErrors bool
|
||||
skipErrors bool // Skip file read errors (log loudly but continue)
|
||||
// ui is the user-facing output; never nil (defaults to a discarding writer).
|
||||
ui *ui.Writer
|
||||
|
||||
@@ -123,9 +121,7 @@ type ScannerConfig struct {
|
||||
EnableProgress bool // Enable the live progress reporter (ETAs, throughput)
|
||||
UI *ui.Writer // Where user-facing scanner messages go; nil = discard
|
||||
Exclude []string // Glob patterns for files/directories to exclude
|
||||
// SkipErrors skips files that cannot be opened or read (log loudly but
|
||||
// continue); packer, database, encryption, and upload errors still abort.
|
||||
SkipErrors bool
|
||||
SkipErrors bool // Skip file read errors (log loudly but continue)
|
||||
}
|
||||
|
||||
// ScanResult contains the results of a scan operation
|
||||
@@ -1340,15 +1336,6 @@ func (s *Scanner) processFileWithErrorHandling(
|
||||
) (bool, error) {
|
||||
err := s.processFileStreaming(ctx, fileToProcess, result)
|
||||
if err != nil {
|
||||
// A packer/database/encryption/upload failure means the chunk's data
|
||||
// may not have been stored. Skipping the file would let the snapshot
|
||||
// record a file whose chunk is in no blob and cannot be restored, so
|
||||
// abort the run even under --skip-errors. Only open and read errors
|
||||
// are skipped below.
|
||||
var pErr *packerError
|
||||
if errors.As(err, &pErr) {
|
||||
return false, fmt.Errorf("processing file %s: %w", fileToProcess.Path, err)
|
||||
}
|
||||
// Handle files that were deleted between scan and process phases
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
log.Warn("File was deleted during backup, skipping",
|
||||
@@ -1358,7 +1345,7 @@ func (s *Scanner) processFileWithErrorHandling(
|
||||
|
||||
return true, nil
|
||||
}
|
||||
// Skip open/read errors if --skip-errors is enabled
|
||||
// Skip file read errors if --skip-errors is enabled
|
||||
if s.skipErrors {
|
||||
log.Error("Failed to process file (skipping due to --skip-errors)",
|
||||
"path", fileToProcess.Path, "error", err)
|
||||
@@ -1725,20 +1712,6 @@ type streamingChunkInfo struct {
|
||||
size int64
|
||||
}
|
||||
|
||||
// packerError marks an error that came from adding a chunk to the packer
|
||||
// (packing, database, encryption, or upload). Such an error means the chunk's
|
||||
// data may not have been stored, so the run must abort even under --skip-errors:
|
||||
// skipping the file would leave the chunk recorded as backed up while it lives
|
||||
// in no blob, and a later snapshot could record a file that cannot be restored.
|
||||
// Only open and read errors are safe to skip.
|
||||
type packerError struct {
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *packerError) Error() string { return e.err.Error() }
|
||||
|
||||
func (e *packerError) Unwrap() error { return e.err }
|
||||
|
||||
// processFileStreaming processes a file by streaming chunks directly to the packer
|
||||
func (s *Scanner) processFileStreaming(
|
||||
ctx context.Context, fileToProcess *FileToProcess, result *ScanResult,
|
||||
@@ -1789,11 +1762,7 @@ func (s *Scanner) processFileStreaming(
|
||||
if !chunkExists {
|
||||
err := s.addChunkToPacker(ctx, chunk)
|
||||
if err != nil {
|
||||
// Mark as a packer error so --skip-errors cannot swallow it:
|
||||
// the chunk was registered as pending before packing, so a
|
||||
// skipped file here would be recorded as backed up while its
|
||||
// data was never stored.
|
||||
return &packerError{err: err}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,216 +0,0 @@
|
||||
package snapshot_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/spf13/afero"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
)
|
||||
|
||||
// errSimTempFail is the one-time temp-file creation failure blobTempFailFs
|
||||
// injects, mirroring a full temp filesystem.
|
||||
var errSimTempFail = errors.New("simulated temp-file creation failure")
|
||||
|
||||
// errSimRead is the read failure readFailFile injects for a file that opens
|
||||
// but cannot be read.
|
||||
var errSimRead = errors.New("simulated read failure")
|
||||
|
||||
// blobTempFailFs fails the first temp-file creation for a packer blob, then
|
||||
// behaves normally, simulating a one-time failure to start a new blob.
|
||||
type blobTempFailFs struct {
|
||||
afero.Fs
|
||||
|
||||
mu sync.Mutex
|
||||
failed bool
|
||||
}
|
||||
|
||||
//nolint:ireturn // afero.Fs.OpenFile is defined to return the interface.
|
||||
func (f *blobTempFailFs) OpenFile(
|
||||
name string, flag int, perm os.FileMode,
|
||||
) (afero.File, error) {
|
||||
if strings.Contains(name, "vaultik-blob-") {
|
||||
f.mu.Lock()
|
||||
firstTime := !f.failed
|
||||
f.failed = true
|
||||
f.mu.Unlock()
|
||||
|
||||
if firstTime {
|
||||
return nil, errSimTempFail
|
||||
}
|
||||
}
|
||||
|
||||
return f.Fs.OpenFile(name, flag, perm)
|
||||
}
|
||||
|
||||
// readFailFile wraps an afero.File whose Read always fails.
|
||||
type readFailFile struct {
|
||||
afero.File
|
||||
}
|
||||
|
||||
func (readFailFile) Read([]byte) (int, error) {
|
||||
return 0, errSimRead
|
||||
}
|
||||
|
||||
// readFailFs fails reads of one target path after a successful open.
|
||||
type readFailFs struct {
|
||||
afero.Fs
|
||||
|
||||
target string
|
||||
}
|
||||
|
||||
//nolint:ireturn // afero.Fs.Open is defined to return the interface.
|
||||
func (f *readFailFs) Open(name string) (afero.File, error) {
|
||||
file, err := f.Fs.Open(name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if name == f.target {
|
||||
return readFailFile{File: file}, nil
|
||||
}
|
||||
|
||||
return file, nil
|
||||
}
|
||||
|
||||
// writeSkipErrorTestFile writes one file into fs with a fixed mtime.
|
||||
func writeSkipErrorTestFile(t *testing.T, fs afero.Fs, path, content string) {
|
||||
t.Helper()
|
||||
|
||||
err := fs.MkdirAll(filepath.Dir(path), 0755)
|
||||
if err != nil {
|
||||
t.Fatalf("mkdir: %v", err)
|
||||
}
|
||||
|
||||
err = afero.WriteFile(fs, path, []byte(content), 0644)
|
||||
if err != nil {
|
||||
t.Fatalf("write %s: %v", path, err)
|
||||
}
|
||||
|
||||
when := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
err = fs.Chtimes(path, when, when)
|
||||
if err != nil {
|
||||
t.Fatalf("chtimes %s: %v", path, err)
|
||||
}
|
||||
}
|
||||
|
||||
// runSkipErrorScan scans /source on fs with the given skip-errors setting and
|
||||
// returns the repositories (for inspection) and the scan error.
|
||||
func runSkipErrorScan(
|
||||
t *testing.T, fs afero.Fs, skipErrors bool,
|
||||
) (*database.Repositories, error) {
|
||||
t.Helper()
|
||||
|
||||
db, err := database.NewTestDB()
|
||||
if err != nil {
|
||||
t.Fatalf("create test db: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() {
|
||||
cerr := db.Close()
|
||||
if cerr != nil {
|
||||
t.Errorf("close db: %v", cerr)
|
||||
}
|
||||
})
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
|
||||
scanner := snapshot.NewScanner(snapshot.ScannerConfig{
|
||||
FS: fs,
|
||||
ChunkSize: int64(1024 * 16),
|
||||
Repositories: repos,
|
||||
MaxBlobSize: int64(1024 * 1024),
|
||||
CompressionLevel: 3,
|
||||
AgeRecipients: []string{testAgePublicKey},
|
||||
SkipErrors: skipErrors,
|
||||
})
|
||||
|
||||
ctx := context.Background()
|
||||
snapshotID := "test-snapshot-skip-errors"
|
||||
createTestSnapshotRecord(ctx, t, repos, snapshotID)
|
||||
|
||||
_, err = scanner.Scan(ctx, "/source", snapshotID)
|
||||
|
||||
return repos, err
|
||||
}
|
||||
|
||||
// TestScannerPackingFailureAbortsUnderSkipErrors checks that a failure to start
|
||||
// a new blob aborts the run even with --skip-errors. Otherwise the file would
|
||||
// be skipped while its chunk had already been registered as pending, letting a
|
||||
// later blob record that chunk in the chunks table with no blob to back it —
|
||||
// a snapshot that completes with a file that cannot be restored.
|
||||
func TestScannerPackingFailureAbortsUnderSkipErrors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Two files with distinct content so each yields a distinct chunk: the
|
||||
// first fails to start a blob, and without the fix the second's blob would
|
||||
// commit the first's orphaned chunk row.
|
||||
fs := &blobTempFailFs{Fs: afero.NewMemMapFs()}
|
||||
writeSkipErrorTestFile(t, fs, "/source/file1.txt", "first file content")
|
||||
writeSkipErrorTestFile(t, fs, "/source/file2.txt", "second file content")
|
||||
|
||||
repos, err := runSkipErrorScan(t, fs, true)
|
||||
if err == nil {
|
||||
t.Fatal("expected scan to abort on the packer error, got nil")
|
||||
}
|
||||
|
||||
// ListUnpacked returns chunks recorded with no blob_chunks row: exactly the
|
||||
// unrestorable state this fix prevents.
|
||||
unpacked, err := repos.Chunks.ListUnpacked(context.Background(), 10)
|
||||
if err != nil {
|
||||
t.Fatalf("listing unpacked chunks: %v", err)
|
||||
}
|
||||
|
||||
if len(unpacked) != 0 {
|
||||
t.Fatalf("expected no chunk recorded without a blob, got %d", len(unpacked))
|
||||
}
|
||||
}
|
||||
|
||||
// TestScannerReadErrorAbortsWithoutSkipErrors checks that a file read error
|
||||
// aborts the run when --skip-errors is not set.
|
||||
func TestScannerReadErrorAbortsWithoutSkipErrors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const target = "/source/unreadable.txt"
|
||||
|
||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||
|
||||
_, err := runSkipErrorScan(t, fs, false)
|
||||
if err == nil {
|
||||
t.Fatal("expected scan to fail on the read error, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestScannerReadErrorSkippedWithSkipErrors checks that a file read error is
|
||||
// skipped and the run completes when --skip-errors is set.
|
||||
func TestScannerReadErrorSkippedWithSkipErrors(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const target = "/source/unreadable.txt"
|
||||
|
||||
fs := &readFailFs{Fs: afero.NewMemMapFs(), target: target}
|
||||
writeSkipErrorTestFile(t, fs, target, "content that cannot be read")
|
||||
|
||||
repos, err := runSkipErrorScan(t, fs, true)
|
||||
if err != nil {
|
||||
t.Fatalf("expected scan to complete with --skip-errors, got %v", err)
|
||||
}
|
||||
|
||||
chunks, err := repos.FileChunks.GetByFile(context.Background(), target)
|
||||
if err != nil {
|
||||
t.Fatalf("getting file chunks: %v", err)
|
||||
}
|
||||
|
||||
if len(chunks) != 0 {
|
||||
t.Fatalf("expected unreadable file skipped, got %d chunks", len(chunks))
|
||||
}
|
||||
}
|
||||
+15
-70
@@ -4,7 +4,6 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -24,10 +23,6 @@ var (
|
||||
ErrUnsupportedScheme = errors.New(
|
||||
"unsupported URL scheme: must start with s3://, file://, or rclone://")
|
||||
ErrUnsupportedStorage = errors.New("unsupported storage scheme")
|
||||
ErrURLCredentials = errors.New(
|
||||
"storage URL must not carry credentials; " +
|
||||
"set s3.access_key_id and s3.secret_access_key in the config instead")
|
||||
ErrURLUnknownParam = errors.New("unknown query parameter in storage URL")
|
||||
)
|
||||
|
||||
// URL represents a parsed storage URL.
|
||||
@@ -64,28 +59,11 @@ func ParseStorageURL(rawURL string) (*URL, error) {
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Handle s3:// URLs
|
||||
if strings.HasPrefix(rawURL, "s3://") {
|
||||
return parseS3URL(rawURL)
|
||||
}
|
||||
|
||||
if strings.HasPrefix(rawURL, "rclone://") {
|
||||
return parseRcloneURL(rawURL)
|
||||
}
|
||||
|
||||
return nil, ErrUnsupportedScheme
|
||||
}
|
||||
|
||||
// parseS3URL parses an s3://bucket/prefix URL. It rejects credentials in
|
||||
// the userinfo and any query parameter other than endpoint, region and
|
||||
// ssl, so a credential-bearing URL is never stored or echoed.
|
||||
func parseS3URL(rawURL string) (*URL, error) {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return nil, wrapParseError(err)
|
||||
}
|
||||
|
||||
if u.User != nil {
|
||||
return nil, ErrURLCredentials
|
||||
return nil, fmt.Errorf("invalid URL: %w", err)
|
||||
}
|
||||
|
||||
bucket := u.Host
|
||||
@@ -93,34 +71,30 @@ func parseS3URL(rawURL string) (*URL, error) {
|
||||
return nil, ErrMissingBucket
|
||||
}
|
||||
|
||||
prefix := strings.TrimPrefix(u.Path, "/")
|
||||
|
||||
query := u.Query()
|
||||
|
||||
err = rejectUnknownParams(query, "endpoint", "region", "ssl")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
useSSL := true
|
||||
if query.Get("ssl") == "false" {
|
||||
useSSL = false
|
||||
}
|
||||
|
||||
return &URL{
|
||||
Scheme: schemeS3,
|
||||
Bucket: bucket,
|
||||
Prefix: strings.TrimPrefix(u.Path, "/"),
|
||||
Prefix: prefix,
|
||||
Endpoint: query.Get("endpoint"),
|
||||
Region: query.Get("region"),
|
||||
UseSSL: query.Get("ssl") != "false",
|
||||
UseSSL: useSSL,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// parseRcloneURL parses an rclone://remote/path URL. rclone:// takes no
|
||||
// query parameters, so credentials in the userinfo and any parameter at
|
||||
// all are rejected rather than silently ignored.
|
||||
func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
// Handle rclone:// URLs
|
||||
if strings.HasPrefix(rawURL, "rclone://") {
|
||||
u, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return nil, wrapParseError(err)
|
||||
}
|
||||
|
||||
if u.User != nil {
|
||||
return nil, ErrURLCredentials
|
||||
return nil, fmt.Errorf("invalid URL: %w", err)
|
||||
}
|
||||
|
||||
remote := u.Host
|
||||
@@ -128,45 +102,16 @@ func parseRcloneURL(rawURL string) (*URL, error) {
|
||||
return nil, ErrMissingRemote
|
||||
}
|
||||
|
||||
err = rejectUnknownParams(u.Query())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
path := strings.TrimPrefix(u.Path, "/")
|
||||
|
||||
return &URL{
|
||||
Scheme: schemeRclone,
|
||||
Prefix: strings.TrimPrefix(u.Path, "/"),
|
||||
Prefix: path,
|
||||
RcloneRemote: remote,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// rejectUnknownParams returns an error naming the first query parameter
|
||||
// not in allowed. The parameter's name is included (so a misspelt
|
||||
// endpoint= is caught), but never its value, which could be a secret,
|
||||
// and never the whole URL.
|
||||
func rejectUnknownParams(query url.Values, allowed ...string) error {
|
||||
for name := range query {
|
||||
if !slices.Contains(allowed, name) {
|
||||
return fmt.Errorf(
|
||||
"%w: %q; put credentials in s3.access_key_id and "+
|
||||
"s3.secret_access_key, not the URL",
|
||||
ErrURLUnknownParam, name)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// wrapParseError wraps only the inner cause of a url.Parse failure. The
|
||||
// *url.Error that url.Parse returns embeds the raw URL in its message, so
|
||||
// wrapping it directly would echo a credential-bearing URL into logs.
|
||||
func wrapParseError(err error) error {
|
||||
var uerr *url.Error
|
||||
if errors.As(err, &uerr) {
|
||||
return fmt.Errorf("invalid URL: %w", uerr.Err)
|
||||
}
|
||||
|
||||
return fmt.Errorf("invalid URL: %w", err)
|
||||
return nil, ErrUnsupportedScheme
|
||||
}
|
||||
|
||||
// String returns a human-readable representation of the storage URL.
|
||||
|
||||
@@ -3,7 +3,6 @@ package storage_test
|
||||
import (
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"sneak.berlin/go/vaultik/internal/storage"
|
||||
@@ -109,100 +108,3 @@ func TestParseStorageURLErrors(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestParseStorageURLRejectsCredentials checks that a URL carrying
|
||||
// credentials in its userinfo or in an unknown query parameter is
|
||||
// rejected, and that the error never echoes the secret-bearing URL back
|
||||
// into logs or output.
|
||||
func TestParseStorageURLRejectsCredentials(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Split so the literals never form a "user:pass@" URL pattern that
|
||||
// tooling would flag as a real hardcoded credential.
|
||||
const (
|
||||
key = "AKIAKEY"
|
||||
secret = "topsecret"
|
||||
)
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
raw string
|
||||
wantErr error
|
||||
secrets []string // must not appear in the error message
|
||||
}{
|
||||
{
|
||||
name: "s3 userinfo",
|
||||
raw: "s3://" + key + ":" + secret + "@mybucket/prefix",
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{key, secret, "mybucket"},
|
||||
},
|
||||
{
|
||||
name: "s3 unknown query param",
|
||||
raw: "s3://mybucket?access_key=" + key + "&secret=" + secret,
|
||||
wantErr: storage.ErrURLUnknownParam,
|
||||
secrets: []string{key, secret},
|
||||
},
|
||||
{
|
||||
name: "s3 misspelt endpoint",
|
||||
raw: "s3://mybucket?endpiont=minio.example.com",
|
||||
wantErr: storage.ErrURLUnknownParam,
|
||||
secrets: nil,
|
||||
},
|
||||
{
|
||||
name: "rclone userinfo",
|
||||
raw: "rclone://user:" + secret + "@gdrive/backups",
|
||||
wantErr: storage.ErrURLCredentials,
|
||||
secrets: []string{secret},
|
||||
},
|
||||
{
|
||||
name: "rclone query param",
|
||||
raw: "rclone://gdrive/backups?token=" + secret,
|
||||
wantErr: storage.ErrURLUnknownParam,
|
||||
secrets: []string{secret},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
_, err := storage.ParseStorageURL(tc.raw)
|
||||
if !errors.Is(err, tc.wantErr) {
|
||||
t.Fatalf("ParseStorageURL(%q) error = %v, want %v",
|
||||
tc.raw, err, tc.wantErr)
|
||||
}
|
||||
|
||||
// The rejection must name the proper config keys so the
|
||||
// operator knows where credentials belong.
|
||||
for _, key := range []string{"s3.access_key_id", "s3.secret_access_key"} {
|
||||
if !strings.Contains(err.Error(), key) {
|
||||
t.Errorf("error %q does not name %q", err.Error(), key)
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
// itself fails, the wrapped error carries only the inner cause, not the
|
||||
// *url.Error whose text embeds the raw (possibly credential-bearing) URL.
|
||||
func TestParseStorageURLParseFailureHidesURL(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
const raw = "s3://mybucket/%zz"
|
||||
|
||||
_, err := storage.ParseStorageURL(raw)
|
||||
if err == nil {
|
||||
t.Fatalf("ParseStorageURL(%q) returned no error", raw)
|
||||
}
|
||||
|
||||
if strings.Contains(err.Error(), "mybucket") {
|
||||
t.Errorf("error message echoed the raw URL: %v", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,146 +0,0 @@
|
||||
package vaultik_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"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/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/types"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
|
||||
// setupConsistencyTest builds a Vaultik whose local database and mock
|
||||
// remote both hold the given snapshots. Remote metadata is stored under
|
||||
// the production layout, metadata/<RemoteSnapshotKey(id)>/manifest.json.zst.
|
||||
// It returns the instance and the mock so a test can inspect the remote.
|
||||
func setupConsistencyTest(
|
||||
t *testing.T, snapshotIDs []string,
|
||||
) (*vaultik.Vaultik, *MockStorer) {
|
||||
t.Helper()
|
||||
|
||||
ctx := context.Background()
|
||||
db, err := database.New(ctx, ":memory:")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
|
||||
repos := database.NewRepositories(db)
|
||||
mockStorage := NewMockStorer()
|
||||
|
||||
for _, id := range snapshotIDs {
|
||||
parts := strings.Split(id, "_")
|
||||
startedAt, err := time.Parse(time.RFC3339, parts[len(parts)-1])
|
||||
require.NoError(t, err, "parsing timestamp from snapshot ID %q", id)
|
||||
|
||||
completedAt := startedAt.Add(5 * time.Minute)
|
||||
snap := &database.Snapshot{
|
||||
ID: types.SnapshotID(id),
|
||||
Hostname: testHostname,
|
||||
VaultikVersion: testLabel,
|
||||
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, "creating snapshot %s", id)
|
||||
|
||||
metadataKey := "metadata/" + snapshot.RemoteSnapshotKey(id) +
|
||||
"/manifest.json.zst"
|
||||
err = mockStorage.Put(ctx, metadataKey, strings.NewReader("stub"))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
v := &vaultik.Vaultik{
|
||||
Storage: mockStorage,
|
||||
Repositories: repos,
|
||||
DB: db,
|
||||
Stdout: &bytes.Buffer{},
|
||||
Stderr: &bytes.Buffer{},
|
||||
Stdin: &bytes.Buffer{},
|
||||
}
|
||||
v.SetContext(ctx)
|
||||
|
||||
return v, mockStorage
|
||||
}
|
||||
|
||||
func remoteHasSnapshot(t *testing.T, m *MockStorer, id string) bool {
|
||||
t.Helper()
|
||||
|
||||
prefix := "metadata/" + snapshot.RemoteSnapshotKey(id) + "/"
|
||||
keys, err := m.List(context.Background(), prefix)
|
||||
require.NoError(t, err)
|
||||
|
||||
return len(keys) > 0
|
||||
}
|
||||
|
||||
// TestPurgeKeepsRemotelyBackedLocalRows guards against issue #160
|
||||
// (https://git.eeqj.de/sneak/vaultik/issues/160): purge reconciles local
|
||||
// rows against the remote first, and that step compared human snapshot IDs
|
||||
// against the hashed remote directory names, which never match — so it
|
||||
// deleted every local record and the purge itself then removed nothing.
|
||||
//
|
||||
// With every snapshot still present remotely and nothing old enough to
|
||||
// purge, all local rows must survive the reconcile untouched.
|
||||
func TestPurgeKeepsRemotelyBackedLocalRows(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ids := []string{snapHomeT0, snapHomeT1, snapSystemT0}
|
||||
|
||||
v, _ := setupConsistencyTest(t, ids)
|
||||
|
||||
err := v.PurgeSnapshotsWithOptions(&vaultik.SnapshotPurgeOptions{
|
||||
// 100 years: nothing is old enough to delete, so the reconcile
|
||||
// is the only thing that touches the rows.
|
||||
OlderThan: "36500d",
|
||||
Force: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
remaining := listRemainingSnapshots(t, v)
|
||||
assert.Len(t, remaining, len(ids),
|
||||
"remotely-backed local rows must survive the reconcile")
|
||||
assert.Contains(t, remaining, snapHomeT0)
|
||||
assert.Contains(t, remaining, snapHomeT1)
|
||||
assert.Contains(t, remaining, snapSystemT0)
|
||||
}
|
||||
|
||||
// TestPurgeRemovesLocalAndRemoteTogether proves the two halves stay
|
||||
// consistent: a purged snapshot is gone both locally and remotely, while a
|
||||
// retained one keeps both. Before the fix, the reconcile dropped every
|
||||
// local row yet the remote metadata was left in place.
|
||||
func TestPurgeRemovesLocalAndRemoteTogether(t *testing.T) {
|
||||
log.Initialize(log.Config{})
|
||||
t.Parallel()
|
||||
|
||||
ids := []string{snapHomeT0, snapHomeT1, snapSystemT0}
|
||||
|
||||
v, mock := setupConsistencyTest(t, ids)
|
||||
|
||||
err := v.PurgeSnapshotsWithOptions(&vaultik.SnapshotPurgeOptions{
|
||||
KeepLatest: true,
|
||||
Force: true,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Keep latest per name: newest home and the lone system are kept.
|
||||
remaining := listRemainingSnapshots(t, v)
|
||||
assert.ElementsMatch(t, []string{snapHomeT1, snapSystemT0}, remaining)
|
||||
|
||||
// Local and remote agree: the older home snapshot is gone from both,
|
||||
// the retained ones are present in both.
|
||||
assert.False(t, remoteHasSnapshot(t, mock, snapHomeT0),
|
||||
"purged snapshot must also be removed remotely")
|
||||
assert.True(t, remoteHasSnapshot(t, mock, snapHomeT1),
|
||||
"retained snapshot must remain remotely")
|
||||
assert.True(t, remoteHasSnapshot(t, mock, snapSystemT0),
|
||||
"retained snapshot must remain remotely")
|
||||
}
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
"sneak.berlin/go/vaultik/internal/database"
|
||||
"sneak.berlin/go/vaultik/internal/log"
|
||||
"sneak.berlin/go/vaultik/internal/snapshot"
|
||||
"sneak.berlin/go/vaultik/internal/types"
|
||||
"sneak.berlin/go/vaultik/internal/vaultik"
|
||||
)
|
||||
@@ -61,11 +60,8 @@ func setupPurgeTest(t *testing.T, snapshotIDs []string) *vaultik.Vaultik {
|
||||
})
|
||||
require.NoError(t, err, "creating snapshot %s", id)
|
||||
|
||||
// Create the remote metadata stub under the production layout so
|
||||
// syncWithRemote keeps the local row. Production stores metadata
|
||||
// under the hashed remote key, not the human snapshot ID.
|
||||
metadataKey := "metadata/" + snapshot.RemoteSnapshotKey(id) +
|
||||
"/manifest.json.zst"
|
||||
// Create remote metadata stub so syncWithRemote keeps it
|
||||
metadataKey := "metadata/" + id + "/manifest.json.zst"
|
||||
err = mockStorage.Put(ctx, metadataKey, strings.NewReader("stub"))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
@@ -935,23 +935,29 @@ func (v *Vaultik) downloadManifestByKey(remoteKey string) (*snapshot.Manifest, e
|
||||
func (v *Vaultik) syncWithRemote() error {
|
||||
log.Info("Syncing with remote snapshots")
|
||||
|
||||
// Remote metadata lives under metadata/<remote-key>/, where the
|
||||
// directory name is snapshot.RemoteSnapshotKey(id), not the human
|
||||
// snapshot ID. Compare each local row's hashed key against that set
|
||||
// so a row still backed by remote metadata is kept. Comparing human
|
||||
// IDs against the hashed directory names matches nothing and deletes
|
||||
// every local snapshot record (issue #160).
|
||||
remoteKeys, err := v.listAllRemoteSnapshotKeys()
|
||||
if err != nil {
|
||||
return fmt.Errorf("listing remote snapshots: %w", err)
|
||||
// Get all remote snapshot IDs
|
||||
remoteSnapshots := make(map[string]bool)
|
||||
objectCh := v.Storage.ListStream(v.ctx, "metadata/")
|
||||
|
||||
for object := range objectCh {
|
||||
if object.Err != nil {
|
||||
return fmt.Errorf("listing remote snapshots: %w", object.Err)
|
||||
}
|
||||
|
||||
remoteKeySet := make(map[string]bool, len(remoteKeys))
|
||||
for _, k := range remoteKeys {
|
||||
remoteKeySet[k] = true
|
||||
// Extract snapshot ID from paths like metadata/hostname-20240115-143052Z/
|
||||
parts := strings.Split(object.Key, "/")
|
||||
if len(parts) >= minSnapshotIDParts &&
|
||||
parts[0] == metadataDirName && parts[1] != "" {
|
||||
// Skip macOS resource fork files (._*) and other hidden files
|
||||
if strings.HasPrefix(parts[1], ".") {
|
||||
continue
|
||||
}
|
||||
|
||||
log.Debug("Found remote snapshots", "count", len(remoteKeySet))
|
||||
remoteSnapshots[parts[1]] = true
|
||||
}
|
||||
}
|
||||
|
||||
log.Debug("Found remote snapshots", "count", len(remoteSnapshots))
|
||||
|
||||
// Get all local snapshots (use a high limit to get all)
|
||||
localSnapshots, err := v.Repositories.Snapshots.ListRecent(v.ctx, listRecentLimit)
|
||||
@@ -959,12 +965,12 @@ func (v *Vaultik) syncWithRemote() error {
|
||||
return fmt.Errorf("listing local snapshots: %w", err)
|
||||
}
|
||||
|
||||
// Remove local snapshots whose metadata is absent from the remote.
|
||||
// Remove local snapshots that don't exist remotely
|
||||
removedCount := 0
|
||||
|
||||
for _, snap := range localSnapshots {
|
||||
snapshotIDStr := snap.ID.String()
|
||||
if !remoteKeySet[snapshot.RemoteSnapshotKey(snapshotIDStr)] {
|
||||
if !remoteSnapshots[snapshotIDStr] {
|
||||
log.Info("Removing local snapshot not found in remote",
|
||||
"snapshot_id", snap.ID)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user