From 952e593d745c607f1f9a6451ce8d3ce5fc8b0e10 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Thu, 23 Jul 2026 23:50:01 +0100 Subject: [PATCH] Fix agent update reconciliation across cached proxies --- README.md | 11 +- docs/UPGRADE_v6.md | 21 ++- .../v6/internal/subsystems/agent-lifecycle.md | 19 +++ .../v6/internal/subsystems/api-contracts.md | 10 ++ .../subsystems/performance-and-scalability.md | 5 + .../internal/subsystems/security-privacy.md | 8 + .../internal/subsystems/storage-recovery.md | 5 + internal/agentupdate/coverage_test.go | 22 +-- internal/agentupdate/export_test.go | 4 +- internal/agentupdate/update.go | 64 +++++++- internal/agentupdate/update_http_test.go | 138 +++++++++++++++++- internal/api/contract_test.go | 29 ++++ internal/api/router.go | 3 + 13 files changed, 306 insertions(+), 33 deletions(-) diff --git a/README.md b/README.md index 576b21712..22337e95b 100644 --- a/README.md +++ b/README.md @@ -114,9 +114,14 @@ Note: this installs the Pulse **server**. Agent installs and v5-to-v6 agent upgr Server and agent updates are separate lifecycle paths. Updating the server sets the version eligible v6 agents should reach, but it does not prove the fleet has -converged: those agents update asynchronously, while v5 agents, PVE host agents, -and agents with auto-update disabled or failed update prerequisites need the -manual per-host command from **Settings → Infrastructure**. +converged. Current Unified Agents on Linux, Windows, and Docker-enabled hosts +check shortly after startup and then hourly, install only a newer semantic +version (including RC-to-stable), and retry after offline or failed checks. +They bind the binary download to the exact server target so intermediary caches +cannot silently return an older agent. v5 agents, PVE host agents, and agents +with auto-update disabled or failed authentication, transport, trust, download, +or self-test prerequisites need the manual per-host command from +**Settings → Infrastructure**. ### Option 2: Docker ```bash diff --git a/docs/UPGRADE_v6.md b/docs/UPGRADE_v6.md index 330be7e15..fa3d877ae 100644 --- a/docs/UPGRADE_v6.md +++ b/docs/UPGRADE_v6.md @@ -88,8 +88,16 @@ No. Use the unified installer to upgrade existing agent deployments in place. Ge No. The server and Unified Agent have separate update lifecycles. After the server changes the target version, eligible v6 agents normally discover and -apply that update asynchronously during their update checks. The server being -current is not proof that every installed agent has checked in or converged. +apply that update asynchronously: once about five seconds after process start, +then hourly while the process remains running. Linux, Windows, and +Docker-enabled installations all use this same Unified Agent updater. It moves +only to a higher semantic version, so RC-to-later-RC, RC-to-stable, and +stable-to-later-stable updates are eligible; a server advertising an older +target never causes an automatic downgrade. Version checks are cache-unique, +and the binary request is bound to the exact target version. Offline checks, +download failures, and pre-replacement failures leave the current process and +binary in place and are retried on a later check. The server being current is +not proof that every installed agent has checked in or converged. Use the manual path for v5 agents, PVE host agents, agents with auto-update disabled, and agents blocked by authentication, missing connection state, @@ -104,6 +112,15 @@ for the operator to copy and run; it does not remotely execute fleet updates. Ag self-update and manual update both still depend on valid authentication, a reachable trusted update channel, and accepted release signing keys. +Early v6 prerelease agents could report successfully to a private plain-HTTP +Pulse URL while their separate updater rejected that same URL before making a +version request. Such an agent cannot download the release that fixes its own +transport policy. Re-run the current per-host installer command once to +preserve the URL/token/trust settings in the current lifecycle format; normal +automatic checks resume after that migration. The same manual recovery rule +applies to an agent whose installed signing trust cannot accept the current +release. + ### Will an upgraded v5 agent keep the same identity in v6? Yes. The v5-to-v6 agent path is expected to preserve one canonical agent diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index f72940525..6a6f76eb9 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -458,6 +458,19 @@ spellings as the same recovered state while preserving the same fail-closed URL-plus-token threshold. Windows stale-agent update commands remain on the existing token-gated install transport until the Windows installer owns an equivalent saved-state update mode. +The current Unified Agent updater is the sole automatic binary-update +authority for Linux, Windows, and Docker-enabled `pulse-agent` installations. +It checks once after the five-second startup delay and then hourly, compares +normalized semantic versions across RC and stable boundaries, and refuses +downgrades. Each version check carries the running agent version plus a +request-unique non-secret cache key; each binary download carries the exact +server target version. The server version response and binary response are +non-cacheable, so a reverse proxy may not reconcile a fresh target with an +older cached artifact. Offline, authentication, checksum, signature, +self-test, or pre-replacement failures retain the running binary and retry on +a later check. The legacy updater in `internal/dockeragent/self_update.go` is +not a second v6 authority: Docker and Podman are modules inside the unified +agent and set `AgentType=unified`, which bypasses that compatibility path. The report contract separates runtime family from OS identity. Newly built agents derive `Host.Platform` from their compiled GOOS and keep the distribution or appliance caption in `Host.OSName`; a Mageia agent therefore @@ -3874,6 +3887,12 @@ the matching base64-encoded `X-Signature-SSHSIG`, and `internal/api/unified_agent.go` must only serve published release installers and agent binaries from local or proxied assets that carry the matching detached signature sidecars. +Target-version cache keys do not relax that trust boundary. A cache-keyed +download must still pass the exact checksum, embedded-key Ed25519 signature, +platform magic, downloaded-binary self-test, same-filesystem atomic +replacement, and backup restoration rules before restart. The cache key is +version selection metadata only and must never be treated as artifact +integrity or downgrade authorization. That same self-update pre-flight must keep the live agent token out of process argv. `internal/agentupdate/update.go` and legacy `internal/dockeragent/self_update.go` may pass a short-lived `0600` token file diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 68589f594..d7d199f4e 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -3598,6 +3598,16 @@ auto-register mutation boundary. ## Current State +### Agent update target responses are reconciliation-safe + +`GET` and `HEAD /api/agent/version` project +`currentAgentTargetVersion()` as the one server-owned agent target and return +explicit `no-store` / `no-cache` response metadata. Current agents add their +running version and a non-secret request-unique cache key, while the subsequent +binary request names the exact returned target version. Query values affect +intermediary cache identity only; they do not select an arbitrary server +release, bypass artifact validation, or replace the canonical response body. + ### vSphere host composition on the vCenter system row Grouped systems in `GET /api/connections` now project ESXi host resources as diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 67f67b8a3..4d2faa0cf 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -2006,3 +2006,8 @@ cluster scope needed for that calculation, but must not add per-row requests, runtime-UID scans, or a second metadata cache. Kubernetes persistence uses the fully scoped cluster, kind, namespace, and name identity; runtime pod UIDs are legacy fallback coordinates only. + +The public agent-version reconciliation path on `internal/api/router.go` +remains an O(1) target projection. Cache-prevention headers and the +agent-supplied non-secret cache key add no inventory scan, persistence read, +metrics fan-out, or background work to `GET /api/agent/version`. diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index 97dd7a175..9b2ef6b61 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -1581,3 +1581,11 @@ be reused to satisfy another organization, including Assistant-authored URL updates, and stable workload identities must retain their full host or cluster, namespace, kind, and name scope so a saved URL cannot cross a tenant or resource boundary. + +Agent update cache reconciliation on the shared router preserves the existing +credential boundary. `/api/agent/version` is non-cacheable, and current agents +put only the running version plus a non-secret request nonce in its query; +runtime tokens remain in protected request headers and redirects remain +rejected. The target-version query on `/download/pulse-agent` is cache identity, +not trust evidence: checksum, embedded-key signature, self-test, and atomic +replacement validation remain mandatory and fail closed. diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index b0b5c953e..09e42426c 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -2293,6 +2293,11 @@ fields and nullable mode/kind metadata before presenting canonical item labels, while storage detail drawers and filter controls must route summary series IDs, source tones, and disk metrics through the shared storage helpers instead of reconstructing them from local table state. + +Unified Agent target cache prevention on the shared router is adjacent +lifecycle/API transport. `/api/agent/version` no-store headers and non-secret +request cache keys must not be interpreted as recovery freshness, retained +backup evidence, rollback coverage, or a storage-local artifact version. Config export/import remains adjacent persistence infrastructure rather than a storage/recovery authority. Guest metadata import must reload the active tenant monitor's live store so recovery and resource projections observe one diff --git a/internal/agentupdate/coverage_test.go b/internal/agentupdate/coverage_test.go index b8fb253ce..70b37caa4 100644 --- a/internal/agentupdate/coverage_test.go +++ b/internal/agentupdate/coverage_test.go @@ -535,7 +535,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { t.Run("Disabled", func(t *testing.T) { u := newUpdaterForTest("https://example") u.cfg.Disabled = true - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update when disabled") return nil } @@ -545,7 +545,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { t.Run("DevCurrent", func(t *testing.T) { u := newUpdaterForTest("https://example") u.cfg.CurrentVersion = "dev" - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update in dev mode") return nil } @@ -554,7 +554,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { t.Run("NoPulseURL", func(t *testing.T) { u := newUpdaterForTest("") - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update without URL") return nil } @@ -563,7 +563,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { t.Run("InsecureHTTPURL", func(t *testing.T) { u := newUpdaterForTest("http://pulse.example.com") - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update with insecure Pulse URL") return nil } @@ -578,7 +578,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { u := newUpdaterForTest(server.URL) u.client = server.Client() - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update on server error") return nil } @@ -593,7 +593,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { u := newUpdaterForTest(server.URL) u.client = server.Client() - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update when server dev") return nil } @@ -608,7 +608,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { u := newUpdaterForTest(server.URL) u.client = server.Client() - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not update when up to date") return nil } @@ -623,7 +623,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { u := newUpdaterForTest(server.URL) u.client = server.Client() - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { t.Fatalf("should not downgrade") return nil } @@ -639,7 +639,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { u := newUpdaterForTest(server.URL) u.client = server.Client() var called int32 - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { atomic.AddInt32(&called, 1) return nil } @@ -658,7 +658,7 @@ func TestCheckAndUpdateBranches(t *testing.T) { u := newUpdaterForTest(server.URL) u.client = server.Client() var called int32 - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { atomic.AddInt32(&called, 1) return errors.New("fail") } @@ -686,7 +686,7 @@ func TestRunLoop(t *testing.T) { defer cancel() var called int32 - u.performUpdateFn = func(context.Context) error { + u.performUpdateFn = func(context.Context, string) error { if atomic.AddInt32(&called, 1) >= 2 { cancel() } diff --git a/internal/agentupdate/export_test.go b/internal/agentupdate/export_test.go index 7d17cb427..abec7232d 100644 --- a/internal/agentupdate/export_test.go +++ b/internal/agentupdate/export_test.go @@ -43,8 +43,8 @@ func UseExecPathForUpdateChecksForTest(u *Updater, execPath string) func() { evalSymlinksFn = func(string) (string, error) { return execPath, nil } restartProcessFn = func(string) error { return nil } u.selfTestFn = func(context.Context, string) error { return nil } - u.performUpdateFn = func(ctx context.Context) error { - return u.performUpdateWithExecPath(ctx, execPath) + u.performUpdateFn = func(ctx context.Context, targetVersion string) error { + return u.performUpdateWithExecPathForVersion(ctx, execPath, targetVersion) } return func() { diff --git a/internal/agentupdate/update.go b/internal/agentupdate/update.go index f49ba2e23..3bd08ac64 100644 --- a/internal/agentupdate/update.go +++ b/internal/agentupdate/update.go @@ -12,6 +12,7 @@ import ( "fmt" "io" "net/http" + "net/url" "os" "os/exec" "path" @@ -112,6 +113,7 @@ var ( readFileFn = os.ReadFile writeFileFn = os.WriteFile retrySleepFn = sleepWithContext + updateRequestNowFn = time.Now ) // Config holds the configuration for the updater. @@ -162,7 +164,7 @@ type Updater struct { statusMu sync.RWMutex status Status - performUpdateFn func(context.Context) error + performUpdateFn func(context.Context, string) error selfTestFn func(context.Context, string) error initialDelay time.Duration newTicker func(time.Duration) *time.Ticker @@ -228,7 +230,7 @@ func New(cfg Config) *Updater { u.status.State = UpdateStateError u.status.LastError = configErr.Error() } - u.performUpdateFn = u.performUpdate + u.performUpdateFn = u.performUpdateForVersion u.selfTestFn = u.runDownloadedBinarySelfTest u.initialDelay = 5 * time.Second u.newTicker = time.NewTicker @@ -428,7 +430,7 @@ func (u *Updater) CheckAndUpdate(ctx context.Context) { status.LastError = "" }) - if err := u.performUpdateFn(ctx); err != nil { + if err := u.performUpdateFn(ctx, serverVersion); err != nil { u.updateStatus(func(status *Status) { status.State = UpdateStateError status.LastError = err.Error() @@ -486,9 +488,18 @@ func (u *Updater) getServerVersion(ctx context.Context) (string, error) { return "", fmt.Errorf("invalid Pulse URL: %w", err) } - url := fmt.Sprintf("%s/api/agent/version", strings.TrimRight(u.cfg.PulseURL, "/")) + versionURL, err := withUpdateQuery( + fmt.Sprintf("%s/api/agent/version", strings.TrimRight(u.cfg.PulseURL, "/")), + map[string]string{ + "agentVersion": strings.TrimSpace(u.cfg.CurrentVersion), + "check": fmt.Sprintf("%d", updateRequestNowFn().UTC().UnixNano()), + }, + ) + if err != nil { + return "", fmt.Errorf("failed to build version check URL: %w", err) + } - resp, err := u.getWithRetry(ctx, url, "version check") + resp, err := u.getWithRetry(ctx, versionURL, "version check") if err != nil { return "", err } @@ -502,6 +513,22 @@ func (u *Updater) getServerVersion(ctx context.Context) (string, error) { return versionResp.Version, nil } +func withUpdateQuery(rawURL string, values map[string]string) (string, error) { + parsed, err := url.Parse(rawURL) + if err != nil { + return "", err + } + + query := parsed.Query() + for key, value := range values { + if value = strings.TrimSpace(value); value != "" { + query.Set(key, value) + } + } + parsed.RawQuery = query.Encode() + return parsed.String(), nil +} + func sleepWithContext(ctx context.Context, d time.Duration) error { timer := time.NewTimer(d) defer timer.Stop() @@ -753,11 +780,15 @@ func (u *Updater) validatePulseURL() error { // performUpdate downloads and installs the new agent binary. func (u *Updater) performUpdate(ctx context.Context) error { + return u.performUpdateForVersion(ctx, "") +} + +func (u *Updater) performUpdateForVersion(ctx context.Context, targetVersion string) error { execPath, err := resolveExecutablePath() if err != nil { return err } - return u.performUpdateWithExecPath(ctx, execPath) + return u.performUpdateWithExecPathForVersion(ctx, execPath, targetVersion) } // resolveExecutablePath returns the running binary's path, falling back to an @@ -787,6 +818,10 @@ func resolveExecutablePath() (string, error) { } func (u *Updater) performUpdateWithExecPath(ctx context.Context, execPath string) error { + return u.performUpdateWithExecPathForVersion(ctx, execPath, "") +} + +func (u *Updater) performUpdateWithExecPathForVersion(ctx context.Context, execPath, targetVersion string) error { agentName, err := normalizeAgentName(u.cfg.AgentName) if err != nil { return fmt.Errorf("invalid agent name: %w", err) @@ -802,9 +837,22 @@ func (u *Updater) performUpdateWithExecPath(ctx context.Context, execPath string // Try architecture-specific binary first, then fall back to default candidates := []string{} if archParam != "" { - candidates = append(candidates, fmt.Sprintf("%s?arch=%s", downloadBase, archParam)) + candidateURL, err := withUpdateQuery(downloadBase, map[string]string{ + "arch": archParam, + "serverVersion": targetVersion, + }) + if err != nil { + return fmt.Errorf("failed to build architecture-specific download URL: %w", err) + } + candidates = append(candidates, candidateURL) } - candidates = append(candidates, downloadBase) + fallbackURL, err := withUpdateQuery(downloadBase, map[string]string{ + "serverVersion": targetVersion, + }) + if err != nil { + return fmt.Errorf("failed to build fallback download URL: %w", err) + } + candidates = append(candidates, fallbackURL) var resp *http.Response lastErr := fmt.Errorf("failed to download binary from all candidate URLs") diff --git a/internal/agentupdate/update_http_test.go b/internal/agentupdate/update_http_test.go index 1b1e2568e..a260fd08d 100644 --- a/internal/agentupdate/update_http_test.go +++ b/internal/agentupdate/update_http_test.go @@ -50,6 +50,12 @@ func TestUpdater_getServerVersion_SetsAuthHeaders(t *testing.T) { if r.Header.Get("Authorization") != "Bearer token" { t.Fatalf("Authorization = %q", r.Header.Get("Authorization")) } + if got := r.URL.Query().Get("agentVersion"); got != "1.0.0" { + t.Fatalf("agentVersion query = %q, want %q", got, "1.0.0") + } + if got := r.URL.Query().Get("check"); got == "" { + t.Fatal("expected a cache-unique version check query") + } sawAuth = true _ = json.NewEncoder(w).Encode(serverVersionResponse{Version: "1.2.3"}) })) @@ -224,6 +230,9 @@ func TestUpdater_performUpdateWithExecPath_RequiresSignatureWhenTrustedKeysConfi data := testBinary() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.URL.Query().Get("serverVersion"); got != "6.0.0" { + t.Fatalf("serverVersion query = %q, want %q", got, "6.0.0") + } w.Header().Set(checksumSHA256Header, checksum(data)) w.Header().Set(signatureHeader, signedUpdateHeader(t, data, privateKey)) _, _ = w.Write(data) @@ -238,8 +247,8 @@ func TestUpdater_performUpdateWithExecPath_RequiresSignatureWhenTrustedKeysConfi t.Cleanup(func() { restartProcessFn = origRestart }) restartProcessFn = func(string) error { return nil } - if err := u.performUpdateWithExecPath(context.Background(), execPath); err != nil { - t.Fatalf("performUpdateWithExecPath: %v", err) + if err := u.performUpdateWithExecPathForVersion(context.Background(), execPath, "6.0.0"); err != nil { + t.Fatalf("performUpdateWithExecPathForVersion: %v", err) } } @@ -264,21 +273,21 @@ func TestUpdater_performUpdateWithExecPath_RejectsMissingSignatureWhenTrustedKey func TestUpdater_CheckAndUpdate_EarlyReturns(t *testing.T) { u := New(Config{Disabled: true}) - u.performUpdateFn = func(ctx context.Context) error { + u.performUpdateFn = func(ctx context.Context, _ string) error { t.Fatalf("performUpdate should not be called") return nil } u.CheckAndUpdate(context.Background()) u = New(Config{CurrentVersion: "dev"}) - u.performUpdateFn = func(ctx context.Context) error { + u.performUpdateFn = func(ctx context.Context, _ string) error { t.Fatalf("performUpdate should not be called") return nil } u.CheckAndUpdate(context.Background()) u = New(Config{CurrentVersion: "1.0.0", PulseURL: ""}) - u.performUpdateFn = func(ctx context.Context) error { + u.performUpdateFn = func(ctx context.Context, _ string) error { t.Fatalf("performUpdate should not be called") return nil } @@ -297,6 +306,10 @@ func TestUpdater_CheckAndUpdate_VersionComparePaths(t *testing.T) { {"server-older", "1.0.1", "1.0.0", false, true}, {"server-dev", "1.0.0", "dev", false, true}, {"server-newer", "1.0.0", "1.0.1", true, true}, + {"release-candidate-to-release-candidate", "6.0.0-rc.1", "6.0.0-rc.6", true, true}, + {"release-candidate-to-stable", "6.0.0-rc.6", "6.0.0", true, true}, + {"stable-to-stable", "6.0.0", "6.1.1", true, true}, + {"stable-does-not-downgrade-to-release-candidate", "6.0.0", "6.0.0-rc.7", false, true}, } for _, tc := range tests { @@ -314,8 +327,11 @@ func TestUpdater_CheckAndUpdate_VersionComparePaths(t *testing.T) { CurrentVersion: tc.current, CheckInterval: time.Minute, }) - u.performUpdateFn = func(ctx context.Context) error { + u.performUpdateFn = func(ctx context.Context, targetVersion string) error { called = true + if targetVersion != tc.server { + t.Fatalf("target version = %q, want %q", targetVersion, tc.server) + } return nil } @@ -328,6 +344,114 @@ func TestUpdater_CheckAndUpdate_VersionComparePaths(t *testing.T) { } } +func TestUpdater_CheckAndUpdate_ReconcilesVersionAndBinaryAcrossCachingProxy(t *testing.T) { + originalKeys := updatesignature.EmbeddedTrustedPublicKeys + updatesignature.EmbeddedTrustedPublicKeys = "" + t.Cleanup(func() { updatesignature.EmbeddedTrustedPublicKeys = originalKeys }) + + targetVersion := "6.0.0" + staleBinary := append(testBinary(), []byte("-stale-rc.1")...) + targetBinary := append(testBinary(), []byte("-stable-6.0.0")...) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/agent/version": + if r.URL.Query().Get("check") == "" { + _ = json.NewEncoder(w).Encode(serverVersionResponse{Version: "6.0.0-rc.1"}) + return + } + if got := r.URL.Query().Get("agentVersion"); got != "6.0.0-rc.1" { + t.Fatalf("agentVersion query = %q, want %q", got, "6.0.0-rc.1") + } + _ = json.NewEncoder(w).Encode(serverVersionResponse{Version: targetVersion}) + case "/download/pulse-agent": + payload := staleBinary + if r.URL.Query().Get("serverVersion") == targetVersion { + payload = targetBinary + } + w.Header().Set(checksumSHA256Header, checksum(payload)) + _, _ = w.Write(payload) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + _, execPath := writeTempExec(t) + u := New(Config{ + PulseURL: server.URL, + AgentName: "pulse-agent", + CurrentVersion: "6.0.0-rc.1", + }) + u.client = server.Client() + u.selfTestFn = func(context.Context, string) error { return nil } + u.performUpdateFn = func(ctx context.Context, version string) error { + return u.performUpdateWithExecPathForVersion(ctx, execPath, version) + } + + originalRestart := restartProcessFn + restartProcessFn = func(string) error { return nil } + t.Cleanup(func() { restartProcessFn = originalRestart }) + + u.CheckAndUpdate(context.Background()) + + got, err := os.ReadFile(execPath) + if err != nil { + t.Fatalf("read updated binary: %v", err) + } + if string(got) != string(targetBinary) { + t.Fatalf("updated binary came from stale cache: got %q want %q", got, targetBinary) + } + status := u.Snapshot() + if status.State != UpdateStateIdle || status.LastSuccessAt == nil || status.LastError != "" { + t.Fatalf("reconciled update status = %+v", status) + } +} + +func TestUpdater_CheckAndUpdate_RecoversAfterOfflineAndPartialFailure(t *testing.T) { + var versionChecks int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&versionChecks, 1) <= updateRequestMaxAttempts { + http.Error(w, "offline", http.StatusServiceUnavailable) + return + } + _ = json.NewEncoder(w).Encode(serverVersionResponse{Version: "6.1.1"}) + })) + defer server.Close() + + originalSleep := retrySleepFn + retrySleepFn = func(context.Context, time.Duration) error { return nil } + t.Cleanup(func() { retrySleepFn = originalSleep }) + + u := New(Config{PulseURL: server.URL, CurrentVersion: "6.1.0"}) + var updateAttempts int32 + u.performUpdateFn = func(context.Context, string) error { + if atomic.AddInt32(&updateAttempts, 1) == 1 { + return errors.New("partial replacement rejected") + } + return nil + } + + u.CheckAndUpdate(context.Background()) + if status := u.Snapshot(); status.State != UpdateStateError || status.LastCheckedAt == nil { + t.Fatalf("offline status = %+v", status) + } + + u.CheckAndUpdate(context.Background()) + if status := u.Snapshot(); status.State != UpdateStateError || status.LastError != "partial replacement rejected" { + t.Fatalf("partial-failure status = %+v", status) + } + + u.CheckAndUpdate(context.Background()) + status := u.Snapshot() + if status.State != UpdateStateIdle || status.LastSuccessAt == nil || status.LastError != "" { + t.Fatalf("recovered status = %+v", status) + } + if got := atomic.LoadInt32(&updateAttempts); got != 2 { + t.Fatalf("update attempts = %d, want 2", got) + } +} + func TestUpdater_CheckAndUpdate_RecordsLifecycleStatus(t *testing.T) { t.Run("current", func(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -354,7 +478,7 @@ func TestUpdater_CheckAndUpdate_RecordsLifecycleStatus(t *testing.T) { defer srv.Close() u := New(Config{PulseURL: srv.URL, CurrentVersion: "1.0.0"}) - u.performUpdateFn = func(context.Context) error { return errors.New("replacement rejected") } + u.performUpdateFn = func(context.Context, string) error { return errors.New("replacement rejected") } u.CheckAndUpdate(context.Background()) status := u.Snapshot() diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index fd1d16c1a..f63bca046 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -65,6 +65,35 @@ type basicActionContractAuthorizer struct { wantUser string } +func TestContract_AgentVersionResponsePreventsStaleReconciliation(t *testing.T) { + router := &Router{} + req := httptest.NewRequest(http.MethodGet, "/api/agent/version?agentVersion=6.0.0-rc.1&check=123", nil) + rec := httptest.NewRecorder() + + router.handleAgentVersion(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK) + } + if got := rec.Header().Get("Cache-Control"); got != "no-store, no-cache, must-revalidate, max-age=0" { + t.Fatalf("Cache-Control = %q", got) + } + if got := rec.Header().Get("Pragma"); got != "no-cache" { + t.Fatalf("Pragma = %q", got) + } + if got := rec.Header().Get("Expires"); got != "0" { + t.Fatalf("Expires = %q", got) + } + + var payload AgentVersionResponse + if err := json.NewDecoder(rec.Body).Decode(&payload); err != nil { + t.Fatalf("decode agent version response: %v", err) + } + if payload.Version == "" { + t.Fatal("agent target version is empty") + } +} + func TestContract_OperationalTrustAttentionActionOfferJSONIsAdditiveAndTyped(t *testing.T) { payload, err := json.Marshal(ai.AttentionActionOffer{ ActionID: "action-1", diff --git a/internal/api/router.go b/internal/api/router.go index 8afd66cae..2804afb2a 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -5755,6 +5755,9 @@ func (r *Router) handleAgentVersion(w http.ResponseWriter, req *http.Request) { Version: version, } + w.Header().Set("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0") + w.Header().Set("Pragma", "no-cache") + w.Header().Set("Expires", "0") w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(response) }