mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 13:10:03 +00:00
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
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user