From be319e82044556eadb656e232dacf7ee2cf8512b Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 4 Oct 2026 17:59:09 +0100 Subject: [PATCH] Guard VM Discovery and command execution at the owning host agent Require local current-node lock evidence and registered guard support before QGA execution. Serialise VM probes, stop on deferral, retain actual in-flight outcomes and saved Discovery evidence, and pause uncertain completion without replay. Config checks do not reserve QGA or prove native thaw; the existing backup monitoring-outage precaution remains. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 43 ++++ .../v6/internal/subsystems/monitoring.md | 27 ++ internal/agentexec/guest_exec_safety.go | 40 +++ internal/agentexec/guest_exec_safety_test.go | 74 ++++++ internal/agentexec/server.go | 36 ++- internal/agentexec/server_websocket_test.go | 9 +- internal/agentexec/types.go | 1 + internal/hostagent/command_client_test.go | 4 + internal/hostagent/commands.go | 62 ++++- internal/hostagent/guest_exec_config_linux.go | 42 +++ internal/hostagent/guest_exec_config_other.go | 14 + .../guest_exec_discovery_integration_test.go | 130 ++++++++++ internal/hostagent/guest_exec_guard.go | 153 +++++++++++ internal/hostagent/guest_exec_guard_test.go | 206 +++++++++++++++ internal/servicediscovery/deep_scanner.go | 239 +++++++++++------- .../guest_exec_safety_test.go | 102 ++++++++ internal/servicediscovery/service.go | 7 + internal/servicediscovery/service_test.go | 4 + 18 files changed, 1090 insertions(+), 103 deletions(-) create mode 100644 internal/agentexec/guest_exec_safety.go create mode 100644 internal/agentexec/guest_exec_safety_test.go create mode 100644 internal/hostagent/guest_exec_config_linux.go create mode 100644 internal/hostagent/guest_exec_config_other.go create mode 100644 internal/hostagent/guest_exec_discovery_integration_test.go create mode 100644 internal/hostagent/guest_exec_guard.go create mode 100644 internal/hostagent/guest_exec_guard_test.go create mode 100644 internal/servicediscovery/guest_exec_safety_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index d59350e064..7a25ae4733 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -741,6 +741,11 @@ installer download and the agent's subsequent Pulse TLS connection. ## Canonical Files +- `internal/agentexec/guest_exec_safety.go` +- `internal/hostagent/guest_exec_guard.go` +- `internal/hostagent/guest_exec_config_linux.go` +- `internal/hostagent/guest_exec_config_other.go` + 1. `internal/api/agent_install_command_shared.go` 2. `internal/api/configapi/config_setup_handlers.go` 2a. `internal/api/configapi/setup_script_render.go` @@ -3527,6 +3532,44 @@ Agent` secondary handoff against the live setup wizard instead of relying ## Current State +### VM guest execution admission (backup precaution) + +`AgentRegisterPayload.guest_exec_guard_version` is optional; version 1 on a +Linux full-trust command session means the agent implements the local guard, +not that the VM is thawed or safe to overlap a backup. The server binds it to +the authenticated session and refuses VM command/file dispatch to absent, +unknown-version or non-Linux support. Version strings, user approval and the +internal Trusted flag cannot substitute for that capability. Host/LXC commands, +tenant admission, approvals and typed action-runner exclusions are unchanged. + +The host agent validates and canonicalises the numeric VMID, requires its +current local-node config, and reads the fixed trusted `qm config --current` +without QGA before and after one absolute-path `qm guest exec`. Missing, +ambiguous, oversized or locked config fails closed. Per-VM process-wide +admission serialises concurrent commands, respects cancellation and bounds +state to 4,096 entries. Unknown handed-off completion or postflight lock/owner +loss imposes a one-minute uncertainty pause; no command is replayed. The next +explicit request must pass fresh config admission. A local process exit or kill +is not proof that guest work stopped; postflight failure retains the actual +exit/output but disqualifies successful live evidence. + +The guard is not shared with the HTTP poller, separate agent processes or PVE +backup workers, and config checks do not reserve QGA. Agent restart also does +not retain the in-memory uncertainty pause. Continue the stop-before-affected- +backup / independent covered-filesystem-write restart precaution until native +overlap, thaw, writes, liveness and resumption are verified. Do not unlock, +thaw or automatically restart from this guard. Raw host-shell commands and +typed VM lifecycle operations are not relabelled as guarded guest reads. + +`TestCommandClientGuestExecutionAdmissionContract` pins no unverified handoff; +`TestGuestExecRealServerAgentDiscoveryAdmission` connects real registration, +server/agent WebSockets and VM/nested-Docker Discovery with fake config/provider +boundaries. `TestVMGuestExecCapabilityBoundToRegisteredSession` pins old/unknown +agent rejection and retained deferral. Config ambiguity, cancellation, +serialization/capacity and explicit-resumption controls are source proofs, not +native PVE/HAOS-QGA, thaw, installed-agent or release acceptance. + + ### Credential-safe container diagnostics bootstrap (1 October 2026) The existing container-runtime migration command now shares the Proxmox diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 4a5675e753..990c203dbe 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -2252,6 +2252,33 @@ service-history reads plus denial/recovery without fabricated samples. ## Current State +### Discovery observes VM guest execution pauses + +Manual, automatic and nested-Docker VM scans use the existing agent execution +boundary; they cannot infer safety from the Web drawer's snapshot. VM routing +requires an exact connected agent ID or a unique owning-node hostname, never a +sole unrelated agent or guest-name hint. The agent independently requires its +current local VM config and checks the operation lock. Discovery runs VM probes +serially and stops on any explicit guest safety deferral, including optional +probes. Deferred in-flight output is not fresh discovery evidence; progress +retains an error and actual partial completion rather than 100% success. + +A manual deferral is returned as an error without replacing saved discoveries, +notes or raw observations with metadata fallback. Background enhancement keeps +its existing evidence rather than analysing partial deferred output. Clearing +a pause triggers no scan or retry by itself; an existing normal trigger or +explicit request starts a new, guarded attempt. Healthy catalogue commands, +non-VM parallelism, permissions, feature gates and saved-result access remain. + +`TestDiscoveryGuestSafetyStopsAndPreservesSavedEvidence` covers both VM/nested +Docker scans, first/later optional-command deferral, manual error and automatic +saved-evidence preservation. Node ambiguity controls exclude misrouting. These +source-connected controls and the agent protocol guard do not reserve QGA +against PVE backup start, establish native thaw, or satisfy the existing native +overlap/covered-write/liveness/resumption dependency. Preserve the monitoring +and alert outage precaution until that operational outcome is verified. + + ### Saved quiet-hours policy before queue activation `Monitor.New` binds `alerts.Manager.QuietHoursNotificationPolicy` to the diff --git a/internal/agentexec/guest_exec_safety.go b/internal/agentexec/guest_exec_safety.go new file mode 100644 index 0000000000..2d1e9324f0 --- /dev/null +++ b/internal/agentexec/guest_exec_safety.go @@ -0,0 +1,40 @@ +package agentexec + +import ( + "fmt" + "strings" +) + +// Version 1 requires local lock admission, per-VM serialization, and an +// uncertainty cooldown. It does NOT reserve QGA against PVE backup workers. +const GuestExecGuardProtocolVersion = 1 + +const guestExecDeferredPrefix = "VM guest execution deferred: " + +// GuestExecDeferred carries only a fixed safety reason, never guest/config data. +func GuestExecDeferred(reason string) error { + return fmt.Errorf("%s%s", guestExecDeferredPrefix, reason) +} + +func IsGuestExecDeferred(message string) bool { + return strings.HasPrefix(message, guestExecDeferredPrefix) +} + +const ( + GuestExecGuardUnavailable = "update the host agent for backup-safety support" + GuestExecInvalidTarget = "the local VM identity is invalid" + GuestExecLockUnverified = "the local VM backup lock could not be verified" + GuestExecVMLocked = "the VM has an active operation lock" + GuestExecCompletionUnknown = "guest command completion could not be verified" + GuestExecCooldown = "a previous guest command is still uncertain" + GuestExecCapacity = "guest command admission is at capacity" + GuestExecTargetUnverified = "the owning node agent could not be uniquely identified" +) + +func requireVMGuestExecGuard(ac *agentConn, targetType string) error { + if strings.EqualFold(strings.TrimSpace(targetType), "vm") && + (ac == nil || ac.guestExecGuardVersion != GuestExecGuardProtocolVersion || ac.agent.Platform != "linux") { + return GuestExecDeferred(GuestExecGuardUnavailable) + } + return nil +} diff --git a/internal/agentexec/guest_exec_safety_test.go b/internal/agentexec/guest_exec_safety_test.go new file mode 100644 index 0000000000..6e9fa2d932 --- /dev/null +++ b/internal/agentexec/guest_exec_safety_test.go @@ -0,0 +1,74 @@ +package agentexec + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" +) + +func testVMGuestExecCapabilityGate(t *testing.T) { + for _, tc := range []struct { + version int + platform string + admitted bool + }{ + {0, "linux", false}, {1, "linux", true}, {2, "linux", false}, {1, "windows", false}, + } { + t.Run(fmt.Sprintf("%s/version-%d", tc.platform, tc.version), func(t *testing.T) { + s := NewServer(allowAllTestTokens) + ts := newWSServer(t, s) + defer ts.Close() + conn, _, err := dialAgentExecWebSocket(t, ts.URL) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{AgentID: "node-agent", Hostname: "node", Version: "future-version-is-not-capability", Platform: tc.platform, Token: "fixture", GuestExecGuardVersion: tc.version})) + if !wsReadRegisteredPayload(t, conn).Success { + t.Fatal("registration rejected") + } + cmd := ExecuteCommandPayload{RequestID: "scan", Command: "echo inspected", TargetType: "vm", TargetID: "105", Trusted: true, Timeout: 2} + if !tc.admitted { + if result, err := s.ExecuteCommand(context.Background(), "node-agent", cmd); result != nil || err == nil || !IsGuestExecDeferred(err.Error()) { + t.Fatalf("legacy target dispatched: result=%#v err=%v", result, err) + } + if result, err := s.ReadFile(context.Background(), "node-agent", ReadFilePayload{RequestID: "read", Path: "/etc/os-release", TargetType: "vm", TargetID: "105"}); result != nil || err == nil || !IsGuestExecDeferred(err.Error()) { + t.Fatalf("legacy file target dispatched: result=%#v err=%v", result, err) + } + // The first message must be the pong, not an unsafe VM dispatch. + wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentPing, "", nil)) + if msg := wsReadRawMessage(t, conn); msg.Type != MsgTypePong { + t.Fatalf("unsafe message queued: %s", msg.Type) + } + return + } + type outcome struct { + result *CommandResultPayload + err error + } + done := make(chan outcome, 1) + go func() { + result, err := s.ExecuteCommand(context.Background(), "node-agent", cmd) + done <- outcome{result, err} + }() + msg, err := wsReadRawMessageWithTimeout(conn, 2*time.Second) + if err != nil || msg.Type != MsgTypeExecuteCmd || msg.Payload == nil { + t.Fatalf("dispatch=%#v err=%v", msg, err) + } + var payload ExecuteCommandPayload + if err := json.Unmarshal(*msg.Payload, &payload); err != nil { + t.Fatal(err) + } + if payload.TargetID != "105" || payload.TargetType != "vm" || payload.Command != cmd.Command || !payload.Trusted { + t.Fatalf("target changed: %#v", payload) + } + wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeCommandResult, payload.RequestID, CommandResultPayload{RequestID: payload.RequestID, Success: false, ExitCode: -1, Error: GuestExecDeferred(GuestExecVMLocked).Error()})) + got := <-done + if got.err != nil || got.result == nil || got.result.Success || !IsGuestExecDeferred(got.result.Error) { + t.Fatalf("deferral lost: %#v %v", got.result, got.err) + } + }) + } +} diff --git a/internal/agentexec/server.go b/internal/agentexec/server.go index 7f5bc48e72..b01182c6e0 100644 --- a/internal/agentexec/server.go +++ b/internal/agentexec/server.go @@ -130,15 +130,16 @@ type CommandAuthorizationRequest struct { } type agentConn struct { - conn *websocket.Conn - agent ConnectedAgent - admission AgentAdmission - sessionKey string - authorityKey string - approvalGrantKey []byte - writeMu sync.Mutex - done chan struct{} - doneOnce sync.Once + conn *websocket.Conn + agent ConnectedAgent + admission AgentAdmission + sessionKey string + authorityKey string + guestExecGuardVersion int + approvalGrantKey []byte + writeMu sync.Mutex + done chan struct{} + doneOnce sync.Once } type pendingHostOperation struct { @@ -1376,11 +1377,12 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { ActionPreflightVersion: reg.ActionPreflightVersion, DockerObservationVersion: reg.DockerObservationVersion, }, - admission: admission, - sessionKey: agentSessionKey(admission.OrganizationID, admission.AgentID), - authorityKey: uuid.NewString(), - approvalGrantKey: DeriveApprovalGrantKey(reg.Token), - done: make(chan struct{}), + admission: admission, + sessionKey: agentSessionKey(admission.OrganizationID, admission.AgentID), + authorityKey: uuid.NewString(), + approvalGrantKey: DeriveApprovalGrantKey(reg.Token), + guestExecGuardVersion: reg.GuestExecGuardVersion, + done: make(chan struct{}), } // Clear deadline for normal operation - both on the WebSocket and underlying connection @@ -2087,6 +2089,9 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute if err := requireLegacyFullTrustConnection(ac, "execute_command"); err != nil { return nil, err } + if err := requireVMGuestExecGuard(ac, cmd.TargetType); err != nil { + return nil, err + } if err := s.authorizeCommandPayload(cmd); err != nil { return nil, err } @@ -2782,6 +2787,9 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo if err := requireLegacyFullTrustConnection(ac, "read_file"); err != nil { return nil, err } + if err := requireVMGuestExecGuard(ac, req.TargetType); err != nil { + return nil, err + } readLog := log.With(). Str("agent_id", agentID). diff --git a/internal/agentexec/server_websocket_test.go b/internal/agentexec/server_websocket_test.go index e622e3a9be..2715c2de33 100644 --- a/internal/agentexec/server_websocket_test.go +++ b/internal/agentexec/server_websocket_test.go @@ -2169,7 +2169,10 @@ func TestTypedOperation_SocketDropAfterSendUnblocksDispatch(t *testing.T) { t.Fatal(err) } done := make(chan error, 1) - go func() { _, err := s.ExecuteProxmoxGuestLifecycle(context.Background(), "typed-a1", request); done <- err }() + go func() { + _, err := s.ExecuteProxmoxGuestLifecycle(context.Background(), "typed-a1", request) + done <- err + }() if sent, ok := agent.nextMessage(3 * time.Second); !ok || sent.Type != MsgTypeProxmoxGuestLifecycle { t.Fatalf("dispatched request = %+v", sent) } @@ -2247,3 +2250,7 @@ func TestRegistrationReservesFirstFrameBeforePublishingSession(t *testing.T) { t.Fatal("published session permits command writes before registration acknowledgement") } } + +func TestVMGuestExecCapabilityBoundToRegisteredSession(t *testing.T) { + testVMGuestExecCapabilityGate(t) +} diff --git a/internal/agentexec/types.go b/internal/agentexec/types.go index afebf7901b..24ea0ff90f 100644 --- a/internal/agentexec/types.go +++ b/internal/agentexec/types.go @@ -107,6 +107,7 @@ type AgentRegisterPayload struct { OperationReceiptVersion int `json:"operation_receipt_version,omitempty"` ActionPreflightVersion int `json:"action_preflight_version,omitempty"` DockerObservationVersion int `json:"docker_observation_version,omitempty"` + GuestExecGuardVersion int `json:"guest_exec_guard_version,omitempty"` } // RegisteredPayload is sent by server after successful registration diff --git a/internal/hostagent/command_client_test.go b/internal/hostagent/command_client_test.go index 22c2d13d08..13f3edf9d8 100644 --- a/internal/hostagent/command_client_test.go +++ b/internal/hostagent/command_client_test.go @@ -518,3 +518,7 @@ func TestCommandClient_ReplayedRequestWaitsForInFlightHandlerInsteadOfDropping(t func TestCommandClientGuestExecutionAdmissionContract(t *testing.T) { testVMGuestExecNoUnverifiedHandoff(t) } + +func TestGuestExecRealServerAgentDiscoveryAdmission(t *testing.T) { + testGuestExecRealServerAgentDiscoveryAdmission(t) +} diff --git a/internal/hostagent/commands.go b/internal/hostagent/commands.go index 5c54070f17..83009ec278 100644 --- a/internal/hostagent/commands.go +++ b/internal/hostagent/commands.go @@ -108,6 +108,7 @@ type CommandClient struct { caCertPath string serverFingerprint string deploySSHUser string + guestExecAdmission *guestExecGuard // nil selects the process-wide local guard commandPolicy *agentexec.CommandPolicy packageUpdates *packageUpdateManager storageCleanup *storageCleanupManager @@ -274,6 +275,7 @@ type registerPayload struct { OperationReceiptVersion int `json:"operation_receipt_version,omitempty"` ActionPreflightVersion int `json:"action_preflight_version,omitempty"` DockerObservationVersion int `json:"docker_observation_version,omitempty"` + GuestExecGuardVersion int `json:"guest_exec_guard_version,omitempty"` } type registeredPayload struct { @@ -538,6 +540,7 @@ func (c *CommandClient) sendRegistration(conn *websocket.Conn) error { OperationReceiptVersion: c.operationReceiptVersion(), ActionPreflightVersion: agentexec.ActionPreflightProtocolVersion, DockerObservationVersion: agentexec.DockerContainerObservationProtocolVersion, + GuestExecGuardVersion: c.guestExecGuardVersion(), }) if err != nil { return fmt.Errorf("marshal registration payload: %w", err) @@ -552,6 +555,13 @@ func (c *CommandClient) sendRegistration(conn *websocket.Conn) error { return conn.WriteJSON(msg) } +func (c *CommandClient) guestExecGuardVersion() int { + if runtime.GOOS == "linux" && !c.actionRunnerOnly { + return agentexec.GuestExecGuardProtocolVersion + } + return 0 +} + func (c *CommandClient) operationReceiptVersion() int { if c != nil && c.operationReceiptErr == nil && c.operationReceipts != nil { return operationreceipt.ProtocolVersion @@ -1556,11 +1566,44 @@ func (c *CommandClient) executeCommand(ctx context.Context, payload executeComma cmdCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() + var guard *guestExecGuard + var vmid string + uncertain := false + if strings.EqualFold(strings.TrimSpace(payload.TargetType), "vm") { + var valid bool + vmid, valid = canonicalGuestExecVMID(payload.TargetID) + if !valid { + result.Error = agentexec.GuestExecDeferred(agentexec.GuestExecInvalidTarget).Error() + result.ExitCode = -1 + return result + } + guard = c.guestExecAdmission + if guard == nil { + guard = localGuestExecGuard + } + release, err := guard.acquire(cmdCtx, vmid) + if err != nil { + result.Error = err.Error() + result.ExitCode = -1 + return result + } + defer func() { release(uncertain) }() + if err := guard.verifyUnlocked(cmdCtx, vmid); err != nil { + result.Error = err.Error() + result.ExitCode = -1 + return result + } + } + command := wrapCommand(payload) // Execute the command var cmd *exec.Cmd - if runtime.GOOS == "windows" { + if guard != nil { + // No host-shell or PATH interpretation of the owning VM/CLI. The + // guest shell still receives the original, policy-approved command. + cmd = execCommandContext(cmdCtx, localGuestExecQM, "guest", "exec", vmid, "--", "sh", "-c", payload.Command) + } else if runtime.GOOS == "windows" { cmd = execCommandContext(cmdCtx, "cmd", "/C", command) } else { cmd = execCommandContext(cmdCtx, "sh", "-c", command) @@ -1585,6 +1628,7 @@ func (c *CommandClient) executeCommand(ctx context.Context, payload executeComma cmd.Stderr = stderr err := cmd.Run() + outputIncomplete := errors.Is(err, exec.ErrWaitDelay) result.Stdout = stdout.String() result.Stderr = stderr.String() @@ -1618,6 +1662,22 @@ func (c *CommandClient) executeCommand(ctx context.Context, payload executeComma result.Success = true } + if guard != nil { + // Killing the local qm process does not prove a handed-off guest + // command stopped. Never retry it or immediately start the next one. + if err != nil || cmdCtx.Err() != nil || outputIncomplete { + uncertain = true + result.Success = false + result.Error = agentexec.GuestExecDeferred(agentexec.GuestExecCompletionUnknown).Error() + } else if err := guard.verifyUnlocked(cmdCtx, vmid); err != nil { + uncertain = true + result.Success = false + result.Error = err.Error() + } + // Retain the actual process exit/output, even if postflight safety + // is unknown. Discovery must not treat that output as fresh evidence. + } + return result } diff --git a/internal/hostagent/guest_exec_config_linux.go b/internal/hostagent/guest_exec_config_linux.go new file mode 100644 index 0000000000..4d306804c1 --- /dev/null +++ b/internal/hostagent/guest_exec_config_linux.go @@ -0,0 +1,42 @@ +//go:build linux + +package hostagent + +import ( + "context" + "errors" + "os" + "os/exec" + "path/filepath" + "time" +) + +const localGuestExecQM = "/usr/sbin/qm" + +func readLocalGuestExecConfig(ctx context.Context, vmid string) ([]byte, error) { + // qm can read a remote node's cluster config too. Require the VM to be + // owned by this node before AND after the non-QGA config query. + configPath := filepath.Join("/etc/pve/local/qemu-server", vmid+".conf") + isLocal := func() bool { + info, err := os.Lstat(configPath) + return err == nil && info.Mode().IsRegular() + } + if !isLocal() { + return nil, errors.New("local guest config unavailable") + } + resolved, err := filepath.EvalSymlinks(localGuestExecQM) + if err != nil || validateTrustedExecutable(resolved, 0) != nil { + return nil, errors.New("trusted qm unavailable") + } + readCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + cmd := exec.CommandContext(readCtx, resolved, "config", vmid, "--current") + configureCommandProcessGroup(cmd) + cmd.WaitDelay = time.Second + stdout, stderr := newCappedBuffer(guestExecConfigLimit), newCappedBuffer(1024) + cmd.Stdout, cmd.Stderr = stdout, stderr + if err := cmd.Run(); err != nil || stdout.truncated || readCtx.Err() != nil || !isLocal() { + return nil, errors.New("local guest config unverified") + } + return []byte(stdout.String()), nil +} diff --git a/internal/hostagent/guest_exec_config_other.go b/internal/hostagent/guest_exec_config_other.go new file mode 100644 index 0000000000..789dd0eda9 --- /dev/null +++ b/internal/hostagent/guest_exec_config_other.go @@ -0,0 +1,14 @@ +//go:build !linux + +package hostagent + +import ( + "context" + "errors" +) + +const localGuestExecQM = "/usr/sbin/qm" + +func readLocalGuestExecConfig(context.Context, string) ([]byte, error) { + return nil, errors.New("local PVE guest config unavailable on this platform") +} diff --git a/internal/hostagent/guest_exec_discovery_integration_test.go b/internal/hostagent/guest_exec_discovery_integration_test.go new file mode 100644 index 0000000000..cb87ac0067 --- /dev/null +++ b/internal/hostagent/guest_exec_discovery_integration_test.go @@ -0,0 +1,130 @@ +package hostagent + +import ( + "context" + "net/http" + "net/http/httptest" + "os/exec" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" + "github.com/rcourtman/pulse-go-rewrite/internal/servicediscovery" + "github.com/rs/zerolog" +) + +type guestSafetyDiscoveryExecutor struct{ server *agentexec.Server } + +func (e guestSafetyDiscoveryExecutor) ExecuteCommand(ctx context.Context, id string, p servicediscovery.ExecuteCommandPayload) (*servicediscovery.CommandResultPayload, error) { + r, err := e.server.ExecuteCommand(ctx, id, agentexec.ExecuteCommandPayload{RequestID: p.RequestID, Command: p.Command, TargetType: p.TargetType, TargetID: p.TargetID, Timeout: p.Timeout, Trusted: true}) + if err != nil || r == nil { + return nil, err + } + return &servicediscovery.CommandResultPayload{RequestID: r.RequestID, Success: r.Success, Stdout: r.Stdout, Stderr: r.Stderr, ExitCode: r.ExitCode, Error: r.Error}, nil +} +func (e guestSafetyDiscoveryExecutor) GetConnectedAgents() []servicediscovery.ConnectedAgent { + var out []servicediscovery.ConnectedAgent + for _, a := range e.server.GetConnectedAgents() { + out = append(out, servicediscovery.ConnectedAgent{AgentID: a.AgentID, Hostname: a.Hostname}) + } + return out +} +func (e guestSafetyDiscoveryExecutor) IsAgentConnected(id string) bool { + return e.server.IsAgentConnected(id) +} + +// Real server/session/registration/agent/Discovery paths, with a fake non-QGA +// config reader and fake provider process. This is NOT native PVE acceptance. +func testGuestExecRealServerAgentDiscoveryAdmission(t *testing.T) { + s := agentexec.NewServer(func(token, agent, host string) bool { + return token == "fixture" && agent == "node-agent" && host == "node" + }) + ts := httptest.NewServer(http.HandlerFunc(s.HandleWebSocket)) + defer ts.Close() + logger := zerolog.Nop() + c := NewCommandClient(Config{PulseURL: ts.URL, APIToken: "fixture", StateDir: t.TempDir(), Logger: &logger}, "node-agent", "node", "linux", "6") + var locked, lockDuringHandoff atomic.Bool + var qgaCalls, configReads atomic.Int32 + locked.Store(true) + c.guestExecAdmission = newGuestExecGuard(func(_ context.Context, vmid string) ([]byte, error) { + configReads.Add(1) + if vmid != "105" { + t.Errorf("wrong VM read: %s", vmid) + } + if locked.Load() { + return []byte("name: vm\nlock: backup\n"), nil + } + return []byte("name: vm\n"), nil + }) + oldExec := execCommandContext + execCommandContext = func(ctx context.Context, name string, args ...string) *exec.Cmd { + if name == localGuestExecQM { + qgaCalls.Add(1) + if len(args) != 7 || args[2] != "105" { + t.Errorf("wrong VM handoff: %#v", args) + } + if lockDuringHandoff.Load() { + locked.Store(true) + } + } + return exec.CommandContext(ctx, "sh", "-c", "printf safe-observation") + } + t.Cleanup(func() { execCommandContext = oldExec }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- c.Run(ctx) }() + defer func() { + cancel() + _ = c.Close() + s.Shutdown() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Error("agent did not stop") + } + }() + deadline := time.Now().Add(3 * time.Second) + for !s.IsAgentConnected("node-agent") && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if !s.IsAgentConnected("node-agent") { + t.Fatal("agent not connected") + } + scanner := servicediscovery.NewDeepScanner(guestSafetyDiscoveryExecutor{s}) + for _, rt := range []servicediscovery.ResourceType{servicediscovery.ResourceTypeVM, servicediscovery.ResourceTypeDockerVM} { + id := "105" + if rt == servicediscovery.ResourceTypeDockerVM { + id = "105:app" + } + got, err := scanner.Scan(context.Background(), servicediscovery.DiscoveryRequest{ResourceType: rt, ResourceID: id, TargetID: "node-agent"}) + if err == nil || !agentexec.IsGuestExecDeferred(err.Error()) || got == nil || len(got.CommandOutputs) != 0 || qgaCalls.Load() != 0 { + t.Fatalf("locked scan escaped: %+v %v calls=%d", got, err, qgaCalls.Load()) + } + } + locked.Store(false) + got, err := scanner.ScanVM(context.Background(), "node-agent", "node", "105") + want := int32(len(servicediscovery.GetCommandsForResource(servicediscovery.ResourceTypeVM))) + if err != nil || got == nil || int32(len(got.CommandOutputs)) != want || qgaCalls.Load() != want { + t.Fatalf("healthy explicit scan changed: %+v %v calls=%d", got, err, qgaCalls.Load()) + } + lockDuringHandoff.Store(true) + got, err = scanner.ScanVM(context.Background(), "node-agent", "node", "105") + if err == nil || !agentexec.IsGuestExecDeferred(err.Error()) || got == nil || len(got.CommandOutputs) != 0 || qgaCalls.Load() != want+1 { + t.Fatalf("in-flight deferral was lost/retried: %+v %v calls=%d", got, err, qgaCalls.Load()) + } + // Host probes do not use this QGA guard, even while VM admission is paused. + if host, err := scanner.ScanHost(context.Background(), "node-agent", "node"); err != nil || host == nil || len(host.CommandOutputs) == 0 || qgaCalls.Load() != want+1 { + t.Fatalf("host control changed: %+v %v", host, err) + } + if !s.IsAgentConnected("node-agent") { + t.Fatal("deferral dropped session liveness") + } + if configReads.Load() != 2+want*2+2 { + t.Fatalf("config retries/unexpected reads: %d", configReads.Load()) + } + if !strings.Contains(err.Error(), "operation lock") { + t.Fatalf("lost lock reason: %v", err) + } +} diff --git a/internal/hostagent/guest_exec_guard.go b/internal/hostagent/guest_exec_guard.go new file mode 100644 index 0000000000..94285746d9 --- /dev/null +++ b/internal/hostagent/guest_exec_guard.go @@ -0,0 +1,153 @@ +package hostagent + +import ( + "context" + "strconv" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" +) + +const ( + guestExecConfigLimit = 64 << 10 + guestExecCooldown = time.Minute + maxGuestExecGuardEntries = 4096 +) + +type guestExecEntry struct { + done chan struct{} + until time.Time +} + +type guestExecGuard struct { + mu sync.Mutex + entries map[string]guestExecEntry + readConfig func(context.Context, string) ([]byte, error) + now func() time.Time +} + +// Shared across reconnects and CommandClients in this agent process. This is +// not a lock shared with the Pulse HTTP poller, other processes, or PVE/QGA. +var localGuestExecGuard = newGuestExecGuard(readLocalGuestExecConfig) + +func newGuestExecGuard(readConfig func(context.Context, string) ([]byte, error)) *guestExecGuard { + return &guestExecGuard{entries: make(map[string]guestExecEntry), readConfig: readConfig, now: time.Now} +} + +func canonicalGuestExecVMID(raw string) (string, bool) { + raw = strings.TrimSpace(raw) + if !safeTargetIDPattern.MatchString(raw) { + return "", false + } + id, err := strconv.ParseUint(raw, 10, 32) + if err != nil || id == 0 { + return "", false + } + return strconv.FormatUint(id, 10), true +} + +// Wait only for another local command, never poll QGA or retry a dispatched +// command. A waiter rechecks cooldown and the authoritative config on admission. +func (g *guestExecGuard) acquire(ctx context.Context, vmid string) (func(bool), error) { + for { + if err := ctx.Err(); err != nil { + return nil, err + } + g.mu.Lock() + now := g.now() + entry, exists := g.entries[vmid] + if entry.done != nil { + done := entry.done + g.mu.Unlock() + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-done: + continue + } + } + if now.Before(entry.until) { + g.mu.Unlock() + return nil, agentexec.GuestExecDeferred(agentexec.GuestExecCooldown) + } + if !exists && len(g.entries) >= maxGuestExecGuardEntries { + for key, e := range g.entries { + if e.done == nil && !now.Before(e.until) { + delete(g.entries, key) + } + } + if len(g.entries) >= maxGuestExecGuardEntries { + g.mu.Unlock() + return nil, agentexec.GuestExecDeferred(agentexec.GuestExecCapacity) + } + } + done := make(chan struct{}) + g.entries[vmid] = guestExecEntry{done: done} + g.mu.Unlock() + return func(uncertain bool) { + g.mu.Lock() + defer g.mu.Unlock() + if uncertain { + g.entries[vmid] = guestExecEntry{until: g.now().Add(guestExecCooldown)} + } else { + delete(g.entries, vmid) + } + close(done) + }, nil + } +} + +func (g *guestExecGuard) verifyUnlocked(ctx context.Context, vmid string) error { + if err := ctx.Err(); err != nil { + return err + } + config, err := g.readConfig(ctx, vmid) + if err != nil { + return agentexec.GuestExecDeferred(agentexec.GuestExecLockUnverified) + } + lock, valid := guestExecConfigLock(config) + if !valid { + return agentexec.GuestExecDeferred(agentexec.GuestExecLockUnverified) + } + if lock != "" { + return agentexec.GuestExecDeferred(agentexec.GuestExecVMLocked) + } + return nil +} + +// qm config --current emits one key/value dictionary, not pending/snapshot +// sections. Reject empty, malformed, oversized or duplicate/case-conflicting +// fields rather than allowing last-key-wins to erase a lock. +func guestExecConfigLock(config []byte) (string, bool) { + if len(config) == 0 || len(config) > guestExecConfigLimit || !utf8.Valid(config) { + return "", false + } + fields := make(map[string]bool) + lock := "" + for _, line := range strings.Split(string(config), "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + key, value, ok := strings.Cut(line, ":") + if !ok || key == "" || strings.TrimSpace(key) != key || fields[key] { + return "", false + } + for _, c := range key { + if !(c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '_') { + return "", false + } + } + fields[key] = true + if key == "lock" { + lock = strings.TrimSpace(value) + if lock == "" { + return "", false + } + } + } + return lock, len(fields) > 0 +} diff --git a/internal/hostagent/guest_exec_guard_test.go b/internal/hostagent/guest_exec_guard_test.go new file mode 100644 index 0000000000..7a03b532ef --- /dev/null +++ b/internal/hostagent/guest_exec_guard_test.go @@ -0,0 +1,206 @@ +package hostagent + +import ( + "context" + "errors" + "os/exec" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" +) + +func TestGuestExecConfigLockDictionary(t *testing.T) { + for _, tc := range []struct { + name, config, lock string + valid bool + }{ + {"unlocked", "cores: 2\nname: home-assistant\n", "", true}, + {"unicode name", "name: cafĂ©\nmemory: 8192\n", "", true}, + {"backup", "name: vm\nlock: backup\n", "backup", true}, + {"other operation", "lock: migrate\n", "migrate", true}, + {"comment", "# lock: backup\nname: vm\n", "", true}, + {"erased duplicate", "lock: backup\nlock:\n", "", false}, + {"duplicate unlocked", "cores: 2\ncores: 2\n", "", false}, + {"case conflict", "lock: backup\nLOCK:\n", "", false}, + {"empty lock", "name: vm\nlock:\n", "", false}, + {"snapshot", "name: vm\n[snapshot]\nlock: backup\n", "", false}, + {"empty", "", "", false}, + {"only comments", "# no config\n", "", false}, + {"malformed", "lock backup\n", "", false}, + {"command error", "ERROR: VM unavailable\n", "", false}, + {"bad key", "name: vm\n lock : backup\n", "", false}, + {"non utf8", "name: \xff\n", "", false}, + {"overflow", "name: " + strings.Repeat("x", guestExecConfigLimit), "", false}, + } { + t.Run(tc.name, func(t *testing.T) { + lock, valid := guestExecConfigLock([]byte(tc.config)) + if valid != tc.valid || lock != tc.lock { + t.Fatalf("lock=%q valid=%v", lock, valid) + } + }) + } +} + +func TestGuestExecCommandPreflightAndPostflight(t *testing.T) { + for _, tc := range []struct { + name string + configs []string + failRead int + wantCalls int + wantSuccess bool + wantReason string + }{ + {"healthy", []string{"name: vm\n", "name: vm\n"}, -1, 1, true, ""}, + {"backup before", []string{"lock: backup\n"}, -1, 0, false, agentexec.GuestExecVMLocked}, + {"migration before", []string{"lock: migrate\n"}, -1, 0, false, agentexec.GuestExecVMLocked}, + {"unverified owner", []string{""}, 0, 0, false, agentexec.GuestExecLockUnverified}, + {"ambiguous lock", []string{"lock: backup\nlock:\n"}, -1, 0, false, agentexec.GuestExecLockUnverified}, + {"backup in flight", []string{"name: vm\n", "lock: backup\n"}, -1, 1, false, agentexec.GuestExecVMLocked}, + {"owner lost in flight", []string{"name: vm\n", ""}, 1, 1, false, agentexec.GuestExecLockUnverified}, + } { + t.Run(tc.name, func(t *testing.T) { + var reads, calls atomic.Int32 + g := newGuestExecGuard(func(_ context.Context, vmid string) ([]byte, error) { + if vmid != "105" { + t.Errorf("config read escaped canonical identity: %s", vmid) + } + i := int(reads.Add(1)) - 1 + if i == tc.failRead { + return nil, errors.New("private config/provider text") + } + if i >= len(tc.configs) { + t.Errorf("unexpected config retry %d", i) + return nil, errors.New("retry") + } + return []byte(tc.configs[i]), nil + }) + c := &CommandClient{guestExecAdmission: g} + oldExec := execCommandContext + execCommandContext = func(ctx context.Context, name string, args ...string) *exec.Cmd { + calls.Add(1) + if name != localGuestExecQM || !reflect.DeepEqual(args, []string{"guest", "exec", "105", "--", "sh", "-c", "echo original"}) { + t.Errorf("changed command/identity: %s %#v", name, args) + } + return exec.CommandContext(ctx, "sh", "-c", "printf 'original-result'") + } + t.Cleanup(func() { execCommandContext = oldExec }) + payload := testApprovedCommandPayload(t, c, executeCommandPayload{RequestID: "scan", Command: "echo original", TargetType: " VM ", TargetID: "000105", Trusted: true}) + got := c.executeCommand(context.Background(), payload) + if got.Success != tc.wantSuccess || int(calls.Load()) != tc.wantCalls { + t.Fatalf("result=%#v calls=%d", got, calls.Load()) + } + if tc.wantReason != "" && got.Error != agentexec.GuestExecDeferred(tc.wantReason).Error() { + t.Fatalf("error=%q", got.Error) + } + if tc.wantCalls == 0 && (got.Stdout != "" || got.Stderr != "") { + t.Fatal("preflight published guest output") + } + if tc.wantCalls == 1 && (got.Stdout != "original-result" || got.ExitCode != 0) { + t.Fatalf("lost actual process completion: %#v", got) + } + if strings.Contains(got.Error, "private") { + t.Fatal("provider/config data leaked") + } + // A lock clearing after a preflight rejection is not an automatic + // retry. A postflight unknown keeps even a NEW request in cooldown. + if tc.wantCalls == 1 && !got.Success { + before := reads.Load() + next := c.executeCommand(context.Background(), payload) + if next.Success || next.Error != agentexec.GuestExecDeferred(agentexec.GuestExecCooldown).Error() || reads.Load() != before || calls.Load() != 1 { + t.Fatalf("uncertainty caused replay: %#v", next) + } + } + }) + } +} + +func TestGuestExecSerializationCancellationCapacityAndExplicitResumption(t *testing.T) { + now := time.Now() + g := newGuestExecGuard(func(context.Context, string) ([]byte, error) { return []byte("name: vm\n"), nil }) + g.now = func() time.Time { return now } + release, err := g.acquire(context.Background(), "105") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := g.acquire(ctx, "105"); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("same-target waiter: %v", err) + } + other, err := g.acquire(context.Background(), "106") + if err != nil { + t.Fatal(err) + } + other(false) + release(true) + if _, err := g.acquire(context.Background(), "105"); err == nil || err.Error() != agentexec.GuestExecDeferred(agentexec.GuestExecCooldown).Error() { + t.Fatalf("no uncertainty fence: %v", err) + } + g.mu.Lock() + now = now.Add(guestExecCooldown) + g.mu.Unlock() + resumed, err := g.acquire(context.Background(), "105") + if err != nil { + t.Fatal(err) + } + resumed(false) + g.mu.Lock() + if len(g.entries) != 0 { + t.Fatal("completed entries retained") + } + for i := 0; i < maxGuestExecGuardEntries; i++ { + g.entries["fixture-"+strings.Repeat("x", i%32)+time.Duration(i).String()] = guestExecEntry{until: now.Add(time.Minute)} + } + g.mu.Unlock() + if _, err := g.acquire(context.Background(), "200"); err == nil || err.Error() != agentexec.GuestExecDeferred(agentexec.GuestExecCapacity).Error() { + t.Fatalf("unbounded admission: %v", err) + } + g.mu.Lock() + now = now.Add(time.Minute) + g.mu.Unlock() + cleaned, err := g.acquire(context.Background(), "200") + if err != nil { + t.Fatal(err) + } + cleaned(false) +} + +func TestGuestExecCanceledHandoffNeverRetriesAndRechecksOnExplicitResumption(t *testing.T) { + var reads, calls atomic.Int32 + g := newGuestExecGuard(func(context.Context, string) ([]byte, error) { reads.Add(1); return []byte("name: vm\n"), nil }) + now := time.Now() + g.now = func() time.Time { return now } + c := &CommandClient{guestExecAdmission: g} + oldExec := execCommandContext + execCommandContext = func(ctx context.Context, _ string, _ ...string) *exec.Cmd { + calls.Add(1) + return exec.CommandContext(ctx, "sh", "-c", "sleep 5") + } + t.Cleanup(func() { execCommandContext = oldExec }) + payload := testApprovedCommandPayload(t, c, executeCommandPayload{Command: "echo original", TargetType: "vm", TargetID: "105", Trusted: true}) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + got := c.executeCommand(ctx, payload) + if got.Success || got.Error != agentexec.GuestExecDeferred(agentexec.GuestExecCompletionUnknown).Error() { + t.Fatalf("canceled result=%#v", got) + } + next := c.executeCommand(context.Background(), payload) + if next.Success || calls.Load() != 1 || reads.Load() != 1 { + t.Fatalf("canceled handoff replayed: %#v calls=%d reads=%d", next, calls.Load(), reads.Load()) + } + g.mu.Lock() + now = now.Add(guestExecCooldown) + g.mu.Unlock() + execCommandContext = func(ctx context.Context, _ string, _ ...string) *exec.Cmd { + calls.Add(1) + return exec.CommandContext(ctx, "sh", "-c", "printf resumed") + } + got = c.executeCommand(context.Background(), payload) + if !got.Success || got.Stdout != "resumed" || reads.Load() != 3 || calls.Load() != 2 { + t.Fatalf("explicit resumption failed: %#v", got) + } +} diff --git a/internal/servicediscovery/deep_scanner.go b/internal/servicediscovery/deep_scanner.go index 9e7232b671..e091af1791 100644 --- a/internal/servicediscovery/deep_scanner.go +++ b/internal/servicediscovery/deep_scanner.go @@ -7,6 +7,7 @@ import ( "time" "github.com/google/uuid" + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rs/zerolog/log" ) @@ -191,7 +192,14 @@ func (s *DeepScanner) Scan(ctx context.Context, req DiscoveryRequest) (*ScanResu } // Find the agent for this host + vmTarget := s.getTargetType(req.ResourceType) == "vm" agentID := s.findAgentForTarget(requestTargetID, req.Hostname) + if vmTarget { + agentID = s.findAgentForVMTarget(requestTargetID) + if agentID == "" { + return nil, agentexec.GuestExecDeferred(agentexec.GuestExecTargetUnverified) + } + } if agentID == "" { scanLog.Warn(). Str("action", "scan_precondition_failed"). @@ -230,109 +238,135 @@ func (s *DeepScanner) Scan(ctx context.Context, req DiscoveryRequest) (*ScanResu var wg sync.WaitGroup var mu sync.Mutex - for _, cmd := range commands { - wg.Add(1) - go func(cmd DiscoveryCommand) { - defer wg.Done() + var safetyError error + runCommand := func(cmd DiscoveryCommand) { - select { - case semaphore <- struct{}{}: - defer func() { <-semaphore }() - case <-ctx.Done(): + select { + case semaphore <- struct{}{}: + defer func() { <-semaphore }() + case <-ctx.Done(): + return + } + + // Build the actual command to run + actualCmd := s.buildCommand(req.ResourceType, req.ResourceID, cmd.Command) + + // Get the target ID for the agent + targetID := s.getTargetID(req.ResourceType, req.ResourceID) + + // Only validate TargetID when it will be interpolated into shell commands + // by the agent (container/vm types). Agent/docker types don't use TargetID + // in command wrapping, so they can have any format (including colons for IPv6). + targetType := s.getTargetType(req.ResourceType) + if targetType == "container" || targetType == "vm" { + if err := ValidateResourceID(targetID); err != nil { + mu.Lock() + result.Errors[cmd.Name] = fmt.Sprintf("invalid target ID: %v", err) + mu.Unlock() return } + } - // Build the actual command to run - actualCmd := s.buildCommand(req.ResourceType, req.ResourceID, cmd.Command) + // Execute the command + cmdCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() - // Get the target ID for the agent - targetID := s.getTargetID(req.ResourceType, req.ResourceID) + cmdResult, err := s.executor.ExecuteCommand(cmdCtx, agentID, ExecuteCommandPayload{ + RequestID: uuid.New().String(), + Command: actualCmd, + TargetType: s.getTargetType(req.ResourceType), + TargetID: targetID, + Timeout: cmd.Timeout, + }) - // Only validate TargetID when it will be interpolated into shell commands - // by the agent (container/vm types). Agent/docker types don't use TargetID - // in command wrapping, so they can have any format (including colons for IPv6). - targetType := s.getTargetType(req.ResourceType) - if targetType == "container" || targetType == "vm" { - if err := ValidateResourceID(targetID); err != nil { - mu.Lock() - result.Errors[cmd.Name] = fmt.Sprintf("invalid target ID: %v", err) - mu.Unlock() - return - } + mu.Lock() + defer mu.Unlock() + + // A safety deferral is never an optional probe failure or usable + // partial evidence. Stop this VM scan, including nested Docker. + if vmTarget { + message := "" + if err != nil { + message = err.Error() + } else if cmdResult != nil { + message = cmdResult.Error + } + if agentexec.IsGuestExecDeferred(message) { + safetyError = fmt.Errorf("%s", message) + result.Errors[cmd.Name] = message + return + } + } + + if err != nil { + if !cmd.Optional { + result.Errors[cmd.Name] = err.Error() + } + scanLog.Debug(). + Str("action", "command_execute_failed"). + Err(err). + Str("command", cmd.Name). + Bool("optional", cmd.Optional). + Msg("Command failed during discovery") + return + } + + if cmdResult != nil { + output := cmdResult.Stdout + if cmdResult.Stderr != "" && output != "" { + output += "\n--- stderr ---\n" + cmdResult.Stderr + } else if cmdResult.Stderr != "" { + output = cmdResult.Stderr } - // Execute the command - cmdCtx, cancel := context.WithTimeout(ctx, timeout) - defer cancel() + if output != "" { + result.CommandOutputs[cmd.Name] = output + } - cmdResult, err := s.executor.ExecuteCommand(cmdCtx, agentID, ExecuteCommandPayload{ - RequestID: uuid.New().String(), - Command: actualCmd, - TargetType: s.getTargetType(req.ResourceType), - TargetID: targetID, - Timeout: cmd.Timeout, - }) + if !cmdResult.Success && cmdResult.Error != "" && !cmd.Optional { + result.Errors[cmd.Name] = cmdResult.Error + } - mu.Lock() - defer mu.Unlock() - - if err != nil { + if !cmdResult.Success { + event := scanLog.Debug() if !cmd.Optional { - result.Errors[cmd.Name] = err.Error() + event = scanLog.Warn() } - scanLog.Debug(). - Str("action", "command_execute_failed"). - Err(err). + event. + Str("action", "command_result_failed"). Str("command", cmd.Name). Bool("optional", cmd.Optional). - Msg("Command failed during discovery") - return + Int("exit_code", cmdResult.ExitCode). + Str("request_id", cmdResult.RequestID). + Str("command_error", cmdResult.Error). + Msg("Deep scan command reported failure") } + } - if cmdResult != nil { - output := cmdResult.Stdout - if cmdResult.Stderr != "" && output != "" { - output += "\n--- stderr ---\n" + cmdResult.Stderr - } else if cmdResult.Stderr != "" { - output = cmdResult.Stderr - } - - if output != "" { - result.CommandOutputs[cmd.Name] = output - } - - if !cmdResult.Success && cmdResult.Error != "" && !cmd.Optional { - result.Errors[cmd.Name] = cmdResult.Error - } - - if !cmdResult.Success { - event := scanLog.Debug() - if !cmd.Optional { - event = scanLog.Warn() - } - event. - Str("action", "command_result_failed"). - Str("command", cmd.Name). - Bool("optional", cmd.Optional). - Int("exit_code", cmdResult.ExitCode). - Str("request_id", cmdResult.RequestID). - Str("command_error", cmdResult.Error). - Msg("Deep scan command reported failure") - } + // Update progress and broadcast + s.mu.Lock() + if prog, ok := s.progress[resourceID]; ok { + prog.CompletedSteps++ + prog.CurrentCommand = cmd.Name + progressCopy := *prog + s.mu.Unlock() + s.notifyProgress(&progressCopy) + } else { + s.mu.Unlock() + } + } + for _, cmd := range commands { + if vmTarget { + // QGA has a serial command channel; parallel Discovery probes are + // not useful and can queue behind a backup or uncertain command. + runCommand(cmd) + if safetyError != nil { + break } - - // Update progress and broadcast - s.mu.Lock() - if prog, ok := s.progress[resourceID]; ok { - prog.CompletedSteps++ - prog.CurrentCommand = cmd.Name - progressCopy := *prog - s.mu.Unlock() - s.notifyProgress(&progressCopy) - } else { - s.mu.Unlock() - } - }(cmd) + } else { + wg.Add(1) + go func(cmd DiscoveryCommand) { defer wg.Done(); runCommand(cmd) }(cmd) + } } wg.Wait() @@ -349,6 +383,12 @@ func (s *DeepScanner) Scan(ctx context.Context, req DiscoveryRequest) (*ScanResu ElapsedMs: result.CompletedAt.Sub(startTime).Milliseconds(), PercentComplete: 100, } + if safetyError != nil { + completionProgress.CurrentStep = "Guest execution paused" + completionProgress.Error = safetyError.Error() + completionProgress.CompletedSteps = len(result.CommandOutputs) + completionProgress.PercentComplete = float64(completionProgress.CompletedSteps) / float64(len(commands)) * 100 + } s.notifyProgress(&completionProgress) scanLog.Info(). @@ -358,7 +398,7 @@ func (s *DeepScanner) Scan(ctx context.Context, req DiscoveryRequest) (*ScanResu Dur("duration", result.CompletedAt.Sub(result.StartedAt)). Msg("Deep scan completed") - return result, nil + return result, safetyError } // buildCommand wraps the command appropriately for the resource type. @@ -479,6 +519,31 @@ func (s *DeepScanner) findAgentForTarget(targetID, hostname string) string { return "" } +// VM execution never falls back to an unrelated sole agent or a guest-name +// hint. A short hostname alias is admitted only when it uniquely names a node. +func (s *DeepScanner) findAgentForVMTarget(targetID string) string { + targetID = strings.TrimSpace(targetID) + if targetID == "" { + return "" + } + agents := s.executor.GetConnectedAgents() + for _, agent := range agents { + if agent.AgentID == targetID { + return agent.AgentID + } + } + matched := "" + for _, agent := range agents { + if unifiedresources.HostnamesEquivalent(agent.Hostname, targetID) { + if matched != "" { + return "" + } + matched = agent.AgentID + } + } + return matched +} + // GetProgress returns a copy of the current progress of a scan. // Returns nil if no scan is in progress for the resource. // A copy is returned to avoid data races with the scan goroutine. diff --git a/internal/servicediscovery/guest_exec_safety_test.go b/internal/servicediscovery/guest_exec_safety_test.go new file mode 100644 index 0000000000..a4610cd029 --- /dev/null +++ b/internal/servicediscovery/guest_exec_safety_test.go @@ -0,0 +1,102 @@ +package servicediscovery + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" +) + +type deferringGuestExecutor struct { + *stubExecutor + calls, deferAt int +} + +func (e *deferringGuestExecutor) ExecuteCommand(ctx context.Context, agentID string, p ExecuteCommandPayload) (*CommandResultPayload, error) { + e.calls++ + if e.calls == e.deferAt { + return &CommandResultPayload{RequestID: p.RequestID, Success: false, ExitCode: 0, Stdout: "in-flight output is not fresh evidence", Error: agentexec.GuestExecDeferred(agentexec.GuestExecVMLocked).Error()}, nil + } + return &CommandResultPayload{RequestID: p.RequestID, Success: true, Stdout: "earlier observation"}, nil +} + +func testDiscoveryGuestSafetyStopsAndPreservesSavedEvidence(t *testing.T) { + for _, rt := range []ResourceType{ResourceTypeVM, ResourceTypeDockerVM} { + for _, deferAt := range []int{1, 2} { + t.Run(string(rt)+"/"+time.Duration(deferAt).String(), func(t *testing.T) { + id := "105" + if rt == ResourceTypeDockerVM { + id = "105:app" + } + e := &deferringGuestExecutor{stubExecutor: &stubExecutor{agents: []ConnectedAgent{{AgentID: "node", Hostname: "node"}}}, deferAt: deferAt} + scanner := NewDeepScanner(e) + var terminal *DiscoveryProgress + scanner.SetProgressCallback(func(p *DiscoveryProgress) { copy := *p; terminal = © }) + req := DiscoveryRequest{ResourceType: rt, ResourceID: id, TargetID: "node", Force: true} + got, err := scanner.Scan(context.Background(), req) + if err == nil || !agentexec.IsGuestExecDeferred(err.Error()) || got == nil || e.calls != deferAt || len(got.CommandOutputs) != deferAt-1 { + t.Fatalf("scan continued: %+v %v calls=%d", got, err, e.calls) + } + if terminal == nil || terminal.Error == "" || terminal.PercentComplete == 100 || scanner.IsScanning(MakeResourceID(rt, "node", id)) { + t.Fatalf("false completion/progress: %+v", terminal) + } + for _, out := range got.CommandOutputs { + if strings.Contains(out, "in-flight") { + t.Fatal("unverified output became fresh evidence") + } + } + store, err := NewStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + store.crypto = nil + saved := &ResourceDiscovery{ID: MakeResourceID(rt, "node", id), ResourceType: rt, ResourceID: id, TargetID: "node", ServiceName: "saved observation", UserNotes: "keep", RawCommandOutput: map[string]string{"old": "retain"}, UpdatedAt: time.Now().Add(-time.Hour), DiscoveredAt: time.Now().Add(-time.Hour)} + if err := store.Save(saved); err != nil { + t.Fatal(err) + } + before, _ := store.Get(saved.ID) + beforeJSON, _ := json.Marshal(before) + cfg := DefaultConfig() + cfg.CommandScanning = true + service := NewService(store, scanner, cfg) + analyzer := &stubAnalyzer{response: `{"service_type":"nginx","confidence":0.9}`} + service.SetAIAnalyzer(analyzer) + e.calls = 0 + if fresh, err := service.DiscoverResource(context.Background(), req); fresh != nil || err == nil || !agentexec.IsGuestExecDeferred(err.Error()) { + t.Fatalf("manual deferral became success: %+v %v", fresh, err) + } + after, _ := store.Get(saved.ID) + afterJSON, _ := json.Marshal(after) + if string(beforeJSON) != string(afterJSON) || analyzer.calls != 0 || e.calls != deferAt { + t.Fatal("manual pause replaced saved evidence, analysed partial output, or retried") + } + e.calls = 0 + if got := service.enhanceWithDeepScan(context.Background(), saved, DockerHost{}); got != saved || analyzer.calls != 0 || e.calls != deferAt { + t.Fatal("automatic deferral fabricated fresh evidence") + } + }) + } + } +} + +func TestVMDiscoveryRequiresUniqueOwningNode(t *testing.T) { + for _, tc := range []struct { + name, target, hint string + agents []ConnectedAgent + }{ + {"sole unrelated agent", "node", "other", []ConnectedAgent{{AgentID: "other", Hostname: "other"}}}, + {"ambiguous short hostname", "node", "", []ConnectedAgent{{AgentID: "a", Hostname: "node.one"}, {AgentID: "b", Hostname: "node.two"}}}, + {"empty target", "", "node", []ConnectedAgent{{AgentID: "a", Hostname: "node"}}}, + } { + t.Run(tc.name, func(t *testing.T) { + e := &deferringGuestExecutor{stubExecutor: &stubExecutor{agents: tc.agents}, deferAt: 1} + scanner := NewDeepScanner(e) + if got, err := scanner.Scan(context.Background(), DiscoveryRequest{ResourceType: ResourceTypeVM, ResourceID: "105", TargetID: tc.target, Hostname: tc.hint}); got != nil || err == nil || !agentexec.IsGuestExecDeferred(err.Error()) || e.calls != 0 { + t.Fatalf("unverified node dispatched: %+v %v calls=%d", got, err, e.calls) + } + }) + } +} diff --git a/internal/servicediscovery/service.go b/internal/servicediscovery/service.go index fd3b7c0efc..d0efd46015 100644 --- a/internal/servicediscovery/service.go +++ b/internal/servicediscovery/service.go @@ -15,6 +15,7 @@ import ( "sync" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/ai/jsonresponse" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rs/zerolog/log" @@ -1883,6 +1884,12 @@ func (s *Service) DiscoverResource(ctx context.Context, req DiscoveryRequest) (* var scanError error if s.scanner != nil && s.IsCommandScanningEnabled() { scanResult, scanError = s.scanner.Scan(ctx, req) + if scanError != nil && agentexec.IsGuestExecDeferred(scanError.Error()) { + // Keep saved observations/notes and the original in-flight outcome; + // metadata fallback must not turn a paused live scan into success. + inProg.err = scanError + return nil, inProg.err + } if scanError != nil { log.Warn(). Err(scanError). diff --git a/internal/servicediscovery/service_test.go b/internal/servicediscovery/service_test.go index ed9074e507..274c9c5867 100644 --- a/internal/servicediscovery/service_test.go +++ b/internal/servicediscovery/service_test.go @@ -2760,3 +2760,7 @@ func TestService_BackfillPreservesConcurrentManualRepair(t *testing.T) { } } } + +func TestDiscoveryGuestSafetyStopsAndPreservesSavedEvidence(t *testing.T) { + testDiscoveryGuestSafetyStopsAndPreservesSavedEvidence(t) +}