diff --git a/cmd/pulse-agent/main.go b/cmd/pulse-agent/main.go index fe9025e67..1a5185623 100644 --- a/cmd/pulse-agent/main.go +++ b/cmd/pulse-agent/main.go @@ -25,6 +25,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" + "github.com/rcourtman/pulse-go-rewrite/internal/agenttarget" "github.com/rcourtman/pulse-go-rewrite/internal/agentupdate" "github.com/rcourtman/pulse-go-rewrite/internal/dockeragent" "github.com/rcourtman/pulse-go-rewrite/internal/hostagent" @@ -310,6 +311,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { logger.Info(). Str("version", Version). Str("pulse_url", cfg.PulseURL). + Int("observer_destinations", len(cfg.Observers)). Bool("host_enabled", cfg.EnableHost). Bool("docker_enabled", cfg.EnableDocker). Bool("kubernetes_enabled", cfg.EnableKubernetes). @@ -322,6 +324,13 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { Str("pulse_url", cfg.PulseURL). Msg("--allow-plaintext-http is set: the agent API token travels in cleartext to any non-loopback Pulse URL; only use this on a network you fully control") } + for _, observer := range cfg.Observers { + if observer.AllowPlaintextHTTP { + logger.Warn(). + Str("destination", observer.Name). + Msg("Observer destination explicitly permits plaintext HTTP; its API token may travel in cleartext") + } + } // 5. Set prometheus info metric agentInfo.WithLabelValues( @@ -395,6 +404,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { AppliedConfig: cfg.AppliedConfig, UpdateStatus: updater.Snapshot, ModuleStatus: runtimeStatus.moduleStatuses, + Observers: hostObserverTargets(cfg.Observers), DockerContainerUpdater: dockerUpdaterBridge, } @@ -439,6 +449,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { IncludeTasks: true, CollectDiskMetrics: false, DiskExclude: cfg.DiskExclude, + Targets: dockerReportTargets(cfg), } dockerAgent, err = newDockerAgent(dockerCfg) @@ -496,6 +507,7 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { IncludeAllPods: cfg.KubeIncludeAllPods, IncludeAllDeployments: cfg.KubeIncludeAllDeployments, MaxPods: cfg.KubeMaxPods, + Targets: kubernetesReportTargets(cfg), } agent, err := newKubeAgent(kubeCfg) @@ -776,6 +788,8 @@ type Config struct { AllowPlaintextHTTP bool CACertPath string ServerFingerprint string + ObserversFile string + Observers []agenttarget.Observer DeploySSHUser string LogLevel zerolog.Level LogFile string @@ -826,6 +840,76 @@ type Config struct { KubeMaxPods int } +func hostObserverTargets(observers []agenttarget.Observer) []hostagent.ObserverTarget { + targets := make([]hostagent.ObserverTarget, 0, len(observers)) + for _, observer := range observers { + targets = append(targets, hostagent.ObserverTarget{ + Name: observer.Name, + ID: observer.ID, + PulseURL: observer.URL, + APIToken: observer.Token, + InsecureSkipVerify: observer.InsecureSkipVerify, + AllowPlaintextHTTP: observer.AllowPlaintextHTTP, + CACertPath: observer.CACertPath, + ServerFingerprint: observer.ServerFingerprint, + ProvisionProxmox: observer.ProvisionProxmox, + }) + } + return targets +} + +func dockerReportTargets(cfg Config) []dockeragent.TargetConfig { + targets := make([]dockeragent.TargetConfig, 0, len(cfg.Observers)+1) + targets = append(targets, dockeragent.TargetConfig{ + Name: "primary", + URL: cfg.PulseURL, + Token: cfg.APIToken, + InsecureSkipVerify: cfg.InsecureSkipVerify, + AllowPlaintextHTTP: cfg.AllowPlaintextHTTP, + CACertPath: cfg.CACertPath, + ServerFingerprint: cfg.ServerFingerprint, + Authoritative: true, + }) + for _, observer := range cfg.Observers { + targets = append(targets, dockeragent.TargetConfig{ + Name: observer.Name, + URL: observer.URL, + Token: observer.Token, + InsecureSkipVerify: observer.InsecureSkipVerify, + AllowPlaintextHTTP: observer.AllowPlaintextHTTP, + CACertPath: observer.CACertPath, + ServerFingerprint: observer.ServerFingerprint, + }) + } + return targets +} + +func kubernetesReportTargets(cfg Config) []kubernetesagent.TargetConfig { + targets := make([]kubernetesagent.TargetConfig, 0, len(cfg.Observers)+1) + targets = append(targets, kubernetesagent.TargetConfig{ + Name: "primary", + URL: cfg.PulseURL, + Token: cfg.APIToken, + InsecureSkipVerify: cfg.InsecureSkipVerify, + AllowPlaintextHTTP: cfg.AllowPlaintextHTTP, + CACertPath: cfg.CACertPath, + ServerFingerprint: cfg.ServerFingerprint, + Authoritative: true, + }) + for _, observer := range cfg.Observers { + targets = append(targets, kubernetesagent.TargetConfig{ + Name: observer.Name, + URL: observer.URL, + Token: observer.Token, + InsecureSkipVerify: observer.InsecureSkipVerify, + AllowPlaintextHTTP: observer.AllowPlaintextHTTP, + CACertPath: observer.CACertPath, + ServerFingerprint: observer.ServerFingerprint, + }) + } + return targets +} + func loadConfig(args []string, getenv func(string) string) (Config, error) { // Environment Variables envURL := strings.TrimSpace(getenv("PULSE_URL")) @@ -867,6 +951,7 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) { envDiskExclude := strings.TrimSpace(getenv("PULSE_DISK_EXCLUDE")) envReportIP := strings.TrimSpace(getenv("PULSE_REPORT_IP")) envDisableCeph := strings.TrimSpace(getenv("PULSE_DISABLE_CEPH")) + envObserversFile := strings.TrimSpace(getenv("PULSE_OBSERVERS_FILE")) // Defaults defaultInterval := 30 * time.Second @@ -920,6 +1005,7 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) { allowPlaintextHTTPFlag := fs.Bool("allow-plaintext-http", utils.ParseBool(strings.TrimSpace(getenv("PULSE_AGENT_ALLOW_PLAINTEXT_HTTP"))), "Allow plain HTTP to a Pulse server that does not look local (sends the API token in cleartext; only for networks you fully control)") caCertFlag := fs.String("cacert", envCACertPath, "Path to custom CA bundle for agent HTTPS transport") serverFingerprintFlag := fs.String("server-fingerprint", envServerFingerprint, "Expected Pulse server TLS certificate fingerprint (SHA256)") + observersFileFlag := fs.String("observers-file", envObserversFile, "Absolute path to a private JSON file defining report-only Pulse observer destinations") deploySSHUserFlag := fs.String("deploy-ssh-user", envDeploySSHUser, "SSH user for peer deploy fan-out (default: root; non-root requires passwordless sudo)") logLevelFlag := fs.String("log-level", defaultLogLevel(envLogLevel), "Log level") logFileFlag := fs.String("log-file", envLogFile, "Write rotating JSON logs to this file") @@ -977,6 +1063,10 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) { // Resolve token with priority: --token > --token-file > env > default file token := resolveToken(*tokenFlag, *tokenFileFlag, envToken) + observers, err := agenttarget.LoadObservers(strings.TrimSpace(*observersFileFlag), pulseURL) + if err != nil { + return Config{}, fmt.Errorf("load observer destinations: %w", err) + } // When --enroll is set and a runtime token already exists from a previous // enrollment, use it instead of the bootstrap token embedded in the service @@ -1057,6 +1147,8 @@ func loadConfig(args []string, getenv func(string) string) (Config, error) { AllowPlaintextHTTP: *allowPlaintextHTTPFlag, CACertPath: strings.TrimSpace(*caCertFlag), ServerFingerprint: strings.TrimSpace(*serverFingerprintFlag), + ObserversFile: strings.TrimSpace(*observersFileFlag), + Observers: observers, DeploySSHUser: deploySSHUser, LogLevel: logLevel, LogFile: strings.TrimSpace(*logFileFlag), diff --git a/cmd/pulse-agent/observers_config_test.go b/cmd/pulse-agent/observers_config_test.go new file mode 100644 index 000000000..1844df48f --- /dev/null +++ b/cmd/pulse-agent/observers_config_test.go @@ -0,0 +1,40 @@ +package main + +import ( + "os" + "path/filepath" + "testing" +) + +func TestLoadConfigBuildsReportOnlyObserverTargets(t *testing.T) { + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + if err := os.WriteFile(tokenPath, []byte("observer-token"), 0o600); err != nil { + t.Fatal(err) + } + configPath := filepath.Join(dir, "observers.json") + config := `{"version":1,"observers":[{"name":"dev","url":"http://127.0.0.1:7656","tokenFile":"` + tokenPath + `"}]}` + if err := os.WriteFile(configPath, []byte(config), 0o600); err != nil { + t.Fatal(err) + } + + cfg, err := loadConfig([]string{ + "--url", "http://127.0.0.1:7655", + "--token", "primary-token", + "--observers-file", configPath, + }, func(string) string { return "" }) + if err != nil { + t.Fatalf("loadConfig: %v", err) + } + if len(cfg.Observers) != 1 || cfg.Observers[0].Token != "observer-token" { + t.Fatalf("observers = %+v", cfg.Observers) + } + dockerTargets := dockerReportTargets(cfg) + if len(dockerTargets) != 2 || !dockerTargets[0].Authoritative || dockerTargets[1].Authoritative { + t.Fatalf("docker targets = %+v", dockerTargets) + } + kubeTargets := kubernetesReportTargets(cfg) + if len(kubeTargets) != 2 || !kubeTargets[0].Authoritative || kubeTargets[1].Authoritative { + t.Fatalf("kubernetes targets = %+v", kubeTargets) + } +} diff --git a/cmd/pulse-agent/service_windows.go b/cmd/pulse-agent/service_windows.go index 16cef61aa..8a6f89024 100644 --- a/cmd/pulse-agent/service_windows.go +++ b/cmd/pulse-agent/service_windows.go @@ -102,6 +102,7 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha AppliedConfig: ws.cfg.AppliedConfig, UpdateStatus: updater.Snapshot, ModuleStatus: runtimeStatus.moduleStatuses, + Observers: hostObserverTargets(ws.cfg.Observers), } agent, err := hostagent.New(hostCfg) if err != nil { @@ -146,6 +147,7 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha IncludeServices: true, IncludeTasks: true, CollectDiskMetrics: true, + Targets: dockerReportTargets(ws.cfg), } agent, err := dockeragent.New(dockerCfg) @@ -191,6 +193,7 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha IncludeAllPods: ws.cfg.KubeIncludeAllPods, IncludeAllDeployments: ws.cfg.KubeIncludeAllDeployments, MaxPods: ws.cfg.KubeMaxPods, + Targets: kubernetesReportTargets(ws.cfg), } agent, err := kubernetesagent.New(kubeCfg) if err != nil { diff --git a/docs/UNIFIED_AGENT.md b/docs/UNIFIED_AGENT.md index 907cefe9f..6392ea5a5 100644 --- a/docs/UNIFIED_AGENT.md +++ b/docs/UNIFIED_AGENT.md @@ -31,11 +31,13 @@ Run it on the host that already has the v5 `pulse-agent` service to replace the binary and service configuration in place; do not uninstall the old service first unless you are intentionally removing that host from Pulse. -An installed agent is configured for one Pulse URL and one token. Do not point -one running service at both a v5 server and a v6 server. After the upgrade, -check the relevant platform page or **Machines** view once the agent has -reported, and confirm the host-local version with `pulse-agent --version` if -the UI has not received a fresh report yet. +An installed agent has one **primary** Pulse URL and token. The primary is the +only server allowed to supply remote configuration, commands, enrollment, or +updates. The same collection can also be sent to explicitly configured, +report-only **observer** instances; see [Observer destinations](#observer-destinations). +After an upgrade, check the relevant platform page or **Machines** view once +the agent has reported, and confirm the host-local version with +`pulse-agent --version` if the UI has not received a fresh report yet. ### Linux (systemd) ```bash @@ -88,6 +90,7 @@ curl -fsSL http://:7655/install.sh | \ |------|---------|-------------|---------| | `--url` | `PULSE_URL` | Pulse server URL | `http://localhost:7655` | | `--token` | `PULSE_TOKEN` | API token | *(required)* | +| `--observers-file` | `PULSE_OBSERVERS_FILE` | Private JSON file defining report-only destinations | *(none)* | | `--token-file` | - | Read API token from file | *(unset)* | | `--interval` | `PULSE_INTERVAL` | Reporting interval | `30s` | | `--enable-host` | `PULSE_ENABLE_HOST` | Enable host metrics | `true` | @@ -124,6 +127,52 @@ health/metrics endpoint over the network. Use `--health-addr ""` or **Token resolution order**: `--token` → `--token-file` → `PULSE_TOKEN` → `/var/lib/pulse-agent/token`. +## Observer destinations + +Observer destinations receive the same already-collected host, Docker/Podman, +and Kubernetes reports. Collection runs once per interval. Delivery, retries, +and persisted host-report buffers are isolated per destination, so an observer +outage does not replay or block the primary stream. Observer responses cannot +change configuration, execute commands, enroll the agent, or select updates. + +Create a separate API token on each observer and store every token in its own +absolute-path file. On Unix, both the JSON file and token files must be regular, +non-symlink files with no group or other permissions (for example mode `0600`). + +```json +{ + "version": 1, + "observers": [ + { + "name": "dev", + "url": "https://pulse-dev.example.test", + "tokenFile": "/etc/pulse-agent/dev-observer.token", + "serverFingerprint": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "provisionProxmox": true + } + ] +} +``` + +Start or install the service with +`--observers-file /etc/pulse-agent/observers.json`. Plaintext remote HTTP is +rejected unless that observer explicitly sets `"allowPlaintextHTTP": true`. +`insecureSkipVerify` is available per observer but should be replaced with a +CA file or certificate fingerprint wherever possible. + +When Proxmox integration is enabled, each observer gets a distinct +destination-scoped PVE/PBS API token and registration-state directory. Pulse +must answer the registration check before the agent creates or rotates any +Proxmox token; an unavailable destination therefore leaves existing +credentials unchanged. Set `"provisionProxmox": false` when an observer should +receive only Unified Agent telemetry and no separately registered PVE/PBS +source. + +Per-destination delivery status is exported on the health listener as +`pulse_agent_destination_configured` and +`pulse_agent_destination_delivery_up`, labelled by module, destination, and +role. + ### Advanced Flags - `--version`: Print the agent version and exit. diff --git a/docs/UPGRADE_v6.md b/docs/UPGRADE_v6.md index 0981c6664..225962423 100644 --- a/docs/UPGRADE_v6.md +++ b/docs/UPGRADE_v6.md @@ -123,14 +123,16 @@ pulse-agent --version systemctl status pulse-agent ``` -### Can one installed Pulse Unified Agent report to both a Pulse v5 instance and a Pulse v6 instance at the same time? +### Can one installed Pulse Unified Agent report to two Pulse instances at the same time? -Not as a supported in-place setup. A running Unified Agent installation is -configured against one Pulse URL and one token, and it fetches remote config -from that one Pulse server. If you need side-by-side evaluation, use a -separate test host or VM, a cloned lab machine, or a separate isolated agent -installation instead of trying to point one running agent service at two Pulse -servers. +Yes. Configure one instance as the primary with `--url` and its token, then add +the other as a report-only observer with `--observers-file`. Only the primary +can supply remote configuration, commands, enrollment, or updates; observer +delivery and retries are isolated. Each instance needs its own Pulse API token, +and Proxmox observers use separate PVE/PBS tokens. See +[Observer destinations](UNIFIED_AGENT.md#observer-destinations) for the file +format and security requirements. Use a v6-capable agent for this topology; +older v5 agents do not understand observer configuration. ### Can I keep Pulse v5 stable while I test Pulse v6? diff --git a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md index a06c958db..72677ee31 100644 --- a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md +++ b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md @@ -1,6 +1,6 @@ # Pulse v6 Source Of Truth -Last updated: 2026-07-14 +Last updated: 2026-07-19 Status: ACTIVE This file is the stable human governance layer for the active v6 release diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index 9e47db896..bb16fca1f 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -1,6 +1,6 @@ { "version": "6.0", - "updated_at": "2026-07-14", + "updated_at": "2026-07-19", "scope": { "active_repos": [ "pulse", @@ -80,6 +80,21 @@ } ], "evidence": [ + { + "repo": "pulse", + "path": "cmd/pulse-agent/main.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "cmd/pulse-agent/observers_config_test.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "cmd/pulse-agent/service_windows.go", + "kind": "file" + }, { "repo": "pulse", "path": "frontend-modern/src/components/Infrastructure/__tests__/UnifiedResourceTable.workloads-link.test.tsx", @@ -5927,7 +5942,7 @@ "status": "target-met", "completion": { "state": "complete", - "summary": "Agent lifecycle and fleet operations are at the current governed RC floor: canonical auto-register now converges on one v6 contract, install-command consumers fail closed through shared validated boundaries, profile-management surfaces preserve the canonical malformed-payload and missing-profile resync contract across settings and deploy surfaces, and the runtime-side Unified Agent reporting path now uses the canonical product terminology.", + "summary": "Agent lifecycle and fleet operations are at the current governed RC floor: canonical auto-register converges on one v6 contract, install-command consumers fail closed through shared validated boundaries, and one Unified Agent can fan a single collection to one authoritative primary plus isolated report-only observers. Observer tokens, TLS policy, retry buffers, delivery metrics, and Proxmox credentials are destination-scoped; Proxmox token mutation is gated on a successful destination registration check.", "tracking": [] }, "blockers": [], @@ -6085,6 +6100,16 @@ "path": "frontend-modern/src/utils/nodeModalPresentation.ts", "kind": "file" }, + { + "repo": "pulse", + "path": "internal/agenttarget/config.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "internal/agenttarget/config_test.go", + "kind": "file" + }, { "repo": "pulse", "path": "internal/agentupdate/update.go", @@ -6120,11 +6145,21 @@ "path": "internal/api/unified_agent.go", "kind": "file" }, + { + "repo": "pulse", + "path": "internal/dockeragent/multi_target_test.go", + "kind": "file" + }, { "repo": "pulse", "path": "internal/hostagent/agent.go", "kind": "file" }, + { + "repo": "pulse", + "path": "internal/hostagent/observer_delivery_test.go", + "kind": "file" + }, { "repo": "pulse", "path": "internal/hostagent/proxmox_setup.go", @@ -6135,6 +6170,16 @@ "path": "internal/hostagent/proxmox_setup_test.go", "kind": "file" }, + { + "repo": "pulse", + "path": "internal/kubernetesagent/multi_target_test.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "scripts/install.ps1", + "kind": "file" + }, { "repo": "pulse", "path": "scripts/install.sh", diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index fd4c092f2..fb426d07e 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -4642,6 +4642,13 @@ file. Proxmox registration is also destination-scoped: the primary retains its legacy token name for upgrade continuity, observers use distinct token names and state markers, and every setup path must obtain a successful registration state response before any create, delete, or rotation command is executed. +Plaintext transport consent is destination-scoped too. The primary keeps the +existing local-network compatibility and explicit process-level override, but +an observer must opt in independently for every non-loopback HTTP URL; primary +consent cannot silently widen an observer. That policy must survive the shared +config loader, host/Docker/Kubernetes target normalization, and observer +Proxmox registration path. Unix and Windows service installation must preserve +the absolute observer-config path across an in-place update. The adjacent recovery handlers under `internal/api/` do not widen this agent lifecycle boundary. Protection posture is a read-only `monitoring:read` diff --git a/docs/release-control/v6/internal/subsystems/deployment-installability.md b/docs/release-control/v6/internal/subsystems/deployment-installability.md index 5e1b6f58a..d22913585 100644 --- a/docs/release-control/v6/internal/subsystems/deployment-installability.md +++ b/docs/release-control/v6/internal/subsystems/deployment-installability.md @@ -2443,3 +2443,13 @@ on community and Pro binaries. The rollback tests in `internal/updates/manager_rollback_test.go`, the rollback handler tests in `internal/api/updates_test.go`, and the route inventory pin for `/api/updates/rollback` are the proof surface for this path. + +### Observer destination installation continuity + +Unix and Windows installers accept `--observers-file` and preserve the absolute +path in the installed service command. Unix installation rejects relative, +missing, and symlink configuration paths before service mutation. The runtime +remains the final schema, permission, token-file, URL, and TLS-policy validator. +Updates recover the observer-file argument from the existing service command so +an in-place binary refresh does not silently collapse a multi-destination +installation back to primary-only reporting. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 2e37b9eab..be5cd673d 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1750,6 +1750,12 @@ The local agent health listener exports to `primary` or `observer`; destination names come from validated configuration. Observer delivery failure is visible but does not make the primary authority unready or merge observer retry state into primary delivery health. +Host, Docker/Podman, and Kubernetes reporters each fan out the already-collected +snapshot without triggering a second collection. Their retry queues and latest +delivery gauges remain per destination, and Kubernetes observer transport uses +the same destination-scoped TLS and explicit plaintext policy as the host and +Docker reporters. Observer acknowledgements never change the canonical +monitoring configuration returned by the primary. ### PBS protection evidence collection diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index 9348833ce..397aa578b 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -1101,6 +1101,7 @@ "contract": "docs/release-control/v6/internal/subsystems/agent-lifecycle.md", "owned_prefixes": [ "internal/agentexec/", + "internal/agenttarget/", "internal/agentupdate/", "internal/hostagent/", "scripts/intelligence_lab/" @@ -1108,6 +1109,7 @@ "owned_files": [ ".github/workflows/unified-agent-native.yml", "cmd/pulse-agent/main.go", + "cmd/pulse-agent/service_windows.go", "frontend-modern/src/api/agentProfiles.ts", "frontend-modern/src/api/nodes.ts", "frontend-modern/src/components/Settings/agentProfileSettings.ts", @@ -1215,12 +1217,14 @@ "label": "pulse-agent CLI entrypoint proof", "match_prefixes": [], "match_files": [ - "cmd/pulse-agent/main.go" + "cmd/pulse-agent/main.go", + "cmd/pulse-agent/service_windows.go" ], "allow_same_subsystem_tests": false, "test_prefixes": [], "exact_files": [ - "cmd/pulse-agent/main_test.go" + "cmd/pulse-agent/main_test.go", + "cmd/pulse-agent/observers_config_test.go" ] }, { @@ -1263,7 +1267,9 @@ { "id": "agent-runtime-transport-trust", "label": "agent runtime transport trust proof", - "match_prefixes": [], + "match_prefixes": [ + "internal/agenttarget/" + ], "match_files": [ "internal/agenttls/config.go", "internal/dockeragent/agent.go", @@ -1275,9 +1281,12 @@ "allow_same_subsystem_tests": false, "test_prefixes": [], "exact_files": [ + "internal/agenttarget/config_test.go", "internal/agenttls/config_test.go", "internal/dockeragent/agent_internal_test.go", + "internal/dockeragent/multi_target_test.go", "internal/kubernetesagent/agent_new_test.go", + "internal/kubernetesagent/multi_target_test.go", "internal/remoteconfig/client_additional_test.go", "internal/remoteconfig/client_test.go", "internal/securityutil/httpurl_test.go" @@ -1317,6 +1326,7 @@ "internal/hostagent/commands_host_update_test.go", "internal/hostagent/commands_storage_cleanup_test.go", "internal/hostagent/docker_lifecycle_test.go", + "internal/hostagent/observer_delivery_test.go", "internal/hostagent/package_updates_test.go", "internal/hostagent/send_report_test.go", "internal/hostagent/storage_cleanup_test.go" @@ -5078,6 +5088,7 @@ "exact_files": [ "internal/kubernetesagent/agent_inventory_test.go", "internal/kubernetesagent/agent_new_test.go", + "internal/kubernetesagent/multi_target_test.go", "internal/monitoring/kubernetes_agents_test.go", "internal/unifiedresources/adapter_coverage_test.go", "internal/unifiedresources/kubernetes_registry_test.go" diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index 716359def..02a1b344e 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -1434,6 +1434,10 @@ enforce TLS, CA, fingerprint, and explicit plaintext policy. Observer payload responses are report acknowledgements only and cannot authorize configuration, commands, enrollment, or updates. Per-destination Proxmox tokens prevent one Pulse instance from rotating credentials used by another. +The process-wide plaintext override belongs only to the authoritative primary: +it cannot satisfy an observer's opt-in. Every non-loopback HTTP observer must +declare its own `allowPlaintextHTTP` consent, and that decision must propagate +unchanged through host, Docker, Kubernetes, and observer Proxmox transports. Operational Trust action offers enforce current plan, approve, and execute authority before a mutating affordance is returned. Planning repeats those diff --git a/internal/agenttarget/config.go b/internal/agenttarget/config.go new file mode 100644 index 000000000..95f99b895 --- /dev/null +++ b/internal/agenttarget/config.go @@ -0,0 +1,242 @@ +package agenttarget + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/url" + "os" + "path/filepath" + "runtime" + "strings" + "unicode" + + "github.com/rcourtman/pulse-go-rewrite/pkg/securityutil" +) + +const ConfigVersion = 1 + +const maxConfigBytes = 256 * 1024 +const maxObservers = 16 +const maxTokenBytes = 64 * 1024 + +// Observer is a report-only Pulse destination. The primary Pulse URL remains +// the sole authority for remote configuration, commands, and agent updates. +type Observer struct { + Name string + URL string + Token string + InsecureSkipVerify bool + AllowPlaintextHTTP bool + CACertPath string + ServerFingerprint string + ProvisionProxmox bool + ID string +} + +type observerFile struct { + Version int `json:"version"` + Observers []observerFileTarget `json:"observers"` +} + +type observerFileTarget struct { + Name string `json:"name"` + URL string `json:"url"` + TokenFile string `json:"tokenFile"` + InsecureSkipVerify bool `json:"insecureSkipVerify,omitempty"` + AllowPlaintextHTTP bool `json:"allowPlaintextHTTP,omitempty"` + CACertPath string `json:"caCertPath,omitempty"` + ServerFingerprint string `json:"serverFingerprint,omitempty"` + ProvisionProxmox *bool `json:"provisionProxmox,omitempty"` +} + +// LoadObservers validates and resolves an owner-private observer configuration. +// Raw tokens are never accepted in the JSON document; each destination must +// reference a separately permissioned token file. +func LoadObservers(path string, primaryURL string) ([]Observer, error) { + path = strings.TrimSpace(path) + if path == "" { + return nil, nil + } + if !filepath.IsAbs(path) { + return nil, errors.New("observer config path must be absolute") + } + if err := requirePrivateRegularFile(path, "observer config", maxConfigBytes); err != nil { + return nil, err + } + + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("open observer config: %w", err) + } + defer f.Close() + + dec := json.NewDecoder(io.LimitReader(f, maxConfigBytes+1)) + dec.DisallowUnknownFields() + var parsed observerFile + if err := dec.Decode(&parsed); err != nil { + return nil, fmt.Errorf("decode observer config: %w", err) + } + if err := ensureJSONEOF(dec); err != nil { + return nil, err + } + if parsed.Version != ConfigVersion { + return nil, fmt.Errorf("observer config version %d is unsupported; expected %d", parsed.Version, ConfigVersion) + } + if len(parsed.Observers) > maxObservers { + return nil, fmt.Errorf("observer config has %d destinations; maximum is %d", len(parsed.Observers), maxObservers) + } + + normalizedPrimary := "" + if strings.TrimSpace(primaryURL) != "" { + primary, err := NormalizePulseURL(primaryURL, true, true) + if err != nil { + return nil, fmt.Errorf("normalize primary Pulse URL: %w", err) + } + normalizedPrimary = primary + } + + result := make([]Observer, 0, len(parsed.Observers)) + seenNames := make(map[string]struct{}, len(parsed.Observers)) + seenURLs := make(map[string]struct{}, len(parsed.Observers)) + for index, item := range parsed.Observers { + observer, err := resolveObserver(item) + if err != nil { + return nil, fmt.Errorf("observer %d: %w", index+1, err) + } + nameKey := strings.ToLower(observer.Name) + if _, exists := seenNames[nameKey]; exists { + return nil, fmt.Errorf("observer %d: duplicate name %q", index+1, observer.Name) + } + if _, exists := seenURLs[observer.URL]; exists { + return nil, fmt.Errorf("observer %d: duplicate URL", index+1) + } + if normalizedPrimary != "" && observer.URL == normalizedPrimary { + return nil, fmt.Errorf("observer %d: URL duplicates the primary Pulse destination", index+1) + } + seenNames[nameKey] = struct{}{} + seenURLs[observer.URL] = struct{}{} + result = append(result, observer) + } + return result, nil +} + +func resolveObserver(item observerFileTarget) (Observer, error) { + name := strings.TrimSpace(item.Name) + if err := validateName(name); err != nil { + return Observer{}, err + } + url, err := NormalizePulseURL(item.URL, false, item.AllowPlaintextHTTP) + if err != nil { + return Observer{}, fmt.Errorf("invalid URL: %w", err) + } + + tokenPath := strings.TrimSpace(item.TokenFile) + if !filepath.IsAbs(tokenPath) { + return Observer{}, errors.New("tokenFile must be an absolute path") + } + if err := requirePrivateRegularFile(tokenPath, "observer token", maxTokenBytes); err != nil { + return Observer{}, err + } + tokenBytes, err := os.ReadFile(tokenPath) + if err != nil { + return Observer{}, fmt.Errorf("read observer token: %w", err) + } + token := strings.TrimSpace(string(tokenBytes)) + if token == "" { + return Observer{}, errors.New("observer token file is empty") + } + if strings.IndexFunc(token, unicode.IsSpace) >= 0 { + return Observer{}, errors.New("observer token contains whitespace") + } + + provisionProxmox := true + if item.ProvisionProxmox != nil { + provisionProxmox = *item.ProvisionProxmox + } + return Observer{ + Name: name, + URL: url, + Token: token, + InsecureSkipVerify: item.InsecureSkipVerify, + AllowPlaintextHTTP: item.AllowPlaintextHTTP, + CACertPath: strings.TrimSpace(item.CACertPath), + ServerFingerprint: strings.TrimSpace(item.ServerFingerprint), + ProvisionProxmox: provisionProxmox, + ID: stableID(url), + }, nil +} + +// NormalizePulseURL applies the transport policy for one report destination. +// Authoritative primary destinations retain the established self-hosted +// local-network HTTP allowance. Report-only observers require an explicit +// per-destination opt-in for every non-loopback plaintext URL, even when the +// process-wide primary override is enabled. +func NormalizePulseURL(raw string, authoritative bool, allowPlaintext bool) (string, error) { + if !authoritative && !allowPlaintext { + parsed, err := url.Parse(strings.TrimSpace(raw)) + if err == nil && parsed.Hostname() != "" && strings.EqualFold(parsed.Scheme, "http") && + !securityutil.IsLoopbackHost(parsed.Hostname()) { + return "", fmt.Errorf("Pulse URL %q must use https unless the observer explicitly allows plaintext HTTP", raw) + } + } + parsed, err := securityutil.NormalizePulseHTTPBaseURLWithOptions(raw, securityutil.PulseURLValidationOptions{ + AllowLocalNetworkHTTP: authoritative, + AllowOperatorPlaintextHTTP: allowPlaintext, + }) + if err != nil { + return "", err + } + return strings.TrimRight(parsed.String(), "/"), nil +} + +func validateName(name string) error { + if name == "" { + return errors.New("name is required") + } + if len(name) > 64 { + return errors.New("name must be 64 characters or fewer") + } + for _, r := range name { + if unicode.IsLetter(r) || unicode.IsDigit(r) || r == '-' || r == '_' || r == '.' { + continue + } + return fmt.Errorf("name %q contains unsupported characters", name) + } + return nil +} + +func requirePrivateRegularFile(path string, label string, maxBytes int64) error { + info, err := os.Lstat(path) + if err != nil { + return fmt.Errorf("stat %s file: %w", label, err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() { + return fmt.Errorf("%s file must be a regular file, not a symlink", label) + } + if info.Size() > maxBytes { + return fmt.Errorf("%s file exceeds the %d-byte size limit", label, maxBytes) + } + if runtime.GOOS != "windows" && info.Mode().Perm()&0o077 != 0 { + return fmt.Errorf("%s file permissions must not grant group or other access", label) + } + return nil +} + +func ensureJSONEOF(dec *json.Decoder) error { + var extra any + if err := dec.Decode(&extra); err == io.EOF { + return nil + } else if err != nil { + return fmt.Errorf("decode observer config trailer: %w", err) + } + return errors.New("observer config contains multiple JSON values") +} + +func stableID(url string) string { + sum := sha256.Sum256([]byte(strings.TrimSpace(url))) + return hex.EncodeToString(sum[:8]) +} diff --git a/internal/agenttarget/config_test.go b/internal/agenttarget/config_test.go new file mode 100644 index 000000000..fb1d2450f --- /dev/null +++ b/internal/agenttarget/config_test.go @@ -0,0 +1,163 @@ +package agenttarget + +import ( + "os" + "path/filepath" + "runtime" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/pkg/securityutil" +) + +func writePrivateFile(t *testing.T, path string, content string) { + t.Helper() + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatal(err) + } +} + +func TestLoadObservers(t *testing.T) { + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret\n") + writePrivateFile(t, configPath, `{ + "version": 1, + "observers": [{ + "name": "dev", + "url": "http://192.168.0.10:7655/", + "tokenFile": "`+tokenPath+`", + "allowPlaintextHTTP": true, + "provisionProxmox": true + }] +}`) + + observers, err := LoadObservers(configPath, "https://prod.example.com") + if err != nil { + t.Fatal(err) + } + if len(observers) != 1 { + t.Fatalf("len(observers) = %d, want 1", len(observers)) + } + got := observers[0] + if got.Name != "dev" || got.URL != "http://192.168.0.10:7655" || got.Token != "observer-secret" || !got.ProvisionProxmox || got.ID == "" { + t.Fatalf("observer = %+v", got) + } +} + +func TestLoadObserversRejectsPrimaryDuplicate(t *testing.T) { + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret") + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://pulse.example.com/","tokenFile":"`+tokenPath+`"}]}`) + + if _, err := LoadObservers(configPath, "https://pulse.example.com"); err == nil { + t.Fatal("expected duplicate primary URL rejection") + } +} + +func TestLoadObserversRejectsInlineOrLooseSecrets(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("POSIX permission check") + } + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret") + if err := os.Chmod(tokenPath, 0o644); err != nil { + t.Fatal(err) + } + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","tokenFile":"`+tokenPath+`"}]}`) + + if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil { + t.Fatal("expected loose token permission rejection") + } +} + +func TestLoadObserversRejectsInlineTokenField(t *testing.T) { + dir := t.TempDir() + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","token":"inline-secret"}]}`) + if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil { + t.Fatal("expected inline token field rejection") + } +} + +func TestLoadObserversRejectsSymlinkedTokenFile(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink creation requires platform-specific privileges") + } + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + linkPath := filepath.Join(dir, "dev-link.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret") + if err := os.Symlink(tokenPath, linkPath); err != nil { + t.Fatal(err) + } + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","tokenFile":"`+linkPath+`"}]}`) + if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil { + t.Fatal("expected symlinked token file rejection") + } +} + +func TestLoadObserversRequiresExplicitPublicPlaintextOptIn(t *testing.T) { + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret") + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://203.0.113.10:7655","tokenFile":"`+tokenPath+`"}]}`) + if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil { + t.Fatal("expected public plaintext URL rejection") + } + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://203.0.113.10:7655","tokenFile":"`+tokenPath+`","allowPlaintextHTTP":true}]}`) + if _, err := LoadObservers(configPath, "https://prod.example.com"); err != nil { + t.Fatalf("explicit plaintext opt-in: %v", err) + } +} + +func TestLoadObserversRequiresExplicitPrivatePlaintextOptIn(t *testing.T) { + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret") + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://192.168.50.10:7655","tokenFile":"`+tokenPath+`"}]}`) + if _, err := LoadObservers(configPath, "https://prod.example.com"); err == nil { + t.Fatal("expected private-network plaintext URL rejection without observer opt-in") + } + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"http://192.168.50.10:7655","tokenFile":"`+tokenPath+`","allowPlaintextHTTP":true}]}`) + if _, err := LoadObservers(configPath, "https://prod.example.com"); err != nil { + t.Fatalf("explicit private-network plaintext opt-in: %v", err) + } +} + +func TestObserverPlaintextPolicyDoesNotInheritPrimaryProcessConsent(t *testing.T) { + securityutil.SetOperatorPlaintextHTTPConsent(true) + defer securityutil.SetOperatorPlaintextHTTPConsent(false) + + if _, err := NormalizePulseURL("http://203.0.113.10:7655", false, false); err == nil { + t.Fatal("observer without explicit opt-in inherited process-wide primary plaintext consent") + } + if got, err := NormalizePulseURL("http://203.0.113.10:7655", false, true); err != nil { + t.Fatalf("explicit observer plaintext opt-in: %v", err) + } else if got != "http://203.0.113.10:7655" { + t.Fatalf("normalized URL = %q", got) + } +} + +func TestLoadObserversDefaultsProxmoxProvisioningOn(t *testing.T) { + dir := t.TempDir() + tokenPath := filepath.Join(dir, "dev.token") + configPath := filepath.Join(dir, "observers.json") + writePrivateFile(t, tokenPath, "observer-secret") + writePrivateFile(t, configPath, `{"version":1,"observers":[{"name":"dev","url":"https://dev.example.com","tokenFile":"`+tokenPath+`"}]}`) + + observers, err := LoadObservers(configPath, "https://prod.example.com") + if err != nil { + t.Fatal(err) + } + if !observers[0].ProvisionProxmox { + t.Fatal("expected Proxmox provisioning to default on") + } +} diff --git a/internal/agenttarget/metrics.go b/internal/agenttarget/metrics.go new file mode 100644 index 000000000..b190017f4 --- /dev/null +++ b/internal/agenttarget/metrics.go @@ -0,0 +1,29 @@ +package agenttarget + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var ( + destinationConfigured = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "pulse_agent_destination_configured", + Help: "Configured Pulse report destinations by module and authority role.", + }, []string{"module", "destination", "role"}) + destinationDeliveryUp = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "pulse_agent_destination_delivery_up", + Help: "Whether the most recent report delivery to a Pulse destination succeeded.", + }, []string{"module", "destination", "role"}) +) + +func MarkConfigured(module, destination, role string) { + destinationConfigured.WithLabelValues(module, destination, role).Set(1) +} + +func MarkDelivery(module, destination, role string, success bool) { + value := 0.0 + if success { + value = 1 + } + destinationDeliveryUp.WithLabelValues(module, destination, role).Set(value) +} diff --git a/internal/dockeragent/agent.go b/internal/dockeragent/agent.go index 68279c74a..62759e7db 100644 --- a/internal/dockeragent/agent.go +++ b/internal/dockeragent/agent.go @@ -15,8 +15,8 @@ import ( systemtypes "github.com/moby/moby/api/types/system" "github.com/moby/moby/client" + "github.com/rcourtman/pulse-go-rewrite/internal/agenttarget" "github.com/rcourtman/pulse-go-rewrite/internal/agenttls" - "github.com/rcourtman/pulse-go-rewrite/internal/securityutil" "github.com/rcourtman/pulse-go-rewrite/internal/utils" agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker" "github.com/rs/zerolog" @@ -24,11 +24,14 @@ import ( // TargetConfig describes a single Pulse backend the agent should report to. type TargetConfig struct { + Name string URL string Token string InsecureSkipVerify bool + AllowPlaintextHTTP bool CACertPath string ServerFingerprint string + Authoritative bool } // Config describes runtime configuration for the Docker / Podman collection module. @@ -124,6 +127,7 @@ type Agent struct { prevContainerCPU map[string]cpuSample cpuMu sync.Mutex // protects prevContainerCPU reportBuffer *utils.Queue[agentsdocker.Report] + reportBuffers map[string]*utils.Queue[agentsdocker.Report] registryChecker *RegistryChecker // For checking container image updates collectMu sync.Mutex // serializes collectOnce calls backgroundMu sync.Mutex // protects updateCheckRunning, cleanupTaskRunning @@ -162,11 +166,13 @@ func New(cfg Config) (*Agent, error) { } targets, err = normalizeTargetsFn([]TargetConfig{{ + Name: "primary", URL: url, Token: token, InsecureSkipVerify: cfg.InsecureSkipVerify, CACertPath: cfg.CACertPath, ServerFingerprint: cfg.ServerFingerprint, + Authoritative: true, }}) if err != nil { return nil, fmt.Errorf("dockeragent.New: normalize fallback target: %w", err) @@ -242,6 +248,11 @@ func New(cfg Config) (*Agent, error) { hasSecure := false hasInsecure := false for _, target := range cfg.Targets { + role := "observer" + if target.Authoritative { + role = "primary" + } + agenttarget.MarkConfigured("docker", target.Name, role) if target.InsecureSkipVerify { hasInsecure = true } else { @@ -288,6 +299,20 @@ func New(cfg Config) (*Agent, error) { const bufferCapacity = 60 + reportBuffers := make(map[string]*utils.Queue[agentsdocker.Report], len(cfg.Targets)) + for _, target := range cfg.Targets { + reportBuffers[target.Name] = utils.New[agentsdocker.Report](bufferCapacity) + } + primaryBuffer := reportBuffers["primary"] + if primaryBuffer == nil { + for _, target := range cfg.Targets { + if target.Authoritative { + primaryBuffer = reportBuffers[target.Name] + break + } + } + } + agent := &Agent{ cfg: cfg, docker: dockerClient, @@ -306,7 +331,8 @@ func New(cfg Config) (*Agent, error) { allowedStates: make(map[string]struct{}, len(stateFilters)), stateFilters: stateFilters, prevContainerCPU: make(map[string]cpuSample), - reportBuffer: utils.New[agentsdocker.Report](bufferCapacity), + reportBuffer: primaryBuffer, + reportBuffers: reportBuffers, registryChecker: newRegistryCheckerWithConfig(*logger, !cfg.DisableUpdateChecks), } @@ -326,8 +352,10 @@ func normalizeTargets(raw []TargetConfig) ([]TargetConfig, error) { normalized := make([]TargetConfig, 0, len(raw)) seen := make(map[string]struct{}, len(raw)) + seenNames := make(map[string]struct{}, len(raw)) - for _, target := range raw { + authoritativeCount := 0 + for index, target := range raw { targetURL := strings.TrimSpace(target.URL) token := strings.TrimSpace(target.Token) if targetURL == "" && token == "" { @@ -341,40 +369,67 @@ func normalizeTargets(raw []TargetConfig) ([]TargetConfig, error) { return nil, fmt.Errorf("pulse target %s is missing API token", targetURL) } - normalizedURL, err := normalizeTargetURL(targetURL) + if len(normalized) == 0 && !target.Authoritative { + target.Authoritative = true + } + normalizedURL, err := normalizeTargetURLWithPolicy( + targetURL, + target.Authoritative, + target.AllowPlaintextHTTP, + ) if err != nil { return nil, fmt.Errorf("invalid pulse target URL %q: %w", targetURL, err) } caCertPath := strings.TrimSpace(target.CACertPath) serverFingerprint := strings.TrimSpace(target.ServerFingerprint) - key := fmt.Sprintf("%s|%s|%t|%s|%s", normalizedURL, token, target.InsecureSkipVerify, caCertPath, serverFingerprint) + name := strings.TrimSpace(target.Name) + if name == "" { + if len(normalized) == 0 { + name = "primary" + } else { + name = fmt.Sprintf("observer-%d", index) + } + } + if _, exists := seenNames[name]; exists { + return nil, fmt.Errorf("duplicate Pulse target name %q", name) + } + key := normalizedURL if _, exists := seen[key]; exists { continue } + if target.Authoritative { + authoritativeCount++ + } seen[key] = struct{}{} + seenNames[name] = struct{}{} normalized = append(normalized, TargetConfig{ + Name: name, URL: normalizedURL, Token: token, InsecureSkipVerify: target.InsecureSkipVerify, + AllowPlaintextHTTP: target.AllowPlaintextHTTP, CACertPath: caCertPath, ServerFingerprint: serverFingerprint, + Authoritative: target.Authoritative, }) } - + if len(normalized) > 0 && authoritativeCount != 1 { + return nil, fmt.Errorf("exactly one authoritative Pulse target is required (got %d)", authoritativeCount) + } return normalized, nil } func normalizeTargetURL(raw string) (string, error) { - parsed, err := securityutil.NormalizePulseHTTPBaseURLWithOptions(raw, securityutil.PulseURLValidationOptions{ - AllowLocalNetworkHTTP: true, - }) + return normalizeTargetURLWithPolicy(raw, true, false) +} + +func normalizeTargetURLWithPolicy(raw string, authoritative bool, allowPlaintext bool) (string, error) { + normalized, err := agenttarget.NormalizePulseURL(raw, authoritative, allowPlaintext) if err != nil { return "", err } - - normalized := strings.TrimRight(parsed.String(), "/") if normalized == "" { return "", errors.New("URL is empty after normalization") } @@ -760,59 +815,116 @@ func (a *Agent) collectOnce(ctx context.Context) error { return fmt.Errorf("build docker report: %w", err) } - if err := a.sendReport(ctx, report); err != nil { - if errors.Is(err, ErrStopRequested) { - return nil - } - a.logger.Warn(). - Err(err). - Int("buffered_reports", a.bufferedReports()). - Int("targets", len(a.targets)). - Msg("Failed to send docker report, buffering") - a.reportBuffer.Push(report) - return nil - } - - a.flushBuffer(ctx) - return nil + return a.deliverReport(ctx, report) } func (a *Agent) flushBuffer(ctx context.Context) { - report, ok := a.reportBuffer.Peek() - if !ok { + a.ensureReportBuffers() + for _, target := range a.targets { + a.flushTargetBuffer(ctx, target) + } +} + +func (a *Agent) deliverReport(ctx context.Context, report agentsdocker.Report) error { + a.ensureReportBuffers() + payload, err := json.Marshal(report) + if err != nil { + return fmt.Errorf("marshal report: %w", err) + } + compressed, err := utils.CompressJSON(payload) + if err != nil { + return fmt.Errorf("compress report: %w", err) + } + for _, target := range a.targets { + if err := a.sendReportToTarget(ctx, target, compressed, len(report.Containers)); err != nil { + agenttarget.MarkDelivery("docker", target.Name, targetRole(target), false) + if errors.Is(err, ErrStopRequested) && target.Authoritative { + return nil + } + a.reportBuffers[target.Name].Push(report) + a.logger.Warn().Err(err).Str("destination", target.Name). + Bool("authoritative", target.Authoritative). + Int("buffered_reports", a.reportBuffers[target.Name].Len()). + Msg("Failed to send docker report, buffering only for this destination") + continue + } + agenttarget.MarkDelivery("docker", target.Name, targetRole(target), true) + a.flushTargetBuffer(ctx, target) + } + return nil +} + +func targetRole(target TargetConfig) string { + if target.Authoritative { + return "primary" + } + return "observer" +} + +func (a *Agent) flushTargetBuffer(ctx context.Context, target TargetConfig) { + a.ensureReportBuffers() + queue := a.reportBuffers[target.Name] + if queue == nil { return } - - a.logger.Info().Int("count", a.reportBuffer.Len()).Msg("Flushing buffered docker reports") - for { - if err := a.sendReport(ctx, report); err != nil { - if errors.Is(err, ErrStopRequested) { - return - } - a.logger.Warn(). - Err(err). - Int("remaining_reports", a.bufferedReports()). - Msg("Failed to flush buffered docker report, stopping flush") - return - } - a.reportBuffer.Pop() - - report, ok = a.reportBuffer.Peek() + report, ok := queue.Peek() if !ok { return } + payload, err := json.Marshal(report) + if err != nil { + queue.Pop() + continue + } + compressed, err := utils.CompressJSON(payload) + if err != nil { + return + } + if err := a.sendReportToTarget(ctx, target, compressed, len(report.Containers)); err != nil { + return + } + queue.Pop() } } func (a *Agent) bufferedReports() int { - if a.reportBuffer == nil { - return 0 + a.ensureReportBuffers() + total := 0 + for _, queue := range a.reportBuffers { + if queue != nil { + total += queue.Len() + } + } + return total +} + +func (a *Agent) ensureReportBuffers() { + if a.reportBuffers != nil { + return + } + a.reportBuffers = make(map[string]*utils.Queue[agentsdocker.Report], len(a.targets)) + for index := range a.targets { + if strings.TrimSpace(a.targets[index].Name) == "" { + if index == 0 { + a.targets[index].Name = "primary" + } else { + a.targets[index].Name = fmt.Sprintf("observer-%d", index) + } + } + if index == 0 { + a.targets[index].Authoritative = true + } + queue := utils.New[agentsdocker.Report](60) + if index == 0 && a.reportBuffer != nil { + queue = a.reportBuffer + } + a.reportBuffers[a.targets[index].Name] = queue } - return a.reportBuffer.Len() } func (a *Agent) sendReport(ctx context.Context, report agentsdocker.Report) error { + a.ensureReportBuffers() payload, err := json.Marshal(report) if err != nil { return fmt.Errorf("marshal report: %w", err) @@ -917,6 +1029,9 @@ func (a *Agent) sendReportToTarget(ctx context.Context, target TargetConfig, pay if len(body) == 0 { return nil } + if !target.Authoritative && target.Name != "" { + return nil + } var reportResp agentsdocker.ReportResponse if err := json.Unmarshal(body, &reportResp); err != nil { @@ -1171,10 +1286,15 @@ func (a *Agent) sendCommandAckWithPayload(ctx context.Context, target TargetConf } func (a *Agent) primaryTarget() TargetConfig { - if len(a.targets) == 0 { - return TargetConfig{} + for _, target := range a.targets { + if target.Authoritative { + return target + } } - return a.targets[0] + if len(a.targets) > 0 { + return a.targets[0] + } + return TargetConfig{} } func (a *Agent) httpClientFor(target TargetConfig) *http.Client { diff --git a/internal/dockeragent/multi_target_test.go b/internal/dockeragent/multi_target_test.go new file mode 100644 index 000000000..a2a3f0682 --- /dev/null +++ b/internal/dockeragent/multi_target_test.go @@ -0,0 +1,71 @@ +package dockeragent + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/utils" + agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker" + "github.com/rs/zerolog" +) + +func TestDeliverReportBuffersOnlyFailedDockerDestination(t *testing.T) { + primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) + defer primary.Close() + observer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusBadGateway) })) + defer observer.Close() + + targets := []TargetConfig{ + {Name: "primary", URL: primary.URL, Token: "p", Authoritative: true}, + {Name: "dev", URL: observer.URL, Token: "o"}, + } + a := &Agent{ + logger: zerolog.Nop(), targets: targets, + httpClients: map[bool]*http.Client{false: http.DefaultClient}, + trustedHTTPClients: map[string]*http.Client{}, + reportBuffer: utils.New[agentsdocker.Report](10), + reportBuffers: map[string]*utils.Queue[agentsdocker.Report]{ + "primary": utils.New[agentsdocker.Report](10), + "dev": utils.New[agentsdocker.Report](10), + }, + } + a.reportBuffer = a.reportBuffers["primary"] + if err := a.deliverReport(context.Background(), agentsdocker.Report{}); err != nil { + t.Fatalf("deliver report: %v", err) + } + if a.reportBuffers["primary"].Len() != 0 || a.reportBuffers["dev"].Len() != 1 { + t.Fatalf("buffer depths primary=%d observer=%d", a.reportBuffers["primary"].Len(), a.reportBuffers["dev"].Len()) + } +} + +func TestDockerObserverCannotIssueCommands(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"commands":[{"id":"cmd1","type":"stop"}]}`)) + })) + defer server.Close() + a := &Agent{logger: zerolog.Nop(), httpClients: map[bool]*http.Client{false: server.Client()}, trustedHTTPClients: map[string]*http.Client{}} + if err := a.sendReportToTarget(context.Background(), TargetConfig{Name: "dev", URL: server.URL, Token: "o"}, []byte(`{}`), 0); err != nil { + t.Fatalf("observer response must be acknowledgement-only: %v", err) + } +} + +func TestDockerObserverPlaintextPolicyIsDestinationScoped(t *testing.T) { + targets := []TargetConfig{ + {Name: "primary", URL: "https://primary.example.test", Token: "p", Authoritative: true}, + {Name: "observer", URL: "http://203.0.113.10:7655", Token: "o"}, + } + if _, err := normalizeTargets(targets); err == nil { + t.Fatal("expected observer plaintext URL rejection without destination opt-in") + } + targets[1].AllowPlaintextHTTP = true + normalized, err := normalizeTargets(targets) + if err != nil { + t.Fatalf("explicit observer plaintext opt-in: %v", err) + } + if !normalized[1].AllowPlaintextHTTP { + t.Fatal("observer plaintext policy was not preserved after normalization") + } +} diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index 9b1323829..29af183d7 100644 --- a/internal/hostagent/agent.go +++ b/internal/hostagent/agent.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "net" "net/http" "net/netip" @@ -19,6 +20,7 @@ import ( "sync" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/agenttarget" "github.com/rcourtman/pulse-go-rewrite/internal/agenttls" "github.com/rcourtman/pulse-go-rewrite/internal/agentupdate" "github.com/rcourtman/pulse-go-rewrite/internal/platformsupport" @@ -44,6 +46,7 @@ type Config struct { InsecureSkipVerify bool CACertPath string ServerFingerprint string + Observers []ObserverTarget RunOnce bool LogLevel zerolog.Level Logger *zerolog.Logger @@ -92,6 +95,28 @@ type Config struct { storageCleanup *storageCleanupManager } +// ObserverTarget is a report-only Pulse destination. It can receive the same +// collected host payload and an independently provisioned Proxmox source, but +// it never supplies remote config, commands, enrollment, or update authority. +type ObserverTarget struct { + Name string + ID string + PulseURL string + APIToken string + InsecureSkipVerify bool + AllowPlaintextHTTP bool + CACertPath string + ServerFingerprint string + ProvisionProxmox bool +} + +type observerReporter struct { + target ObserverTarget + httpClient *http.Client + reportBuffer *utils.Queue[agentshost.Report] + lastAuthFailureLog time.Time +} + // Agent is responsible for collecting host metrics and shipping them to Pulse. type Agent struct { cfg Config @@ -117,6 +142,7 @@ type Agent struct { configMu sync.RWMutex remoteConfigChanged chan struct{} reportBuffer *utils.Queue[agentshost.Report] + observerReporters []*observerReporter commandClient *CommandClient commandClientMu sync.Mutex commandClientRunCancel context.CancelFunc @@ -270,23 +296,18 @@ func New(cfg Config) (*Agent, error) { if arch == "" { arch = runtime.GOARCH } - tlsConfig, err := agenttls.NewClientTLSConfig(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint) + client, err := newAgentHTTPClient(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint) if err != nil { return nil, fmt.Errorf("invalid TLS configuration: %w", err) } - client := &http.Client{ - Timeout: 15 * time.Second, - Transport: &http.Transport{ - Proxy: http.ProxyFromEnvironment, - TLSClientConfig: tlsConfig, - }, - // Disallow redirects for agent API calls. If a reverse proxy redirects - // HTTP to HTTPS, Go's default behavior converts POST to GET (per HTTP spec), - // causing 405 errors. Return an error with guidance instead. - CheckRedirect: func(req *http.Request, via []*http.Request) error { - return fmt.Errorf("server returned redirect to %s - if using a reverse proxy, ensure you use the correct protocol (https:// instead of http://) in your --url flag", req.URL) - }, + observerReporters, err := newObserverReporters(cfg.Observers) + if err != nil { + return nil, err + } + agenttarget.MarkConfigured("host", "primary", "primary") + for _, observer := range observerReporters { + agenttarget.MarkConfigured("host", observer.target.Name, "observer") } trimmedTags := make([]string, 0, len(cfg.Tags)) @@ -349,6 +370,7 @@ func New(cfg Config) (*Agent, error) { trimmedPulseURL: pulseURL, remoteConfigChanged: make(chan struct{}, 1), reportBuffer: utils.New[agentshost.Report](bufferCapacity), + observerReporters: observerReporters, collector: collector, newCommandClient: newCommandClientFn, runCommandClient: runCommandClientFn, @@ -367,6 +389,58 @@ func New(cfg Config) (*Agent, error) { return agent, nil } +func newAgentHTTPClient(caCertPath string, insecureSkipVerify bool, serverFingerprint string) (*http.Client, error) { + tlsConfig, err := agenttls.NewClientTLSConfig(caCertPath, insecureSkipVerify, serverFingerprint) + if err != nil { + return nil, err + } + return &http.Client{ + Timeout: 15 * time.Second, + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + TLSClientConfig: tlsConfig, + }, + // Redirects can rewrite report POSTs to GETs and must never cross an + // authority boundary implicitly. + CheckRedirect: func(req *http.Request, _ []*http.Request) error { + return fmt.Errorf("server returned redirect to %s - use the final Pulse URL explicitly", req.URL) + }, + }, nil +} + +func newObserverReporters(targets []ObserverTarget) ([]*observerReporter, error) { + const bufferCapacity = 60 + result := make([]*observerReporter, 0, len(targets)) + seen := make(map[string]struct{}, len(targets)) + for _, target := range targets { + target.Name = strings.TrimSpace(target.Name) + target.ID = strings.TrimSpace(target.ID) + target.APIToken = strings.TrimSpace(target.APIToken) + if target.Name == "" || target.ID == "" || target.APIToken == "" { + return nil, errors.New("observer destination requires name, id, and API token") + } + url, err := agenttarget.NormalizePulseURL(target.PulseURL, false, target.AllowPlaintextHTTP) + if err != nil { + return nil, fmt.Errorf("invalid observer %q Pulse URL: %w", target.Name, err) + } + target.PulseURL = url + if _, exists := seen[target.ID]; exists { + return nil, fmt.Errorf("duplicate observer destination id for %q", target.Name) + } + seen[target.ID] = struct{}{} + client, err := newAgentHTTPClient(target.CACertPath, target.InsecureSkipVerify, target.ServerFingerprint) + if err != nil { + return nil, fmt.Errorf("configure observer %q TLS: %w", target.Name, err) + } + result = append(result, &observerReporter{ + target: target, + httpClient: client, + reportBuffer: utils.New[agentshost.Report](bufferCapacity), + }) + } + return result, nil +} + func normalizeProxmoxType(raw string) (string, error) { value := strings.TrimSpace(strings.ToLower(raw)) switch value { @@ -442,10 +516,12 @@ func (a *Agent) Run(ctx context.Context) error { // Load any reports buffered from a previous shutdown a.loadPersistedBuffer() + a.loadPersistedObserverBuffers() ticker := time.NewTicker(a.currentInterval()) defer ticker.Stop() defer a.persistBuffer() + defer a.persistObserverBuffers() if err := a.process(ctx); err != nil && !errors.Is(err, context.Canceled) { a.logger.Error(). @@ -619,7 +695,16 @@ func (a *Agent) process(ctx context.Context) error { if err != nil { return fmt.Errorf("build report: %w", err) } + primaryErr := a.deliverPrimaryReport(ctx, report) + for _, observer := range a.observerReporters { + a.deliverObserverReport(ctx, observer, report) + } + return primaryErr +} + +func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Report) error { if err := a.sendReport(ctx, report); err != nil { + agenttarget.MarkDelivery("host", "primary", "primary", false) var statusErr *reportHTTPStatusError if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusForbidden { a.logger.Error(). @@ -652,6 +737,7 @@ func (a *Agent) process(ctx context.Context) error { event.Msg("Failed to send report, buffering") return nil } + agenttarget.MarkDelivery("host", "primary", "primary", true) // A successful report means the token is accepted again; reset the auth // failure throttle so a later rejection is reported promptly. @@ -667,6 +753,56 @@ func (a *Agent) process(ctx context.Context) error { return nil } +func (a *Agent) deliverObserverReport(ctx context.Context, observer *observerReporter, report agentshost.Report) { + if err := a.sendReportToDestination(ctx, report, observer.target.PulseURL, observer.target.APIToken, observer.httpClient, false); err != nil { + agenttarget.MarkDelivery("host", observer.target.Name, "observer", false) + var statusErr *reportHTTPStatusError + if errors.As(err, &statusErr) && (statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) { + if time.Since(observer.lastAuthFailureLog) >= authFailureLogInterval { + observer.lastAuthFailureLog = time.Now() + a.logger.Error().Err(err). + Str("destination", observer.target.Name). + Str("role", "observer"). + Str("pulse_url", observer.target.PulseURL). + Msg("Pulse observer rejected its API token; reports for this destination are dropped until the token is replaced") + } + return + } + observer.reportBuffer.Push(report) + a.logger.Warn().Err(err). + Str("destination", observer.target.Name). + Str("role", "observer"). + Int("buffered_reports", observer.reportBuffer.Len()). + Msg("Failed to send observer report, buffering only for this destination") + return + } + + agenttarget.MarkDelivery("host", observer.target.Name, "observer", true) + observer.lastAuthFailureLog = time.Time{} + a.flushObserverBuffer(ctx, observer) +} + +func (a *Agent) flushObserverBuffer(ctx context.Context, observer *observerReporter) { + for !observer.reportBuffer.IsEmpty() { + report, ok := observer.reportBuffer.Peek() + if !ok { + return + } + if err := a.sendReportToDestination(ctx, report, observer.target.PulseURL, observer.target.APIToken, observer.httpClient, false); err != nil { + var statusErr *reportHTTPStatusError + if errors.As(err, &statusErr) && (statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) { + for { + if _, ok := observer.reportBuffer.Pop(); !ok { + break + } + } + } + return + } + observer.reportBuffer.Pop() + } +} + // logAuthFailure emits an actionable, throttled error when the server rejects // the agent's API token with 401. It is only called from the single-threaded // report loop, so lastAuthFailureLog needs no additional synchronisation. @@ -739,11 +875,25 @@ func (a *Agent) flushBuffer(ctx context.Context) { const bufferFileName = "report-buffer.json" +func observerBufferFileName(id string) string { + return "report-buffer-observer-" + id + ".json" +} + // persistBuffer writes buffered reports to disk on shutdown so they can be // retransmitted on the next startup. Uses atomic write (tmp + rename) to // prevent corruption if the process is killed mid-write. func (a *Agent) persistBuffer() { - items := a.reportBuffer.Items() + a.persistReportQueue(a.reportBuffer, bufferFileName, "primary") +} + +func (a *Agent) persistObserverBuffers() { + for _, observer := range a.observerReporters { + a.persistReportQueue(observer.reportBuffer, observerBufferFileName(observer.target.ID), observer.target.Name) + } +} + +func (a *Agent) persistReportQueue(queue *utils.Queue[agentshost.Report], fileName, destination string) { + items := queue.Items() if len(items) == 0 { return } @@ -764,7 +914,7 @@ func (a *Agent) persistBuffer() { return } - path := filepath.Join(a.stateDir, bufferFileName) + path := filepath.Join(a.stateDir, fileName) tmpPath := path + ".tmp" if err := os.WriteFile(tmpPath, data, 0600); err != nil { a.logger.Warn().Err(err).Str("path", tmpPath).Msg("Failed to write report buffer temp file") @@ -776,18 +926,28 @@ func (a *Agent) persistBuffer() { return } - a.logger.Info().Int("count", len(items)).Str("path", path).Msg("Persisted report buffer to disk") + a.logger.Info().Int("count", len(items)).Str("path", path).Str("destination", destination).Msg("Persisted report buffer to disk") } // loadPersistedBuffer loads buffered reports from a previous shutdown and // attempts to flush them. The file is deleted after loading regardless of // whether the flush succeeds (items are pushed back into the in-memory buffer). func (a *Agent) loadPersistedBuffer() { + a.loadPersistedReportQueue(a.reportBuffer, bufferFileName, "primary") +} + +func (a *Agent) loadPersistedObserverBuffers() { + for _, observer := range a.observerReporters { + a.loadPersistedReportQueue(observer.reportBuffer, observerBufferFileName(observer.target.ID), observer.target.Name) + } +} + +func (a *Agent) loadPersistedReportQueue(queue *utils.Queue[agentshost.Report], fileName, destination string) { if a.stateDir == "" { return } - path := filepath.Join(a.stateDir, bufferFileName) + path := filepath.Join(a.stateDir, fileName) data, err := os.ReadFile(path) if err != nil { if !os.IsNotExist(err) { @@ -810,10 +970,10 @@ func (a *Agent) loadPersistedBuffer() { } for _, item := range items { - a.reportBuffer.Push(item) + queue.Push(item) } - a.logger.Info().Int("count", len(items)).Msg("Loaded persisted report buffer from disk") + a.logger.Info().Int("count", len(items)).Str("destination", destination).Msg("Loaded persisted report buffer from disk") } func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { @@ -960,6 +1120,10 @@ func (a *Agent) currentStorageCleanupStatus(ctx context.Context) *agentshost.Sto } func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error { + return a.sendReportToDestination(ctx, report, a.trimmedPulseURL, a.cfg.APIToken, a.httpClient, true) +} + +func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.Report, pulseURL, token string, client *http.Client, authoritative bool) error { payload, err := json.Marshal(report) if err != nil { return fmt.Errorf("marshal report: %w", err) @@ -971,7 +1135,7 @@ func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error } endpoint := agentReportEndpoint - url := fmt.Sprintf("%s%s", a.trimmedPulseURL, endpoint) + url := fmt.Sprintf("%s%s", strings.TrimRight(pulseURL, "/"), endpoint) req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(compressed)) if err != nil { return fmt.Errorf("create request: %w", err) @@ -979,13 +1143,13 @@ func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Encoding", "gzip") - if token := strings.TrimSpace(a.cfg.APIToken); token != "" { + if token := strings.TrimSpace(token); token != "" { req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("X-API-Token", token) } req.Header.Set("User-Agent", "pulse-agent/"+Version) - resp, err := a.httpClient.Do(req) + resp, err := client.Do(req) if err != nil { return fmt.Errorf("send request: %w", err) } @@ -1002,6 +1166,10 @@ func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error StatusCode: resp.StatusCode, } } + if !authoritative { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64*1024)) + return nil + } // Parse response to check for server-side config overrides var reportResp struct { @@ -1795,50 +1963,56 @@ func (a *Agent) collectSMARTData(ctx context.Context, diskExclude []string) []ag // Supports hosts with multiple Proxmox products (e.g., PVE + PBS on same host). func (a *Agent) runProxmoxSetup(ctx context.Context) { a.logger.Info().Msg("Proxmox mode enabled, checking setup...") - - setup := NewProxmoxSetup( - a.logger, - a.httpClient, - a.collector, - a.trimmedPulseURL, - a.cfg.APIToken, - a.cfg.ProxmoxType, - a.hostname, - a.currentReportIP(), - a.stateDir, - a.cfg.InsecureSkipVerify, - ) - - // Use RunAll to detect and register all Proxmox products on this host - results, err := setup.RunAll(ctx) - if err != nil { - a.logger.Error().Err(err).Msg("Proxmox setup failed") - return - } - - if len(results) == 0 { - // All types already registered - a.logger.Info().Msg("All detected Proxmox products already registered") - return - } - - // Log results for each registered type - for _, result := range results { - if result.Registered { - a.logger.Info(). - Str("type", result.ProxmoxType). - Str("host", result.NodeHost). - Str("token_id", result.TokenID). - Msg("Proxmox node registered successfully") - } else { - a.logger.Warn(). - Str("type", result.ProxmoxType). - Str("host", result.NodeHost). - Msg("Proxmox token created but registration failed (node may need manual configuration)") + for _, destination := range a.proxmoxDestinations() { + results, err := destination.setup.RunAll(ctx) + if err != nil { + a.logger.Error().Err(err).Str("destination", destination.name).Msg("Proxmox setup failed") + continue + } + for _, result := range results { + if result.Registered { + a.logger.Info().Str("destination", destination.name). + Str("type", result.ProxmoxType). + Str("host", result.NodeHost). + Str("token_id", result.TokenID). + Msg("Proxmox node registered successfully") + } else { + a.logger.Warn().Str("destination", destination.name). + Str("type", result.ProxmoxType). + Str("host", result.NodeHost). + Msg("Proxmox token created but registration failed") + } } } } +type proxmoxDestination struct { + name string + setup *ProxmoxSetup +} + +func (a *Agent) proxmoxDestinations() []proxmoxDestination { + destinations := []proxmoxDestination{{ + name: "primary", + setup: NewProxmoxSetup(a.logger, a.httpClient, a.collector, a.trimmedPulseURL, a.cfg.APIToken, + a.cfg.ProxmoxType, a.hostname, a.currentReportIP(), a.stateDir, a.cfg.InsecureSkipVerify), + }} + for _, observer := range a.observerReporters { + if !observer.target.ProvisionProxmox { + continue + } + stateDir := filepath.Join(a.stateDir, "observers", observer.target.ID) + tokenName := "pulse-" + proxmoxTokenScope(a.hostname, a.trimmedPulseURL) + "-" + observer.target.ID + setup := NewProxmoxSetup(a.logger, observer.httpClient, a.collector, observer.target.PulseURL, + observer.target.APIToken, a.cfg.ProxmoxType, a.hostname, a.currentReportIP(), stateDir, + observer.target.InsecureSkipVerify). + WithObserverPlaintextPolicy(observer.target.AllowPlaintextHTTP). + WithMonitorTokenName(tokenName) + destinations = append(destinations, proxmoxDestination{name: observer.target.Name, setup: setup}) + } + return destinations +} + const ( proxmoxHealthCheckInitialDelay = 2 * time.Minute proxmoxHealthCheckInterval = 5 * time.Minute @@ -1865,34 +2039,22 @@ func (a *Agent) runProxmoxHealthCheckLoop(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - setup := NewProxmoxSetup( - a.logger, - a.httpClient, - a.collector, - a.trimmedPulseURL, - a.cfg.APIToken, - a.cfg.ProxmoxType, - a.hostname, - a.currentReportIP(), - a.stateDir, - a.cfg.InsecureSkipVerify, - ) - results, err := setup.RunHealthCheck(ctx) - if err != nil { - a.logger.Warn().Err(err).Msg("Proxmox health check failed") - continue - } - for _, result := range results { - if result.Registered { - a.logger.Info(). + for _, destination := range a.proxmoxDestinations() { + results, err := destination.setup.RunHealthCheck(ctx) + if err != nil { + a.logger.Warn().Err(err).Str("destination", destination.name).Msg("Proxmox health check failed") + continue + } + for _, result := range results { + event := a.logger.Info() + if !result.Registered { + event = a.logger.Warn() + } + event.Str("destination", destination.name). Str("type", result.ProxmoxType). Str("host", result.NodeHost). - Msg("Proxmox node re-registered via health check") - } else { - a.logger.Warn(). - Str("type", result.ProxmoxType). - Str("host", result.NodeHost). - Msg("Proxmox health check: token rotated but registration failed") + Bool("registered", result.Registered). + Msg("Proxmox destination health repair completed") } } } diff --git a/internal/hostagent/observer_delivery_test.go b/internal/hostagent/observer_delivery_test.go new file mode 100644 index 000000000..ad0c0257a --- /dev/null +++ b/internal/hostagent/observer_delivery_test.go @@ -0,0 +1,94 @@ +package hostagent + +import ( + "context" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/utils" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" + "github.com/rs/zerolog" +) + +func TestObserverDeliveryIsIsolatedAndReportOnly(t *testing.T) { + primaryRequests := atomic.Int32{} + primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + primaryRequests.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"config":{"commandsEnabled":false}}`)) + })) + defer primary.Close() + + observerRequests := atomic.Int32{} + observerHealthy := atomic.Bool{} + observer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + observerRequests.Add(1) + if !observerHealthy.Load() { + w.WriteHeader(http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"config":{"commandsEnabled":true}}`)) + })) + defer observer.Close() + + logger := zerolog.Nop() + a := &Agent{ + cfg: Config{APIToken: "primary-token"}, + logger: logger, + httpClient: primary.Client(), + trimmedPulseURL: primary.URL, + reportBuffer: utils.New[agentshost.Report](10), + remoteConfigChanged: make(chan struct{}, 1), + observerReporters: []*observerReporter{{ + target: ObserverTarget{Name: "dev", ID: "dev12345", PulseURL: observer.URL, APIToken: "observer-token"}, + httpClient: observer.Client(), reportBuffer: utils.New[agentshost.Report](10), + }}, + } + report := agentshost.Report{Host: agentshost.HostInfo{Hostname: "node"}} + + if err := a.deliverPrimaryReport(context.Background(), report); err != nil { + t.Fatalf("primary delivery: %v", err) + } + a.deliverObserverReport(context.Background(), a.observerReporters[0], report) + if got := a.reportBuffer.Len(); got != 0 { + t.Fatalf("primary buffer depth = %d, want 0", got) + } + if got := a.observerReporters[0].reportBuffer.Len(); got != 1 { + t.Fatalf("observer buffer depth = %d, want 1", got) + } + + observerHealthy.Store(true) + a.deliverObserverReport(context.Background(), a.observerReporters[0], report) + if got := a.observerReporters[0].reportBuffer.Len(); got != 0 { + t.Fatalf("observer buffer depth after recovery = %d, want 0", got) + } + if a.cfg.EnableCommands { + t.Fatal("observer response changed authoritative command configuration") + } + if primaryRequests.Load() != 1 || observerRequests.Load() != 3 { + t.Fatalf("requests primary=%d observer=%d, want 1 and 3", primaryRequests.Load(), observerRequests.Load()) + } +} + +func TestObserverReporterPlaintextPolicyIsDestinationScoped(t *testing.T) { + target := ObserverTarget{ + Name: "observer", + ID: "observer-id", + PulseURL: "http://203.0.113.10:7655", + APIToken: "observer-token", + } + if _, err := newObserverReporters([]ObserverTarget{target}); err == nil { + t.Fatal("expected observer plaintext URL rejection without destination opt-in") + } + target.AllowPlaintextHTTP = true + reporters, err := newObserverReporters([]ObserverTarget{target}) + if err != nil { + t.Fatalf("explicit observer plaintext opt-in: %v", err) + } + if len(reporters) != 1 || !reporters[0].target.AllowPlaintextHTTP { + t.Fatalf("observer plaintext policy was not preserved: %+v", reporters) + } +} diff --git a/internal/hostagent/proxmox_setup.go b/internal/hostagent/proxmox_setup.go index a2f5f34df..b172538e7 100644 --- a/internal/hostagent/proxmox_setup.go +++ b/internal/hostagent/proxmox_setup.go @@ -16,6 +16,7 @@ import ( "strings" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/agenttarget" "github.com/rs/zerolog" ) @@ -32,6 +33,9 @@ type ProxmoxSetup struct { collector SystemCollector stateDir string // directory for registration state files retryBackoffs []time.Duration // overridable for testing; nil uses defaults + monitorToken string // optional destination-scoped token name + authoritative bool // primary destinations retain local-network HTTP compatibility + allowPlaintextHTTP bool // explicit observer plaintext transport policy } // ProxmoxSetupResult contains the result of a successful Proxmox setup. @@ -200,6 +204,9 @@ func proxmoxTokenScope(candidates ...string) string { } func (p *ProxmoxSetup) monitorTokenName() string { + if tokenName := strings.TrimSpace(p.monitorToken); tokenName != "" { + return tokenName + } return "pulse-" + proxmoxTokenScope(p.hostname, p.pulseURL) } @@ -347,15 +354,33 @@ func NewProxmoxSetup(logger zerolog.Logger, httpClient *http.Client, collector S reportIP: reportIP, stateDir: stateDir, insecureSkipVerify: insecure, + authoritative: true, } } +// WithObserverPlaintextPolicy marks this setup as an observer transport. Unlike +// the primary, every non-loopback plaintext observer requires its own explicit +// opt-in and cannot inherit process-wide primary transport consent. +func (p *ProxmoxSetup) WithObserverPlaintextPolicy(allowPlaintext bool) *ProxmoxSetup { + p.authoritative = false + p.allowPlaintextHTTP = allowPlaintext + return p +} + +// WithMonitorTokenName isolates Proxmox credentials for an additional Pulse +// destination. The primary destination deliberately keeps the legacy token +// name for upgrade compatibility. +func (p *ProxmoxSetup) WithMonitorTokenName(tokenName string) *ProxmoxSetup { + p.monitorToken = strings.TrimSpace(tokenName) + return p +} + // Run executes the Proxmox setup process: // 1. Detects Proxmox type (if not specified) // 2. Creates the monitoring user and API token // 3. Registers the node with Pulse via auto-register func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) { - pulseURL, err := normalizePulseURL(p.pulseURL) + pulseURL, err := agenttarget.NormalizePulseURL(p.pulseURL, p.authoritative, p.allowPlaintextHTTP) if err != nil { return nil, fmt.Errorf("invalid pulse URL: %w", err) } @@ -377,53 +402,11 @@ func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) { p.logger.Info().Str("type", string(ptype)).Msg("Auto-detected Proxmox type") } - hostURL := p.getHostURL(ctx, ptype) - - // Check if already registered (idempotency) - if p.isAlreadyRegistered() { - registered, err := p.checkRegistrationWithPulse(ctx, ptype, hostURL) - if err != nil { - p.logger.Warn().Err(err).Msg("Failed to verify Proxmox registration state with Pulse; keeping local marker behavior") - return nil, nil - } - if registered { - p.logger.Info().Msg("Proxmox node already registered, skipping setup") - return nil, nil - } - p.logger.Info().Str("type", string(ptype)).Str("host", hostURL).Msg("Local Proxmox registration marker exists but Pulse has no matching node; re-registering") - } - - // Create monitoring user and token - tokenID, tokenValue, err := p.setupToken(ctx, ptype) - if err != nil { - return nil, fmt.Errorf("failed to create Proxmox API token: %w", err) - } - - p.logger.Info().Str("token_id", tokenID).Msg("Created Proxmox API token") - - // Register with Pulse - registered := false - registerResp, err := p.registerWithPulse(ctx, ptype, hostURL, tokenID, tokenValue) - if err != nil { - p.logger.Warn().Err(err).Msg("Failed to register with Pulse (node may already exist)") - } else { - registered = true + result, err := p.runForType(ctx, ptype) + if result != nil && result.Registered { p.markAsRegistered() - p.markTypeAsRegistered(ptype) - if strings.TrimSpace(registerResp.Host) != "" { - hostURL = registerResp.Host - } - p.logger.Info().Str("host", hostURL).Str("node", registerResp.NodeName).Msg("Successfully registered Proxmox node with Pulse") } - - return &ProxmoxSetupResult{ - ProxmoxType: string(ptype), - TokenID: tokenID, - TokenValue: tokenValue, - NodeHost: hostURL, - NodeName: registerResp.NodeName, - Registered: registered, - }, nil + return result, err } // RunAll detects and registers ALL Proxmox products on this system. @@ -431,7 +414,7 @@ func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) { // supported configuration). Each type gets its own registration and state tracking. // Returns results for all types that were processed (skipping already-registered ones). func (p *ProxmoxSetup) RunAll(ctx context.Context) ([]*ProxmoxSetupResult, error) { - pulseURL, err := normalizePulseURL(p.pulseURL) + pulseURL, err := agenttarget.NormalizePulseURL(p.pulseURL, p.authoritative, p.allowPlaintextHTTP) if err != nil { return nil, fmt.Errorf("invalid pulse URL: %w", err) } @@ -490,7 +473,7 @@ func (p *ProxmoxSetup) RunAll(ctx context.Context) ([]*ProxmoxSetupResult, error // health-check rotation, which would cause uncontrolled token churn if Pulse // is temporarily unreachable. func (p *ProxmoxSetup) RunHealthCheck(ctx context.Context) ([]*ProxmoxSetupResult, error) { - pulseURL, err := normalizePulseURL(p.pulseURL) + pulseURL, err := agenttarget.NormalizePulseURL(p.pulseURL, p.authoritative, p.allowPlaintextHTTP) if err != nil { return nil, fmt.Errorf("invalid pulse URL: %w", err) } @@ -552,20 +535,26 @@ func (p *ProxmoxSetup) runForType(ctx context.Context, ptype proxmoxProductType) ptype = proxmoxProductType(normalizedTypeStr) hostURL := p.getHostURL(ctx, ptype) - // Check if this type is already registered + // Always establish destination reachability and registration state before + // mutating a Proxmox token. This prevents a missing local marker or a Pulse + // startup outage from rotating credentials that a healthy Pulse instance is + // still using. + registered, err := p.checkRegistrationWithPulse(ctx, ptype, hostURL) + if err != nil { + p.logger.Warn(). + Err(err). + Str("type", string(ptype)). + Msg("Failed to verify Proxmox registration state with Pulse; leaving credentials unchanged") + return nil, nil + } + if registered { + if !p.isTypeRegistered(ptype) { + p.markTypeAsRegistered(ptype) + } + p.logger.Info().Str("type", string(ptype)).Msg("Proxmox type already registered, skipping") + return nil, nil + } if p.isTypeRegistered(ptype) { - registered, err := p.checkRegistrationWithPulse(ctx, ptype, hostURL) - if err != nil { - p.logger.Warn(). - Err(err). - Str("type", string(ptype)). - Msg("Failed to verify Proxmox registration state with Pulse; keeping local marker behavior") - return nil, nil - } - if registered { - p.logger.Info().Str("type", string(ptype)).Msg("Proxmox type already registered, skipping") - return nil, nil - } p.logger.Info(). Str("type", string(ptype)). Str("host", hostURL). @@ -588,7 +577,7 @@ func (p *ProxmoxSetup) runForType(ctx context.Context, ptype proxmoxProductType) p.logger.Info().Str("type", string(ptype)).Str("token_id", tokenID).Msg("Created Proxmox API token") // Register with Pulse - registered := false + registered = false registerResp, err := p.registerWithPulse(ctx, ptype, hostURL, tokenID, tokenValue) if err != nil { p.logger.Warn().Err(err).Str("type", string(ptype)).Msg("Failed to register with Pulse (node may already exist)") diff --git a/internal/hostagent/proxmox_setup_test.go b/internal/hostagent/proxmox_setup_test.go index fdb1a2430..7c1723185 100644 --- a/internal/hostagent/proxmox_setup_test.go +++ b/internal/hostagent/proxmox_setup_test.go @@ -560,6 +560,7 @@ func TestGetIPThatReachesPulse_IPv6Target(t *testing.T) { func TestProxmoxSetup_RunForType(t *testing.T) { mc := &mockCollector{} var expectedTokenID string + registrationExists := true server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/api/setup-script-url": @@ -573,7 +574,7 @@ func TestProxmoxSetup_RunForType(t *testing.T) { } if check, _ := payload["checkRegistration"].(bool); check { w.Header().Set("Content-Type", "application/json") - _, _ = w.Write([]byte(`{"registered":true}`)) + _, _ = w.Write([]byte(fmt.Sprintf(`{"registered":%t}`, registrationExists))) return } w.Header().Set("Content-Type", "application/json") @@ -601,6 +602,7 @@ func TestProxmoxSetup_RunForType(t *testing.T) { }) t.Run("performs registration successfully", func(t *testing.T) { + registrationExists = false mc.statFn = func(name string) (os.FileInfo, error) { return nil, os.ErrNotExist } mc.commandCombinedOutputFn = func(ctx context.Context, name string, arg ...string) (string, error) { // pveum user token add pulse-monitor@pve ... --privsep 1 @@ -653,6 +655,44 @@ func TestProxmoxSetup_MonitorTokenNameIsNodeScoped(t *testing.T) { } } +func TestProxmoxSetup_DoesNotMutateTokenBeforeDestinationCheck(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + mutations := 0 + mc := &mockCollector{ + statFn: func(string) (os.FileInfo, error) { return nil, os.ErrNotExist }, + dialTimeoutFn: func(string, string, time.Duration) (net.Conn, error) { + return &mockConn{localAddr: &net.UDPAddr{IP: net.ParseIP("10.0.0.1")}}, nil + }, + commandCombinedOutputFn: func(_ context.Context, name string, _ ...string) (string, error) { + if name == "pveum" || name == "proxmox-backup-manager" { + mutations++ + } + return "", nil + }, + } + p := NewProxmoxSetup(zerolog.Nop(), server.Client(), mc, server.URL, "token", "pve", "node", "", t.TempDir(), false) + p.retryBackoffs = []time.Duration{} + result, err := p.runForType(context.Background(), proxmoxProductPVE) + if err != nil { + t.Fatalf("runForType: %v", err) + } + if result != nil { + t.Fatalf("result = %+v, want nil while destination is unavailable", result) + } + if mutations != 0 { + t.Fatalf("Proxmox command mutations = %d, want 0", mutations) + } + + p.WithMonitorTokenName("pulse-node-observer-deadbeef") + if got := p.monitorTokenName(); got != "pulse-node-observer-deadbeef" { + t.Fatalf("observer token name = %q", got) + } +} + func TestProxmoxSetup_SetupPVETokenUsesPrivilegeSeparatedTokenACLs(t *testing.T) { mc := &mockCollector{} calls := make([]commandCall, 0) @@ -967,8 +1007,8 @@ func TestProxmoxSetup_RunAll(t *testing.T) { } results, _ := p.RunAll(context.Background()) - if len(results) != 1 || results[0].ProxmoxType != "pbs" { - t.Errorf("expected pbs result") + if len(results) != 0 { + t.Errorf("expected no token mutation while Pulse registration state is unreachable") } }) @@ -1437,8 +1477,8 @@ func TestProxmoxSetup_Run_TopLevel(t *testing.T) { if gotPayload.AuthToken != "setup-token-optional" { t.Fatalf("authToken = %q, want %q", gotPayload.AuthToken, "setup-token-optional") } - if requestCount != 2 { - t.Fatalf("requestCount = %d, want 2", requestCount) + if requestCount != 4 { + t.Fatalf("requestCount = %d, want 4 (reachability check plus registration)", requestCount) } }) @@ -1456,6 +1496,27 @@ func TestProxmoxSetup_Run_TopLevel(t *testing.T) { }) } +func TestObserverProxmoxSetupRequiresDestinationPlaintextOptIn(t *testing.T) { + setup := NewProxmoxSetup( + zerolog.Nop(), + http.DefaultClient, + nil, + "http://203.0.113.10:7655", + "observer-token", + "pve", + "node", + "", + t.TempDir(), + false, + ).WithObserverPlaintextPolicy(false) + + if _, err := setup.Run(context.Background()); err == nil { + t.Fatal("expected observer Proxmox setup to reject plaintext without destination opt-in") + } else if !strings.Contains(err.Error(), "explicitly allows plaintext HTTP") { + t.Fatalf("unexpected error: %v", err) + } +} + // Helpers type mockConn struct { diff --git a/internal/kubernetesagent/agent.go b/internal/kubernetesagent/agent.go index 4e4f1f6fa..565307e14 100644 --- a/internal/kubernetesagent/agent.go +++ b/internal/kubernetesagent/agent.go @@ -19,6 +19,7 @@ import ( "time" "github.com/IGLOU-EU/go-wildcard/v2" + "github.com/rcourtman/pulse-go-rewrite/internal/agenttarget" "github.com/rcourtman/pulse-go-rewrite/internal/agenttls" "github.com/rcourtman/pulse-go-rewrite/internal/securityutil" "github.com/rcourtman/pulse-go-rewrite/internal/utils" @@ -57,6 +58,7 @@ type Config struct { InsecureSkipVerify bool CACertPath string ServerFingerprint string + Targets []TargetConfig LogLevel zerolog.Level Logger *zerolog.Logger @@ -72,6 +74,25 @@ type Config struct { MaxPods int // Max pods included in the report } +// TargetConfig describes one Pulse report destination. Exactly one target is +// authoritative; additional targets are report-only observers. +type TargetConfig struct { + Name string + URL string + Token string + InsecureSkipVerify bool + AllowPlaintextHTTP bool + CACertPath string + ServerFingerprint string + Authoritative bool +} + +type reportTarget struct { + config TargetConfig + client *http.Client + buffer *utils.Queue[agentsk8s.Report] +} + // Agent collects and reports Kubernetes cluster state to Pulse. // It periodically gathers pod, deployment, and node metrics, then sends // them to the configured Pulse URL. The agent handles authentication, @@ -100,6 +121,7 @@ type Agent struct { excludeNamespaces []string reportBuffer *utils.Queue[agentsk8s.Report] + targets []*reportTarget } const ( @@ -147,19 +169,32 @@ func New(cfg Config) (*Agent, error) { logger = &scoped } - if strings.TrimSpace(cfg.APIToken) == "" { - return nil, fmt.Errorf("api token is required") - } - - pulseURL := strings.TrimSpace(cfg.PulseURL) - if pulseURL == "" { - pulseURL = "http://localhost:7655" - } - pulseURL, err := normalizePulseURL(pulseURL) + legacyTarget := len(cfg.Targets) == 0 + targetConfigs, err := normalizeKubernetesTargets(cfg) if err != nil { - return nil, fmt.Errorf("invalid pulse URL: %w", err) + if legacyTarget { + if strings.TrimSpace(cfg.APIToken) == "" { + return nil, fmt.Errorf("api token is required") + } + return nil, fmt.Errorf("invalid pulse URL: %w", err) + } + return nil, err } - cfg.PulseURL = pulseURL + cfg.Targets = targetConfigs + primary := targetConfigs[0] + for _, target := range targetConfigs { + role := "observer" + if target.Authoritative { + role = "primary" + } + agenttarget.MarkConfigured("kubernetes", target.Name, role) + if target.Authoritative { + primary = target + } + } + cfg.PulseURL = primary.URL + cfg.APIToken = primary.Token + pulseURL := primary.URL restCfg, contextName, err := buildRESTConfig(cfg.KubeconfigPath, cfg.KubeContext) if err != nil { @@ -183,22 +218,23 @@ func New(cfg Config) (*Agent, error) { agentVersion = Version } - tlsConfig, err := agenttls.NewClientTLSConfig(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint) - if err != nil { - return nil, fmt.Errorf("configure Pulse TLS client: %w", err) - } - httpClient := &http.Client{ - Timeout: 15 * time.Second, - Transport: &http.Transport{ - Proxy: http.ProxyFromEnvironment, - TLSClientConfig: tlsConfig, - }, - // Disallow redirects for agent API calls. If a reverse proxy redirects - // HTTP to HTTPS, Go's default behavior converts POST to GET (per HTTP spec), - // causing 405 errors. Return an error with guidance instead. - CheckRedirect: func(req *http.Request, via []*http.Request) error { - return fmt.Errorf("server returned redirect to %s - if using a reverse proxy, ensure you use the correct protocol (https:// instead of http://) in your --url flag", req.URL) - }, + reportTargets := make([]*reportTarget, 0, len(targetConfigs)) + var httpClient *http.Client + var primaryBuffer *utils.Queue[agentsk8s.Report] + for _, target := range targetConfigs { + client, err := newKubernetesHTTPClient(target) + if err != nil { + if legacyTarget { + return nil, fmt.Errorf("configure Pulse TLS client: %w", err) + } + return nil, fmt.Errorf("configure Pulse target %q: %w", target.Name, err) + } + buffer := utils.New[agentsk8s.Report](60) + reportTargets = append(reportTargets, &reportTarget{config: target, client: client, buffer: buffer}) + if target.Authoritative { + httpClient = client + primaryBuffer = buffer + } } clusterServer := strings.TrimSpace(restCfg.Host) @@ -228,7 +264,8 @@ func New(cfg Config) (*Agent, error) { clusterContext: clusterContext, includeNamespaces: cfg.IncludeNamespaces, excludeNamespaces: cfg.ExcludeNamespaces, - reportBuffer: utils.New[agentsk8s.Report](60), + reportBuffer: primaryBuffer, + targets: reportTargets, } if err := agent.discoverClusterMetadata(context.Background()); err != nil { @@ -256,6 +293,85 @@ func normalizePulseURL(rawURL string) (string, error) { return parsed.String(), nil } +func normalizeKubernetesTargets(cfg Config) ([]TargetConfig, error) { + raw := append([]TargetConfig(nil), cfg.Targets...) + if len(raw) == 0 { + url := strings.TrimSpace(cfg.PulseURL) + if url == "" { + url = "http://localhost:7655" + } + raw = []TargetConfig{{ + Name: "primary", URL: url, Token: cfg.APIToken, InsecureSkipVerify: cfg.InsecureSkipVerify, + CACertPath: cfg.CACertPath, ServerFingerprint: cfg.ServerFingerprint, Authoritative: true, + }} + } + + result := make([]TargetConfig, 0, len(raw)) + seenURLs := make(map[string]struct{}, len(raw)) + seenNames := make(map[string]struct{}, len(raw)) + authoritative := 0 + for index, target := range raw { + name := strings.TrimSpace(target.Name) + if name == "" { + if index == 0 { + name = "primary" + } else { + name = fmt.Sprintf("observer-%d", index) + } + } + if _, exists := seenNames[name]; exists { + return nil, fmt.Errorf("duplicate Pulse target name %q", name) + } + if index == 0 && !target.Authoritative { + target.Authoritative = true + } + url, err := agenttarget.NormalizePulseURL( + strings.TrimSpace(target.URL), + target.Authoritative, + target.AllowPlaintextHTTP, + ) + if err != nil { + return nil, fmt.Errorf("invalid Pulse target %q URL: %w", name, err) + } + if _, exists := seenURLs[url]; exists { + return nil, fmt.Errorf("duplicate Pulse target URL %q", url) + } + token := strings.TrimSpace(target.Token) + if token == "" { + return nil, fmt.Errorf("Pulse target %q API token is required", name) + } + if target.Authoritative { + authoritative++ + } + target.Name = name + target.URL = url + target.Token = token + target.CACertPath = strings.TrimSpace(target.CACertPath) + target.ServerFingerprint = strings.TrimSpace(target.ServerFingerprint) + seenNames[name] = struct{}{} + seenURLs[url] = struct{}{} + result = append(result, target) + } + if authoritative != 1 { + return nil, fmt.Errorf("exactly one authoritative Pulse target is required (got %d)", authoritative) + } + return result, nil +} + +func newKubernetesHTTPClient(target TargetConfig) (*http.Client, error) { + tlsConfig, err := agenttls.NewClientTLSConfig(target.CACertPath, target.InsecureSkipVerify, target.ServerFingerprint) + if err != nil { + return nil, err + } + return &http.Client{ + Timeout: 15 * time.Second, + Transport: &http.Transport{Proxy: http.ProxyFromEnvironment, TLSClientConfig: tlsConfig}, + CheckRedirect: func(req *http.Request, _ []*http.Request) error { + return fmt.Errorf("server returned redirect to %s - use the final Pulse URL explicitly", req.URL) + }, + }, nil +} + func buildRESTConfig(kubeconfigPath, kubeContext string) (*rest.Config, string, error) { kubeconfigPath = strings.TrimSpace(kubeconfigPath) kubeContext = strings.TrimSpace(kubeContext) @@ -335,8 +451,16 @@ func (a *Agent) discoverClusterMetadata(ctx context.Context) error { } func (a *Agent) closeIdleConnections() { - if a.httpClient != nil { - a.httpClient.CloseIdleConnections() + if len(a.targets) == 0 { + if a.httpClient != nil { + a.httpClient.CloseIdleConnections() + } + return + } + for _, target := range a.targets { + if target.client != nil { + target.client.CloseIdleConnections() + } } } @@ -359,10 +483,20 @@ func (a *Agent) Run(ctx context.Context) error { } func (a *Agent) bufferedReportCount() int { - if a == nil || a.reportBuffer == nil { + if a == nil { return 0 } - return a.reportBuffer.Len() + if len(a.targets) == 0 { + if a.reportBuffer == nil { + return 0 + } + return a.reportBuffer.Len() + } + total := 0 + for _, target := range a.targets { + total += target.buffer.Len() + } + return total } func (a *Agent) runOnce(ctx context.Context) { @@ -379,59 +513,75 @@ func (a *Agent) runOnce(ctx context.Context) { return } - if err := a.sendReport(ctx, report); err != nil { - a.logger.Warn(). - Err(err). - Str("phase", "send_report"). - Str("cluster_id", a.clusterID). - Str("agent_id", a.agentID). - Int("report_nodes", len(report.Nodes)). - Int("report_pods", len(report.Pods)). - Int("report_deployments", len(report.Deployments)). - Int("buffer_depth_before", a.bufferedReportCount()). - Msg("Failed to send Kubernetes report, buffering") - a.reportBuffer.Push(report) - a.logger.Debug(). - Str("phase", "buffer_report"). - Str("cluster_id", a.clusterID). - Int("buffer_depth_after", a.bufferedReportCount()). - Msg("Buffered Kubernetes report for retry") + for _, target := range a.targets { + if err := a.sendReportToTarget(ctx, report, target); err != nil { + agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), false) + target.buffer.Push(report) + a.logger.Warn().Err(err). + Str("destination", target.config.Name). + Bool("authoritative", target.config.Authoritative). + Int("buffer_depth", target.buffer.Len()). + Msg("Failed to send Kubernetes report, buffering only for this destination") + continue + } + agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), true) + a.flushTargetReports(ctx, target) + } + if len(a.targets) == 0 { + if err := a.sendReport(ctx, report); err != nil { + a.reportBuffer.Push(report) + } } } +func kubernetesTargetRole(target TargetConfig) string { + if target.Authoritative { + return "primary" + } + return "observer" +} + func (a *Agent) flushReports(ctx context.Context) { - flushed := 0 - for { - report, ok := a.reportBuffer.Peek() - if !ok { - if flushed > 0 { - a.logger.Debug(). - Str("phase", "flush_buffered_reports"). - Str("cluster_id", a.clusterID). - Str("agent_id", a.agentID). - Int("flushed_reports", flushed). - Int("buffer_depth_remaining", a.bufferedReportCount()). - Msg("Flushed buffered Kubernetes reports") + if len(a.targets) == 0 { + for { + report, ok := a.reportBuffer.Peek() + if !ok || a.sendReport(ctx, report) != nil { + return } + a.reportBuffer.Pop() + } + } + for _, target := range a.targets { + a.flushTargetReports(ctx, target) + } +} + +func (a *Agent) flushTargetReports(ctx context.Context, target *reportTarget) { + for { + report, ok := target.buffer.Peek() + if !ok { return } - if err := a.sendReport(ctx, report); err != nil { + if err := a.sendReportToTarget(ctx, report, target); err != nil { + agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), false) a.logger.Warn(). Err(err). + Str("destination", target.config.Name). Str("phase", "flush_buffered_report"). Str("cluster_id", a.clusterID). Str("agent_id", a.agentID). Int("report_nodes", len(report.Nodes)). Int("report_pods", len(report.Pods)). Int("report_deployments", len(report.Deployments)). - Int("buffer_depth", a.bufferedReportCount()). + Int("buffer_depth", target.buffer.Len()). Msg("Failed to flush buffered Kubernetes report") return } - if _, ok := a.reportBuffer.Pop(); !ok { + if _, ok := target.buffer.Pop(); !ok { a.logger.Debug().Msg("Failed to remove buffered report after successful send") return } + agenttarget.MarkDelivery("kubernetes", target.config.Name, kubernetesTargetRole(target.config), true) } } @@ -3476,6 +3626,19 @@ func isProblemDeployment(dep appsv1.Deployment) bool { } func (a *Agent) sendReport(ctx context.Context, report agentsk8s.Report) (retErr error) { + for _, target := range a.targets { + if target.config.Authoritative { + return a.sendReportToTarget(ctx, report, target) + } + } + // Preserve direct-construction test and legacy behavior. + return a.sendReportToTarget(ctx, report, &reportTarget{ + config: TargetConfig{URL: a.pulseURL, Token: a.cfg.APIToken, Authoritative: true}, + client: a.httpClient, + }) +} + +func (a *Agent) sendReportToTarget(ctx context.Context, report agentsk8s.Report, target *reportTarget) (retErr error) { payload, err := json.Marshal(report) if err != nil { return fmt.Errorf("marshal report: %w", err) @@ -3486,7 +3649,7 @@ func (a *Agent) sendReport(ctx context.Context, report agentsk8s.Report) (retErr return fmt.Errorf("compress report: %w", err) } - reportURL := fmt.Sprintf("%s/api/agents/kubernetes/report", a.pulseURL) + reportURL := fmt.Sprintf("%s/api/agents/kubernetes/report", target.config.URL) req, err := http.NewRequestWithContext(ctx, http.MethodPost, reportURL, bytes.NewReader(compressed)) if err != nil { return fmt.Errorf("create request for %s: %w", reportURL, err) @@ -3494,11 +3657,15 @@ func (a *Agent) sendReport(ctx context.Context, report agentsk8s.Report) (retErr req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Encoding", "gzip") - req.Header.Set("Authorization", "Bearer "+a.cfg.APIToken) - req.Header.Set("X-API-Token", a.cfg.APIToken) + req.Header.Set("Authorization", "Bearer "+target.config.Token) + req.Header.Set("X-API-Token", target.config.Token) req.Header.Set("User-Agent", reportUserAgent+a.agentVersion) - resp, err := a.httpClient.Do(req) + client := target.client + if client == nil { + client = a.httpClient + } + resp, err := client.Do(req) if err != nil { return fmt.Errorf("send request to %s: %w", reportURL, err) } diff --git a/internal/kubernetesagent/multi_target_test.go b/internal/kubernetesagent/multi_target_test.go new file mode 100644 index 000000000..7d977b05a --- /dev/null +++ b/internal/kubernetesagent/multi_target_test.go @@ -0,0 +1,52 @@ +package kubernetesagent + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/utils" + agentsk8s "github.com/rcourtman/pulse-go-rewrite/pkg/agents/kubernetes" + "github.com/rs/zerolog" +) + +func TestKubernetesDestinationBuffersAreIndependent(t *testing.T) { + primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) + defer primary.Close() + observer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) })) + defer observer.Close() + + primaryTarget := &reportTarget{config: TargetConfig{Name: "primary", URL: primary.URL, Token: "p", Authoritative: true}, client: primary.Client(), buffer: utils.New[agentsk8s.Report](10)} + observerTarget := &reportTarget{config: TargetConfig{Name: "dev", URL: observer.URL, Token: "o"}, client: observer.Client(), buffer: utils.New[agentsk8s.Report](10)} + a := &Agent{logger: zerolog.Nop(), agentVersion: "test", targets: []*reportTarget{primaryTarget, observerTarget}, reportBuffer: primaryTarget.buffer} + report := agentsk8s.Report{} + for _, target := range a.targets { + if err := a.sendReportToTarget(context.Background(), report, target); err != nil { + target.buffer.Push(report) + } + } + if primaryTarget.buffer.Len() != 0 || observerTarget.buffer.Len() != 1 { + t.Fatalf("buffer depths primary=%d observer=%d", primaryTarget.buffer.Len(), observerTarget.buffer.Len()) + } +} + +func TestKubernetesObserverPlaintextPolicyIsDestinationScoped(t *testing.T) { + cfg := Config{ + Targets: []TargetConfig{ + {Name: "primary", URL: "https://primary.example.test", Token: "p", Authoritative: true}, + {Name: "observer", URL: "http://203.0.113.10:7655", Token: "o"}, + }, + } + if _, err := normalizeKubernetesTargets(cfg); err == nil { + t.Fatal("expected observer plaintext URL rejection without destination opt-in") + } + cfg.Targets[1].AllowPlaintextHTTP = true + normalized, err := normalizeKubernetesTargets(cfg) + if err != nil { + t.Fatalf("explicit observer plaintext opt-in: %v", err) + } + if !normalized[1].AllowPlaintextHTTP { + t.Fatal("observer plaintext policy was not preserved after normalization") + } +} diff --git a/scripts/install.ps1 b/scripts/install.ps1 index 56d7de28b..dc5a4e11d 100644 --- a/scripts/install.ps1 +++ b/scripts/install.ps1 @@ -20,6 +20,7 @@ param ( [bool]$Uninstall = $false, [string]$CACertPath = $env:PULSE_CACERT, [string]$ServerFingerprint = $env:PULSE_SERVER_FINGERPRINT, + [string]$ObserversFile = $env:PULSE_OBSERVERS_FILE, [string]$AgentId = $env:PULSE_AGENT_ID, [string]$Hostname = $env:PULSE_HOSTNAME, [string]$TokenFile = $env:PULSE_TOKEN_FILE, @@ -436,6 +437,9 @@ function Save-ConnectionState { if (-not [string]::IsNullOrWhiteSpace($ServerFingerprint)) { $lines += "PULSE_SERVER_FINGERPRINT='$ServerFingerprint'" } + if (-not [string]::IsNullOrWhiteSpace($ObserversFile)) { + $lines += "PULSE_OBSERVERS_FILE='$ObserversFile'" + } New-Item -ItemType Directory -Path $StateDir -Force | Out-Null Set-Content -Path $ConnectionStatePath -Value ($lines -join "`n") -Encoding UTF8 @@ -614,6 +618,18 @@ if ([string]::IsNullOrWhiteSpace($Token) -and -not [string]::IsNullOrWhiteSpace( } } +if (-not [string]::IsNullOrWhiteSpace($ObserversFile)) { + try { + $ObserversFile = $ExecutionContext.SessionState.Path.GetUnresolvedProviderPathFromPSPath($ObserversFile) + if (-not [System.IO.Path]::IsPathRooted($ObserversFile) -or -not (Test-Path $ObserversFile -PathType Leaf)) { + throw "path must be absolute and identify an existing file" + } + } catch { + Show-Error "Invalid observer config file.`nProvided: $ObserversFile`nError: $_" + Exit 1 + } +} + # --- Input Validation --- Write-Host "Validating parameters..." -ForegroundColor Cyan @@ -936,6 +952,7 @@ if ($EnableCommands) { $ServiceArgs += "--enable-commands" } if ($Insecure) { $ServiceArgs += "--insecure" } if (-not [string]::IsNullOrWhiteSpace($CACertPath)) { $ServiceArgs += @("--cacert", "`"$CACertPath`"") } if (-not [string]::IsNullOrWhiteSpace($ServerFingerprint)) { $ServiceArgs += @("--server-fingerprint", "`"$ServerFingerprint`"") } +if (-not [string]::IsNullOrWhiteSpace($ObserversFile)) { $ServiceArgs += @("--observers-file", "`"$ObserversFile`"") } if (-not [string]::IsNullOrWhiteSpace($AgentId)) { $ServiceArgs += @("--agent-id", "`"$AgentId`"") } if (-not [string]::IsNullOrWhiteSpace($Hostname)) { $ServiceArgs += @("--hostname", "`"$Hostname`"") } $ServiceArgs += @("--log-file", "`"$LogFile`"") diff --git a/scripts/install.sh b/scripts/install.sh index 3e474da30..9ca019d62 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -23,6 +23,7 @@ # --disk-exclude Exclude mount points matching pattern (repeatable) # --insecure Skip TLS certificate verification # --server-fingerprint Pin the Pulse server leaf certificate +# --observers-file Report to additional observer Pulse instances # --enable-commands Enable Pulse command execution on agent (disabled by default; required for Patrol actions and Proxmox LXC Docker inventory) # --health-addr Health/metrics listener address (default: 127.0.0.1:9191, use "" to disable) # --update Update an existing agent using saved connection state @@ -79,6 +80,7 @@ UPDATE_ONLY="false" UNINSTALL="false" INSECURE="false" SERVER_FINGERPRINT="${PULSE_SERVER_FINGERPRINT:-}" +OBSERVERS_FILE="${PULSE_OBSERVERS_FILE:-}" AGENT_ID="" HOSTNAME_OVERRIDE="" ENABLE_COMMANDS="false" @@ -304,6 +306,7 @@ Options: --insecure Skip TLS verification (auto-enabled for http:// URLs) --cacert Custom CA certificate for TLS (used by curl and agent) --server-fingerprint Pin the Pulse server leaf certificate for agent connections + --observers-file Absolute path to private JSON config for report-only observer Pulse destinations --enable-commands Enable Pulse command execution (disabled by default; required for Patrol actions and Proxmox LXC Docker inventory) --health-addr Health/metrics listener address (default: 127.0.0.1:9191; use "" to disable) --enroll Exchange bootstrap token for runtime token (deploy wizard) @@ -1451,6 +1454,7 @@ build_exec_arg_items() { if [[ -n "$PROXMOX_TYPE" ]]; then EXEC_ARG_ITEMS+=(--proxmox-type "$PROXMOX_TYPE"); fi if [[ "$INSECURE" == "true" ]]; then EXEC_ARG_ITEMS+=(--insecure); fi if [[ -n "$SERVER_FINGERPRINT" ]]; then EXEC_ARG_ITEMS+=(--server-fingerprint "$SERVER_FINGERPRINT"); fi + if [[ -n "$OBSERVERS_FILE" ]]; then EXEC_ARG_ITEMS+=(--observers-file "$OBSERVERS_FILE"); fi if [[ "$ENABLE_COMMANDS" == "true" ]]; then EXEC_ARG_ITEMS+=(--enable-commands); fi if [[ "$HEALTH_ADDR_SET" == "true" ]]; then EXEC_ARG_ITEMS+=(--health-addr "$HEALTH_ADDR"); fi if [[ "$ENROLL" == "true" ]]; then EXEC_ARG_ITEMS+=(--enroll); fi @@ -1718,6 +1722,10 @@ apply_recovered_agent_arg_value() { if [[ -z "$SERVER_FINGERPRINT" ]]; then SERVER_FINGERPRINT="$value"; fi RECOVERED_AGENT_ARG_STATE="true" ;; + observers-file) + if [[ -z "$OBSERVERS_FILE" ]]; then OBSERVERS_FILE="$value"; fi + RECOVERED_AGENT_ARG_STATE="true" + ;; health-addr) if [[ "$HEALTH_ADDR_SET" != "true" ]]; then HEALTH_ADDR="$value" @@ -1772,10 +1780,10 @@ recover_connection_state_from_arg_stream() { fi case "$arg" in - --url|--pulse-url|--token|--token-file|--interval|--agent-id|--hostname|--cacert|--server-fingerprint|--health-addr|--state-dir|--kubeconfig|--proxmox-type|--disk-exclude|-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude) + --url|--pulse-url|--token|--token-file|--interval|--agent-id|--hostname|--cacert|--server-fingerprint|--observers-file|--health-addr|--state-dir|--kubeconfig|--proxmox-type|--disk-exclude|-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-observers-file|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude) pending_key=$(normalize_recovered_agent_arg_key "$arg") ;; - --url=*|--pulse-url=*|--token=*|--token-file=*|--interval=*|--agent-id=*|--hostname=*|--cacert=*|--server-fingerprint=*|--health-addr=*|--state-dir=*|--kubeconfig=*|--proxmox-type=*|--disk-exclude=*|-url=*|-pulse-url=*|-token=*|-token-file=*|-interval=*|-agent-id=*|-hostname=*|-cacert=*|-server-fingerprint=*|-health-addr=*|-state-dir=*|-kubeconfig=*|-proxmox-type=*|-disk-exclude=*) + --url=*|--pulse-url=*|--token=*|--token-file=*|--interval=*|--agent-id=*|--hostname=*|--cacert=*|--server-fingerprint=*|--observers-file=*|--health-addr=*|--state-dir=*|--kubeconfig=*|--proxmox-type=*|--disk-exclude=*|-url=*|-pulse-url=*|-token=*|-token-file=*|-interval=*|-agent-id=*|-hostname=*|-cacert=*|-server-fingerprint=*|-observers-file=*|-health-addr=*|-state-dir=*|-kubeconfig=*|-proxmox-type=*|-disk-exclude=*) key="${arg%%=*}" value="${arg#*=}" apply_recovered_agent_arg_value "$key" "$value" @@ -2288,6 +2296,7 @@ while [[ $# -gt 0 ]]; do --insecure) INSECURE="true"; shift ;; --cacert) CURL_CA_BUNDLE="$2"; shift 2 ;; --server-fingerprint) SERVER_FINGERPRINT="$2"; shift 2 ;; + --observers-file) OBSERVERS_FILE="$2"; shift 2 ;; --enable-commands) ENABLE_COMMANDS="true"; shift ;; --health-addr) HEALTH_ADDR="$2"; HEALTH_ADDR_SET="true"; shift 2 ;; --enroll) ENROLL="true"; shift ;; @@ -2308,6 +2317,15 @@ while [[ $# -gt 0 ]]; do esac done +if [[ -n "$OBSERVERS_FILE" ]]; then + if [[ "$OBSERVERS_FILE" != /* ]]; then + fail "Observer config path must be absolute: ${OBSERVERS_FILE}" "$EXIT_MISSING_ARGS" + fi + if [[ ! -f "$OBSERVERS_FILE" || -L "$OBSERVERS_FILE" ]]; then + fail "Observer config must be a regular non-symlink file: ${OBSERVERS_FILE}" "$EXIT_MISSING_ARGS" + fi +fi + # Read token from file if --token-file was provided if [[ -n "$TOKEN_FILE_PATH" ]]; then if [[ ! -f "$TOKEN_FILE_PATH" ]]; then diff --git a/scripts/installtests/install_ps1_test.go b/scripts/installtests/install_ps1_test.go index 5c25583b6..e89e5dc2e 100644 --- a/scripts/installtests/install_ps1_test.go +++ b/scripts/installtests/install_ps1_test.go @@ -136,6 +136,27 @@ func TestInstallPS1PersistsAndVerifiesServerFingerprint(t *testing.T) { } } +func TestInstallPS1PreservesObserverDestinationConfig(t *testing.T) { + content, err := os.ReadFile(repoFile("scripts", "install.ps1")) + if err != nil { + t.Fatalf("read install.ps1: %v", err) + } + + script := string(content) + required := []string{ + `[string]$ObserversFile = $env:PULSE_OBSERVERS_FILE,`, + `$lines += "PULSE_OBSERVERS_FILE='$ObserversFile'"`, + `[System.IO.Path]::IsPathRooted($ObserversFile)`, + `Test-Path $ObserversFile -PathType Leaf`, + `$ServiceArgs += @("--observers-file", "` + "`" + `"$ObserversFile` + "`" + `"")`, + } + for _, needle := range required { + if !strings.Contains(script, needle) { + t.Fatalf("install.ps1 missing observer-config lifecycle contract: %s", needle) + } + } +} + func TestInstallPS1AllowsMissingTokenForOptionalAuth(t *testing.T) { content, err := os.ReadFile(repoFile("scripts", "install.ps1")) if err != nil { diff --git a/scripts/installtests/install_sh_test.go b/scripts/installtests/install_sh_test.go index 2655b1bbc..c06121436 100644 --- a/scripts/installtests/install_sh_test.go +++ b/scripts/installtests/install_sh_test.go @@ -711,7 +711,7 @@ func TestInstallSHSupportsSavedStateUpdateMode(t *testing.T) { `recover_connection_state_from_arg_stream`, `recover_token_from_default_agent_token_file() {`, `normalize_recovered_agent_arg_key() {`, - `-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude)`, + `-url|-pulse-url|-token|-token-file|-interval|-agent-id|-hostname|-cacert|-server-fingerprint|-observers-file|-health-addr|-state-dir|-kubeconfig|-proxmox-type|-disk-exclude)`, `--enable-host|-enable-host|--enable-host=true|-enable-host=true)`, `recover_connection_state_from_env_stream`, `recovered_connection_state_ready() {`, diff --git a/scripts/release_control/canonical_completion_guard_test.py b/scripts/release_control/canonical_completion_guard_test.py index 664b0f973..62a8f2706 100644 --- a/scripts/release_control/canonical_completion_guard_test.py +++ b/scripts/release_control/canonical_completion_guard_test.py @@ -556,6 +556,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "internal/hostagent/commands_host_update_test.go", "internal/hostagent/commands_storage_cleanup_test.go", "internal/hostagent/docker_lifecycle_test.go", + "internal/hostagent/observer_delivery_test.go", "internal/hostagent/package_updates_test.go", "internal/hostagent/send_report_test.go", "internal/hostagent/storage_cleanup_test.go", diff --git a/scripts/release_control/subsystem_lookup_test.py b/scripts/release_control/subsystem_lookup_test.py index af702eb48..bbf1756cf 100644 --- a/scripts/release_control/subsystem_lookup_test.py +++ b/scripts/release_control/subsystem_lookup_test.py @@ -4437,6 +4437,7 @@ class SubsystemLookupTest(unittest.TestCase): "internal/hostagent/commands_host_update_test.go", "internal/hostagent/commands_storage_cleanup_test.go", "internal/hostagent/docker_lifecycle_test.go", + "internal/hostagent/observer_delivery_test.go", "internal/hostagent/package_updates_test.go", "internal/hostagent/send_report_test.go", "internal/hostagent/storage_cleanup_test.go",