fix(offsite): 桶内记录密钥指纹，密钥不符时 sync 拒绝写入和清理，安装时大声提示

internal/offsite/keymark.go

@@ -0,0 +1,152 @@
+package offsite
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"regexp"
+	"sort"
+	"strings"
+)
+
+// keyMark is the one object the bucket holds in the clear: the KeyID of the
+// key every other object is sealed with. The id names the key without
+// revealing it, so a host holding another key (a reinstall that generated a
+// fresh one, an offsite.env from elsewhere) is refused before it writes an
+// object or prunes one the right key still opens.
+const keyMark = "felis-key-id"
+
+// ErrKeyMismatch is a bucket whose objects are sealed with another key.
+var ErrKeyMismatch = errors.New("offsite: the bucket's objects are sealed with another key")
+
+const keyMismatchFix = "set FELIS_OFFSITE_KEY in /etc/felis/offsite.env to the key they were sealed with, or point [offsite] at an empty bucket or prefix (docs/troubleshooting.md §16)"
+
+var keyIDPattern = regexp.MustCompile(`^[0-9a-f]{16}$`)
+
+// KeyFit is how CheckKey matched a key to a bucket. The zero value is no match:
+// what CheckKey returns with an error.
+type KeyFit int
+
+const (
+	// KeyRecorded: the bucket records this key's id.
+	KeyRecorded KeyFit = iota + 1
+	// KeyOpens: the bucket records no id, and the key opens its newest objects.
+	KeyOpens
+	// KeyUnused: the bucket holds no sealed object yet.
+	KeyUnused
+)
+
+// A bucket without a recorded id is judged by its newest sealed objects: the
+// key must open the first segment of one of them. keyRefusals objects that
+// refuse the key decide a mismatch; at most keyTries are read, so a run of
+// damaged objects cannot stall the check.
+const (
+	keyRefusals = 3
+	keyTries    = 10
+)
+
+// BucketKeyID reads the key id the bucket records, "" when it records none.
+func BucketKeyID(ctx context.Context, b Bucket) (string, error) {
+	rc, err := b.Get(ctx, keyMark)
+	if errors.Is(err, ErrNotFound) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("read %s: %w", keyMark, err)
+	}
+	defer rc.Close()
+	raw, err := io.ReadAll(io.LimitReader(rc, 256))
+	if err != nil {
+		return "", fmt.Errorf("read %s: %w", keyMark, err)
+	}
+	id := strings.TrimSpace(string(raw))
+	if !keyIDPattern.MatchString(id) {
+		return "", fmt.Errorf("offsite: %s in the bucket is not a key id Felis wrote", keyMark)
+	}
+	return id, nil
+}
+
+// errOpened stops Decrypt once the first segment has authenticated.
+var errOpened = errors.New("offsite: first segment opened")
+
+type firstSegment struct{}
+
+func (firstSegment) Write([]byte) (int, error) { return 0, errOpened }
+
+// CheckKey tells whether key is the one the bucket's objects are sealed with.
+// It writes nothing; a mismatch is ErrKeyMismatch.
+func CheckKey(ctx context.Context, b Bucket, key []byte) (KeyFit, error) {
+	mine := KeyID(key)
+	id, err := BucketKeyID(ctx, b)
+	if err != nil {
+		return 0, err
+	}
+	if id != "" {
+		if id != mine {
+			return 0, fmt.Errorf("%w: the bucket records key id %s, this key is %s; %s", ErrKeyMismatch, id, mine, keyMismatchFix)
+		}
+		return KeyRecorded, nil
+	}
+	var sealed []Object
+	for _, dir := range []string{dbDir, worldsDir, registryDir, uploadsDir} {
+		objs, err := b.List(ctx, dir)
+		if err != nil {
+			return 0, fmt.Errorf("list %s: %w", dir, err)
+		}
+		for _, o := range objs {
+			if strings.HasSuffix(o.Key, objExt) {
+				sealed = append(sealed, o)
+			}
+		}
+	}
+	if len(sealed) == 0 {
+		return KeyUnused, nil
+	}
+	sort.Slice(sealed, func(i, j int) bool { return sealed[i].Modified.After(sealed[j].Modified) })
+	var refused []string
+	var unjudged error
+	for i, o := range sealed {
+		if i == keyTries || len(refused) == keyRefusals {
+			break
+		}
+		rc, err := b.Get(ctx, o.Key)
+		if errors.Is(err, ErrNotFound) {
+			continue // pruned since the listing
+		}
+		if err != nil {
+			return 0, fmt.Errorf("%s: %w", o.Key, err)
+		}
+		err = Decrypt(firstSegment{}, rc, key)
+		rc.Close()
+		switch {
+		case err == nil || errors.Is(err, errOpened):
+			return KeyOpens, nil
+		case errors.Is(err, ErrAuth):
+			refused = append(refused, o.Key)
+		case unjudged == nil:
+			unjudged = fmt.Errorf("%s: %w", o.Key, err)
+		}
+	}
+	if len(refused) > 0 {
+		return 0, fmt.Errorf("%w: this key (key id %s) opens none of %s; %s", ErrKeyMismatch, mine, strings.Join(refused, ", "), keyMismatchFix)
+	}
+	if unjudged != nil {
+		return 0, fmt.Errorf("offsite: cannot tell which key sealed the bucket's objects: %w", unjudged)
+	}
+	return KeyUnused, nil
+}
+
+// ClaimKey is CheckKey, then records key's id in a bucket that has none, so
+// that every later check reads the id.
+func ClaimKey(ctx context.Context, b Bucket, key []byte) error {
+	fit, err := CheckKey(ctx, b, key)
+	if err != nil || fit == KeyRecorded {
+		return err
+	}
+	id := KeyID(key) + "\n"
+	if err := b.Put(ctx, keyMark, strings.NewReader(id), int64(len(id))); err != nil {
+		return fmt.Errorf("record the key id in %s: %w", keyMark, err)
+	}
+	return nil
+}

internal/offsite/keymark_test.go

@@ -0,0 +1,248 @@
+package offsite
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+// putAt stores data under key as modified at t0 plus min minutes.
+func putAt(b *memBucket, key string, data []byte, min int) {
+	if b.modified == nil {
+		b.modified = map[string]time.Time{}
+	}
+	b.objs[key] = data
+	b.modified[key] = now.Add(time.Duration(min) * time.Minute)
+}
+
+func TestCheckKey(t *testing.T) {
+	key, other := testKey(t), testKey(t)
+	big := make([]byte, 3*segmentSize)
+	tailDamaged := seal(t, big, key)
+	tailDamaged[len(tailDamaged)-1] ^= 1
+	for _, tc := range []struct {
+		what   string
+		bucket func(b *memBucket)
+		fit    KeyFit
+		err    string   // "" for none
+		named  []string // what a refusal names
+		skip   []string // what it must leave out
+	}{
+		{what: "an empty bucket", bucket: func(b *memBucket) {}, fit: KeyUnused},
+		{what: "nothing Felis sealed", bucket: func(b *memBucket) { putAt(b, "db/notes.txt", []byte("hi"), 0) }, fit: KeyUnused},
+		{
+			what: "the recorded id", fit: KeyRecorded,
+			bucket: func(b *memBucket) {
+				b.objs[keyMark] = []byte(KeyID(key) + "\n")
+				putAt(b, "worlds/a.tar.gz.fenc", seal(t, []byte("a"), other), 0) // the id is the judge
+			},
+		},
+		{
+			what:   "another recorded id",
+			bucket: func(b *memBucket) { b.objs[keyMark] = []byte(KeyID(other) + "\n") },
+			err:    "sealed with another key", named: []string{KeyID(other), KeyID(key), "offsite.env"},
+		},
+		{
+			what:   "a marker Felis did not write",
+			bucket: func(b *memBucket) { b.objs[keyMark] = []byte("hello") },
+			err:    "not a key id Felis wrote",
+		},
+		{
+			what: "an unmarked bucket the key opens", fit: KeyOpens,
+			bucket: func(b *memBucket) {
+				putAt(b, "db/felis-db-20260101T030000Z-daily.tar.fenc", seal(t, []byte("old"), other), 0)
+				putAt(b, "registry/blobs/aa.fenc", seal(t, []byte("new"), key), 10)
+			},
+		},
+		{
+			what: "only the first segment is read", fit: KeyOpens,
+			bucket: func(b *memBucket) { putAt(b, "uploads/blobs/bb.fenc", tailDamaged, 0) },
+		},
+		{
+			what: "an unmarked bucket sealed with another key",
+			bucket: func(b *memBucket) {
+				for i, k := range []string{"worlds/1.fenc", "worlds/2.fenc", "db/3.fenc", "uploads/index/4.json.fenc"} {
+					putAt(b, k, seal(t, []byte(k), other), i)
+				}
+			},
+			err: "sealed with another key", named: []string{"uploads/index/4.json.fenc", "db/3.fenc", "worlds/2.fenc", KeyID(key)}, skip: []string{"worlds/1.fenc"},
+		},
+		{
+			what: "two refusals, then an object the key opens", fit: KeyOpens,
+			bucket: func(b *memBucket) {
+				putAt(b, "worlds/1.fenc", seal(t, []byte("1"), key), 0)
+				putAt(b, "worlds/2.fenc", seal(t, []byte("2"), other), 1)
+				putAt(b, "worlds/3.fenc", seal(t, []byte("3"), other), 2)
+			},
+		},
+		{
+			what: "damaged objects are passed over", fit: KeyOpens,
+			bucket: func(b *memBucket) {
+				putAt(b, "worlds/1.fenc", seal(t, []byte("1"), key), 0)
+				putAt(b, "worlds/2.fenc", []byte("partial"), 1)
+			},
+		},
+		{
+			what: "a refusal among damaged objects",
+			bucket: func(b *memBucket) {
+				putAt(b, "worlds/1.fenc", seal(t, []byte("1"), other), 0)
+				putAt(b, "worlds/2.fenc", []byte("partial"), 1)
+			},
+			err: "sealed with another key", named: []string{"worlds/1.fenc"},
+		},
+		{
+			what: "nothing it can judge",
+			bucket: func(b *memBucket) {
+				putAt(b, "worlds/1.fenc", seal(t, []byte("1"), key), 0)
+				for i := range keyTries {
+					putAt(b, fmt.Sprintf("worlds/junk-%d.fenc", i), []byte("partial"), 1+i)
+				}
+			},
+			err: "cannot tell which key", named: []string{"worlds/junk-"},
+		},
+	} {
+		t.Run(tc.what, func(t *testing.T) {
+			b := newMemBucket()
+			tc.bucket(b)
+			fit, err := CheckKey(context.Background(), b, key)
+			if b.puts != 0 {
+				t.Errorf("CheckKey wrote %d objects", b.puts)
+			}
+			if tc.err == "" {
+				if err != nil || fit != tc.fit {
+					t.Fatalf("CheckKey = %v, %v; want %v", fit, err, tc.fit)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tc.err) {
+				t.Fatalf("CheckKey = %v, %v; want an error saying %q", fit, err, tc.err)
+			}
+			if mismatch := tc.err == "sealed with another key"; errors.Is(err, ErrKeyMismatch) != mismatch {
+				t.Errorf("errors.Is(err, ErrKeyMismatch) = %v, want %v: %v", !mismatch, mismatch, err)
+			}
+			for _, s := range tc.named {
+				if !strings.Contains(err.Error(), s) {
+					t.Errorf("error lacks %q: %v", s, err)
+				}
+			}
+			for _, s := range tc.skip {
+				if strings.Contains(err.Error(), s) {
+					t.Errorf("error names %q: %v", s, err)
+				}
+			}
+		})
+	}
+
+	// A bucket that does not answer is not a mismatch.
+	b := newMemBucket()
+	putAt(b, "worlds/1.fenc", seal(t, []byte("1"), other), 0)
+	down := errors.New("connection reset")
+	b.getErr = map[string]error{"worlds/1.fenc": down}
+	if _, err := CheckKey(context.Background(), b, key); !errors.Is(err, down) || errors.Is(err, ErrKeyMismatch) {
+		t.Errorf("unreachable object: err = %v, want the read error", err)
+	}
+	b.getErr = map[string]error{keyMark: down}
+	if _, err := CheckKey(context.Background(), b, key); !errors.Is(err, down) || errors.Is(err, ErrKeyMismatch) {
+		t.Errorf("unreachable marker: err = %v, want the read error", err)
+	}
+
+	// An object pruned between the listing and the read is passed over.
+	b = newMemBucket()
+	putAt(b, "worlds/1.fenc", seal(t, []byte("1"), key), 0)
+	putAt(b, "db/2.fenc", seal(t, []byte("2"), key), 1)
+	b.getErr = map[string]error{"db/2.fenc": fmt.Errorf("%w: db/2.fenc", ErrNotFound)}
+	if fit, err := CheckKey(context.Background(), b, key); fit != KeyOpens || err != nil {
+		t.Errorf("an object gone since the listing: CheckKey = %v, %v; want KeyOpens", fit, err)
+	}
+}
+
+func TestClaimKey(t *testing.T) {
+	key, other := testKey(t), testKey(t)
+	marker := func(b *memBucket) string { return string(b.objs[keyMark]) }
+
+	b := newMemBucket()
+	if err := ClaimKey(context.Background(), b, key); err != nil {
+		t.Fatal(err)
+	}
+	if marker(b) != KeyID(key)+"\n" {
+		t.Fatalf("empty bucket: marker = %q, want %s", marker(b), KeyID(key))
+	}
+	if err := ClaimKey(context.Background(), b, key); err != nil || b.puts != 1 {
+		t.Errorf("second claim: err %v, puts %d; want the marker written once", err, b.puts)
+	}
+	if fit, err := CheckKey(context.Background(), b, key); fit != KeyRecorded || err != nil {
+		t.Errorf("after the claim: CheckKey = %v, %v", fit, err)
+	}
+	if err := ClaimKey(context.Background(), b, other); !errors.Is(err, ErrKeyMismatch) || marker(b) != KeyID(key)+"\n" {
+		t.Errorf("another key: err %v, marker %q; want a refusal that leaves the marker", err, marker(b))
+	}
+
+	b = newMemBucket()
+	putAt(b, "worlds/1.fenc", seal(t, []byte("1"), key), 0)
+	if err := ClaimKey(context.Background(), b, key); err != nil || marker(b) != KeyID(key)+"\n" {
+		t.Errorf("unmarked bucket the key opens: err %v, marker %q", err, marker(b))
+	}
+
+	b = newMemBucket()
+	putAt(b, "worlds/1.fenc", seal(t, []byte("1"), other), 0)
+	if err := ClaimKey(context.Background(), b, key); !errors.Is(err, ErrKeyMismatch) || b.puts != 0 {
+		t.Errorf("unmarked bucket under another key: err %v, puts %d; want a refusal that writes nothing", err, b.puts)
+	}
+}
+
+// TestSyncRefusesAnotherKeysBucket: a host whose key does not match the
+// bucket's objects neither copies into it nor prunes what only the right key
+// opens.
+func TestSyncRefusesAnotherKeysBucket(t *testing.T) {
+	for _, marked := range []bool{true, false} {
+		t.Run(fmt.Sprintf("marked=%v", marked), func(t *testing.T) {
+			cat := &fakeCatalog{rows: []*row{
+				{WorldBackup: WorldBackup{ID: "b1", Server: "alpha", Ref: "/a/alpha-1.tar.gz"}, status: "present"},
+			}}
+			s, b := newSyncer(t, cat)
+			other := testKey(t)
+			delete(b.objs, keyMark)
+			if marked {
+				b.objs[keyMark] = []byte(KeyID(other) + "\n")
+			}
+			writeFile(t, s.ArchiveDir, "alpha-1.tar.gz", 100)
+			writeFile(t, s.DBDir, "felis-db-20260924T030000Z-daily.tar", 50)
+			for i, name := range []string{"felis-db-20260901T030000Z-daily.tar", "felis-db-20260902T030000Z-daily.tar", "felis-db-20260903T030000Z-daily.tar"} {
+				putAt(b, DBKey(name), seal(t, []byte(name), other), i)
+			}
+			before := len(b.objs)
+
+			_, err := s.Run(context.Background())
+			if !errors.Is(err, ErrKeyMismatch) {
+				t.Fatalf("Run error = %v, want ErrKeyMismatch", err)
+			}
+			if len(b.started) != 0 || len(b.removed) != 0 || len(b.objs) != before {
+				t.Errorf("the refused run began puts %v and removed %v", b.started, b.removed)
+			}
+			if !cat.rows[0].offsite.IsZero() {
+				t.Error("the refused run recorded alpha as copied")
+			}
+		})
+	}
+}
+
+// TestSyncRecordsTheKey: the first run into a bucket records its key before
+// the first upload.
+func TestSyncRecordsTheKey(t *testing.T) {
+	s, b := newSyncer(t, &fakeCatalog{})
+	delete(b.objs, keyMark)
+	writeFile(t, s.DBDir, "felis-db-20260924T030000Z-daily.tar", 50)
+	if _, err := s.Run(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(b.objs[keyMark], []byte(KeyID(s.Key)+"\n")) {
+		t.Fatalf("marker = %q, want %s", b.objs[keyMark], KeyID(s.Key))
+	}
+	if len(b.started) != 2 || b.started[0] != keyMark {
+		t.Errorf("puts began in the order %v, want the marker, then the bundle", b.started)
+	}
+}

internal/offsite/offsite_test.go

@@ -122,6 +122,10 @@ type memBucket struct {
 	stall map[string]bool
 	// started lists every Put in the order it began.
 	started []string
+	// modified is what List reports as each key's modification time.
+	modified map[string]time.Time
+	// getErr fails Get for a key the way an unreachable bucket would.
+	getErr map[string]error
 }
 
 func newMemBucket() *memBucket {
@@ -157,6 +161,9 @@ func (b *memBucket) Put(ctx context.Context, key string, r io.Reader, size int64
 func (b *memBucket) Get(_ context.Context, key string) (io.ReadCloser, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
+	if err := b.getErr[key]; err != nil {
+		return nil, err
+	}
 	data, ok := b.objs[key]
 	if !ok {
 		return nil, fmt.Errorf("%w: %s", ErrNotFound, key)
@@ -170,7 +177,7 @@ func (b *memBucket) List(_ context.Context, prefix string) ([]Object, error) {
 	var out []Object
 	for k, v := range b.objs {
 		if strings.HasPrefix(k, prefix) {
-			out = append(out, Object{Key: k, Size: int64(len(v))})
+			out = append(out, Object{Key: k, Size: int64(len(v)), Modified: b.modified[k]})
 		}
 	}
 	sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
@@ -245,11 +252,15 @@ func writeFile(t *testing.T, dir, name string, size int) []byte {
 
 var now = time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC)
 
+// newSyncer syncs into a bucket that already records its key, as every bucket
+// does after its first run.
 func newSyncer(t *testing.T, cat *fakeCatalog) (*Syncer, *memBucket) {
 	t.Helper()
 	b := newMemBucket()
+	key := testKey(t)
+	b.objs[keyMark] = []byte(KeyID(key) + "\n")
 	return &Syncer{
-		Bucket: b, Catalog: cat, Key: testKey(t),
+		Bucket: b, Catalog: cat, Key: key,
 		ArchiveDir: t.TempDir(), DBDir: t.TempDir(), DBKeep: 2,
 		Now: func() time.Time { return now },
 	}, b
@@ -393,7 +404,7 @@ func TestSyncDBKeepsNewest(t *testing.T) {
 		keys = append(keys, k)
 	}
 	sort.Strings(keys)
-	want := "db/felis-db-20260923T030000Z-manual.tar.fenc db/felis-db-20260924T030000Z-daily.tar.fenc db/notes.txt"
+	want := "db/felis-db-20260923T030000Z-manual.tar.fenc db/felis-db-20260924T030000Z-daily.tar.fenc db/notes.txt " + keyMark
 	if strings.Join(keys, " ") != want {
 		t.Fatalf("bucket = %v, want %s", keys, want)
 	}

internal/offsite/status.go

@@ -28,6 +28,10 @@ type Status struct {
 	Prefix      string    `json:"prefix,omitempty"`
 	KeyID       string    `json:"key_id"`
 	Result      Result    `json:"result"`
+	// KeyMismatch is a run refused because the bucket's objects are sealed
+	// with another key (ErrKeyMismatch): no later run copies anything until
+	// the key is fixed, so the watchdog reports it at once.
+	KeyMismatch bool `json:"key_mismatch,omitempty"`
 }
 
 // ReadStatus reads the status file. A missing file is (nil, nil): no sync has

internal/offsite/sync.go

@@ -200,6 +200,11 @@ func (s *Syncer) Run(ctx context.Context) (Result, error) {
 		s.logf("%s", msg)
 	}
 
+	// Before anything is written or pruned: objects sealed with another key
+	// are copies only that key opens, and DBKeep would prune them.
+	if err := ClaimKey(ctx, s.Bucket, s.Key); err != nil {
+		return res, err
+	}
 	remoteWorlds, err := s.listSizes(ctx, worldsDir)
 	if err != nil {
 		return res, fmt.Errorf("list %s in the bucket: %w", worldsDir, err)

internal/watchdog/probes.go

@@ -360,6 +360,14 @@ func OffsiteFinding(statusFile string, now time.Time) *Finding {
 			Hint:      hint,
 		}
 	}
+	if st != nil && st.KeyMismatch {
+		return &Finding{
+			Key: "offsite", Severity: Warning, For: backupFor,
+			Summary:   "异地备份已停止：桶里的对象是用另一把密钥加密的，本机不往桶里写入也不清理任何对象（" + st.LastError + "）",
+			SummaryEN: "the off-site copy has stopped: the bucket's objects are sealed with another key, and this host writes and prunes nothing there (" + st.LastError + ")",
+			Hint:      "`sudo felis offsite check-key`; set FELIS_OFFSITE_KEY in /etc/felis/offsite.env to the key the bucket was written with (docs/troubleshooting.md §16)",
+		}
+	}
 	if st != nil && !st.LastSuccess.IsZero() && now.Sub(st.LastSuccess) <= offsite.StaleAfter {
 		return nil
 	}

internal/watchdog/probes_test.go

@@ -183,6 +183,11 @@ func TestOffsiteFinding(t *testing.T) {
 	if f := OffsiteFinding(path, t0); f != nil {
 		t.Fatalf("a success within %s reported: %+v", offsite.StaleAfter, f)
 	}
+	// A key mismatch stops every later run too: reported without waiting out StaleAfter.
+	write(offsite.Status{LastAttempt: t0.Add(-time.Hour), LastSuccess: t0.Add(-2 * time.Hour), LastError: "sealed with another key", KeyMismatch: true})
+	if f := OffsiteFinding(path, t0); f == nil || !strings.Contains(f.SummaryEN, "has stopped") || !strings.Contains(f.SummaryEN, "(sealed with another key)") || !strings.Contains(f.Hint, "check-key") {
+		t.Fatalf("key mismatch: %+v", f)
+	}
 }
 
 func TestMemoryFinding(t *testing.T) {