diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index bb986360b0..a32223caff 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8278,6 +8278,52 @@ report path relies on: an applied report carries its token's ID, a report the ordering watermark rejects returns the stored host and hostname, and mock mode's discard acknowledgement carries no token. +### The command channel registers under the identity Pulse acknowledges + +A token binds the first agent ID its command channel presents and never moves +afterwards, while Pulse resolves the agent's reports to its own host ID, which +the acknowledgement returns as `agentId` and the agent persists and presents +after a restart. When the two differed (a configured `--agent-id`, an ID Pulse +forked for a cloned machine ID, or one it kept from an earlier enrollment), the +first registration bound the presented ID, the config gate, which judges the +resolved ID, turned commands off, and the next restart was refused for good. +The agent therefore keeps its command channel staged until Pulse acknowledges a +report, then registers under the identity the acknowledgement names as +`commandAgentId`. Only a real acknowledgement negotiates: one that reports +success and names the host, for a report presenting the agent ID and hostname +this process registers with, so a report buffered by an earlier run under +another identity neither names nor releases the channel. Pulse names the +identity under the rule in the security-privacy contract, only when the +token's binding holds the resolved ID or would take it on that registration, +which also lets an agent restarted under a different `--agent-id` register +under the identity its token recorded. An acknowledgement +without it (an older Pulse, or a token already bound to the ID the agent +presents) leaves the channel on the identity it already uses, the agent's own +ID by default. An unreadable acknowledgement is no answer and the channel keeps +waiting; a 403, which means the token lacks report scope and Pulse can never +answer, releases it under the agent's own ID as before. A client Pulse has +admitted is never rebuilt under a newly named identity, because a second client +would reopen the operation receipt store and mark in-flight operations +interrupted; a client whose registration Pulse has never accepted has received +no operation, and the agent retires and rebuilds it under the named identity, +with a compare-and-swap on the client so a retired client can never be admitted +afterwards. Observer acknowledgements neither name nor release the channel. A +token bound to an ID its binding will not give up keeps it, since no report +moves a binding: re-running the install command issues a fresh token that +binds the resolved identity. +`TestCommandChannelRegistersUnderTheIdentityPulseAcknowledges` and the other +`TestCommandChannel*`, `TestRunningCommandChannelIsNeverReplacedByALaterIdentity`, +`TestRefusedCommandChannelMovesToTheIdentityALaterAckNames`, +`TestReplayedReportAckNeverNegotiatesTheCurrentChannel`, +`TestObserverAckNeverReleasesOrNamesTheCommandChannel`, +`TestSupersededCommandClientCannotOccupyTheRunSlot`, and +`TestCommandClientRegistrationAdmitsOnlyAnUnretiredClient` tests in +`internal/hostagent/command_client_test.go` pin the agent side, and +`TestHostAgentCommandChannelBindsTheIdentityPulseAcknowledges` in +`internal/api/host_agent_removal_lifecycle_integration_test.go` replays a +configured-ID install through the router: report, registration, restart, and +the gate. + ### Alert push routing does not create an agent lifecycle channel The shared API router may classify an external-probe outage for specialized diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index a331164414..905bf26451 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -10607,6 +10607,19 @@ router echoes it end-to-end on enrollment acks (`TestHostAgentRemovalLifecycleThroughAuthenticatedRouterAndRestart`). Agents use it to trigger an immediate self-update check after a server upgrade. The field is additive: agents that predate it ignore it. +### Unified-agent report ack names the command identity + +The `POST /api/agents/agent/report` acknowledgement may also carry +`commandAgentId`: the identity the agent's command channel must register under. +When present it equals `agentId`, and it is present only when the reporting +token's command binding holds that ID or would take it on registration, under +the rule in the security-privacy contract. It is omitted otherwise, including +for tokens without `agent:exec`, tokens command admission refuses outright, and +mock mode's unresolved acknowledgements, and agents read its absence as "keep +the command identity you have". The field is additive: agents that predate it +ignore it and keep registering under their own ID. +`TestContract_UnifiedAgentReportAckNamesCommandIdentityOnlyWhenTheBindingAdmitsIt` +in `internal/api/contract_test.go` pins the field. ### Security status serves an infrastructureRead settings capability `GET /api/security/status` adds `settingsCapabilities.infrastructureRead` to the diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index e9c82794b3..a2dc2b0c6e 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -1509,6 +1509,33 @@ admission holds, and a failed save restores the previous metadata. `internal/api/agentbinding/policy_test.go`, and the `TestReportedIdentity*` tests and `TestReportHandlerOffersOnlyQualifyingLiveReportsToTheRecorder` in `internal/api/security_regression_test.go`, pin that boundary. +Command-channel identity converges by naming, never by moving a binding. +`commandIdentityForToken` in `internal/api/agent_ingest.go` returns the +server-resolved host ID, which `/api/agents/agent/report` acknowledges as +`commandAgentId` for a host the monitor resolved and holds under the reporting +token, judged for the hostname the report presents. It names that ID only for +a credential command admission would take on its legacy path (`agent:exec`, at +most one organization, no runner, collector, or unsupported runtime role), and +only when the shared `evaluateAgentExecBinding` decision admits it: the token +already holds it, or a registration presenting it would take the first bind, +backfill, legacy migration, hostname rebind, or one-time deploy repair it would +get anyway. The name grants nothing that registration would not, and it +discloses no ID the acknowledgement's `agentId` does not already carry. A token +whose binding would refuse the resolved ID gets no name and keeps its binding. +Identity resolution can map a report onto another host's continuity record by +machine ID without a token check, so a stale-ID report and a forged one are +indistinguishable, and an established binding therefore moves only through the +moves registration could already make (legacy migration, hostname rebind, and +the one-time deploy placeholder repair, each still decided at admission), never +from report contents alone. Mock mode's +unresolved acknowledgements name nothing. The agent config gate consumes the +same predicate, so it never enables commands for an identity the channel would +refuse. `TestReportAckNeverMovesAnEstablishedCommandBinding`, +`TestReportAckNamesTheResolvedIdentityForAHostnameBoundDeployToken`, +`TestReportAckLandsTheDeployPlaceholderRepairOnTheResolvedIdentity`, +`TestReportAckNamesTheRecordedIdentityToALaterReconfiguredAgent`, and +`TestReportAckNamesNoCommandIdentityInMockMode` in +`internal/api/security_regression_test.go` pin that boundary. Pulse-minted install tokens now carry a server-authored command-policy intent in addition to first-use binding metadata. That intent grants no independent authority: first-report convergence requires the same shared binding decision diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 1f7866d37c..42eb4c900a 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -6497,6 +6497,18 @@ half-bound token. This adds no backup, restore, retention, snapshot, or recovery authority. `TestReportedIdentityRestoresTheTokenWhenTheSaveFails` in `internal/api/security_regression_test.go` pins the restore and the retry. +### Naming a command identity persists nothing + +Naming `commandAgentId` on a report acknowledgement reads the request's token +snapshot and writes nothing: the command binding is still persisted only by +command admission and by the report-side identity recording, each in its one +existing save. The agent keeps the named identity in memory for the process and +asks again after every restart. This adds no backup, restore, retention, +snapshot, or recovery authority. +`TestHostAgentCommandChannelBindsTheIdentityPulseAcknowledges` in +`internal/api/host_agent_removal_lifecycle_integration_test.go` pins the +binding the command registration persists. + ### Destination severity policy preserves notification recovery identity Email, webhook, Apprise, and Relay minimum-severity floors persist through diff --git a/internal/api/agent_ingest.go b/internal/api/agent_ingest.go index b3b13a0d6b..9195a50cd2 100644 --- a/internal/api/agent_ingest.go +++ b/internal/api/agent_ingest.go @@ -159,6 +159,17 @@ func (h *UnifiedAgentHandlers) HandleReport(w http.ResponseWriter, r *http.Reque if h.serverVersion != "" { resp["serverVersion"] = h.serverVersion } + // Name the identity this agent's command channel must register under, for + // a host the monitor resolved and holds under this token, judged for the + // hostname this report presents, as its registration will. Mock mode's + // discard acknowledgement carries no token ID, so it names nothing. + if tokenRecord != nil && host.TokenID == tokenRecord.ID { + commandHost := host + commandHost.Hostname = reportedHostname + if commandAgentID := commandIdentityForToken(tokenRecord, commandHost); commandAgentID != "" { + resp["commandAgentId"] = commandAgentID + } + } // Only include config if there are actual overrides if serverConfig.CommandsEnabled != nil { @@ -688,7 +699,7 @@ func commandConfigAllowedForToken(record *config.APITokenRecord, host models.Hos // commands are enabled when its channel registration would be rejected // strands the host on "Remote control blocked" (the agent reports // CommandsEnabled=true forever while no channel can be admitted). - if !evaluateAgentExecBinding(record, host.ID, host.Hostname).admit { + if commandIdentityForToken(record, host) == "" { // This gate runs before the agent ever attempts command-channel // registration, so without a log here a refused binding leaves no // trace anywhere: the agent never learns commands were requested and @@ -707,6 +718,42 @@ func commandConfigAllowedForToken(record *config.APITokenRecord, host models.Hos return true } +// commandIdentityForToken names the identity a host agent's command channel +// must register under for its token to admit it: the host ID Pulse resolved +// for the agent's report, which the ack returns as agentId and the agent +// persists and presents after a restart. A token binds the first ID its +// channel presents, so an agent that registered a configured --agent-id, or +// an ID Pulse forked or kept from an earlier enrollment, was refused on every +// restart while the config gate judged the resolved ID. +// +// The ID is named only for a credential that command admission would take on +// its legacy path (agent:exec, at most one organization, no runner, +// collector, or unsupported runtime role), and only when the shared binding +// decision admits it: the token already holds it, or a registration +// presenting it would take the first bind, backfill, legacy migration, +// hostname rebind, or one-time deploy repair it would get anyway. Naming it +// grants nothing beyond that registration. An established binding to any +// other ID is never moved by report contents, which can be steered onto +// another host's continuity record; the result is empty and the agent keeps +// registering under the ID it presents. The config gate consumes the same +// decision, so commands are never reported enabled for an identity the +// channel would refuse. +func commandIdentityForToken(record *config.APITokenRecord, host models.Host) string { + hostID := strings.TrimSpace(host.ID) + if record == nil || hostID == "" || !record.HasScope(config.ScopeAgentExec) || len(record.GetBoundOrgs()) > 1 { + return "" + } + switch strings.TrimSpace(record.Metadata[agenttokens.RuntimeRoleMetadataKey]) { + case "", agenttokens.CredentialKindLegacyFullTrust: + default: + return "" + } + if !evaluateAgentExecBinding(record, hostID, host.Hostname).admit { + return "" + } + return hostID +} + func (h *UnifiedAgentHandlers) ensureAgentTokenMatch(w http.ResponseWriter, r *http.Request, agentID string) bool { record := getAPITokenRecordFromRequest(r) if record == nil { diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 1f3bd9db9a..32bfb19a12 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -24560,6 +24560,91 @@ func TestUnifiedAgentHandlers_HandleReportAckOmitsEmptyServerVersion(t *testing. } } +// The report acknowledgement names the command identity only as the identity +// the token's binding admits: equal to agentId when present, and absent for a +// token bound elsewhere, without agent:exec, or one command admission refuses +// outright (a collector role, more than one organization), which agents read +// as "keep the identity you have". +func TestContract_UnifiedAgentReportAckNamesCommandIdentityOnlyWhenTheBindingAdmitsIt(t *testing.T) { + cases := []struct { + name string + scopes []string + metadata map[string]string + orgs []string + wantName bool + }{ + { + name: "unbound install token", + scopes: []string{config.ScopeAgentReport, config.ScopeAgentExec}, + metadata: map[string]string{"issued_via": agentInstallIssuedViaConfig, "install_type": "host"}, + wantName: true, + }, + { + name: "bound to another identity", + scopes: []string{config.ScopeAgentReport, config.ScopeAgentExec}, + metadata: map[string]string{ + "issued_via": agentInstallIssuedViaConfig, "install_type": "host", + "bound_agent_id": "configured-id", "bound_hostname": "contract-host", + agentExecBindingVersionKey: agentExecBindingVersion, + }, + }, + { + name: "no agent:exec", + scopes: []string{config.ScopeAgentReport}, + metadata: map[string]string{"issued_via": agentInstallIssuedViaConfig, "install_type": "host"}, + }, + { + name: "monitoring collector", + scopes: []string{config.ScopeAgentReport, config.ScopeAgentExec}, + metadata: map[string]string{ + "issued_via": agentInstallIssuedViaConfig, "install_type": "host", + agenttokens.RuntimeRoleMetadataKey: agenttokens.CredentialKindMonitoringCollector, + }, + }, + { + name: "bound to two organizations", + scopes: []string{config.ScopeAgentReport, config.ScopeAgentExec}, + metadata: map[string]string{"issued_via": agentInstallIssuedViaConfig, "install_type": "host"}, + orgs: []string{"default", "tenant-b"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + record := newTokenRecord(t, "contract-command-identity-token.12345678", tc.scopes, tc.metadata) + record.OrgIDs = tc.orgs + cfg := newTestConfigWithTokens(t, record) + handler, _ := newUnifiedAgentHandlers(t, cfg) + + body, err := json.Marshal(agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "configured-id", Version: "6.5.0"}, + Host: agentshost.HostInfo{ID: "machine-contract", MachineID: "machine-contract", Hostname: "contract-host", Platform: "linux"}, + Timestamp: time.Now().UTC(), + }) + if err != nil { + t.Fatalf("marshal report: %v", err) + } + req := httptest.NewRequest(http.MethodPost, "/api/agents/agent/report", bytes.NewReader(body)) + attachAPITokenRecord(req, &cfg.APITokens[0]) + rec := httptest.NewRecorder() + handler.HandleReport(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d: %s", rec.Code, rec.Body.String()) + } + var ack map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &ack); err != nil { + t.Fatalf("decode ack: %v", err) + } + named, present := ack["commandAgentId"] + if present != tc.wantName { + t.Fatalf("commandAgentId present = %v (%v), want %v", present, named, tc.wantName) + } + if present && named != ack["agentId"] { + t.Fatalf("commandAgentId = %v, want it equal to agentId %v", named, ack["agentId"]) + } + }) + } +} + // App bootstrap requires presentation defaults for every authenticated role. // Pin both sides of the privilege boundary: the narrow runtime projection is // readable by a viewer, while the complete system settings payload remains diff --git a/internal/api/host_agent_removal_lifecycle_integration_test.go b/internal/api/host_agent_removal_lifecycle_integration_test.go index c32a44cd9b..4164ecfc0c 100644 --- a/internal/api/host_agent_removal_lifecycle_integration_test.go +++ b/internal/api/host_agent_removal_lifecycle_integration_test.go @@ -594,6 +594,93 @@ func TestHostAgentFreshInstallTokenReplacesStaleDisabledCommandPolicyOnce(t *tes } } +// An agent started with a configured --agent-id presents that ID while Pulse +// resolves its reports to the machine's host ID, acknowledges that ID, and the +// agent persists it. Its command channel used to bind the install token to the +// configured ID on first registration, so the config gate (which judges the +// resolved ID) turned the operator's commands-enabled install off on the next +// report, and a restart under the persisted ID was refused for good. The +// acknowledgement now names the resolved ID as the command identity, the +// agent's channel binds it, and every later report, restart and gate decision +// agrees on it. +func TestHostAgentCommandChannelBindsTheIdentityPulseAcknowledges(t *testing.T) { + dataPath := t.TempDir() + const adminRaw = "command-identity-admin-token-123.12345678" + admin := newTokenRecord(t, adminRaw, []string{config.ScopeWildcard}, nil) + if err := config.NewConfigPersistence(dataPath).SaveAPITokens([]config.APITokenRecord{admin}); err != nil { + t.Fatalf("SaveAPITokens: %v", err) + } + runtime := newHostRemovalLifecycleHTTPRuntime(t, dataPath, []config.APITokenRecord{admin}) + t.Cleanup(runtime.stop) + + installRec := serveHostRemovalLifecycleRequest(t, runtime, http.MethodPost, "/api/agent-install-command", adminRaw, + []byte(`{"type":"host","enableCommands":true,"name":"configured-agent-id-install"}`)) + install := decodeHostAgentInstallResponse(t, installRec) + if install.Token == "" || install.Record == nil { + t.Fatalf("commands-enabled install omitted token record: %+v", install) + } + + const ( + presentedID = "configured-agent-id" + machineID = "machine-configured-agent-id" + hostname = "configured-host" + ) + report := agentshost.Report{ + Agent: agentshost.AgentInfo{ID: presentedID, Version: "6.5.0", Type: "unified", CommandsEnabled: true}, + Host: agentshost.HostInfo{ID: machineID, MachineID: machineID, Hostname: hostname, Platform: "linux"}, + Timestamp: time.Now().UTC(), + } + ack := func() map[string]any { + t.Helper() + body, err := json.Marshal(report) + if err != nil { + t.Fatalf("marshal report: %v", err) + } + rec := serveHostRemovalLifecycleRequest(t, runtime, http.MethodPost, "/api/agents/agent/report", install.Token, body) + if rec.Code != http.StatusOK { + t.Fatalf("report status = %d: %s", rec.Code, rec.Body.String()) + } + var decoded map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &decoded); err != nil { + t.Fatalf("decode ack: %v", err) + } + report.Timestamp = report.Timestamp.Add(time.Minute) + return decoded + } + + first := ack() + resolved, _ := first["agentId"].(string) + if resolved != machineID { + t.Fatalf("ack agentId = %q, want the resolved host ID %q", resolved, machineID) + } + if got, _ := first["commandAgentId"].(string); got != resolved { + t.Fatalf("ack commandAgentId = %v, want the resolved %q", first["commandAgentId"], resolved) + } + + // The agent registers its command channel under the named identity, then + // restarts under the identity it persisted from the acknowledgement. + if _, ok := runtime.router.admitAgentExecToken(install.Token, resolved, hostname); !ok { + t.Fatal("first command registration under the named identity was refused") + } + if _, ok := runtime.router.admitAgentExecToken(install.Token, resolved, hostname); !ok { + t.Fatal("restart registration under the acknowledged identity was refused") + } + + // The agent still presents its configured ID in reports. The gate keeps + // the operator's commands-enabled install on, and Pulse keeps naming the + // bound identity. + later := ack() + if got, _ := later["commandAgentId"].(string); got != resolved { + t.Fatalf("later ack commandAgentId = %v, want %q", later["commandAgentId"], resolved) + } + if cfg, _ := later["config"].(map[string]any); cfg == nil || cfg["commandsEnabled"] != true { + t.Fatalf("later ack config = %v, want commands kept enabled", later["config"]) + } + if _, ok := runtime.router.admitAgentExecToken(install.Token, presentedID, hostname); ok { + t.Fatal("the install token admitted the presented identity as a second command identity") + } +} + func tokenRecordByID(records []config.APITokenRecord, tokenID string) *config.APITokenRecord { for index := range records { if records[index].ID == tokenID { diff --git a/internal/api/security_regression_test.go b/internal/api/security_regression_test.go index 53794d9fb7..6176eed3b1 100644 --- a/internal/api/security_regression_test.go +++ b/internal/api/security_regression_test.go @@ -6046,3 +6046,164 @@ func TestReportHandlerOffersOnlyQualifyingLiveReportsToTheRecorder(t *testing.T) t.Fatalf("recorder calls in mock mode = %+v, want none beyond the live report", calls[1:]) } } + +// postReportIdentityAck posts an authenticated report through the router and +// returns the whole acknowledgement. +func postReportIdentityAck(t *testing.T, router *Router, token string, report agentshost.Report) map[string]any { + t.Helper() + code, payload := issue1753PostReport(t, router, token, report) + if code != http.StatusOK { + t.Fatalf("report rejected with %d (%v)", code, payload) + } + return payload +} + +// A token already bound to one identity is never moved by report contents. +// Identity resolution can map a report onto another host's continuity record +// by machine ID without a token check, so a report resolving elsewhere names +// no command identity, leaves the binding as it was, and the agent keeps +// registering under the ID it presents. +func TestReportAckNeverMovesAnEstablishedCommandBinding(t *testing.T) { + const raw = "command-identity-established-token.12345678" + record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{ + "issued_via": agentbinding.IssuedViaConfig, + "install_type": "host", + "bound_agent_id": "configured-id", + "bound_hostname": "host-established", + agentExecBindingVersionKey: agentExecBindingVersion, + agenttokens.RuntimeRoleMetadataKey: agenttokens.CredentialKindLegacyFullTrust, + }) + record.OrgID = "default" + cfg := newTestConfigWithTokens(t, record) + router := newReportIdentityRouter(t, cfg) + + ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "host-established")) + if ack["agentId"] != reportIdentityMachineID { + t.Fatalf("precondition: acknowledged agentId = %v, want the machine ID", ack["agentId"]) + } + if named, present := ack["commandAgentId"]; present { + t.Fatalf("ack named command identity %v for a token bound to another identity", named) + } + if got := reportIdentityToken(t, cfg, record.ID).Metadata["bound_agent_id"]; got != "configured-id" { + t.Fatalf("bound_agent_id = %q after the report, want it unchanged", got) + } + if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "host-established"); ok { + t.Fatal("the established binding moved to the resolved identity") + } + if _, ok := router.admitAgentExecToken(raw, "configured-id", "host-established"); !ok { + t.Fatal("the agent lost its established command identity") + } +} + +// Deploy enrollment binds the runtime token to the node's hostname only. When +// the agent presents another ID than Pulse resolves (a configured --agent-id), +// the report path records nothing, and the acknowledgement names the resolved +// ID so the agent's command channel backfills that one instead. +func TestReportAckNamesTheResolvedIdentityForAHostnameBoundDeployToken(t *testing.T) { + const raw = "command-identity-deploy-token.12345678" + record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{ + "bound_hostname": "pve-node5", + "deploy_job_id": "dep_5", + agentExecBindingVersionKey: agentExecBindingVersion, + agentbinding.DeployIdentityKey: agentbinding.DeployIdentityAgent, + }) + record.OrgID = "default" + cfg := newTestConfigWithTokens(t, record) + router := newReportIdentityRouter(t, cfg) + + ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "pve-node5")) + if ack["commandAgentId"] != reportIdentityMachineID || ack["agentId"] != reportIdentityMachineID { + t.Fatalf("ack agentId=%v commandAgentId=%v, want both the resolved %q", ack["agentId"], ack["commandAgentId"], reportIdentityMachineID) + } + if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "pve-node5"); !ok { + t.Fatal("the command channel could not backfill the named identity") + } + if _, ok := router.admitAgentExecToken(raw, "configured-id", "pve-node5"); ok { + t.Fatal("the backfilled token admitted the presented identity too") + } + bound := reportIdentityToken(t, cfg, record.ID) + if !commandConfigAllowedForToken(&bound, models.Host{ID: reportIdentityMachineID, Hostname: "pve-node5"}) { + t.Fatal("the config gate refused the identity the command channel admits") + } +} + +// A historical deploy token still bound to the invented agent- may +// move once, on exactly its hostname, to the identity that registers. Naming +// the resolved ID lands that one move on the identity the agent's reports and +// restarts use, not on whatever ID the agent happened to present. +func TestReportAckLandsTheDeployPlaceholderRepairOnTheResolvedIdentity(t *testing.T) { + const raw = "command-identity-placeholder-token.12345678" + record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{ + "bound_agent_id": "agent-pve-node6", + "bound_hostname": "pve-node6", + "deploy_job_id": "dep_6", + agentExecBindingVersionKey: agentExecBindingVersion, + }) + record.OrgID = "default" + cfg := newTestConfigWithTokens(t, record) + router := newReportIdentityRouter(t, cfg) + + ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "pve-node6")) + if ack["commandAgentId"] != reportIdentityMachineID { + t.Fatalf("ack commandAgentId = %v, want the resolved %q", ack["commandAgentId"], reportIdentityMachineID) + } + if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "pve-node6"); !ok { + t.Fatal("the placeholder token did not take its one-time repair to the named identity") + } + if got := reportIdentityToken(t, cfg, record.ID).Metadata; got["bound_agent_id"] != reportIdentityMachineID || + got[agentbinding.DeployIdentityKey] != agentbinding.DeployIdentityRepaired { + t.Fatalf("binding after repair = %v, want %q marked repaired", got, reportIdentityMachineID) + } + if _, ok := router.admitAgentExecToken(raw, "configured-id", "pve-node6"); ok { + t.Fatal("the repaired token moved a second time") + } +} + +// The report path records the first identity an agent reports under, and it +// is final. An agent later restarted under a different --agent-id is refused +// under that ID, so the acknowledgement names the recorded identity and the +// agent's command channel registers under it instead. +func TestReportAckNamesTheRecordedIdentityToALaterReconfiguredAgent(t *testing.T) { + const raw = "command-identity-reconfigured-token.12345678" + record := autoRegisteredInstallToken(t, raw, "pve10", true) + cfg := newTestConfigWithTokens(t, record) + router := newReportIdentityRouter(t, cfg) + + postReportIdentityReport(t, router, raw, reportIdentityReport(reportIdentityMachineID, reportIdentityMachineID, "pve10")) + ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "pve10")) + if ack["commandAgentId"] != reportIdentityMachineID { + t.Fatalf("ack commandAgentId = %v, want the recorded %q", ack["commandAgentId"], reportIdentityMachineID) + } + if _, ok := router.admitAgentExecToken(raw, reportIdentityMachineID, "pve10"); !ok { + t.Fatal("the reconfigured agent's registration under the named identity was refused") + } +} + +// Mock mode discards reports and acknowledges the presented ID unresolved, on +// a host held under no token. Naming that ID would steer a fresh token's first +// command registration onto an identity Pulse never resolved, so a mock ack +// names no command identity. +func TestReportAckNamesNoCommandIdentityInMockMode(t *testing.T) { + const raw = "command-identity-mock-token.12345678" + record := newTokenRecord(t, raw, agenttokens.HostScopes(true), map[string]string{ + "issued_via": agentbinding.IssuedViaConfig, + "install_type": "host", + }) + record.OrgID = "default" + cfg := newTestConfigWithTokens(t, record) + router := newReportIdentityRouter(t, cfg) + + previous := mock.IsMockEnabled() + if err := mock.SetEnabled(true); err != nil { + t.Fatalf("enable mock mode: %v", err) + } + t.Cleanup(func() { _ = mock.SetEnabled(previous) }) + + ack := postReportIdentityAck(t, router, raw, reportIdentityReport("configured-id", reportIdentityMachineID, "host-mock")) + if ack["agentId"] != "configured-id" { + t.Fatalf("precondition: mock ack agentId = %v, want the presented ID echoed", ack["agentId"]) + } + if named, present := ack["commandAgentId"]; present { + t.Fatalf("mock ack named command identity %v", named) + } +} diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index 72d4c31c7d..d58abadfc4 100644 --- a/internal/hostagent/agent.go +++ b/internal/hostagent/agent.go @@ -191,6 +191,16 @@ type Agent struct { commandClientMu sync.Mutex commandClientRunCancel context.CancelFunc commandClientParentCtx context.Context + // commandClientRunCtx holds Run's context until the first acknowledged + // report releases it as commandClientParentCtx. The command channel must + // not register before Pulse has said which identity its token binding + // will hold: a first registration binds whatever ID it presents, and an + // agent that later restarts under the acknowledged ID is refused for good. + commandClientRunCtx context.Context + // commandIdentity is the command-channel identity Pulse acknowledged + // (commandAgentId on a report ack). Empty means Pulse named none, and the + // channel registers under agentID as it always has. + commandIdentity string // commandAuthorityProfile is the immutable local install-time ceiling for // this process. Remote configuration may disable and later re-enable a // command-capable or legacy install, but it cannot promote monitoring-only. @@ -626,12 +636,15 @@ func (a *Agent) Run(ctx context.Context) error { return a.runOnce(ctx) } + // The command channel waits for the first acknowledged report, which + // releases this context to it (adoptCommandIdentity). a.commandClientMu.Lock() - a.commandClientParentCtx = ctx + a.commandClientRunCtx = ctx commandClient := a.commandClient a.commandClientMu.Unlock() defer func() { a.commandClientMu.Lock() + a.commandClientRunCtx = nil a.commandClientParentCtx = nil a.commandClientMu.Unlock() a.stopCommandClient(true) @@ -647,7 +660,7 @@ func (a *Agent) Run(ctx context.Context) error { // original was built with the bootstrap token during New(). if a.cfg.EnableCommands { a.commandClientMu.Lock() - a.commandClient = a.newCommandClient(a.cfg, a.agentID, a.hostname, a.platform, a.agentVersion) + a.commandClient = a.newCommandClient(a.cfg, a.commandRegistrationIdentityLocked(), a.hostname, a.platform, a.agentVersion) commandClient = a.commandClient a.commandClientMu.Unlock() } @@ -659,7 +672,8 @@ func (a *Agent) Run(ctx context.Context) error { go a.runProxmoxHealthCheckLoop(ctx) } - // Start command client in background for AI command execution + // Stage the command client for AI command execution. It starts once the + // first report is acknowledged. if commandClient != nil { a.startCommandClient(commandClient) } @@ -813,6 +827,13 @@ func (a *Agent) startCommandClient(client *CommandClient) bool { a.commandClientMu.Unlock() return true } + // A caller may hold a client that a concurrent disable already closed and + // cleared. Running it would occupy the run slot with a dead client, and + // the next enabled client would never start. + if a.commandClient != client { + a.commandClientMu.Unlock() + return false + } parentCtx := a.commandClientParentCtx if parentCtx == nil { @@ -831,6 +852,84 @@ func (a *Agent) startCommandClient(client *CommandClient) bool { return true } +// commandRegistrationIdentityLocked returns the agent ID the command channel +// registers under: the identity Pulse acknowledged for it, or this agent's own +// ID when Pulse named none. Callers hold commandClientMu. +func (a *Agent) commandRegistrationIdentityLocked() string { + if a.commandIdentity != "" { + return a.commandIdentity + } + return a.agentID +} + +// adoptCommandIdentity runs after every acknowledged report. Pulse resolves +// this agent to its own host ID, which the agent persists and presents after a +// restart, and names that ID as commandAgentId whenever the token's binding +// holds it or would take it on first registration. The command channel must +// register under it, because a token binds the first ID its channel presents +// and never moves afterwards: registering under a configured --agent-id, or an +// ID the server forked or kept from an earlier enrollment, strands the channel +// on the next restart. +// +// An ack without commandAgentId keeps the identity already in use (an older +// Pulse, or a token already bound to the ID this agent presents). The first +// ack also releases Run's context, so the channel never registers before Pulse +// has answered. A client under another ID is replaced while Pulse has never +// admitted it (staged, or still retrying a refused registration). An admitted +// client is never replaced: its token accepted it, and building a second +// client would reopen the operation receipt store, which marks in-flight +// operations interrupted. A new identity then applies to the next client this +// process builds. +func (a *Agent) adoptCommandIdentity(commandAgentID string) { + commandAgentID = strings.TrimSpace(commandAgentID) + + a.configMu.RLock() + commandCfg := a.cfg + a.configMu.RUnlock() + + a.commandClientMu.Lock() + if commandAgentID != "" && commandAgentID != a.commandIdentity { + a.commandIdentity = commandAgentID + if commandAgentID != a.agentID { + a.logger.Info(). + Str("agent_id", a.agentID). + Str("command_agent_id", commandAgentID). + Msg("Registering the command channel under the identity Pulse acknowledged for this agent") + } + } + if a.commandClientParentCtx == nil && a.commandClientRunCtx != nil { + a.commandClientParentCtx = a.commandClientRunCtx + } + identity := a.commandRegistrationIdentityLocked() + client := a.commandClient + if client != nil && client.agentID != identity && + (a.commandClientRunCancel == nil || client.retireIfNeverAdmitted()) { + if a.commandClientRunCancel != nil { + a.commandClientRunCancel() + a.commandClientRunCancel = nil + } + if err := client.Close(); err != nil { + a.logger.Debug().Err(err).Msg("Error closing command client staged under a superseded identity") + } + client = a.newCommandClient(commandCfg, identity, a.hostname, a.platform, a.agentVersion) + a.commandClient = client + } + a.commandClientMu.Unlock() + + if client != nil { + a.startCommandClient(client) + } +} + +// releaseCommandChannelWithoutReports lets the command channel register under +// the identity it already has when Pulse refuses this token's reports outright +// (403: the token lacks report scope). Pulse can then never name a command +// identity, and a token that may still run commands keeps the channel it had +// before registration waited for an acknowledgement. +func (a *Agent) releaseCommandChannelWithoutReports() { + a.adoptCommandIdentity("") +} + func (a *Agent) stopCommandClient(clearClient bool) { a.commandClientMu.Lock() client := a.commandClient @@ -889,6 +988,7 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo Str("endpoint", statusErr.Endpoint). Int("status_code", statusErr.StatusCode). Msg("Failed to send Unified Agent report (403 Forbidden). API token may lack 'Unified Agent reporting' scope. Set PULSE_ENABLE_HOST=false if host monitoring is not needed.") + a.releaseCommandChannelWithoutReports() return nil } if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusUnauthorized { @@ -1057,6 +1157,9 @@ func (a *Agent) flushBuffer(ctx context.Context) { // accepted with it. Drop them so the buffer cannot grow // unbounded across restarts, and surface the actionable error. a.logAuthFailure(statusErr) + if statusErr.StatusCode == http.StatusForbidden { + a.releaseCommandChannelWithoutReports() + } for { if _, ok := a.reportBuffer.Pop(); !ok { break @@ -1440,15 +1543,18 @@ func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.R // Parse response to check for server-side config overrides var reportResp struct { - Success bool `json:"success"` - AgentID string `json:"agentId"` - ServerVersion string `json:"serverVersion"` - Config *struct { + Success bool `json:"success"` + AgentID string `json:"agentId"` + CommandAgentID string `json:"commandAgentId"` + ServerVersion string `json:"serverVersion"` + Config *struct { CommandsEnabled *bool `json:"commandsEnabled"` } `json:"config,omitempty"` } if err := json.NewDecoder(resp.Body).Decode(&reportResp); err != nil { - // Non-fatal: just log and continue + // Non-fatal: just log and continue. An unreadable ack is no answer, so + // the command channel keeps waiting: registering now could bind a + // fresh token to an ID the next readable ack would not name. a.logger.Debug().Err(err).Msg("Failed to parse report response, ignoring config") return nil } @@ -1473,6 +1579,17 @@ func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.R a.applyRemoteConfig(*reportResp.Config.CommandsEnabled) } + // Only a real acknowledgement answers for this process's identity: one + // that accepted the report and named its host, for a report that presents + // the agent ID and hostname this process registers with. A replayed report + // buffered by an earlier run may carry another of either, and Pulse judged + // its command identity for that one. + if reportResp.Success && canonicalAgentID != "" && + strings.TrimSpace(report.Agent.ID) == a.agentID && + strings.TrimSpace(report.Host.Hostname) == a.hostname { + a.adoptCommandIdentity(reportResp.CommandAgentID) + } + return nil } @@ -1636,7 +1753,7 @@ func (a *Agent) applyRemoteConfig(commandsEnabled bool) (bool, bool) { currentlyEnabled := a.commandClient != nil switch { case effectiveState && !currentlyEnabled: - clientToStart = a.newCommandClient(commandCfg, a.agentID, a.hostname, a.platform, a.agentVersion) + clientToStart = a.newCommandClient(commandCfg, a.commandRegistrationIdentityLocked(), a.hostname, a.platform, a.agentVersion) a.commandClient = clientToStart case !effectiveState && currentlyEnabled: shouldStop = true diff --git a/internal/hostagent/command_client_test.go b/internal/hostagent/command_client_test.go index 8b3c0952a9..e7934b327c 100644 --- a/internal/hostagent/command_client_test.go +++ b/internal/hostagent/command_client_test.go @@ -2,14 +2,21 @@ package hostagent import ( "context" + "encoding/json" + "errors" "fmt" "io" + "net/http" + "net/http/httptest" "reflect" + "strings" + "sync" "testing" "time" "github.com/gorilla/websocket" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" "github.com/rs/zerolog" ) @@ -531,3 +538,527 @@ func TestCommandClientGuestOutcomeContract(t *testing.T) { t.Run("execution-and-admission", testGuestExecProjectsGuestOutcome) t.Run("terminal-dictionary", TestGuestExecTerminalDictionary) } + +// commandIdentityHarness runs an agent with commands enabled against a fake +// Pulse whose report acknowledgement the test scripts, and records the +// identity of every command client that starts. +type commandIdentityHarness struct { + agent *Agent + server *httptest.Server + started chan string + + mu sync.Mutex + reports int + built int + // refuse makes every started client behave as if Pulse refused its + // registration; otherwise a started client counts as admitted. + refuse bool +} + +func newCommandIdentityHarness(t *testing.T, presentedID string, interval time.Duration, respond func(report int, w http.ResponseWriter)) *commandIdentityHarness { + t.Helper() + h := &commandIdentityHarness{started: make(chan string, 8)} + h.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + h.mu.Lock() + h.reports++ + report := h.reports + h.mu.Unlock() + respond(report, w) + })) + t.Cleanup(h.server.Close) + + logger := zerolog.Nop() + agent, err := New(Config{ + APIToken: "token", + PulseURL: "https://pulse", + AgentID: presentedID, + EnableCommands: true, + Interval: interval, + StateDir: t.TempDir(), + Collector: &mockCollector{}, + Logger: &logger, + newCommandClientFn: func(cfg Config, agentID, hostname, platform, version string) *CommandClient { + h.mu.Lock() + h.built++ + h.mu.Unlock() + return NewCommandClient(cfg, agentID, hostname, platform, version) + }, + runCommandClientFn: func(client *CommandClient, ctx context.Context) error { + h.mu.Lock() + refuse := h.refuse + h.mu.Unlock() + if !refuse { + client.markAdmitted() + } + h.started <- client.agentID + <-ctx.Done() + return ctx.Err() + }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + agent.httpClient = h.server.Client() + agent.trimmedPulseURL = h.server.URL + h.agent = agent + return h +} + +func (h *commandIdentityHarness) run(t *testing.T) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + _ = h.agent.Run(ctx) + }() + t.Cleanup(func() { + cancel() + <-done + }) +} + +func (h *commandIdentityHarness) waitForReports(t *testing.T, want int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + h.mu.Lock() + reports := h.reports + h.mu.Unlock() + if reports >= want { + return + } + if time.Now().After(deadline) { + t.Fatalf("only %d reports were sent, want %d", reports, want) + } + time.Sleep(20 * time.Millisecond) + } +} + +func (h *commandIdentityHarness) clientIdentity() string { + h.agent.commandClientMu.Lock() + defer h.agent.commandClientMu.Unlock() + if h.agent.commandClient == nil { + return "" + } + return h.agent.commandClient.agentID +} + +func (h *commandIdentityHarness) expectStart(t *testing.T, want string) { + t.Helper() + select { + case got := <-h.started: + if got != want { + t.Fatalf("command channel registered as %q, want %q", got, want) + } + case <-time.After(5 * time.Second): + t.Fatalf("command channel never started; want it registered as %q", want) + } +} + +func (h *commandIdentityHarness) expectNoStart(t *testing.T, within time.Duration) { + t.Helper() + select { + case got := <-h.started: + t.Fatalf("command channel registered as %q before Pulse acknowledged a report", got) + case <-time.After(within): + } +} + +func (h *commandIdentityHarness) expectNoRestart(t *testing.T) { + t.Helper() + select { + case got := <-h.started: + t.Fatalf("command channel registered again, as %q", got) + default: + } +} + +func writeAck(w http.ResponseWriter, body string) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) +} + +// A token binds the first identity its command channel presents and never +// moves. An agent started with a configured ID (or an ID Pulse forked, or +// kept from an earlier enrollment) used to register that ID before its first +// report, while Pulse resolved its reports to another host ID that the agent +// persisted and presented after the next restart: the channel was refused for +// good. The channel now waits for the acknowledgement and registers under the +// identity Pulse names for it. +func TestCommandChannelRegistersUnderTheIdentityPulseAcknowledges(t *testing.T) { + release := make(chan struct{}) + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(report int, w http.ResponseWriter) { + if report == 1 { + <-release + } + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + }) + h.run(t) + + h.expectNoStart(t, 300*time.Millisecond) + close(release) + h.expectStart(t, "server-resolved") +} + +// An older Pulse names no command identity, and a current Pulse names none for +// a token already bound to the ID this agent presents (for example a +// configured --agent-id). The channel then registers under the agent's own ID +// exactly as before, once a report is acknowledged. +func TestCommandChannelKeepsItsOwnIdentityWhenPulseNamesNone(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) { + writeAck(w, `{"success":true,"agentId":"server-resolved"}`) + }) + h.run(t) + + h.expectStart(t, "configured-id") +} + +// While Pulse is unreachable or refusing reports the agent has no answer, so +// the channel stays staged and registers once a later report is acknowledged. +func TestCommandChannelWaitsForAnAcknowledgedReport(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) { + if report < 3 { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + return + } + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + }) + h.run(t) + + h.expectStart(t, "server-resolved") + h.mu.Lock() + reports := h.reports + h.mu.Unlock() + if reports < 3 { + t.Fatalf("command channel started after %d report attempts, want it to wait for the accepted third", reports) + } +} + +// Once Pulse has named the channel's identity, a later acknowledgement that +// names none (Pulse now resolves the host elsewhere, and the binding refuses +// that ID) must not drop the channel back to a different ID: the token is +// bound to the acknowledged one. +func TestCommandChannelKeepsTheAcknowledgedIdentityWhenALaterAckNamesNone(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) { + if report == 1 { + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + return + } + writeAck(w, `{"success":true,"agentId":"healed-id"}`) + }) + h.run(t) + + h.expectStart(t, "server-resolved") + h.waitForReports(t, 4) + h.expectNoRestart(t) + if got := h.clientIdentity(); got != "server-resolved" { + t.Fatalf("command client identity = %q, want the acknowledged %q", got, "server-resolved") + } + + // A client rebuilt later in this process (commands disabled, then enabled + // again) still registers under the acknowledged identity. + h.agent.applyRemoteConfig(false) + h.agent.applyRemoteConfig(true) + h.expectStart(t, "server-resolved") +} + +// Commands enabled by Pulse after start-up (rather than by --enable-commands) +// stage a client before any report is acknowledged. That client must register +// under the acknowledged identity, not the presented one. +func TestRemotelyEnabledCommandChannelRegistersUnderTheAcknowledgedIdentity(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) { + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved","config":{"commandsEnabled":true}}`) + }) + h.agent.stopCommandClient(true) + h.agent.configMu.Lock() + h.agent.cfg.EnableCommands = false + h.agent.configMu.Unlock() + h.run(t) + + h.expectStart(t, "server-resolved") +} + +// An unreadable acknowledgement is no answer. Registering on it could bind a +// fresh token to the presented ID, which the next readable ack would not name, +// so the channel waits for an ack it can read. +func TestCommandChannelIgnoresAnUnreadableAck(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) { + if report == 1 { + w.Header().Set("Content-Type", "text/html") + _, _ = w.Write([]byte("proxy")) + return + } + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + }) + h.run(t) + + h.expectStart(t, "server-resolved") +} + +// A token Pulse refuses reports for (403: no report scope) can never be named +// a command identity, and it may still run commands. Its channel registers +// under the agent's own ID as it did before registration waited for an ack. +func TestCommandChannelRegistersUnderItsOwnIdentityWhenReportsAreForbidden(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) { + http.Error(w, "missing scope", http.StatusForbidden) + }) + h.run(t) + + h.expectStart(t, "configured-id") +} + +// A running channel was admitted by its token. A later ack naming another +// identity must not tear it down and build a second client: that reopens the +// operation receipt store and marks in-flight operations interrupted. +func TestRunningCommandChannelIsNeverReplacedByALaterIdentity(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) { + if report == 1 { + writeAck(w, `{"success":true,"agentId":"server-resolved"}`) + return + } + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + }) + h.run(t) + + h.expectStart(t, "configured-id") + h.waitForReports(t, 3) + h.expectNoRestart(t) + if got := h.clientIdentity(); got != "configured-id" { + t.Fatalf("running command client identity = %q, want it left on %q", got, "configured-id") + } + h.mu.Lock() + built := h.built + h.mu.Unlock() + if built != 1 { + t.Fatalf("command clients built = %d, want only the one New staged", built) + } +} + +// Observer destinations may run any Pulse version and never speak for this +// agent's identity: their acknowledgements neither name the command identity +// nor release the channel. +func TestObserverAckNeverReleasesOrNamesTheCommandChannel(t *testing.T) { + server := newServerVersionAckServer(`{"success":true,"agentId":"observer-id","commandAgentId":"observer-id"}`) + defer server.Close() + + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) { + writeAck(w, `{"success":true,"agentId":"server-resolved"}`) + }) + runCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + h.agent.commandClientMu.Lock() + h.agent.commandClientRunCtx = runCtx + h.agent.commandClientMu.Unlock() + + report := agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "configured-id"}, + Host: agentshost.HostInfo{Hostname: "test-host"}, + } + if err := h.agent.sendReportToDestination(context.Background(), report, server.URL, "observer-token", server.Client(), false); err != nil { + t.Fatalf("sendReportToDestination: %v", err) + } + h.agent.commandClientMu.Lock() + released := h.agent.commandClientParentCtx != nil + identity := h.agent.commandIdentity + h.agent.commandClientMu.Unlock() + if released || identity != "" { + t.Fatalf("observer ack released=%v identity=%q, want neither", released, identity) + } + h.expectNoStart(t, 100*time.Millisecond) +} + +// A remote disable can close and clear a client another path is about to +// start. Starting that superseded client must fail without occupying the run +// slot, or the next enabled client never runs. +func TestSupersededCommandClientCannotOccupyTheRunSlot(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) { + writeAck(w, `{"success":true,"agentId":"server-resolved"}`) + }) + runCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + h.agent.commandClientMu.Lock() + h.agent.commandClientParentCtx = runCtx + superseded := h.agent.commandClient + h.agent.commandClientMu.Unlock() + + h.agent.applyRemoteConfig(false) + if h.agent.startCommandClient(superseded) { + t.Fatal("a cleared command client was started") + } + h.agent.commandClientMu.Lock() + occupied := h.agent.commandClientRunCancel != nil + h.agent.commandClientMu.Unlock() + if occupied { + t.Fatal("a cleared command client occupied the run slot") + } + h.expectNoStart(t, 100*time.Millisecond) + + h.agent.applyRemoteConfig(true) + h.expectStart(t, "configured-id") +} + +// A client whose registration Pulse keeps refusing (for example one released +// by a mock-mode acknowledgement that named nothing, for a token bound to the +// resolved identity) has accepted no operation. When a later acknowledgement +// names the identity the token admits, it is retired and rebuilt under that +// identity instead of retrying the refused one for the life of the process. +func TestRefusedCommandChannelMovesToTheIdentityALaterAckNames(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) { + if report == 1 { + writeAck(w, `{"success":true,"agentId":"configured-id"}`) + return + } + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + }) + h.refuse = true + h.run(t) + + h.expectStart(t, "configured-id") + h.expectStart(t, "server-resolved") + if got := h.clientIdentity(); got != "server-resolved" { + t.Fatalf("command client identity = %q, want %q", got, "server-resolved") + } +} + +// A decodable body that is not a Pulse acknowledgement (an empty object, or a +// refusal) is no answer either: the channel keeps waiting for a real ack. +func TestCommandChannelIgnoresAResponseThatIsNotAnAck(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", 50*time.Millisecond, func(report int, w http.ResponseWriter) { + switch report { + case 1: + writeAck(w, `{}`) + case 2: + writeAck(w, `{"success":false}`) + default: + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + } + }) + h.run(t) + + h.expectStart(t, "server-resolved") +} + +// A report buffered by an earlier run may present another agent ID or +// hostname, and Pulse judges the command identity for the report it +// acknowledges. Its acknowledgement must neither name nor release this +// process's channel. +func TestReplayedReportAckNeverNegotiatesTheCurrentChannel(t *testing.T) { + h := newCommandIdentityHarness(t, "configured-id", time.Hour, func(_ int, w http.ResponseWriter) { + writeAck(w, `{"success":true,"agentId":"server-resolved","commandAgentId":"server-resolved"}`) + }) + runCtx, cancel := context.WithCancel(context.Background()) + defer cancel() + h.agent.commandClientMu.Lock() + h.agent.commandClientRunCtx = runCtx + h.agent.commandClientMu.Unlock() + + for _, replayed := range []agentshost.Report{ + {Agent: agentshost.AgentInfo{ID: "earlier-id"}, Host: agentshost.HostInfo{Hostname: h.agent.hostname}}, + {Agent: agentshost.AgentInfo{ID: "configured-id"}, Host: agentshost.HostInfo{Hostname: "earlier-hostname"}}, + } { + if err := h.agent.sendReport(context.Background(), replayed); err != nil { + t.Fatalf("sendReport: %v", err) + } + } + h.agent.commandClientMu.Lock() + released := h.agent.commandClientParentCtx != nil + identity := h.agent.commandIdentity + h.agent.commandClientMu.Unlock() + if released || identity != "" { + t.Fatalf("replayed acks released=%v identity=%q, want neither", released, identity) + } + h.expectNoStart(t, 100*time.Millisecond) + + current := agentshost.Report{Agent: agentshost.AgentInfo{ID: "configured-id"}, Host: agentshost.HostInfo{Hostname: h.agent.hostname}} + if err := h.agent.sendReport(context.Background(), current); err != nil { + t.Fatalf("sendReport: %v", err) + } + h.expectStart(t, "server-resolved") +} + +// Pulse admits a client's registration before sending it anything. A client +// it admitted can no longer be retired, and one the agent retired can never be +// admitted afterwards. +func TestCommandClientAdmissionAndRetirementAreExclusive(t *testing.T) { + admitted := &CommandClient{} + if !admitted.markAdmitted() || !admitted.markAdmitted() { + t.Fatal("an admitted client did not stay admitted across reconnects") + } + if admitted.retireIfNeverAdmitted() { + t.Fatal("an admitted client was retired") + } + retired := &CommandClient{} + if !retired.retireIfNeverAdmitted() { + t.Fatal("a never-admitted client could not be retired") + } + if retired.markAdmitted() { + t.Fatal("a retired client was admitted") + } +} + +// The admission state is set by the real registration handshake: a client +// Pulse accepts becomes admitted, and a client the agent retired before that +// stops right after registering instead of handling anything. +func TestCommandClientRegistrationAdmitsOnlyAnUnretiredClient(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + var registration wsMessage + if err := conn.ReadJSON(®istration); err != nil { + return + } + payload, _ := json.Marshal(registeredPayload{Success: true, Message: "Registered"}) + if err := conn.WriteJSON(wsMessage{Type: msgTypeRegistered, Timestamp: time.Now(), Payload: payload}); err != nil { + return + } + _, _, _ = conn.ReadMessage() + })) + defer server.Close() + + newClient := func() *CommandClient { + return &CommandClient{ + pulseURL: strings.TrimRight(server.URL, "/"), + apiToken: "token", + agentID: "agent-1", + hostname: "host-1", + platform: "linux", + version: "1.2.3", + logger: zerolog.Nop(), + done: make(chan struct{}), + } + } + + admitted := newClient() + ctx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- admitted.connectAndHandle(ctx) }() + deadline := time.Now().Add(5 * time.Second) + for admitted.admission.Load() != commandClientAdmitted { + if time.Now().After(deadline) { + t.Fatal("an accepted registration did not admit the client") + } + time.Sleep(10 * time.Millisecond) + } + cancel() + <-errCh + + retired := newClient() + if !retired.retireIfNeverAdmitted() { + t.Fatal("precondition: a fresh client could not be retired") + } + done := make(chan error, 1) + go func() { done <- retired.connectAndHandle(context.Background()) }() + select { + case err := <-done: + if !errors.Is(err, errCommandClientSuperseded) { + t.Fatalf("retired client connectAndHandle = %v, want errCommandClientSuperseded", err) + } + case <-time.After(5 * time.Second): + t.Fatal("a retired client kept running after its registration was accepted") + } +} diff --git a/internal/hostagent/commands.go b/internal/hostagent/commands.go index 07cd888e02..1246f7565e 100644 --- a/internal/hostagent/commands.go +++ b/internal/hostagent/commands.go @@ -16,6 +16,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "time" "github.com/gorilla/websocket" @@ -146,6 +147,12 @@ type CommandClient struct { healthCapabilities []string actionActivationNonce string actionHealthWriter func(bool) error + // admission is commandClientPending until Pulse first accepts this + // client's registration (commandClientAdmitted), or until the agent + // retires it unadmitted under a superseded identity + // (commandClientSuperseded). Only an unadmitted client may be retired: + // operations arrive only after registration, so it has accepted none. + admission atomic.Int32 } const maxCancellableRequestsPerConnection = 256 @@ -452,6 +459,9 @@ func (c *CommandClient) connectAndHandle(ctx context.Context) error { if err := c.waitForRegistration(conn); err != nil { return fmt.Errorf("registration failed: %w", err) } + if !c.markAdmitted() { + return errCommandClientSuperseded + } if c.actionRunnerOnly { if err := c.persistActionRunnerHealth(false); err != nil { return fmt.Errorf("write action-runner health: %w", err) @@ -600,6 +610,31 @@ func (c *CommandClient) waitForRegistration(conn *websocket.Conn) error { return nil } +const ( + commandClientPending int32 = iota + commandClientAdmitted + commandClientSuperseded +) + +var errCommandClientSuperseded = errors.New("command client was retired under a superseded identity before Pulse admitted it") + +// markAdmitted records that Pulse accepted this client's registration. It +// reports false for a client the agent already retired, which must then +// handle nothing. +func (c *CommandClient) markAdmitted() bool { + if c.admission.CompareAndSwap(commandClientPending, commandClientAdmitted) { + return true + } + return c.admission.Load() == commandClientAdmitted +} + +// retireIfNeverAdmitted retires a client Pulse has never admitted, so a later +// registration success cannot let it handle anything. It reports false once +// the client has been admitted. +func (c *CommandClient) retireIfNeverAdmitted() bool { + return c.admission.CompareAndSwap(commandClientPending, commandClientSuperseded) +} + func (c *CommandClient) pingLoop(ctx context.Context, conn *websocket.Conn, done chan struct{}) { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop()