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

Harden agentexec token binding and disk filtering

Browse Source
This commit is contained in:
rcourtman
2026-04-23 15:54:48 +01:00
parent 3d82140997
commit 60d7db6ef9
16 changed files with 238 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/agentexec/deploy_test.go
+7 -7
View File
@@ -167,7 +167,7 @@ func TestDeployCancelPayloadRoundTrip(t *testing.T) {
}
func TestSubscribeDeployProgress(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
ch := s.SubscribeDeployProgress("agent-1", "job-1", 10)
if ch == nil {
@@ -196,7 +196,7 @@ func TestSubscribeDeployProgress(t *testing.T) {
}
func TestSubscribeDeployProgressDefaultBuffer(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
ch := s.SubscribeDeployProgress("agent-1", "job-2", 0)
if cap(ch) != 64 {
t.Errorf("expected default capacity 64, got %d", cap(ch))
@@ -205,7 +205,7 @@ func TestSubscribeDeployProgressDefaultBuffer(t *testing.T) {
}
func TestSubscribeDeployProgressAgentIsolation(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
ch1 := s.SubscribeDeployProgress("agent-1", "job-1", 10)
ch2 := s.SubscribeDeployProgress("agent-2", "job-1", 10)
@@ -226,7 +226,7 @@ func TestSubscribeDeployProgressAgentIsolation(t *testing.T) {
}
func TestSendDeployPreflightAgentNotConnected(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
err := s.SendDeployPreflight(nil, "missing-agent", DeployPreflightPayload{
RequestID: "req-1",
@@ -238,7 +238,7 @@ func TestSendDeployPreflightAgentNotConnected(t *testing.T) {
}
func TestSendDeployInstallAgentNotConnected(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
err := s.SendDeployInstall(nil, "missing-agent", DeployInstallPayload{
RequestID: "req-1",
@@ -250,7 +250,7 @@ func TestSendDeployInstallAgentNotConnected(t *testing.T) {
}
func TestSendDeployCancelAgentNotConnected(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
err := s.SendDeployCancel(nil, "missing-agent", DeployCancelPayload{
RequestID: "req-1",
@@ -262,7 +262,7 @@ func TestSendDeployCancelAgentNotConnected(t *testing.T) {
}
func TestSendDeployCommandEmptyAgentID(t *testing.T) {
s := NewServer(func(string, string) bool { return true })
s := NewServer(func(string, string, string) bool { return true })
err := s.SendDeployPreflight(nil, "", DeployPreflightPayload{RequestID: "req-1"})
if err == nil {
internal/agentexec/server.go
+14 -4
View File
@@ -61,7 +61,7 @@ type Server struct {
agents map[string]*agentConn // agentID -> connection
pendingReqs map[string]chan CommandResultPayload // scoped request key -> response channel
deploySubs map[string]chan DeployProgressPayload // deploySubKey(agentID, jobID) -> progress subscriber
validateToken func(token string, agentID string) bool
validateToken func(token string, agentID string, hostname string) bool
commandPolicy *CommandPolicy
ipConnCounts map[string]int
maxConnsPerIP int
@@ -88,8 +88,18 @@ func (ac *agentConn) signalDone() {
})
}
// NewServer creates a new agent execution server
func NewServer(validateToken func(token string, agentID string) bool) *Server {
// NewServer creates a new agent execution server.
//
// validateToken is invoked during WebSocket agent registration with the token,
// the agent-claimed agentID, and the hostname from the register payload. The
// hostname is provided because enrollment-minted tokens bind to bound_hostname
// rather than to a predictable agent ID: agents derive their runtime agentID
// from /etc/machine-id (or an override), which the server cannot know when it
// mints the token. Matching on hostname preserves the trust boundary ("the
// bearer is running on the bound host") without requiring the agent to know a
// server-canonical ID format. See internal/api/router.go for the production
// validator.
func NewServer(validateToken func(token string, agentID string, hostname string) bool) *Server {
if validateToken == nil {
panic("agentexec: validateToken is required")
}
@@ -414,7 +424,7 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
}
// Validate token
if !s.validateToken(reg.Token, reg.AgentID) {
if !s.validateToken(reg.Token, reg.AgentID, reg.Hostname) {
log.Warn().Str("agent_id", reg.AgentID).Msg("Agent registration rejected: invalid token")
rejectedMsg, err := NewMessage(MsgTypeRegistered, "", RegisteredPayload{Success: false, Message: "Invalid token"})
if err != nil {
internal/agentexec/server_coverage_test.go
+1 -1
View File
@@ -150,7 +150,7 @@ func TestHandleWebSocket_InvalidTokenRejectionSendFailure(t *testing.T) {
return errors.New("write failure")
}
s := NewServer(func(string, string) bool { return false })
s := NewServer(func(string, string, string) bool { return false })
ts := newWSServer(t, s)
defer ts.Close()
internal/agentexec/server_test.go
+1 -1
View File
@@ -7,7 +7,7 @@ import (
"time"
)
func allowAllTestTokens(string, string) bool { return true }
func allowAllTestTokens(string, string, string) bool { return true }
func TestNewServerRequiresValidateToken(t *testing.T) {
defer func() {
internal/agentexec/server_websocket_test.go
+2 -2
View File
@@ -118,7 +118,7 @@ func waitFor(t *testing.T, timeout time.Duration, cond func() bool) {
}
func TestHandleWebSocket_RegistrationSuccessAndDisconnectRemovesAgent(t *testing.T) {
s := NewServer(func(token string, agentID string) bool { return token == "ok" })
s := NewServer(func(token string, agentID string, hostname string) bool { return token == "ok" })
ts := newWSServer(t, s)
defer ts.Close()
@@ -227,7 +227,7 @@ func TestHandleWebSocket_RejectsPerIPConnectionFlood(t *testing.T) {
}
func TestHandleWebSocket_InvalidTokenRejected(t *testing.T) {
s := NewServer(func(string, string) bool { return false })
s := NewServer(func(string, string, string) bool { return false })
ts := newWSServer(t, s)
defer ts.Close()
internal/ai/discovery_adapter_test.go
+3 -3
View File
@@ -50,7 +50,7 @@ func TestDiscoveryCommandAdapter_NilServer(t *testing.T) {
}
func TestDiscoveryCommandAdapter_ExecuteCommandNotConnected(t *testing.T) {
server := agentexec.NewServer(func(string, string) bool { return true })
server := agentexec.NewServer(func(string, string, string) bool { return true })
adapter := newDiscoveryCommandAdapter(server)
cmd := servicediscovery.ExecuteCommandPayload{
@@ -76,7 +76,7 @@ func TestDiscoveryCommandAdapter_ExecuteCommandNotConnected(t *testing.T) {
}
func TestDiscoveryCommandAdapter_ConnectedAgentsAndLookup(t *testing.T) {
server := agentexec.NewServer(func(string, string) bool { return true })
server := agentexec.NewServer(func(string, string, string) bool { return true })
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
server.HandleWebSocket(w, r)
}))
@@ -153,7 +153,7 @@ func TestDiscoveryCommandAdapter_ConnectedAgentsAndLookup(t *testing.T) {
}
func TestDiscoveryCommandAdapter_ExecuteCommandSuccess(t *testing.T) {
server := agentexec.NewServer(func(string, string) bool { return true })
server := agentexec.NewServer(func(string, string, string) bool { return true })
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
server.HandleWebSocket(w, r)
}))
internal/ai/patrol_prober_test.go
+2 -2
View File
@@ -82,7 +82,7 @@ func TestAgentExecProber_PingGuestsNilServer(t *testing.T) {
}
func TestAgentExecProber_PingGuestsEmptyIPs(t *testing.T) {
prober := NewAgentExecProber(agentexec.NewServer(func(string, string) bool { return true }))
prober := NewAgentExecProber(agentexec.NewServer(func(string, string, string) bool { return true }))
results, err := prober.PingGuests(context.Background(), "agent-1", nil)
if err != nil {
t.Fatalf("PingGuests with empty ips returned error: %v", err)
@@ -93,7 +93,7 @@ func TestAgentExecProber_PingGuestsEmptyIPs(t *testing.T) {
}
func TestAgentExecProber_RoundTripViaAgentExecServer(t *testing.T) {
server := agentexec.NewServer(func(string, string) bool { return true })
server := agentexec.NewServer(func(string, string, string) bool { return true })
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
server.HandleWebSocket(w, r)
}))
internal/api/ai_handlers_investigation_additional_test.go
+1 -1
View File
@@ -660,7 +660,7 @@ func registerAgent(t *testing.T, url, agentID, hostname string) *websocket.Conn
}
func TestAgentCommandAdapter_FindAgentForTarget(t *testing.T) {
server := agentexec.NewServer(func(string, string) bool { return true })
server := agentexec.NewServer(func(string, string, string) bool { return true })
ts := newIPv4HTTPServer(t, http.HandlerFunc(server.HandleWebSocket))
defer ts.Close()
internal/api/ai_handlers_test.go
+4 -4
View File
@@ -1826,7 +1826,7 @@ func TestHandleRunCommand_ConsumesApproval(t *testing.T) {
tmp := t.TempDir()
cfg := &config.Config{DataPath: tmp}
persistence := config.NewConfigPersistence(tmp)
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string) bool { return true }))
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string, string) bool { return true }))
store, err := approval.NewStore(approval.StoreConfig{
DataDir: tmp,
@@ -1862,7 +1862,7 @@ func TestHandleRunCommand_RejectsCommandMismatch(t *testing.T) {
tmp := t.TempDir()
cfg := &config.Config{DataPath: tmp}
persistence := config.NewConfigPersistence(tmp)
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string) bool { return true }))
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string, string) bool { return true }))
store, err := approval.NewStore(approval.StoreConfig{
DataDir: tmp,
@@ -1898,7 +1898,7 @@ func TestHandleRunCommand_RejectsUnsupportedTargetType(t *testing.T) {
tmp := t.TempDir()
cfg := &config.Config{DataPath: tmp}
persistence := config.NewConfigPersistence(tmp)
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string) bool { return true }))
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string, string) bool { return true }))
store, err := approval.NewStore(approval.StoreConfig{
DataDir: tmp,
@@ -1935,7 +1935,7 @@ func TestHandleRunCommand_RejectsCrossOrgApproval(t *testing.T) {
tmp := t.TempDir()
cfg := &config.Config{DataPath: tmp}
persistence := config.NewConfigPersistence(tmp)
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string) bool { return true }))
handler := newTestAISettingsHandler(cfg, persistence, agentexec.NewServer(func(string, string, string) bool { return true }))
store, err := approval.NewStore(approval.StoreConfig{
DataDir: tmp,
internal/api/deploy_handlers_test.go
+1 -1
View File
@@ -40,7 +40,7 @@ func newTestDeployHandlers(t *testing.T, nodes []models.Node, hosts []models.Hos
state.UpsertHost(h)
}
execServer := agentexec.NewServer(func(string, string) bool { return true })
execServer := agentexec.NewServer(func(string, string, string) bool { return true })
reservation := deploy.NewReservationManager()
cfg := &config.Config{
internal/api/security_regression_test.go
+84
View File
@@ -553,6 +553,90 @@ func TestAgentExecTokenBindingEnforced(t *testing.T) {
conn.Close()
}
// 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
// agent_id is derived locally from /etc/machine-id and does NOT match
// bound_agent_id. The hostname is the authoritative binding — the agent
// proves it is running on the bound host by registering with that hostname.
//
// Regression: prior to this test, the hardening commit 3ec2c0779 enforced
// strict bound_agent_id equality, which silently rejected every deploy-flow
// agent (they never produce "agent-<hostname>" as their runtime ID) and made
// the AI command tool report "No agents are currently connected" despite
// agents appearing online via HTTP reports.
func TestAgentExecTokenBindingAcceptsHostnameMatch(t *testing.T) {
rawToken := "agent-deploy-token-123.12345678"
record := newTokenRecord(t, rawToken, []string{config.ScopeAgentExec}, map[string]string{
"bound_agent_id": "agent-prox97",
"bound_hostname": "prox97",
})
cfg := newTestConfigWithTokens(t, record)
router := NewRouter(cfg, nil, nil, nil, nil, "1.0.0")
ts := newIPv4HTTPServer(t, router.Handler())
defer ts.Close()
wsURL := wsURLForHTTP(ts.URL) + "/api/agent/ws"
// Agent registers with machine-id-style agent_id that does NOT match
// bound_agent_id, but hostname matches bound_hostname. Must succeed.
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: "f0c1b2a3e4d5f60718293a4b5c6d7e8f",
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)
}
reg := readRegisteredPayload(t, conn)
if !reg.Success {
conn.Close()
t.Fatalf("expected registration to be accepted when hostname matches bound_hostname, got %q", reg.Message)
}
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
// agent_id.
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: "attacker-id",
Hostname: "attacker-host",
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)
}
reg = readRegisteredPayload(t, conn)
if reg.Success {
conn.Close()
t.Fatalf("expected registration to be rejected when neither hostname nor agent_id match the binding")
}
conn.Close()
}
func TestSecurityTokens_AgentExecRejectsUnboundToken(t *testing.T) {
rawToken := "agent-unbound-token-123.12345678"
record := newTokenRecord(t, rawToken, []string{config.ScopeAgentExec}, nil)
internal/hostagent/smartctl.go
+10
View File
@@ -17,6 +17,8 @@ import (
"time"
"github.com/rs/zerolog/log"
"github.com/rcourtman/pulse-go-rewrite/pkg/fsfilters"
)
const smartctlComponent = "smartctl_collector"
@@ -355,6 +357,14 @@ func parseSmartctlScanOpenTargets(output []byte, diskExclude []string) []smartct
}
name := filepath.Base(path)
if fsfilters.IsVirtualBlockDevice(name) {
log.Debug().
Str("component", smartctlComponent).
Str("action", "skip_virtual_device").
Str("device", path).
Msg("Skipping non-physical device reported by smartctl --scan-open")
continue
}
if matchesDeviceExclude(name, path, diskExclude) {
continue
}
internal/unifiedresources/registry.go
+4
View File
@@ -11,6 +11,7 @@ import (
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/pkg/fsfilters"
)
const autoMergeThreshold = 0.85
@@ -816,6 +817,9 @@ func (rr *ResourceRegistry) ingestHostSMARTDisks(host models.Host) {
hostParentID := rr.sourceResourceID(SourceAgent, host.ID)
unraidStorageID := rr.sourceResourceID(SourceAgent, hostUnraidStorageSourceID(host))
for _, disk := range host.Sensors.SMART {
if fsfilters.IsVirtualBlockDevice(disk.Device) {
continue
}
resource, identity := resourceFromHostSMARTDisk(host, disk)
if resource.PhysicalDisk == nil {
continue
internal/unifiedresources/registry_test.go
+34
View File
@@ -1276,6 +1276,40 @@ func TestResourceRegistry_IngestSnapshotCreatesPhysicalDisksFromHostSMART(t *tes
}
}
func TestResourceRegistry_IngestSnapshotSkipsVirtualBlockDevicesFromHostSMART(t *testing.T) {
rr := NewRegistry(nil)
now := time.Date(2026, 4, 23, 12, 0, 0, 0, time.UTC)
rr.IngestSnapshot(models.StateSnapshot{
Hosts: []models.Host{
{
ID: "host-minipc",
Hostname: "minipc",
Status: "online",
LastSeen: now,
Sensors: models.HostSensorSummary{
SMART: []models.HostDiskSMART{
{Device: "/dev/nvme0n1", Model: "Samsung 970 EVO", Serial: "REAL-DISK", Type: "nvme", Health: "PASSED"},
{Device: "/dev/zd0", Model: "", Serial: "", Type: ""},
{Device: "/dev/zd16", Model: "", Serial: "", Type: ""},
{Device: "zram0", Model: "", Serial: "", Type: ""},
{Device: "/dev/loop3", Model: "", Serial: "", Type: ""},
{Device: "/dev/dm-1", Model: "", Serial: "", Type: ""},
},
},
},
},
})
disks := rr.ListByType(ResourceTypePhysicalDisk)
if len(disks) != 1 {
t.Fatalf("expected 1 physical disk resource (virtual devices filtered), got %d", len(disks))
}
if disks[0].PhysicalDisk == nil || disks[0].PhysicalDisk.Serial != "REAL-DISK" {
t.Fatalf("expected only the real nvme disk, got %+v", disks[0].PhysicalDisk)
}
}
func TestResourceRegistry_IngestSnapshotMergesAgentAndProxmoxPhysicalDisksByIdentity(t *testing.T) {
rr := NewRegistry(nil)
now := time.Date(2026, 3, 7, 12, 0, 0, 0, time.UTC)
pkg/fsfilters/filters.go
+38
View File
@@ -236,6 +236,44 @@ func MatchesDiskExclude(device, mountpoint string, excludePatterns []string) boo
return false
}
// virtualBlockDevicePrefixes are device-name prefixes for virtual or
// pseudo block devices that should never be treated as physical disks.
// The list is shared between the host agent (which skips these during
// SMART collection) and the server-side resource registry (which refuses
// to surface them as physical_disk resources even if an older agent
// reports them).
var virtualBlockDevicePrefixes = []string{
"dm-",
"drbd",
"loop",
"md",
"nbd",
"pmem",
"ram",
"rbd",
"vd",
"xvd",
"zd",
"zram",
}
// IsVirtualBlockDevice reports whether a block-device name (with or
// without the /dev/ prefix) looks like a virtual or pseudo device that
// cannot provide SMART data or meaningful physical-disk metrics.
func IsVirtualBlockDevice(name string) bool {
trimmed := strings.ToLower(strings.TrimSpace(name))
trimmed = strings.TrimPrefix(trimmed, "/dev/")
if trimmed == "" {
return false
}
for _, prefix := range virtualBlockDevicePrefixes {
if strings.HasPrefix(trimmed, prefix) {
return true
}
}
return false
}
// MatchesDeviceExclude checks if a device name/path matches exclusion patterns.
// For disk I/O collection where we only have device names (not mountpoints).
func MatchesDeviceExclude(device string, excludePatterns []string) bool {
pkg/fsfilters/filters_test.go
+32
View File
@@ -321,3 +321,35 @@ func TestMatchesDeviceExclude(t *testing.T) {
})
}
}
func TestIsVirtualBlockDevice(t *testing.T) {
tests := []struct {
name string
device string
expected bool
}{
{"zfs zvol", "zd0", true},
{"zfs zvol numeric tail", "zd48", true},
{"zram device", "zram0", true},
{"loopback", "loop3", true},
{"device mapper", "dm-7", true},
{"raid md", "md0", true},
{"virtio disk", "vda", true},
{"xen disk", "xvda", true},
{"with dev prefix", "/dev/zram0", true},
{"uppercase prefix", "/DEV/ZD0", true},
{"sata disk", "sda", false},
{"sata partition-ish", "sdb3", false},
{"nvme", "nvme0n1", false},
{"empty", "", false},
{"whitespace", " ", false},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
if got := IsVirtualBlockDevice(tc.device); got != tc.expected {
t.Errorf("IsVirtualBlockDevice(%q) = %t, want %t", tc.device, got, tc.expected)
}
})
}
}