gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-09 18:15:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix agent command channel admission

Browse Source
This commit is contained in:
rcourtman
2026-07-24 12:55:44 +01:00
parent 478a9e933d
commit c41edb65a0
46 changed files with 1629 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/AGENT_SECURITY.md
+1 -1
View File
@@ -159,4 +159,4 @@ Unified agent (`pulse-agent`):
- **Token Authentication**: All agent-to-server communication requires a valid API token.
- **TLS**: Encrypted by default (unless specifically disabled).
- **Network Isolation (optional)**: Agent check-in can be served on a dedicated, separately firewalled port that exposes only the agent-ingest routes (`/api/agents/*`), so a host that can reach the agent endpoint over an untrusted network cannot pivot to the web UI or management API. See [Split-Port Agent Ingest](CONFIGURATION.md#split-port-agent-ingest-network-isolation).
- **Network Isolation (optional)**: The agent control plane can be served on a dedicated, separately firewalled port. It exposes the bounded report/config, command WebSocket, version, and bootstrap routes needed for the full agent lifecycle, but not the web UI or management API. See [Split-Port Agent Ingest](CONFIGURATION.md#split-port-agent-ingest-network-isolation).
docs/CONFIGURATION.md
+4 -4
View File
@@ -202,7 +202,7 @@ Environment variables take precedence over `system.json`.
| ---------- | ------------- | --------- |
| `FRONTEND_PORT` | Public listening port (web UI, API, and agent ingest) | `7655` |
| `PORT` | **Deprecated** legacy alias for `FRONTEND_PORT`, honored only when `FRONTEND_PORT` is unset. Logs a deprecation warning at startup; switch to `FRONTEND_PORT`. | *(unset)* |
| `PULSE_AGENT_INGEST_PORT` | Optional dedicated port that serves **only** agent ingest (`/api/agents/*`), network-isolated from the web UI and the rest of the API. `0` = disabled (single port). See [Split-Port Agent Ingest](#split-port-agent-ingest-network-isolation). | `0` |
| `PULSE_AGENT_INGEST_PORT` | Optional dedicated port for the complete agent control plane: reports/config (`/api/agents/*`), command admission (`/api/agent/ws`), version checks, and bootstrap downloads. The web UI and management API stay isolated. `0` = disabled (single port). See [Split-Port Agent Ingest](#split-port-agent-ingest-network-isolation). | `0` |
| `LOG_LEVEL` | Log verbosity (see below) | `info` |
| `LOG_FORMAT` | Log output format (`auto`, `json`, `console`) | `auto` |
| `LOG_FILE` | Log file path (enables file logging) | *(unset)* |
@@ -249,7 +249,7 @@ Environment variables take precedence over `system.json`.
### Split-Port Agent Ingest (Network Isolation)
By default Pulse serves the web UI, the REST API, and agent check-in together on `FRONTEND_PORT`. For deployments that expose Pulse to monitored hosts across an untrusted network (for example, a managed service provider whose clients' Proxmox nodes reach a central Pulse server over the internet), you can move agent check-in onto its own dedicated port and keep the web UI and management API on a separate, firewalled port.
By default Pulse serves the web UI, the REST API, and the agent control plane together on `FRONTEND_PORT`. For deployments that expose Pulse to monitored hosts across an untrusted network (for example, a managed service provider whose clients' Proxmox nodes reach a central Pulse server over the internet), you can expose the agent control plane on its own dedicated port and keep the web UI and management API on a separate, firewalled port.
Set `PULSE_AGENT_INGEST_PORT` to a port other than `FRONTEND_PORT`:
@@ -259,7 +259,7 @@ PULSE_AGENT_INGEST_PORT=7656
When enabled:
- The dedicated port serves **only** the agent-ingest routes (`/api/agents/*`). Every other path, including the web UI, login, and the management API, returns `404`. A host that can reach the agent port cannot pivot to the management interface.
- The dedicated port serves only the agent-owned routes required for a complete lifecycle: `/api/agents/*`, `/api/agent/ws`, `/api/agent/version`, `/api/server/info`, `/install.sh`, `/install.ps1`, and `/download/pulse-agent`. Every other path, including the web UI, login, and management APIs, returns `404`. A host that can reach the agent port cannot pivot to the management interface.
- The main `FRONTEND_PORT` listener is unchanged and still serves everything (including agent ingest), so existing single-port installs keep working. The dedicated listener is purely additive.
- The value is validated at startup: it must be between 1 and 65535 and must differ from `FRONTEND_PORT` and the HTTP redirect port. An invalid value is rejected.
@@ -270,7 +270,7 @@ PULSE_AGENT_INGEST_PORT=7656
PULSE_AGENT_CONNECT_URL=https://agents.example.com:7656
```
Agents then post telemetry to `https://agents.example.com:7656/api/agents/agent/report`, while the web UI and management API remain reachable only on the private `FRONTEND_PORT` listener.
Agents then post telemetry to `https://agents.example.com:7656/api/agents/agent/report` and establish their command channel at `wss://agents.example.com:7656/api/agent/ws`, while the web UI and management API remain reachable only on the private `FRONTEND_PORT` listener. If command execution is enabled, both routes must traverse the same proxy/firewall path; a successful report does not prove that the WebSocket is admitted.
### Iframe Embedding (system.json)
docs/MSP.md
+14 -8
View File
@@ -84,7 +84,7 @@ for the full reference.
```bash
FRONTEND_PORT=7655 # management UI + API: private network / VPN only
PULSE_AGENT_INGEST_PORT=7656 # agent check-in only: reachable from client sites
PULSE_AGENT_INGEST_PORT=7656 # agent reports + command/control: reachable from client sites
PULSE_AGENT_CONNECT_URL=https://agents.example.com:7656
```
@@ -93,14 +93,16 @@ Firewall baseline:
| Surface | Port | Reachable from |
|---------|------|----------------|
| Management UI + API | `FRONTEND_PORT` (7655) | Provider staff network / VPN only |
| Agent ingest | `PULSE_AGENT_INGEST_PORT` (7656) | Client sites (or client VPN tunnels) |
| Agent control plane | `PULSE_AGENT_INGEST_PORT` (7656) | Client sites (or client VPN tunnels) |
| Prometheus metrics | 9091 | Provider monitoring network only |
The dedicated agent port serves **only** `/api/agents/*`; every other path,
including login and the management API, returns `404`. Agent check-in
authenticates with an `agent:report`-scoped API token, which cannot read
monitoring data or change settings — the token scope and the port isolation
are independent layers.
The dedicated agent port serves only the report/config, command WebSocket,
version, and bootstrap routes documented in
[Split-Port Agent Ingest](CONFIGURATION.md#split-port-agent-ingest-network-isolation);
login and management APIs return `404`. Report and command access use separate
least-privilege scopes (`agent:report` and `agent:exec`) on the same
host-bound enrollment token. Token scope, immutable command-session binding,
and port isolation are independent layers.
If agents reach the central server over per-client VPN tunnels instead of the
public internet, the same split still applies: expose only the agent port into
@@ -108,13 +110,17 @@ the tunnels and keep the management port out of them.
### Validation checklist (run after setup, repeat after network changes)
1. **Agent port serves agent ingest only.** Both must return `404`:
1. **Agent port excludes management surfaces.** Both must return `404`:
```bash
curl -sk -o /dev/null -w '%{http_code}\n' https://agents.example.com:7656/ # 404
curl -sk -o /dev/null -w '%{http_code}\n' https://agents.example.com:7656/api/login # 404
```
When commands are enabled, also verify that `/api/agent/ws` reaches Pulse
through the proxy. Agent Doctor reports the command channel as disconnected
if telemetry is current but WebSocket admission is absent.
2. **Management port is not reachable from a client site.** From a client
network (or through a client tunnel), a connection to `FRONTEND_PORT` must
time out or be refused by your firewall — not answer.
docs/MULTI_TENANT.md
+1 -1
View File
@@ -134,7 +134,7 @@ To onboard an internal estate:
2. **Create an org-bound API token** with the `agent:report` scope, bound to that estate's organization (`orgId`). A token bound to a single organization automatically routes every agent that uses it into that organization, with no extra header required. Binding also scopes the token: an org-bound token cannot access other organizations, including the default org (bind `default` explicitly if a token genuinely needs it). Legacy unbound tokens keep their default-org access.
3. **Install the estate's agents** (Proxmox host, Docker, Kubernetes) using that token. Their telemetry lands in the selected organization.
4. **(Optional) Alias node names per estate.** If two estates both use the default `pve` hostname and you want them visually distinct, set `--hostname` (or the `PULSE_HOSTNAME` environment variable) on the agent, for example `--hostname "acme-pve1"`. See [UNIFIED_AGENT.md](UNIFIED_AGENT.md).
5. **(Optional) Isolate agent check-in on its own port.** When remote nodes reach the central server across the internet, enable [Split-Port Agent Ingest](CONFIGURATION.md#split-port-agent-ingest-network-isolation) so agents connect on a dedicated, firewalled port that exposes only `/api/agents/*` and never the web UI or management API.
5. **(Optional) Isolate the agent control plane on its own port.** When remote nodes reach the central server across the internet, enable [Split-Port Agent Ingest](CONFIGURATION.md#split-port-agent-ingest-network-isolation) so reports and command WebSockets share a dedicated, firewalled agent port that never exposes the web UI or management API.
Route each estate's alerts into the right internal system with per-organization webhooks or the org-scoped alerts API. See the multi-tenant section of [WEBHOOKS.md](WEBHOOKS.md).
docs/release-control/v6/internal/subsystems/agent-lifecycle.md
+30 -3
View File
@@ -1230,11 +1230,20 @@ the intentionally sparse public response.
job, target, and expected node, but not the human owner.
Command-agent WebSocket identity follows the same lifecycle/auth split:
install-command tokens that enable command execution may be minted before
the final agent ID is known, but only Pulse-minted PVE/PBS install-command
tokens may bind on first `/api/agent/ws` registration. That first-use bind
the final agent ID is known, but only Pulse-minted PVE/PBS or host
install-command tokens may bind on first `/api/agent/ws` registration. That first-use bind
must persist the registering agent ID and hostname and become authoritative
for later command registration attempts; generic unbound `agent:exec`
tokens are not lifecycle credentials and must fail closed.
tokens are not lifecycle credentials and must fail closed. A pre-v6.1.1
hostname-bound record carrying a server-synthesized ID may migrate to the
registering runtime ID exactly once; the versioned result must then require
both fields. The live registry keys the session by organization plus agent
ID, retains only non-secret token identity, isolates equal IDs across
organizations, rejects same-organization hostname collisions, and
revalidates token existence, expiry, scope, organization, and binding before
reporting connectivity or dispatching work. Reconnect replacement and stale
socket cleanup must use pointer identity so an old reader cannot remove the
newly admitted session.
The canonical durable-principal vocabulary for those shared auth routes is
recorded in `docs/release-control/v6/internal/IDENTITY_INVARIANTS.md`; agent
lifecycle work may consume that identity but must not define a parallel
@@ -1872,6 +1881,16 @@ the intentionally sparse public response.
## Completion Obligations
Command-capable agent completion must prove more than fresh telemetry. The
dedicated agent listener must admit the full bootstrap/report/WebSocket
lifecycle, command sessions must be keyed by organization plus canonical bound
identity, and every dispatch must revalidate the non-secret token admission.
Reconnect replacement may replace only that exact tenant identity; duplicate
hostnames, stale sockets, revoked or rebound tokens, and ambiguous hostname or
token resolution fail closed. Fleet/Doctor projections must keep adapter health
separate from command-channel admission and report an enabled-but-disconnected
command runtime as a blocking condition.
Any Docker / Podman report-size change must update the shared contract, agent
diagnostic proof, API encoded/decoded boundary proof, and the API-contract
dependency in one slice. A handler-local literal, agent-local threshold, or
@@ -2191,6 +2210,14 @@ structured fleet diagnostics that have no ledger binding (agents reporting
only workload telemetry, e.g. Docker-only or Kubernetes-only agents) are now
appended as diagnostics-only targets, honoring the scoped-agent filter, so a
critical workload-only agent can no longer vanish from the fleet view.
Telemetry liveness and an applied `commandsEnabled` report are not command
connectivity. The connections ledger must consult the tenant- and
token-scoped live session registry: an enabled report without an admitted
socket projects `remoteControl="disconnected"` and a blocked command policy,
while a deliberately disabled command policy remains disabled rather than
becoming a transport alarm. Agent Doctor merges that ledger fact with
structured diagnostics, so a diagnostic snapshot marked healthy cannot hide
the critical `command_channel_disconnected` reason.
Diagnostics-only rows render the diagnostic's status, reasons, and evidence
but offer no host-local update command (there is no ledger connection to
derive an update from). Removed rows are the deliberate exception in the
docs/release-control/v6/internal/subsystems/ai-runtime.md
+8
View File
@@ -3198,6 +3198,14 @@ query...`, and `Reading storage...` before streamed tool arguments are
## Completion Obligations
Every per-organization Assistant or legacy AI service that can discover or
dispatch through the host-agent command transport must receive an
organization-pinned command-server view. A tenant service must never enumerate
the default organization or another tenant and must not rely on an unscoped
agent ID to select a session. Loss, revocation, or replacement of the pinned
session fails the invocation before command dispatch while the ordinary
approval, action-audit, timeout, and cancellation contracts remain in force.
Qualification floor: Patrol model launch and product-claim qualification must use
`cmd/patrol-qualify` against reviewed scenario-owned ground truth. Expected
faults and independent postconditions must be declared before the run and
docs/release-control/v6/internal/subsystems/api-contracts.md
+18
View File
@@ -3018,6 +3018,15 @@ a new API state machine, queue contract, or verification-accounting field.
## Completion Obligations
The public connection ledger and action APIs must project telemetry liveness
and command admission independently. An agent may be adapter-healthy while
remote control is `disconnected`; in that state command policy is blocked and
action dispatch returns the canonical unavailable/disconnected error rather
than inferring command readiness from reports. Resource actions resolve the
live session inside the request organization, prefer the immutable report
token binding when present, and refuse stale-token fallback to hostname or
agent ID.
Docker / Podman report transport changes must prove uncompressed and gzip
ingress, exact inclusive boundaries, encoded and decoded 413 rejection,
ordinary and 163-container Docker/Podman fleets, and typed compression
@@ -6316,6 +6325,15 @@ request/response surface is the Pulse Unified Agent route family, while
`/api/agents/host/*` stays a compatibility alias and must not leak back into
handler naming, router-owned state, or proof labels as if it were a second
product-facing API surface.
The optional split listener must carry that report family and the exact agent
lifecycle routes `/api/agent/ws`, `/api/agent/version`, `/api/server/info`,
`/install.sh`, `/install.ps1`, and `/download/pulse-agent`; allowing reports
while returning `404` for the command WebSocket is an invalid partial control
plane. The connections response must not infer `fleet.remoteControl="enabled"`
from the report flag alone. It projects `disconnected` when command policy is
applied enabled but the report token has no current admitted session in the
request organization, and returns a blocked command-policy reason so API and
Agent Doctor clients cannot present report health as action readiness.
That confirmation marker must survive the legacy setup-script transport too:
script-generated `/api/auto-register` payloads must send `source="script"`,
and canonical callers must send that source explicitly, so later canonical
docs/release-control/v6/internal/subsystems/performance-and-scalability.md
+8
View File
@@ -834,6 +834,14 @@ shell clickable behind another overlay.
## Completion Obligations
Command-channel routing must remain bounded by tenant-scoped session keys.
Router consumers may enumerate only the current organization and must resolve
token or hostname aliases to exactly one revalidated live connection before
dispatch. Reconnect replacement and token invalidation remove the indexed
session without fleet-wide retry loops, and pending request correlation uses
the same scoped key so concurrent tenants can reuse local agent/request IDs
without contention or cross-delivery.
1. Update benchmarks, SLOs, or query-plan tests when hot-path behavior changes
2. Update this contract when a new protected hot path is adopted
3. Route runtime changes through the explicit performance proof policies in `registry.json`; default fallback proof routing is not allowed
docs/release-control/v6/internal/subsystems/security-privacy.md
+32 -9
View File
@@ -471,6 +471,15 @@ the `white_label` branding entitlement.
## Completion Obligations
Split-port agent exposure is complete only when its exact allowlist covers
report/config routes, command WebSocket admission, version/server bootstrap,
and the supported installer/download endpoints while every UI and management
route remains unavailable. Registration binds one execution-scoped token to
one organization, agent ID, and hostname; the server stores only the resulting
non-secret admission, revalidates it before visibility or dispatch, and rejects
multi-organization authority, identity drift, replay after migration, revoked
tokens, and path-normalization variants.
1. Update privacy/security docs and the telemetry runtime together when outbound-data behavior changes.
2. Keep shared API-contract proof routing aligned whenever auth, token, or telemetry settings payloads change.
3. Keep shared frontend settings proof routing aligned whenever security/privacy presentation changes.
@@ -1471,17 +1480,31 @@ Stopping the watcher is the synchronization point that lets tests and runtime
teardown restore auth/config state without racing a background reload.
That same server-bind config boundary now also owns optional agent-ingest
network isolation. `internal/config/config.go` may accept
`PULSE_AGENT_INGEST_PORT` as a dedicated listener for agent report and
management traffic so operators can place `/api/agents/*` on its own network or
firewall boundary, but the option must fail closed at validation: the agent
`PULSE_AGENT_INGEST_PORT` as a dedicated listener for the bounded agent control
plane so operators can place report/config traffic, command WebSocket
admission, version checks, and bootstrap downloads on one network or firewall
boundary, but the option must fail closed at validation: the agent
ingest port stays disabled at `0`, must be a valid `1`-`65535` port, and must
differ from both the frontend port and any HTTP redirect port. When that
listener is active the runtime must serve only the `/api/agents/*` surface on
it and must never expose the web UI or the rest of the REST API through that
port, so a port reachable from an untrusted agent network cannot widen into the
operator console. Enabling the dedicated port is additive: the main listener
keeps serving agent ingest too, so the default single-port deployment and
existing agents are unaffected.
listener is active the runtime must serve only `/api/agents/*` plus the exact
agent-owned `/api/agent/ws`, `/api/agent/version`, `/api/server/info`,
`/install.sh`, `/install.ps1`, and `/download/pulse-agent` routes. It must never
expose the web UI or management REST API through that port, so a port reachable
from an untrusted agent network cannot widen into the operator console.
Enabling the dedicated port is additive: the main listener keeps serving the
agent control plane too, so the default single-port deployment and existing
agents are unaffected.
Command admission is a separate security fact from report health. A live
session is owned by one organization, one token identity, one runtime agent ID,
and one hostname. Multi-organization exec tokens are ambiguous and must be
rejected; duplicate IDs remain isolated across organizations but fail closed
inside one organization when hostnames conflict. Legacy hostname-bound tokens
may migrate a synthesized ID to the first observed runtime ID once, after
which both identity fields are immutable. Revoked, expired, re-scoped, or
re-bound token records invalidate an existing socket before connectivity is
reported or work is dispatched. Raw bearer values must not be retained in the
session registry.
Locale-catalog additions for shared mobile copy controls remain contract-neutral
to security and privacy only while they preserve every governed token, scope,
docs/release-control/v6/internal/subsystems/storage-recovery.md
+19
View File
@@ -1613,6 +1613,15 @@ recovery scope, or a storage/recovery-owned secret source.
## Completion Obligations
Agent-backed lifecycle recovery must namespace every in-memory pending request,
typed result, operation query, and deployment progress subscription by the
admitted organization session. That transport namespace must not alter the
durable receipt identity: action ID, attempt ID, operation kind/version,
request digest, subject, and canonical agent ID remain immutable. Reconnect and
restart recovery may query the owning live session for a terminal receipt, but
may never resend an ambiguous mutation or accept a response from another
tenant/session.
Legacy RBAC JSON import is an adjacent security-owned migration, not recovery
inventory or restore evidence. When shared `internal/api/` construction
triggers that import, validation and SQLite writes must complete atomically,
@@ -2025,6 +2034,16 @@ Storage and recovery consumers may observe action detail, pending/settled
projections, and correlated receipt state, but may not infer restore,
verification, rollback, or compensation truth from those transport-only fields
or create a local retry path when the core attempt is `receipt_pending`.
The adjacent command transport now namespaces live sessions and pending
response channels by organization plus agent identity, while durable receipt
identity remains the immutable attempt/action/operation/digest/agent tuple.
Action dispatch and query contexts must carry the owning organization into
that transport; token rotation, server restart, socket replacement, or stale
reader cleanup may make the executor unavailable, but must not rewrite the
bound attempt, admit a cross-tenant response, replay a committed mutation, or
convert report health into receipt evidence. Recovery continues query-only
against the same bound operation and fails closed while its admitted session
is absent.
Unified Agent lifecycle fields added to the shared host and connections API are
adjacent monitoring/API state only. Applied config fingerprints, updater
frontend-modern/public/docs/CONFIGURATION.md
+6 -4
View File
@@ -80,6 +80,7 @@ You can pre-configure Pulse by setting environment variables. Plain text credent
```bash
# Docker Example
docker run -d \
-e PULSE_DEPLOYMENT_METHOD=docker_run \
-e PULSE_AUTH_USER=admin \
-e PULSE_AUTH_PASS=secret123 \
rcourtman/pulse:latest
@@ -201,7 +202,7 @@ Environment variables take precedence over `system.json`.
| ---------- | ------------- | --------- |
| `FRONTEND_PORT` | Public listening port (web UI, API, and agent ingest) | `7655` |
| `PORT` | **Deprecated** legacy alias for `FRONTEND_PORT`, honored only when `FRONTEND_PORT` is unset. Logs a deprecation warning at startup; switch to `FRONTEND_PORT`. | *(unset)* |
| `PULSE_AGENT_INGEST_PORT` | Optional dedicated port that serves **only** agent ingest (`/api/agents/*`), network-isolated from the web UI and the rest of the API. `0` = disabled (single port). See [Split-Port Agent Ingest](#split-port-agent-ingest-network-isolation). | `0` |
| `PULSE_AGENT_INGEST_PORT` | Optional dedicated port for the complete agent control plane: reports/config (`/api/agents/*`), command admission (`/api/agent/ws`), version checks, and bootstrap downloads. The web UI and management API stay isolated. `0` = disabled (single port). See [Split-Port Agent Ingest](#split-port-agent-ingest-network-isolation). | `0` |
| `LOG_LEVEL` | Log verbosity (see below) | `info` |
| `LOG_FORMAT` | Log output format (`auto`, `json`, `console`) | `auto` |
| `LOG_FILE` | Log file path (enables file logging) | *(unset)* |
@@ -248,7 +249,7 @@ Environment variables take precedence over `system.json`.
### Split-Port Agent Ingest (Network Isolation)
By default Pulse serves the web UI, the REST API, and agent check-in together on `FRONTEND_PORT`. For deployments that expose Pulse to monitored hosts across an untrusted network (for example, a managed service provider whose clients' Proxmox nodes reach a central Pulse server over the internet), you can move agent check-in onto its own dedicated port and keep the web UI and management API on a separate, firewalled port.
By default Pulse serves the web UI, the REST API, and the agent control plane together on `FRONTEND_PORT`. For deployments that expose Pulse to monitored hosts across an untrusted network (for example, a managed service provider whose clients' Proxmox nodes reach a central Pulse server over the internet), you can expose the agent control plane on its own dedicated port and keep the web UI and management API on a separate, firewalled port.
Set `PULSE_AGENT_INGEST_PORT` to a port other than `FRONTEND_PORT`:
@@ -258,7 +259,7 @@ PULSE_AGENT_INGEST_PORT=7656
When enabled:
- The dedicated port serves **only** the agent-ingest routes (`/api/agents/*`). Every other path, including the web UI, login, and the management API, returns `404`. A host that can reach the agent port cannot pivot to the management interface.
- The dedicated port serves only the agent-owned routes required for a complete lifecycle: `/api/agents/*`, `/api/agent/ws`, `/api/agent/version`, `/api/server/info`, `/install.sh`, `/install.ps1`, and `/download/pulse-agent`. Every other path, including the web UI, login, and management APIs, returns `404`. A host that can reach the agent port cannot pivot to the management interface.
- The main `FRONTEND_PORT` listener is unchanged and still serves everything (including agent ingest), so existing single-port installs keep working. The dedicated listener is purely additive.
- The value is validated at startup: it must be between 1 and 65535 and must differ from `FRONTEND_PORT` and the HTTP redirect port. An invalid value is rejected.
@@ -269,7 +270,7 @@ PULSE_AGENT_INGEST_PORT=7656
PULSE_AGENT_CONNECT_URL=https://agents.example.com:7656
```
Agents then post telemetry to `https://agents.example.com:7656/api/agents/agent/report`, while the web UI and management API remain reachable only on the private `FRONTEND_PORT` listener.
Agents then post telemetry to `https://agents.example.com:7656/api/agents/agent/report` and establish their command channel at `wss://agents.example.com:7656/api/agent/ws`, while the web UI and management API remain reachable only on the private `FRONTEND_PORT` listener. If command execution is enabled, both routes must traverse the same proxy/firewall path; a successful report does not prove that the WebSocket is admitted.
### Iframe Embedding (system.json)
@@ -308,6 +309,7 @@ When `allowEmbedding` is `false`, Pulse sends `X-Frame-Options: DENY` and `frame
| `PULSE_ENABLE_PROXMOX_GUEST_DOCKER_INVENTORY` | Allow Proxmox-side minimal LXC Docker inventory collection with `pct exec`; collects Docker host/container summary, not inspect/env/mount/process data | `false` |
| `PULSE_PROXMOX_GUEST_DOCKER_INVENTORY_VMIDS` | Optional comma-separated VMID allowlist for Proxmox-side LXC Docker inventory; empty means all running Docker-enabled LXCs are eligible when inventory is enabled | *(unset)* |
| `PULSE_TELEMETRY` | Outbound usage telemetry ([details](PRIVACY.md)); set `false` to disable | `true` |
| `PULSE_DEPLOYMENT_METHOD` | Optional closed telemetry label: `docker_compose`, `docker_run`, `container_other`, `systemd`, `binary_other`, or `other`; invalid values are reported only as the safe runtime fallback | Inferred as `container_other` or `binary_other` |
### Logging Overrides
frontend-modern/src/api/__tests__/connections.test.ts
+10 -11
View File
@@ -153,7 +153,7 @@ describe('ConnectionsAPI', () => {
configRollout: 'reported',
credentialStatus: 'verified',
updateStatus: 'update-available',
remoteControl: 'enabled',
remoteControl: 'disconnected',
configDrift: {
status: 'drifted',
desired: { version: 'host-agent-config/v1', hash: 'sha256:desired' },
@@ -172,12 +172,11 @@ describe('ConnectionsAPI', () => {
lastVerifiedAt: '2026-04-22T20:00:00Z',
},
commandPolicy: {
status: 'enabled',
desired: 'disabled',
status: 'blocked',
desired: 'enabled',
applied: 'enabled',
enforcement: 'drifted',
reason:
'agent still reports command execution enabled while desired policy disables it',
enforcement: 'blocked',
reason: 'agent command channel is not connected',
},
},
capabilities: { supportsPause: false, supportsScope: false, supportsTest: false },
@@ -195,7 +194,7 @@ describe('ConnectionsAPI', () => {
configRollout: 'reported',
credentialStatus: 'verified',
updateStatus: 'update-available',
remoteControl: 'enabled',
remoteControl: 'disconnected',
configDrift: {
status: 'drifted',
desired: { version: 'host-agent-config/v1', hash: 'sha256:desired' },
@@ -214,11 +213,11 @@ describe('ConnectionsAPI', () => {
lastVerifiedAt: '2026-04-22T20:00:00Z',
},
commandPolicy: {
status: 'enabled',
desired: 'disabled',
status: 'blocked',
desired: 'enabled',
applied: 'enabled',
enforcement: 'drifted',
reason: 'agent still reports command execution enabled while desired policy disables it',
enforcement: 'blocked',
reason: 'agent command channel is not connected',
},
});
});
frontend-modern/src/api/connections.ts
+2 -1
View File
@@ -25,7 +25,8 @@ export type ConnectionFleetUpdateStatus =
| 'unknown'
| 'update-available'
| 'updating';
export type ConnectionFleetRemoteControl = 'disabled' | 'enabled' | 'not-applicable' | 'unknown';
export type ConnectionFleetRemoteControl =
'disabled' | 'disconnected' | 'enabled' | 'not-applicable' | 'unknown';
export type ConnectionFleetConfigDriftStatus =
'current' | 'drifted' | 'not-applicable' | 'paused' | 'pending' | 'unknown';
export type ConnectionFleetRolloutStatus =
frontend-modern/src/components/Settings/__tests__/infrastructureAgentDoctorModel.test.ts
+40
View File
@@ -121,6 +121,46 @@ describe('Agent Doctor model', () => {
expect(targets[0].commandBlockedReason).toBeUndefined();
});
it('does not report healthy when telemetry is fresh but command admission is disconnected', () => {
const connection = connectionFixture({
agentUpdateAvailable: false,
agentVersion: '6.1.1',
expectedAgentVersion: '6.1.1',
agentIdentity: {
hostname: 'host-1',
platform: 'ubuntu',
architecture: 'amd64',
commandsEnabled: true,
},
fleet: {
versionDrift: 'current',
remoteControl: 'disconnected',
commandPolicy: {
status: 'blocked',
desired: 'enabled',
applied: 'enabled',
enforcement: 'blocked',
reason: 'Agent reports commands enabled, but no admitted command channel is connected.',
},
} as Connection['fleet'],
});
const [target] = collectInfrastructureAgentDoctorTargets({
rows: [rowFixture(connection)],
connections: [connection],
diagnostics: [diagnosticFixture({ status: 'healthy', reasons: [] })],
diagnosticsAvailable: true,
targetVersion: '6.1.1',
});
expect(target.status).toBe('critical');
expect(target.reasons).toContainEqual(
expect.objectContaining({
code: 'command_channel_disconnected',
severity: 'critical',
}),
);
});
it('falls back to ledger classification and blocks commands for unknown platforms', () => {
const connection = connectionFixture({
state: 'stale',
frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts
+2
View File
@@ -108,11 +108,13 @@ const settingsRuntimeSources = import.meta.glob(['../*.tsx', '../ConnectionEdito
describe('settings architecture guardrails', () => {
it('keeps Unified Agent lifecycle failures on the shared connections contract', () => {
expect(connectionsApiSource).toContain("| 'failed'");
expect(connectionsApiSource).toContain("'disabled' | 'disconnected' | 'enabled'");
expect(connectionsApiSource).toContain('agentUpdate?: ConnectionAgentUpdateStatus;');
expect(connectionsApiSource).toContain('agentModules?: ConnectionAgentModuleStatus[];');
expect(connectionsTableModelSource).toContain("key: 'module-health'");
expect(connectionsTableModelSource).toContain("label: 'Agent update failed'");
expect(connectionsTableModelSource).toContain('update?.lastError');
expect(connectionsTableModelSource).toContain("case 'disconnected':");
});
it('keeps Settings on the canonical page shell boundary', () => {
frontend-modern/src/components/Settings/connectionsTableModel.ts
+8
View File
@@ -321,6 +321,14 @@ const commandPolicyFromFleet = (fleet: ConnectionFleetGovernance): ConnectionFle
if (fleet.commandPolicy) return fleet.commandPolicy;
switch (fleet.remoteControl) {
case 'disconnected':
return {
status: 'blocked',
desired: 'unknown',
applied: 'enabled',
enforcement: 'blocked',
reason: 'The agent reports command execution enabled, but no command channel is connected.',
};
case 'enabled':
return {
status: 'enabled',
frontend-modern/src/components/Settings/infrastructureAgentUpdateCommandsModel.ts
+31 -2
View File
@@ -333,6 +333,20 @@ const ledgerFallbackReasons = (
);
}
}
if (
connection.fleet?.remoteControl === 'disconnected' ||
(connection.fleet?.commandPolicy?.status === 'blocked' &&
connection.agentIdentity?.commandsEnabled)
) {
reasons.push(
fallbackReason(
'command_channel_disconnected',
'critical',
'The agent is reporting, but its command channel is not connected.',
connection.fleet?.commandPolicy?.reason ? [connection.fleet.commandPolicy.reason] : [],
),
);
}
if (needsUpdate) {
reasons.push(
fallbackReason(
@@ -463,7 +477,17 @@ const doctorTargetFromBinding = (
const expectedVersion = expectedVersionFor(connection, targetVersion);
const needsUpdate = connectionNeedsUpdate(connection, expectedVersion);
const fallbackReasons = ledgerFallbackReasons(connection, needsUpdate);
const reasons = diagnosticsAvailable && diagnostic ? (diagnostic.reasons ?? []) : fallbackReasons;
const reasons =
diagnosticsAvailable && diagnostic
? Array.from(
new Map(
[
...fallbackReasons.filter((reason) => reason.code === 'command_channel_disconnected'),
...(diagnostic.reasons ?? []),
].map((reason) => [reason.code, reason]),
).values(),
)
: fallbackReasons;
const updater = updaterPresentation(connection, diagnostic, needsUpdate);
let status: InfrastructureAgentDoctorStatus =
diagnosticsAvailable && diagnostic
@@ -472,7 +496,12 @@ const doctorTargetFromBinding = (
// The ledger can advance one poll ahead of diagnostics. Never hide a live
// update/error signal while the structured endpoint catches up.
if (status === 'healthy' && fallbackReasons.length > 0) status = 'warning';
const fallbackDoctorStatus = fallbackStatus(connection, fallbackReasons);
if (fallbackDoctorStatus === 'critical') {
status = 'critical';
} else if (status === 'healthy' && fallbackReasons.length > 0) {
status = 'warning';
}
const nonVersionReasons = reasons.filter((reason) => reason.code !== 'agent_version_stale');
if (
internal/agentexec/operation_query_test.go
+1 -1
View File
@@ -27,7 +27,7 @@ func registeredTestAgent(t *testing.T, s *Server, agentID string) (*websocket.Co
if err != nil {
t.Fatal(err)
}
wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{AgentID: agentID, Hostname: "host", Version: "6", Platform: "linux", Token: "ok", OperationReceiptVersion: operationreceipt.ProtocolVersion}))
wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{AgentID: agentID, Hostname: agentID + "-host", Version: "6", Platform: "linux", Token: "ok", OperationReceiptVersion: operationreceipt.ProtocolVersion}))
if !wsReadRegisteredPayload(t, conn).Success {
t.Fatal("registration failed")
}
internal/agentexec/server.go
+418 -87
View File
@@ -62,7 +62,7 @@ var hostStorageCleanupFingerprintPattern = regexp.MustCompile(`^sha256:[a-f0-9]{
// Server manages WebSocket connections from agents
type Server struct {
mu sync.RWMutex
agents map[string]*agentConn // agentID -> connection
agents map[string]*agentConn // organizationID + agentID -> connection
pendingReqs map[string]chan CommandResultPayload // scoped request key -> response channel
pendingHostStorageCleanups map[string]chan HostStorageCleanupResultPayload // scoped request key -> typed storage-cleanup response
pendingHostUpdates map[string]chan HostUpdateResultPayload // scoped request key -> typed host-update response
@@ -71,7 +71,8 @@ type Server struct {
pendingHostOperations map[string]pendingHostOperation // scoped request key -> exact typed APT operation/query identity
pendingOperationQueries map[string]pendingOperationQuery
deploySubs map[string]chan DeployProgressPayload // deploySubKey(agentID, jobID) -> progress subscriber
validateToken func(token string, agentID string, hostname string) bool
admitToken AgentRegistrationValidator
validateSession AgentSessionValidator
commandPolicy *CommandPolicy
ipConnCounts map[string]int
maxConnsPerIP int
@@ -81,9 +82,31 @@ type Server struct {
commandAuthorizationVerifier func(CommandAuthorizationRequest) error
newCommandApprovalGrant func([]byte, string, ExecuteCommandPayload, time.Time, time.Duration) (*CommandApprovalGrant, error)
now func() time.Time
agentRegisteredNotifier func(agentID string)
agentRegisteredNotifier func(AgentAdmission)
}
const defaultOrganizationID = "default"
type organizationContextKey struct{}
// AgentAdmission is the immutable server-owned identity of an admitted command
// session. The raw bearer token is deliberately not retained after
// registration.
type AgentAdmission struct {
OrganizationID string
TokenID string
AgentID string
Hostname string
}
// AgentRegistrationValidator authenticates and binds a registration to one
// organization, token, agent identity, and hostname.
type AgentRegistrationValidator func(token string, agentID string, hostname string) (AgentAdmission, bool)
// AgentSessionValidator revalidates the non-secret admission immediately
// before the server treats a socket as connected or dispatches work to it.
type AgentSessionValidator func(AgentAdmission) bool
// CommandAuthorizationRequest is the complete server-side approval scope
// verified and consumed immediately before an approval grant is signed.
type CommandAuthorizationRequest struct {
@@ -99,6 +122,8 @@ type CommandAuthorizationRequest struct {
type agentConn struct {
conn *websocket.Conn
agent ConnectedAgent
admission AgentAdmission
sessionKey string
approvalGrantKey []byte
writeMu sync.Mutex
done chan struct{}
@@ -143,6 +168,26 @@ func NewServer(validateToken func(token string, agentID string, hostname string)
panic("agentexec: validateToken is required")
}
return NewServerWithAdmissionValidator(func(token string, agentID string, hostname string) (AgentAdmission, bool) {
if !validateToken(token, agentID, hostname) {
return AgentAdmission{}, false
}
return AgentAdmission{
OrganizationID: defaultOrganizationID,
AgentID: strings.TrimSpace(agentID),
Hostname: strings.TrimSpace(hostname),
}, true
}, nil)
}
// NewServerWithAdmissionValidator creates a command server whose sessions are
// tenant-scoped and can be invalidated after registration without retaining
// bearer tokens in memory.
func NewServerWithAdmissionValidator(admit AgentRegistrationValidator, validateSession AgentSessionValidator) *Server {
if admit == nil {
panic("agentexec: admission validator is required")
}
return &Server{
agents: make(map[string]*agentConn),
pendingReqs: make(map[string]chan CommandResultPayload),
@@ -153,7 +198,8 @@ func NewServer(validateToken func(token string, agentID string, hostname string)
pendingHostOperations: make(map[string]pendingHostOperation),
pendingOperationQueries: make(map[string]pendingOperationQuery),
deploySubs: make(map[string]chan DeployProgressPayload),
validateToken: validateToken,
admitToken: admit,
validateSession: validateSession,
commandPolicy: DefaultPolicy(),
ipConnCounts: make(map[string]int),
maxConnsPerIP: defaultMaxWebSocketConnectionsPerIP,
@@ -164,6 +210,83 @@ func NewServer(validateToken func(token string, agentID string, hostname string)
}
}
// WithOrganizationID scopes command-session lookup and dispatch to a tenant.
// Empty values normalize to the single-tenant default for compatibility.
func WithOrganizationID(ctx context.Context, organizationID string) context.Context {
if ctx == nil {
ctx = context.Background()
}
return context.WithValue(ctx, organizationContextKey{}, normalizeOrganizationID(organizationID))
}
func normalizeOrganizationID(organizationID string) string {
if organizationID = strings.TrimSpace(organizationID); organizationID != "" {
return organizationID
}
return defaultOrganizationID
}
func organizationIDFromContext(ctx context.Context) string {
if ctx != nil {
if organizationID, ok := ctx.Value(organizationContextKey{}).(string); ok {
return normalizeOrganizationID(organizationID)
}
}
return defaultOrganizationID
}
// OrganizationServer is a tenant-pinned view of a command server. It exists
// for consumers whose interface cannot carry the request context through
// discovery and dispatch as separate calls (for example, long-lived per-tenant
// Assistant services).
type OrganizationServer struct {
server *Server
organizationID string
}
// ForOrganization returns a command-server view that can only discover and
// dispatch sessions admitted to organizationID.
func (s *Server) ForOrganization(organizationID string) *OrganizationServer {
return &OrganizationServer{
server: s,
organizationID: normalizeOrganizationID(organizationID),
}
}
func (s *OrganizationServer) GetConnectedAgents() []ConnectedAgent {
if s == nil || s.server == nil {
return nil
}
return s.server.GetConnectedAgentsForOrganization(s.organizationID)
}
func (s *OrganizationServer) ExecuteCommand(ctx context.Context, agentID string, cmd ExecuteCommandPayload) (*CommandResultPayload, error) {
if s == nil || s.server == nil {
return nil, fmt.Errorf("agent execution server is unavailable")
}
return s.server.ExecuteCommand(WithOrganizationID(ctx, s.organizationID), agentID, cmd)
}
func agentSessionKey(organizationID, agentID string) string {
organizationID = normalizeOrganizationID(organizationID)
agentID = strings.TrimSpace(agentID)
if organizationID == defaultOrganizationID {
// Preserve the historical key for direct single-tenant users and tests.
return agentID
}
return organizationID + "\x00" + agentID
}
func connectionSessionKey(ac *agentConn) string {
if ac == nil {
return ""
}
if strings.TrimSpace(ac.sessionKey) != "" {
return ac.sessionKey
}
return agentSessionKey(ac.admission.OrganizationID, ac.agent.AgentID)
}
// SetCommandAuthorizationVerifier installs the server-owned authorization
// consumer used for approval-gated arbitrary commands.
func (s *Server) SetCommandAuthorizationVerifier(verifier func(CommandAuthorizationRequest) error) {
@@ -181,6 +304,20 @@ func (s *Server) SetCommandAuthorizationVerifier(verifier func(CommandAuthorizat
// goroutine because the query response can only be read once this server
// enters the connection's read loop.
func (s *Server) SetAgentRegisteredNotifier(notify func(agentID string)) {
if s == nil {
return
}
if notify == nil {
s.agentRegisteredNotifier = nil
return
}
s.agentRegisteredNotifier = func(admission AgentAdmission) {
notify(admission.AgentID)
}
}
// SetAgentAdmissionNotifier installs the tenant-aware registration callback.
func (s *Server) SetAgentAdmissionNotifier(notify func(AgentAdmission)) {
if s == nil {
return
}
@@ -200,6 +337,40 @@ func pendingRequestKey(agentID, requestID string) string {
return agentID + "\x00" + requestID
}
func (s *Server) connectionForOrganization(organizationID, agentID string) (*agentConn, bool) {
if s == nil {
return nil, false
}
key := agentSessionKey(organizationID, agentID)
s.mu.RLock()
ac, ok := s.agents[key]
s.mu.RUnlock()
if !ok {
return nil, false
}
if s.validateSession == nil || s.validateSession(ac.admission) {
return ac, true
}
// A revoked, expired, re-bound, or otherwise stale token must stop being a
// command authority immediately. Pointer equality prevents an old
// validation result from evicting a replacement session.
s.mu.Lock()
if current, exists := s.agents[key]; exists && current == ac {
delete(s.agents, key)
}
s.mu.Unlock()
ac.signalDone()
if ac.conn != nil {
_ = ac.conn.Close()
}
return nil, false
}
func (s *Server) connectionForContext(ctx context.Context, agentID string) (*agentConn, bool) {
return s.connectionForOrganization(organizationIDFromContext(ctx), agentID)
}
func (s *Server) claimPendingHostOperation(agentID, requestID, actionID, operation string) (string, error) {
key := pendingRequestKey(agentID, requestID)
s.mu.Lock()
@@ -223,11 +394,15 @@ func (s *Server) matchesPendingHostOperation(agentID, requestID, actionID, opera
}
func (s *Server) claimPendingDockerOperation(identity operationreceipt.Identity, containerID string) (string, error) {
return s.claimPendingDockerOperationForSession(identity.AgentID, identity, containerID)
}
func (s *Server) claimPendingDockerOperationForSession(sessionKey string, identity operationreceipt.Identity, containerID string) (string, error) {
identity, err := operationreceipt.NormalizeIdentity(identity)
if err != nil {
return "", err
}
key := pendingRequestKey(identity.AgentID, identity.AttemptID)
key := pendingRequestKey(sessionKey, identity.AttemptID)
s.mu.Lock()
defer s.mu.Unlock()
if _, exists := s.pendingHostOperations[key]; exists {
@@ -238,7 +413,11 @@ func (s *Server) claimPendingDockerOperation(identity operationreceipt.Identity,
}
func (s *Server) matchesPendingDockerOperation(agentID string, result DockerContainerLifecycleResultPayload) bool {
key := pendingRequestKey(agentID, result.RequestID)
return s.matchesPendingDockerOperationForSession(agentID, agentID, result)
}
func (s *Server) matchesPendingDockerOperationForSession(sessionKey, agentID string, result DockerContainerLifecycleResultPayload) bool {
key := pendingRequestKey(sessionKey, result.RequestID)
s.mu.RLock()
expected, ok := s.pendingHostOperations[key]
s.mu.RUnlock()
@@ -247,7 +426,11 @@ func (s *Server) matchesPendingDockerOperation(agentID string, result DockerCont
}
func (s *Server) matchesPendingDockerUpdateOperation(agentID string, result DockerContainerUpdateResultPayload) bool {
key := pendingRequestKey(agentID, result.RequestID)
return s.matchesPendingDockerUpdateOperationForSession(agentID, agentID, result)
}
func (s *Server) matchesPendingDockerUpdateOperationForSession(sessionKey, agentID string, result DockerContainerUpdateResultPayload) bool {
key := pendingRequestKey(sessionKey, result.RequestID)
s.mu.RLock()
expected, ok := s.pendingHostOperations[key]
s.mu.RUnlock()
@@ -723,8 +906,24 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
return
}
// Validate token
if !s.validateToken(reg.Token, reg.AgentID, reg.Hostname) {
// Validate and canonicalize the command-session admission. Reporting and
// command admission are intentionally separate trust decisions.
admission, admitted := s.admitToken(reg.Token, reg.AgentID, reg.Hostname)
admission.OrganizationID = normalizeOrganizationID(admission.OrganizationID)
admission.TokenID = strings.TrimSpace(admission.TokenID)
admission.AgentID = strings.TrimSpace(admission.AgentID)
admission.Hostname = strings.TrimSpace(admission.Hostname)
if admission.AgentID == "" {
admission.AgentID = reg.AgentID
}
if admission.Hostname == "" {
admission.Hostname = strings.TrimSpace(reg.Hostname)
}
if admission.AgentID != reg.AgentID ||
!unifiedresources.HostnamesEquivalent(admission.Hostname, reg.Hostname) {
admitted = false
}
if !admitted {
log.Warn().Str("agent_id", reg.AgentID).Msg("Agent registration rejected: invalid token")
// Actionable message instead of a bare "Invalid token": the agent logs
// this verbatim, and the dominant causes (token not recognised, or not
@@ -748,14 +947,18 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
ac := &agentConn{
conn: conn,
agent: ConnectedAgent{
AgentID: reg.AgentID,
Hostname: reg.Hostname,
OrganizationID: admission.OrganizationID,
TokenID: admission.TokenID,
AgentID: admission.AgentID,
Hostname: admission.Hostname,
Version: reg.Version,
Platform: reg.Platform,
Tags: reg.Tags,
ConnectedAt: time.Now(),
OperationReceiptVersion: reg.OperationReceiptVersion,
},
admission: admission,
sessionKey: agentSessionKey(admission.OrganizationID, admission.AgentID),
approvalGrantKey: DeriveApprovalGrantKey(reg.Token),
done: make(chan struct{}),
}
@@ -789,23 +992,59 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
// Register agent - after this point, other goroutines can access the connection
s.mu.Lock()
// Close existing connection if any
if existing, ok := s.agents[reg.AgentID]; ok {
log.Info().
Str("agent_id", reg.AgentID).
Str("hostname", reg.Hostname).
Msg("Replacing existing agent connection")
close(existing.done)
if err := existing.conn.Close(); err != nil {
log.Debug().Err(err).Str("agent_id", reg.AgentID).Msg("Failed to close existing connection during reconnect")
for key, existing := range s.agents {
if key != ac.sessionKey &&
normalizeOrganizationID(existing.admission.OrganizationID) == admission.OrganizationID &&
unifiedresources.HostnamesEquivalent(existing.agent.Hostname, ac.agent.Hostname) {
s.mu.Unlock()
log.Warn().
Str("organization_id", admission.OrganizationID).
Str("connected_agent_id", existing.agent.AgentID).
Str("requested_agent_id", admission.AgentID).
Str("hostname", admission.Hostname).
Msg("Agent registration rejected: hostname is already owned by another command identity")
rejectedMsg, err := NewMessage(MsgTypeRegistered, "", RegisteredPayload{Success: false, Message: "agent hostname is already connected under another identity"})
if err == nil {
_ = s.sendMessage(conn, rejectedMsg)
}
closeConn("Failed to close duplicate agent hostname connection")
return
}
}
s.agents[reg.AgentID] = ac
// Close existing connection if any
if existing, ok := s.agents[ac.sessionKey]; ok {
if !unifiedresources.HostnamesEquivalent(existing.agent.Hostname, ac.agent.Hostname) {
s.mu.Unlock()
log.Warn().
Str("organization_id", admission.OrganizationID).
Str("agent_id", admission.AgentID).
Str("connected_hostname", existing.agent.Hostname).
Str("requested_hostname", admission.Hostname).
Msg("Agent registration rejected: duplicate identity is already connected from another host")
rejectedMsg, err := NewMessage(MsgTypeRegistered, "", RegisteredPayload{Success: false, Message: "agent identity is already connected from another host"})
if err == nil {
_ = s.sendMessage(conn, rejectedMsg)
}
closeConn("Failed to close duplicate agent identity connection")
return
}
log.Info().
Str("organization_id", admission.OrganizationID).
Str("agent_id", admission.AgentID).
Str("hostname", admission.Hostname).
Msg("Replacing existing agent connection")
existing.signalDone()
if err := existing.conn.Close(); err != nil {
log.Debug().Err(err).Str("agent_id", admission.AgentID).Msg("Failed to close existing connection during reconnect")
}
}
s.agents[ac.sessionKey] = ac
s.mu.Unlock()
log.Info().
Str("agent_id", reg.AgentID).
Str("hostname", reg.Hostname).
Str("organization_id", admission.OrganizationID).
Str("agent_id", admission.AgentID).
Str("hostname", admission.Hostname).
Str("version", reg.Version).
Str("platform", reg.Platform).
Msg("Agent connected")
@@ -824,6 +1063,15 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
Str("agent_id", reg.AgentID).
Str("hostname", reg.Hostname).
Msg("Failed to send registration ack")
ac.writeMu.Unlock()
s.mu.Lock()
if existing, ok := s.agents[ac.sessionKey]; ok && existing == ac {
delete(s.agents, ac.sessionKey)
}
s.mu.Unlock()
ac.signalDone()
_ = conn.Close()
return
}
ac.writeMu.Unlock()
@@ -833,7 +1081,7 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
defer close(pingDone)
if notify := s.agentRegisteredNotifier; notify != nil {
go notify(reg.AgentID)
go notify(admission)
}
// Run read loop (blocking) - don't use goroutine, or HTTP handler will close connection
@@ -843,18 +1091,23 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
func (s *Server) readLoop(ac *agentConn) {
defer func() {
agentID := ac.agent.AgentID
sessionKey := connectionSessionKey(ac)
s.mu.Lock()
if existing, ok := s.agents[agentID]; ok && existing == ac {
delete(s.agents, agentID)
existing, sessionExists := s.agents[sessionKey]
ownsSession := !sessionExists || existing == ac
if sessionExists && existing == ac {
delete(s.agents, sessionKey)
}
// Close all deploy progress subscriptions for this agent so
// processPreflightProgress goroutines unblock and detect disconnect.
var closeChs []chan DeployProgressPayload
prefix := agentID + "\x00"
for key, ch := range s.deploySubs {
if strings.HasPrefix(key, prefix) {
closeChs = append(closeChs, ch)
delete(s.deploySubs, key)
if ownsSession {
prefix := sessionKey + "\x00"
for key, ch := range s.deploySubs {
if strings.HasPrefix(key, prefix) {
closeChs = append(closeChs, ch)
delete(s.deploySubs, key)
}
}
}
s.mu.Unlock()
@@ -926,7 +1179,7 @@ func (s *Server) readLoop(ac *agentConn) {
}
s.mu.RLock()
ch, ok := s.pendingReqs[pendingRequestKey(ac.agent.AgentID, result.RequestID)]
ch, ok := s.pendingReqs[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
s.mu.RUnlock()
if ok {
@@ -958,12 +1211,12 @@ func (s *Server) readLoop(ac *agentConn) {
log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid host update result")
continue
}
if !s.matchesPendingHostOperation(ac.agent.AgentID, result.RequestID, result.ActionID, HostUpdateOperationInstall) {
if !s.matchesPendingHostOperation(connectionSessionKey(ac), result.RequestID, result.ActionID, HostUpdateOperationInstall) {
log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated host update result")
continue
}
s.mu.RLock()
ch, ok := s.pendingHostUpdates[pendingRequestKey(ac.agent.AgentID, result.RequestID)]
ch, ok := s.pendingHostUpdates[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
s.mu.RUnlock()
if ok {
select {
@@ -979,12 +1232,12 @@ func (s *Server) readLoop(ac *agentConn) {
log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid host storage cleanup result")
continue
}
if !s.matchesPendingHostOperation(ac.agent.AgentID, result.RequestID, result.ActionID, HostStorageCleanupOperationPackageCache) {
if !s.matchesPendingHostOperation(connectionSessionKey(ac), result.RequestID, result.ActionID, HostStorageCleanupOperationPackageCache) {
log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated host storage cleanup result")
continue
}
s.mu.RLock()
ch, ok := s.pendingHostStorageCleanups[pendingRequestKey(ac.agent.AgentID, result.RequestID)]
ch, ok := s.pendingHostStorageCleanups[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
s.mu.RUnlock()
if ok {
select {
@@ -1000,12 +1253,12 @@ func (s *Server) readLoop(ac *agentConn) {
log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid docker container lifecycle result")
continue
}
if !s.matchesPendingDockerOperation(ac.agent.AgentID, result) {
if !s.matchesPendingDockerOperationForSession(connectionSessionKey(ac), ac.agent.AgentID, result) {
log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated docker lifecycle result")
continue
}
s.mu.RLock()
ch, ok := s.pendingDockerContainerLifecycles[pendingRequestKey(ac.agent.AgentID, result.RequestID)]
ch, ok := s.pendingDockerContainerLifecycles[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
s.mu.RUnlock()
if ok {
select {
@@ -1021,12 +1274,12 @@ func (s *Server) readLoop(ac *agentConn) {
log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid docker container update result")
continue
}
if !s.matchesPendingDockerUpdateOperation(ac.agent.AgentID, result) {
if !s.matchesPendingDockerUpdateOperationForSession(connectionSessionKey(ac), ac.agent.AgentID, result) {
log.Warn().Str("agent_id", ac.agent.AgentID).Str("request_id", result.RequestID).Msg("Dropping uncorrelated docker update result")
continue
}
s.mu.RLock()
ch, ok := s.pendingDockerContainerUpdates[pendingRequestKey(ac.agent.AgentID, result.RequestID)]
ch, ok := s.pendingDockerContainerUpdates[pendingRequestKey(connectionSessionKey(ac), result.RequestID)]
s.mu.RUnlock()
if ok {
select {
@@ -1042,7 +1295,7 @@ func (s *Server) readLoop(ac *agentConn) {
log.Warn().Err(decodeErr).Str("agent_id", ac.agent.AgentID).Msg("Dropping invalid operation query result")
continue
}
key := pendingRequestKey(ac.agent.AgentID, strings.TrimSpace(msg.ID))
key := pendingRequestKey(connectionSessionKey(ac), strings.TrimSpace(msg.ID))
s.mu.RLock()
pending, ok := s.pendingOperationQueries[key]
s.mu.RUnlock()
@@ -1072,7 +1325,7 @@ func (s *Server) readLoop(ac *agentConn) {
continue
}
subKey := deploySubKey(ac.agent.AgentID, progress.JobID)
subKey := deploySubKey(connectionSessionKey(ac), progress.JobID)
// Hold the read lock across map lookup AND the non-blocking send to
// prevent UnsubscribeDeployProgress from closing the channel between
@@ -1257,10 +1510,7 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
startedAt := time.Now()
s.mu.RLock()
ac, ok := s.agents[agentID]
s.mu.RUnlock()
ac, ok := s.connectionForContext(ctx, agentID)
if !ok {
log.Warn().
Str("agent_id", agentID).
@@ -1307,8 +1557,12 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
// Create response channel
respCh := make(chan CommandResultPayload, 1)
reqKey := pendingRequestKey(agentID, cmd.RequestID)
reqKey := pendingRequestKey(connectionSessionKey(ac), cmd.RequestID)
s.mu.Lock()
if _, exists := s.pendingReqs[reqKey]; exists {
s.mu.Unlock()
return nil, fmt.Errorf("command request %q is already pending", cmd.RequestID)
}
s.pendingReqs[reqKey] = respCh
s.mu.Unlock()
@@ -1360,7 +1614,7 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
Dur("duration", time.Since(startedAt)).
Msg("Agent command completed")
return &result, nil
case <-time.After(timeout):
case <-timer.C:
execLog.Warn().
Dur("timeout", timeout).
Dur("duration", time.Since(startedAt)).
@@ -1372,6 +1626,8 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
Dur("duration", time.Since(startedAt)).
Msg("Agent command canceled")
return nil, ctx.Err()
case <-ac.done:
return nil, fmt.Errorf("agent %s disconnected before command result", agentID)
case <-s.shutdown:
return nil, errServerShuttingDown
}
@@ -1426,9 +1682,7 @@ func prepareHostOperationRequest(s *Server, agentID string, requestID *string, b
func dispatchHostOperation[Req hostOperationPayload, Res any](ctx context.Context, s *Server, agentID string, req Req, op hostOperationDispatch[Req, Res]) (*Res, error) {
requestID, actionID, operation, timeoutSeconds := req.hostOperationIdentity()
s.mu.RLock()
ac, ok := s.agents[agentID]
s.mu.RUnlock()
ac, ok := s.connectionForContext(ctx, agentID)
if !ok {
return nil, fmt.Errorf("agent %s not connected", agentID)
}
@@ -1437,8 +1691,9 @@ func dispatchHostOperation[Req hostOperationPayload, Res any](ctx context.Contex
}
respCh := make(chan Res, 1)
reqKey := pendingRequestKey(agentID, requestID)
hostOperationKey, err := s.claimPendingHostOperation(agentID, requestID, actionID, operation)
sessionKey := connectionSessionKey(ac)
reqKey := pendingRequestKey(sessionKey, requestID)
hostOperationKey, err := s.claimPendingHostOperation(sessionKey, requestID, actionID, operation)
if err != nil {
return nil, err
}
@@ -1557,9 +1812,7 @@ func dispatchTypedDockerContainerOperation[Res any](
pending map[string]chan Res, label string,
validate func(Res) error,
) (*Res, error) {
s.mu.RLock()
ac, ok := s.agents[agentID]
s.mu.RUnlock()
ac, ok := s.connectionForContext(ctx, agentID)
if !ok {
return nil, fmt.Errorf("agent %s not connected", agentID)
}
@@ -1568,8 +1821,9 @@ func dispatchTypedDockerContainerOperation[Res any](
}
respCh := make(chan Res, 1)
reqKey := pendingRequestKey(agentID, requestID)
hostOperationKey, err := s.claimPendingDockerOperation(identity, containerID)
sessionKey := connectionSessionKey(ac)
reqKey := pendingRequestKey(sessionKey, requestID)
hostOperationKey, err := s.claimPendingDockerOperationForSession(sessionKey, identity, containerID)
if err != nil {
return nil, err
}
@@ -1652,12 +1906,16 @@ func (s *Server) currentTime() time.Time {
}
func (s *Server) AgentOperationReceiptVersion(agentID string) int {
return s.AgentOperationReceiptVersionForOrganization(defaultOrganizationID, agentID)
}
// AgentOperationReceiptVersionForOrganization reports the live protocol
// version only for a currently admitted tenant-scoped session.
func (s *Server) AgentOperationReceiptVersionForOrganization(organizationID, agentID string) int {
if s == nil {
return 0
}
s.mu.RLock()
defer s.mu.RUnlock()
connection, ok := s.agents[strings.TrimSpace(agentID)]
connection, ok := s.connectionForOrganization(organizationID, agentID)
if !ok {
return 0
}
@@ -1674,9 +1932,7 @@ func (s *Server) QueryAgentOperation(ctx context.Context, agentID string, identi
if identity.AgentID != agentID {
return operationreceipt.QueryResult{}, operationreceipt.ErrBindingConflict
}
s.mu.RLock()
ac, ok := s.agents[agentID]
s.mu.RUnlock()
ac, ok := s.connectionForContext(ctx, agentID)
if !ok {
return operationreceipt.QueryResult{}, fmt.Errorf("agent %s not connected", agentID)
}
@@ -1684,7 +1940,7 @@ func (s *Server) QueryAgentOperation(ctx context.Context, agentID string, identi
return operationreceipt.QueryResult{}, fmt.Errorf("agent does not support durable operation receipts")
}
queryID := identity.AttemptID + ".query." + uuid.NewString()
key := pendingRequestKey(agentID, queryID)
key := pendingRequestKey(connectionSessionKey(ac), queryID)
ch := make(chan operationreceipt.QueryResult, 1)
s.mu.Lock()
if _, exists := s.pendingOperationQueries[key]; exists {
@@ -1734,10 +1990,7 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
return nil, err
}
s.mu.RLock()
ac, ok := s.agents[agentID]
s.mu.RUnlock()
ac, ok := s.connectionForContext(ctx, agentID)
if !ok {
log.Warn().
Str("agent_id", agentID).
@@ -1759,8 +2012,12 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
// Create response channel
respCh := make(chan CommandResultPayload, 1)
reqKey := pendingRequestKey(agentID, req.RequestID)
reqKey := pendingRequestKey(connectionSessionKey(ac), req.RequestID)
s.mu.Lock()
if _, exists := s.pendingReqs[reqKey]; exists {
s.mu.Unlock()
return nil, fmt.Errorf("read_file request %q is already pending", req.RequestID)
}
s.pendingReqs[reqKey] = respCh
s.mu.Unlock()
@@ -1819,6 +2076,8 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
return nil, fmt.Errorf("read_file timed out after %v", timeout)
case <-ctx.Done():
return nil, fmt.Errorf("read_file %q on agent %q canceled: %w", req.RequestID, agentID, ctx.Err())
case <-ac.done:
return nil, fmt.Errorf("agent %s disconnected before read_file result", agentID)
case <-s.shutdown:
return nil, errServerShuttingDown
}
@@ -1826,33 +2085,100 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
// GetConnectedAgents returns a list of currently connected agents
func (s *Server) GetConnectedAgents() []ConnectedAgent {
s.mu.RLock()
defer s.mu.RUnlock()
return s.GetConnectedAgentsForOrganization(defaultOrganizationID)
}
agents := make([]ConnectedAgent, 0, len(s.agents))
// GetConnectedAgentsForOrganization returns only currently admitted sessions
// owned by one tenant.
func (s *Server) GetConnectedAgentsForOrganization(organizationID string) []ConnectedAgent {
if s == nil {
return nil
}
organizationID = normalizeOrganizationID(organizationID)
s.mu.RLock()
ids := make([]string, 0, len(s.agents))
for _, ac := range s.agents {
agents = append(agents, ac.agent)
if normalizeOrganizationID(ac.admission.OrganizationID) == organizationID {
ids = append(ids, ac.agent.AgentID)
}
}
s.mu.RUnlock()
agents := make([]ConnectedAgent, 0, len(ids))
for _, agentID := range ids {
if ac, ok := s.connectionForOrganization(organizationID, agentID); ok {
agents = append(agents, ac.agent)
}
}
return agents
}
// IsAgentConnected checks if an agent is currently connected
func (s *Server) IsAgentConnected(agentID string) bool {
s.mu.RLock()
defer s.mu.RUnlock()
_, ok := s.agents[agentID]
return s.IsAgentConnectedForOrganization(defaultOrganizationID, agentID)
}
// IsAgentConnectedForOrganization checks command-channel admission rather than
// telemetry liveness.
func (s *Server) IsAgentConnectedForOrganization(organizationID, agentID string) bool {
_, ok := s.connectionForOrganization(organizationID, agentID)
return ok
}
// GetAgentForHost finds the agent for a given hostname using the canonical
// hostname-equivalence contract shared with the unified identity layer.
func (s *Server) GetAgentForHost(hostname string) (string, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
return s.GetAgentForHostForOrganization(defaultOrganizationID, hostname)
}
// GetAgentForHostForOrganization resolves a hostname only within one tenant.
func (s *Server) GetAgentForHostForOrganization(organizationID, hostname string) (string, bool) {
if s == nil {
return "", false
}
organizationID = normalizeOrganizationID(organizationID)
s.mu.RLock()
ids := make([]string, 0, len(s.agents))
for _, ac := range s.agents {
if unifiedresources.HostnamesEquivalent(ac.agent.Hostname, hostname) {
return ac.agent.AgentID, true
if normalizeOrganizationID(ac.admission.OrganizationID) == organizationID &&
unifiedresources.HostnamesEquivalent(ac.agent.Hostname, hostname) {
ids = append(ids, ac.agent.AgentID)
}
}
s.mu.RUnlock()
if len(ids) != 1 {
return "", false
}
for _, agentID := range ids {
if _, ok := s.connectionForOrganization(organizationID, agentID); ok {
return agentID, true
}
}
return "", false
}
// GetAgentForTokenForOrganization resolves the canonical live command session
// for the same enrollment token that owns a telemetry resource.
func (s *Server) GetAgentForTokenForOrganization(organizationID, tokenID string) (string, bool) {
if s == nil || strings.TrimSpace(tokenID) == "" {
return "", false
}
organizationID = normalizeOrganizationID(organizationID)
tokenID = strings.TrimSpace(tokenID)
s.mu.RLock()
ids := make([]string, 0, 1)
for _, ac := range s.agents {
if normalizeOrganizationID(ac.admission.OrganizationID) == organizationID &&
strings.TrimSpace(ac.admission.TokenID) == tokenID {
ids = append(ids, ac.agent.AgentID)
}
}
s.mu.RUnlock()
if len(ids) != 1 {
return "", false
}
for _, agentID := range ids {
if _, ok := s.connectionForOrganization(organizationID, agentID); ok {
return agentID, true
}
}
return "", false
@@ -1864,12 +2190,16 @@ func (s *Server) GetAgentForHost(hostname string) (string, bool) {
// events for the given agent and job ID. Returns a buffered channel. The caller
// must call UnsubscribeDeployProgress when done.
func (s *Server) SubscribeDeployProgress(agentID, jobID string, bufSize int) chan DeployProgressPayload {
return s.SubscribeDeployProgressForOrganization(defaultOrganizationID, agentID, jobID, bufSize)
}
func (s *Server) SubscribeDeployProgressForOrganization(organizationID, agentID, jobID string, bufSize int) chan DeployProgressPayload {
if bufSize <= 0 {
bufSize = 64
}
ch := make(chan DeployProgressPayload, bufSize)
s.mu.Lock()
s.deploySubs[deploySubKey(agentID, jobID)] = ch
s.deploySubs[deploySubKey(agentSessionKey(organizationID, agentID), jobID)] = ch
s.mu.Unlock()
return ch
}
@@ -1877,7 +2207,11 @@ func (s *Server) SubscribeDeployProgress(agentID, jobID string, bufSize int) cha
// UnsubscribeDeployProgress removes and closes the progress subscriber for an agent's job.
// Safe to call multiple times — a no-op if already unsubscribed (e.g. by readLoop cleanup).
func (s *Server) UnsubscribeDeployProgress(agentID, jobID string) {
key := deploySubKey(agentID, jobID)
s.UnsubscribeDeployProgressForOrganization(defaultOrganizationID, agentID, jobID)
}
func (s *Server) UnsubscribeDeployProgressForOrganization(organizationID, agentID, jobID string) {
key := deploySubKey(agentSessionKey(organizationID, agentID), jobID)
s.mu.Lock()
ch, exists := s.deploySubs[key]
delete(s.deploySubs, key)
@@ -1915,10 +2249,7 @@ func (s *Server) sendDeployCommand(ctx context.Context, agentID string, msgType
return fmt.Errorf("agent id is required")
}
s.mu.RLock()
ac, ok := s.agents[agentID]
s.mu.RUnlock()
ac, ok := s.connectionForContext(ctx, agentID)
if !ok {
return fmt.Errorf("agent %s not connected", agentID)
}
internal/agentexec/server_coverage_test.go
+3 -5
View File
@@ -229,15 +229,13 @@ func TestHandleWebSocket_RegistrationAckSendFailure(t *testing.T) {
Token: "any",
}))
waitFor(t, 2*time.Second, func() bool { return s.IsAgentConnected("a1") })
_ = conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
if _, _, err := conn.ReadMessage(); err == nil {
t.Fatalf("expected no registration ack when send fails")
}
conn.Close()
waitFor(t, 2*time.Second, func() bool { return !s.IsAgentConnected("a1") })
if s.IsAgentConnected("a1") {
t.Fatal("agent must not remain command-connected when registration acknowledgement fails")
}
}
func TestHandleWebSocket_PongHandler(t *testing.T) {
internal/agentexec/server_websocket_test.go
+162
View File
@@ -846,6 +846,22 @@ func TestHandleWebSocket_ReconnectSameAgentIDClosesOldConnection(t *testing.T) {
}))
_ = wsReadRegisteredPayload(t, c1)
progressCh := s.SubscribeDeployProgress("a1", "job-reconnect", 1)
defer s.UnsubscribeDeployProgress("a1", "job-reconnect")
commandDone := make(chan error, 1)
go func() {
_, err := s.ExecuteCommand(context.Background(), "a1", ExecuteCommandPayload{
RequestID: "command-before-reconnect",
Command: "true",
Timeout: 10,
Trusted: true,
})
commandDone <- err
}()
if command := wsReadRawMessage(t, c1); command.Type != MsgTypeExecuteCmd {
t.Fatalf("old session received %q, want %q", command.Type, MsgTypeExecuteCmd)
}
c2 := dial()
defer c2.Close()
wsWriteMessage(t, c2, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{
@@ -862,4 +878,150 @@ func TestHandleWebSocket_ReconnectSameAgentIDClosesOldConnection(t *testing.T) {
if err == nil {
t.Fatalf("expected old connection to be closed")
}
select {
case commandErr := <-commandDone:
if commandErr == nil || !strings.Contains(commandErr.Error(), "disconnected") {
t.Fatalf("in-flight command reconnect result = %v, want disconnected", commandErr)
}
case <-time.After(time.Second):
t.Fatal("in-flight command did not stop when its session was replaced")
}
progress := DeployProgressPayload{
RequestID: "deploy-after-reconnect",
JobID: "job-reconnect",
Phase: DeployPhasePreflightSSH,
Status: DeployStepOK,
}
wsWriteMessage(t, c2, mustNewMessage(t, MsgTypeDeployProgress, progress.RequestID, progress))
select {
case received, ok := <-progressCh:
if !ok {
t.Fatal("replacement cleanup closed the active deploy subscription")
}
if received.RequestID != progress.RequestID {
t.Fatalf("deploy progress request id = %q, want %q", received.RequestID, progress.RequestID)
}
case <-time.After(time.Second):
t.Fatal("replacement session did not retain deploy progress subscription")
}
}
func TestCommandSessionsAreTenantScopedAndDuplicateIdentityFailsClosed(t *testing.T) {
admissions := map[string]AgentAdmission{
"token-a": {OrganizationID: "org-a", TokenID: "token-a", AgentID: "shared", Hostname: "host-a"},
"token-b": {OrganizationID: "org-b", TokenID: "token-b", AgentID: "shared", Hostname: "host-b"},
"token-c": {OrganizationID: "org-a", TokenID: "token-c", AgentID: "shared", Hostname: "other-host"},
"token-d": {OrganizationID: "org-a", TokenID: "token-d", AgentID: "other-id", Hostname: "host-a"},
}
s := NewServerWithAdmissionValidator(func(token, _, _ string) (AgentAdmission, bool) {
admission, ok := admissions[token]
return admission, ok
}, func(AgentAdmission) bool { return true })
ts := newWSServer(t, s)
defer ts.Close()
register := func(token, agentID, hostname string) (*websocket.Conn, RegisteredPayload) {
t.Helper()
conn, _, err := dialAgentExecWebSocket(t, ts.URL)
if err != nil {
t.Fatalf("Dial: %v", err)
}
wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{
AgentID: agentID, Hostname: hostname, Token: token,
}))
return conn, wsReadRegisteredPayload(t, conn)
}
orgA, ack := register("token-a", "shared", "host-a")
defer orgA.Close()
if !ack.Success {
t.Fatalf("org-a registration failed: %s", ack.Message)
}
orgB, ack := register("token-b", "shared", "host-b")
defer orgB.Close()
if !ack.Success {
t.Fatalf("org-b registration failed: %s", ack.Message)
}
if !s.IsAgentConnectedForOrganization("org-a", "shared") ||
!s.IsAgentConnectedForOrganization("org-b", "shared") {
t.Fatal("same agent id must remain independently connected in both organizations")
}
if s.IsAgentConnected("shared") {
t.Fatal("tenant-scoped sessions must not leak into the default organization")
}
orgAView := s.ForOrganization("org-a")
orgAAgents := orgAView.GetConnectedAgents()
if len(orgAAgents) != 1 || orgAAgents[0].Hostname != "host-a" {
t.Fatalf("org-a server view leaked another tenant: %#v", orgAAgents)
}
orgBAgents := s.ForOrganization("org-b").GetConnectedAgents()
if len(orgBAgents) != 1 || orgBAgents[0].Hostname != "host-b" {
t.Fatalf("org-b server view leaked another tenant: %#v", orgBAgents)
}
duplicate, ack := register("token-c", "shared", "other-host")
defer duplicate.Close()
if ack.Success {
t.Fatal("same-tenant duplicate identity from another hostname was admitted")
}
if !s.IsAgentConnectedForOrganization("org-a", "shared") {
t.Fatal("rejected duplicate identity evicted the original session")
}
duplicateHost, ack := register("token-d", "other-id", "host-a")
defer duplicateHost.Close()
if ack.Success {
t.Fatal("same-tenant hostname was admitted under a second identity")
}
if !s.IsAgentConnectedForOrganization("org-a", "shared") {
t.Fatal("rejected duplicate hostname evicted the original session")
}
}
func TestRevokedAdmissionInvalidatesStaleSocketBeforeDispatch(t *testing.T) {
valid := true
admission := AgentAdmission{
OrganizationID: "org-a",
TokenID: "token-a",
AgentID: "agent-a",
Hostname: "host-a",
}
s := NewServerWithAdmissionValidator(func(token, _, _ string) (AgentAdmission, bool) {
return admission, token == admission.TokenID
}, func(candidate AgentAdmission) bool {
return valid && candidate == admission
})
ts := newWSServer(t, s)
defer ts.Close()
conn, _, err := dialAgentExecWebSocket(t, ts.URL)
if err != nil {
t.Fatalf("Dial: %v", err)
}
defer conn.Close()
wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{
AgentID: admission.AgentID, Hostname: admission.Hostname, Token: admission.TokenID,
}))
if ack := wsReadRegisteredPayload(t, conn); !ack.Success {
t.Fatalf("registration failed: %s", ack.Message)
}
if !s.IsAgentConnectedForOrganization("org-a", "agent-a") {
t.Fatal("expected admitted session")
}
valid = false
ctx := WithOrganizationID(context.Background(), "org-a")
if _, err := s.ExecuteCommand(ctx, "agent-a", ExecuteCommandPayload{
RequestID: "after-revocation",
Command: "true",
TargetType: "agent",
Trusted: true,
}); err == nil || !strings.Contains(err.Error(), "not connected") {
t.Fatalf("revoked stale socket remained dispatchable: %v", err)
}
if s.IsAgentConnectedForOrganization("org-a", "agent-a") {
t.Fatal("revoked session remained visible as connected")
}
}
internal/agentexec/types.go
+2
View File
@@ -400,6 +400,8 @@ const (
// ConnectedAgent represents an agent connected via WebSocket
type ConnectedAgent struct {
OrganizationID string
TokenID string
AgentID string
Hostname string
Version string
internal/api/action_executor.go
+66
View File
@@ -16,6 +16,72 @@ type actionAgentCommander interface {
IsAgentConnected(agentID string) bool
}
type scopedActionAgentCommander interface {
IsAgentConnectedForOrganization(organizationID, agentID string) bool
GetAgentForHostForOrganization(organizationID, hostname string) (string, bool)
GetAgentForTokenForOrganization(organizationID, tokenID string) (string, bool)
}
type scopedAgentOperationReceiptCapability interface {
AgentOperationReceiptVersionForOrganization(organizationID, agentID string) int
}
type tenantAgentServer interface {
GetConnectedAgents() []agentexec.ConnectedAgent
ExecuteCommand(ctx context.Context, agentID string, cmd agentexec.ExecuteCommandPayload) (*agentexec.CommandResultPayload, error)
}
func tenantAgentServerForOrganization(server *agentexec.Server, organizationID string) tenantAgentServer {
if server == nil {
return nil
}
return server.ForOrganization(organizationID)
}
func agentCommandContext(ctx context.Context) context.Context {
return agentexec.WithOrganizationID(ctx, GetOrgID(ctx))
}
func isAgentCommandConnected(ctx context.Context, agents any, agentID string) bool {
if scoped, ok := agents.(scopedActionAgentCommander); ok {
return scoped.IsAgentConnectedForOrganization(GetOrgID(ctx), agentID)
}
if legacy, ok := agents.(interface{ IsAgentConnected(string) bool }); ok {
return legacy.IsAgentConnected(agentID)
}
return false
}
func commandAgentForHost(ctx context.Context, agents actionAgentCommander, hostname string) (string, bool) {
if scoped, ok := agents.(scopedActionAgentCommander); ok {
return scoped.GetAgentForHostForOrganization(GetOrgID(ctx), hostname)
}
if agents == nil {
return "", false
}
return agents.GetAgentForHost(hostname)
}
func commandAgentForToken(ctx context.Context, agents actionAgentCommander, tokenID string) (string, bool) {
if strings.TrimSpace(tokenID) == "" {
return "", false
}
if scoped, ok := agents.(scopedActionAgentCommander); ok {
return scoped.GetAgentForTokenForOrganization(GetOrgID(ctx), tokenID)
}
return "", false
}
func liveAgentOperationReceiptVersion(ctx context.Context, agents any, agentID string) int {
if scoped, ok := agents.(scopedAgentOperationReceiptCapability); ok {
return scoped.AgentOperationReceiptVersionForOrganization(GetOrgID(ctx), agentID)
}
if capability, ok := agents.(agentOperationReceiptCapability); ok {
return capability.AgentOperationReceiptVersion(agentID)
}
return 0
}
type actionHandlerProvider interface {
ActionHandlerNames() []string
}
internal/api/agent_exec_token_binding.go
+183 -16
View File
@@ -4,6 +4,7 @@ import (
"strings"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rs/zerolog/log"
)
@@ -11,11 +12,42 @@ import (
const (
agentInstallIssuedViaConfig = "config_agent_install_command"
agentInstallIssuedViaHosted = "hosted_agent_install_command"
agentExecBindingVersionKey = "agent_exec_binding_version"
agentExecBindingVersion = "2"
)
type agentExecMetadataValue struct {
value string
present bool
}
func snapshotAgentExecMetadata(metadata map[string]string, keys ...string) map[string]agentExecMetadataValue {
snapshot := make(map[string]agentExecMetadataValue, len(keys))
for _, key := range keys {
value, present := metadata[key]
snapshot[key] = agentExecMetadataValue{value: value, present: present}
}
return snapshot
}
func restoreAgentExecMetadata(metadata map[string]string, snapshot map[string]agentExecMetadataValue) {
for key, previous := range snapshot {
if previous.present {
metadata[key] = previous.value
} else {
delete(metadata, key)
}
}
}
func (r *Router) validateAgentExecToken(token string, agentID string, hostname string) bool {
_, ok := r.admitAgentExecToken(token, agentID, hostname)
return ok
}
func (r *Router) admitAgentExecToken(token string, agentID string, hostname string) (agentexec.AgentAdmission, bool) {
if r == nil || r.config == nil {
return false
return agentexec.AgentAdmission{}, false
}
requestedID := strings.TrimSpace(agentID)
@@ -33,7 +65,7 @@ func (r *Router) validateAgentExecToken(token string, agentID string, hostname s
Str("agent_id", requestedID).
Str("hostname", requestedHost).
Msg("Agent exec token not recognized by this server — re-run the agent installer to re-enroll this agent")
return false
return agentexec.AgentAdmission{}, false
}
tokenID := record.ID
@@ -42,7 +74,20 @@ func (r *Router) validateAgentExecToken(token string, agentID string, hostname s
log.Warn().
Str("token_id", tokenID).
Msg("Agent exec token missing required scope: agent:exec")
return false
return agentexec.AgentAdmission{}, false
}
orgs := record.GetBoundOrgs()
if len(orgs) > 1 {
config.Mu.Unlock()
log.Warn().
Str("token_id", tokenID).
Strs("organization_ids", orgs).
Msg("Agent exec token rejected because command sessions require one organization binding")
return agentexec.AgentAdmission{}, false
}
organizationID := "default"
if len(orgs) == 1 && strings.TrimSpace(orgs[0]) != "" {
organizationID = strings.TrimSpace(orgs[0])
}
boundID := strings.TrimSpace(record.Metadata["bound_agent_id"])
@@ -53,22 +98,30 @@ func (r *Router) validateAgentExecToken(token string, agentID string, hostname s
if record.Metadata == nil {
record.Metadata = make(map[string]string)
}
previousMetadata := snapshotAgentExecMetadata(
record.Metadata,
"bound_agent_id",
"bound_hostname",
"bound_at",
agentExecBindingVersionKey,
)
record.Metadata["bound_agent_id"] = requestedID
record.Metadata["bound_hostname"] = requestedHost
record.Metadata["bound_at"] = time.Now().UTC().Format(time.RFC3339)
tokens := make([]config.APITokenRecord, len(r.config.APITokens))
copy(tokens, r.config.APITokens)
config.Mu.Unlock()
record.Metadata[agentExecBindingVersionKey] = agentExecBindingVersion
if r.persistence != nil {
if err := r.persistence.SaveAPITokens(tokens); err != nil {
log.Warn().
if err := r.persistence.SaveAPITokens(r.config.APITokens); err != nil {
restoreAgentExecMetadata(record.Metadata, previousMetadata)
config.Mu.Unlock()
log.Error().
Err(err).
Str("token_id", tokenID).
Str("hostname", requestedHost).
Msg("Failed to persist first-use Proxmox agent exec token binding")
Msg("Failed to persist first-use agent exec token binding; command registration denied")
return agentexec.AgentAdmission{}, false
}
}
config.Mu.Unlock()
log.Info().
Str("token_id", tokenID).
@@ -76,7 +129,12 @@ func (r *Router) validateAgentExecToken(token string, agentID string, hostname s
Str("issued_via", issuedVia).
Str("install_type", installType).
Msg("Bound agent install token to first command agent registration")
return true
return agentexec.AgentAdmission{
OrganizationID: organizationID,
TokenID: tokenID,
AgentID: requestedID,
Hostname: requestedHost,
}, true
}
if boundID == "" && boundHost == "" {
@@ -84,17 +142,93 @@ func (r *Router) validateAgentExecToken(token string, agentID string, hostname s
log.Warn().
Str("token_id", tokenID).
Msg("Agent exec token missing binding metadata")
return false
return agentexec.AgentAdmission{}, false
}
if boundHost != "" && strings.EqualFold(boundHost, requestedHost) {
// Pre-v6.1.1 deploy tokens could carry a server-synthesized agent ID even
// though the runtime derives its ID from machine-id. Migrate that
// hostname-bound legacy record exactly once, then enforce both fields.
if strings.TrimSpace(record.Metadata[agentExecBindingVersionKey]) != agentExecBindingVersion &&
boundHost != "" && strings.EqualFold(boundHost, requestedHost) {
previousID := boundID
previousMetadata := snapshotAgentExecMetadata(
record.Metadata,
"bound_agent_id",
"bound_at",
agentExecBindingVersionKey,
)
record.Metadata["bound_agent_id"] = requestedID
record.Metadata["bound_at"] = time.Now().UTC().Format(time.RFC3339)
record.Metadata[agentExecBindingVersionKey] = agentExecBindingVersion
if r.persistence != nil {
if err := r.persistence.SaveAPITokens(r.config.APITokens); err != nil {
restoreAgentExecMetadata(record.Metadata, previousMetadata)
config.Mu.Unlock()
log.Error().
Err(err).
Str("token_id", tokenID).
Msg("Failed to persist legacy agent exec identity migration; command registration denied")
return agentexec.AgentAdmission{}, false
}
}
config.Mu.Unlock()
return true
log.Info().
Str("token_id", tokenID).
Str("previous_agent_id", previousID).
Str("agent_id", requestedID).
Str("hostname", requestedHost).
Msg("Migrated legacy hostname-bound agent exec token to immutable runtime identity")
return agentexec.AgentAdmission{
OrganizationID: organizationID,
TokenID: tokenID,
AgentID: requestedID,
Hostname: requestedHost,
}, true
}
if boundID != "" && boundID == requestedID {
idMatches := boundID == "" || boundID == requestedID
hostMatches := boundHost == "" || strings.EqualFold(boundHost, requestedHost)
if idMatches && hostMatches {
previousMetadata := snapshotAgentExecMetadata(
record.Metadata,
"bound_agent_id",
"bound_hostname",
"bound_at",
agentExecBindingVersionKey,
)
metadataChanged := false
if boundID == "" && boundHost != "" {
record.Metadata["bound_agent_id"] = requestedID
boundID = requestedID
metadataChanged = true
}
if boundHost == "" && boundID != "" {
record.Metadata["bound_hostname"] = requestedHost
boundHost = requestedHost
metadataChanged = true
}
if metadataChanged {
record.Metadata["bound_at"] = time.Now().UTC().Format(time.RFC3339)
record.Metadata[agentExecBindingVersionKey] = agentExecBindingVersion
}
if metadataChanged && r.persistence != nil {
if err := r.persistence.SaveAPITokens(r.config.APITokens); err != nil {
restoreAgentExecMetadata(record.Metadata, previousMetadata)
config.Mu.Unlock()
log.Error().
Err(err).
Str("token_id", tokenID).
Msg("Failed to persist migrated agent exec token binding; command registration denied")
return agentexec.AgentAdmission{}, false
}
}
config.Mu.Unlock()
return true
return agentexec.AgentAdmission{
OrganizationID: organizationID,
TokenID: tokenID,
AgentID: requestedID,
Hostname: requestedHost,
}, true
}
config.Mu.Unlock()
@@ -105,6 +239,39 @@ func (r *Router) validateAgentExecToken(token string, agentID string, hostname s
Str("requested_id", requestedID).
Str("requested_hostname", requestedHost).
Msg("Agent token mismatch: token is not bound to the registering agent")
return agentexec.AgentAdmission{}, false
}
func (r *Router) validateAgentExecSession(admission agentexec.AgentAdmission) bool {
if r == nil || r.config == nil {
return false
}
tokenID := strings.TrimSpace(admission.TokenID)
requestedID := strings.TrimSpace(admission.AgentID)
requestedHost := strings.TrimSpace(admission.Hostname)
if tokenID == "" || requestedID == "" || requestedHost == "" {
return false
}
config.Mu.Lock()
defer config.Mu.Unlock()
for index := range r.config.APITokens {
record := &r.config.APITokens[index]
if record.ID != tokenID || record.IsExpired() || !record.HasScope(config.ScopeAgentExec) {
continue
}
orgs := record.GetBoundOrgs()
organizationID := "default"
if len(orgs) > 1 {
return false
}
if len(orgs) == 1 && strings.TrimSpace(orgs[0]) != "" {
organizationID = strings.TrimSpace(orgs[0])
}
return organizationID == strings.TrimSpace(admission.OrganizationID) &&
strings.TrimSpace(record.Metadata["bound_agent_id"]) == requestedID &&
strings.EqualFold(strings.TrimSpace(record.Metadata["bound_hostname"]), requestedHost)
}
return false
}
internal/api/ai_handler.go
+2 -2
View File
@@ -636,7 +636,7 @@ func (h *AIHandler) initTenantService(ctx context.Context, orgID string) AIServi
chatCfg := chat.Config{
AIConfig: aiCfg,
DataDir: dataDir,
AgentServer: h.agentServer,
AgentServer: tenantAgentServerForOrganization(h.agentServer, orgID),
ReadState: h.readStateForOrg(orgID),
OrgID: orgID,
ControlLevelResolver: func(next *config.AIConfig) string {
@@ -886,7 +886,7 @@ func (h *AIHandler) startWithConfig(ctx context.Context, monitor *monitoring.Mon
AIConfig: aiCfg,
DataDir: dataDir,
StateProvider: monitor,
AgentServer: h.agentServer,
AgentServer: tenantAgentServerForOrganization(h.agentServer, orgID),
ReadState: h.readStateForOrg(orgID),
OrgID: orgID,
ControlLevelResolver: func(next *config.AIConfig) string {
internal/api/ai_handlers.go
+4 -4
View File
@@ -287,7 +287,7 @@ func (h *AISettingsHandler) providerSnapshot() aiSettingsProviderSnapshot {
}
func (h *AISettingsHandler) newFailClosedTenantService(orgID string) *ai.Service {
svc := ai.NewService(nil, h.agentServer)
svc := ai.NewService(nil, tenantAgentServerForOrganization(h.agentServer, orgID))
svc.SetOrgID(orgID)
h.stateMu.RLock()
patrolAutopilotPolicy := h.patrolAutopilotPolicy
@@ -338,7 +338,7 @@ func NewAISettingsHandler(mtp *config.MultiTenantPersistence, mtm *monitoring.Mu
incidentRecorders: make(map[string]*metrics.IncidentRecorder),
}
defaultAIService = ai.NewService(defaultPersistence, agentServer)
defaultAIService = ai.NewService(defaultPersistence, tenantAgentServerForOrganization(agentServer, "default"))
defaultAIService.SetOrgID("default")
defaultAIService.SetAlertAnalyzerFactory(getCreateAlertAnalyzer())
if defaultPersistence != nil {
@@ -435,7 +435,7 @@ func (h *AISettingsHandler) GetAIService(ctx context.Context) *ai.Service {
return h.newFailClosedTenantService(orgID)
}
svc = ai.NewService(persistence, h.agentServer)
svc = ai.NewService(persistence, tenantAgentServerForOrganization(h.agentServer, orgID))
svc.SetOrgID(orgID)
h.stateMu.RLock()
patrolAutopilotPolicy := h.patrolAutopilotPolicy
@@ -4747,7 +4747,7 @@ func (h *AISettingsHandler) HandleGetConnectedAgents(w http.ResponseWriter, r *h
var agents []agentInfo
if h.agentServer != nil {
for _, a := range h.agentServer.GetConnectedAgents() {
for _, a := range h.agentServer.GetConnectedAgentsForOrganization(GetOrgID(r.Context())) {
agents = append(agents, agentInfo{
AgentID: a.AgentID,
Hostname: a.Hostname,
internal/api/ai_handlers_more_test.go
+67
View File
@@ -5,10 +5,13 @@ import (
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
"unsafe"
"github.com/gorilla/websocket"
"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
"github.com/rcourtman/pulse-go-rewrite/internal/ai"
"github.com/rcourtman/pulse-go-rewrite/internal/ai/memory"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
@@ -28,6 +31,70 @@ type recordingMetadataProvider struct {
guestURLs map[string]string
}
func TestTenantAgentServerForOrganizationFailsClosedAcrossTenants(t *testing.T) {
admissions := map[string]agentexec.AgentAdmission{
"token-a": {OrganizationID: "tenant-a", TokenID: "token-a", AgentID: "agent-a", Hostname: "host-a"},
"token-b": {OrganizationID: "tenant-b", TokenID: "token-b", AgentID: "agent-b", Hostname: "host-b"},
}
server := agentexec.NewServerWithAdmissionValidator(
func(token, _, _ string) (agentexec.AgentAdmission, bool) {
admission, ok := admissions[token]
return admission, ok
},
func(agentexec.AgentAdmission) bool { return true },
)
websocketServer := httptest.NewServer(http.HandlerFunc(server.HandleWebSocket))
defer websocketServer.Close()
register := func(admission agentexec.AgentAdmission) *websocket.Conn {
t.Helper()
url := "ws" + strings.TrimPrefix(websocketServer.URL, "http")
conn, _, err := websocket.DefaultDialer.Dial(url, http.Header{
"Origin": []string{websocketServer.URL},
})
if err != nil {
t.Fatalf("dial agent websocket: %v", err)
}
message, err := agentexec.NewMessage(agentexec.MsgTypeAgentRegister, "", agentexec.AgentRegisterPayload{
AgentID: admission.AgentID,
Hostname: admission.Hostname,
Token: admission.TokenID,
})
if err != nil {
t.Fatalf("create registration message: %v", err)
}
if err := conn.WriteJSON(message); err != nil {
t.Fatalf("write registration: %v", err)
}
var response agentexec.Message
if err := conn.ReadJSON(&response); err != nil {
t.Fatalf("read registration response: %v", err)
}
var registered agentexec.RegisteredPayload
if err := response.DecodePayload(&registered); err != nil || !registered.Success {
t.Fatalf("registration failed: payload=%+v err=%v", registered, err)
}
return conn
}
connA := register(admissions["token-a"])
defer connA.Close()
connB := register(admissions["token-b"])
defer connB.Close()
tenantA := tenantAgentServerForOrganization(server, "tenant-a")
agents := tenantA.GetConnectedAgents()
if len(agents) != 1 || agents[0].AgentID != "agent-a" {
t.Fatalf("tenant-a command view leaked another tenant: %#v", agents)
}
if _, err := tenantA.ExecuteCommand(context.Background(), "agent-b", agentexec.ExecuteCommandPayload{
Command: "true",
Trusted: true,
}); err == nil || !strings.Contains(err.Error(), "not connected") {
t.Fatalf("cross-tenant command dispatch did not fail closed: %v", err)
}
}
func (p *recordingMetadataProvider) SetGuestURL(id, url string) error {
p.guestURLs[id] = url
return nil
internal/api/connections_aggregator.go
+15
View File
@@ -38,6 +38,7 @@ const (
fleetStateCurrent = "current"
fleetStateDegraded = "degraded"
fleetStateDisabled = "disabled"
fleetStateDisconnected = "disconnected"
fleetStateEnabled = "enabled"
fleetStateEnrolled = "enrolled"
fleetStateHealthy = "healthy"
@@ -543,6 +544,8 @@ func buildAgentConnection(host models.Host, expectedAgentVersion string, now tim
AgentUpdate: connectionAgentUpdateStatus(host.AgentUpdate),
AgentModules: connectionAgentModuleStatuses(host.AgentModules),
Capabilities: ConnectionCapabilities{SupportsPause: false, SupportsScope: false, SupportsTest: false},
agentID: strings.TrimSpace(host.ID),
agentTokenID: strings.TrimSpace(host.TokenID),
}, now)
conn.Fleet.ConfigDrift = connectionFleetAgentConfigDrift(conn, desiredConfig, host.AppliedConfig)
conn.Fleet.CredentialHealth = connectionFleetAgentCredentialHealth(conn, host, now)
@@ -725,6 +728,9 @@ func connectionFleetRemoteControl(conn Connection) string {
return fleetStateUnknown
}
if conn.AgentIdentity != nil && conn.AgentIdentity.CommandsEnabled {
if conn.commandChannelConnected != nil && !*conn.commandChannelConnected {
return fleetStateDisconnected
}
return fleetStateEnabled
}
return fleetStateDisabled
@@ -1008,6 +1014,15 @@ func connectionFleetAgentCommandPolicy(conn Connection, host models.Host, desire
}
applied := connectionFleetCommandPolicyState(host.CommandsEnabled)
if applied == fleetStateEnabled && conn.commandChannelConnected != nil && !*conn.commandChannelConnected {
return &ConnectionFleetCommandPolicy{
Status: fleetStateBlocked,
Desired: desired,
Applied: applied,
Enforcement: fleetStateBlocked,
Reason: "agent reports command execution enabled, but no admitted command channel is connected",
}
}
policy := &ConnectionFleetCommandPolicy{
Status: applied,
Desired: desired,
internal/api/connections_handlers.go
+49 -6
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"net/http"
"strings"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
@@ -13,11 +14,23 @@ import (
// any persistence of its own — it composes per-type stores and the
// monitoring scheduler's in-memory health data into a single list.
type ConnectionsHandlers struct {
getConfig func(ctx context.Context) *config.Config
getPersistence func(ctx context.Context) *config.ConfigPersistence
getMonitor func(ctx context.Context) *monitoring.Monitor
getTrueNASPoller func(ctx context.Context) *monitoring.TrueNASPoller
getVMwarePoller func(ctx context.Context) *monitoring.VMwarePoller
getConfig func(ctx context.Context) *config.Config
getPersistence func(ctx context.Context) *config.ConfigPersistence
getMonitor func(ctx context.Context) *monitoring.Monitor
getTrueNASPoller func(ctx context.Context) *monitoring.TrueNASPoller
getVMwarePoller func(ctx context.Context) *monitoring.VMwarePoller
agentCommandSessionConnected func(organizationID, tokenID, agentID, hostname string) bool
}
// SetAgentCommandSessionProvider wires the live command-channel registry into
// the ledger. Telemetry liveness alone is not command readiness.
func (h *ConnectionsHandlers) SetAgentCommandSessionProvider(
connected func(organizationID, tokenID, agentID, hostname string) bool,
) {
if h == nil {
return
}
h.agentCommandSessionConnected = connected
}
// NewConnectionsHandlers wires the aggregator behind the request-scoped
@@ -62,9 +75,39 @@ func (h *ConnectionsHandlers) HandleList(w http.ResponseWriter, r *http.Request)
persistence := h.getPersistence(ctx)
monitor := h.getMonitor(ctx)
inputs := buildAggregatorInputsWithRuntimeSources(ctx, cfg, persistence, monitor, h.runtimeSources(ctx, resolveTenantOrgID(r)))
organizationID := resolveTenantOrgID(r)
inputs := buildAggregatorInputsWithRuntimeSources(ctx, cfg, persistence, monitor, h.runtimeSources(ctx, organizationID))
connections := buildConnections(inputs)
if h.agentCommandSessionConnected != nil {
for index := range connections {
conn := &connections[index]
if conn.Type != ConnectionTypeAgent || conn.AgentIdentity == nil {
continue
}
connected := h.agentCommandSessionConnected(
organizationID,
conn.agentTokenID,
conn.agentID,
conn.AgentIdentity.Hostname,
)
conn.commandChannelConnected = &connected
conn.Fleet.RemoteControl = connectionFleetRemoteControl(*conn)
if conn.AgentIdentity.CommandsEnabled && !connected {
desired := fleetStateUnknown
if conn.Fleet.CommandPolicy != nil && strings.TrimSpace(conn.Fleet.CommandPolicy.Desired) != "" {
desired = conn.Fleet.CommandPolicy.Desired
}
conn.Fleet.CommandPolicy = &ConnectionFleetCommandPolicy{
Status: fleetStateBlocked,
Desired: desired,
Applied: fleetStateEnabled,
Enforcement: fleetStateBlocked,
Reason: "agent reports command execution enabled, but no admitted command channel is connected",
}
}
}
}
writeJSON(w, http.StatusOK, ConnectionsListResponse{
Connections: connections,
Systems: buildConnectionSystems(connections, monitor),
internal/api/connections_handlers_continuity_test.go
+73
View File
@@ -90,6 +90,79 @@ func TestConnectionsHandleListIncludesContinuityBackedHostAgents(t *testing.T) {
}
}
func TestConnectionsLedgerSeparatesTelemetryHealthFromCommandAdmission(t *testing.T) {
dir := t.TempDir()
cfg := &config.Config{DataPath: dir}
now := time.Now().UTC()
monitor, err := monitoring.New(cfg)
if err != nil {
t.Fatalf("monitoring.New: %v", err)
}
t.Cleanup(monitor.Stop)
for _, report := range []agentshost.Report{
{
Agent: agentshost.AgentInfo{ID: "agent-enabled", Version: "6.1.1", IntervalSeconds: 30, CommandsEnabled: true},
Host: agentshost.HostInfo{ID: "machine-enabled", MachineID: "machine-enabled", Hostname: "docker-host", Platform: "linux"},
Timestamp: now,
},
{
Agent: agentshost.AgentInfo{ID: "agent-disabled", Version: "6.1.1", IntervalSeconds: 30, CommandsEnabled: false},
Host: agentshost.HostInfo{ID: "machine-disabled", MachineID: "machine-disabled", Hostname: "policy-disabled", Platform: "linux"},
Timestamp: now,
},
} {
tokenID := "token-enabled"
if report.Agent.ID == "agent-disabled" {
tokenID = "token-disabled"
}
if _, err := monitor.ApplyHostReport(report, &config.APITokenRecord{ID: tokenID, Name: tokenID}); err != nil {
t.Fatalf("ApplyHostReport(%s): %v", report.Agent.ID, err)
}
}
handler := NewConnectionsHandlers(
func(context.Context) *config.Config { return cfg },
func(context.Context) *config.ConfigPersistence { return nil },
func(context.Context) *monitoring.Monitor { return monitor },
)
handler.SetAgentCommandSessionProvider(func(_, tokenID, _, _ string) bool {
return tokenID == "token-connected"
})
req := httptest.NewRequest(http.MethodGet, "/api/connections", nil)
rec := httptest.NewRecorder()
handler.HandleList(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String())
}
var response ConnectionsListResponse
if err := json.NewDecoder(rec.Body).Decode(&response); err != nil {
t.Fatalf("decode: %v", err)
}
byID := make(map[string]Connection, len(response.Connections))
for _, connection := range response.Connections {
byID[connection.ID] = connection
}
enabled := byID["agent:machine-enabled"]
if enabled.State != ConnectionStateActive || enabled.Fleet.AdapterHealth != fleetStateHealthy {
t.Fatalf("telemetry health changed when command channel was absent: %+v", enabled.Fleet)
}
if enabled.Fleet.RemoteControl != fleetStateDisconnected ||
enabled.Fleet.CommandPolicy == nil ||
enabled.Fleet.CommandPolicy.Status != fleetStateBlocked {
t.Fatalf("enabled report without admitted socket must be disconnected/blocked: %+v", enabled.Fleet)
}
disabled := byID["agent:machine-disabled"]
if disabled.Fleet.RemoteControl != fleetStateDisabled ||
disabled.Fleet.CommandPolicy == nil ||
disabled.Fleet.CommandPolicy.Status != fleetStateDisabled {
t.Fatalf("disabled command policy must not be misreported as a transport failure: %+v", disabled.Fleet)
}
}
func TestConnectionsHandleListUsesTrueNASPollerRuntimeSummary(t *testing.T) {
persistence := config.NewConfigPersistence(t.TempDir())
connection := config.TrueNASInstance{
internal/api/connections_types.go
+4
View File
@@ -181,6 +181,10 @@ type Connection struct {
AgentModules []ConnectionAgentModuleStatus `json:"agentModules,omitempty"`
Fleet ConnectionFleetGovernance `json:"fleet"`
Capabilities ConnectionCapabilities `json:"capabilities"`
agentID string
agentTokenID string
commandChannelConnected *bool
}
type ConnectionSystemComponentRole string
internal/api/contract_test.go
+14 -10
View File
@@ -20678,9 +20678,11 @@ func TestContract_DockerLifecycleActionsResolveCommandAgentAndDispatchOneTypedOp
}
sharedSrc := string(sharedSource)
for _, snippet := range []string{
"func (e dockerContainerActionExecutor) connectedDockerCommandAgentID(resource unified.Resource) (string, error)",
"func (e dockerContainerActionExecutor) connectedDockerCommandAgentID(ctx context.Context, resource unified.Resource) (string, error)",
"resource.Docker.TokenID",
"commandAgentForToken(ctx, e.agents, resource.Docker.TokenID)",
"resource.Docker.AgentID",
"e.agents.GetAgentForHost(strings.TrimSpace(resource.Docker.Hostname))",
"commandAgentForHost(ctx, e.agents, strings.TrimSpace(resource.Docker.Hostname))",
"ExecuteDockerContainerLifecycle(context.Context, string, agentexec.DockerContainerLifecyclePayload)",
"ActionDispatchOperationKinds",
"BindActionDispatch",
@@ -20715,9 +20717,11 @@ func TestContract_DockerLifecycleActionsResolveCommandAgentAndDispatchOneTypedOp
if strings.Contains(reconcileSource, "executorForAction(ctx, record.Request)") {
t.Fatal("durable reconciliation must not rediscover its executor from current resource inventory")
}
if strings.Index(src, "if agentID := strings.TrimSpace(resource.Docker.AgentID)") >
strings.Index(src, "e.agents.GetAgentForHost(strings.TrimSpace(resource.Docker.Hostname))") {
t.Fatal("docker lifecycle executor must try the Docker reporting agent id before falling back to hostname resolution")
if strings.Index(src, "commandAgentForToken(ctx, e.agents, resource.Docker.TokenID)") >
strings.Index(src, "if agentID := strings.TrimSpace(resource.Docker.AgentID)") ||
strings.Index(src, "if agentID := strings.TrimSpace(resource.Docker.AgentID)") >
strings.Index(src, "commandAgentForHost(ctx, e.agents, strings.TrimSpace(resource.Docker.Hostname))") {
t.Fatal("docker lifecycle executor must resolve the immutable reporting token before legacy agent-id and hostname fallbacks")
}
}
@@ -20750,9 +20754,9 @@ func TestContract_ProxmoxLifecycleActionsResolveNodeCommandAgentAndVerifyState(t
}
src := string(source)
for _, snippet := range []string{
"func (e proxmoxGuestActionExecutor) connectedProxmoxNodeCommandAgentID(resource unified.Resource) (string, error)",
"func (e proxmoxGuestActionExecutor) connectedProxmoxNodeCommandAgentID(ctx context.Context, resource unified.Resource) (string, error)",
"resource.Proxmox.LinkedAgentID",
"e.agents.GetAgentForHost(strings.TrimSpace(resource.Proxmox.NodeName))",
"commandAgentForHost(ctx, e.agents, strings.TrimSpace(resource.Proxmox.NodeName))",
"Trusted: true",
"func (e proxmoxGuestActionExecutor) verifyProxmoxGuestState(",
"proxmoxGuestStatusCommand(kind, vmid)",
@@ -20764,7 +20768,7 @@ func TestContract_ProxmoxLifecycleActionsResolveNodeCommandAgentAndVerifyState(t
}
}
if strings.Index(src, "if agentID := strings.TrimSpace(resource.Proxmox.LinkedAgentID)") >
strings.Index(src, "e.agents.GetAgentForHost(strings.TrimSpace(resource.Proxmox.NodeName))") {
strings.Index(src, "commandAgentForHost(ctx, e.agents, strings.TrimSpace(resource.Proxmox.NodeName))") {
t.Fatal("proxmox lifecycle executor must try the linked Proxmox node agent before falling back to node hostname resolution")
}
@@ -20935,8 +20939,8 @@ func TestContract_DurableOperationReceiptCapabilityGatesAPTActions(t *testing.T)
files := map[string][]string{
"../../pkg/agents/host/report.go": {"OperationReceiptVersion int", `json:"operationReceiptVersion,omitempty"`},
"../unifiedresources/adapters.go": {"host.OperationReceiptVersion != operationreceipt.ProtocolVersion", "hostPackageUpdateCapabilities(host)", "hostStorageCleanupCapability(host)"},
"host_update_action_executor.go": {"resource.Agent.OperationReceiptVersion != operationreceipt.ProtocolVersion", "AgentOperationReceiptVersion(agentID) != operationreceipt.ProtocolVersion"},
"host_storage_cleanup_action_executor.go": {"resource.Agent.OperationReceiptVersion != operationreceipt.ProtocolVersion", "AgentOperationReceiptVersion(agentID) != operationreceipt.ProtocolVersion"},
"host_update_action_executor.go": {"resource.Agent.OperationReceiptVersion != operationreceipt.ProtocolVersion", "liveAgentOperationReceiptVersion(ctx, e.agents, agentID) != operationreceipt.ProtocolVersion"},
"host_storage_cleanup_action_executor.go": {"resource.Agent.OperationReceiptVersion != operationreceipt.ProtocolVersion", "liveAgentOperationReceiptVersion(ctx, e.agents, agentID) != operationreceipt.ProtocolVersion"},
"../ai/findings_apt_workflows.go": {"aptHostHasCapability(host, \"install_os_updates\", \"host.package_updates\")", "aptHostHasCapability(host, \"clean_package_cache\", \"host.storage_cleanup\")"},
}
for path, fragments := range files {
internal/api/deploy_handlers.go
+21 -21
View File
@@ -113,7 +113,7 @@ func (h *DeployHandlers) HandleCandidates(w http.ResponseWriter, r *http.Request
// Build connected agents set
connectedAgents := make(map[string]bool)
for _, agent := range h.execServer.GetConnectedAgents() {
for _, agent := range h.execServer.GetConnectedAgentsForOrganization(GetOrgID(r.Context())) {
connectedAgents[agent.AgentID] = true
}
@@ -226,7 +226,7 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
}
// Verify source agent is connected.
if !h.execServer.IsAgentConnected(req.SourceAgentID) {
if !h.execServer.IsAgentConnectedForOrganization(resolveTenantOrgID(r), req.SourceAgentID) {
writeErrorResponse(w, http.StatusConflict, "source_agent_offline", "Source agent is not connected", nil)
return
}
@@ -361,11 +361,11 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
})
// Subscribe to progress before sending command to avoid race.
progressCh := h.execServer.SubscribeDeployProgress(req.SourceAgentID, jobID, 64)
progressCh := h.execServer.SubscribeDeployProgressForOrganization(job.OrgID, req.SourceAgentID, jobID, 64)
// Send command to agent.
if err := h.execServer.SendDeployPreflight(ctx, req.SourceAgentID, payload); err != nil {
h.execServer.UnsubscribeDeployProgress(req.SourceAgentID, jobID)
if err := h.execServer.SendDeployPreflight(agentCommandContext(ctx), req.SourceAgentID, payload); err != nil {
h.execServer.UnsubscribeDeployProgressForOrganization(job.OrgID, req.SourceAgentID, jobID)
_ = h.store.UpdateJobStatus(ctx, jobID, deploy.JobFailed)
log.Error().Err(err).Str("job_id", jobID).Msg("Failed to send preflight command")
writeErrorResponse(w, http.StatusInternalServerError, "send_failed",
@@ -374,7 +374,7 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
}
// Start background goroutine to process progress events.
go h.processPreflightProgress(jobID, req.SourceAgentID, progressCh)
go h.processPreflightProgress(job.OrgID, jobID, req.SourceAgentID, progressCh)
resp := createPreflightResponse{
PreflightID: jobID,
@@ -554,8 +554,8 @@ func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Re
// processPreflightProgress reads deploy progress events from the agent and
// persists them as deploy events, also broadcasting to SSE clients.
func (h *DeployHandlers) processPreflightProgress(jobID, agentID string, ch <-chan agentexec.DeployProgressPayload) {
defer h.execServer.UnsubscribeDeployProgress(agentID, jobID)
func (h *DeployHandlers) processPreflightProgress(organizationID, jobID, agentID string, ch <-chan agentexec.DeployProgressPayload) {
defer h.execServer.UnsubscribeDeployProgressForOrganization(organizationID, agentID, jobID)
ctx := context.Background()
@@ -1016,7 +1016,7 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request)
}
// Verify source agent is connected.
if !h.execServer.IsAgentConnected(req.SourceAgentID) {
if !h.execServer.IsAgentConnectedForOrganization(resolveTenantOrgID(r), req.SourceAgentID) {
writeErrorResponse(w, http.StatusConflict, "source_agent_offline", "Source agent is not connected", nil)
return
}
@@ -1224,11 +1224,11 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request)
}
// Subscribe to progress before sending command.
progressCh := h.execServer.SubscribeDeployProgress(req.SourceAgentID, jobID, 64)
progressCh := h.execServer.SubscribeDeployProgressForOrganization(orgID, req.SourceAgentID, jobID, 64)
// Send install command to agent.
if err := h.execServer.SendDeployInstall(ctx, req.SourceAgentID, payload); err != nil {
h.execServer.UnsubscribeDeployProgress(req.SourceAgentID, jobID)
if err := h.execServer.SendDeployInstall(agentCommandContext(ctx), req.SourceAgentID, payload); err != nil {
h.execServer.UnsubscribeDeployProgressForOrganization(orgID, req.SourceAgentID, jobID)
_ = h.store.UpdateJobStatus(ctx, jobID, deploy.JobFailed)
// Mark pending targets as failed so they're eligible for retry.
for _, it := range installTargets {
@@ -1241,7 +1241,7 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request)
}
// Start background goroutine to process progress events.
go h.processInstallProgress(jobID, req.SourceAgentID, job.RetryMax, progressCh)
go h.processInstallProgress(orgID, jobID, req.SourceAgentID, job.RetryMax, progressCh)
resp := createJobResponse{
JobID: jobID,
@@ -1315,7 +1315,7 @@ func (h *DeployHandlers) HandleCancelJob(w http.ResponseWriter, r *http.Request)
RequestID: generateID("req"),
JobID: jobID,
}
if err := h.execServer.SendDeployCancel(ctx, job.SourceAgentID, cancelPayload); err != nil {
if err := h.execServer.SendDeployCancel(agentCommandContext(ctx), job.SourceAgentID, cancelPayload); err != nil {
log.Error().Err(err).Str("job_id", jobID).Msg("Failed to send cancel command")
// Don't fail the request — the agent may have already disconnected.
// processInstallProgress will handle the channel close.
@@ -1392,7 +1392,7 @@ func (h *DeployHandlers) HandleRetryJob(w http.ResponseWriter, r *http.Request)
}
// Verify source agent is connected.
if !h.execServer.IsAgentConnected(job.SourceAgentID) {
if !h.execServer.IsAgentConnectedForOrganization(orgID, job.SourceAgentID) {
writeErrorResponse(w, http.StatusConflict, "source_agent_offline", "Source agent is not connected", nil)
return
}
@@ -1523,9 +1523,9 @@ func (h *DeployHandlers) HandleRetryJob(w http.ResponseWriter, r *http.Request)
}
// Subscribe and send.
progressCh := h.execServer.SubscribeDeployProgress(job.SourceAgentID, jobID, 64)
if err := h.execServer.SendDeployInstall(ctx, job.SourceAgentID, payload); err != nil {
h.execServer.UnsubscribeDeployProgress(job.SourceAgentID, jobID)
progressCh := h.execServer.SubscribeDeployProgressForOrganization(orgID, job.SourceAgentID, jobID, 64)
if err := h.execServer.SendDeployInstall(agentCommandContext(ctx), job.SourceAgentID, payload); err != nil {
h.execServer.UnsubscribeDeployProgressForOrganization(orgID, job.SourceAgentID, jobID)
_ = h.store.UpdateJobStatus(ctx, jobID, deploy.JobFailed)
// Mark retried targets back to failed so they can be retried again.
for _, it := range installTargets {
@@ -1537,7 +1537,7 @@ func (h *DeployHandlers) HandleRetryJob(w http.ResponseWriter, r *http.Request)
return
}
go h.processInstallProgress(jobID, job.SourceAgentID, job.RetryMax, progressCh)
go h.processInstallProgress(orgID, jobID, job.SourceAgentID, job.RetryMax, progressCh)
resp := map[string]any{
"jobId": jobID,
@@ -1553,8 +1553,8 @@ func (h *DeployHandlers) HandleRetryJob(w http.ResponseWriter, r *http.Request)
// processInstallProgress reads install progress events from the agent and
// persists them as deploy events, also broadcasting to SSE clients.
func (h *DeployHandlers) processInstallProgress(jobID, agentID string, retryMax int, ch <-chan agentexec.DeployProgressPayload) {
defer h.execServer.UnsubscribeDeployProgress(agentID, jobID)
func (h *DeployHandlers) processInstallProgress(organizationID, jobID, agentID string, retryMax int, ch <-chan agentexec.DeployProgressPayload) {
defer h.execServer.UnsubscribeDeployProgressForOrganization(organizationID, agentID, jobID)
ctx := context.Background()
internal/api/deploy_handlers_test.go
+1 -1
View File
@@ -1622,7 +1622,7 @@ func TestProcessInstallProgress_AgentDisconnect(t *testing.T) {
close(ch)
// Run in foreground for testing.
h.processInstallProgress("dep_disc", "agent-1", 3, ch)
h.processInstallProgress("default", "dep_disc", "agent-1", 3, ch)
// Job should be failed.
job, _ := h.store.GetJob(ctx, "dep_disc")
internal/api/docker_container_action_executor.go
+22 -15
View File
@@ -68,7 +68,7 @@ func (e dockerContainerActionExecutor) BindActionDispatch(ctx context.Context, r
if err != nil {
return unified.ActionDispatchAttempt{}, err
}
agentID, err := e.connectedDockerCommandAgentID(resource)
agentID, err := e.connectedDockerCommandAgentID(ctx, resource)
if err != nil {
return unified.ActionDispatchAttempt{}, err
}
@@ -90,7 +90,7 @@ func (e dockerContainerActionExecutor) BindActionDispatch(ctx context.Context, r
if err != nil {
return unified.ActionDispatchAttempt{}, err
}
agentID, err := e.connectedDockerCommandAgentID(resource)
agentID, err := e.connectedDockerCommandAgentID(ctx, resource)
if err != nil {
return unified.ActionDispatchAttempt{}, err
}
@@ -149,7 +149,7 @@ func (e dockerContainerActionExecutor) ExecuteAction(ctx context.Context, record
if dockerContainerRef(resource) == "" {
return nil, fmt.Errorf("docker container resource %q has no executable container id", record.Request.ResourceID)
}
agentID, err := e.connectedDockerCommandAgentID(resource)
agentID, err := e.connectedDockerCommandAgentID(ctx, resource)
if err != nil {
return nil, err
}
@@ -169,7 +169,7 @@ func (e dockerContainerActionExecutor) ExecuteAction(ctx context.Context, record
if agentexec.DockerContainerLifecycleOperationIdentity(agentID, req) != (operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID}) {
return nil, fmt.Errorf("docker container lifecycle dispatch binding drift")
}
result, err := typedAgents.ExecuteDockerContainerLifecycle(ctx, agentID, req)
result, err := typedAgents.ExecuteDockerContainerLifecycle(agentCommandContext(ctx), agentID, req)
if err != nil {
return nil, err
}
@@ -202,7 +202,7 @@ func (e dockerContainerActionExecutor) executeDockerContainerUpdate(ctx context.
if dockerContainerRef(resource) == "" {
return nil, fmt.Errorf("docker container resource %q has no executable container id", record.Request.ResourceID)
}
agentID, err := e.connectedDockerCommandAgentID(resource)
agentID, err := e.connectedDockerCommandAgentID(ctx, resource)
if err != nil {
return nil, err
}
@@ -217,7 +217,7 @@ func (e dockerContainerActionExecutor) executeDockerContainerUpdate(ctx context.
if agentexec.DockerContainerUpdateOperationIdentity(agentID, req) != (operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID}) {
return nil, fmt.Errorf("docker container update dispatch binding drift")
}
result, err := typedAgents.ExecuteDockerContainerUpdate(ctx, agentID, req)
result, err := typedAgents.ExecuteDockerContainerUpdate(agentCommandContext(ctx), agentID, req)
if err != nil {
return nil, err
}
@@ -261,7 +261,7 @@ func (e dockerContainerActionExecutor) ReconcileActionDispatch(ctx context.Conte
return nil, unified.ActionDispatchReceipt{}, false, nil
}
identity := operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID}
query, err := querier.QueryAgentOperation(ctx, attempt.AgentID, identity)
query, err := querier.QueryAgentOperation(agentCommandContext(ctx), attempt.AgentID, identity)
if err != nil {
return nil, unified.ActionDispatchReceipt{}, false, err
}
@@ -329,12 +329,11 @@ func (e dockerContainerActionExecutor) CheckActionAvailable(ctx context.Context,
if _, err := e.executableDockerContainerResource(ctx, resource, operation); err != nil {
return unavailableDockerActionReadiness(operation, dockerActionUnavailableReasonCode(err), dockerActionUnavailableReason(err))
}
if _, err := e.connectedDockerCommandAgentID(resource); err != nil {
if _, err := e.connectedDockerCommandAgentID(ctx, resource); err != nil {
return unavailableDockerActionReadiness(operation, "command_agent_disconnected", "Docker / Podman command agent is not connected.")
}
agentID, _ := e.connectedDockerCommandAgentID(resource)
liveCapability, supported := e.agents.(agentOperationReceiptCapability)
if !supported || liveCapability.AgentOperationReceiptVersion(agentID) != operationreceipt.ProtocolVersion {
agentID, _ := e.connectedDockerCommandAgentID(ctx, resource)
if liveAgentOperationReceiptVersion(ctx, e.agents, agentID) != operationreceipt.ProtocolVersion {
return unavailableDockerActionReadiness(operation, "operation_receipt_unsupported", "The Pulse agent on this host cannot run reviewed actions: it is on an older version, or its durable state directory is unavailable. Update the agent, or check the agent logs if it is already current, then retry.")
}
return readiness
@@ -377,19 +376,27 @@ func (e dockerContainerActionExecutor) executableDockerContainerResource(_ conte
return resource, nil
}
func (e dockerContainerActionExecutor) connectedDockerCommandAgentID(resource unified.Resource) (string, error) {
func (e dockerContainerActionExecutor) connectedDockerCommandAgentID(ctx context.Context, resource unified.Resource) (string, error) {
if e.agents == nil {
return "", fmt.Errorf("docker container command agent is not connected")
}
if resource.Docker == nil {
return "", fmt.Errorf("docker resource metadata missing")
}
if agentID := strings.TrimSpace(resource.Docker.AgentID); agentID != "" && e.agents.IsAgentConnected(agentID) {
if agentID, ok := commandAgentForToken(ctx, e.agents, resource.Docker.TokenID); ok {
return strings.TrimSpace(agentID), nil
}
// When telemetry names a token, it is the immutable session binding. Do
// not fall back to a different identity or hostname after rotation.
if strings.TrimSpace(resource.Docker.TokenID) != "" {
return "", fmt.Errorf("docker container command agent is not connected")
}
if agentID := strings.TrimSpace(resource.Docker.AgentID); agentID != "" && isAgentCommandConnected(ctx, e.agents, agentID) {
return agentID, nil
}
if agentID, ok := e.agents.GetAgentForHost(strings.TrimSpace(resource.Docker.Hostname)); ok {
if agentID, ok := commandAgentForHost(ctx, e.agents, strings.TrimSpace(resource.Docker.Hostname)); ok {
agentID = strings.TrimSpace(agentID)
if agentID != "" && e.agents.IsAgentConnected(agentID) {
if agentID != "" && isAgentCommandConnected(ctx, e.agents, agentID) {
return agentID, nil
}
}
internal/api/docker_container_action_executor_test.go
+52
View File
@@ -31,6 +31,28 @@ type fakeDockerActionAgentCommander struct {
queries []operationreceipt.Identity
}
type scopedFakeDockerActionAgentCommander struct {
*fakeDockerActionAgentCommander
tokenAgents map[string]string
lastOrg string
}
func (f *scopedFakeDockerActionAgentCommander) IsAgentConnectedForOrganization(organizationID, agentID string) bool {
f.lastOrg = organizationID
return f.IsAgentConnected(agentID)
}
func (f *scopedFakeDockerActionAgentCommander) GetAgentForHostForOrganization(organizationID, hostname string) (string, bool) {
f.lastOrg = organizationID
return f.GetAgentForHost(hostname)
}
func (f *scopedFakeDockerActionAgentCommander) GetAgentForTokenForOrganization(organizationID, tokenID string) (string, bool) {
f.lastOrg = organizationID
agentID, ok := f.tokenAgents[tokenID]
return agentID, ok
}
func dockerActionDispatchContext(t *testing.T, executor dockerContainerActionExecutor, record unified.ActionAuditRecord) context.Context {
t.Helper()
attempt, err := unified.NewActionDispatchAttempt(record.ID, time.Now())
@@ -122,6 +144,36 @@ func dockerActionReadinessByName(readinesses []unified.ResourceActionReadiness,
return unified.ResourceActionReadiness{}, false
}
func TestDockerCommandSessionResolutionUsesTenantAndImmutableReportToken(t *testing.T) {
agents := &scopedFakeDockerActionAgentCommander{
fakeDockerActionAgentCommander: &fakeDockerActionAgentCommander{
connected: map[string]bool{"canonical-agent": true, "stale-agent": true},
agentByHost: map[string]string{"docker-host": "stale-agent"},
},
tokenAgents: map[string]string{"fresh-token": "canonical-agent"},
}
executor := dockerContainerActionExecutor{agents: agents}
ctx := context.WithValue(context.Background(), OrgIDContextKey, "org-b")
resource := unified.Resource{Docker: &unified.DockerData{
TokenID: "fresh-token",
AgentID: "stale-agent",
Hostname: "docker-host",
}}
agentID, err := executor.connectedDockerCommandAgentID(ctx, resource)
if err != nil {
t.Fatalf("resolve command session: %v", err)
}
if agentID != "canonical-agent" || agents.lastOrg != "org-b" {
t.Fatalf("resolved agent/org = %q/%q, want canonical-agent/org-b", agentID, agents.lastOrg)
}
delete(agents.tokenAgents, "fresh-token")
if agentID, err := executor.connectedDockerCommandAgentID(ctx, resource); err == nil {
t.Fatalf("stale token unexpectedly fell back to report identity %q", agentID)
}
}
func TestDockerContainerActionExecutorDispatchesPodmanRestartAndVerification(t *testing.T) {
now := time.Now().UTC()
h := newActionTestResourceHandlers(t, &config.Config{DataPath: t.TempDir()})
internal/api/host_storage_cleanup_action_executor.go
+8 -9
View File
@@ -58,7 +58,7 @@ func (e hostStorageCleanupActionExecutor) ActionDispatchOperationKinds() []strin
return []string{agentexec.HostStorageCleanupOperationPackageCache}
}
func (e hostStorageCleanupActionExecutor) CheckActionAvailable(_ context.Context, req unified.ActionRequest, resource unified.Resource) unified.ResourceActionReadiness {
func (e hostStorageCleanupActionExecutor) CheckActionAvailable(ctx context.Context, req unified.ActionRequest, resource unified.Resource) unified.ResourceActionReadiness {
if strings.TrimSpace(req.CapabilityName) != hostStorageCleanupCapability {
return unified.ResourceActionReadiness{}
}
@@ -66,7 +66,7 @@ func (e hostStorageCleanupActionExecutor) CheckActionAvailable(_ context.Context
if !ok || capability.InternalHandler != hostStorageCleanupActionHandler {
return unified.ResourceActionReadiness{}
}
if err := e.validateResource(resource); err != nil {
if err := e.validateResource(ctx, resource); err != nil {
return unified.ResourceActionReadiness{
Name: hostStorageCleanupCapability,
Available: false,
@@ -107,7 +107,7 @@ func (e hostStorageCleanupActionExecutor) ExecuteAction(ctx context.Context, rec
if agentexec.HostStorageCleanupOperationIdentity(agentID, req) != (operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID}) {
return nil, fmt.Errorf("host cleanup dispatch binding drift")
}
result, err := e.agents.ExecuteHostStorageCleanup(ctx, agentID, req)
result, err := e.agents.ExecuteHostStorageCleanup(agentCommandContext(ctx), agentID, req)
if err != nil {
return nil, err
}
@@ -130,7 +130,7 @@ func (e hostStorageCleanupActionExecutor) ReconcileActionDispatch(ctx context.Co
if !ok {
return nil, unified.ActionDispatchReceipt{}, false, nil
}
query, err := querier.QueryAgentOperation(ctx, attempt.AgentID, identity)
query, err := querier.QueryAgentOperation(agentCommandContext(ctx), attempt.AgentID, identity)
if err != nil {
return nil, unified.ActionDispatchReceipt{}, false, err
}
@@ -174,13 +174,13 @@ func (e hostStorageCleanupActionExecutor) currentResource(ctx context.Context, r
if !ok || resource == nil {
return unified.Resource{}, fmt.Errorf("resource %q is no longer present", resourceID)
}
if err := e.validateResource(*resource); err != nil {
if err := e.validateResource(ctx, *resource); err != nil {
return unified.Resource{}, err
}
return *resource, nil
}
func (e hostStorageCleanupActionExecutor) validateResource(resource unified.Resource) error {
func (e hostStorageCleanupActionExecutor) validateResource(ctx context.Context, resource unified.Resource) error {
if resource.Type != unified.ResourceTypeAgent || resource.Agent == nil {
return fmt.Errorf("resource is not an agent-managed host")
}
@@ -214,11 +214,10 @@ func (e hostStorageCleanupActionExecutor) validateResource(resource unified.Reso
return fmt.Errorf("package cache filesystem is not under storage pressure")
}
agentID := strings.TrimSpace(resource.Agent.AgentID)
liveCapability, supported := e.agents.(agentOperationReceiptCapability)
if agentID == "" || e.agents == nil || !e.agents.IsAgentConnected(agentID) {
if agentID == "" || e.agents == nil || !isAgentCommandConnected(ctx, e.agents, agentID) {
return fmt.Errorf("host command agent is disconnected")
}
if !supported || liveCapability.AgentOperationReceiptVersion(agentID) != operationreceipt.ProtocolVersion {
if liveAgentOperationReceiptVersion(ctx, e.agents, agentID) != operationreceipt.ProtocolVersion {
return fmt.Errorf("live durable operation receipt protocol is unsupported")
}
return nil
internal/api/host_update_action_executor.go
+7 -8
View File
@@ -71,7 +71,7 @@ func (e hostUpdateActionExecutor) CheckActionAvailable(ctx context.Context, req
if !ok || capability.InternalHandler != hostPackageUpdateActionHandler {
return unified.ResourceActionReadiness{}
}
if err := e.validateResource(resource); err != nil {
if err := e.validateResource(ctx, resource); err != nil {
return unified.ResourceActionReadiness{
Name: hostPackageUpdateCapability,
Available: false,
@@ -112,7 +112,7 @@ func (e hostUpdateActionExecutor) ExecuteAction(ctx context.Context, record unif
if agentexec.HostUpdateOperationIdentity(agentID, req) != (operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID}) {
return nil, fmt.Errorf("host update dispatch binding drift")
}
result, err := e.agents.ExecuteHostUpdate(ctx, agentID, req)
result, err := e.agents.ExecuteHostUpdate(agentCommandContext(ctx), agentID, req)
if err != nil {
return nil, err
}
@@ -135,7 +135,7 @@ func (e hostUpdateActionExecutor) ReconcileActionDispatch(ctx context.Context, r
if !ok {
return nil, unified.ActionDispatchReceipt{}, false, nil
}
query, err := querier.QueryAgentOperation(ctx, attempt.AgentID, identity)
query, err := querier.QueryAgentOperation(agentCommandContext(ctx), attempt.AgentID, identity)
if err != nil {
return nil, unified.ActionDispatchReceipt{}, false, err
}
@@ -186,13 +186,13 @@ func (e hostUpdateActionExecutor) currentResource(ctx context.Context, resourceI
if !ok || resource == nil {
return unified.Resource{}, fmt.Errorf("resource %q is no longer present", resourceID)
}
if err := e.validateResource(*resource); err != nil {
if err := e.validateResource(ctx, *resource); err != nil {
return unified.Resource{}, err
}
return *resource, nil
}
func (e hostUpdateActionExecutor) validateResource(resource unified.Resource) error {
func (e hostUpdateActionExecutor) validateResource(ctx context.Context, resource unified.Resource) error {
if resource.Type != unified.ResourceTypeAgent || resource.Agent == nil {
return fmt.Errorf("resource is not an agent-managed host")
}
@@ -223,11 +223,10 @@ func (e hostUpdateActionExecutor) validateResource(resource unified.Resource) er
return fmt.Errorf("host has no pending package updates")
}
agentID := strings.TrimSpace(resource.Agent.AgentID)
liveCapability, supported := e.agents.(agentOperationReceiptCapability)
if agentID == "" || e.agents == nil || !e.agents.IsAgentConnected(agentID) {
if agentID == "" || e.agents == nil || !isAgentCommandConnected(ctx, e.agents, agentID) {
return fmt.Errorf("host command agent is disconnected")
}
if !supported || liveCapability.AgentOperationReceiptVersion(agentID) != operationreceipt.ProtocolVersion {
if liveAgentOperationReceiptVersion(ctx, e.agents, agentID) != operationreceipt.ProtocolVersion {
return fmt.Errorf("live durable operation receipt protocol is unsupported")
}
return nil
internal/api/proxmox_guest_action_executor.go
+8 -8
View File
@@ -82,7 +82,7 @@ func (e proxmoxGuestActionExecutor) ExecuteAction(ctx context.Context, record un
return nil, err
}
vmid := resource.Proxmox.VMID
agentID, err := e.connectedProxmoxNodeCommandAgentID(resource)
agentID, err := e.connectedProxmoxNodeCommandAgentID(ctx, resource)
if err != nil {
return nil, err
}
@@ -95,7 +95,7 @@ func (e proxmoxGuestActionExecutor) ExecuteAction(ctx context.Context, record un
command := proxmoxGuestLifecycleCommand(kind, operation, vmid)
actionStartedAt := time.Now().UTC()
result, err := e.agents.ExecuteCommand(ctx, agentID, agentexec.ExecuteCommandPayload{
result, err := e.agents.ExecuteCommand(agentCommandContext(ctx), agentID, agentexec.ExecuteCommandPayload{
RequestID: attempt.ID,
Command: command,
ApprovalID: record.ID,
@@ -129,7 +129,7 @@ func (e proxmoxGuestActionExecutor) CheckActionAvailable(ctx context.Context, re
if _, _, err := e.executableProxmoxGuestResource(resource, operation); err != nil {
return unavailableProxmoxActionReadiness(operation, proxmoxActionUnavailableReasonCode(err), proxmoxActionUnavailableReason(err))
}
if _, err := e.connectedProxmoxNodeCommandAgentID(resource); err != nil {
if _, err := e.connectedProxmoxNodeCommandAgentID(ctx, resource); err != nil {
return unavailableProxmoxActionReadiness(operation, "command_agent_disconnected", "Proxmox node command agent is not connected.")
}
return readiness
@@ -183,19 +183,19 @@ func (e proxmoxGuestActionExecutor) executableProxmoxGuestResource(resource unif
return resource, kind, nil
}
func (e proxmoxGuestActionExecutor) connectedProxmoxNodeCommandAgentID(resource unified.Resource) (string, error) {
func (e proxmoxGuestActionExecutor) connectedProxmoxNodeCommandAgentID(ctx context.Context, resource unified.Resource) (string, error) {
if e.agents == nil {
return "", fmt.Errorf("proxmox node command agent is not connected")
}
if resource.Proxmox == nil {
return "", fmt.Errorf("proxmox resource metadata missing")
}
if agentID := strings.TrimSpace(resource.Proxmox.LinkedAgentID); agentID != "" && e.agents.IsAgentConnected(agentID) {
if agentID := strings.TrimSpace(resource.Proxmox.LinkedAgentID); agentID != "" && isAgentCommandConnected(ctx, e.agents, agentID) {
return agentID, nil
}
if agentID, ok := e.agents.GetAgentForHost(strings.TrimSpace(resource.Proxmox.NodeName)); ok {
if agentID, ok := commandAgentForHost(ctx, e.agents, strings.TrimSpace(resource.Proxmox.NodeName)); ok {
agentID = strings.TrimSpace(agentID)
if agentID != "" && e.agents.IsAgentConnected(agentID) {
if agentID != "" && isAgentCommandConnected(ctx, e.agents, agentID) {
return agentID, nil
}
}
@@ -344,7 +344,7 @@ func (e proxmoxGuestActionExecutor) verifyProxmoxGuestState(ctx context.Context,
}
}
result, err := e.agents.ExecuteCommand(ctx, agentID, agentexec.ExecuteCommandPayload{
result, err := e.agents.ExecuteCommand(agentCommandContext(ctx), agentID, agentexec.ExecuteCommandPayload{
RequestID: fmt.Sprintf("%s-verify-%d", actionID, attempt+1),
Command: command,
ApprovalID: actionID,
internal/api/router.go
+14 -1
View File
@@ -673,8 +673,21 @@ func (r *Router) setupRoutes() {
r.systemSettingsHandler = NewSystemSettingsHandler(r.config, r.persistence, r.wsHub, r.mtMonitor, r.monitor, r.reloadSystemSettings, r.reloadFunc)
// Agent execution server for AI tool use
r.agentExecServer = agentexec.NewServer(r.validateAgentExecToken)
r.agentExecServer = agentexec.NewServerWithAdmissionValidator(r.admitAgentExecToken, r.validateAgentExecSession)
r.agentExecServer.SetCommandAuthorizationVerifier(verifyAndConsumeCommandAuthorization)
if r.connectionsHandlers != nil {
r.connectionsHandlers.SetAgentCommandSessionProvider(func(organizationID, tokenID, agentID, hostname string) bool {
if strings.TrimSpace(tokenID) != "" {
_, connected := r.agentExecServer.GetAgentForTokenForOrganization(organizationID, tokenID)
return connected
}
if strings.TrimSpace(agentID) != "" && r.agentExecServer.IsAgentConnectedForOrganization(organizationID, agentID) {
return true
}
_, connected := r.agentExecServer.GetAgentForHostForOrganization(organizationID, hostname)
return connected
})
}
if r.resourceHandlers != nil {
r.resourceHandlers.SetActionExecutor(newRoutedActionExecutor(
r.resourceHandlers,
internal/api/router_routes_ai_relay.go
+4 -2
View File
@@ -752,7 +752,7 @@ func (a *agentCommandAdapter) ExecuteCommand(ctx context.Context, agentID, comma
if a.handler.agentServer == nil {
return "", "", -1, fmt.Errorf("agent server not available")
}
result, execErr := a.handler.agentServer.ExecuteCommand(ctx, agentID, agentexec.ExecuteCommandPayload{
result, execErr := a.handler.agentServer.ExecuteCommand(agentCommandContext(ctx), agentID, agentexec.ExecuteCommandPayload{
Command: command,
TargetType: "agent",
})
@@ -766,7 +766,9 @@ func (a *agentCommandAdapter) FindAgentForTarget(targetHost string) string {
if a.handler.agentServer == nil {
return ""
}
agents := a.handler.agentServer.GetConnectedAgents()
// This legacy resolver is only used by the default-organization adapter.
// Tenant Assistant services receive an organization-pinned server view.
agents := a.handler.agentServer.GetConnectedAgentsForOrganization("default")
if len(agents) == 0 {
return ""
}
internal/api/security_regression_test.go
+91
View File
@@ -562,6 +562,60 @@ func TestAgentExecTokenBindingEnforced(t *testing.T) {
conn.Close()
}
func TestAgentExecTokenRejectsAmbiguousMultiOrganizationAuthority(t *testing.T) {
rawToken := "multi-org-agent-token-123.12345678"
record := newTokenRecord(t, rawToken, []string{config.ScopeAgentExec}, map[string]string{
"bound_agent_id": "agent-1",
"bound_hostname": "host-1",
agentExecBindingVersionKey: agentExecBindingVersion,
})
record.OrgIDs = []string{"org-a", "org-b"}
cfg := newTestConfigWithTokens(t, record)
router := NewRouter(cfg, nil, nil, nil, nil, "1.0.0")
if _, ok := router.admitAgentExecToken(rawToken, "agent-1", "host-1"); ok {
t.Fatal("multi-organization exec token was admitted to an ambiguous command session")
}
}
func TestAgentExecTokenBindingFailsClosedAndRestoresMetadataWhenPersistenceFails(t *testing.T) {
rawToken := "persist-failure-agent-token-123.12345678"
record := newTokenRecord(t, rawToken, []string{config.ScopeAgentExec}, map[string]string{
"install_type": "docker",
"issued_via": agentInstallIssuedViaConfig,
})
cfg := newTestConfigWithTokens(t, record)
notDirectory := filepath.Join(t.TempDir(), "not-a-directory")
persistence := config.NewConfigPersistence(notDirectory)
if err := os.RemoveAll(notDirectory); err != nil {
t.Fatalf("remove persistence directory: %v", err)
}
if err := os.WriteFile(notDirectory, []byte("blocked"), 0600); err != nil {
t.Fatalf("create persistence blocker: %v", err)
}
router := &Router{
config: cfg,
persistence: persistence,
}
if _, ok := router.admitAgentExecToken(rawToken, "agent-1", "host-1"); ok {
t.Fatal("first-use command admission succeeded without durable identity binding")
}
config.Mu.RLock()
defer config.Mu.RUnlock()
for _, key := range []string{
"bound_agent_id",
"bound_hostname",
"bound_at",
agentExecBindingVersionKey,
} {
if _, present := cfg.APITokens[0].Metadata[key]; present {
t.Fatalf("failed persistence left transient %q binding metadata behind", key)
}
}
}
// TestAgentExecTokenBindingAcceptsHostnameMatch covers the deploy/enroll flow
// where the runtime token carries both bound_agent_id (server-canonical
// "agent-<hostname>" form) and bound_hostname, but the agent's runtime
@@ -615,6 +669,43 @@ func TestAgentExecTokenBindingAcceptsHostnameMatch(t *testing.T) {
t.Fatalf("expected registration to be accepted when hostname matches bound_hostname, got %q", reg.Message)
}
conn.Close()
config.Mu.RLock()
if got := cfg.APITokens[0].Metadata["bound_agent_id"]; got != "f0c1b2a3e4d5f60718293a4b5c6d7e8f" {
config.Mu.RUnlock()
t.Fatalf("legacy hostname binding migrated agent id = %q", got)
}
if got := cfg.APITokens[0].Metadata[agentExecBindingVersionKey]; got != agentExecBindingVersion {
config.Mu.RUnlock()
t.Fatalf("legacy binding version = %q", got)
}
config.Mu.RUnlock()
// The hostname exception is a one-time upgrade migration. Once the runtime
// identity is persisted, replay with a second ID on the same host fails.
conn, _, err = websocket.DefaultDialer.Dial(wsURL, wsHeadersForHTTP(t, ts.URL))
if err != nil {
t.Fatalf("Dial: %v", err)
}
regMsg, err = agentexec.NewMessage(agentexec.MsgTypeAgentRegister, "", agentexec.AgentRegisterPayload{
AgentID: "second-runtime-id",
Hostname: "prox97",
Version: "1.0.0",
Platform: "linux",
Token: rawToken,
})
if err != nil {
conn.Close()
t.Fatalf("NewMessage: %v", err)
}
if err := conn.WriteJSON(regMsg); err != nil {
conn.Close()
t.Fatalf("WriteJSON: %v", err)
}
if replay := readRegisteredPayload(t, conn); replay.Success {
conn.Close()
t.Fatal("migrated token accepted replay under a second runtime identity")
}
conn.Close()
// Mismatched hostname AND mismatched agent_id must still be rejected —
// a leaked token cannot be used from a different host claiming a different
pkg/server/server.go
+30 -12
View File
@@ -10,6 +10,7 @@ import (
"net/http"
"os"
"os/signal"
"path"
"path/filepath"
"strconv"
"strings"
@@ -186,10 +187,9 @@ func Run(ctx context.Context, version string) error {
}
defer mainListener.Close()
// Optional dedicated agent-ingest listener. When PULSE_AGENT_INGEST_PORT is
// set, agent report/management traffic (/api/agents/*) is also served on this
// separate port so operators can expose it on its own network/firewall
// boundary without exposing the web UI or the rest of the REST API there.
// Optional dedicated agent-control listener. Reports, command admission,
// version checks, and bootstrap downloads share this network boundary;
// the web UI and management API are not exposed on it.
var agentListener net.Listener
if cfg.AgentIngestPort > 0 {
agentAddr := fmt.Sprintf("%s:%d", cfg.BindAddress, cfg.AgentIngestPort)
@@ -633,9 +633,8 @@ func Run(ctx context.Context, version string) error {
},
}
// When a dedicated agent-ingest listener is configured, serve only the
// /api/agents/* surface on it so that port never exposes the web UI or the
// rest of the REST API.
// The dedicated listener serves the complete agent control plane, while
// excluding the web UI and management REST API.
var agentSrv *http.Server
if agentListener != nil {
agentSrv = &http.Server{
@@ -876,13 +875,12 @@ shutdown:
return runErr
}
// agentIngestHandler restricts a handler to the agent-ingest surface
// (/api/agents/*). It backs the optional dedicated agent-ingest listener so
// that port serves only agent report/management endpoints and never the web UI
// or the rest of the REST API. Every other path returns 404.
// agentIngestHandler restricts a handler to the complete agent control plane.
// Reports, command admission, version checks, and bootstrap downloads must
// share one reachable listener; management APIs and the web UI remain absent.
func agentIngestHandler(inner http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if !strings.HasPrefix(r.URL.Path, "/api/agents/") {
if !isAgentControlPlanePath(r.URL.Path) {
http.NotFound(w, r)
return
}
@@ -890,6 +888,26 @@ func agentIngestHandler(inner http.Handler) http.Handler {
})
}
func isAgentControlPlanePath(requestPath string) bool {
if requestPath == "" || path.Clean(requestPath) != requestPath {
return false
}
if strings.HasPrefix(requestPath, "/api/agents/") {
return true
}
switch requestPath {
case "/api/agent/ws",
"/api/agent/version",
"/api/server/info",
"/install.sh",
"/install.ps1",
"/download/pulse-agent":
return true
default:
return false
}
}
// startMetricsServer starts the Prometheus /metrics endpoint. When metricsToken
// is non-empty, requests must include a matching Authorization: Bearer <token> header.
func startMetricsServer(ctx context.Context, addr string, metricsToken string, allowInsecureRemote bool) error {
pkg/server/server_test.go
+56
View File
@@ -12,7 +12,10 @@ import (
"testing"
"time"
"github.com/gorilla/websocket"
"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
"github.com/rcourtman/pulse-go-rewrite/pkg/extensions"
pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing"
"github.com/rcourtman/pulse-go-rewrite/pkg/metrics"
@@ -36,6 +39,12 @@ func TestAgentIngestHandler(t *testing.T) {
{"/api/agents/kubernetes/report", true, http.StatusOK},
{"/api/agents/agent/lookup", true, http.StatusOK},
{"/api/agents/agent/config", true, http.StatusOK},
{"/api/agent/ws", true, http.StatusOK},
{"/api/agent/version", true, http.StatusOK},
{"/api/server/info", true, http.StatusOK},
{"/install.sh", true, http.StatusOK},
{"/install.ps1", true, http.StatusOK},
{"/download/pulse-agent", true, http.StatusOK},
// Everything outside the agent-ingest surface must be rejected so the
// dedicated port never exposes the web UI or the rest of the REST API.
{"/", false, http.StatusNotFound},
@@ -44,6 +53,10 @@ func TestAgentIngestHandler(t *testing.T) {
{"/api/state", false, http.StatusNotFound},
{"/api/security/status", false, http.StatusNotFound},
{"/api/agents", false, http.StatusNotFound},
{"/api/agents/../security/status", false, http.StatusNotFound},
{"/api/agents//agent/report", false, http.StatusNotFound},
{"/api/agent/ws/extra", false, http.StatusNotFound},
{"/install.sh/extra", false, http.StatusNotFound},
}
for _, tc := range cases {
innerCalled = false
@@ -59,6 +72,49 @@ func TestAgentIngestHandler(t *testing.T) {
}
}
func TestAgentControlPlaneListenerAdmitsCommandWebSocket(t *testing.T) {
execServer := agentexec.NewServer(func(token, agentID, hostname string) bool {
return token == "exec-token" && agentID == "docker-agent" && hostname == "docker-host"
})
t.Cleanup(execServer.Shutdown)
server := httptest.NewServer(agentIngestHandler(http.HandlerFunc(execServer.HandleWebSocket)))
defer server.Close()
origin, err := securityutil.HTTPOriginForWebSocketBaseURL(server.URL)
if err != nil {
t.Fatalf("origin: %v", err)
}
headers := http.Header{}
headers.Set("Origin", origin)
wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/agent/ws"
conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
if err != nil {
t.Fatalf("dial dedicated agent command websocket: %v", err)
}
defer conn.Close()
registration, err := agentexec.NewMessage(agentexec.MsgTypeAgentRegister, "", agentexec.AgentRegisterPayload{
AgentID: "docker-agent", Hostname: "docker-host", Token: "exec-token",
})
if err != nil {
t.Fatalf("registration message: %v", err)
}
if err := conn.WriteJSON(registration); err != nil {
t.Fatalf("write registration: %v", err)
}
var response agentexec.Message
if err := conn.ReadJSON(&response); err != nil {
t.Fatalf("read registration acknowledgement: %v", err)
}
var acknowledged agentexec.RegisteredPayload
if err := response.DecodePayload(&acknowledged); err != nil {
t.Fatalf("decode acknowledgement: %v", err)
}
if !acknowledged.Success || !execServer.IsAgentConnected("docker-agent") {
t.Fatalf("dedicated listener did not admit command channel: %+v", acknowledged)
}
}
func TestBusinessHooks(t *testing.T) {
called := false
hook := func(store *metrics.Store) {
scripts/release_control/canonical_completion_guard_test.py
+10
View File
@@ -287,6 +287,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"internal/monitoring/availability_poller_test.go",
"internal/monitoring/availability_udp_test.go",
"internal/monitoring/canonical_guardrails_test.go",
"internal/monitoring/ceph_test.go",
"internal/monitoring/issue1485_unraid_lifecycle_test.go",
"internal/monitoring/issue1595_collection_trust_test.go",
"internal/monitoring/issue1613_contract_test.go",
@@ -300,6 +301,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"internal/monitoring/monitor_pve_cluster_refresh_test.go",
"internal/monitoring/monitor_pve_guest_lxc_test.go",
"internal/monitoring/ratetracker_test.go",
"internal/monitoring/truenas_poller_test.go",
"internal/unifiedresources/code_standards_test.go",
],
}
@@ -466,6 +468,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"internal/api/resources_test.go",
"internal/storagehealth/risk_test.go",
"internal/storagehealth/topology_test.go",
"internal/storagehealth/zfs_pool_health_contract_test.go",
],
}
],
@@ -1163,11 +1166,13 @@ class CanonicalCompletionGuardTest(unittest.TestCase):
"frontend-modern/src/types/api.ts",
"internal/api/ai_handlers_more_test.go",
"internal/api/ai_handlers_patrol_actions_additional_test.go",
"internal/api/audit_handlers_test.go",
"internal/api/contract_test.go",
"internal/api/docker_agents_report_size_test.go",
"internal/api/host_agent_removal_lifecycle_integration_test.go",
"internal/api/metadata_handlers_test.go",
"internal/api/patrol_autopilot_test.go",
"pulse-enterprise:test/extensions_contract_test.go",
],
}
],
@@ -2572,6 +2577,7 @@ None yet.
"exact_files": [
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
}
],
@@ -2597,6 +2603,7 @@ None yet.
"exact_files": [
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
}
],
@@ -2622,6 +2629,7 @@ None yet.
"exact_files": [
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
}
],
@@ -2653,6 +2661,7 @@ None yet.
"exact_files": [
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
}
],
@@ -2722,6 +2731,7 @@ None yet.
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
scripts/release_control/subsystem_lookup_test.py
+8
View File
@@ -2205,6 +2205,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
@@ -2263,6 +2264,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
@@ -2294,6 +2296,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
@@ -2325,6 +2328,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
@@ -2356,6 +2360,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
@@ -2387,6 +2392,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
security_match = next(
@@ -2435,6 +2441,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)
@@ -2518,6 +2525,7 @@ class SubsystemLookupTest(unittest.TestCase):
[
"frontend-modern/src/components/Settings/__tests__/dataHandlingPanelModel.test.ts",
"frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts",
"frontend-modern/src/components/Settings/__tests__/useAuditLogPanelState.test.tsx",
],
)