fix: declare grpc dependency in shared go.mod

The workspace supplied it; a standalone build of the module could not
resolve google.golang.org/grpc at all.

backup/archive.go

@@ -0,0 +1,253 @@
+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 := r.OpenCollection(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) {
+	p, err := safeJoin(r.dir, collectionMember(name))
+	if err != nil {
+		return nil, err
+	}
+	return os.Open(p)
+}
+
+// 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) {
+	p, err := safeJoin(r.dir, indexMember(name))
+	if err != nil {
+		return nil, err
+	}
+	raw, err := os.ReadFile(p)
+	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) }

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)
+		}
+	}
+}

backup/dump.go

@@ -0,0 +1,184 @@
+package backup
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"sort"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+// DumpOptions configures one backup.
+type DumpOptions struct {
+	Client   *mongo.Client
+	Database string
+	Exclude  []string
+
+	// KeyHex is KEY_ENCRYPTION_KEY. It is fingerprinted and discarded; it is
+	// never written to the archive.
+	KeyHex string
+
+	// AllowNoKey permits a backup of a deployment that stores no encrypted
+	// material. The manifest then records a null fingerprint, which restore
+	// reports rather than treating as a match.
+	AllowNoKey bool
+
+	VantageVersion string
+	Out            io.Writer
+}
+
+// Dump writes a complete archive of one database to opt.Out.
+//
+// Collections are enumerated live rather than read from a list. A backup tool
+// has no equivalent of AssertNoScopedCollectionMissed to catch a hardcoded list
+// drifting, and the first symptom of that drift would be a restore silently
+// missing a collection added since the list was written.
+func Dump(ctx context.Context, opt DumpOptions) (Manifest, error) {
+	fingerprint, err := dumpFingerprint(opt)
+	if err != nil {
+		return Manifest{}, err
+	}
+
+	db := opt.Client.Database(opt.Database)
+	names, err := db.ListCollectionNames(ctx, bson.M{})
+	if err != nil {
+		return Manifest{}, fmt.Errorf("list collections: %w", err)
+	}
+	sort.Strings(names)
+
+	excluded := map[string]bool{}
+	for _, e := range opt.Exclude {
+		excluded[e] = true
+	}
+
+	serverVersion, err := mongoServerVersion(ctx, opt.Client)
+	if err != nil {
+		return Manifest{}, err
+	}
+	hostname, err := os.Hostname()
+	if err != nil {
+		hostname = "unknown"
+	}
+
+	w := NewWriter(opt.Out)
+	entries := make([]CollectionEntry, 0, len(names))
+	for _, name := range names {
+		if excluded[name] {
+			continue
+		}
+		entry, err := dumpCollection(ctx, w, db, name)
+		if err != nil {
+			return Manifest{}, err
+		}
+		entries = append(entries, entry)
+	}
+
+	m := Manifest{
+		FormatVersion:      FormatVersion,
+		CreatedAt:          time.Now().UTC(),
+		VantageVersion:     opt.VantageVersion,
+		Hostname:           hostname,
+		MongoDB:            opt.Database,
+		MongoServerVersion: serverVersion,
+		KeyFingerprint:     fingerprint,
+		Collections:        entries,
+		Excluded:           append([]string{}, opt.Exclude...),
+	}
+	if err := w.Close(m); err != nil {
+		return Manifest{}, err
+	}
+	return m, nil
+}
+
+// dumpFingerprint applies the key policy before any output is produced. An
+// archive of ciphertext whose key was never recorded is worse than no archive,
+// because it looks like a backup.
+func dumpFingerprint(opt DumpOptions) (*string, error) {
+	if opt.KeyHex == "" {
+		if opt.AllowNoKey {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("%w: pass --allow-no-key only if this deployment stores no encrypted data", ErrNoKey)
+	}
+	fp, err := FingerprintHex(opt.KeyHex)
+	if err != nil {
+		return nil, err
+	}
+	return &fp, nil
+}
+
+func dumpCollection(ctx context.Context, w *Writer, db *mongo.Database, name string) (CollectionEntry, error) {
+	cur, err := db.Collection(name).Find(ctx, bson.M{})
+	if err != nil {
+		return CollectionEntry{}, fmt.Errorf("find %s: %w", name, err)
+	}
+	defer cur.Close(ctx)
+
+	var docs [][]byte
+	for cur.Next(ctx) {
+		// cur.Current is only valid until the next Next, and it is written to
+		// the archive verbatim rather than through a map, so every BSON type
+		// survives exactly as the server stored it.
+		docs = append(docs, append([]byte(nil), cur.Current...))
+	}
+	if err := cur.Err(); err != nil {
+		return CollectionEntry{}, fmt.Errorf("iterate %s: %w", name, err)
+	}
+
+	entry, err := w.WriteCollection(name, docs)
+	if err != nil {
+		return CollectionEntry{}, err
+	}
+	if err := dumpIndexes(ctx, w, db, name); err != nil {
+		return CollectionEntry{}, err
+	}
+	return entry, nil
+}
+
+func dumpIndexes(ctx context.Context, w *Writer, db *mongo.Database, name string) error {
+	cur, err := db.Collection(name).Indexes().List(ctx)
+	if err != nil {
+		return fmt.Errorf("list indexes on %s: %w", name, err)
+	}
+	defer cur.Close(ctx)
+
+	// The specs are read as raw BSON and re-encoded as extended JSON, one
+	// element per index, so key order and every option the server reported —
+	// partialFilterExpression, collation, weights and the rest — survive
+	// verbatim. Decoding into bson.M would lose compound key order, and
+	// reconstructing an index from a hand-picked set of options would drop
+	// whatever was not picked.
+	var specs []bson.Raw
+	if err := cur.All(ctx, &specs); err != nil {
+		return fmt.Errorf("read indexes on %s: %w", name, err)
+	}
+	encoded := make([]json.RawMessage, 0, len(specs))
+	for _, spec := range specs {
+		ej, err := bson.MarshalExtJSON(spec, false, false)
+		if err != nil {
+			return fmt.Errorf("encode indexes on %s: %w", name, err)
+		}
+		encoded = append(encoded, ej)
+	}
+	raw, err := json.Marshal(encoded)
+	if err != nil {
+		return fmt.Errorf("encode indexes on %s: %w", name, err)
+	}
+	return w.WriteIndexes(name, raw)
+}
+
+func mongoServerVersion(ctx context.Context, client *mongo.Client) (string, error) {
+	var res struct {
+		Version string `bson:"version"`
+	}
+	err := client.Database("admin").RunCommand(ctx, bson.D{{Key: "buildInfo", Value: 1}}).Decode(&res)
+	if err != nil {
+		return "", fmt.Errorf("buildInfo: %w", err)
+	}
+	return res.Version, nil
+}

backup/dump_test.go

@@ -0,0 +1,206 @@
+package backup
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+func seed(t *testing.T, client *mongo.Client, dbName string) {
+	t.Helper()
+	ctx := context.Background()
+	db := client.Database(dbName)
+	if _, err := db.Collection("servers").InsertMany(ctx, []any{
+		bson.M{"_id": bson.NewObjectID(), "name": "alpha", "instance_id": "i1"},
+		bson.M{"_id": bson.NewObjectID(), "name": "beta", "instance_id": "i1"},
+	}); err != nil {
+		t.Fatalf("insert servers: %v", err)
+	}
+	if _, err := db.Collection("audit_logs").InsertOne(ctx, bson.M{"action": "login"}); err != nil {
+		t.Fatalf("insert audit_logs: %v", err)
+	}
+}
+
+func dumpToFile(t *testing.T, opt DumpOptions) (string, Manifest) {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "out.tar.gz")
+	f, err := os.Create(path)
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	opt.Out = f
+	m, err := Dump(context.Background(), opt)
+	if cerr := f.Close(); cerr != nil {
+		t.Fatalf("close: %v", cerr)
+	}
+	if err != nil {
+		t.Fatalf("Dump: %v", err)
+	}
+	return path, m
+}
+
+func TestDumpEnumeratesEveryCollection(t *testing.T) {
+	client, dbName := testDB(t)
+	seed(t, client, dbName)
+
+	_, m := dumpToFile(t, DumpOptions{
+		Client: client, Database: dbName, KeyHex: validKeyHex, VantageVersion: "test",
+	})
+
+	if _, ok := m.Collection("servers"); !ok {
+		t.Fatal("servers missing from the manifest")
+	}
+	if _, ok := m.Collection("audit_logs"); !ok {
+		t.Fatal("audit_logs missing; enumeration must not filter by a hardcoded list")
+	}
+	servers, _ := m.Collection("servers")
+	if servers.Documents != 2 {
+		t.Fatalf("servers documents %d, want 2", servers.Documents)
+	}
+	if m.MongoDB != dbName {
+		t.Fatalf("manifest database %q, want %q", m.MongoDB, dbName)
+	}
+	if m.MongoServerVersion == "" {
+		t.Fatal("manifest records no MongoDB server version")
+	}
+	if m.Hostname == "" {
+		t.Fatal("manifest records no hostname")
+	}
+}
+
+func TestDumpRecordsKeyFingerprint(t *testing.T) {
+	client, dbName := testDB(t)
+	seed(t, client, dbName)
+
+	_, m := dumpToFile(t, DumpOptions{Client: client, Database: dbName, KeyHex: validKeyHex})
+
+	want, err := FingerprintHex(validKeyHex)
+	if err != nil {
+		t.Fatalf("FingerprintHex: %v", err)
+	}
+	if m.KeyFingerprint == nil || *m.KeyFingerprint != want {
+		t.Fatalf("fingerprint %v, want %s", m.KeyFingerprint, want)
+	}
+}
+
+func TestDumpRefusesWithoutAKey(t *testing.T) {
+	client, dbName := testDB(t)
+	seed(t, client, dbName)
+
+	var buf bytes.Buffer
+	_, err := Dump(context.Background(), DumpOptions{
+		Client: client, Database: dbName, Out: &buf,
+	})
+	if !errors.Is(err, ErrNoKey) {
+		t.Fatalf("got %v, want ErrNoKey", err)
+	}
+	if buf.Len() != 0 {
+		t.Fatal("refusal must happen before anything is written")
+	}
+}
+
+func TestDumpAllowNoKeyStampsNull(t *testing.T) {
+	client, dbName := testDB(t)
+	seed(t, client, dbName)
+
+	_, m := dumpToFile(t, DumpOptions{Client: client, Database: dbName, AllowNoKey: true})
+	if m.KeyFingerprint != nil {
+		t.Fatalf("want a null fingerprint, got %v", *m.KeyFingerprint)
+	}
+}
+
+func TestDumpRejectsMalformedKey(t *testing.T) {
+	client, dbName := testDB(t)
+	var buf bytes.Buffer
+	_, err := Dump(context.Background(), DumpOptions{
+		Client: client, Database: dbName, KeyHex: "nonsense", Out: &buf,
+	})
+	if !errors.Is(err, ErrBadKey) {
+		t.Fatalf("got %v, want ErrBadKey", err)
+	}
+}
+
+func TestDumpExcludeIsRecordedAndOmitted(t *testing.T) {
+	client, dbName := testDB(t)
+	seed(t, client, dbName)
+
+	path, m := dumpToFile(t, DumpOptions{
+		Client: client, Database: dbName, KeyHex: validKeyHex,
+		Exclude: []string{"audit_logs"},
+	})
+
+	if _, ok := m.Collection("audit_logs"); ok {
+		t.Fatal("excluded collection is in the manifest's collection list")
+	}
+	if len(m.Excluded) != 1 || m.Excluded[0] != "audit_logs" {
+		t.Fatalf("excluded recorded as %v", m.Excluded)
+	}
+
+	r, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer r.Close()
+	if _, err := r.OpenCollection("audit_logs"); err == nil {
+		t.Fatal("excluded collection is present in the archive")
+	}
+}
+
+func TestDumpPreservesAwkwardBSONTypes(t *testing.T) {
+	client, dbName := testDB(t)
+	ctx := context.Background()
+
+	dec, err := bson.ParseDecimal128("1234.5678")
+	if err != nil {
+		t.Fatalf("ParseDecimal128: %v", err)
+	}
+	doc := bson.M{
+		"_id":     bson.NewObjectID(),
+		"decimal": dec,
+		"when":    bson.NewDateTimeFromTime(mustTime(t)),
+		"binary":  bson.Binary{Subtype: 0x00, Data: []byte{0x01, 0x02, 0x03}},
+		"nothing": nil,
+		"nested":  bson.A{bson.M{"deep": bson.A{1, 2, 3}}},
+	}
+	if _, err := client.Database(dbName).Collection("odd").InsertOne(ctx, doc); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	path, _ := dumpToFile(t, DumpOptions{Client: client, Database: dbName, KeyHex: validKeyHex})
+
+	original, err := client.Database(dbName).Collection("odd").FindOne(ctx, bson.M{}).Raw()
+	if err != nil {
+		t.Fatalf("read back: %v", err)
+	}
+
+	r, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer r.Close()
+	rc, err := r.OpenCollection("odd")
+	if err != nil {
+		t.Fatalf("OpenCollection: %v", err)
+	}
+	defer rc.Close()
+	archived, err := io.ReadAll(rc)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if !bytes.Equal(archived, []byte(original)) {
+		t.Fatal("archived BSON differs from what the driver returned")
+	}
+}
+
+func mustTime(t *testing.T) time.Time {
+	t.Helper()
+	return time.Date(2026, 9, 7, 12, 0, 0, 0, time.UTC)
+}

backup/fingerprint.go

@@ -11,7 +11,7 @@ import (
 	"errors"
 	"fmt"
 
-	"gitea.hostxtra.co.uk/mrhid6/vantage/shared/cryptobox"
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/cryptobox"
 )
 
 // ErrNoKey is returned when no key was supplied at all. It is distinct from

backup/manifest.go

@@ -3,6 +3,7 @@ package backup
 import (
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 )
 
@@ -16,6 +17,10 @@ const ManifestName = "manifest.json"
 // ErrUnknownFormat is returned for an archive this build cannot read.
 var ErrUnknownFormat = errors.New("unsupported archive format version")
 
+// ErrBadCollectionName is returned for a manifest naming a collection that
+// cannot safely be used as a path component.
+var ErrBadCollectionName = errors.New("manifest names an unusable collection")
+
 // CollectionEntry describes one collection in the archive. Bytes and SHA256
 // cover the uncompressed .bson member, which is what restore verifies before
 // writing anything.
@@ -49,6 +54,27 @@ func (m Manifest) Check() error {
 		return fmt.Errorf("%w: archive is version %d, this build reads version %d",
 			ErrUnknownFormat, m.FormatVersion, FormatVersion)
 	}
+	// Collection names become path components inside the extraction directory,
+	// and an archive is operator-supplied input that may not be one we wrote.
+	for _, c := range m.Collections {
+		if err := checkCollectionName(c.Name); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// checkCollectionName refuses a name that could escape a directory when joined
+// as a path component.
+func checkCollectionName(name string) error {
+	switch {
+	case name == "":
+		return fmt.Errorf("%w: a collection entry has no name", ErrBadCollectionName)
+	case name == "." || name == "..":
+		return fmt.Errorf("%w: %q", ErrBadCollectionName, name)
+	case strings.ContainsAny(name, "/\\"), strings.Contains(name, ".."):
+		return fmt.Errorf("%w: %q", ErrBadCollectionName, name)
+	}
 	return nil
 }
 
@@ -68,6 +94,9 @@ func (m Manifest) Collection(name string) (CollectionEntry, bool) {
 // match the archive, this is the list of what will be unreadable afterwards,
 // and an operator deserves to see it before the write rather than discover it
 // a week later.
+//
+// settings is not in the list: it holds no encrypted material. Its ESO read
+// token is a SHA-256 hash, not ciphertext.
 func CiphertextCollections() []string {
-	return []string{"keys", "secrets", "auth_providers", "console_sessions", "settings"}
+	return []string{"keys", "secrets", "auth_providers", "console_sessions"}
 }

backup/manifest_test.go

@@ -69,7 +69,7 @@ func TestManifestCollectionLookup(t *testing.T) {
 
 func TestCiphertextCollections(t *testing.T) {
 	got := CiphertextCollections()
-	want := []string{"keys", "secrets", "auth_providers", "console_sessions", "settings"}
+	want := []string{"keys", "secrets", "auth_providers", "console_sessions"}
 	if len(got) != len(want) {
 		t.Fatalf("got %v, want %v", got, want)
 	}
@@ -79,3 +79,25 @@ func TestCiphertextCollections(t *testing.T) {
 		}
 	}
 }
+
+// TestManifestRefusesUnusableCollectionNames covers names being used as path
+// components inside the extraction directory. An archive is operator-supplied
+// input and may not be one we wrote.
+func TestManifestRefusesUnusableCollectionNames(t *testing.T) {
+	for _, name := range []string{"", ".", "..", "../etc/passwd", "a/b", "a..b"} {
+		m := Manifest{
+			FormatVersion: FormatVersion,
+			Collections:   []CollectionEntry{{Name: name}},
+		}
+		if err := m.Check(); !errors.Is(err, ErrBadCollectionName) {
+			t.Fatalf("collection name %q was accepted (err %v)", name, err)
+		}
+	}
+	m := Manifest{
+		FormatVersion: FormatVersion,
+		Collections:   []CollectionEntry{{Name: "workflow_log_lines"}},
+	}
+	if err := m.Check(); err != nil {
+		t.Fatalf("an ordinary collection name was refused: %v", err)
+	}
+}

backup/mongo_test.go

@@ -0,0 +1,44 @@
+package backup
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+// testDB connects to the MongoDB named by MONGO_TEST_URI and returns a client
+// plus a database name unique to this test, dropped when the test ends.
+//
+// Skips rather than fails when the variable is unset: these tests need a real
+// server, and a developer without one should still be able to run the rest of
+// the suite.
+func testDB(t *testing.T) (*mongo.Client, string) {
+	t.Helper()
+	uri := os.Getenv("MONGO_TEST_URI")
+	if uri == "" {
+		t.Skip("MONGO_TEST_URI is not set; skipping tests that need MongoDB")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	client, err := mongo.Connect(options.Client().ApplyURI(uri))
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	if err := client.Ping(ctx, nil); err != nil {
+		t.Fatalf("ping: %v", err)
+	}
+	name := fmt.Sprintf("vantage_test_%d", time.Now().UnixNano())
+	t.Cleanup(func() {
+		c, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		_ = client.Database(name).Drop(c)
+		_ = client.Disconnect(c)
+	})
+	return client, name
+}

backup/restore.go

@@ -0,0 +1,393 @@
+package backup
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"sort"
+	"strings"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+// BatchSize is how many documents are inserted per bulk write.
+const BatchSize = 1000
+
+// ErrTargetNotEmpty is returned when the target database already holds data and
+// Force was not set.
+var ErrTargetNotEmpty = errors.New("target database is not empty")
+
+// ErrKeyMismatch is returned when the archive's key fingerprint does not match
+// the key supplied.
+var ErrKeyMismatch = errors.New("KEY_ENCRYPTION_KEY does not match the archive")
+
+// ErrIndexBuild is returned when a unique index in the archive cannot be built
+// on the restored data.
+var ErrIndexBuild = errors.New("index could not be built on the restored data")
+
+// RestoredCollection is what one collection's restore produced.
+type RestoredCollection struct {
+	Name      string
+	Documents int64
+	Indexes   int
+}
+
+// RestoreResult is the summary a caller prints.
+type RestoreResult struct {
+	Collections []RestoredCollection
+}
+
+// RestoreOptions configures one restore.
+type RestoreOptions struct {
+	Client   *mongo.Client
+	Database string
+	Archive  *Reader
+
+	// Force drops each collection in the archive before loading it. Without it
+	// a non-empty target is refused.
+	Force bool
+
+	KeyHex string
+
+	// IgnoreKeyMismatch proceeds past a fingerprint mismatch, having first
+	// warned which collections will hold unreadable ciphertext afterwards.
+	IgnoreKeyMismatch bool
+
+	// Warn receives operator-facing warnings. A nil Warn discards them.
+	Warn func(string)
+}
+
+func (o RestoreOptions) warn(format string, args ...any) {
+	if o.Warn != nil {
+		o.Warn(fmt.Sprintf(format, args...))
+	}
+}
+
+// Restore loads an archive into a database.
+//
+// The order is fixed and every check that can refuse does so before the first
+// write: format, checksums (done by Open), key policy, then target inspection.
+// A restore that has begun writing and then fails leaves a partial database
+// which the next run refuses to touch, which is correct — the alternative is a
+// silent merge, and merging two control planes reconciles nothing.
+func Restore(ctx context.Context, opt RestoreOptions) (RestoreResult, error) {
+	m := opt.Archive.Manifest()
+
+	if err := checkKey(m, opt); err != nil {
+		return RestoreResult{}, err
+	}
+	if err := checkTarget(ctx, opt); err != nil {
+		return RestoreResult{}, err
+	}
+	if err := warnLeftovers(ctx, opt, m); err != nil {
+		return RestoreResult{}, err
+	}
+	if len(m.Excluded) > 0 {
+		opt.warn("this archive excluded %s; those collections will be empty after the restore",
+			strings.Join(m.Excluded, ", "))
+	}
+	opt.warn("Redis is not restored. Sessions are the only state it holds, so everyone signs in again.")
+
+	res := RestoreResult{}
+	if err := warnVersionGap(ctx, opt, m); err != nil {
+		return res, err
+	}
+
+	for _, entry := range m.Collections {
+		rc, err := restoreCollection(ctx, opt, entry)
+		if err != nil {
+			return res, err
+		}
+		res.Collections = append(res.Collections, rc)
+	}
+	return res, nil
+}
+
+// checkKey applies the fingerprint policy.
+func checkKey(m Manifest, opt RestoreOptions) error {
+	if m.KeyFingerprint == nil {
+		opt.warn("this archive carries no key fingerprint, so nothing here proves your " +
+			"KEY_ENCRYPTION_KEY opens its ciphertext")
+		return nil
+	}
+	if opt.KeyHex == "" {
+		return fmt.Errorf("%w: the archive records a key fingerprint, so a key is required "+
+			"(pass --ignore-key-mismatch only if you accept unreadable secrets)", ErrNoKey)
+	}
+	got, err := FingerprintHex(opt.KeyHex)
+	if err != nil {
+		return err
+	}
+	if got == *m.KeyFingerprint {
+		return nil
+	}
+	if !opt.IgnoreKeyMismatch {
+		return fmt.Errorf("%w: archive fingerprint %s, your key fingerprints as %s",
+			ErrKeyMismatch, *m.KeyFingerprint, got)
+	}
+	opt.warn("proceeding past a key mismatch: ciphertext in %s will be permanently unreadable",
+		strings.Join(CiphertextCollections(), ", "))
+	return nil
+}
+
+// checkTarget refuses a non-empty database unless Force was set.
+func checkTarget(ctx context.Context, opt RestoreOptions) error {
+	db := opt.Client.Database(opt.Database)
+	names, err := db.ListCollectionNames(ctx, bson.M{})
+	if err != nil {
+		return fmt.Errorf("inspect target: %w", err)
+	}
+	if len(names) == 0 || opt.Force {
+		return nil
+	}
+	sort.Strings(names)
+
+	var found []string
+	for _, n := range names {
+		count, err := db.Collection(n).CountDocuments(ctx, bson.M{})
+		if err != nil {
+			return fmt.Errorf("count %s: %w", n, err)
+		}
+		found = append(found, fmt.Sprintf("%s (%d)", n, count))
+	}
+	return fmt.Errorf("%w: %s holds %s", ErrTargetNotEmpty, opt.Database, strings.Join(found, ", "))
+}
+
+// warnLeftovers names the collections already in the target that the archive
+// does not carry.
+//
+// They are named rather than dropped. A --force restore of an archive taken
+// with --exclude workflow_log_lines leaves the old logs joined to restored
+// runs, which the operator must know; but dropping a collection the archive
+// never mentioned would delete data nobody asked to delete, and there is no
+// way back from that.
+func warnLeftovers(ctx context.Context, opt RestoreOptions, m Manifest) error {
+	if !opt.Force {
+		// Without Force the target was already proven empty.
+		return nil
+	}
+	names, err := opt.Client.Database(opt.Database).ListCollectionNames(ctx, bson.M{})
+	if err != nil {
+		return fmt.Errorf("inspect target: %w", err)
+	}
+	inArchive := map[string]bool{}
+	for _, c := range m.Collections {
+		inArchive[c.Name] = true
+	}
+	var leftover []string
+	for _, n := range names {
+		if !inArchive[n] {
+			leftover = append(leftover, n)
+		}
+	}
+	if len(leftover) == 0 {
+		return nil
+	}
+	sort.Strings(leftover)
+	opt.warn("this archive does not carry %s, which already exist in %s and are left "+
+		"untouched: their contents will sit alongside the restored data",
+		strings.Join(leftover, ", "), opt.Database)
+	return nil
+}
+
+// warnVersionGap reports a major version difference between the server that
+// produced the archive and the one receiving it. It warns rather than refuses:
+// restoring across a major version is a normal part of an upgrade, and a tool
+// that refused would be blocking the migration it exists to make safe.
+func warnVersionGap(ctx context.Context, opt RestoreOptions, m Manifest) error {
+	if m.MongoServerVersion == "" {
+		return nil
+	}
+	target, err := mongoServerVersion(ctx, opt.Client)
+	if err != nil {
+		return err
+	}
+	if majorOf(m.MongoServerVersion) != majorOf(target) {
+		opt.warn("this archive came from MongoDB %s and you are restoring onto %s",
+			m.MongoServerVersion, target)
+	}
+	return nil
+}
+
+func majorOf(version string) string {
+	if i := strings.IndexByte(version, '.'); i >= 0 {
+		return version[:i]
+	}
+	return version
+}
+
+func restoreCollection(ctx context.Context, opt RestoreOptions, entry CollectionEntry) (RestoredCollection, error) {
+	coll := opt.Client.Database(opt.Database).Collection(entry.Name)
+	if opt.Force {
+		if err := coll.Drop(ctx); err != nil {
+			return RestoredCollection{}, fmt.Errorf("drop %s: %w", entry.Name, err)
+		}
+	}
+
+	written, err := insertDocuments(ctx, opt, coll, entry)
+	if err != nil {
+		return RestoredCollection{}, err
+	}
+	indexes, err := replayIndexes(ctx, opt, coll, entry.Name)
+	if err != nil {
+		return RestoredCollection{}, err
+	}
+	return RestoredCollection{Name: entry.Name, Documents: written, Indexes: indexes}, nil
+}
+
+func insertDocuments(ctx context.Context, opt RestoreOptions, coll *mongo.Collection, entry CollectionEntry) (int64, error) {
+	rc, err := opt.Archive.OpenCollection(entry.Name)
+	if err != nil {
+		return 0, fmt.Errorf("open %s in archive: %w", entry.Name, err)
+	}
+	defer rc.Close()
+
+	raw, err := io.ReadAll(rc)
+	if err != nil {
+		return 0, fmt.Errorf("read %s: %w", entry.Name, err)
+	}
+
+	var written int64
+	batch := make([]any, 0, BatchSize)
+	flush := func() error {
+		if len(batch) == 0 {
+			return nil
+		}
+		if _, err := coll.InsertMany(ctx, batch, options.InsertMany().SetOrdered(false)); err != nil {
+			return fmt.Errorf("insert into %s: %w", entry.Name, err)
+		}
+		written += int64(len(batch))
+		batch = batch[:0]
+		return nil
+	}
+
+	for len(raw) > 0 {
+		doc, rest, err := splitBSON(raw)
+		if err != nil {
+			return 0, fmt.Errorf("%s: %w", entry.Name, err)
+		}
+		batch = append(batch, doc)
+		raw = rest
+		if len(batch) == BatchSize {
+			if err := flush(); err != nil {
+				return 0, err
+			}
+		}
+	}
+	if err := flush(); err != nil {
+		return 0, err
+	}
+	return written, nil
+}
+
+// splitBSON peels one document off the front of a concatenated BSON stream. A
+// BSON document declares its own length in its first four bytes.
+func splitBSON(raw []byte) (bson.Raw, []byte, error) {
+	if len(raw) < 4 {
+		return nil, nil, fmt.Errorf("truncated BSON: %d trailing bytes", len(raw))
+	}
+	n := int(int32(raw[0]) | int32(raw[1])<<8 | int32(raw[2])<<16 | int32(raw[3])<<24)
+	if n < 5 || n > len(raw) {
+		return nil, nil, fmt.Errorf("BSON document declares length %d with %d bytes remaining", n, len(raw))
+	}
+	return bson.Raw(raw[:n]), raw[n:], nil
+}
+
+// replayIndexes recreates the archived indexes.
+//
+// The specs are handed to the createIndexes command exactly as the source
+// server reported them, rather than reconstructed into a mongo.IndexModel from
+// a hand-picked set of options. Reconstruction dropped every option nobody had
+// thought to pick — partialFilterExpression above all, which this codebase
+// relies on for partial unique indexes, and which replayed as a full unique
+// index fails on any real database. It also lost compound key order, which is
+// significant.
+//
+// A unique index that will not build means the restored data violates it, and
+// the unique indexes here — (instance_id, email), instance slug, settings
+// instance, the ESO token hash — are tenant-isolation properties rather than
+// optimisations. That aborts. A non-unique index failing is a performance
+// problem and warns.
+func replayIndexes(ctx context.Context, opt RestoreOptions, coll *mongo.Collection, name string) (int, error) {
+	raw, err := opt.Archive.IndexesJSON(name)
+	if err != nil {
+		return 0, fmt.Errorf("read index specs for %s: %w", name, err)
+	}
+	if len(raw) == 0 {
+		return 0, nil
+	}
+	var encoded []json.RawMessage
+	if err := json.Unmarshal(raw, &encoded); err != nil {
+		return 0, fmt.Errorf("parse index specs for %s: %w", name, err)
+	}
+
+	db := coll.Database()
+	created := 0
+	for _, ej := range encoded {
+		spec, indexName, unique, ok, err := indexSpecFrom(ej)
+		if err != nil {
+			return created, fmt.Errorf("parse index specs for %s: %w", name, err)
+		}
+		if !ok {
+			continue
+		}
+		cmd := bson.D{
+			{Key: "createIndexes", Value: name},
+			{Key: "indexes", Value: bson.A{spec}},
+		}
+		if err := db.RunCommand(ctx, cmd).Err(); err != nil {
+			if unique {
+				return created, fmt.Errorf("%w: %s on %s: %v", ErrIndexBuild, indexName, name, err)
+			}
+			opt.warn("index %s on %s was not created: %v", indexName, name, err)
+			continue
+		}
+		created++
+	}
+	return created, nil
+}
+
+// droppedIndexSpecFields are the fields the server reports on an existing index
+// but rejects when creating one. Everything else is passed through untouched.
+var droppedIndexSpecFields = map[string]bool{"v": true, "ns": true}
+
+// indexSpecFrom decodes one archived extended-JSON index specification into an
+// ordered bson.D suitable for createIndexes.
+//
+// The _id_ index is skipped: MongoDB creates it itself and refuses an explicit
+// attempt to create it.
+func indexSpecFrom(ej []byte) (bson.D, string, bool, bool, error) {
+	var d bson.D
+	if err := bson.UnmarshalExtJSON(ej, false, &d); err != nil {
+		return nil, "", false, false, err
+	}
+
+	out := make(bson.D, 0, len(d))
+	var name string
+	unique := false
+	hasKey := false
+	for _, e := range d {
+		switch e.Key {
+		case "name":
+			name, _ = e.Value.(string)
+		case "unique":
+			if u, ok := e.Value.(bool); ok {
+				unique = u
+			}
+		case "key":
+			hasKey = true
+		}
+		if droppedIndexSpecFields[e.Key] {
+			continue
+		}
+		out = append(out, e)
+	}
+	if name == "_id_" || !hasKey {
+		return nil, name, false, false, nil
+	}
+	return out, name, unique, true, nil
+}

backup/restore_test.go

@@ -0,0 +1,439 @@
+package backup
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+// archiveOf seeds a database, dumps it, and returns an opened Reader.
+func archiveOf(t *testing.T, client *mongo.Client, keyHex string, allowNoKey bool) *Reader {
+	t.Helper()
+	_, srcDB := testDB(t)
+	seed(t, client, srcDB)
+	path, _ := dumpToFile(t, DumpOptions{
+		Client: client, Database: srcDB, KeyHex: keyHex, AllowNoKey: allowNoKey,
+	})
+	r, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	t.Cleanup(func() { r.Close() })
+	return r
+}
+
+func countIn(t *testing.T, client *mongo.Client, dbName, coll string) int64 {
+	t.Helper()
+	n, err := client.Database(dbName).Collection(coll).CountDocuments(context.Background(), bson.M{})
+	if err != nil {
+		t.Fatalf("count %s: %v", coll, err)
+	}
+	return n
+}
+
+func TestRestoreIntoEmptyDatabase(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+
+	res, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+	})
+	if err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+	if countIn(t, client, target, "servers") != 2 {
+		t.Fatal("servers not restored")
+	}
+	if len(res.Collections) == 0 {
+		t.Fatal("result reports no collections")
+	}
+}
+
+func TestRestoreRefusesNonEmptyTarget(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+	if _, err := client.Database(target).Collection("servers").
+		InsertOne(context.Background(), bson.M{"name": "existing"}); err != nil {
+		t.Fatalf("seed target: %v", err)
+	}
+
+	_, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+	})
+	if !errors.Is(err, ErrTargetNotEmpty) {
+		t.Fatalf("got %v, want ErrTargetNotEmpty", err)
+	}
+	if countIn(t, client, target, "servers") != 1 {
+		t.Fatal("a refused restore modified the target")
+	}
+}
+
+func TestRestoreForceReplaces(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+	if _, err := client.Database(target).Collection("servers").
+		InsertOne(context.Background(), bson.M{"name": "existing"}); err != nil {
+		t.Fatalf("seed target: %v", err)
+	}
+
+	if _, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex, Force: true,
+	}); err != nil {
+		t.Fatalf("Restore --force: %v", err)
+	}
+	if got := countIn(t, client, target, "servers"); got != 2 {
+		t.Fatalf("servers has %d documents, want 2; force must drop, not merge", got)
+	}
+	n, err := client.Database(target).Collection("servers").
+		CountDocuments(context.Background(), bson.M{"name": "existing"})
+	if err != nil {
+		t.Fatalf("count: %v", err)
+	}
+	if n != 0 {
+		t.Fatal("the pre-existing document survived --force")
+	}
+}
+
+func TestRestoreRefusesKeyMismatch(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+	other := "0000000000000000000000000000000000000000000000000000000000000002"
+
+	_, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: other,
+	})
+	if !errors.Is(err, ErrKeyMismatch) {
+		t.Fatalf("got %v, want ErrKeyMismatch", err)
+	}
+	names, err := client.Database(target).ListCollectionNames(context.Background(), bson.M{})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(names) != 0 {
+		t.Fatalf("a refused restore wrote %v", names)
+	}
+}
+
+func TestRestoreRefusesWhenArchiveHasKeyAndEnvironmentDoesNot(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+
+	_, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive,
+	})
+	if !errors.Is(err, ErrNoKey) {
+		t.Fatalf("got %v, want ErrNoKey", err)
+	}
+}
+
+func TestRestoreIgnoreKeyMismatchWarnsAndProceeds(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+	other := "0000000000000000000000000000000000000000000000000000000000000002"
+
+	var warnings []string
+	if _, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive,
+		KeyHex: other, IgnoreKeyMismatch: true,
+		Warn: func(s string) { warnings = append(warnings, s) },
+	}); err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+	joined := strings.Join(warnings, "\n")
+	for _, name := range CiphertextCollections() {
+		if !strings.Contains(joined, name) {
+			t.Fatalf("warning does not name %s\ngot:\n%s", name, joined)
+		}
+	}
+	if countIn(t, client, target, "servers") != 2 {
+		t.Fatal("restore did not proceed")
+	}
+}
+
+func TestRestoreNullFingerprintIsReportedNotAssumed(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, "", true)
+	_, target := testDB(t)
+
+	var warnings []string
+	if _, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+		Warn: func(s string) { warnings = append(warnings, s) },
+	}); err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+	if !strings.Contains(strings.Join(warnings, "\n"), "no key fingerprint") {
+		t.Fatalf("a null fingerprint must be reported, got: %v", warnings)
+	}
+}
+
+func TestRestoreReplaysIndexes(t *testing.T) {
+	client, _ := testDB(t)
+	ctx := context.Background()
+	_, srcDB := testDB(t)
+	seed(t, client, srcDB)
+	if _, err := client.Database(srcDB).Collection("servers").Indexes().
+		CreateOne(ctx, mongo.IndexModel{Keys: bson.D{{Key: "name", Value: 1}}}); err != nil {
+		t.Fatalf("create index: %v", err)
+	}
+	path, _ := dumpToFile(t, DumpOptions{Client: client, Database: srcDB, KeyHex: validKeyHex})
+	archive, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer archive.Close()
+
+	_, target := testDB(t)
+	res, err := Restore(ctx, RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+	})
+	if err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+
+	cur, err := client.Database(target).Collection("servers").Indexes().List(ctx)
+	if err != nil {
+		t.Fatalf("list indexes: %v", err)
+	}
+	var specs []bson.M
+	if err := cur.All(ctx, &specs); err != nil {
+		t.Fatalf("read indexes: %v", err)
+	}
+	found := false
+	for _, s := range specs {
+		if s["name"] == "name_1" {
+			found = true
+		}
+	}
+	if !found {
+		t.Fatalf("index name_1 not replayed; got %v", specs)
+	}
+	for _, c := range res.Collections {
+		if c.Name == "servers" && c.Indexes < 1 {
+			t.Fatal("result reports no indexes created for servers")
+		}
+	}
+}
+
+// TestRestoreAbortsWhenAUniqueIndexCannotBuild replaces the brief's
+// TestRestoreAbortsOnUniqueIndexViolation per ruling 1: creating an index on
+// the target also creates the collection, so that version's assertion
+// (err == nil) would have passed on the wrong error (ErrTargetNotEmpty), and
+// Force: true does not rescue it because the drop removes the index before
+// replayIndexes runs. This version builds a hand-made archive whose data and
+// index specification directly contradict each other, which is the actual
+// shape of a corrupted archive that replayIndexes must refuse to load.
+func TestRestoreAbortsWhenAUniqueIndexCannotBuild(t *testing.T) {
+	client, _ := testDB(t)
+	ctx := context.Background()
+
+	// Built by hand rather than dumped: two documents that collide on email
+	// alongside an index specification declaring email unique. No live database
+	// would let those coexist, which is exactly the point — this is the shape
+	// of a corrupted or hand-edited archive, and restore must refuse rather
+	// than load the rows and leave the index missing.
+	a, err := bson.Marshal(bson.M{"email": "a@example.com"})
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	b, err := bson.Marshal(bson.M{"email": "a@example.com"})
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+
+	path := filepath.Join(t.TempDir(), "dupes.tar.gz")
+	f, err := os.Create(path)
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	w := NewWriter(f)
+	entry, err := w.WriteCollection("users", [][]byte{a, b})
+	if err != nil {
+		t.Fatalf("WriteCollection: %v", err)
+	}
+	if err := w.WriteIndexes("users",
+		[]byte(`[{"name":"email_1","key":{"email":1},"unique":true}]`)); err != nil {
+		t.Fatalf("WriteIndexes: %v", err)
+	}
+	fp, err := FingerprintHex(validKeyHex)
+	if err != nil {
+		t.Fatalf("FingerprintHex: %v", err)
+	}
+	if err := w.Close(Manifest{
+		FormatVersion:  FormatVersion,
+		CreatedAt:      time.Now().UTC(),
+		MongoDB:        "handmade",
+		KeyFingerprint: &fp,
+		Collections:    []CollectionEntry{entry},
+	}); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if err := f.Close(); err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	archive, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer archive.Close()
+
+	_, target := testDB(t)
+	if _, err := Restore(ctx, RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+	}); !errors.Is(err, ErrIndexBuild) {
+		t.Fatalf("got %v, want ErrIndexBuild", err)
+	} else if !strings.Contains(err.Error(), "email_1") {
+		t.Fatalf("the error must name the offending index, got: %v", err)
+	}
+}
+
+func TestRestoreSameVersionDoesNotWarnAboutIt(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	_, target := testDB(t)
+
+	var warnings []string
+	if _, err := Restore(context.Background(), RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+		Warn: func(s string) { warnings = append(warnings, s) },
+	}); err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+	for _, w := range warnings {
+		if strings.Contains(w, "you are restoring onto") {
+			t.Fatalf("same-version restore warned about a version gap: %s", w)
+		}
+	}
+}
+
+func TestCompoundIndexKeyOrderIsPreserved(t *testing.T) {
+	spec, name, unique, ok, err := indexSpecFrom([]byte(
+		`{"v":2,"key":{"b":1,"a":1},"name":"b_1_a_1","ns":"db.c"}`))
+	if err != nil {
+		t.Fatalf("indexSpecFrom: %v", err)
+	}
+	if !ok {
+		t.Fatal("spec rejected")
+	}
+	if unique {
+		t.Fatal("index reported as unique")
+	}
+	if name != "b_1_a_1" {
+		t.Fatalf("name %q", name)
+	}
+
+	var keys bson.D
+	for _, e := range spec {
+		switch e.Key {
+		case "v", "ns":
+			t.Fatalf("%q must be stripped before createIndexes, got %v", e.Key, spec)
+		case "key":
+			d, isD := e.Value.(bson.D)
+			if !isD {
+				t.Fatalf("key is %T, want bson.D", e.Value)
+			}
+			keys = d
+		}
+	}
+	// Compound index key order is significant, so it is carried through
+	// verbatim rather than reconstructed from an unordered map.
+	if len(keys) != 2 || keys[0].Key != "b" || keys[1].Key != "a" {
+		t.Fatalf("key order not preserved, got %v", keys)
+	}
+}
+
+func TestIdIndexIsSkipped(t *testing.T) {
+	_, name, _, ok, err := indexSpecFrom([]byte(`{"v":2,"key":{"_id":1},"name":"_id_"}`))
+	if err != nil {
+		t.Fatalf("indexSpecFrom: %v", err)
+	}
+	if ok {
+		t.Fatal("_id_ must be skipped; MongoDB creates it itself")
+	}
+	if name != "_id_" {
+		t.Fatalf("name %q", name)
+	}
+}
+
+// TestRestoreReplaysPartialUniqueIndex is the regression guard for the defect
+// that made a restore abort on any real database: a partial unique index —
+// this codebase has them on workflow_steps and settings — replayed as a full
+// unique index hits duplicate keys, and a failing unique index is fatal.
+func TestRestoreReplaysPartialUniqueIndex(t *testing.T) {
+	client, _ := testDB(t)
+	ctx := context.Background()
+	_, srcDB := testDB(t)
+	seed(t, client, srcDB)
+
+	coll := client.Database(srcDB).Collection("workflow_steps")
+	docs := []any{
+		bson.M{"instance_id": "i1", "slug": "same", "source": "default"},
+		bson.M{"instance_id": "i1", "slug": "same", "source": "custom"},
+		bson.M{"instance_id": "i1", "slug": "same", "source": "custom"},
+	}
+	if _, err := coll.InsertMany(ctx, docs); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	if _, err := coll.Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys: bson.D{{Key: "instance_id", Value: 1}, {Key: "slug", Value: 1}},
+		Options: options.Index().SetName("default_step_slug").SetUnique(true).
+			SetPartialFilterExpression(bson.M{"source": "default"}),
+	}); err != nil {
+		t.Fatalf("create partial index: %v", err)
+	}
+
+	path, _ := dumpToFile(t, DumpOptions{Client: client, Database: srcDB, KeyHex: validKeyHex})
+	archive, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer archive.Close()
+
+	_, target := testDB(t)
+	if _, err := Restore(ctx, RestoreOptions{
+		Client: client, Database: target, Archive: archive, KeyHex: validKeyHex,
+	}); err != nil {
+		t.Fatalf("Restore: %v", err)
+	}
+
+	cur, err := client.Database(target).Collection("workflow_steps").Indexes().List(ctx)
+	if err != nil {
+		t.Fatalf("list indexes: %v", err)
+	}
+	var specs []bson.M
+	if err := cur.All(ctx, &specs); err != nil {
+		t.Fatalf("read indexes: %v", err)
+	}
+	for _, s := range specs {
+		if s["name"] != "default_step_slug" {
+			continue
+		}
+		if s["unique"] != true {
+			t.Fatalf("index lost its uniqueness: %v", s)
+		}
+		if s["partialFilterExpression"] == nil {
+			t.Fatalf("partialFilterExpression was dropped: %v", s)
+		}
+		keys, isD := s["key"].(bson.D)
+		if isD && (len(keys) != 2 || keys[0].Key != "instance_id" || keys[1].Key != "slug") {
+			t.Fatalf("compound key order not preserved: %v", keys)
+		}
+		return
+	}
+	t.Fatalf("partial unique index not replayed; got %v", specs)
+}

backup/verify.go

@@ -0,0 +1,197 @@
+package backup
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/cryptobox"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+// VerifyOptions configures a verification. Client and Database are optional;
+// supplying them turns on the live probe.
+type VerifyOptions struct {
+	Archive  *Reader
+	KeyHex   string
+	Client   *mongo.Client
+	Database string
+}
+
+// VerifyReport is what verify found.
+type VerifyReport struct {
+	ArchiveFingerprint *string
+	KeyFingerprint     *string
+	KeyMatchesArchive  bool
+
+	// ProbeAttempted is false when no client was supplied, and also when the
+	// database holds no ciphertext to probe.
+	ProbeAttempted  bool
+	ProbeCollection string
+	ProbeDecrypted  bool
+
+	Problems []string
+}
+
+// OK reports whether this archive is usable with the key in hand.
+func (r VerifyReport) OK() bool { return len(r.Problems) == 0 }
+
+func (r *VerifyReport) problem(format string, args ...any) {
+	r.Problems = append(r.Problems, fmt.Sprintf(format, args...))
+}
+
+// Verify checks an already-opened archive against the key in hand and, when a
+// client is supplied, against a live database.
+//
+// Open has already verified every member's checksum, so integrity is not
+// rechecked here. What this adds is the question an operator actually has:
+// will the key I hold open the data this archive carries. A fingerprint
+// comparison proves two archives agree; only the probe proves the key opens
+// real ciphertext.
+func Verify(ctx context.Context, opt VerifyOptions) (VerifyReport, error) {
+	m := opt.Archive.Manifest()
+	rep := VerifyReport{ArchiveFingerprint: m.KeyFingerprint}
+
+	if opt.KeyHex != "" {
+		fp, err := FingerprintHex(opt.KeyHex)
+		if err != nil {
+			return rep, err
+		}
+		rep.KeyFingerprint = &fp
+	}
+
+	switch {
+	case m.KeyFingerprint == nil && rep.KeyFingerprint == nil:
+		rep.problem("neither the archive nor this environment names a key; nothing here " +
+			"proves the archive's ciphertext can ever be read")
+	case m.KeyFingerprint == nil:
+		rep.problem("the archive carries no key fingerprint, so it cannot be matched " +
+			"against the key you hold")
+	case rep.KeyFingerprint == nil:
+		rep.problem("KEY_ENCRYPTION_KEY is not set, so the archive's fingerprint %s "+
+			"cannot be checked against anything", *m.KeyFingerprint)
+	case *m.KeyFingerprint == *rep.KeyFingerprint:
+		rep.KeyMatchesArchive = true
+	default:
+		rep.problem("key mismatch: archive fingerprint %s, your key fingerprints as %s",
+			*m.KeyFingerprint, *rep.KeyFingerprint)
+	}
+
+	if opt.Client == nil || opt.Database == "" || opt.KeyHex == "" {
+		return rep, nil
+	}
+	if err := probe(ctx, opt, &rep); err != nil {
+		return rep, err
+	}
+	return rep, nil
+}
+
+// probe reads one ciphertext field from the live database and tries to open it.
+func probe(ctx context.Context, opt VerifyOptions, rep *VerifyReport) error {
+	key, err := ParseKey(opt.KeyHex)
+	if err != nil {
+		return err
+	}
+	for _, coll := range CiphertextCollections() {
+		ciphertext, ok, err := findCiphertext(ctx, opt.Client.Database(opt.Database), coll)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			continue
+		}
+		rep.ProbeAttempted = true
+		rep.ProbeCollection = coll
+		if _, err := cryptobox.Open(key, ciphertext); err != nil {
+			rep.problem("the key in hand does not decrypt live ciphertext in %s", coll)
+			return nil
+		}
+		rep.ProbeDecrypted = true
+		return nil
+	}
+	// No ciphertext anywhere is an ordinary state — a deployment that has
+	// stored no secrets, keys or SSO configuration yet — and is not a failure.
+	return nil
+}
+
+// ciphertextFields names, per collection, the fields that hold hex ciphertext.
+// A value is a candidate only if it is a hex string long enough to carry a GCM
+// nonce and tag, which is what keeps this from probing a plaintext field.
+//
+// This map MIRRORS BY HAND the bson tags in server/internal/models, which this
+// package cannot import: shared/ is a separate module and models is under
+// server/internal. It must change in the same commit as any rename of the
+// fields below — the same mirrored-constant hazard as web/lib/targets.ts and
+// services.MaxWorkloadLogLines. The sources are:
+//
+//	keys             — models/key.go: private_key_enc, passphrase_enc
+//	secrets          — models/secret.go: encrypted_value
+//	auth_providers   — models/auth_provider.go: client_secret_enc
+//	console_sessions — models/console_session.go: rdp_user_enc, rdp_pass_enc
+//
+// settings is deliberately absent: it holds no ciphertext at all. The ESO read
+// token is stored as a SHA-256 hash, which no key opens.
+var ciphertextFields = map[string][]string{
+	"keys":             {"private_key_enc", "passphrase_enc"},
+	"secrets":          {"encrypted_value"},
+	"auth_providers":   {"client_secret_enc"},
+	"console_sessions": {"rdp_user_enc", "rdp_pass_enc"},
+}
+
+func findCiphertext(ctx context.Context, db *mongo.Database, coll string) (string, bool, error) {
+	fields, ok := ciphertextFields[coll]
+	if !ok {
+		return "", false, nil
+	}
+	cur, err := db.Collection(coll).Find(ctx, bson.M{})
+	if err != nil {
+		return "", false, fmt.Errorf("probe %s: %w", coll, err)
+	}
+	defer cur.Close(ctx)
+
+	for cur.Next(ctx) {
+		var doc bson.M
+		if err := cur.Decode(&doc); err != nil {
+			return "", false, fmt.Errorf("probe %s: %w", coll, err)
+		}
+		for _, f := range fields {
+			if v, ok := looksLikeCiphertext(doc[f]); ok {
+				return v, true, nil
+			}
+		}
+	}
+	return "", false, cur.Err()
+}
+
+// looksLikeCiphertext accepts a hex string long enough to be a sealed value. It
+// descends into a sub-document so a field that holds a map of sealed values is
+// still reachable.
+func looksLikeCiphertext(v any) (string, bool) {
+	switch t := v.(type) {
+	case string:
+		// 12-byte nonce plus a 16-byte tag is 56 hex characters before any
+		// plaintext at all, so anything shorter is not a sealed value.
+		if len(t) < 56 || !isHex(t) {
+			return "", false
+		}
+		return t, true
+	case bson.M:
+		for _, inner := range t {
+			if s, ok := looksLikeCiphertext(inner); ok {
+				return s, true
+			}
+		}
+	}
+	return "", false
+}
+
+func isHex(s string) bool {
+	for _, c := range s {
+		switch {
+		case c >= '0' && c <= '9', c >= 'a' && c <= 'f', c >= 'A' && c <= 'F':
+		default:
+			return false
+		}
+	}
+	return true
+}

backup/verify_test.go

@@ -0,0 +1,156 @@
+package backup
+
+import (
+	"context"
+	"testing"
+
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/cryptobox"
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+func TestVerifyMatchingKey(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+
+	rep, err := Verify(context.Background(), VerifyOptions{Archive: archive, KeyHex: validKeyHex})
+	if err != nil {
+		t.Fatalf("Verify: %v", err)
+	}
+	if !rep.KeyMatchesArchive {
+		t.Fatal("matching key reported as a mismatch")
+	}
+	if rep.ProbeAttempted {
+		t.Fatal("probe ran with no client supplied")
+	}
+	if !rep.OK() {
+		t.Fatalf("report not OK: %v", rep.Problems)
+	}
+}
+
+func TestVerifyMismatchedKeyIsNotOK(t *testing.T) {
+	client, _ := testDB(t)
+	archive := archiveOf(t, client, validKeyHex, false)
+	other := "0000000000000000000000000000000000000000000000000000000000000002"
+
+	rep, err := Verify(context.Background(), VerifyOptions{Archive: archive, KeyHex: other})
+	if err != nil {
+		t.Fatalf("Verify: %v", err)
+	}
+	if rep.KeyMatchesArchive {
+		t.Fatal("mismatched key reported as matching")
+	}
+	if rep.OK() {
+		t.Fatal("a mismatch must not report OK")
+	}
+}
+
+func TestVerifyProbeDecryptsLiveCiphertext(t *testing.T) {
+	client, dbName := testDB(t)
+	ctx := context.Background()
+
+	key, err := ParseKey(validKeyHex)
+	if err != nil {
+		t.Fatalf("ParseKey: %v", err)
+	}
+	sealed, err := cryptobox.Seal(key, "s3cret")
+	if err != nil {
+		t.Fatalf("Seal: %v", err)
+	}
+	if _, err := client.Database(dbName).Collection("secrets").InsertOne(ctx, bson.M{
+		"instance_id": "i1",
+		"values":      bson.M{"TOKEN": sealed},
+	}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	path, _ := dumpToFile(t, DumpOptions{Client: client, Database: dbName, KeyHex: validKeyHex})
+	archive, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer archive.Close()
+
+	rep, err := Verify(ctx, VerifyOptions{
+		Archive: archive, KeyHex: validKeyHex, Client: client, Database: dbName,
+	})
+	if err != nil {
+		t.Fatalf("Verify: %v", err)
+	}
+	if !rep.ProbeAttempted {
+		t.Fatal("probe did not run with a client supplied")
+	}
+	if !rep.ProbeDecrypted {
+		t.Fatalf("probe failed to decrypt live ciphertext: %v", rep.Problems)
+	}
+	if rep.ProbeCollection != "secrets" {
+		t.Fatalf("probe collection %q, want secrets", rep.ProbeCollection)
+	}
+	if !rep.OK() {
+		t.Fatalf("report not OK: %v", rep.Problems)
+	}
+}
+
+func TestVerifyProbeFailsWithWrongKey(t *testing.T) {
+	client, dbName := testDB(t)
+	ctx := context.Background()
+
+	key, err := ParseKey(validKeyHex)
+	if err != nil {
+		t.Fatalf("ParseKey: %v", err)
+	}
+	sealed, err := cryptobox.Seal(key, "s3cret")
+	if err != nil {
+		t.Fatalf("Seal: %v", err)
+	}
+	if _, err := client.Database(dbName).Collection("secrets").InsertOne(ctx, bson.M{
+		"values": bson.M{"TOKEN": sealed},
+	}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	other := "0000000000000000000000000000000000000000000000000000000000000002"
+	path, _ := dumpToFile(t, DumpOptions{Client: client, Database: dbName, KeyHex: other})
+	archive, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer archive.Close()
+
+	rep, err := Verify(ctx, VerifyOptions{
+		Archive: archive, KeyHex: other, Client: client, Database: dbName,
+	})
+	if err != nil {
+		t.Fatalf("Verify: %v", err)
+	}
+	if rep.ProbeDecrypted {
+		t.Fatal("probe decrypted with the wrong key")
+	}
+	if rep.OK() {
+		t.Fatal("a failed probe must not report OK")
+	}
+}
+
+func TestVerifyProbeAbsentCiphertextIsNotAFailure(t *testing.T) {
+	client, dbName := testDB(t)
+	seed(t, client, dbName)
+
+	path, _ := dumpToFile(t, DumpOptions{Client: client, Database: dbName, KeyHex: validKeyHex})
+	archive, err := Open(path)
+	if err != nil {
+		t.Fatalf("Open: %v", err)
+	}
+	defer archive.Close()
+
+	rep, err := Verify(context.Background(), VerifyOptions{
+		Archive: archive, KeyHex: validKeyHex, Client: client, Database: dbName,
+	})
+	if err != nil {
+		t.Fatalf("Verify: %v", err)
+	}
+	if rep.ProbeAttempted {
+		t.Fatal("probe claims to have run against a database with no ciphertext")
+	}
+	if !rep.OK() {
+		t.Fatalf("a database storing no secrets must still verify: %v", rep.Problems)
+	}
+}

cmd/lkctl/main.go

@@ -15,7 +15,7 @@ import (
 	"strings"
 	"time"
 
-	"gitea.hostxtra.co.uk/mrhid6/vantage/shared/license"
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/license"
 	"github.com/google/uuid"
 	"github.com/hyperboloide/lk"
 )

go.mod

@@ -1,4 +1,4 @@
-module gitea.hostxtra.co.uk/mrhid6/vantage/shared
+module gitea.hostxtra.co.uk/vantage/vantage-shared
 
 go 1.26
 
@@ -7,6 +7,7 @@ require (
 	github.com/hyperboloide/lk v0.0.0-20251220053519-b291812e3216
 	go.mongodb.org/mongo-driver/v2 v2.8.0
 	golang.org/x/crypto v0.54.0
+	google.golang.org/grpc v1.64.0
 )
 
 require (
@@ -15,6 +16,10 @@ require (
 	github.com/xdg-go/scram v1.2.0 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
+	golang.org/x/net v0.56.0 // indirect
 	golang.org/x/sync v0.22.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 )

go.sum

@@ -31,6 +31,8 @@ golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.56.0 h1:Rw8j/hFzGvJUZwNBXnAtf5sVDVt+65SK2C7IxCxZt5o=
+golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
@@ -40,6 +42,8 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -52,5 +56,11 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 h1:NnYq6UN9ReLM9/Y01KWNOWyI5xQ9kbIms5GGJVwS/Yc=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
+google.golang.org/grpc v1.64.0 h1:KH3VH9y/MgNQg1dE7b3XfVK0GsPSIzJwdF617gUSbvY=
+google.golang.org/grpc v1.64.0/go.mod h1:oxjF8E3FBnjp+/gVFYdWacaLDx9na1aqy9oovLpxQYg=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

grpc/codec/codec.go

@@ -0,0 +1,19 @@
+package codec
+
+import (
+	"encoding/json"
+)
+
+type JSONCodec struct{}
+
+func (JSONCodec) Marshal(v interface{}) ([]byte, error) {
+	return json.Marshal(v)
+}
+
+func (JSONCodec) Unmarshal(data []byte, v interface{}) error {
+	return json.Unmarshal(data, v)
+}
+
+func (JSONCodec) Name() string {
+	return "proto"
+}

grpc/pb/vantage.pb.go

@@ -0,0 +1,685 @@
+package pb
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type RegisterRequest struct {
+	ServerId    string `json:"server_id"`
+	PreRegToken string `json:"pre_reg_token"`
+	Hostname    string `json:"hostname"`
+	IpAddress   string `json:"ip_address"`
+	OsInfo      string `json:"os_info"`
+}
+
+type RegisterResponse struct {
+	AgentToken string `json:"agent_token"`
+}
+
+type SyncRequest struct {
+	ServerId     string `json:"server_id"`
+	AgentToken   string `json:"agent_token"`
+	AgentVersion string `json:"agent_version,omitempty"`
+}
+
+type SyncResponse struct {
+	PublicKeys []string `json:"public_keys"`
+	// CollectPackages tells the agent whether this instance's licence grants
+	// vulnerability scanning. Absent decodes as false, which is the safe
+	// direction: an older server leaves agents collecting nothing.
+	CollectPackages bool `json:"collect_packages,omitempty"`
+}
+
+type OSRelease struct {
+	Family string `json:"family"`
+	// VersionId is not optional: Ubuntu 22.04 and 24.04 publish different fixed
+	// versions for the same CVE, so a scan without it is guesswork.
+	VersionId string `json:"version_id"`
+	Arch      string `json:"arch,omitempty"`
+}
+
+type InstalledPackage struct {
+	Name    string `json:"name"`
+	Version string `json:"version"`
+	Epoch   int32  `json:"epoch,omitempty"`
+	Arch    string `json:"arch,omitempty"`
+	// SourceName is what the Debian and Ubuntu feeds are keyed on: one advisory
+	// against "openssl" covers libssl3, openssl and libssl-dev.
+	SourceName string `json:"source_name,omitempty"`
+}
+
+// ReportPackagesRequest carries a server's installed package set.
+//
+// The agent calls twice at most: first with Packages empty, offering only the
+// hash. If the server already holds it, NeedFull is false and the ~150KB body
+// is never sent.
+type ReportPackagesRequest struct {
+	ServerId   string             `json:"server_id"`
+	AgentToken string             `json:"agent_token"`
+	Hash       string             `json:"hash"`
+	Os         OSRelease          `json:"os"`
+	Packages   []InstalledPackage `json:"packages,omitempty"`
+}
+
+type ReportPackagesResponse struct {
+	NeedFull bool `json:"need_full"`
+}
+
+type UploadKeyRequest struct {
+	ServerId   string `json:"server_id"`
+	AgentToken string `json:"agent_token"`
+	PublicKey  string `json:"public_key"`
+	Label      string `json:"label"`
+	PrivateKey string `json:"private_key,omitempty"`
+}
+
+type UploadKeyResponse struct {
+	KeyId string `json:"key_id"`
+}
+
+type PackageUpdate struct {
+	Name           string `json:"name"`
+	CurrentVersion string `json:"current_version,omitempty"`
+	NewVersion     string `json:"new_version"`
+}
+
+type ReportUpdatesRequest struct {
+	ServerId   string          `json:"server_id"`
+	AgentToken string          `json:"agent_token"`
+	Updates    []PackageUpdate `json:"updates"`
+}
+
+type ReportUpdatesResponse struct{}
+
+type CPUReport struct {
+	Model    string  `json:"model,omitempty"`
+	Cores    int     `json:"cores,omitempty"`
+	UsagePct float64 `json:"usage_pct"`
+	Load1    float64 `json:"load1,omitempty"`
+}
+type MemReport struct {
+	TotalBytes uint64 `json:"total_bytes"`
+	UsedBytes  uint64 `json:"used_bytes"`
+}
+type PartitionReport struct {
+	Device     string `json:"device"`
+	Mountpoint string `json:"mountpoint"`
+	Fstype     string `json:"fstype,omitempty"`
+	TotalBytes uint64 `json:"total_bytes"`
+	UsedBytes  uint64 `json:"used_bytes"`
+}
+type InventoryReport struct {
+	ServerId       string            `json:"server_id"`
+	AgentToken     string            `json:"agent_token"`
+	IncludeStatic  bool              `json:"include_static"`
+	CPU            *CPUReport        `json:"cpu,omitempty"`
+	Memory         *MemReport        `json:"memory,omitempty"`
+	SwapTotal      uint64            `json:"swap_total"`
+	SwapUsed       uint64            `json:"swap_used"`
+	Partitions     []PartitionReport `json:"partitions,omitempty"`
+	Kernel         string            `json:"kernel,omitempty"`
+	RebootRequired bool              `json:"reboot_required,omitempty"`
+}
+type InventoryReportResponse struct{}
+
+type MonitorSpec struct {
+	MonitorId      string `json:"monitor_id"`
+	Type           string `json:"type"`
+	URL            string `json:"url,omitempty"`
+	Host           string `json:"host,omitempty"`
+	Port           int    `json:"port,omitempty"`
+	Method         string `json:"method,omitempty"`
+	ExpectedStatus int    `json:"expected_status,omitempty"`
+	Keyword        string `json:"keyword,omitempty"`
+	TLSWarnDays    int    `json:"tls_warn_days,omitempty"`
+	Insecure       bool   `json:"insecure,omitempty"`
+	IntervalSec    int    `json:"interval_sec"`
+	Retries        int    `json:"retries"`
+}
+type SyncMonitorsRequest struct {
+	ServerId   string `json:"server_id"`
+	AgentToken string `json:"agent_token"`
+}
+type SyncMonitorsResponse struct {
+	Monitors []MonitorSpec `json:"monitors,omitempty"`
+}
+type CheckResult struct {
+	MonitorId      string `json:"monitor_id"`
+	Up             bool   `json:"up"`
+	LatencyMs      int    `json:"latency_ms"`
+	Message        string `json:"message,omitempty"`
+	CertExpiryUnix int64  `json:"cert_expiry_unix,omitempty"`
+}
+type ReportChecksRequest struct {
+	ServerId   string        `json:"server_id"`
+	AgentToken string        `json:"agent_token"`
+	Results    []CheckResult `json:"results,omitempty"`
+}
+type ReportChecksResponse struct{}
+
+type ApplyUpdatesCmd struct{}
+
+type OpenProxyCmd struct {
+	ProxyId string `json:"proxy_id"`
+	Port    uint32 `json:"port"`
+}
+
+type ProxyOpen struct {
+	ServerId   string `json:"server_id"`
+	AgentToken string `json:"agent_token"`
+	ProxyId    string `json:"proxy_id"`
+}
+
+type ProxyClose struct {
+	Reason string `json:"reason,omitempty"`
+}
+
+type ProxyClientMsg struct {
+	Open  *ProxyOpen  `json:"open,omitempty"`
+	Data  []byte      `json:"data,omitempty"`
+	Close *ProxyClose `json:"close,omitempty"`
+}
+
+type ProxyServerMsg struct {
+	Data  []byte      `json:"data,omitempty"`
+	Close *ProxyClose `json:"close,omitempty"`
+}
+
+type ServerCommand struct {
+	CommandId        string               `json:"command_id"`
+	GenerateKey      *GenerateKeyCmd      `json:"generate_key,omitempty"`
+	DeleteKey        *DeleteKeyCmd        `json:"delete_key,omitempty"`
+	UpdateAgent      *UpdateAgentCmd      `json:"update_agent,omitempty"`
+	ApplyUpdates     *ApplyUpdatesCmd     `json:"apply_updates,omitempty"`
+	RunStep          *RunStepCmd          `json:"run_step,omitempty"`
+	CleanupWorkspace *CleanupWorkspaceCmd `json:"cleanup_workspace,omitempty"`
+	OpenProxy        *OpenProxyCmd        `json:"open_proxy,omitempty"`
+	Ping             *PingCmd             `json:"ping,omitempty"`
+
+	RefreshWorkloads *RefreshWorkloadsCmd `json:"refresh_workloads,omitempty"`
+	ControlWorkload  *ControlWorkloadCmd  `json:"control_workload,omitempty"`
+	WorkloadLogs     *WorkloadLogsCmd     `json:"workload_logs,omitempty"`
+}
+
+// PingCmd is a server-originated liveness beat. It carries nothing and expects
+// no reply: its arrival is the entire message. See the .proto for why gRPC
+// keepalive is not sufficient on its own.
+type PingCmd struct{}
+
+type CleanupWorkspaceCmd struct {
+	WorkspaceId string `json:"workspace_id"`
+}
+
+type DeleteKeyCmd struct {
+	Label string `json:"label"`
+}
+
+type UpdateAgentCmd struct {
+	Version      string `json:"version"`
+	GiteaBaseURL string `json:"gitea_base_url"`
+}
+
+type GenerateKeyCmd struct {
+	Label      string `json:"label"`
+	KeyType    string `json:"key_type,omitempty"`
+	KeySize    int    `json:"key_size,omitempty"`
+	Passphrase string `json:"passphrase,omitempty"`
+	Comment    string `json:"comment,omitempty"`
+}
+
+type AgentMessage struct {
+	ServerId   string           `json:"server_id"`
+	AgentToken string           `json:"agent_token"`
+	Ready      *AgentReady      `json:"ready,omitempty"`
+	Result     *CommandResult   `json:"result,omitempty"`
+	StepResult *StepResult      `json:"step_result,omitempty"`
+	StepOutput *StepOutputChunk `json:"step_output,omitempty"`
+
+	WorkloadLogsResult *WorkloadLogsResult `json:"workload_logs_result,omitempty"`
+}
+
+type AgentReady struct{}
+
+type CommandResult struct {
+	CommandId string `json:"command_id"`
+	Success   bool   `json:"success"`
+	Message   string `json:"message"`
+}
+
+type RunStepCmd struct {
+	Interpreter    string            `json:"interpreter"`
+	Script         string            `json:"script"`
+	Env            map[string]string `json:"env,omitempty"`
+	TimeoutSeconds int               `json:"timeout_seconds,omitempty"`
+
+	WorkspaceId string `json:"workspace_id,omitempty"`
+}
+
+type StepResult struct {
+	CommandId string            `json:"command_id"`
+	ExitCode  int               `json:"exit_code"`
+	Stdout    string            `json:"stdout,omitempty"`
+	Stderr    string            `json:"stderr,omitempty"`
+	OutputEnv map[string]string `json:"output_env,omitempty"`
+}
+
+type StepOutputChunk struct {
+	CommandId string `json:"command_id"`
+	Seq       uint64 `json:"seq"`
+	Data      []byte `json:"data,omitempty"`
+	Eof       bool   `json:"eof,omitempty"`
+}
+
+type Vantage_CommandStreamServer interface {
+	Send(*ServerCommand) error
+	Recv() (*AgentMessage, error)
+	grpc.ServerStream
+}
+
+type vantageCommandStreamServer struct {
+	grpc.ServerStream
+}
+
+func (s *vantageCommandStreamServer) Send(m *ServerCommand) error {
+	return s.ServerStream.SendMsg(m)
+}
+
+func (s *vantageCommandStreamServer) Recv() (*AgentMessage, error) {
+	m := new(AgentMessage)
+	if err := s.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+type Vantage_CommandStreamClient interface {
+	Send(*AgentMessage) error
+	Recv() (*ServerCommand, error)
+	grpc.ClientStream
+}
+
+type vantageCommandStreamClient struct {
+	grpc.ClientStream
+}
+
+func (c *vantageCommandStreamClient) Send(m *AgentMessage) error {
+	return c.ClientStream.SendMsg(m)
+}
+
+func (c *vantageCommandStreamClient) Recv() (*ServerCommand, error) {
+	m := new(ServerCommand)
+	if err := c.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+type Vantage_ProxyStreamServer interface {
+	Send(*ProxyServerMsg) error
+	Recv() (*ProxyClientMsg, error)
+	grpc.ServerStream
+}
+
+type vantageProxyStreamServer struct {
+	grpc.ServerStream
+}
+
+func (s *vantageProxyStreamServer) Send(m *ProxyServerMsg) error {
+	return s.ServerStream.SendMsg(m)
+}
+
+func (s *vantageProxyStreamServer) Recv() (*ProxyClientMsg, error) {
+	m := new(ProxyClientMsg)
+	if err := s.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+type Vantage_ProxyStreamClient interface {
+	Send(*ProxyClientMsg) error
+	Recv() (*ProxyServerMsg, error)
+	CloseSend() error
+	grpc.ClientStream
+}
+
+type vantageProxyStreamClient struct {
+	grpc.ClientStream
+}
+
+func (c *vantageProxyStreamClient) Send(m *ProxyClientMsg) error {
+	return c.ClientStream.SendMsg(m)
+}
+
+func (c *vantageProxyStreamClient) Recv() (*ProxyServerMsg, error) {
+	m := new(ProxyServerMsg)
+	if err := c.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func _Vantage_ProxyStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(VantageServer).ProxyStream(&vantageProxyStreamServer{stream})
+}
+
+type VantageServer interface {
+	Register(context.Context, *RegisterRequest) (*RegisterResponse, error)
+	SyncKeys(context.Context, *SyncRequest) (*SyncResponse, error)
+	UploadGeneratedKey(context.Context, *UploadKeyRequest) (*UploadKeyResponse, error)
+	ReportUpdates(context.Context, *ReportUpdatesRequest) (*ReportUpdatesResponse, error)
+	ReportPackages(context.Context, *ReportPackagesRequest) (*ReportPackagesResponse, error)
+	ReportWorkloads(context.Context, *ReportWorkloadsRequest) (*ReportWorkloadsResponse, error)
+	ReportInventory(context.Context, *InventoryReport) (*InventoryReportResponse, error)
+	SyncMonitors(context.Context, *SyncMonitorsRequest) (*SyncMonitorsResponse, error)
+	ReportChecks(context.Context, *ReportChecksRequest) (*ReportChecksResponse, error)
+	CommandStream(Vantage_CommandStreamServer) error
+	ProxyStream(Vantage_ProxyStreamServer) error
+}
+
+type UnimplementedVantageServer struct{}
+
+func (UnimplementedVantageServer) Register(context.Context, *RegisterRequest) (*RegisterResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Register not implemented")
+}
+
+func (UnimplementedVantageServer) SyncKeys(context.Context, *SyncRequest) (*SyncResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method SyncKeys not implemented")
+}
+
+func (UnimplementedVantageServer) UploadGeneratedKey(context.Context, *UploadKeyRequest) (*UploadKeyResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method UploadGeneratedKey not implemented")
+}
+
+func (UnimplementedVantageServer) ReportUpdates(context.Context, *ReportUpdatesRequest) (*ReportUpdatesResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReportUpdates not implemented")
+}
+
+func (UnimplementedVantageServer) ReportPackages(context.Context, *ReportPackagesRequest) (*ReportPackagesResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReportPackages not implemented")
+}
+
+func (UnimplementedVantageServer) ReportInventory(context.Context, *InventoryReport) (*InventoryReportResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReportInventory not implemented")
+}
+
+func (UnimplementedVantageServer) SyncMonitors(context.Context, *SyncMonitorsRequest) (*SyncMonitorsResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method SyncMonitors not implemented")
+}
+
+func (UnimplementedVantageServer) ReportChecks(context.Context, *ReportChecksRequest) (*ReportChecksResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReportChecks not implemented")
+}
+
+func (UnimplementedVantageServer) CommandStream(Vantage_CommandStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method CommandStream not implemented")
+}
+
+func (UnimplementedVantageServer) ProxyStream(Vantage_ProxyStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method ProxyStream not implemented")
+}
+
+type VantageClient interface {
+	Register(ctx context.Context, in *RegisterRequest, opts ...grpc.CallOption) (*RegisterResponse, error)
+	SyncKeys(ctx context.Context, in *SyncRequest, opts ...grpc.CallOption) (*SyncResponse, error)
+	UploadGeneratedKey(ctx context.Context, in *UploadKeyRequest, opts ...grpc.CallOption) (*UploadKeyResponse, error)
+	ReportUpdates(ctx context.Context, in *ReportUpdatesRequest, opts ...grpc.CallOption) (*ReportUpdatesResponse, error)
+	ReportPackages(ctx context.Context, in *ReportPackagesRequest, opts ...grpc.CallOption) (*ReportPackagesResponse, error)
+	ReportWorkloads(ctx context.Context, in *ReportWorkloadsRequest, opts ...grpc.CallOption) (*ReportWorkloadsResponse, error)
+	ReportInventory(ctx context.Context, in *InventoryReport, opts ...grpc.CallOption) (*InventoryReportResponse, error)
+	SyncMonitors(ctx context.Context, in *SyncMonitorsRequest, opts ...grpc.CallOption) (*SyncMonitorsResponse, error)
+	ReportChecks(ctx context.Context, in *ReportChecksRequest, opts ...grpc.CallOption) (*ReportChecksResponse, error)
+	CommandStream(ctx context.Context, opts ...grpc.CallOption) (Vantage_CommandStreamClient, error)
+	ProxyStream(ctx context.Context, opts ...grpc.CallOption) (Vantage_ProxyStreamClient, error)
+}
+
+type keyManagerClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewVantageClient(cc grpc.ClientConnInterface) VantageClient {
+	return &keyManagerClient{cc}
+}
+
+func (c *keyManagerClient) Register(ctx context.Context, in *RegisterRequest, opts ...grpc.CallOption) (*RegisterResponse, error) {
+	out := new(RegisterResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/Register", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) SyncKeys(ctx context.Context, in *SyncRequest, opts ...grpc.CallOption) (*SyncResponse, error) {
+	out := new(SyncResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/SyncKeys", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) UploadGeneratedKey(ctx context.Context, in *UploadKeyRequest, opts ...grpc.CallOption) (*UploadKeyResponse, error) {
+	out := new(UploadKeyResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/UploadGeneratedKey", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) ReportUpdates(ctx context.Context, in *ReportUpdatesRequest, opts ...grpc.CallOption) (*ReportUpdatesResponse, error) {
+	out := new(ReportUpdatesResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/ReportUpdates", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) ReportPackages(ctx context.Context, in *ReportPackagesRequest, opts ...grpc.CallOption) (*ReportPackagesResponse, error) {
+	out := new(ReportPackagesResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/ReportPackages", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) ReportInventory(ctx context.Context, in *InventoryReport, opts ...grpc.CallOption) (*InventoryReportResponse, error) {
+	out := new(InventoryReportResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/ReportInventory", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) SyncMonitors(ctx context.Context, in *SyncMonitorsRequest, opts ...grpc.CallOption) (*SyncMonitorsResponse, error) {
+	out := new(SyncMonitorsResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/SyncMonitors", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) ReportChecks(ctx context.Context, in *ReportChecksRequest, opts ...grpc.CallOption) (*ReportChecksResponse, error) {
+	out := new(ReportChecksResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/ReportChecks", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *keyManagerClient) CommandStream(ctx context.Context, opts ...grpc.CallOption) (Vantage_CommandStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Vantage_ServiceDesc.Streams[0], "/vantage.v1.Vantage/CommandStream", opts...)
+	if err != nil {
+		return nil, err
+	}
+	return &vantageCommandStreamClient{stream}, nil
+}
+
+func (c *keyManagerClient) ProxyStream(ctx context.Context, opts ...grpc.CallOption) (Vantage_ProxyStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Vantage_ServiceDesc.Streams[1], "/vantage.v1.Vantage/ProxyStream", opts...)
+	if err != nil {
+		return nil, err
+	}
+	return &vantageProxyStreamClient{stream}, nil
+}
+
+func RegisterVantageServer(s grpc.ServiceRegistrar, srv VantageServer) {
+	s.RegisterService(&Vantage_ServiceDesc, srv)
+}
+
+var Vantage_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "vantage.v1.Vantage",
+	HandlerType: (*VantageServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{MethodName: "Register", Handler: _Vantage_Register_Handler},
+		{MethodName: "SyncKeys", Handler: _Vantage_SyncKeys_Handler},
+		{MethodName: "UploadGeneratedKey", Handler: _Vantage_UploadGeneratedKey_Handler},
+		{MethodName: "ReportUpdates", Handler: _Vantage_ReportUpdates_Handler},
+		{MethodName: "ReportPackages", Handler: _Vantage_ReportPackages_Handler},
+		{MethodName: "ReportWorkloads", Handler: _Vantage_ReportWorkloads_Handler},
+		{MethodName: "ReportInventory", Handler: _Vantage_ReportInventory_Handler},
+		{MethodName: "SyncMonitors", Handler: _Vantage_SyncMonitors_Handler},
+		{MethodName: "ReportChecks", Handler: _Vantage_ReportChecks_Handler},
+	},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "CommandStream",
+			Handler:       _Vantage_CommandStream_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+		{
+			StreamName:    "ProxyStream",
+			Handler:       _Vantage_ProxyStream_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+	},
+	Metadata: "vantage/v1/vantage.proto",
+}
+
+func _Vantage_Register_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(RegisterRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).Register(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/Register"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).Register(ctx, req.(*RegisterRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_SyncKeys_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SyncRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).SyncKeys(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/SyncKeys"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).SyncKeys(ctx, req.(*SyncRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_UploadGeneratedKey_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(UploadKeyRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).UploadGeneratedKey(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/UploadGeneratedKey"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).UploadGeneratedKey(ctx, req.(*UploadKeyRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_ReportUpdates_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReportUpdatesRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).ReportUpdates(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/ReportUpdates"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).ReportUpdates(ctx, req.(*ReportUpdatesRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_ReportPackages_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReportPackagesRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).ReportPackages(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/ReportPackages"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).ReportPackages(ctx, req.(*ReportPackagesRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_ReportInventory_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(InventoryReport)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).ReportInventory(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/ReportInventory"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).ReportInventory(ctx, req.(*InventoryReport))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_SyncMonitors_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SyncMonitorsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).SyncMonitors(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/SyncMonitors"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).SyncMonitors(ctx, req.(*SyncMonitorsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_ReportChecks_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReportChecksRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).ReportChecks(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/ReportChecks"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).ReportChecks(ctx, req.(*ReportChecksRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Vantage_CommandStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(VantageServer).CommandStream(&vantageCommandStreamServer{stream})
+}

grpc/pb/workloads.pb.go

@@ -0,0 +1,101 @@
+package pb
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// Workload registry messages. Hand-written like the rest of this package: the
+// .proto is the contract, this file is the Go side of it, and the two must be
+// changed together.
+
+// Workload is one container or one systemd unit.
+type Workload struct {
+	Kind      string   `json:"kind"`
+	Id        string   `json:"id"`
+	Name      string   `json:"name"`
+	State     string   `json:"state"`
+	Health    string   `json:"health,omitempty"`
+	Image     string   `json:"image,omitempty"`
+	Stack     string   `json:"stack,omitempty"`
+	Ports     []string `json:"ports,omitempty"`
+	Restarts  int32    `json:"restarts,omitempty"`
+	StartedAt string   `json:"started_at,omitempty"` // RFC3339, empty when not running
+	Protected bool     `json:"protected,omitempty"`
+}
+
+// ReportWorkloadsRequest carries what a server is running.
+//
+// Offer-then-send, the same handshake as ReportPackages: the agent calls once
+// with Workloads empty, and resends with the body only if NeedFull is set.
+type ReportWorkloadsRequest struct {
+	ServerId     string     `json:"server_id"`
+	AgentToken   string     `json:"agent_token"`
+	Hash         string     `json:"hash"`
+	DockerOk     bool       `json:"docker_ok"`
+	DockerError  string     `json:"docker_error,omitempty"`
+	SystemdOk    bool       `json:"systemd_ok"`
+	SystemdError string     `json:"systemd_error,omitempty"`
+	Workloads    []Workload `json:"workloads,omitempty"` // empty on the offer call
+	// Full marks the second call. It is not inferred from an empty Workloads
+	// slice: a host running nothing sends an empty list as its full report.
+	Full bool `json:"full,omitempty"`
+}
+
+type ReportWorkloadsResponse struct {
+	NeedFull bool `json:"need_full"`
+}
+
+// RefreshWorkloadsCmd carries no payload back. It makes the agent report
+// immediately through ReportWorkloads, so there is exactly one writer for the
+// server_workloads collection rather than two arriving by different routes.
+type RefreshWorkloadsCmd struct{}
+
+type ControlWorkloadCmd struct {
+	Kind   string `json:"kind"`
+	Id     string `json:"id"`
+	Action string `json:"action"` // start | stop | restart
+}
+
+type WorkloadLogsCmd struct {
+	Kind string `json:"kind"`
+	Id   string `json:"id"`
+	Tail int32  `json:"tail,omitempty"`
+}
+
+type WorkloadLogsResult struct {
+	CommandId string `json:"command_id"`
+	Text      string `json:"text,omitempty"`
+	Truncated bool   `json:"truncated,omitempty"`
+	Error     string `json:"error,omitempty"`
+}
+
+func (UnimplementedVantageServer) ReportWorkloads(context.Context, *ReportWorkloadsRequest) (*ReportWorkloadsResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReportWorkloads not implemented")
+}
+
+func (c *keyManagerClient) ReportWorkloads(ctx context.Context, in *ReportWorkloadsRequest, opts ...grpc.CallOption) (*ReportWorkloadsResponse, error) {
+	out := new(ReportWorkloadsResponse)
+	if err := c.cc.Invoke(ctx, "/vantage.v1.Vantage/ReportWorkloads", in, out, opts...); err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func _Vantage_ReportWorkloads_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReportWorkloadsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(VantageServer).ReportWorkloads(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{Server: srv, FullMethod: "/vantage.v1.Vantage/ReportWorkloads"}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(VantageServer).ReportWorkloads(ctx, req.(*ReportWorkloadsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}

provision/instance.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"time"
 
-	"gitea.hostxtra.co.uk/mrhid6/vantage/shared/models"
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/models"
 	"github.com/google/uuid"
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/mongo"

provision/user.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 	"time"
 
-	"gitea.hostxtra.co.uk/mrhid6/vantage/shared/models"
+	"gitea.hostxtra.co.uk/vantage/vantage-shared/models"
 	"github.com/google/uuid"
 	"go.mongodb.org/mongo-driver/v2/mongo"
 	"golang.org/x/crypto/bcrypt"