From 69ce9220cea2971d584c9bd1d5440c98af874da2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Sat, 7 Nov 2026 20:44:17 -0500 Subject: [PATCH] feat(natsstore): mirror the buckets into a disposable spool The rest of the process reads config and credentials from files, so the buckets need a local mirror, but the mirror must never be trusted over the buckets. Rebuilding it on every start and sending local writes back with revision checks keeps replicas from diverging through stale disk state. Signed-off-by: Yordis Prieto --- internal/store/natsstore/authstore.go | 39 ++++ internal/store/natsstore/store.go | 247 ++++++++++++++++++++ internal/store/natsstore/store_test.go | 300 +++++++++++++++++++++++++ 3 files changed, 586 insertions(+) create mode 100644 internal/store/natsstore/store.go create mode 100644 internal/store/natsstore/store_test.go diff --git a/internal/store/natsstore/authstore.go b/internal/store/natsstore/authstore.go index b22a61a084a..28b1fe7e440 100644 --- a/internal/store/natsstore/authstore.go +++ b/internal/store/natsstore/authstore.go @@ -397,3 +397,42 @@ func labelFor(metadata map[string]any) string { } return "" } + +// PutFile stores the bytes of a credential file that something else already +// wrote into the spool. A file equal to the stored value is a no-op, which +// absorbs the watcher echoing this process's own writes. Otherwise the write is +// checked against the last revision seen, and a stale or removed credential is +// reported as a conflict, so a stale local file never replaces newer +// credentials. +func (s *AuthStore) PutFile(ctx context.Context, id string, data []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + + expected := s.revisions[id] + current, err := s.kv.Get(ctx, id) + switch { + case errors.Is(err, ErrNotFound): + if expected != 0 { + return &ConflictError{Bucket: s.kv.bucket, ID: id} + } + rev, err := s.kv.Create(ctx, id, data) + if err != nil { + return err + } + s.revisions[id] = rev + return nil + case err != nil: + return err + case sameJSON(current.Value, data): + s.revisions[id] = current.Revision + return nil + case current.Revision != expected: + return &ConflictError{Bucket: s.kv.bucket, ID: id} + } + rev, err := s.kv.Update(ctx, id, data, expected) + if err != nil { + return err + } + s.revisions[id] = rev + return nil +} diff --git a/internal/store/natsstore/store.go b/internal/store/natsstore/store.go new file mode 100644 index 00000000000..24a67fce57f --- /dev/null +++ b/internal/store/natsstore/store.go @@ -0,0 +1,247 @@ +package natsstore + +import ( + "bytes" + "context" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/nats-io/nats.go" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/config" + "github.com/router-for-me/CLIProxyAPI/v8/internal/misc" + cliproxyauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth" + log "github.com/sirupsen/logrus" +) + +// Store is the NATS-backed token store. The buckets are the source of truth and +// the spool directory is a disposable mirror rebuilt on every start, so the +// config and credential files the rest of the process reads keep working. +type Store struct { + nc *nats.Conn + auth *AuthStore + config *ConfigStore + spoolRoot string + configPath string + + mu sync.Mutex + // configData is the config last written to or read from the spool, so the + // watcher echoing that write back is not stored again. + configData []byte +} + +// NewStore wires a bootstrapped connection to a spool rooted at spoolDir. The +// store closes nc when it is closed. +func NewStore(nc *nats.Conn, boot *Bootstrapped, settings StoreSettings, spoolDir string) (*Store, error) { + root, err := filepath.Abs(spoolDir) + if err != nil { + return nil, fmt.Errorf("nats store: resolve spool directory: %w", err) + } + configDir := filepath.Join(root, "config") + if err := os.MkdirAll(configDir, 0o700); err != nil { + return nil, fmt.Errorf("nats store: create config directory: %w", err) + } + auth, err := NewAuthStore(NewKV(settings.BucketName(BucketAuth), boot.Bucket(settings, BucketAuth)), filepath.Join(root, "auths")) + if err != nil { + return nil, err + } + cfg := NewConfigStore(NewKV(settings.BucketName(BucketConfig), boot.Bucket(settings, BucketConfig))) + return &Store{ + nc: nc, + auth: auth, + config: cfg, + spoolRoot: root, + configPath: filepath.Join(configDir, "config.yaml"), + }, nil +} + +// Close closes the NATS connection. +func (s *Store) Close() error { + if s != nil && s.nc != nil { + s.nc.Close() + } + return nil +} + +// ConfigPath returns the config file inside the spool. +func (s *Store) ConfigPath() string { return s.configPath } + +// AuthDir returns the directory credential files are mirrored into. +func (s *Store) AuthDir() string { return s.auth.AuthDir() } + +// WorkDir returns the spool root. +func (s *Store) WorkDir() string { return s.spoolRoot } + +// SetBaseDir satisfies the optional authenticator interface; the store owns its workspace. +func (s *Store) SetBaseDir(string) {} + +// Save stores a credential. +func (s *Store) Save(ctx context.Context, auth *cliproxyauth.Auth) (string, error) { + return s.auth.Save(ctx, auth) +} + +// List returns every stored credential. +func (s *Store) List(ctx context.Context) ([]*cliproxyauth.Auth, error) { return s.auth.List(ctx) } + +// Delete removes a credential. +func (s *Store) Delete(ctx context.Context, id string) error { return s.auth.Delete(ctx, id) } + +// Bootstrap rebuilds the spool from the buckets. A config missing from the +// bucket is seeded from the local file or the example template, and replicas +// starting together converge on whichever one created it first. Running it +// again changes nothing. +func (s *Store) Bootstrap(ctx context.Context, exampleConfigPath string) error { + if err := s.syncConfig(ctx, exampleConfigPath); err != nil { + return err + } + return s.syncAuth(ctx) +} + +func (s *Store) syncConfig(ctx context.Context, exampleConfigPath string) error { + data, _, err := s.config.Read(ctx) + if errors.Is(err, ErrNotFound) { + if err := s.seedConfig(ctx, exampleConfigPath); err == nil { + return nil + } else if !errors.Is(err, ErrConflict) { + return err + } + // Another replica seeded it first; adopt theirs. + data, _, err = s.config.Read(ctx) + } + if err != nil { + return fmt.Errorf("nats store: load config: %w", err) + } + return s.writeConfigFile(data) +} + +// seedConfig stores the local config, or the example template, in the bucket. +func (s *Store) seedConfig(ctx context.Context, exampleConfigPath string) error { + if _, err := os.Stat(s.configPath); errors.Is(err, fs.ErrNotExist) { + if exampleConfigPath != "" { + if err := misc.CopyConfigTemplate(exampleConfigPath, s.configPath); err != nil { + return fmt.Errorf("nats store: copy example config: %w", err) + } + } else if err := os.WriteFile(s.configPath, nil, 0o600); err != nil { + return fmt.Errorf("nats store: create empty config: %w", err) + } + } + // Seed with the file the startup load leaves behind, so a plaintext + // management secret never reaches the bucket. + if err := config.PrepareConfigFileForUpload(s.configPath); err != nil { + return fmt.Errorf("nats store: prepare local config for seeding: %w", err) + } + data, err := os.ReadFile(s.configPath) + if err != nil { + return fmt.Errorf("nats store: read local config: %w", err) + } + if _, err := s.config.Write(ctx, data); err != nil { + return err + } + s.rememberConfig(data) + return nil +} + +func (s *Store) writeConfigFile(data []byte) error { + if err := writeFileAtomic(s.configPath, data); err != nil { + return fmt.Errorf("nats store: write config to spool: %w", err) + } + s.rememberConfig(data) + return nil +} + +func (s *Store) rememberConfig(data []byte) { + s.mu.Lock() + defer s.mu.Unlock() + s.configData = []byte(normalizeLineEndings(string(data))) +} + +func (s *Store) syncAuth(ctx context.Context) error { + entries, err := s.auth.Snapshot(ctx) + if err != nil { + return fmt.Errorf("nats store: load auth: %w", err) + } + if err := os.RemoveAll(s.auth.AuthDir()); err != nil { + return fmt.Errorf("nats store: reset auth directory: %w", err) + } + if err := os.MkdirAll(s.auth.AuthDir(), 0o700); err != nil { + return fmt.Errorf("nats store: recreate auth directory: %w", err) + } + for _, entry := range entries { + path, err := s.auth.absoluteAuthPath(entry.ID) + if err != nil { + log.WithError(err).Warnf("nats store: skipping auth %s outside spool", entry.ID) + continue + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return fmt.Errorf("nats store: create auth subdirectory: %w", err) + } + if err := os.WriteFile(path, entry.Value, 0o600); err != nil { + return fmt.Errorf("nats store: write auth file: %w", err) + } + } + return nil +} + +// PersistConfig stores the spool config file in the bucket. A missing or empty +// file is ignored, since it is more likely a write in progress than an intent +// to wipe the config every replica shares. +func (s *Store) PersistConfig(ctx context.Context) error { + data, err := os.ReadFile(s.configPath) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + log.Warn("nats store: config file is missing, not persisting") + return nil + } + return fmt.Errorf("nats store: read config file: %w", err) + } + if len(bytes.TrimSpace(data)) == 0 { + log.Warn("nats store: config file is empty, not persisting") + return nil + } + s.mu.Lock() + unchanged := bytes.Equal(s.configData, []byte(normalizeLineEndings(string(data)))) + s.mu.Unlock() + if unchanged { + return nil + } + if _, err := s.config.Write(ctx, data); err != nil { + return fmt.Errorf("nats store: persist config: %w", err) + } + s.rememberConfig(data) + return nil +} + +// PersistAuthFiles stores the given spool credential files in the bucket, and +// removes the credentials whose files are gone or empty. +func (s *Store) PersistAuthFiles(ctx context.Context, _ string, paths ...string) error { + for _, raw := range paths { + path := strings.TrimSpace(raw) + if path == "" { + continue + } + id, err := s.auth.relativeAuthID(path) + if err != nil { + log.WithError(err).Warnf("nats store: ignoring auth path %s", path) + continue + } + data, err := os.ReadFile(filepath.Join(s.auth.AuthDir(), filepath.FromSlash(id))) + if err != nil && !errors.Is(err, fs.ErrNotExist) { + return fmt.Errorf("nats store: read auth file: %w", err) + } + if err != nil || len(data) == 0 { + if err := s.auth.Delete(ctx, id); err != nil { + return err + } + continue + } + if err := s.auth.PutFile(ctx, id, data); err != nil { + return err + } + } + return nil +} diff --git a/internal/store/natsstore/store_test.go b/internal/store/natsstore/store_test.go new file mode 100644 index 00000000000..303c4bdf155 --- /dev/null +++ b/internal/store/natsstore/store_test.go @@ -0,0 +1,300 @@ +package natsstore + +import ( + "errors" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/store/natsstore/natstest" +) + +// storeFixture is one NATS server that several stores, each with its own spool, share. +type storeFixture struct { + t *testing.T + url string + settings StoreSettings + example string +} + +func newStoreFixture(t *testing.T) *storeFixture { + t.Helper() + srv := natstest.Start(t) + example := filepath.Join(t.TempDir(), "config.example.yaml") + if err := os.WriteFile(example, []byte("port: 8317\nhost: \"\"\n"), 0o600); err != nil { + t.Fatal(err) + } + return &storeFixture{t: t, url: srv.URL, settings: autoSettings("spooltest"), example: example} +} + +// open connects a new replica with an empty spool. +func (f *storeFixture) open() *Store { + f.t.Helper() + nc, err := connect(f.t, map[string]string{EnvURL: f.url}, 5*time.Second) + if err != nil { + f.t.Fatal(err) + } + return f.wrap(nc) +} + +func (f *storeFixture) wrap(nc *nats.Conn) *Store { + f.t.Helper() + boot, err := Bootstrap(testContext(f.t), nc, f.settings) + if err != nil { + f.t.Fatal(err) + } + store, err := NewStore(nc, boot, f.settings, f.t.TempDir()) + if err != nil { + f.t.Fatal(err) + } + return store +} + +func readFile(t *testing.T, path string) string { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return string(data) +} + +func TestStoreBootstrapSeedsConfigOnce(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + first, second := f.open(), f.open() + + if err := first.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + if got := readFile(t, first.ConfigPath()); got != "port: 8317\nhost: \"\"\n" { + t.Fatalf("seeded config = %q", got) + } + // A different template must not replace what is already stored. + other := filepath.Join(t.TempDir(), "other.yaml") + if err := os.WriteFile(other, []byte("port: 1\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := second.Bootstrap(ctx, other); err != nil { + t.Fatal(err) + } + if got := readFile(t, second.ConfigPath()); got != "port: 8317\nhost: \"\"\n" { + t.Fatalf("second replica config = %q, want the stored one", got) + } + rev := first.config.Revision() + if err := first.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + if first.config.Revision() != rev { + t.Fatalf("repeat bootstrap rewrote the config: revision %d -> %d", rev, first.config.Revision()) + } +} + +func TestStoreBootstrapConcurrentSeedsConverge(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + stores := []*Store{f.open(), f.open(), f.open(), f.open()} + + var wg sync.WaitGroup + errs := make([]error, len(stores)) + for i, store := range stores { + wg.Add(1) + go func() { + defer wg.Done() + errs[i] = store.Bootstrap(ctx, f.example) + }() + } + wg.Wait() + for i, err := range errs { + if err != nil { + t.Fatalf("replica %d: %v", i, err) + } + } + for i, store := range stores { + if got := readFile(t, store.ConfigPath()); got != "port: 8317\nhost: \"\"\n" { + t.Fatalf("replica %d config = %q", i, got) + } + } +} + +func TestStoreBootstrapRebuildsAuthSpool(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + writer, reader := f.open(), f.open() + if err := writer.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + if _, err := writer.Save(ctx, testAuth("team/a.json", map[string]any{"type": "test", "access_token": "tok-a"})); err != nil { + t.Fatal(err) + } + + stale := filepath.Join(reader.AuthDir(), "stale.json") + if err := os.WriteFile(stale, []byte(`{"type":"test"}`), 0o600); err != nil { + t.Fatal(err) + } + for range 2 { + if err := reader.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("stale spool file survived bootstrap: %v", err) + } + if got := readFile(t, filepath.Join(reader.AuthDir(), "team", "a.json")); got != readFile(t, filepath.Join(writer.AuthDir(), "team", "a.json")) { + t.Fatalf("mirrored file = %q", got) + } + info, err := os.Stat(filepath.Join(reader.AuthDir(), "team", "a.json")) + if err != nil || info.Mode().Perm() != 0o600 { + t.Fatalf("mirrored file mode = %v, %v", info, err) + } + } + auths, err := reader.List(ctx) + if err != nil || len(auths) != 1 || auths[0].ID != "team/a.json" { + t.Fatalf("List = %v, %v", auths, err) + } +} + +func TestStorePersistAuthFiles(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + store, other := f.open(), f.open() + if err := store.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + + path := filepath.Join(store.AuthDir(), "uploaded.json") + if err := os.WriteFile(path, []byte(`{"type":"test","access_token":"tok-1"}`), 0o600); err != nil { + t.Fatal(err) + } + if err := store.PersistAuthFiles(ctx, "", path); err != nil { + t.Fatal(err) + } + entry, err := store.auth.kv.Get(ctx, "uploaded.json") + if err != nil || string(entry.Value) != `{"type":"test","access_token":"tok-1"}` { + t.Fatalf("stored = %+v, %v", entry, err) + } + // The watcher echoing the same file must not write a new revision. + if err := store.PersistAuthFiles(ctx, "", path); err != nil { + t.Fatal(err) + } + if again, _ := store.auth.kv.Get(ctx, "uploaded.json"); again.Revision != entry.Revision { + t.Fatalf("echo wrote revision %d over %d", again.Revision, entry.Revision) + } + + if err := os.WriteFile(path, []byte(`{"type":"test","access_token":"tok-2"}`), 0o600); err != nil { + t.Fatal(err) + } + if err := store.PersistAuthFiles(ctx, "", path); err != nil { + t.Fatal(err) + } + if changed, _ := store.auth.kv.Get(ctx, "uploaded.json"); string(changed.Value) != `{"type":"test","access_token":"tok-2"}` { + t.Fatalf("edit not stored: %s", changed.Value) + } + + // A replica holding an older revision must not overwrite the newer value. + if err := other.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + if err := store.PersistAuthFiles(ctx, "", path); err != nil { + t.Fatal(err) + } + stale := filepath.Join(other.AuthDir(), "uploaded.json") + if err := os.WriteFile(stale, []byte(`{"type":"test","access_token":"tok-stale"}`), 0o600); err != nil { + t.Fatal(err) + } + if err := store.auth.Delete(ctx, "uploaded.json"); err != nil { + t.Fatal(err) + } + if _, err := store.auth.Save(ctx, testAuth("uploaded.json", map[string]any{"type": "test", "access_token": "tok-3"})); err != nil { + t.Fatal(err) + } + if err := other.PersistAuthFiles(ctx, "", stale); !errors.Is(err, ErrConflict) { + t.Fatalf("stale persist = %v, want conflict", err) + } + if kept, _ := store.auth.kv.Get(ctx, "uploaded.json"); string(kept.Value) == `{"type":"test","access_token":"tok-stale"}` { + t.Fatal("stale file replaced newer credentials") + } + + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := store.PersistAuthFiles(ctx, "", path, filepath.Join(t.TempDir(), "outside.json")); err != nil { + t.Fatal(err) + } + if _, err := store.auth.kv.Get(ctx, "uploaded.json"); !errors.Is(err, ErrNotFound) { + t.Fatalf("removed file left its credential behind: %v", err) + } +} + +func TestStorePersistConfig(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + store, other := f.open(), f.open() + if err := store.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + if err := other.Bootstrap(ctx, f.example); err != nil { + t.Fatal(err) + } + rev := store.config.Revision() + + // The watcher echoing the bootstrap write stores nothing. + if err := store.PersistConfig(ctx); err != nil || store.config.Revision() != rev { + t.Fatalf("echo persist: %v, revision %d -> %d", err, rev, store.config.Revision()) + } + if err := os.WriteFile(store.ConfigPath(), []byte("port: 9000\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := store.PersistConfig(ctx); err != nil { + t.Fatal(err) + } + data, _, err := f.open().config.Read(ctx) + if err != nil || string(data) != "port: 9000\n" { + t.Fatalf("stored config = %q, %v", data, err) + } + + // A stale replica is told it lost. + if err := os.WriteFile(other.ConfigPath(), []byte("port: 1\n"), 0o600); err != nil { + t.Fatal(err) + } + if err := other.PersistConfig(ctx); !errors.Is(err, ErrConflict) { + t.Fatalf("stale persist = %v, want conflict", err) + } + + // A missing or empty file never wipes the shared config. + if err := os.WriteFile(store.ConfigPath(), nil, 0o600); err != nil { + t.Fatal(err) + } + if err := store.PersistConfig(ctx); err != nil { + t.Fatal(err) + } + if err := os.Remove(store.ConfigPath()); err != nil { + t.Fatal(err) + } + if err := store.PersistConfig(ctx); err != nil { + t.Fatal(err) + } + if data, _, err := f.open().config.Read(ctx); err != nil || string(data) != "port: 9000\n" { + t.Fatalf("config after empty and missing files = %q, %v", data, err) + } +} + +func TestStoreSeedRefusesPlaintextManagementSecret(t *testing.T) { + f := newStoreFixture(t) + ctx := testContext(t) + store := f.open() + template := filepath.Join(t.TempDir(), "template.yaml") + body := "remote-management:\n secret-key: &s plain-secret-value\n other: *s\n" + if err := os.WriteFile(template, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + if err := store.Bootstrap(ctx, template); err == nil { + t.Fatal("seeded a config that still holds a plaintext management secret") + } + if _, _, err := f.open().config.Read(ctx); !errors.Is(err, ErrNotFound) { + t.Fatalf("refused config reached the bucket: %v", err) + } +}