From 525dc6af007a7c8469fb7cfc851b1f9a4f0aadb7 Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Mon, 7 Sep 2026 11:10:17 +0000 Subject: [PATCH] feat: Add the backup archive writer and reader Open extracts and verifies every member against the manifest before the reader is usable, so a corrupt archive is refused before a restore writes its first document rather than halfway through. --- backup/archive.go | 245 +++++++++++++++++++++++++++++++++++++++++ backup/archive_test.go | 216 ++++++++++++++++++++++++++++++++++++ 2 files changed, 461 insertions(+) create mode 100644 backup/archive.go create mode 100644 backup/archive_test.go diff --git a/backup/archive.go b/backup/archive.go new file mode 100644 index 0000000..1d83e57 --- /dev/null +++ b/backup/archive.go @@ -0,0 +1,245 @@ +package backup + +import ( + "archive/tar" + "compress/gzip" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path" + "path/filepath" + "strings" + "time" +) + +// ErrChecksum is returned when an archive member does not match the checksum +// the manifest recorded for it. +var ErrChecksum = errors.New("archive member failed its checksum") + +// Writer streams a tar.gz. Members are written in the order they are produced +// and the manifest goes last, because its per-collection checksums are only +// known once every collection has been written. +type Writer struct { + gz *gzip.Writer + tar *tar.Writer +} + +// NewWriter starts an archive on out. out may be a file or stdout; nothing here +// seeks. +func NewWriter(out io.Writer) *Writer { + gz := gzip.NewWriter(out) + return &Writer{gz: gz, tar: tar.NewWriter(gz)} +} + +func (w *Writer) writeMember(name string, body []byte) error { + h := &tar.Header{ + Name: name, + Mode: 0o600, + Size: int64(len(body)), + ModTime: time.Now().UTC(), + Typeflag: tar.TypeReg, + } + if err := w.tar.WriteHeader(h); err != nil { + return fmt.Errorf("write header %s: %w", name, err) + } + if _, err := w.tar.Write(body); err != nil { + return fmt.Errorf("write %s: %w", name, err) + } + return nil +} + +// WriteCollection writes the concatenated raw BSON of one collection and +// returns the manifest entry describing it. +func (w *Writer) WriteCollection(name string, docs [][]byte) (CollectionEntry, error) { + var body []byte + for _, d := range docs { + body = append(body, d...) + } + sum := sha256.Sum256(body) + entry := CollectionEntry{ + Name: name, + Documents: int64(len(docs)), + Bytes: int64(len(body)), + SHA256: hex.EncodeToString(sum[:]), + } + if err := w.writeMember(collectionMember(name), body); err != nil { + return CollectionEntry{}, err + } + return entry, nil +} + +// WriteIndexes writes a collection's index specifications verbatim. +func (w *Writer) WriteIndexes(name string, specsJSON []byte) error { + return w.writeMember(indexMember(name), specsJSON) +} + +// Close writes the manifest and finishes the archive. +func (w *Writer) Close(m Manifest) error { + raw, err := json.MarshalIndent(m, "", " ") + if err != nil { + return fmt.Errorf("marshal manifest: %w", err) + } + if err := w.writeMember(ManifestName, raw); err != nil { + return err + } + if err := w.tar.Close(); err != nil { + return err + } + return w.gz.Close() +} + +func collectionMember(name string) string { return "collections/" + name + ".bson" } +func indexMember(name string) string { return "indexes/" + name + ".json" } + +// Reader is an opened archive. +// +// Open extracts to a temporary directory rather than streaming, because gzip +// offers no random access and the manifest — which carries the checksums every +// other member is judged against — is written last. Verifying before writing a +// single document to the target is worth one pass over local disk. This is why +// the container image needs a /tmp. +type Reader struct { + dir string + manifest Manifest +} + +// Open extracts, verifies and returns the archive at path. The caller must +// Close it. +func Open(archivePath string) (*Reader, error) { + dir, err := os.MkdirTemp("", "vantage-restore-*") + if err != nil { + return nil, fmt.Errorf("temp dir: %w", err) + } + r := &Reader{dir: dir} + + if err := r.extract(archivePath); err != nil { + r.Close() + return nil, err + } + if err := r.loadManifest(); err != nil { + r.Close() + return nil, err + } + if err := r.verifyMembers(); err != nil { + r.Close() + return nil, err + } + return r, nil +} + +func (r *Reader) extract(archivePath string) error { + f, err := os.Open(archivePath) + if err != nil { + return fmt.Errorf("open archive: %w", err) + } + defer f.Close() + + gz, err := gzip.NewReader(f) + if err != nil { + return fmt.Errorf("archive is not gzip: %w", err) + } + defer gz.Close() + + tr := tar.NewReader(gz) + for { + h, err := tr.Next() + if err == io.EOF { + return nil + } + if err != nil { + return fmt.Errorf("read archive: %w", err) + } + if h.Typeflag != tar.TypeReg { + continue + } + dest, err := safeJoin(r.dir, h.Name) + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(dest), 0o700); err != nil { + return fmt.Errorf("mkdir: %w", err) + } + out, err := os.OpenFile(dest, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) + if err != nil { + return fmt.Errorf("create %s: %w", h.Name, err) + } + if _, err := io.Copy(out, tr); err != nil { + out.Close() + return fmt.Errorf("extract %s: %w", h.Name, err) + } + if err := out.Close(); err != nil { + return err + } + } +} + +// safeJoin refuses a member name that escapes the extraction directory. An +// archive is operator-supplied input and may not be one we wrote. +func safeJoin(dir, name string) (string, error) { + clean := path.Clean("/" + name) + dest := filepath.Join(dir, filepath.FromSlash(strings.TrimPrefix(clean, "/"))) + if !strings.HasPrefix(dest, filepath.Clean(dir)+string(os.PathSeparator)) { + return "", fmt.Errorf("archive member %q escapes the extraction directory", name) + } + return dest, nil +} + +func (r *Reader) loadManifest() error { + raw, err := os.ReadFile(filepath.Join(r.dir, ManifestName)) + if err != nil { + return fmt.Errorf("archive has no %s: %w", ManifestName, err) + } + if err := json.Unmarshal(raw, &r.manifest); err != nil { + return fmt.Errorf("parse %s: %w", ManifestName, err) + } + return r.manifest.Check() +} + +func (r *Reader) verifyMembers() error { + for _, c := range r.manifest.Collections { + f, err := os.Open(filepath.Join(r.dir, collectionMember(c.Name))) + if err != nil { + return fmt.Errorf("%w: %s is named in the manifest but absent from the archive", + ErrChecksum, c.Name) + } + h := sha256.New() + n, err := io.Copy(h, f) + f.Close() + if err != nil { + return fmt.Errorf("read %s: %w", c.Name, err) + } + if n != c.Bytes { + return fmt.Errorf("%w: %s is %d bytes, manifest says %d", ErrChecksum, c.Name, n, c.Bytes) + } + if got := hex.EncodeToString(h.Sum(nil)); got != c.SHA256 { + return fmt.Errorf("%w: %s checksum %s, manifest says %s", ErrChecksum, c.Name, got, c.SHA256) + } + } + return nil +} + +// Manifest returns the verified manifest. +func (r *Reader) Manifest() Manifest { return r.manifest } + +// OpenCollection returns the raw BSON stream for one collection. +func (r *Reader) OpenCollection(name string) (io.ReadCloser, error) { + return os.Open(filepath.Join(r.dir, collectionMember(name))) +} + +// IndexesJSON returns a collection's index specifications, or nil when the +// archive holds none. A collection with no indexes beyond _id_ is ordinary and +// is not an error. +func (r *Reader) IndexesJSON(name string) ([]byte, error) { + raw, err := os.ReadFile(filepath.Join(r.dir, indexMember(name))) + if os.IsNotExist(err) { + return nil, nil + } + return raw, err +} + +// Close removes the extraction directory. +func (r *Reader) Close() error { return os.RemoveAll(r.dir) } diff --git a/backup/archive_test.go b/backup/archive_test.go new file mode 100644 index 0000000..62c7a67 --- /dev/null +++ b/backup/archive_test.go @@ -0,0 +1,216 @@ +package backup + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "errors" + "io" + "os" + "path/filepath" + "testing" + "time" +) + +// writeSampleArchive builds a two-collection archive on disk and returns its path. +func writeSampleArchive(t *testing.T) string { + t.Helper() + path := filepath.Join(t.TempDir(), "sample.tar.gz") + f, err := os.Create(path) + if err != nil { + t.Fatalf("create: %v", err) + } + defer f.Close() + + w := NewWriter(f) + servers, err := w.WriteCollection("servers", [][]byte{[]byte("one"), []byte("two")}) + if err != nil { + t.Fatalf("WriteCollection: %v", err) + } + if err := w.WriteIndexes("servers", []byte(`[{"name":"idx"}]`)); err != nil { + t.Fatalf("WriteIndexes: %v", err) + } + keys, err := w.WriteCollection("keys", [][]byte{[]byte("k")}) + if err != nil { + t.Fatalf("WriteCollection: %v", err) + } + if err := w.Close(Manifest{ + FormatVersion: FormatVersion, + CreatedAt: time.Now().UTC(), + MongoDB: "vantage", + Collections: []CollectionEntry{servers, keys}, + }); err != nil { + t.Fatalf("Close: %v", err) + } + return path +} + +func TestWriterRecordsCountsAndChecksums(t *testing.T) { + var buf bytes.Buffer + w := NewWriter(&buf) + e, err := w.WriteCollection("servers", [][]byte{[]byte("one"), []byte("two")}) + if err != nil { + t.Fatalf("WriteCollection: %v", err) + } + if e.Name != "servers" { + t.Fatalf("name %q", e.Name) + } + if e.Documents != 2 { + t.Fatalf("documents %d, want 2", e.Documents) + } + if e.Bytes != 6 { + t.Fatalf("bytes %d, want 6", e.Bytes) + } + if len(e.SHA256) != 64 { + t.Fatalf("sha256 %q is not 64 hex chars", e.SHA256) + } +} + +func TestRoundTrip(t *testing.T) { + r, err := Open(writeSampleArchive(t)) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer r.Close() + + if r.Manifest().MongoDB != "vantage" { + t.Fatalf("manifest not read back: %+v", r.Manifest()) + } + + rc, err := r.OpenCollection("servers") + if err != nil { + t.Fatalf("OpenCollection: %v", err) + } + defer rc.Close() + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("read: %v", err) + } + if string(got) != "onetwo" { + t.Fatalf("got %q, want %q", got, "onetwo") + } + + idx, err := r.IndexesJSON("servers") + if err != nil { + t.Fatalf("IndexesJSON: %v", err) + } + if string(idx) != `[{"name":"idx"}]` { + t.Fatalf("indexes round-tripped as %q", idx) + } +} + +func TestIndexesJSONAbsentIsEmptyNotError(t *testing.T) { + r, err := Open(writeSampleArchive(t)) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer r.Close() + idx, err := r.IndexesJSON("keys") + if err != nil { + t.Fatalf("a collection with no index member must not error: %v", err) + } + if len(idx) != 0 { + t.Fatalf("want empty, got %q", idx) + } +} + +func TestOpenRejectsCorruptedMember(t *testing.T) { + path := writeSampleArchive(t) + + // Rewrite the archive with one byte of a collection member flipped, leaving + // the manifest's checksum describing the original. + corrupt := filepath.Join(t.TempDir(), "corrupt.tar.gz") + rewriteFlippingCollectionByte(t, path, corrupt, "servers") + + if _, err := Open(corrupt); !errors.Is(err, ErrChecksum) { + t.Fatalf("got %v, want ErrChecksum", err) + } +} + +func TestOpenRejectsUnknownFormatVersion(t *testing.T) { + path := filepath.Join(t.TempDir(), "future.tar.gz") + f, err := os.Create(path) + if err != nil { + t.Fatalf("create: %v", err) + } + w := NewWriter(f) + if err := w.Close(Manifest{FormatVersion: 99}); err != nil { + t.Fatalf("Close: %v", err) + } + f.Close() + + if _, err := Open(path); !errors.Is(err, ErrUnknownFormat) { + t.Fatalf("got %v, want ErrUnknownFormat", err) + } +} + +func TestCloseRemovesTempDir(t *testing.T) { + r, err := Open(writeSampleArchive(t)) + if err != nil { + t.Fatalf("Open: %v", err) + } + dir := r.dir + if _, err := os.Stat(dir); err != nil { + t.Fatalf("temp dir missing while open: %v", err) + } + if err := r.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("temp dir %s survived Close", dir) + } +} + +// rewriteFlippingCollectionByte copies an archive, flipping one byte inside the +// named collection's .bson member so its content no longer matches the checksum +// the manifest recorded. +func rewriteFlippingCollectionByte(t *testing.T, src, dst, collection string) { + t.Helper() + + in, err := os.Open(src) + if err != nil { + t.Fatalf("open src: %v", err) + } + defer in.Close() + gz, err := gzip.NewReader(in) + if err != nil { + t.Fatalf("gzip: %v", err) + } + defer gz.Close() + + out, err := os.Create(dst) + if err != nil { + t.Fatalf("create dst: %v", err) + } + defer out.Close() + gw := gzip.NewWriter(out) + defer gw.Close() + tw := tar.NewWriter(gw) + defer tw.Close() + + tr := tar.NewReader(gz) + target := "collections/" + collection + ".bson" + for { + h, err := tr.Next() + if err == io.EOF { + return + } + if err != nil { + t.Fatalf("tar next: %v", err) + } + body, err := io.ReadAll(tr) + if err != nil { + t.Fatalf("read member: %v", err) + } + if h.Name == target && len(body) > 0 { + body[0] ^= 0xFF + } + h.Size = int64(len(body)) + if err := tw.WriteHeader(h); err != nil { + t.Fatalf("write header: %v", err) + } + if _, err := tw.Write(body); err != nil { + t.Fatalf("write body: %v", err) + } + } +}