From 3e6207804a8533f7ed2f95b505ef76eb3d45dd64 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 25 Oct 2026 08:29:34 +0000 Subject: [PATCH] Prevent configuration-import replays from overwriting newer edits Keep one active import per workspace through persistence and runtime apply. Reject equal committed archive/passphrase requests for ten minutes, including after lost responses or failed apply, rather than repeat writes/reloads or report a cached success. Bound process-local admission without evicting live protection and release uncommitted failures for corrected retries. Preserve existing authentication, CSRF, tenant scope and complete legacy archives. Cover the real handler with response-loss, overlap, newer-edit, router and admission controls. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 11 + .../v6/internal/subsystems/api-contracts.md | 26 ++ .../internal/subsystems/storage-recovery.md | 17 + .../config_export_import_handlers.go | 16 + internal/api/configapi/config_handlers.go | 1 + .../configapi/config_import_admission_test.go | 109 ++++++ .../api/configapi/config_import_replay.go | 93 +++++ .../configapi/config_import_replay_test.go | 320 ++++++++++++++++++ internal/api/contract_test.go | 81 +++++ 9 files changed, 674 insertions(+) create mode 100644 internal/api/configapi/config_import_admission_test.go create mode 100644 internal/api/configapi/config_import_replay.go create mode 100644 internal/api/configapi/config_import_replay_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 14db19d65e..60b59e5fa5 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -15,6 +15,17 @@ ## Purpose +### Configuration-import retries preserve the remote-agent boundary + +The import API's process-local replay admission rejects duplicate committed +archives and overlapping workspace imports without another persistence or +runtime reload. Its existing remote-agent-endpoint disclosure stays unchanged: +configuration recovery does not retarget installed agents, replace enrollment +or prove installed recovery. Different tenant imports retain separate admission +and the normal authentication/CSRF/write-scope boundaries. The consuming-router +control is `TestContract_ConfigImportReplayRemainsAuthenticatedAndScoped`; +loopback loss and failed-apply controls live in `config_import_replay_test.go`. + Infrastructure Add shortcuts and update chips follow the frontend primitives visible-label naming contract. Their names preserve `Add` or the displayed update wording and append source or host context, while keeping the existing diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index cd1648af0d..17bcb05d24 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -21,6 +21,32 @@ ## Purpose +### Configuration-import transport replays are bounded, not fresh recovery + +The existing import POST keeps authentication, write scope, CSRF and tenant +resolution ahead of replay admission. A handler admits one import per workspace +through persistence and runtime apply. Equal archive/passphrase requests that +committed are rejected with HTTP 409 for ten minutes after completion, including +when runtime apply failed or the response was lost. Neither a replacement write +nor a reload is repeated, and a cached success cannot imply that a newer edit +was restored. Different archives can be imported after the active import ends; +uncommitted failures can retry after their cause is corrected. + +Admission stores only scoped full-request digests and completion times, never +archives, passphrases or response bodies. Empty/default scope aliases share +one admission; different tenants do not. The process-local registry holds at +most 128 active/recent attempts and fails closed with HTTP 503 rather than +evict an unexpired attempt. Active imports never expire; completed entries +expire without extending their window on replay. This is a bounded transport +safety measure, not durable cross-restart idempotency or installed recovery. + +`config_import_replay_test.go` exercises real persistence, newer credentials/ +settings/metadata, overlapping archives, runtime failure and loopback response +loss. `config_import_admission_test.go` covers concurrency, expiry and capacity. +`TestContract_ConfigImportReplayRemainsAuthenticatedAndScoped` exercises the +consuming router's tenant, authentication and CSRF boundaries. No guest/QGA, +backup execution, database/driver/schema or native recovery route is added. + ### Host-agent disk history distinguishes absence from zero History and full/infrastructure summary charts synthesize host-agent disk usage diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 4bbef90025..30b6a1d718 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -21,6 +21,23 @@ ## Purpose +### Configuration-import replays cannot repeat recent replacement + +Import admission spans the existing file transaction and runtime apply. A +committed equal archive/passphrase request is rejected for ten minutes after +completion, even after a lost response or failed runtime apply; it cannot +replace newer settings, credentials, metadata or recovery backups. Concurrent +workspace imports are rejected, not queued. An uncommitted failure can retry +after correction. The bounded, process-local registry fails closed rather +than evict unexpired protection; it is not durable cross-restart idempotency. + +`config_import_replay_test.go` exercises real handler/persistence replacement, +three loopback POSTs with response loss and newer legitimate edits, overlapping +archives and failed apply. The scoped router and admission concurrency/expiry/ +capacity controls retain authentication and tenant separation. Complete legacy +archives and successful transaction semantics remain unchanged. None of this +is guest thaw, backup execution or original native recovery evidence. + ### Agent startup is not storage evidence An agent's pending first report after a server restart is a monitoring-owned diff --git a/internal/api/configapi/config_export_import_handlers.go b/internal/api/configapi/config_export_import_handlers.go index cc3a9e7fb1..8379b48d38 100644 --- a/internal/api/configapi/config_export_import_handlers.go +++ b/internal/api/configapi/config_export_import_handlers.go @@ -3,6 +3,7 @@ package configapi import ( "encoding/json" "net/http" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/api/apicontext" "github.com/rcourtman/pulse-go-rewrite/internal/api/apihttp" @@ -119,12 +120,27 @@ func (h *ConfigHandlers) handleImportConfig(w http.ResponseWriter, r *http.Reque return } + // A browser or proxy can resend a POST even when application retries are + // disabled. Admit only one import per workspace through persistence AND + // runtime apply, and retain committed attempts before writing the response. + attempt, rejection := h.importRequests.begin(orgID, req, time.Now()) + if rejection != nil { + http.Error(w, rejection.message, rejection.status) + return + } + retainAttempt := true // Fail closed if an unexpected panic interrupts apply. + defer func() { h.importRequests.finish(attempt, retainAttempt, time.Now()) }() + // Capture actor identity before config swap (config replacement changes auth resolution) deps := h.runtimeDependencies() importUser := deps.AuthUsername(currentConfig, r) // Import configuration. if err := persistence.ImportConfig(req.Data, req.Passphrase); err != nil { + // No archive committed: a corrected storage failure or passphrase may + // be retried. Once persistence succeeds, even failed runtime apply + // must not replay the replacement over a newer legitimate edit. + retainAttempt = false deps.AuditEvent(apicontext.OrgID(r.Context()), "config_imported", importUser, deps.ClientIP(r), r.URL.Path, false, "Import failed") log.Error().Err(err).Msg("Failed to import configuration") diff --git a/internal/api/configapi/config_handlers.go b/internal/api/configapi/config_handlers.go index b5c08ca01c..c07d162741 100644 --- a/internal/api/configapi/config_handlers.go +++ b/internal/api/configapi/config_handlers.go @@ -231,6 +231,7 @@ type ConfigHandlers struct { recentAutoRegistered map[string]time.Time recentAutoRegMutex sync.Mutex autoRegisterMutex sync.Mutex + importRequests configImportRequests } // NewConfigHandlers creates a new ConfigHandlers instance diff --git a/internal/api/configapi/config_import_admission_test.go b/internal/api/configapi/config_import_admission_test.go new file mode 100644 index 0000000000..684f552d8b --- /dev/null +++ b/internal/api/configapi/config_import_admission_test.go @@ -0,0 +1,109 @@ +package configapi + +import ( + "fmt" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestConfigImportAdmissionScopeLifetimeAndCapacity(t *testing.T) { + now := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC) + req := ImportConfigRequest{Data: "synthetic-ciphertext", Passphrase: "synthetic-passphrase"} + t.Run("default aliases, other org and different passphrase", func(t *testing.T) { + var guard configImportRequests + first, rejection := guard.begin("", req, now) + if rejection != nil { + t.Fatal(rejection) + } + if _, rejection := guard.begin("default", req, now); rejection == nil || rejection.status != http.StatusConflict { + t.Fatal("default alias admitted overlapping import") + } + other, rejection := guard.begin("other", req, now) + if rejection != nil { + t.Fatal("another org inherited active import") + } + guard.finish(other, true, now) + guard.finish(first, true, now) + if _, rejection := guard.begin("default", req, now.Add(configImportReplayWindow-time.Nanosecond)); rejection == nil || rejection.status != http.StatusConflict { + t.Fatal("unexpired completed attempt was evicted") + } + wrong := req + wrong.Passphrase += "-different" + attempt, rejection := guard.begin("default", wrong, now) + if rejection != nil { + t.Fatal("another passphrase reused the completed admission") + } + guard.finish(attempt, false, now) + if _, rejection := guard.begin("default", req, now.Add(configImportReplayWindow)); rejection != nil { + t.Fatal("expired attempt prevented an intentional later import") + } + }) + t.Run("in-flight attempts never expire and completion starts window", func(t *testing.T) { + var guard configImportRequests + first, _ := guard.begin("default", req, now) + later := now.Add(2 * configImportReplayWindow) + if _, rejection := guard.begin("default", req, later); rejection == nil { + t.Fatal("long-running import expired") + } + guard.finish(first, true, later) + if _, rejection := guard.begin("default", req, later.Add(configImportReplayWindow-time.Nanosecond)); rejection == nil { + t.Fatal("window did not start at completion") + } + }) + t.Run("bounded capacity fails closed without evicting attempts", func(t *testing.T) { + var guard configImportRequests + for i := 0; i < configImportRequestLimit; i++ { + attempt, rejection := guard.begin(fmt.Sprintf("org-%d", i), req, now) + if rejection != nil { + t.Fatal(rejection) + } + guard.finish(attempt, true, now) + } + if _, rejection := guard.begin("overflow", req, now); rejection == nil || rejection.status != http.StatusServiceUnavailable { + t.Fatal("full registry did not fail closed") + } + if len(guard.attempts) != configImportRequestLimit { + t.Fatal("capacity changed") + } + if _, rejection := guard.begin("org-0", req, now); rejection == nil || rejection.status != http.StatusConflict { + t.Fatal("capacity overflow evicted original replay protection") + } + if _, rejection := guard.begin("overflow", req, now.Add(configImportReplayWindow)); rejection != nil { + t.Fatal("expired registry did not reclaim capacity") + } + }) + t.Run("failed transaction releases its admission", func(t *testing.T) { + var guard configImportRequests + attempt, _ := guard.begin("default", req, now) + guard.finish(attempt, false, now) + if _, rejection := guard.begin("default", req, now); rejection != nil { + t.Fatal("uncommitted request could not retry") + } + }) +} + +func TestConfigImportAdmissionConcurrentRequestsHaveOneOwner(t *testing.T) { + var guard configImportRequests + var admitted atomic.Int32 + var requests sync.WaitGroup + start := make(chan struct{}) + now := time.Now() + for i := 0; i < 64; i++ { + requests.Add(1) + go func() { + defer requests.Done() + <-start + if _, rejection := guard.begin("default", ImportConfigRequest{Data: "synthetic", Passphrase: "synthetic"}, now); rejection == nil { + admitted.Add(1) + } + }() + } + close(start) + requests.Wait() + if admitted.Load() != 1 || len(guard.attempts) != 1 { + t.Fatalf("concurrent owners = %d, want one", admitted.Load()) + } +} diff --git a/internal/api/configapi/config_import_replay.go b/internal/api/configapi/config_import_replay.go new file mode 100644 index 0000000000..3c46abab0c --- /dev/null +++ b/internal/api/configapi/config_import_replay.go @@ -0,0 +1,93 @@ +package configapi + +import ( + "crypto/sha256" + "encoding/binary" + "io" + "net/http" + "sync" + "time" +) + +const ( + configImportReplayWindow = 10 * time.Minute + configImportRequestLimit = 128 +) + +type configImportKey struct { + orgID string + fingerprint [sha256.Size]byte +} + +type configImportAttempt struct { + key configImportKey + completedAt time.Time // Zero while persistence or runtime apply is in flight. +} + +type configImportRejection struct { + status int + message string +} + +// configImportRequests protects the existing no-request-ID API without keeping +// archives, passphrases or responses. It is deliberately process-local and +// bounded: one in-flight import per org and at most 128 attempts for ten minutes +// after completion. Never evict an unexpired attempt to admit a new one. +type configImportRequests struct { + mu sync.Mutex + attempts map[configImportKey]*configImportAttempt +} + +func (g *configImportRequests) begin(orgID string, req ImportConfigRequest, now time.Time) (*configImportAttempt, *configImportRejection) { + if orgID == "" { + orgID = "default" + } + // Length-prefix the data, including malformed input. Include the + // passphrase so a different password cannot reuse an earlier admission. + // Only this full-request digest survives the request, never either input. + hash := sha256.New() + var length [8]byte + binary.BigEndian.PutUint64(length[:], uint64(len(req.Data))) + _, _ = hash.Write(length[:]) + _, _ = io.WriteString(hash, req.Data) + _, _ = io.WriteString(hash, req.Passphrase) + key := configImportKey{orgID: orgID} + hash.Sum(key.fingerprint[:0]) + + g.mu.Lock() + defer g.mu.Unlock() + for k, attempt := range g.attempts { + if !attempt.completedAt.IsZero() && now.Sub(attempt.completedAt) >= configImportReplayWindow { + delete(g.attempts, k) + } + } + if previous := g.attempts[key]; previous != nil && !previous.completedAt.IsZero() { + // Do not return a cached success: the configuration may have changed + // since that attempt. A replay must not claim a fresh restoration. + return nil, &configImportRejection{http.StatusConflict, "This backup was submitted recently and may already have changed the configuration. It has not been imported again; check the current configuration before retrying in 10 minutes."} + } + for k, attempt := range g.attempts { + if k.orgID == orgID && attempt.completedAt.IsZero() { + return nil, &configImportRejection{http.StatusConflict, "A configuration import is still running for this workspace. This request has not been applied; wait for it to finish and check the configuration before retrying."} + } + } + if len(g.attempts) >= configImportRequestLimit { + return nil, &configImportRejection{http.StatusServiceUnavailable, "Configuration imports are temporarily unavailable. This request has not been applied; try again in 10 minutes."} + } + if g.attempts == nil { + g.attempts = make(map[configImportKey]*configImportAttempt) + } + attempt := &configImportAttempt{key: key} + g.attempts[key] = attempt + return attempt, nil +} + +func (g *configImportRequests) finish(attempt *configImportAttempt, retain bool, now time.Time) { + g.mu.Lock() + defer g.mu.Unlock() + if retain { + attempt.completedAt = now + } else { + delete(g.attempts, attempt.key) + } +} diff --git a/internal/api/configapi/config_import_replay_test.go b/internal/api/configapi/config_import_replay_test.go new file mode 100644 index 0000000000..47b2417536 --- /dev/null +++ b/internal/api/configapi/config_import_replay_test.go @@ -0,0 +1,320 @@ +package configapi + +import ( + "bytes" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" +) + +const replayTestPassphrase = "synthetic-replay-control-passphrase" + +func replayTestHandler(t *testing.T) (*ConfigHandlers, *atomic.Int32) { + t.Helper() + dir := t.TempDir() + t.Setenv("PULSE_DATA_DIR", dir) + t.Setenv("PULSE_DEV", "true") + p := config.NewConfigPersistence(dir) + if err := p.SaveSystemSettings(*config.DefaultSystemSettings()); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "config.json"), []byte("{}"), 0600); err != nil { + t.Fatal(err) + } + calls := &atomic.Int32{} + h := &ConfigHandlers{ + defaultConfig: &config.Config{DataPath: dir, ConfigPath: dir}, + defaultPersistence: p, + reloadFunc: func() error { calls.Add(1); return nil }, + } + h.SetRuntimeDependencies(RuntimeDependencies{}) + store := p.GetGuestMetadataStore() + t.Cleanup(func() { + if err := store.Close(3 * time.Second); err != nil { + t.Error(err) + } + }) + return h, calls +} + +func replayTestRequest(t *testing.T, name string) ImportConfigRequest { + t.Helper() + data := config.ExportData{ + Version: "4.4", + Nodes: config.NodesConfig{PVEInstances: []config.PVEInstance{{Name: name, Host: "https://" + name + ".invalid:8006", Disabled: true, VerifySSL: true, TokenValue: "synthetic-" + name + "-secret"}}}, + System: *config.DefaultSystemSettings(), + GuestMetadata: map[string]*config.GuestMetadata{"synthetic:101": {Description: name}}, + APITokens: []config.APITokenRecord{{ID: name, Hash: "synthetic-" + name + "-hash", OrgID: "default"}}, + } + data.System.PublicURL = "https://" + name + ".invalid" + plain, err := json.Marshal(data) + if err != nil { + t.Fatal(err) + } + return ImportConfigRequest{Data: encryptedImportValidationFixture(t, plain, replayTestPassphrase), Passphrase: replayTestPassphrase} +} + +func replayTestBody(t *testing.T, request ImportConfigRequest) []byte { + t.Helper() + body, err := json.Marshal(request) + if err != nil { + t.Fatal(err) + } + return body +} + +func replayTestSubmit(h *ConfigHandlers, body []byte) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + h.HandleImportConfig(rec, httptest.NewRequest(http.MethodPost, "/api/config/import", bytes.NewReader(body))) + return rec +} + +func replayTestNewerEdit(t *testing.T, h *ConfigHandlers) { + t.Helper() + p := h.defaultPersistence + if err := p.SaveNodesConfig([]config.PVEInstance{{Name: "newer-edit", Host: "https://newer.invalid:8006", TokenValue: "synthetic-newer-secret", Disabled: true, VerifySSL: true}}, nil, nil); err != nil { + t.Fatal(err) + } + settings := config.DefaultSystemSettings() + settings.PublicURL = "https://newer.invalid" + if err := p.SaveSystemSettings(*settings); err != nil { + t.Fatal(err) + } + if err := p.SaveAPITokens([]config.APITokenRecord{{ID: "newer-token", Hash: "synthetic-newer-hash", OrgID: "default"}}); err != nil { + t.Fatal(err) + } + if err := p.GetGuestMetadataStore().Set("synthetic:101", &config.GuestMetadata{Description: "newer metadata"}); err != nil { + t.Fatal(err) + } +} + +// Real loopback HTTP drops two response bodies after running the real handler. +// Three equal POSTs must apply once, even if a legitimate edit follows the +// unseen first response. No monitored guest or external endpoint is contacted. +func TestHandleImportConfigLostResponsePreservesNewerEdit(t *testing.T) { + h, calls := replayTestHandler(t) + request := replayTestRequest(t, "restored") + body := replayTestBody(t, request) + var wire atomic.Int32 + var filesAfterEdit map[string]string + var events []*httptest.ResponseRecorder + observed := make(chan *httptest.ResponseRecorder, 3) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := wire.Add(1) + rec := httptest.NewRecorder() + h.HandleImportConfig(rec, r) + if n <= 2 { + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + conn.Close() + } else { + for name, values := range rec.Header() { + w.Header()[name] = values + } + w.WriteHeader(rec.Code) + _, _ = w.Write(rec.Body.Bytes()) + } + observed <- rec + })) + defer srv.Close() + client := srv.Client() + client.Timeout = 10 * time.Second + for i := 0; i < 3; i++ { + resp, err := client.Post(srv.URL+"/api/config/import", "application/json", bytes.NewReader(body)) + if i < 2 { + if err == nil { + resp.Body.Close() + t.Fatal("response-loss control unexpectedly received a response") + } + } else { + if err != nil { + t.Fatal(err) + } + responseBody, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil { + t.Fatal(readErr) + } + if resp.StatusCode != http.StatusConflict { + t.Errorf("third wire response = %d %s; want 409", resp.StatusCode, responseBody) + } + } + select { + case rec := <-observed: + events = append(events, rec) + case <-time.After(10 * time.Second): + t.Fatal("handler did not complete") + } + if i == 0 { + if events[0].Code != http.StatusOK { + t.Fatalf("first import = %d %s", events[0].Code, events[0].Body.String()) + } + replayTestNewerEdit(t, h) + filesAfterEdit = metadataImportFiles(t, h.defaultPersistence.GetConfigDir()) + } + if i > 0 && !reflect.DeepEqual(filesAfterEdit, metadataImportFiles(t, h.defaultPersistence.GetConfigDir())) { + t.Error("response-loss replay changed persisted files or recovery backups") + } + if i > 0 && events[i].Code != http.StatusConflict { + t.Errorf("replay %d = %d, want conflict", i, events[i].Code) + } + } + if wire.Load() != 3 || calls.Load() != 1 { + t.Errorf("three POSTs caused %d reloads, want exactly one", calls.Load()) + } + settings, err := h.defaultPersistence.LoadSystemSettings() + if err != nil { + t.Fatal(err) + } + tokens, err := h.defaultPersistence.LoadAPITokens() + if err != nil { + t.Fatal(err) + } + nodes, err := h.defaultPersistence.LoadNodesConfig() + if err != nil { + t.Fatal(err) + } + metadata := h.defaultPersistence.GetGuestMetadataStore().Get("synthetic:101") + if settings.PublicURL != "https://newer.invalid" || len(tokens) != 1 || tokens[0].ID != "newer-token" || len(nodes.PVEInstances) != 1 || nodes.PVEInstances[0].Name != "newer-edit" || metadata == nil || metadata.Description != "newer metadata" { + t.Error("response-loss replays overwrote a newer legitimate edit") + } + logReplayProducer(t, "completed", request, events[2]) +} + +func TestHandleImportConfigOverlappingRequestsDoNotWriteOrReload(t *testing.T) { + for _, overlap := range []string{"same archive", "different archive"} { + t.Run(overlap, func(t *testing.T) { + h, calls := replayTestHandler(t) + first := replayTestBody(t, replayTestRequest(t, "first")) + request := replayTestRequest(t, "second") + if overlap == "same archive" { + if err := json.Unmarshal(first, &request); err != nil { + t.Fatal(err) + } + } + started, release, done := make(chan struct{}), make(chan struct{}), make(chan *httptest.ResponseRecorder, 1) + h.reloadFunc = func() error { + if calls.Add(1) == 1 { + close(started) + <-release + } + return nil + } + go func() { done <- replayTestSubmit(h, first) }() + released := false + defer func() { + if !released { + close(release) + } + <-done + }() + select { + case <-started: + case <-time.After(10 * time.Second): + t.Fatal("first import did not reach apply") + } + before := metadataImportFiles(t, h.defaultPersistence.GetConfigDir()) + rec := replayTestSubmit(h, replayTestBody(t, request)) + if rec.Code != http.StatusConflict { + t.Errorf("overlap = %d %s, want 409", rec.Code, rec.Body.String()) + } + if calls.Load() != 1 || !reflect.DeepEqual(before, metadataImportFiles(t, h.defaultPersistence.GetConfigDir())) { + t.Error("overlap repeated disk writes/backups or reloaded monitoring") + } + logReplayProducer(t, "in-flight", request, rec) + close(release) + released = true + }) + } +} + +func TestHandleImportConfigAppliedFailureCannotReplay(t *testing.T) { + h, calls := replayTestHandler(t) + h.reloadFunc = func() error { calls.Add(1); return errors.New("synthetic apply failure") } + request := replayTestRequest(t, "applied-before-error") + body := replayTestBody(t, request) + rec := replayTestSubmit(h, body) + if rec.Code != http.StatusInternalServerError { + t.Fatalf("first apply failure = %d", rec.Code) + } + replayTestNewerEdit(t, h) + before := metadataImportFiles(t, h.defaultPersistence.GetConfigDir()) + rec = replayTestSubmit(h, body) + if rec.Code != http.StatusConflict || calls.Load() != 1 || !reflect.DeepEqual(before, metadataImportFiles(t, h.defaultPersistence.GetConfigDir())) { + t.Error("a committed import with a failed runtime apply was reapplied") + } + logReplayProducer(t, "applied-with-error", request, rec) +} + +func TestHandleImportConfigUncommittedFailureCanRetry(t *testing.T) { + h, calls := replayTestHandler(t) + request := replayTestRequest(t, "retry-after-storage-repair") + body := replayTestBody(t, request) + path := filepath.Join(h.defaultPersistence.GetConfigDir(), "guest_metadata.json") + if err := os.Mkdir(path, 0700); err != nil { + t.Fatal(err) + } + if rec := replayTestSubmit(h, body); rec.Code != http.StatusBadRequest { + t.Fatalf("failed commit status = %d", rec.Code) + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if rec := replayTestSubmit(h, body); rec.Code != http.StatusOK || calls.Load() != 1 { + t.Fatalf("corrected retry = %d %s", rec.Code, rec.Body.String()) + } +} + +func TestHandleImportConfigWrongPassphraseCannotReuseSuccess(t *testing.T) { + h, calls := replayTestHandler(t) + request := replayTestRequest(t, "passphrase-bound") + if rec := replayTestSubmit(h, replayTestBody(t, request)); rec.Code != http.StatusOK { + t.Fatalf("first import = %d", rec.Code) + } + before := metadataImportFiles(t, h.defaultPersistence.GetConfigDir()) + request.Passphrase = "synthetic-wrong-passphrase" + rec := replayTestSubmit(h, replayTestBody(t, request)) + if rec.Code != http.StatusBadRequest || calls.Load() != 1 || !reflect.DeepEqual(before, metadataImportFiles(t, h.defaultPersistence.GetConfigDir())) { + t.Error("wrong passphrase reused successful admission or changed state") + } +} + +func logReplayProducer(t *testing.T, state string, request ImportConfigRequest, rec *httptest.ResponseRecorder) { + t.Helper() + data, err := json.Marshal(map[string]any{"state": state, "request": request, "status": rec.Code, "contentType": rec.Header().Get("Content-Type"), "body": rec.Body.String()}) + if err != nil { + t.Fatal(err) + } + t.Logf("SYNTHETIC_IMPORT_REPLAY_READBACK %s", data) +} + +func TestHandleImportConfigDifferentArchiveThenOldReplay(t *testing.T) { + h, calls := replayTestHandler(t) + first := replayTestBody(t, replayTestRequest(t, "first")) + second := replayTestBody(t, replayTestRequest(t, "second")) + for _, body := range [][]byte{first, second} { + if rec := replayTestSubmit(h, body); rec.Code != http.StatusOK { + t.Fatalf("distinct import = %d %s", rec.Code, rec.Body.String()) + } + } + before := metadataImportFiles(t, h.defaultPersistence.GetConfigDir()) + if rec := replayTestSubmit(h, first); rec.Code != http.StatusConflict { + t.Fatalf("old replay = %d", rec.Code) + } + if calls.Load() != 2 || !reflect.DeepEqual(before, metadataImportFiles(t, h.defaultPersistence.GetConfigDir())) { + t.Error("old replay replaced the newer legitimate import") + } +} diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 0f79e11b21..804456e174 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -27960,3 +27960,84 @@ func TestHostDiskUsageHistoryAndSummaryRespectPresence(t *testing.T) { }) } } + +// Replay protection is behind the normal router authentication/CSRF and is +// scoped to the selected workspace, not a global archive hash. Every monitored +// node is disabled and every credential is synthetic. +func TestContract_ConfigImportReplayRemainsAuthenticatedAndScoped(t *testing.T) { + SetMultiTenantEnabled(true) + t.Cleanup(func() { SetMultiTenantEnabled(false) }) + t.Setenv("PULSE_DEV", "true") + t.Setenv("PULSE_HOSTED_MODE", "false") + dir := t.TempDir() + t.Setenv("PULSE_DATA_DIR", dir) + mtp := config.NewMultiTenantPersistence(dir) + for _, orgID := range []string{"default", "tenant-a", "tenant-b"} { + if err := mtp.SaveOrganization(&models.Organization{ID: orgID, DisplayName: orgID, OwnerUserID: "import-owner"}); err != nil { + t.Fatal(err) + } + p, err := mtp.GetPersistence(orgID) + if err != nil { + t.Fatal(err) + } + seedConfigImportRuntime(t, p, orgID, orgID, 71) + } + cfg, err := config.LoadWithoutLoggingInit() + if err != nil { + t.Fatal(err) + } + cfg.AuthUser, cfg.AuthPass = "import-owner", "synthetic-password-hash" + mtm := monitoring.NewMultiTenantMonitor(cfg, mtp, nil) + t.Cleanup(mtm.Stop) + defaultMonitor, err := mtm.GetMonitor("default") + if err != nil { + t.Fatal(err) + } + router := NewRouter(cfg, defaultMonitor, mtm, nil, nil, "test") + cleanupTestRouter(t, router) + source := config.NewConfigPersistence(t.TempDir()) + archive, err := source.ExportConfig("synthetic-replay-passphrase") + if err != nil { + t.Fatal(err) + } + body, err := json.Marshal(ImportConfigRequest{Data: archive, Passphrase: "synthetic-replay-passphrase"}) + if err != nil { + t.Fatal(err) + } + const session = "synthetic-replay-owner-session" + GetSessionStore().CreateSession(session, time.Hour, "browser", "127.0.0.1", "import-owner") + for _, orgID := range []string{"tenant-a", "tenant-b"} { + t.Run(orgID, func(t *testing.T) { + for _, want := range []int{http.StatusOK, http.StatusConflict} { + req := httptest.NewRequest(http.MethodPost, "/api/config/import", bytes.NewReader(body)) + req.RemoteAddr = "127.0.0.1:1234" + req.Header.Set("X-Pulse-Org-ID", orgID) + req.Header.Set("X-CSRF-Token", generateCSRFToken(session)) + req.AddCookie(&http.Cookie{Name: cookieNameSession, Value: session}) + rec := httptest.NewRecorder() + router.Handler().ServeHTTP(rec, req) + if rec.Code != want { + t.Fatalf("scoped request = %d %s; want %d", rec.Code, rec.Body.String(), want) + } + if want == http.StatusConflict && (!strings.Contains(rec.Body.String(), "has not been imported again") || strings.Contains(rec.Body.String(), archive) || strings.Contains(rec.Body.String(), "synthetic-replay-passphrase")) { + t.Fatal("replay response lost truthful/private error boundary") + } + } + }) + } + // A known archive must not become an authentication or CSRF bypass. + for _, authenticated := range []bool{false, true} { + counted := &countingConfigTransferBody{reader: bytes.NewReader(body)} + req := httptest.NewRequest(http.MethodPost, "/api/config/import", counted) + req.Header.Set("X-Pulse-Org-ID", "tenant-a") + req.RemoteAddr = "127.0.0.1:1234" + if authenticated { + req.AddCookie(&http.Cookie{Name: cookieNameSession, Value: session}) + } + rec := httptest.NewRecorder() + router.Handler().ServeHTTP(rec, req) + if (rec.Code != http.StatusUnauthorized && rec.Code != http.StatusForbidden) || counted.reads != 0 { + t.Fatalf("replay bypassed auth/CSRF: status=%d reads=%d", rec.Code, counted.reads) + } + } +}