Watch
1
0
Fork
You've already forked RedFlag
0

feat: process explorer — on-demand /proc scanning with full osquery parity

Agent-side: reads /proc for all PIDs with 25+ fields (identity, resources,
state, disk I/O, elevation) plus related data on drill-down (open files,
sockets, pipes, env keys, memory map, namespaces, listening ports). Pure
/proc reads, no subprocess spawns.

Server-side: dedicated tables (agent_process_snapshots, agent_processes,
agent_process_related) with JSONB for related data. On-demand scan via
scan_processes command, last-10-snapshot retention. Four endpoints:
report, get latest, get detail, trigger scan.

UI: new Processes tab in agent detail with sortable/filterable table,
search by name/cmdline, state/user filters, and ProcessDetailModal with
tabs for Overview, Network, Files, Environment, Memory, Namespaces.
This commit is contained in:
Fimeg 2026-06-10 21:28:34 -04:00
commit 244d9091ee
31 changed files with 4127 additions and 13 deletions

View file

@ -59,16 +59,19 @@ jobs:
rust_target: aarch64-unknown-linux-gnu
linker: gcc-aarch64-linux-gnu
use_zigbuild: false
skip_helper: false
- goos: windows
goarch: amd64
rust_target: x86_64-pc-windows-gnu
rust_target: ""
linker: gcc-mingw-w64-x86-64
use_zigbuild: false
skip_helper: true
- goos: darwin
goarch: arm64
rust_target: aarch64-apple-darwin
linker: ""
use_zigbuild: true
skip_helper: false
steps:
- uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
- uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5
@ -85,7 +88,7 @@ jobs:
- name: Install cargo-zigbuild
if: matrix.use_zigbuild
run: pip3 install cargo-zigbuild
run: pip3 install --break-system-packages cargo-zigbuild
- name: Cross-compile Go (server)
env:
@ -102,6 +105,7 @@ jobs:
run: cd agent && go build -o /dev/null ./cmd/agent/
- name: Cross-compile Rust (helper)
if: "!matrix.skip_helper"
run: |
cd helper
if [ "${{ matrix.use_zigbuild }}" = "true" ]; then

View file

@ -117,24 +117,28 @@ jobs:
suffix: linux-amd64
linker: ""
use_zigbuild: false
skip_helper: false
- goos: linux
goarch: arm64
rust_target: aarch64-unknown-linux-gnu
suffix: linux-arm64
linker: gcc-aarch64-linux-gnu
use_zigbuild: false
skip_helper: false
- goos: windows
goarch: amd64
rust_target: x86_64-pc-windows-gnu
rust_target: ""
suffix: windows-amd64
linker: gcc-mingw-w64-x86-64
use_zigbuild: false
skip_helper: true
- goos: darwin
goarch: arm64
rust_target: aarch64-apple-darwin
suffix: darwin-arm64
linker: ""
use_zigbuild: true
skip_helper: false
steps:
- uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
with:
@ -155,7 +159,7 @@ jobs:
- name: Install cargo-zigbuild
if: matrix.use_zigbuild
run: pip3 install cargo-zigbuild
run: pip3 install --break-system-packages cargo-zigbuild
- name: Download web UI
uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4
@ -193,6 +197,7 @@ jobs:
-o ../dist/redflag-agent-${{ matrix.suffix }}${EXT} ./cmd/agent/
- name: Build helper
if: "!matrix.skip_helper"
run: |
cd helper
if [ "${{ matrix.use_zigbuild }}" = "true" ]; then
@ -223,8 +228,10 @@ jobs:
run: |
VERSION=${GITHUB_REF#refs/tags/v}
cd dist
ls -la
if [ "${{ matrix.goos }}" = "windows" ]; then
zip redflag-$VERSION-${{ matrix.suffix }}.zip redflag-*-${{ matrix.suffix }}.exe
# Windows: zip (no helper — it's Unix-only)
zip redflag-$VERSION-${{ matrix.suffix }}.zip redflag-server-${{ matrix.suffix }}.exe redflag-agent-${{ matrix.suffix }}.exe
sha256sum redflag-$VERSION-${{ matrix.suffix }}.zip > checksums-$VERSION-${{ matrix.suffix }}.txt
else
tar czf redflag-$VERSION-${{ matrix.suffix }}.tar.gz redflag-*-${{ matrix.suffix }}

View file

@ -200,6 +200,97 @@ func (c *Client) ReportEvents(agentID uuid.UUID, events []*models.SystemEvent) (
return result.Accepted, result.Rejected, nil
}
// AgentSecurityEvent is the wire format for security events sent from agent
// to server. The server handler maps this onto its SecurityEvent model,
// stamping AgentID from the URL parameter.
type AgentSecurityEvent struct {
Timestamp time.Time `json:"timestamp"`
Level string `json:"level"`
EventType string `json:"event_type"`
Message string `json:"message"`
Details map[string]interface{} `json:"details,omitempty"`
}
// ReportSecurityEvents sends buffered security events to the server.
// POST /api/v1/agents/:id/security-events
// Returns (accepted, rejected, error).
func (c *Client) ReportSecurityEvents(agentID uuid.UUID, events []AgentSecurityEvent) (int, int, error) {
if len(events) == 0 {
return 0, 0, nil
}
url := fmt.Sprintf("%s/api/v1/agents/%s/security-events", c.baseURL, agentID.String())
body, err := json.Marshal(map[string]interface{}{"events": events})
if err != nil {
return 0, 0, fmt.Errorf("failed to marshal security events: %w", err)
}
req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
if err != nil {
return 0, 0, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+c.token)
c.addMachineIDHeader(req)
resp, err := c.http.Do(req)
if err != nil {
return 0, 0, fmt.Errorf("failed to send security events: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
return 0, 0, fmt.Errorf("server returned status %d", resp.StatusCode)
}
var result struct {
Accepted int `json:"accepted"`
Rejected int `json:"rejected"`
Errors []string `json:"errors,omitempty"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return 0, 0, fmt.Errorf("failed to decode response: %w", err)
}
return result.Accepted, result.Rejected, nil
}
// ReportInventory sends inventory data to the server.
// POST /api/v1/agents/:id/inventory
func (c *Client) ReportInventory(agentID uuid.UUID, report InventoryReport) error {
url := fmt.Sprintf("%s/api/v1/agents/%s/inventory", c.baseURL, agentID.String())
body, err := json.Marshal(report)
if err != nil {
return fmt.Errorf("failed to marshal inventory report: %w", err)
}
req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+c.token)
c.addMachineIDHeader(req)
resp, err := c.http.Do(req)
if err != nil {
return fmt.Errorf("failed to send inventory report: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
bodyBytes, _ := io.ReadAll(resp.Body)
return fmt.Errorf("server returned status %d: %s", resp.StatusCode, string(bodyBytes))
}
return nil
}
// addMachineIDHeader adds X-Machine-ID header to authenticated requests (v0.1.22+)
func (c *Client) addMachineIDHeader(req *http.Request) {
if c.machineID != "" {
@ -942,6 +1033,45 @@ func (c *Client) ReportStorageMetrics(agentID uuid.UUID, report models.StorageMe
return nil
}
// ProcessScanReport represents a full process scan result
type ProcessScanReport struct {
AgentID uuid.UUID `json:"agent_id"`
CommandID string `json:"command_id"`
Timestamp time.Time `json:"timestamp"`
Snapshot system.FullProcessSnapshot `json:"snapshot"`
}
// ReportProcessScan sends a full process scan to the server via dedicated endpoint
func (c *Client) ReportProcessScan(agentID uuid.UUID, report ProcessScanReport) error {
url := fmt.Sprintf("%s/api/v1/agents/%s/process-scan", c.baseURL, agentID)
body, err := json.Marshal(report)
if err != nil {
return fmt.Errorf("failed to marshal process scan: %w", err)
}
req, err := http.NewRequest("POST", url, bytes.NewBuffer(body))
if err != nil {
return fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+c.token)
c.addMachineIDHeader(req)
resp, err := c.http.Do(req)
if err != nil {
return fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
bodyBytes, _ := io.ReadAll(resp.Body)
return fmt.Errorf("failed to report process scan: %s - %s", resp.Status, string(bodyBytes))
}
return nil
}
// LogReport represents an execution log
type LogReport struct {
CommandID string `json:"command_id"`

View file

@ -0,0 +1,31 @@
package client
import "time"
// InventoryItem represents a single inventory record -- something present on
// the system. Distinct from UpdateReportItem (which represents an available
// update) and MetricsReportItem (which represents a point-in-time measurement).
//
// Inspired by osquery's per-ecosystem table model (rpm_packages, deb_packages,
// programs, docker_images) where each item has typed, indexed fields rather
// than a freeform Metadata map.
type InventoryItem struct {
Ecosystem string `json:"inventory_ecosystem"` // "docker", "system", "apt", "dnf", etc.
ItemName string `json:"item_name"` // Primary identifier within ecosystem
ItemVersion string `json:"item_version"` // Installed version / digest
Description string `json:"description,omitempty"`
Arch string `json:"arch,omitempty"`
InstallTime string `json:"install_time,omitempty"` // RFC3339 or ecosystem-specific
SizeBytes int64 `json:"size_bytes,omitempty"`
Vendor string `json:"vendor,omitempty"` // Registry, repo, manufacturer
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// InventoryReport is sent by agents when reporting inventory data.
type InventoryReport struct {
CommandID string `json:"command_id"`
Timestamp time.Time `json:"timestamp"`
Ecosystem string `json:"inventory_ecosystem"`
Items []InventoryItem `json:"items"`
ScanSucceeded bool `json:"scan_succeeded"`
}

View file

@ -0,0 +1,158 @@
// Package event provides event helper functions and buffering for the RedFlag agent.
//
// TeeLogger emits an ETHOS-tagged log.Printf AND a buffered SystemEvent in a
// single call. When the internal buffer is nil (e.g. during early boot or in
// tests), it degrades to log-only mode.
//
// ETHOS Compliance:
// (1) Errors are History, Not /dev/null
// - Every TeeLogger call produces a log line (stderr/journald)
// - Every TeeLogger call buffers a SystemEvent for server delivery
// - Buffer failures are logged, never silently dropped
//
// (3) Assume Failure; Build for Resilience
// - Nil buffer degrades gracefully to log-only
// - Buffer write is best-effort; log line is the durability guarantee
package event
import (
"log"
"time"
"github.com/Fimeg/RedFlag/agent/internal/models"
"github.com/gofrs/uuid/v5"
)
// TeeLogger emits an ETHOS-tagged log.Printf AND a buffered SystemEvent
// in a single call. When the internal buffer is nil (e.g. during early boot
// or in test), it degrades to log-only mode.
type TeeLogger struct {
buffer *Buffer
agentID uuid.UUID
}
// NewTeeLogger creates a TeeLogger. Pass a nil buffer for log-only mode.
func NewTeeLogger(buffer *Buffer, agentID uuid.UUID) *TeeLogger {
return &TeeLogger{buffer: buffer, agentID: agentID}
}
// LogParams carries all structured fields for a dual-output event.
type LogParams struct {
Level string // ETHOS level tag: "INFO", "WARNING", "ERROR", "CRITICAL"
System string // ETHOS system tag: "agent"
Component string // ETHOS component tag: "migration", "crypto", "config"
EventType string // SystemEvent.EventType
EventSubtype string // SystemEvent.EventSubtype
Severity string // SystemEvent.Severity
ServerComponent string // SystemEvent.Component
Message string // Human-readable message
Metadata map[string]interface{} // Structured key-value pairs
}
// Log is the core dual-output method.
func (l *TeeLogger) Log(params LogParams) {
// Nil receiver: log-only mode (no buffer, no agent ID).
if l == nil {
log.Printf("[%s] [%s] [%s] %s", params.Level, params.System, params.Component, params.Message)
return
}
// 1. ETHOS-tagged log line
log.Printf("[%s] [%s] [%s] %s", params.Level, params.System, params.Component, params.Message)
// 2. Buffer SystemEvent (best-effort, nil-safe)
if l.buffer == nil {
return
}
var agentIDPtr *uuid.UUID
if l.agentID != uuid.Nil {
agentIDPtr = &l.agentID
}
evt := &models.SystemEvent{
ID: uuid.Must(uuid.NewV4()),
AgentID: agentIDPtr,
EventType: params.EventType,
EventSubtype: params.EventSubtype,
Severity: params.Severity,
Component: params.ServerComponent,
Message: params.Message,
Metadata: params.Metadata,
CreatedAt: time.Now().UTC(),
}
if err := l.buffer.BufferEvent(evt); err != nil {
log.Printf("[WARNING] [agent] [event] tee_buffer_failed error=%v", err)
}
}
// Info emits an INFO-level dual-output event.
func (l *TeeLogger) Info(system, component, serverComponent, message string, metadata map[string]interface{}) {
l.Log(LogParams{
Level: "INFO", System: system, Component: component,
EventType: deriveEventType(component), EventSubtype: models.SubtypeInfo,
Severity: models.SeverityInfo, ServerComponent: serverComponent,
Message: message, Metadata: metadata,
})
}
// Warning emits a WARNING-level dual-output event.
func (l *TeeLogger) Warning(system, component, serverComponent, message string, metadata map[string]interface{}) {
l.Log(LogParams{
Level: "WARNING", System: system, Component: component,
EventType: deriveEventType(component), EventSubtype: models.SubtypeWarning,
Severity: models.SeverityWarning, ServerComponent: serverComponent,
Message: message, Metadata: metadata,
})
}
// Error emits an ERROR-level dual-output event.
func (l *TeeLogger) Error(system, component, serverComponent, message string, metadata map[string]interface{}) {
l.Log(LogParams{
Level: "ERROR", System: system, Component: component,
EventType: deriveEventType(component), EventSubtype: models.SubtypeFailed,
Severity: models.SeverityError, ServerComponent: serverComponent,
Message: message, Metadata: metadata,
})
}
// Critical emits a CRITICAL-level dual-output event.
func (l *TeeLogger) Critical(system, component, serverComponent, message string, metadata map[string]interface{}) {
l.Log(LogParams{
Level: "CRITICAL", System: system, Component: component,
EventType: deriveEventType(component), EventSubtype: models.SubtypeCritical,
Severity: models.SeverityCritical, ServerComponent: serverComponent,
Message: message, Metadata: metadata,
})
}
// deriveEventType maps an ETHOS component tag to a SystemEvent.EventType.
// Callers can always override by using Log() directly with explicit params.
func deriveEventType(component string) string {
switch component {
case "migration":
return models.EventTypeAgentMigration
case "crypto":
return models.EventTypeAgentCrypto
case "config":
return models.EventTypeAgentConfig
case "docker":
return models.EventTypeAgentDocker
case "installer":
return models.EventTypeAgentInstall
case "kernel":
return models.EventTypeAgentStartup
case "acknowledgment":
return models.EventTypeAgentCheckIn
case "receipt":
return models.EventTypeAgentCheckIn
case "localapi":
return models.EventTypeAgentStartup
case "loop":
return models.EventTypeAgentStartup
default:
return models.EventTypeError
}
}

View file

@ -0,0 +1,169 @@
package event
import (
"testing"
"github.com/Fimeg/RedFlag/agent/internal/models"
"github.com/gofrs/uuid/v5"
)
func TestTeeLoggerNilBuffer(t *testing.T) {
// TeeLogger with nil buffer should not panic
logger := NewTeeLogger(nil, uuid.Nil)
// Should not panic — log-only mode
logger.Info("agent", "test", "test_component", "test message", nil)
logger.Warning("agent", "test", "test_component", "test warning", nil)
logger.Error("agent", "test", "test_component", "test error", nil)
logger.Critical("agent", "test", "test_component", "test critical", nil)
}
func TestTeeLoggerNilAgentID(t *testing.T) {
// TeeLogger with Nil agent ID should produce events with nil AgentID
buffer := NewBuffer(t.TempDir() + "/events.json")
logger := NewTeeLogger(buffer, uuid.Nil)
logger.Info("agent", "test", "test_component", "test message", nil)
events, err := buffer.GetBufferedEvents()
if err != nil {
t.Fatalf("GetBufferedEvents failed: %v", err)
}
if len(events) != 1 {
t.Fatalf("expected 1 event, got %d", len(events))
}
if events[0].AgentID != nil {
t.Errorf("expected nil AgentID, got %v", events[0].AgentID)
}
}
func TestTeeLoggerEventFields(t *testing.T) {
buffer := NewBuffer(t.TempDir() + "/events.json")
agentID := uuid.Must(uuid.NewV4())
logger := NewTeeLogger(buffer, agentID)
metadata := map[string]interface{}{
"key1": "value1",
"key2": 42,
}
logger.Warning("agent", "crypto", "pubkey", "key fetch failed", metadata)
events, err := buffer.GetBufferedEvents()
if err != nil {
t.Fatalf("GetBufferedEvents failed: %v", err)
}
if len(events) != 1 {
t.Fatalf("expected 1 event, got %d", len(events))
}
evt := events[0]
if evt.AgentID == nil || *evt.AgentID != agentID {
t.Errorf("expected AgentID %v, got %v", agentID, evt.AgentID)
}
if evt.EventType != models.EventTypeAgentCrypto {
t.Errorf("expected EventType %q, got %q", models.EventTypeAgentCrypto, evt.EventType)
}
if evt.EventSubtype != models.SubtypeWarning {
t.Errorf("expected EventSubtype %q, got %q", models.SubtypeWarning, evt.EventSubtype)
}
if evt.Severity != models.SeverityWarning {
t.Errorf("expected Severity %q, got %q", models.SeverityWarning, evt.Severity)
}
if evt.Component != "pubkey" {
t.Errorf("expected Component %q, got %q", "pubkey", evt.Component)
}
if evt.Message != "key fetch failed" {
t.Errorf("expected Message %q, got %q", "key fetch failed", evt.Message)
}
if evt.Metadata["key1"] != "value1" {
t.Errorf("expected metadata key1=value1, got %v", evt.Metadata["key1"])
}
// JSON round-trip converts int to float64
if int(evt.Metadata["key2"].(float64)) != 42 {
t.Errorf("expected metadata key2=42, got %v", evt.Metadata["key2"])
}
}
func TestTeeLoggerConvenienceMethods(t *testing.T) {
buffer := NewBuffer(t.TempDir() + "/events.json")
logger := NewTeeLogger(buffer, uuid.Must(uuid.NewV4()))
tests := []struct {
name string
call func()
expectedLevel string
expectedSubtype string
expectedSeverity string
}{
{"Info", func() { logger.Info("agent", "config", "config", "loaded", nil) }, "INFO", models.SubtypeInfo, models.SeverityInfo},
{"Warning", func() { logger.Warning("agent", "config", "config", "stale", nil) }, "WARNING", models.SubtypeWarning, models.SeverityWarning},
{"Error", func() { logger.Error("agent", "config", "config", "failed", nil) }, "ERROR", models.SubtypeFailed, models.SeverityError},
{"Critical", func() { logger.Critical("agent", "config", "config", "corrupt", nil) }, "CRITICAL", models.SubtypeCritical, models.SeverityCritical},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
tt.call()
})
}
events, err := buffer.GetBufferedEvents()
if err != nil {
t.Fatalf("GetBufferedEvents failed: %v", err)
}
if len(events) != len(tests) {
t.Fatalf("expected %d events, got %d", len(tests), len(events))
}
for i, tt := range tests {
evt := events[i]
if evt.EventSubtype != tt.expectedSubtype {
t.Errorf("%s: expected EventSubtype %q, got %q", tt.name, tt.expectedSubtype, evt.EventSubtype)
}
if evt.Severity != tt.expectedSeverity {
t.Errorf("%s: expected Severity %q, got %q", tt.name, tt.expectedSeverity, evt.Severity)
}
}
}
func TestDeriveEventType(t *testing.T) {
tests := []struct {
component string
expected string
}{
{"migration", models.EventTypeAgentMigration},
{"crypto", models.EventTypeAgentCrypto},
{"config", models.EventTypeAgentConfig},
{"docker", models.EventTypeAgentDocker},
{"installer", models.EventTypeAgentInstall},
{"unknown", models.EventTypeError},
}
for _, tt := range tests {
t.Run(tt.component, func(t *testing.T) {
got := deriveEventType(tt.component)
if got != tt.expected {
t.Errorf("deriveEventType(%q) = %q, want %q", tt.component, got, tt.expected)
}
})
}
}
func TestTeeLoggerMetadataNil(t *testing.T) {
buffer := NewBuffer(t.TempDir() + "/events.json")
logger := NewTeeLogger(buffer, uuid.Must(uuid.NewV4()))
// Should not panic with nil metadata
logger.Info("agent", "test", "test_component", "test message", nil)
events, err := buffer.GetBufferedEvents()
if err != nil {
t.Fatalf("GetBufferedEvents failed: %v", err)
}
if len(events) != 1 {
t.Fatalf("expected 1 event, got %d", len(events))
}
if events[0].Metadata != nil {
t.Errorf("expected nil metadata, got %v", events[0].Metadata)
}
}

View file

@ -76,6 +76,8 @@ func DispatchCrossPlatformCommand(
err = HandleReboot(apiClient, cfg, ackTracker, cmd.ID, cmd.Params)
case "capture_screenshot":
err = HandleCaptureScreenshot(apiClient, cfg, ackTracker, cmd.ID)
case "scan_processes":
err = HandleScanProcesses(apiClient, cfg, ackTracker, cmd.ID)
default:
return false
}

View file

@ -0,0 +1,62 @@
package handlers
import (
"fmt"
"log"
"time"
"github.com/Fimeg/RedFlag/agent/internal/acknowledgment"
"github.com/Fimeg/RedFlag/agent/internal/client"
"github.com/Fimeg/RedFlag/agent/internal/config"
"github.com/Fimeg/RedFlag/agent/internal/system"
)
// HandleScanProcesses performs a full on-demand process scan and reports
// the results to the server. Triggered by the "scan_processes" command
// (issued when a user opens the Processes tab in the dashboard).
func HandleScanProcesses(apiClient *client.Client, cfg *config.Config, ackTracker *acknowledgment.Tracker, commandID string) error {
log.Printf("[INFO] [agent] [processes] scan_started command_id=%s", commandID)
startTime := time.Now()
snapshot, err := system.GetFullProcessSnapshot()
duration := time.Since(startTime)
if err != nil {
log.Printf("[ERROR] [agent] [processes] scan_failed command_id=%s duration=%s error=%v", commandID, duration, err)
return fmt.Errorf("process scan failed: %w", err)
}
log.Printf("[INFO] [agent] [processes] scan_complete command_id=%s process_count=%d duration=%s", commandID, snapshot.ProcessCount, duration)
// Report to server
report := client.ProcessScanReport{
AgentID: cfg.AgentID,
CommandID: commandID,
Timestamp: time.Now().UTC(),
Snapshot: *snapshot,
}
if err := apiClient.ReportProcessScan(cfg.AgentID, report); err != nil {
log.Printf("[ERROR] [agent] [processes] report_failed command_id=%s error=%v", commandID, err)
return fmt.Errorf("failed to report process scan: %w", err)
}
// Audit log
logReport := client.LogReport{
CommandID: commandID,
Action: "scan_processes",
Result: "success",
ExitCode: 0,
DurationSeconds: int(duration.Seconds()),
Metadata: map[string]string{
"subsystem_label": "Processes",
"subsystem": "processes",
"process_count": fmt.Sprintf("%d", snapshot.ProcessCount),
},
}
if err := reportLogWithAck(apiClient, cfg, ackTracker, logReport); err != nil {
log.Printf("[ERROR] [agent] [processes] report_log_failed: %v", err)
}
return nil
}

View file

@ -0,0 +1,474 @@
package supplychain
import (
"context"
"crypto/ed25519"
"crypto/sha256"
"encoding/hex"
"fmt"
"io"
"log"
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"time"
"github.com/Fimeg/RedFlag/agent/internal/capability"
"github.com/Fimeg/RedFlag/agent/internal/constants"
)
const (
AgentSelfPackageType = "agent-self"
HelperSelfPackageType = "helper-self"
DesktopSelfPackageType = "desktop-self"
selfUpdateOperation = "upgrade"
agentSelfStagingPath = "/var/lib/redflag/agent/pending-upgrade.bin"
helperSelfStagingPath = "/var/lib/redflag/agent/pending-helper.bin"
maxSelfUpdateBinarySize = 500 * 1024 * 1024
)
// Exit codes mirror helper/src/main.rs so capability receipts have one taxonomy
// across helper-executed and agent-local self-update paths.
const (
policyExitOK = 0
policyExitBadToken = 10
policyExitTimeWindow = 12
policyExitKeyNotFound = 14
policyExitSignature = 15
policyExitArtifact = 16
policyExitReplay = 17
policyExitUnsupportedOp = 18
policyExitExecFailed = 19
policyExitInternal = 20
)
func (c *Consumer) processAgentSelfToken(ctx context.Context, token *capability.Token) (*PolicyResult, error) {
if token.Operation != selfUpdateOperation {
return nil, fmt.Errorf("agent-self operation %q not supported", token.Operation)
}
agentEntry, err := findClosureEntry(token, "redflag-agent")
if err != nil {
return nil, err
}
stagedAgent, err := c.stageClosureArtifact(agentEntry, agentSelfStagingPath)
if err != nil {
return nil, fmt.Errorf("stage agent-self artifact: %w", err)
}
defer os.Remove(stagedAgent)
var helperArgs []string
if helperEntry, err := findClosureEntry(token, "redflag-helper"); err == nil {
stagedHelper, err := c.stageClosureArtifact(helperEntry, helperSelfStagingPath)
if err != nil {
return nil, fmt.Errorf("stage agent-self helper artifact: %w", err)
}
defer os.Remove(stagedHelper)
helperArgs = []string{"--helper-file", stagedHelper}
}
return c.executor.Execute(ctx, token, helperArgs...)
}
func (c *Consumer) processHelperSelfToken(ctx context.Context, token *capability.Token) (*PolicyResult, error) {
if token.Operation != selfUpdateOperation {
return nil, fmt.Errorf("helper-self operation %q not supported", token.Operation)
}
entry, err := findClosureEntry(token, "redflag-helper")
if err != nil {
return nil, err
}
staged, err := c.stageClosureArtifact(entry, helperSelfStagingPath)
if err != nil {
return nil, fmt.Errorf("stage helper-self artifact: %w", err)
}
defer os.Remove(staged)
return c.executor.Execute(ctx, token)
}
func (c *Consumer) processDesktopSelfToken(ctx context.Context, token *capability.Token) (*PolicyResult, error) {
if err := validateDirectSelfToken(token); err != nil {
return selfPolicyResult(token, "denied", "desktop_self_token_invalid", policyExitSignature, 0, err), nil
}
if token.Operation != selfUpdateOperation {
return selfPolicyResult(
token,
"denied",
"operation_not_allowed",
policyExitUnsupportedOp,
0,
fmt.Errorf("operation=%s (desktop-self requires upgrade)", token.Operation),
), nil
}
// Bound the in-process staging + install to match the executor timeout.
// The agent-self and helper-self paths go through executor.Execute which
// carries its own context timeout; desktop-self runs in-process and needs
// an equivalent guard against hung downloads or stalled I/O.
ctx, cancel := context.WithTimeout(ctx, executorTimeout)
defer cancel()
entry, err := findClosureEntry(token, "redflag-desktop")
if err != nil {
return selfPolicyResult(token, "denied", "desktop_self_no_entry", policyExitBadToken, 0, err), nil
}
// Check context before starting potentially long staging operation.
if err := ctx.Err(); err != nil {
return selfPolicyResult(token, "failed", "desktop_self_cancelled", policyExitExecFailed, 0, err), nil
}
stagingPath := filepath.Join(constants.GetAgentStateDir(), "desktop-self-upgrade.bin")
staged, err := c.stageClosureArtifact(entry, stagingPath)
if err != nil {
return selfPolicyResult(token, "failed", "desktop_self_stage_failed", policyExitArtifact, 0, err), nil
}
defer os.Remove(staged)
if err := replayCheckAndRecordAgent(token.TokenID); err != nil {
return selfPolicyResult(token, "denied", "token_already_consumed", policyExitReplay, 1, err), nil
}
target, err := installDesktopBinary(staged)
if err != nil {
return selfPolicyResult(token, "failed", "desktop_self_install_failed", policyExitExecFailed, 1, err), nil
}
if err := signalDesktopRestart(target); err != nil {
log.Printf("[WARNING] [agent] [supplychain] desktop_restart_signal_failed token_id=%s error=%v", token.TokenID, err)
}
log.Printf("[SECURITY] [agent] [supplychain] desktop_self_upgraded token_id=%s path=%s", token.TokenID, target)
return selfPolicyResult(token, "executed", "desktop_self_upgraded", policyExitOK, 1, nil), nil
}
func findClosureEntry(token *capability.Token, name string) (*capability.ClosureEntry, error) {
for i := range token.Closure {
if token.Closure[i].Name == name {
return &token.Closure[i], nil
}
}
return nil, fmt.Errorf("closure missing %s entry", name)
}
func (c *Consumer) stageClosureArtifact(entry *capability.ClosureEntry, dstPath string) (string, error) {
if entry.ArtifactPath == "" {
return "", fmt.Errorf("closure entry %s missing artifact_path", entry.Name)
}
if strings.TrimSpace(entry.SHA256) == "" {
return "", fmt.Errorf("closure entry %s missing sha256", entry.Name)
}
dstDir := filepath.Dir(dstPath)
if err := os.MkdirAll(dstDir, 0o700); err != nil {
return "", fmt.Errorf("create staging dir: %w", err)
}
tmp, err := os.CreateTemp(dstDir, filepath.Base(dstPath)+".")
if err != nil {
return "", fmt.Errorf("create staging temp: %w", err)
}
tmpPath := tmp.Name()
if err := tmp.Close(); err != nil {
os.Remove(tmpPath)
return "", fmt.Errorf("close staging temp: %w", err)
}
cleanup := true
defer func() {
if cleanup {
os.Remove(tmpPath)
}
}()
if err := c.fetchArtifactToFile(entry.ArtifactPath, tmpPath); err != nil {
return "", err
}
actual, err := computeFileSHA256Hex(tmpPath)
if err != nil {
return "", err
}
if !strings.EqualFold(actual, entry.SHA256) {
return "", fmt.Errorf("hash mismatch for %s: expected=%s actual=%s", entry.Name, strings.ToLower(entry.SHA256), actual)
}
if err := os.Chmod(tmpPath, 0o644); err != nil {
return "", fmt.Errorf("chmod staged artifact: %w", err)
}
if err := os.Rename(tmpPath, dstPath); err != nil {
return "", fmt.Errorf("commit staged artifact: %w", err)
}
cleanup = false
return dstPath, nil
}
func (c *Consumer) fetchArtifactToFile(ref, dstPath string) error {
if info, err := os.Stat(ref); err == nil {
if info.IsDir() {
return fmt.Errorf("artifact path is a directory: %s", ref)
}
return copyRegularFile(ref, dstPath)
} else if !os.IsNotExist(err) {
// Stat failed for a reason other than "not found" (e.g. permission denied).
return fmt.Errorf("stat artifact %s: %w", ref, err)
}
// Local absolute path that doesn't exist — don't fall through to the
// downloader; the file was expected on disk and is missing.
if strings.HasPrefix(ref, "/") {
return fmt.Errorf("artifact not staged at %s", ref)
}
if isDownloadRef(ref) {
if c.downloader == nil {
return fmt.Errorf("artifact downloader unavailable for %s", ref)
}
if _, err := c.downloader.DownloadAuthenticatedToFile(ref, dstPath, maxSelfUpdateBinarySize); err != nil {
return fmt.Errorf("download artifact %s: %w", ref, err)
}
return nil
}
return copyRegularFile(ref, dstPath)
}
func isDownloadRef(ref string) bool {
return strings.HasPrefix(ref, "http://") || strings.HasPrefix(ref, "https://")
}
func validateDirectSelfToken(token *capability.Token) error {
if token.Version != capability.Version {
return fmt.Errorf("unsupported token version=%d supported=%d", token.Version, capability.Version)
}
now := time.Now().UTC().Unix()
if now < token.NotBefore {
return fmt.Errorf("token not yet valid: now=%d not_before=%d", now, token.NotBefore)
}
if now > token.ExpiresAt {
return fmt.Errorf("token expired: now=%d expires_at=%d", now, token.ExpiresAt)
}
pub, err := loadCapabilityPublicKey()
if err != nil {
return err
}
keyID := capability.KeyIDFor(ed25519.PublicKey(pub))
if token.KeyID != keyID {
return fmt.Errorf("key_id mismatch: token=%s local=%s", token.KeyID, keyID)
}
if err := token.Verify(ed25519.PublicKey(pub)); err != nil {
return err
}
return nil
}
func loadCapabilityPublicKey() ([]byte, error) {
keyPath := constants.GetServerPublicKeyPath()
data, err := os.ReadFile(keyPath)
if err != nil {
return nil, fmt.Errorf("public key not found at %s: %w", keyPath, err)
}
if len(data) == ed25519.PublicKeySize {
return data, nil
}
trimmed := strings.TrimSpace(string(data))
decoded, decErr := hex.DecodeString(trimmed)
if decErr == nil && len(decoded) == ed25519.PublicKeySize {
return decoded, nil
}
return nil, fmt.Errorf("invalid public key size at %s: raw=%d decoded=%d", keyPath, len(data), len(decoded))
}
func replayCheckAndRecordAgent(tokenID string) error {
if _, err := safeTokenFilename(tokenID); err != nil {
return err
}
if strings.ContainsAny(tokenID, "\r\n") {
return fmt.Errorf("unsafe token_id: contains newline")
}
statePath := filepath.Join(constants.GetAgentStateDir(), "consumed-self-tokens")
if contents, err := os.ReadFile(statePath); err == nil {
for _, line := range strings.Split(string(contents), "\n") {
if strings.TrimSpace(line) == tokenID {
return fmt.Errorf("token_id=%s", tokenID)
}
}
} else if !os.IsNotExist(err) {
return fmt.Errorf("read replay state: %w", err)
}
if err := os.MkdirAll(filepath.Dir(statePath), 0o700); err != nil {
return fmt.Errorf("create replay state dir: %w", err)
}
f, err := os.OpenFile(statePath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
if err != nil {
return fmt.Errorf("open replay state: %w", err)
}
defer f.Close()
if _, err := fmt.Fprintln(f, tokenID); err != nil {
return fmt.Errorf("write replay state: %w", err)
}
return nil
}
func installDesktopBinary(staged string) (string, error) {
target, err := desktopBinaryPath()
if err != nil {
return "", err
}
targetDir := filepath.Dir(target)
if info, err := os.Stat(targetDir); err != nil {
return "", fmt.Errorf("desktop install dir unavailable: %w", err)
} else if !info.IsDir() {
return "", fmt.Errorf("desktop install parent is not a directory: %s", targetDir)
}
if info, err := os.Stat(target); err == nil {
if info.IsDir() {
return "", fmt.Errorf("desktop target is a directory: %s", target)
}
// One generation back: overwrite stale .bak with current before replacing.
os.Remove(target + ".bak")
if err := copyRegularFile(target, target+".bak"); err != nil {
return "", fmt.Errorf("backup desktop binary: %w", err)
}
} else if !os.IsNotExist(err) {
return "", fmt.Errorf("stat desktop binary: %w", err)
}
if err := atomicReplaceFile(staged, target, 0o755); err != nil {
return "", err
}
return target, nil
}
func desktopBinaryPath() (string, error) {
if p := os.Getenv("REDFLAG_DESKTOP_BINARY"); p != "" {
return p, nil
}
execPath, err := os.Executable()
if err != nil {
return "", fmt.Errorf("determine agent executable path: %w", err)
}
name := "redflag-desktop"
if runtime.GOOS == "windows" {
name += ".exe"
}
return filepath.Join(filepath.Dir(execPath), name), nil
}
func signalDesktopRestart(target string) error {
if runtime.GOOS != "linux" {
return nil
}
name := filepath.Base(target)
cmd := exec.Command("pkill", "-x", name)
out, err := cmd.CombinedOutput()
if err == nil {
log.Printf("[INFO] [agent] [supplychain] desktop_restart_signal_sent binary=%s", name)
return nil
}
if exitErr, ok := err.(*exec.ExitError); ok && exitErr.ExitCode() == 1 {
log.Printf("[INFO] [agent] [supplychain] desktop_restart_signal_skipped binary=%s reason=not_running", name)
return nil
}
return fmt.Errorf("pkill -x %s failed: %w output=%s", name, err, strings.TrimSpace(string(out)))
}
func atomicReplaceFile(staged, target string, mode os.FileMode) error {
tmp, err := os.CreateTemp(filepath.Dir(target), filepath.Base(target)+".new.")
if err != nil {
return fmt.Errorf("create install temp: %w", err)
}
tmpPath := tmp.Name()
if err := tmp.Close(); err != nil {
os.Remove(tmpPath)
return fmt.Errorf("close install temp: %w", err)
}
cleanup := true
defer func() {
if cleanup {
os.Remove(tmpPath)
}
}()
if err := copyRegularFile(staged, tmpPath); err != nil {
return err
}
if err := os.Chmod(tmpPath, mode); err != nil {
return fmt.Errorf("chmod install temp: %w", err)
}
if runtime.GOOS == "windows" {
_ = os.Remove(target)
}
if err := os.Rename(tmpPath, target); err != nil {
return fmt.Errorf("replace binary: %w", err)
}
cleanup = false
return nil
}
func copyRegularFile(src, dst string) error {
in, err := os.Open(src)
if err != nil {
return fmt.Errorf("open source: %w", err)
}
defer in.Close()
if info, err := in.Stat(); err != nil {
return fmt.Errorf("stat source: %w", err)
} else if info.IsDir() {
return fmt.Errorf("source is a directory: %s", src)
}
out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
if err != nil {
return fmt.Errorf("open destination: %w", err)
}
if _, err := io.Copy(out, in); err != nil {
out.Close()
return fmt.Errorf("copy file: %w", err)
}
if err := out.Close(); err != nil {
return fmt.Errorf("close destination: %w", err)
}
return nil
}
func computeFileSHA256Hex(path string) (string, error) {
f, err := os.Open(path)
if err != nil {
return "", fmt.Errorf("open for hash: %w", err)
}
defer f.Close()
h := sha256.New()
if _, err := io.Copy(h, f); err != nil {
return "", fmt.Errorf("hash file: %w", err)
}
return hex.EncodeToString(h.Sum(nil)), nil
}
func selfPolicyResult(token *capability.Token, decision, reason string, exitCode, verifiedArtifacts int, resultErr error) *PolicyResult {
var errText string
if resultErr != nil {
errText = resultErr.Error()
}
return &PolicyResult{
TokenID: token.TokenID,
AgentID: token.AgentID,
PackageType: token.PackageType,
Operation: token.Operation,
Decision: decision,
Reason: reason,
Executed: decision == "executed",
VerifiedArtifacts: verifiedArtifacts,
ExitCode: exitCode,
Error: errText,
Timestamp: time.Now().UTC().Unix(),
}
}

View file

@ -0,0 +1,128 @@
package system
import "time"
// FullProcess represents a complete process snapshot with osquery-level detail.
// Fields mirror the osquery `processes` table schema plus related data for
// drill-down (open files, sockets, env, memory map, namespaces, pipes).
type FullProcess struct {
// Core identification
PID int `json:"pid"`
Name string `json:"name"`
Path string `json:"path,omitempty"` // /proc/[pid]/exe symlink target
Cmdline string `json:"cmdline"` // /proc/[pid]/cmdline (NUL→space)
Cwd string `json:"cwd,omitempty"` // /proc/[pid]/cwd symlink target
// State and identity
State string `json:"state"` // R/S/D/Z/T from /proc/[pid]/stat
UID uint32 `json:"uid"`
GID uint32 `json:"gid"`
EUID uint32 `json:"euid"`
EGID uint32 `json:"egid"`
User string `json:"user"` // resolved from /etc/passwd
Group string `json:"group"` // resolved from /etc/group
// Terminal
TTY int `json:"tty"`
TTYName string `json:"tty_name,omitempty"`
// Resource usage
CPUSecondsUser float64 `json:"cpu_seconds_user"` // utime / clock_ticks
CPUSecondsSystem float64 `json:"cpu_seconds_system"` // stime / clock_ticks
CPUPercent float64 `json:"cpu_percent"` // vs total CPU ticks
RSSBytes uint64 `json:"rss_bytes"` // VmRSS from /proc/[pid]/status
VMSBytes uint64 `json:"vms_bytes"` // VmSize from /proc/[pid]/status
MemPercent float64 `json:"mem_percent"`
Threads int `json:"threads"` // Threads field from /proc/[pid]/status
// Process metadata
Nice int `json:"nice"`
StartTimeSeconds uint64 `json:"start_time_seconds"` // starttime from /proc/[pid]/stat
ParentPID int `json:"parent_pid"`
ProcessGroupID int `json:"process_group_id"`
// On-disk check
OnDisk int `json:"on_disk"` // 1=yes, 0=no, -1=unknown (matches osquery)
// Elevation
ElevationStatus string `json:"elevation_status,omitempty"` // "elevated" if uid!=euid
// Disk I/O (from /proc/[pid]/io, may be empty for other users' processes)
DiskBytesRead uint64 `json:"disk_bytes_read,omitempty"`
DiskBytesWritten uint64 `json:"disk_bytes_written,omitempty"`
// Related data — populated only on drill-down (GetProcessDetail), not on list scan
OpenFiles []ProcessOpenFile `json:"open_files,omitempty"`
OpenSockets []ProcessOpenSocket `json:"open_sockets,omitempty"`
OpenPipes []ProcessOpenPipe `json:"open_pipes,omitempty"`
EnvironmentVars []string `json:"environment_vars,omitempty"` // keys only
MemoryMap []ProcessMemoryMap `json:"memory_map,omitempty"`
Namespaces []ProcessNamespace `json:"namespaces,omitempty"`
ListeningPorts []ProcessListeningPort `json:"listening_ports,omitempty"`
}
// Related data types — stored as JSONB per relation in the database.
type ProcessOpenFile struct {
FD int `json:"fd"`
Path string `json:"path"`
Type string `json:"type,omitempty"` // file, socket, pipe, etc.
}
type ProcessOpenSocket struct {
FD int `json:"fd"`
Family string `json:"family"` // IPv4, IPv6, UNIX
Protocol string `json:"protocol"` // TCP, UDP
LocalAddr string `json:"local_addr"`
LocalPort int `json:"local_port"`
RemoteAddr string `json:"remote_addr"`
RemotePort int `json:"remote_port"`
State string `json:"state"`
Inode uint64 `json:"inode"`
Path string `json:"path,omitempty"` // for UNIX sockets
}
type ProcessOpenPipe struct {
FD int `json:"fd"`
Inode uint64 `json:"inode"`
Mode string `json:"mode,omitempty"` // r/w
Type string `json:"type,omitempty"` // named vs anonymous
}
type ProcessMemoryMap struct {
Start uint64 `json:"start"`
End uint64 `json:"end"`
Permissions string `json:"permissions"` // r/w/x/p
Offset uint64 `json:"offset"`
Device string `json:"device"`
Inode uint64 `json:"inode"`
Path string `json:"path,omitempty"`
}
type ProcessNamespace struct {
Type string `json:"type"` // cgroup, ipc, mnt, net, pid, user, uts
Inode string `json:"inode"` // string because kernel shows as "[4026531835]"
}
type ProcessListeningPort struct {
Protocol string `json:"protocol"` // TCP, UDP
LocalAddr string `json:"local_addr"`
LocalPort int `json:"local_port"`
FD int `json:"fd,omitempty"`
Socket uint64 `json:"socket,omitempty"`
}
// FullProcessSnapshot is the complete result of an on-demand process scan.
type FullProcessSnapshot struct {
Processes []FullProcess `json:"processes"`
ProcessCount int `json:"process_count"`
ScannedAt time.Time `json:"scanned_at"`
DurationMs int64 `json:"duration_ms"`
}
// GetFullProcessSnapshot reads /proc to build a complete process inventory.
// Platform-specific implementations live in process_detail_linux.go and
// process_detail_other.go (stub returning "not supported").
func GetFullProcessSnapshot() (*FullProcessSnapshot, error) {
return getFullProcessSnapshot()
}

View file

@ -0,0 +1,706 @@
//go:build linux
// +build linux
package system
import (
"bytes"
"fmt"
"log"
"os"
"strconv"
"strings"
"time"
)
// getFullProcessSnapshot reads /proc to build a complete process inventory.
// No subprocess spawns — pure /proc reads following the pattern of process_linux.go.
func getFullProcessSnapshot() (*FullProcessSnapshot, error) {
start := time.Now()
totalCPU, err := readTotalCPUTicks()
if err != nil {
return nil, fmt.Errorf("read total cpu ticks: %w", err)
}
if totalCPU == 0 {
return nil, fmt.Errorf("total CPU ticks is zero")
}
memTotal, err := readMemTotal()
if err != nil {
return nil, fmt.Errorf("read mem total: %w", err)
}
if memTotal == 0 {
return nil, fmt.Errorf("mem total is zero")
}
entries, err := os.ReadDir("/proc")
if err != nil {
return nil, fmt.Errorf("read /proc: %w", err)
}
// Pre-load /etc/passwd and /etc/group for uid→name resolution.
passwd := readPasswd()
group := readGroup()
var procs []FullProcess
for _, entry := range entries {
if !entry.IsDir() {
continue
}
pid, err := strconv.Atoi(entry.Name())
if err != nil {
continue
}
proc, err := readFullProc(pid, totalCPU, memTotal, passwd, group)
if err != nil {
continue // vanished or permission denied
}
procs = append(procs, *proc)
}
return &FullProcessSnapshot{
Processes: procs,
ProcessCount: len(procs),
ScannedAt: start.UTC(),
DurationMs: time.Since(start).Milliseconds(),
}, nil
}
// readFullProc reads all /proc/[pid]/* files for a single process.
func readFullProc(pid int, totalCPU, memTotal uint64, passwd, group map[uint32]string) (*FullProcess, error) {
// /proc/[pid]/stat — core fields
statData, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
if err != nil {
return nil, err
}
proc := &FullProcess{PID: pid, OnDisk: -1}
// Parse comm: between first '(' and last ')'
firstParen := bytes.IndexByte(statData, '(')
lastParen := bytes.LastIndexByte(statData, ')')
if firstParen < 0 || lastParen <= firstParen {
return nil, fmt.Errorf("malformed stat")
}
proc.Name = string(statData[firstParen+1 : lastParen])
// Fields after ')'
afterName := bytes.TrimSpace(statData[lastParen+1:])
fields := strings.Fields(string(afterName))
// Need at least 20 fields (state through nice)
if len(fields) < 20 {
return nil, fmt.Errorf("not enough fields in stat")
}
// Field 0 (index 0) = state
proc.State = fields[0]
// fields[0]=state, fields[3]=ppid, fields[4]=pgrp, fields[6]=tty
proc.ParentPID = atoi(fields[3])
proc.ProcessGroupID = atoi(fields[4])
proc.TTY = atoi(fields[6])
// Field 13 = utime, Field 14 = stime, Field 17 = starttime, Field 18 = nice
utime := atouint64(fields[13])
stime := atouint64(fields[14])
proc.StartTimeSeconds = atouint64(fields[19]) // starttime is field 21 (0-indexed after name: 19)
proc.Nice = atoi(fields[18])
// CPU seconds
if totalCPU > 0 {
proc.CPUSecondsUser = float64(utime) / float64(totalCPU) * 100
proc.CPUSecondsSystem = float64(stime) / float64(totalCPU) * 100
proc.CPUPercent = (float64(utime+stime) / float64(totalCPU)) * 100.0
}
// /proc/[pid]/status — memory, threads, uid/gid
statusData, err := os.ReadFile(fmt.Sprintf("/proc/%d/status", pid))
if err == nil {
for _, line := range bytes.Split(statusData, []byte("\n")) {
switch {
case bytes.HasPrefix(line, []byte("Uid:")):
// Uid:\t<real>\t<effective>\t<saved>\t<fs>
parts := strings.Fields(string(line))
if len(parts) >= 3 {
proc.UID = uint32(atoi(parts[1]))
proc.EUID = uint32(atoi(parts[2]))
}
case bytes.HasPrefix(line, []byte("Gid:")):
parts := strings.Fields(string(line))
if len(parts) >= 3 {
proc.GID = uint32(atoi(parts[1]))
proc.EGID = uint32(atoi(parts[2]))
}
case bytes.HasPrefix(line, []byte("VmRSS:")):
parts := strings.Fields(string(line))
if len(parts) >= 2 {
proc.RSSBytes = atouint64(parts[1]) * 1024 // KB to bytes
}
case bytes.HasPrefix(line, []byte("VmSize:")):
parts := strings.Fields(string(line))
if len(parts) >= 2 {
proc.VMSBytes = atouint64(parts[1]) * 1024
}
case bytes.HasPrefix(line, []byte("Threads:")):
parts := strings.Fields(string(line))
if len(parts) >= 2 {
proc.Threads = atoi(parts[1])
}
}
}
}
// Memory percent
if memTotal > 0 {
proc.MemPercent = (float64(proc.RSSBytes) / float64(memTotal*1024)) * 100.0
}
// Resolved user/group names
proc.User = passwd[proc.UID]
proc.Group = group[proc.GID]
// Elevation
if proc.UID != proc.EUID || proc.GID != proc.EGID {
proc.ElevationStatus = "elevated"
}
// /proc/[pid]/exe — binary path
exePath := fmt.Sprintf("/proc/%d/exe", pid)
if link, err := os.Readlink(exePath); err == nil {
proc.Path = link
// On-disk check: does the binary exist at the resolved path?
if _, err := os.Stat(link); err == nil {
proc.OnDisk = 1
} else {
proc.OnDisk = 0
}
}
// /proc/[pid]/cmdline — full command line
cmdlineData, err := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid))
if err == nil {
// NUL-delimited, join with spaces, trim trailing NUL
proc.Cmdline = strings.ReplaceAll(strings.TrimRight(string(cmdlineData), "\x00"), "\x00", " ")
}
// /proc/[pid]/cwd — working directory
cwdPath := fmt.Sprintf("/proc/%d/cwd", pid)
if link, err := os.Readlink(cwdPath); err == nil {
proc.Cwd = link
}
// TTY name resolution
if proc.TTY > 0 {
proc.TTYName = resolveTTY(proc.TTY)
}
// /proc/[pid]/io — disk I/O (may fail for other users' processes)
ioData, err := os.ReadFile(fmt.Sprintf("/proc/%d/io", pid))
if err == nil {
for _, line := range bytes.Split(ioData, []byte("\n")) {
switch {
case bytes.HasPrefix(line, []byte("read_bytes:")):
proc.DiskBytesRead = atouint64(strings.TrimSpace(strings.TrimPrefix(string(line), "read_bytes:")))
case bytes.HasPrefix(line, []byte("write_bytes:")):
proc.DiskBytesWritten = atouint64(strings.TrimSpace(strings.TrimPrefix(string(line), "write_bytes:")))
}
}
}
return proc, nil
}
// readPasswd parses /etc/passwd into a uid→username map.
func readPasswd() map[uint32]string {
m := make(map[uint32]string)
data, err := os.ReadFile("/etc/passwd")
if err != nil {
return m
}
for _, line := range bytes.Split(data, []byte("\n")) {
parts := strings.SplitN(string(line), ":", 7)
if len(parts) >= 3 {
if uid, err := strconv.ParseUint(parts[2], 10, 32); err == nil {
m[uint32(uid)] = parts[0]
}
}
}
return m
}
// readGroup parses /etc/group into a gid→groupname map.
func readGroup() map[uint32]string {
m := make(map[uint32]string)
data, err := os.ReadFile("/etc/group")
if err != nil {
return m
}
for _, line := range bytes.Split(data, []byte("\n")) {
parts := strings.SplitN(string(line), ":", 4)
if len(parts) >= 3 {
if gid, err := strconv.ParseUint(parts[2], 10, 32); err == nil {
m[uint32(gid)] = parts[0]
}
}
}
return m
}
// resolveTTY maps a TTY device number (from /proc/[pid]/stat field 7) to /dev/pts/N or /dev/ttyN.
func resolveTTY(tty int) string {
major := (tty >> 8) & 0xfff
minor := (tty & 0xff) | ((tty >> 12) & 0xfff00)
// /dev/pts/N — minor is the pts number
if major == 136 || major == 188 {
return fmt.Sprintf("pts/%d", minor)
}
// /dev/ttyN — major 4
if major == 4 {
if minor == 0 {
return "tty0"
}
return fmt.Sprintf("tty%d", minor)
}
// /dev/tty (controlling terminal)
if major == 5 && minor == 0 {
return "tty"
}
return fmt.Sprintf("%d:%d", major, minor)
}
// GetProcessDetail reads the related data (open files, sockets, env, etc.) for a
// single process. Called on drill-down, not on the initial list scan.
func GetProcessDetail(pid int) (*FullProcess, error) {
totalCPU, err := readTotalCPUTicks()
if err != nil || totalCPU == 0 {
return nil, fmt.Errorf("read total cpu ticks: %w", err)
}
memTotal, err := readMemTotal()
if err != nil || memTotal == 0 {
return nil, fmt.Errorf("read mem total: %w", err)
}
passwd := readPasswd()
group := readGroup()
proc, err := readFullProc(pid, totalCPU, memTotal, passwd, group)
if err != nil {
return nil, err
}
pidStr := strconv.Itoa(pid)
// Open files from /proc/[pid]/fd/
proc.OpenFiles = readOpenFiles(pidStr)
// Sockets from /proc/[pid]/net/tcp, /proc/[pid]/net/tcp6, /proc/[pid]/net/unix
proc.OpenSockets = readProcSockets(pidStr)
// Pipes from /proc/[pid]/fd/ (symlink targets starting with "pipe:")
proc.OpenPipes = readOpenPipes(pidStr)
// Environment variable names from /proc/[pid]/environ (keys only, no values)
proc.EnvironmentVars = readEnvKeys(pidStr)
// Memory map from /proc/[pid]/maps
proc.MemoryMap = readMemoryMap(pidStr)
// Namespaces from /proc/[pid]/ns/
proc.Namespaces = readNamespaces(pidStr)
// Listening ports from /proc/net/tcp (state 0A = LISTEN)
proc.ListeningPorts = readListeningPorts(pid)
return proc, nil
}
// readOpenFiles reads /proc/[pid]/fd/ and follows symlinks.
func readOpenFiles(pidStr string) []ProcessOpenFile {
dir := fmt.Sprintf("/proc/%s/fd", pidStr)
entries, err := os.ReadDir(dir)
if err != nil {
return nil
}
var files []ProcessOpenFile
for _, entry := range entries {
fd, err := strconv.Atoi(entry.Name())
if err != nil {
continue
}
link := fmt.Sprintf("%s/%d", dir, fd)
target, err := os.Readlink(link)
if err != nil {
continue
}
ftype := "file"
if strings.HasPrefix(target, "socket:") {
ftype = "socket"
} else if strings.HasPrefix(target, "pipe:") {
ftype = "pipe"
} else if strings.HasPrefix(target, "anon_inode:") {
ftype = "anon_inode"
}
files = append(files, ProcessOpenFile{
FD: fd,
Path: target,
Type: ftype,
})
}
return files
}
// readProcSockets reads /proc/[pid]/net/tcp, tcp6, and unix for socket info.
func readProcSockets(pidStr string) []ProcessOpenSocket {
var sockets []ProcessOpenSocket
// TCP sockets
for _, proto := range []struct {
file string
family string
prot string
}{
{fmt.Sprintf("/proc/%s/net/tcp", pidStr), "IPv4", "TCP"},
{fmt.Sprintf("/proc/%s/net/tcp6", pidStr), "IPv6", "TCP"},
} {
data, err := os.ReadFile(proto.file)
if err != nil {
continue
}
lines := strings.Split(string(data), "\n")
for i, line := range lines {
if i == 0 || strings.TrimSpace(line) == "" {
continue // skip header
}
fields := strings.Fields(line)
if len(fields) < 10 {
continue
}
localAddr, localPort := parseHexAddr(fields[1])
remoteAddr, remotePort := parseHexAddr(fields[2])
state := tcpState(fields[3])
inode := atouint64(fields[9])
sockets = append(sockets, ProcessOpenSocket{
Family: proto.family,
Protocol: proto.prot,
LocalAddr: localAddr,
LocalPort: localPort,
RemoteAddr: remoteAddr,
RemotePort: remotePort,
State: state,
Inode: inode,
})
}
}
// UNIX sockets
unixFile := fmt.Sprintf("/proc/%s/net/unix", pidStr)
data, err := os.ReadFile(unixFile)
if err == nil {
lines := strings.Split(string(data), "\n")
for i, line := range lines {
if i == 0 || strings.TrimSpace(line) == "" {
continue
}
fields := strings.Fields(line)
if len(fields) < 7 {
continue
}
inode := atouint64(fields[6])
path := ""
if len(fields) > 7 {
path = fields[7]
}
sockets = append(sockets, ProcessOpenSocket{
Family: "UNIX",
Protocol: "UNIX",
Inode: inode,
Path: path,
})
}
}
return sockets
}
// readOpenPipes identifies pipe file descriptors from /proc/[pid]/fd/.
func readOpenPipes(pidStr string) []ProcessOpenPipe {
dir := fmt.Sprintf("/proc/%s/fd", pidStr)
entries, err := os.ReadDir(dir)
if err != nil {
return nil
}
var pipes []ProcessOpenPipe
for _, entry := range entries {
fd, err := strconv.Atoi(entry.Name())
if err != nil {
continue
}
link := fmt.Sprintf("%s/%d", dir, fd)
target, err := os.Readlink(link)
if err != nil {
continue
}
if !strings.HasPrefix(target, "pipe:") {
continue
}
// Extract inode from "pipe:[12345]"
inodeStr := strings.TrimPrefix(target, "pipe:[")
inodeStr = strings.TrimSuffix(inodeStr, "]")
inode := atouint64(inodeStr)
pipes = append(pipes, ProcessOpenPipe{
FD: fd,
Inode: inode,
})
}
return pipes
}
// readEnvKeys reads /proc/[pid]/environ and returns only the key names.
// Values are never transmitted (security: env vars may contain secrets).
func readEnvKeys(pidStr string) []string {
data, err := os.ReadFile(fmt.Sprintf("/proc/%s/environ", pidStr))
if err != nil {
return nil
}
var keys []string
for _, part := range bytes.Split(data, []byte{0}) {
s := string(part)
if idx := strings.IndexByte(s, '='); idx > 0 {
keys = append(keys, s[:idx])
}
}
return keys
}
// readMemoryMap reads /proc/[pid]/maps for memory-mapped regions.
func readMemoryMap(pidStr string) []ProcessMemoryMap {
data, err := os.ReadFile(fmt.Sprintf("/proc/%s/maps", pidStr))
if err != nil {
return nil
}
var regions []ProcessMemoryMap
for _, line := range strings.Split(string(data), "\n") {
fields := strings.Fields(line)
if len(fields) < 5 {
continue
}
// Address range: start-end
addrs := strings.SplitN(fields[0], "-", 2)
if len(addrs) != 2 {
continue
}
start, _ := strconv.ParseUint(addrs[0], 16, 64)
end, _ := strconv.ParseUint(addrs[1], 16, 64)
offset, _ := strconv.ParseUint(fields[2], 16, 64)
inode := atouint64(fields[4])
path := ""
if len(fields) > 5 {
path = strings.Join(fields[5:], " ")
}
regions = append(regions, ProcessMemoryMap{
Start: start,
End: end,
Permissions: fields[1],
Offset: offset,
Device: fields[3],
Inode: inode,
Path: path,
})
}
return regions
}
// readNamespaces reads /proc/[pid]/ns/ symlinks for namespace info.
func readNamespaces(pidStr string) []ProcessNamespace {
dir := fmt.Sprintf("/proc/%s/ns", pidStr)
entries, err := os.ReadDir(dir)
if err != nil {
return nil
}
var ns []ProcessNamespace
for _, entry := range entries {
link := fmt.Sprintf("%s/%s", dir, entry.Name())
target, err := os.Readlink(link)
if err != nil {
continue
}
// Target format: "type:[inode]" e.g. "cgroup:[4026531835]"
inode := ""
if idx := strings.Index(target, "["); idx >= 0 {
inode = strings.TrimSuffix(target[idx+1:], "]")
}
ns = append(ns, ProcessNamespace{
Type: entry.Name(),
Inode: inode,
})
}
return ns
}
// readListeningPorts finds TCP listening ports that belong to a specific process
// by correlating socket inodes from /proc/[pid]/fd/ with /proc/net/tcp entries.
func readListeningPorts(pid int) []ProcessListeningPort {
pidStr := strconv.Itoa(pid)
// Collect socket inodes from /proc/[pid]/fd/
socketInodes := make(map[uint64]bool)
dir := fmt.Sprintf("/proc/%s/fd", pidStr)
entries, err := os.ReadDir(dir)
if err != nil {
return nil
}
for _, entry := range entries {
fd, ferr := strconv.Atoi(entry.Name())
if ferr != nil {
continue
}
target, lerr := os.Readlink(fmt.Sprintf("%s/%d", dir, fd))
if lerr != nil {
continue
}
if strings.HasPrefix(target, "socket:[") {
inodeStr := strings.TrimPrefix(target, "socket:[")
inodeStr = strings.TrimSuffix(inodeStr, "]")
inode := atouint64(inodeStr)
if inode > 0 {
socketInodes[inode] = true
}
}
}
if len(socketInodes) == 0 {
return nil
}
// Read system-wide /proc/net/tcp and /proc/net/tcp6, keep only LISTEN
// entries whose inode matches one of this process's sockets.
var ports []ProcessListeningPort
for _, file := range []struct {
path string
}{
{"/proc/net/tcp"},
{"/proc/net/tcp6"},
} {
data, err := os.ReadFile(file.path)
if err != nil {
continue
}
lines := strings.Split(string(data), "\n")
for i, line := range lines {
if i == 0 || strings.TrimSpace(line) == "" {
continue
}
fields := strings.Fields(line)
if len(fields) < 10 {
continue
}
if fields[3] != "0A" { // LISTEN only
continue
}
inode := atouint64(fields[9])
if !socketInodes[inode] {
continue // not this process
}
localAddr, localPort := parseHexAddr(fields[1])
ports = append(ports, ProcessListeningPort{
Protocol: "TCP",
LocalAddr: localAddr,
LocalPort: localPort,
Socket: inode,
})
}
}
return ports
}
// parseHexAddr parses "HEXADDR:HEXPORT" from /proc/net/tcp format.
// e.g. "0100007F:0050" → "127.0.0.1", 80
func parseHexAddr(hex string) (string, int) {
parts := strings.SplitN(hex, ":", 2)
if len(parts) != 2 {
return "", 0
}
port, _ := strconv.ParseUint(parts[1], 16, 16)
// Parse IP (little-endian hex for IPv4)
addrHex := parts[0]
if len(addrHex) == 8 {
// IPv4: 4 bytes in little-endian
b0, _ := strconv.ParseUint(addrHex[6:8], 16, 8)
b1, _ := strconv.ParseUint(addrHex[4:6], 16, 8)
b2, _ := strconv.ParseUint(addrHex[2:4], 16, 8)
b3, _ := strconv.ParseUint(addrHex[0:2], 16, 8)
return fmt.Sprintf("%d.%d.%d.%d", b0, b1, b2, b3), int(port)
}
if len(addrHex) == 32 {
// IPv6: 16 bytes in little-endian groups of 4 hex chars
var b [16]uint64
for i := 0; i < 16; i++ {
start := (15 - i) * 2
b[i], _ = strconv.ParseUint(addrHex[start:start+2], 16, 8)
}
return fmt.Sprintf("%x%x:%x%x:%x%x:%x%x:%x%x:%x%x:%x%x:%x%x",
b[0], b[1], b[2], b[3], b[4], b[5], b[6], b[7],
b[8], b[9], b[10], b[11], b[12], b[13], b[14], b[15]), int(port)
}
return addrHex, int(port)
}
// tcpState maps the hex state field from /proc/net/tcp to a human-readable name.
func tcpState(hex string) string {
switch strings.ToUpper(hex) {
case "01":
return "ESTABLISHED"
case "02":
return "SYN_SENT"
case "03":
return "SYN_RECV"
case "04":
return "FIN_WAIT1"
case "05":
return "FIN_WAIT2"
case "06":
return "TIME_WAIT"
case "07":
return "CLOSE"
case "08":
return "CLOSE_WAIT"
case "09":
return "LAST_ACK"
case "0A":
return "LISTEN"
case "0B":
return "CLOSING"
default:
return hex
}
}
// Helpers — avoid importing strconv for every tiny conversion.
func atoi(s string) int {
v, _ := strconv.Atoi(s)
return v
}
func atouint64(s string) uint64 {
v, _ := strconv.ParseUint(s, 10, 64)
return v
}
// Ensure log is used (ETHOS: no silent imports).
var _ = log.Printf

View file

@ -0,0 +1,10 @@
//go:build !linux
// +build !linux
package system
import "fmt"
func getFullProcessSnapshot() (*FullProcessSnapshot, error) {
return nil, fmt.Errorf("process detail scanning not supported on this platform")
}

View file

@ -122,16 +122,18 @@ func startWelcomeModeServer() {
c.JSON(200, gin.H{"status": "waiting for configuration"})
})
// Welcome page with setup instructions
router.GET("/", setupHandler.ShowSetupPage)
// Setup endpoint for web configuration
router.POST("/api/setup/configure", setupHandler.ConfigureServer)
router.POST("/api/setup/generate-keys", setupHandler.GenerateSigningKeys)
router.POST("/api/setup/configure-secrets", setupHandler.ConfigureSecrets)
// Setup endpoint for web configuration
router.GET("/setup", setupHandler.ShowSetupPage)
if webui.Present() {
registerWebUI(router)
} else {
// Fallback for UI-less development binaries.
router.GET("/", setupHandler.ShowSetupPage)
router.GET("/setup", setupHandler.ShowSetupPage)
}
log.Printf("Welcome mode server started on :8080")
log.Printf("Waiting for configuration...")
@ -226,6 +228,7 @@ func main() {
metricsQueries := queries.NewMetricsQueries(db.DB)
dockerQueries := queries.NewDockerQueries(db.DB)
storageMetricsQueries := queries.NewStorageMetricsQueries(db.DB.DB)
processQueries := queries.NewProcessQueries(db.DB.DB)
adminQueries := queries.NewAdminQueries(db.DB)
// Create PackageQueries for accessing signed agent update packages
@ -456,6 +459,7 @@ func main() {
metricsHandler := handlers.NewMetricsHandler(metricsQueries, agentQueries, commandQueries)
dockerReportsHandler := handlers.NewDockerReportsHandler(dockerQueries, agentQueries, commandQueries)
storageMetricsHandler := handlers.NewStorageMetricsHandler(storageMetricsQueries)
processHandler := handlers.NewProcessHandler(processQueries, agentQueries, commandQueries, signingService)
agentSetupHandler := handlers.NewAgentSetupHandler(agentQueries)
// Initialize scanner config handler (for user-configurable scanner timeouts)
@ -674,6 +678,8 @@ func main() {
// Initialize events handler [TD-003]
eventsHandler := handlers.NewEventsHandler(agentQueries)
agentSecurityEventsHandler := handlers.NewAgentSecurityEventsHandler(agentQueries)
inventoryHandler := handlers.NewInventoryHandler(queries.NewInventoryQueries(db.DB), agentQueries)
// Add routes that depend on agentHandler (must be after agentHandler creation)
api.POST("/agents/register", rateLimiter.RateLimit("agent_registration", middleware.KeyByIP), agentHandler.RegisterAgent)
@ -718,11 +724,21 @@ func main() {
// Dedicated storage metrics endpoint (proper separation from generic metrics)
agents.POST("/:id/storage-metrics", rateLimiter.RateLimit("agent_reports", middleware.KeyByAgentID), storageMetricsHandler.ReportStorageMetrics)
// Process scan reporting (on-demand, triggered by dashboard)
agents.POST("/:id/process-scan", rateLimiter.RateLimit("agent_reports", middleware.KeyByAgentID), processHandler.ReportProcessScan)
// Circuit breaker health reporting [ISSUE-004]
agents.POST("/:id/circuit-breakers", rateLimiter.RateLimit("agent_reports", middleware.KeyByAgentID), agentHandler.ReportCircuitBreakerStats)
// Event reporting [TD-003]
agents.POST("/:id/events", rateLimiter.RateLimit("agent_reports", middleware.KeyByAgentID), eventsHandler.ReportEvents)
// Security event reporting (agent → server security_events table)
agents.POST("/:id/security-events", rateLimiter.RateLimit("agent_reports", middleware.KeyByAgentID), agentSecurityEventsHandler.ReportSecurityEvents)
// Inventory reporting (agent → server agent_inventory table)
agents.POST("/:id/inventory", rateLimiter.RateLimit("agent_reports", middleware.KeyByAgentID), inventoryHandler.ReportInventory)
agents.GET("/:id/inventory", inventoryHandler.GetAgentInventory)
}
// Dashboard/Web routes (protected by web auth)
@ -730,9 +746,13 @@ func main() {
dashboard.Use(authHandler.WebAuthMiddleware())
{
dashboard.GET("/stats/summary", statsHandler.GetDashboardStats)
dashboard.GET("/inventory", inventoryHandler.GetFleetInventory)
dashboard.GET("/agents", agentHandler.ListAgents)
dashboard.GET("/agents/:id", agentHandler.GetAgent)
dashboard.GET("/agents/:id/storage-metrics", storageMetricsHandler.GetStorageMetrics)
dashboard.GET("/agents/:id/processes", processHandler.GetLatestProcessSnapshot)
dashboard.GET("/agents/:id/processes/:processId", processHandler.GetProcessDetail)
dashboard.POST("/agents/:id/processes/scan", processHandler.TriggerProcessScan)
dashboard.POST("/agents/:id/heartbeat", agentHandler.TriggerHeartbeat)
dashboard.GET("/agents/:id/heartbeat", agentHandler.GetHeartbeatStatus)
dashboard.POST("/agents/:id/reboot", agentHandler.TriggerReboot)

View file

@ -0,0 +1,101 @@
package handlers
import (
"fmt"
"net/http"
"time"
"github.com/Fimeg/RedFlag/server/internal/database/queries"
"github.com/Fimeg/RedFlag/server/internal/models"
"github.com/gin-gonic/gin"
"github.com/gofrs/uuid/v5"
)
// AgentSecurityEventsHandler handles security event reporting from agents.
type AgentSecurityEventsHandler struct {
agentQueries *queries.AgentQueries
}
// NewAgentSecurityEventsHandler creates a new agent security events handler.
func NewAgentSecurityEventsHandler(aq *queries.AgentQueries) *AgentSecurityEventsHandler {
return &AgentSecurityEventsHandler{agentQueries: aq}
}
// ReportSecurityEventsRequest is the JSON body for agent security event reports.
type ReportSecurityEventsRequest struct {
Events []AgentSecurityEventPayload `json:"events" binding:"required,dive"`
}
// AgentSecurityEventPayload is the wire format sent by the agent.
type AgentSecurityEventPayload struct {
Timestamp time.Time `json:"timestamp"`
Level string `json:"level"`
EventType string `json:"event_type"`
Message string `json:"message"`
Details map[string]interface{} `json:"details,omitempty"`
}
// ReportSecurityEventsResponse mirrors the ReportEventsResponse pattern.
type ReportSecurityEventsResponse struct {
Accepted int `json:"accepted"`
Rejected int `json:"rejected"`
Errors []string `json:"errors,omitempty"`
}
// ReportSecurityEvents handles POST /api/v1/agents/:id/security-events
// Accepts up to 100 security events per request. AgentID is taken from the
// URL parameter, matching the pattern used by ReportEvents.
func (h *AgentSecurityEventsHandler) ReportSecurityEvents(c *gin.Context) {
agentIDStr := c.Param("id")
agentID, err := uuid.FromString(agentIDStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid agent ID"})
return
}
var req ReportSecurityEventsRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"})
return
}
const maxEventsPerRequest = 100
if len(req.Events) > maxEventsPerRequest {
c.JSON(http.StatusBadRequest, gin.H{
"error": "too many events",
"max": maxEventsPerRequest,
"actual": len(req.Events),
})
return
}
response := ReportSecurityEventsResponse{}
for i, payload := range req.Events {
event := &models.SecurityEvent{
Timestamp: payload.Timestamp,
Level: payload.Level,
EventType: payload.EventType,
AgentID: agentID,
Message: payload.Message,
Details: payload.Details,
Metadata: map[string]interface{}{"source": "agent"},
}
if err := h.agentQueries.CreateSecurityEvent(event); err != nil {
response.Rejected++
response.Errors = append(response.Errors, fmt.Sprintf("event %d: %s", i, err.Error()))
} else {
response.Accepted++
}
}
status := http.StatusOK
if response.Rejected > 0 && response.Accepted == 0 {
status = http.StatusInternalServerError
} else if response.Rejected > 0 {
status = http.StatusPartialContent // 206
}
c.JSON(status, response)
}

View file

@ -0,0 +1,136 @@
package handlers
import (
"fmt"
"net/http"
"time"
"github.com/Fimeg/RedFlag/server/internal/database/queries"
"github.com/Fimeg/RedFlag/server/internal/models"
"github.com/gin-gonic/gin"
"github.com/gofrs/uuid/v5"
)
// InventoryHandler handles agent inventory reporting and queries.
type InventoryHandler struct {
inventoryQueries *queries.InventoryQueries
agentQueries *queries.AgentQueries
}
// NewInventoryHandler creates a new inventory handler.
func NewInventoryHandler(iq *queries.InventoryQueries, aq *queries.AgentQueries) *InventoryHandler {
return &InventoryHandler{
inventoryQueries: iq,
agentQueries: aq,
}
}
// ReportInventory handles POST /api/v1/agents/:id/inventory
// Accepts inventory items from an agent and upserts them into agent_inventory.
func (h *InventoryHandler) ReportInventory(c *gin.Context) {
agentIDStr := c.Param("id")
agentID, err := uuid.FromString(agentIDStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid agent ID"})
return
}
var req models.InventoryReportRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body"})
return
}
if len(req.Items) == 0 {
c.JSON(http.StatusOK, gin.H{"accepted": 0, "message": "no items to report"})
return
}
// Convert wire-format items to database model
items := make([]models.AgentInventoryItem, 0, len(req.Items))
for _, itemReq := range req.Items {
item := models.AgentInventoryItem{
ID: uuid.Must(uuid.NewV4()),
AgentID: agentID,
InventoryEcosystem: req.Ecosystem,
ItemName: itemReq.ItemName,
ItemVersion: itemReq.ItemVersion,
Description: itemReq.Description,
Arch: itemReq.Arch,
SizeBytes: itemReq.SizeBytes,
Vendor: itemReq.Vendor,
Metadata: models.JSONB(itemReq.Metadata),
}
// Parse install_time if provided
if itemReq.InstallTime != "" {
if t, err := time.Parse(time.RFC3339, itemReq.InstallTime); err == nil {
item.InstallTime = &t
}
}
items = append(items, item)
}
// Upsert all items
if err := h.inventoryQueries.UpsertBatch(agentID, items); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to upsert inventory: %s", err.Error())})
return
}
// Update agent last_seen
_ = h.agentQueries.UpdateAgentLastSeen(agentID)
c.JSON(http.StatusOK, gin.H{
"accepted": len(items),
"ecosystem": req.Ecosystem,
"scan_succeeded": req.ScanSucceeded,
})
}
// GetAgentInventory handles GET /api/v1/agents/:id/inventory
// Returns inventory items for an agent with optional ecosystem filter.
func (h *InventoryHandler) GetAgentInventory(c *gin.Context) {
agentIDStr := c.Param("id")
agentID, err := uuid.FromString(agentIDStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid agent ID"})
return
}
ecosystem := c.Query("ecosystem")
limit := 100
offset := 0
if l := c.Query("limit"); l != "" {
fmt.Sscanf(l, "%d", &limit)
}
if o := c.Query("offset"); o != "" {
fmt.Sscanf(o, "%d", &offset)
}
items, total, err := h.inventoryQueries.GetByAgent(agentID, ecosystem, limit, offset)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to query inventory"})
return
}
c.JSON(http.StatusOK, gin.H{
"items": items,
"total": total,
"limit": limit,
"offset": offset,
})
}
// GetFleetInventory handles GET /api/v1/inventory
// Returns a fleet-wide inventory summary.
func (h *InventoryHandler) GetFleetInventory(c *gin.Context) {
summaries, err := h.inventoryQueries.GetFleetSummary()
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to query fleet inventory"})
return
}
c.JSON(http.StatusOK, gin.H{"summary": summaries})
}

View file

@ -0,0 +1,371 @@
package handlers
import (
"encoding/json"
"fmt"
"log"
"net/http"
"time"
"github.com/Fimeg/RedFlag/server/internal/database/queries"
"github.com/Fimeg/RedFlag/server/internal/models"
"github.com/Fimeg/RedFlag/server/internal/services"
"github.com/gin-gonic/gin"
"github.com/gofrs/uuid/v5"
)
// ProcessHandler handles process scan endpoints.
type ProcessHandler struct {
processQueries *queries.ProcessQueries
agentQueries *queries.AgentQueries
commandQueries *queries.CommandQueries
signingService *services.SigningService
}
// NewProcessHandler creates a new process handler.
func NewProcessHandler(
pq *queries.ProcessQueries,
aq *queries.AgentQueries,
cq *queries.CommandQueries,
ss *services.SigningService,
) *ProcessHandler {
return &ProcessHandler{
processQueries: pq,
agentQueries: aq,
commandQueries: cq,
signingService: ss,
}
}
// ReportProcessScan handles POST /api/v1/agents/:id/process-scan
// Agent reports a full process scan result.
func (h *ProcessHandler) ReportProcessScan(c *gin.Context) {
agentID := c.MustGet("agent_id").(uuid.UUID)
var req models.ProcessScanRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body"})
return
}
if req.AgentID != agentID {
c.JSON(http.StatusBadRequest, gin.H{"error": "Agent ID mismatch"})
return
}
ctx := c.Request.Context()
now := time.Now().UTC()
// Insert snapshot header
snapID := uuid.Must(uuid.NewV4())
snap := models.ProcessSnapshot{
ID: snapID,
AgentID: agentID,
CommandID: req.CommandID,
ProcessCount: req.Snapshot.ProcessCount,
ScannedAt: req.Snapshot.ScannedAt,
ScanDurationMs: int(req.Snapshot.DurationMs),
CreatedAt: now,
}
if err := h.processQueries.InsertSnapshot(ctx, snap); err != nil {
log.Printf("[ERROR] [server] [processes] insert_snapshot agent=%s error=%v", agentID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to save snapshot"})
return
}
// Convert and insert processes
procs := make([]models.Process, 0, len(req.Snapshot.Processes))
for _, p := range req.Snapshot.Processes {
procID := uuid.Must(uuid.NewV4())
procs = append(procs, models.Process{
ID: procID,
SnapshotID: snapID,
AgentID: agentID,
PID: p.PID,
Name: p.Name,
Path: p.Path,
Cmdline: p.Cmdline,
Cwd: p.Cwd,
State: p.State,
UID: int(p.UID),
GID: int(p.GID),
EUID: int(p.EUID),
EGID: int(p.EGID),
User: p.User,
Group: p.Group,
TTY: p.TTY,
TTYName: p.TTYName,
CPUSecondsUser: p.CPUSecondsUser,
CPUSecondsSystem: p.CPUSecondsSystem,
CPUPercent: p.CPUPercent,
RSSBytes: int64(p.RSSBytes),
VMSBytes: int64(p.VMSBytes),
MemPercent: p.MemPercent,
Threads: p.Threads,
Nice: p.Nice,
StartTimeSeconds: int64(p.StartTimeSeconds),
ParentPID: p.ParentPID,
ProcessGroupID: p.ProcessGroupID,
ElevationStatus: p.ElevationStatus,
OnDisk: p.OnDisk,
DiskBytesRead: int64(p.DiskBytesRead),
DiskBytesWritten: int64(p.DiskBytesWritten),
CreatedAt: now,
})
}
if err := h.processQueries.InsertProcesses(ctx, procs); err != nil {
log.Printf("[ERROR] [server] [processes] insert_processes agent=%s error=%v", agentID, err)
// Clean up the snapshot header so the UI doesn't show an empty snapshot as latest.
_ = h.processQueries.DeleteSnapshot(ctx, snapID)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to save processes"})
return
}
// Insert related data for each process
var relatedEntries []models.ProcessRelated
for i, p := range req.Snapshot.Processes {
if i >= len(procs) {
break
}
procID := procs[i].ID
now := time.Now().UTC()
for _, relType := range []struct {
name string
data interface{}
}{
{"open_file", p.OpenFiles},
{"socket", p.OpenSockets},
{"pipe", p.OpenPipes},
{"environment", p.EnvironmentVars},
{"memory_map", p.MemoryMap},
{"namespace", p.Namespaces},
{"listening_port", p.ListeningPorts},
} {
if relType.data == nil {
continue
}
// Check if the slice is empty (non-nil but zero-length)
dataJSON, err := json.Marshal(relType.data)
if err != nil {
continue
}
if string(dataJSON) == "null" || string(dataJSON) == "[]" {
continue
}
var jb models.JSONB
if err := json.Unmarshal(dataJSON, &jb); err != nil {
continue
}
relatedEntries = append(relatedEntries, models.ProcessRelated{
ID: uuid.Must(uuid.NewV4()),
ProcessID: procID,
RelationType: relType.name,
Data: jb,
CreatedAt: now,
})
}
}
if len(relatedEntries) > 0 {
if err := h.processQueries.InsertRelatedData(ctx, relatedEntries); err != nil {
log.Printf("[ERROR] [server] [processes] insert_related agent=%s error=%v", agentID, err)
// Non-fatal: process data is saved, related data failed
}
}
// Cleanup old snapshots (keep last 10)
deleted, err := h.processQueries.CleanupOldSnapshots(ctx, agentID, 10)
if err != nil {
log.Printf("[WARN] [server] [processes] cleanup_old agent=%s error=%v", agentID, err)
} else if deleted > 0 {
log.Printf("[INFO] [server] [processes] cleanup_old agent=%s deleted=%d", agentID, deleted)
}
log.Printf("[INFO] [server] [processes] scan_stored agent=%s snapshot=%s processes=%d duration=%dms",
agentID, snapID, len(procs), snap.ScanDurationMs)
c.JSON(http.StatusOK, gin.H{
"status": "success",
"snapshot_id": snapID,
"processes": len(procs),
})
}
// GetLatestProcessSnapshot handles GET /api/v1/agents/:id/processes
// Dashboard reads the latest snapshot.
func (h *ProcessHandler) GetLatestProcessSnapshot(c *gin.Context) {
agentIDStr := c.Param("id")
agentID, err := uuid.FromString(agentIDStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid agent ID"})
return
}
ctx := c.Request.Context()
snap, err := h.processQueries.GetLatestSnapshot(ctx, agentID)
if err != nil {
log.Printf("[ERROR] [server] [processes] get_snapshot agent=%s error=%v", agentID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to retrieve snapshot"})
return
}
if snap == nil {
c.JSON(http.StatusOK, gin.H{"snapshot": nil, "processes": []interface{}{}})
return
}
// Parse query params for filtering
filter := queries.ProcessFilter{
Name: c.Query("name"),
User: c.Query("user"),
State: c.Query("state"),
SortBy: c.DefaultQuery("sort_by", "cpu"),
SortDir: c.DefaultQuery("sort_dir", "desc"),
Limit: 500, // cap at 500 for UI
}
if v := c.Query("limit"); v != "" {
fmt.Sscanf(v, "%d", &filter.Limit)
}
if v := c.Query("offset"); v != "" {
fmt.Sscanf(v, "%d", &filter.Offset)
}
procs, total, err := h.processQueries.GetProcessesBySnapshot(ctx, snap.ID, filter)
if err != nil {
log.Printf("[ERROR] [server] [processes] get_processes agent=%s error=%v", agentID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to retrieve processes"})
return
}
c.JSON(http.StatusOK, gin.H{
"snapshot": snap,
"processes": procs,
"total": total,
})
}
// GetProcessDetail handles GET /api/v1/agents/:id/processes/:processId
// Dashboard reads a single process with its related data.
func (h *ProcessHandler) GetProcessDetail(c *gin.Context) {
processIDStr := c.Param("processId")
processID, err := uuid.FromString(processIDStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid process ID"})
return
}
ctx := c.Request.Context()
proc, err := h.processQueries.GetProcessByID(ctx, processID)
if err != nil {
log.Printf("[ERROR] [server] [processes] get_detail error=%v", err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to retrieve process"})
return
}
if proc == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "Process not found"})
return
}
// Get all related data
related, err := h.processQueries.GetProcessRelated(ctx, processID, "")
if err != nil {
log.Printf("[WARN] [server] [processes] get_related error=%v", err)
}
// Group by type
resp := models.ProcessDetailResponse{Process: *proc}
for _, r := range related {
switch r.RelationType {
case "open_file":
resp.OpenFiles = append(resp.OpenFiles, r)
case "socket":
resp.OpenSockets = append(resp.OpenSockets, r)
case "pipe":
resp.OpenPipes = append(resp.OpenPipes, r)
case "environment":
resp.Environment = append(resp.Environment, r)
case "memory_map":
resp.MemoryMap = append(resp.MemoryMap, r)
case "namespace":
resp.Namespaces = append(resp.Namespaces, r)
case "listening_port":
resp.ListeningPorts = append(resp.ListeningPorts, r)
}
}
c.JSON(http.StatusOK, resp)
}
// TriggerProcessScan handles POST /api/v1/agents/:id/processes/scan
// Dashboard triggers an on-demand process scan.
func (h *ProcessHandler) TriggerProcessScan(c *gin.Context) {
agentIDStr := c.Param("id")
agentID, err := uuid.FromString(agentIDStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid agent ID"})
return
}
// Verify agent exists
agent, err := h.agentQueries.GetAgentByID(agentID)
if err != nil || agent == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "Agent not found"})
return
}
// Check for existing pending scan_processes command (dedup)
existingCmds, err := h.commandQueries.GetCommandsByAgentID(agentID)
if err == nil {
for _, cmd := range existingCmds {
if cmd.CommandType == "scan_processes" && (cmd.Status == "pending" || cmd.Status == "received") {
c.JSON(http.StatusOK, gin.H{
"message": "Scan already in progress",
"command_id": cmd.ID.String(),
})
return
}
}
}
// Create signed command
cmdID := uuid.Must(uuid.NewV4())
cmd := &models.AgentCommand{
ID: cmdID,
AgentID: agentID,
CommandType: "scan_processes",
Status: "pending",
CreatedAt: time.Now().UTC(),
UpdatedAt: time.Now().UTC(),
}
if h.signingService == nil || !h.signingService.IsEnabled() {
log.Printf("[ERROR] [server] [processes] signing_unavailable agent=%s", agentID)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Signing service unavailable"})
return
}
signature, err := h.signingService.SignCommand(cmd)
if err != nil {
log.Printf("[ERROR] [server] [processes] sign_failed agent=%s error=%v", agentID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to sign command"})
return
}
cmd.Signature = signature
if err := h.commandQueries.CreateCommand(cmd); err != nil {
log.Printf("[ERROR] [server] [processes] create_command agent=%s error=%v", agentID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create command"})
return
}
log.Printf("[INFO] [server] [processes] scan_triggered agent=%s command=%s", agentID, cmdID)
c.JSON(http.StatusOK, gin.H{
"message": "Process scan triggered",
"command_id": cmdID,
})
}

View file

@ -0,0 +1,3 @@
DROP TABLE IF EXISTS agent_process_related;
DROP TABLE IF EXISTS agent_processes;
DROP TABLE IF EXISTS agent_process_snapshots;

View file

@ -0,0 +1,69 @@
-- Process scan snapshots: one row per on-demand scan
CREATE TABLE IF NOT EXISTS agent_process_snapshots (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
command_id TEXT,
process_count INT NOT NULL DEFAULT 0,
scanned_at TIMESTAMP NOT NULL,
scan_duration_ms INT NOT NULL DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_process_snapshots_agent_latest
ON agent_process_snapshots(agent_id, created_at DESC);
-- Individual processes within a snapshot
CREATE TABLE IF NOT EXISTS agent_processes (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
snapshot_id UUID NOT NULL REFERENCES agent_process_snapshots(id) ON DELETE CASCADE,
agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
pid INT NOT NULL,
name TEXT NOT NULL,
path TEXT,
cmdline TEXT,
cwd TEXT,
state CHAR(1),
uid INT,
gid INT,
euid INT,
egid INT,
"user" TEXT,
"group" TEXT,
tty INT,
tty_name TEXT,
cpu_seconds_user FLOAT DEFAULT 0,
cpu_seconds_system FLOAT DEFAULT 0,
cpu_percent FLOAT DEFAULT 0,
rss_bytes BIGINT DEFAULT 0,
vms_bytes BIGINT DEFAULT 0,
mem_percent FLOAT DEFAULT 0,
threads INT DEFAULT 0,
nice INT DEFAULT 0,
start_time_seconds BIGINT DEFAULT 0,
parent_pid INT,
process_group_id INT,
elevation_status TEXT,
on_disk INT DEFAULT -1,
disk_bytes_read BIGINT DEFAULT 0,
disk_bytes_written BIGINT DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_processes_snapshot ON agent_processes(snapshot_id);
CREATE INDEX IF NOT EXISTS idx_processes_agent ON agent_processes(agent_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_processes_name ON agent_processes(name);
CREATE INDEX IF NOT EXISTS idx_processes_user ON agent_processes("user");
CREATE INDEX IF NOT EXISTS idx_processes_state ON agent_processes(state);
CREATE INDEX IF NOT EXISTS idx_processes_cpu ON agent_processes(cpu_percent DESC);
-- Related data (open files, sockets, pipes, env, memory map, namespaces, listening ports)
CREATE TABLE IF NOT EXISTS agent_process_related (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
process_id UUID NOT NULL REFERENCES agent_processes(id) ON DELETE CASCADE,
relation_type TEXT NOT NULL,
data JSONB NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_process_related_process ON agent_process_related(process_id);
CREATE INDEX IF NOT EXISTS idx_process_related_type ON agent_process_related(relation_type);

View file

@ -0,0 +1 @@
DROP TABLE IF EXISTS agent_inventory;

View file

@ -0,0 +1,31 @@
CREATE TABLE IF NOT EXISTS agent_inventory (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
agent_id UUID NOT NULL REFERENCES agents(id) ON DELETE CASCADE,
inventory_ecosystem VARCHAR(50) NOT NULL,
item_name TEXT NOT NULL,
item_version TEXT NOT NULL,
description TEXT,
arch VARCHAR(50),
install_time TIMESTAMPTZ,
size_bytes BIGINT,
vendor TEXT,
metadata JSONB DEFAULT '{}',
last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
first_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (agent_id, inventory_ecosystem, item_name)
);
CREATE INDEX IF NOT EXISTS idx_agent_inventory_agent
ON agent_inventory (agent_id);
CREATE INDEX IF NOT EXISTS idx_agent_inventory_ecosystem
ON agent_inventory (inventory_ecosystem);
CREATE INDEX IF NOT EXISTS idx_agent_inventory_agent_ecosystem
ON agent_inventory (agent_id, inventory_ecosystem);
CREATE INDEX IF NOT EXISTS idx_agent_inventory_last_seen
ON agent_inventory (last_seen_at);

View file

@ -0,0 +1,155 @@
package queries
import (
"fmt"
"time"
"github.com/Fimeg/RedFlag/server/internal/models"
"github.com/gofrs/uuid/v5"
"github.com/jmoiron/sqlx"
)
// InventoryQueries handles database operations for agent inventory.
type InventoryQueries struct {
db *sqlx.DB
}
// NewInventoryQueries creates a new InventoryQueries instance.
func NewInventoryQueries(db *sqlx.DB) *InventoryQueries {
return &InventoryQueries{db: db}
}
// UpsertBatch inserts or updates inventory items for an agent.
// Uses ON CONFLICT to update existing items (same agent + ecosystem + item_name).
func (q *InventoryQueries) UpsertBatch(agentID uuid.UUID, items []models.AgentInventoryItem) error {
if len(items) == 0 {
return nil
}
query := `
INSERT INTO agent_inventory (
agent_id, inventory_ecosystem, item_name, item_version,
description, arch, install_time, size_bytes, vendor, metadata,
last_seen_at, first_seen_at
) VALUES (
:agent_id, :inventory_ecosystem, :item_name, :item_version,
:description, :arch, :install_time, :size_bytes, :vendor, :metadata,
now(), now()
)
ON CONFLICT (agent_id, inventory_ecosystem, item_name) DO UPDATE SET
item_version = EXCLUDED.item_version,
description = EXCLUDED.description,
arch = EXCLUDED.arch,
install_time = EXCLUDED.install_time,
size_bytes = EXCLUDED.size_bytes,
vendor = EXCLUDED.vendor,
metadata = EXCLUDED.metadata,
last_seen_at = now(),
updated_at = now()
`
for _, item := range items {
item.AgentID = agentID
if item.ID == uuid.Nil {
item.ID = uuid.Must(uuid.NewV4())
}
_, err := q.db.NamedExec(query, item)
if err != nil {
return fmt.Errorf("failed to upsert inventory item %s: %w", item.ItemName, err)
}
}
return nil
}
// GetByAgent retrieves inventory items for an agent with optional ecosystem filter.
func (q *InventoryQueries) GetByAgent(agentID uuid.UUID, ecosystem string, limit, offset int) ([]models.AgentInventoryItem, int, error) {
// Count query
countQuery := `SELECT COUNT(*) FROM agent_inventory WHERE agent_id = $1`
countArgs := []interface{}{agentID}
if ecosystem != "" {
countQuery += ` AND inventory_ecosystem = $2`
countArgs = append(countArgs, ecosystem)
}
var total int
if err := q.db.Get(&total, countQuery, countArgs...); err != nil {
return nil, 0, fmt.Errorf("failed to count inventory items: %w", err)
}
// Data query
query := `
SELECT id, agent_id, inventory_ecosystem, item_name, item_version,
description, arch, install_time, size_bytes, vendor, metadata,
last_seen_at, first_seen_at, created_at, updated_at
FROM agent_inventory
WHERE agent_id = $1
`
args := []interface{}{agentID}
argIdx := 2
if ecosystem != "" {
query += fmt.Sprintf(` AND inventory_ecosystem = $%d`, argIdx)
args = append(args, ecosystem)
argIdx++
}
query += ` ORDER BY last_seen_at DESC`
query += fmt.Sprintf(` LIMIT $%d OFFSET $%d`, argIdx, argIdx+1)
args = append(args, limit, offset)
var items []models.AgentInventoryItem
if err := q.db.Select(&items, query, args...); err != nil {
return nil, 0, fmt.Errorf("failed to query inventory items: %w", err)
}
return items, total, nil
}
// MarkStale deletes inventory items not seen in the latest scan.
// Called after a successful scan to remove items that are no longer present.
func (q *InventoryQueries) MarkStale(agentID uuid.UUID, ecosystem string, scanTime time.Time) error {
query := `
DELETE FROM agent_inventory
WHERE agent_id = $1 AND inventory_ecosystem = $2 AND last_seen_at < $3
`
result, err := q.db.Exec(query, agentID, ecosystem, scanTime)
if err != nil {
return fmt.Errorf("failed to mark stale inventory items: %w", err)
}
affected, _ := result.RowsAffected()
if affected > 0 {
// Log but don't error — stale cleanup is best-effort
_ = affected
}
return nil
}
// GetFleetSummary returns a summary of inventory across all agents.
func (q *InventoryQueries) GetFleetSummary() ([]FleetInventorySummary, error) {
query := `
SELECT inventory_ecosystem, COUNT(DISTINCT agent_id) AS agent_count,
COUNT(*) AS item_count, SUM(size_bytes) AS total_size_bytes
FROM agent_inventory
GROUP BY inventory_ecosystem
ORDER BY inventory_ecosystem
`
var summaries []FleetInventorySummary
if err := q.db.Select(&summaries, query); err != nil {
return nil, fmt.Errorf("failed to query fleet inventory summary: %w", err)
}
return summaries, nil
}
// FleetInventorySummary is a summary row for fleet-wide inventory.
type FleetInventorySummary struct {
InventoryEcosystem string `db:"inventory_ecosystem" json:"inventory_ecosystem"`
AgentCount int `db:"agent_count" json:"agent_count"`
ItemCount int `db:"item_count" json:"item_count"`
TotalSizeBytes int64 `db:"total_size_bytes" json:"total_size_bytes"`
}

View file

@ -0,0 +1,356 @@
package queries
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"github.com/Fimeg/RedFlag/server/internal/models"
"github.com/gofrs/uuid/v5"
)
// ProcessQueries handles process snapshot database operations.
type ProcessQueries struct {
db *sql.DB
}
// NewProcessQueries creates a new process queries instance.
func NewProcessQueries(db *sql.DB) *ProcessQueries {
return &ProcessQueries{db: db}
}
// InsertSnapshot inserts a process snapshot header and returns its ID.
func (q *ProcessQueries) InsertSnapshot(ctx context.Context, snap models.ProcessSnapshot) error {
query := `
INSERT INTO agent_process_snapshots (id, agent_id, command_id, process_count, scanned_at, scan_duration_ms, created_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`
_, err := q.db.ExecContext(ctx, query,
snap.ID, snap.AgentID, snap.CommandID, snap.ProcessCount,
snap.ScannedAt, snap.ScanDurationMs, snap.CreatedAt,
)
if err != nil {
return fmt.Errorf("insert snapshot: %w", err)
}
return nil
}
// InsertProcesses bulk-inserts process rows within a single transaction.
func (q *ProcessQueries) InsertProcesses(ctx context.Context, procs []models.Process) error {
if len(procs) == 0 {
return nil
}
tx, err := q.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, `
INSERT INTO agent_processes (
id, snapshot_id, agent_id, pid, name, path, cmdline, cwd, state,
uid, gid, euid, egid, "user", "group", tty, tty_name,
cpu_seconds_user, cpu_seconds_system, cpu_percent,
rss_bytes, vms_bytes, mem_percent, threads, nice,
start_time_seconds, parent_pid, process_group_id,
elevation_status, on_disk, disk_bytes_read, disk_bytes_written, created_at
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9,
$10, $11, $12, $13, $14, $15, $16, $17,
$18, $19, $20,
$21, $22, $23, $24, $25,
$26, $27, $28,
$29, $30, $31, $32, $33
)
`)
if err != nil {
return fmt.Errorf("prepare: %w", err)
}
defer stmt.Close()
for _, p := range procs {
_, err := stmt.ExecContext(ctx,
p.ID, p.SnapshotID, p.AgentID, p.PID, p.Name, p.Path, p.Cmdline, p.Cwd, p.State,
p.UID, p.GID, p.EUID, p.EGID, p.User, p.Group, p.TTY, p.TTYName,
p.CPUSecondsUser, p.CPUSecondsSystem, p.CPUPercent,
p.RSSBytes, p.VMSBytes, p.MemPercent, p.Threads, p.Nice,
p.StartTimeSeconds, p.ParentPID, p.ProcessGroupID,
p.ElevationStatus, p.OnDisk, p.DiskBytesRead, p.DiskBytesWritten, p.CreatedAt,
)
if err != nil {
return fmt.Errorf("insert process pid=%d: %w", p.PID, err)
}
}
return tx.Commit()
}
// InsertRelatedData inserts related data rows for a process.
func (q *ProcessQueries) InsertRelatedData(ctx context.Context, entries []models.ProcessRelated) error {
if len(entries) == 0 {
return nil
}
tx, err := q.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("begin tx: %w", err)
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, `
INSERT INTO agent_process_related (id, process_id, relation_type, data, created_at)
VALUES ($1, $2, $3, $4, $5)
`)
if err != nil {
return fmt.Errorf("prepare: %w", err)
}
defer stmt.Close()
for _, e := range entries {
_, err := stmt.ExecContext(ctx, e.ID, e.ProcessID, e.RelationType, e.Data, e.CreatedAt)
if err != nil {
return fmt.Errorf("insert related type=%s: %w", e.RelationType, err)
}
}
return tx.Commit()
}
// GetLatestSnapshot returns the most recent snapshot for an agent, or nil if none.
func (q *ProcessQueries) GetLatestSnapshot(ctx context.Context, agentID uuid.UUID) (*models.ProcessSnapshot, error) {
query := `
SELECT id, agent_id, command_id, process_count, scanned_at, scan_duration_ms, created_at
FROM agent_process_snapshots
WHERE agent_id = $1
ORDER BY created_at DESC
LIMIT 1
`
var snap models.ProcessSnapshot
err := q.db.QueryRowContext(ctx, query, agentID).Scan(
&snap.ID, &snap.AgentID, &snap.CommandID, &snap.ProcessCount,
&snap.ScannedAt, &snap.ScanDurationMs, &snap.CreatedAt,
)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("get latest snapshot: %w", err)
}
return &snap, nil
}
// GetProcessesBySnapshot returns processes for a snapshot, with optional filtering and sorting.
func (q *ProcessQueries) GetProcessesBySnapshot(ctx context.Context, snapshotID uuid.UUID, filter ProcessFilter) ([]models.Process, int, error) {
// Count total
countQuery := `SELECT COUNT(*) FROM agent_processes WHERE snapshot_id = $1`
var total int
if err := q.db.QueryRowContext(ctx, countQuery, snapshotID).Scan(&total); err != nil {
return nil, 0, fmt.Errorf("count processes: %w", err)
}
// Build query with optional filters
query := `
SELECT id, snapshot_id, agent_id, pid, name, path, cmdline, cwd, state,
uid, gid, euid, egid, "user", "group", tty, tty_name,
cpu_seconds_user, cpu_seconds_system, cpu_percent,
rss_bytes, vms_bytes, mem_percent, threads, nice,
start_time_seconds, parent_pid, process_group_id,
elevation_status, on_disk, disk_bytes_read, disk_bytes_written, created_at
FROM agent_processes
WHERE snapshot_id = $1
`
args := []interface{}{snapshotID}
argIdx := 2
if filter.Name != "" {
query += fmt.Sprintf(" AND name ILIKE $%d", argIdx)
args = append(args, "%"+filter.Name+"%")
argIdx++
}
if filter.User != "" {
query += fmt.Sprintf(` AND "user" = $%d`, argIdx)
args = append(args, filter.User)
argIdx++
}
if filter.State != "" {
query += fmt.Sprintf(" AND state = $%d", argIdx)
args = append(args, filter.State)
argIdx++
}
// Sort
sortCol := "cpu_percent"
switch filter.SortBy {
case "mem", "mem_percent":
sortCol = "mem_percent"
case "pid":
sortCol = "pid"
case "name":
sortCol = "name"
case "threads":
sortCol = "threads"
case "rss":
sortCol = "rss_bytes"
}
sortDir := "DESC"
if filter.SortDir == "asc" {
sortDir = "ASC"
}
query += fmt.Sprintf(" ORDER BY %s %s", sortCol, sortDir)
// Pagination
if filter.Limit > 0 {
query += fmt.Sprintf(" LIMIT $%d", argIdx)
args = append(args, filter.Limit)
argIdx++
}
if filter.Offset > 0 {
query += fmt.Sprintf(" OFFSET $%d", argIdx)
args = append(args, filter.Offset)
argIdx++
}
rows, err := q.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, 0, fmt.Errorf("query processes: %w", err)
}
defer rows.Close()
var procs []models.Process
for rows.Next() {
var p models.Process
err := rows.Scan(
&p.ID, &p.SnapshotID, &p.AgentID, &p.PID, &p.Name, &p.Path, &p.Cmdline, &p.Cwd, &p.State,
&p.UID, &p.GID, &p.EUID, &p.EGID, &p.User, &p.Group, &p.TTY, &p.TTYName,
&p.CPUSecondsUser, &p.CPUSecondsSystem, &p.CPUPercent,
&p.RSSBytes, &p.VMSBytes, &p.MemPercent, &p.Threads, &p.Nice,
&p.StartTimeSeconds, &p.ParentPID, &p.ProcessGroupID,
&p.ElevationStatus, &p.OnDisk, &p.DiskBytesRead, &p.DiskBytesWritten, &p.CreatedAt,
)
if err != nil {
return nil, 0, fmt.Errorf("scan process: %w", err)
}
procs = append(procs, p)
}
if err := rows.Err(); err != nil {
return nil, 0, fmt.Errorf("iterate processes: %w", err)
}
return procs, total, nil
}
// GetProcessByID returns a single process by its ID.
func (q *ProcessQueries) GetProcessByID(ctx context.Context, processID uuid.UUID) (*models.Process, error) {
query := `
SELECT id, snapshot_id, agent_id, pid, name, path, cmdline, cwd, state,
uid, gid, euid, egid, "user", "group", tty, tty_name,
cpu_seconds_user, cpu_seconds_system, cpu_percent,
rss_bytes, vms_bytes, mem_percent, threads, nice,
start_time_seconds, parent_pid, process_group_id,
elevation_status, on_disk, disk_bytes_read, disk_bytes_written, created_at
FROM agent_processes
WHERE id = $1
`
var p models.Process
err := q.db.QueryRowContext(ctx, query, processID).Scan(
&p.ID, &p.SnapshotID, &p.AgentID, &p.PID, &p.Name, &p.Path, &p.Cmdline, &p.Cwd, &p.State,
&p.UID, &p.GID, &p.EUID, &p.EGID, &p.User, &p.Group, &p.TTY, &p.TTYName,
&p.CPUSecondsUser, &p.CPUSecondsSystem, &p.CPUPercent,
&p.RSSBytes, &p.VMSBytes, &p.MemPercent, &p.Threads, &p.Nice,
&p.StartTimeSeconds, &p.ParentPID, &p.ProcessGroupID,
&p.ElevationStatus, &p.OnDisk, &p.DiskBytesRead, &p.DiskBytesWritten, &p.CreatedAt,
)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("get process: %w", err)
}
return &p, nil
}
// GetProcessRelated returns related data for a process, optionally filtered by type.
func (q *ProcessQueries) GetProcessRelated(ctx context.Context, processID uuid.UUID, relationType string) ([]models.ProcessRelated, error) {
query := `SELECT id, process_id, relation_type, data, created_at FROM agent_process_related WHERE process_id = $1`
args := []interface{}{processID}
if relationType != "" {
query += " AND relation_type = $2"
args = append(args, relationType)
}
query += " ORDER BY relation_type, created_at"
rows, err := q.db.QueryContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("query related: %w", err)
}
defer rows.Close()
var entries []models.ProcessRelated
for rows.Next() {
var e models.ProcessRelated
if err := rows.Scan(&e.ID, &e.ProcessID, &e.RelationType, &e.Data, &e.CreatedAt); err != nil {
return nil, fmt.Errorf("scan related: %w", err)
}
entries = append(entries, e)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("iterate related: %w", err)
}
return entries, nil
}
// DeleteSnapshot deletes a snapshot by ID (cascades to processes and related data).
func (q *ProcessQueries) DeleteSnapshot(ctx context.Context, snapID uuid.UUID) error {
_, err := q.db.ExecContext(ctx, "DELETE FROM agent_process_snapshots WHERE id = $1", snapID)
if err != nil {
return fmt.Errorf("delete snapshot: %w", err)
}
return nil
}
// CleanupOldSnapshots keeps only the most recent N snapshots per agent.
func (q *ProcessQueries) CleanupOldSnapshots(ctx context.Context, agentID uuid.UUID, keepCount int) (int, error) {
query := `
DELETE FROM agent_process_snapshots
WHERE agent_id = $1
AND id NOT IN (
SELECT id FROM agent_process_snapshots
WHERE agent_id = $1
ORDER BY created_at DESC
LIMIT $2
)
`
result, err := q.db.ExecContext(ctx, query, agentID, keepCount)
if err != nil {
return 0, fmt.Errorf("cleanup snapshots: %w", err)
}
rows, _ := result.RowsAffected()
return int(rows), nil
}
// ProcessFilter defines filtering/sorting options for process queries.
type ProcessFilter struct {
Name string
User string
State string
SortBy string // cpu, mem, pid, name, threads, rss
SortDir string // asc, desc
Limit int
Offset int
}
// helper to marshal related data to JSONB
func marshalJSONB(v interface{}) (models.JSONB, error) {
data, err := json.Marshal(v)
if err != nil {
return nil, err
}
var jb models.JSONB
if err := json.Unmarshal(data, &jb); err != nil {
return nil, err
}
return jb, nil
}

View file

@ -109,6 +109,7 @@ const (
CommandTypeDisableHeartbeat = "disable_heartbeat"
CommandTypeReboot = "reboot"
CommandTypeCaptureScreenshot = "capture_screenshot"
CommandTypeScanProcesses = "scan_processes"
)
// Command statuses

View file

@ -0,0 +1,58 @@
package models
import (
"time"
"github.com/gofrs/uuid/v5"
)
// AgentInventoryItem represents a current-state inventory record for an agent.
// Distinct from current_package_state (which tracks available updates) and
// metrics (which track point-in-time measurements).
type AgentInventoryItem struct {
ID uuid.UUID `db:"id" json:"id"`
AgentID uuid.UUID `db:"agent_id" json:"agent_id"`
InventoryEcosystem string `db:"inventory_ecosystem" json:"inventory_ecosystem"`
ItemName string `db:"item_name" json:"item_name"`
ItemVersion string `db:"item_version" json:"item_version"`
Description string `db:"description" json:"description,omitempty"`
Arch string `db:"arch" json:"arch,omitempty"`
InstallTime *time.Time `db:"install_time" json:"install_time,omitempty"`
SizeBytes int64 `db:"size_bytes" json:"size_bytes,omitempty"`
Vendor string `db:"vendor" json:"vendor,omitempty"`
Metadata JSONB `db:"metadata" json:"metadata,omitempty"`
LastSeenAt time.Time `db:"last_seen_at" json:"last_seen_at"`
FirstSeenAt time.Time `db:"first_seen_at" json:"first_seen_at"`
CreatedAt time.Time `db:"created_at" json:"created_at"`
UpdatedAt time.Time `db:"updated_at" json:"updated_at"`
}
// InventoryReportRequest is the JSON body for agent inventory reports.
type InventoryReportRequest struct {
CommandID string `json:"command_id"`
Timestamp time.Time `json:"timestamp"`
Ecosystem string `json:"inventory_ecosystem"`
Items []AgentInventoryItemRequest `json:"items"`
ScanSucceeded bool `json:"scan_succeeded"`
}
// AgentInventoryItemRequest is a single inventory item from the agent wire format.
type AgentInventoryItemRequest struct {
Ecosystem string `json:"inventory_ecosystem"`
ItemName string `json:"item_name"`
ItemVersion string `json:"item_version"`
Description string `json:"description,omitempty"`
Arch string `json:"arch,omitempty"`
InstallTime string `json:"install_time,omitempty"`
SizeBytes int64 `json:"size_bytes,omitempty"`
Vendor string `json:"vendor,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// InventoryFilter supports paginated, filtered queries for inventory data.
type InventoryFilter struct {
AgentID *uuid.UUID
InventoryEcosystem *string
Limit *int
Offset *int
}

View file

@ -0,0 +1,189 @@
package models
import (
"time"
"github.com/gofrs/uuid/v5"
)
// ProcessSnapshot represents one on-demand process scan for an agent.
type ProcessSnapshot struct {
ID uuid.UUID `json:"id" db:"id"`
AgentID uuid.UUID `json:"agent_id" db:"agent_id"`
CommandID string `json:"command_id" db:"command_id"`
ProcessCount int `json:"process_count" db:"process_count"`
ScannedAt time.Time `json:"scanned_at" db:"scanned_at"`
ScanDurationMs int `json:"scan_duration_ms" db:"scan_duration_ms"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
}
// Process represents a single process within a snapshot.
type Process struct {
ID uuid.UUID `json:"id" db:"id"`
SnapshotID uuid.UUID `json:"snapshot_id" db:"snapshot_id"`
AgentID uuid.UUID `json:"agent_id" db:"agent_id"`
PID int `json:"pid" db:"pid"`
Name string `json:"name" db:"name"`
Path string `json:"path" db:"path"`
Cmdline string `json:"cmdline" db:"cmdline"`
Cwd string `json:"cwd" db:"cwd"`
State string `json:"state" db:"state"`
UID int `json:"uid" db:"uid"`
GID int `json:"gid" db:"gid"`
EUID int `json:"euid" db:"euid"`
EGID int `json:"egid" db:"egid"`
User string `json:"user" db:"user"`
Group string `json:"group" db:"group"`
TTY int `json:"tty" db:"tty"`
TTYName string `json:"tty_name" db:"tty_name"`
CPUSecondsUser float64 `json:"cpu_seconds_user" db:"cpu_seconds_user"`
CPUSecondsSystem float64 `json:"cpu_seconds_system" db:"cpu_seconds_system"`
CPUPercent float64 `json:"cpu_percent" db:"cpu_percent"`
RSSBytes int64 `json:"rss_bytes" db:"rss_bytes"`
VMSBytes int64 `json:"vms_bytes" db:"vms_bytes"`
MemPercent float64 `json:"mem_percent" db:"mem_percent"`
Threads int `json:"threads" db:"threads"`
Nice int `json:"nice" db:"nice"`
StartTimeSeconds int64 `json:"start_time_seconds" db:"start_time_seconds"`
ParentPID int `json:"parent_pid" db:"parent_pid"`
ProcessGroupID int `json:"process_group_id" db:"process_group_id"`
ElevationStatus string `json:"elevation_status" db:"elevation_status"`
OnDisk int `json:"on_disk" db:"on_disk"`
DiskBytesRead int64 `json:"disk_bytes_read" db:"disk_bytes_read"`
DiskBytesWritten int64 `json:"disk_bytes_written" db:"disk_bytes_written"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
}
// ProcessRelated represents related data (files, sockets, etc.) for a process.
type ProcessRelated struct {
ID uuid.UUID `json:"id" db:"id"`
ProcessID uuid.UUID `json:"process_id" db:"process_id"`
RelationType string `json:"relation_type" db:"relation_type"`
Data JSONB `json:"data" db:"data"`
CreatedAt time.Time `json:"created_at" db:"created_at"`
}
// ProcessScanRequest is the payload from the agent's process scan report.
type ProcessScanRequest struct {
AgentID uuid.UUID `json:"agent_id"`
CommandID string `json:"command_id"`
Timestamp time.Time `json:"timestamp"`
Snapshot ProcessSnapshotData `json:"snapshot"`
}
// ProcessSnapshotData mirrors the agent's FullProcessSnapshot for deserialization.
type ProcessSnapshotData struct {
Processes []ProcessData `json:"processes"`
ProcessCount int `json:"process_count"`
ScannedAt time.Time `json:"scanned_at"`
DurationMs int64 `json:"duration_ms"`
}
// ProcessData mirrors a single FullProcess from the agent.
type ProcessData struct {
PID int `json:"pid"`
Name string `json:"name"`
Path string `json:"path,omitempty"`
Cmdline string `json:"cmdline"`
Cwd string `json:"cwd,omitempty"`
State string `json:"state"`
UID uint32 `json:"uid"`
GID uint32 `json:"gid"`
EUID uint32 `json:"euid"`
EGID uint32 `json:"egid"`
User string `json:"user"`
Group string `json:"group"`
TTY int `json:"tty"`
TTYName string `json:"tty_name,omitempty"`
CPUSecondsUser float64 `json:"cpu_seconds_user"`
CPUSecondsSystem float64 `json:"cpu_seconds_system"`
CPUPercent float64 `json:"cpu_percent"`
RSSBytes uint64 `json:"rss_bytes"`
VMSBytes uint64 `json:"vms_bytes"`
MemPercent float64 `json:"mem_percent"`
Threads int `json:"threads"`
Nice int `json:"nice"`
StartTimeSeconds uint64 `json:"start_time_seconds"`
ParentPID int `json:"parent_pid"`
ProcessGroupID int `json:"process_group_id"`
OnDisk int `json:"on_disk"`
ElevationStatus string `json:"elevation_status,omitempty"`
DiskBytesRead uint64 `json:"disk_bytes_read,omitempty"`
DiskBytesWritten uint64 `json:"disk_bytes_written,omitempty"`
OpenFiles []ProcessOpenFileData `json:"open_files,omitempty"`
OpenSockets []ProcessOpenSocketData `json:"open_sockets,omitempty"`
OpenPipes []ProcessOpenPipeData `json:"open_pipes,omitempty"`
EnvironmentVars []string `json:"environment_vars,omitempty"`
MemoryMap []ProcessMemoryMapData `json:"memory_map,omitempty"`
Namespaces []ProcessNamespaceData `json:"namespaces,omitempty"`
ListeningPorts []ProcessListeningPortData `json:"listening_ports,omitempty"`
}
// Related data types for deserialization.
type ProcessOpenFileData struct {
FD int `json:"fd"`
Path string `json:"path"`
Type string `json:"type,omitempty"`
}
type ProcessOpenSocketData struct {
FD int `json:"fd"`
Family string `json:"family"`
Protocol string `json:"protocol"`
LocalAddr string `json:"local_addr"`
LocalPort int `json:"local_port"`
RemoteAddr string `json:"remote_addr"`
RemotePort int `json:"remote_port"`
State string `json:"state"`
Inode uint64 `json:"inode"`
Path string `json:"path,omitempty"`
}
type ProcessOpenPipeData struct {
FD int `json:"fd"`
Inode uint64 `json:"inode"`
Mode string `json:"mode,omitempty"`
Type string `json:"type,omitempty"`
}
type ProcessMemoryMapData struct {
Start uint64 `json:"start"`
End uint64 `json:"end"`
Permissions string `json:"permissions"`
Offset uint64 `json:"offset"`
Device string `json:"device"`
Inode uint64 `json:"inode"`
Path string `json:"path,omitempty"`
}
type ProcessNamespaceData struct {
Type string `json:"type"`
Inode string `json:"inode"`
}
type ProcessListeningPortData struct {
Protocol string `json:"protocol"`
LocalAddr string `json:"local_addr"`
LocalPort int `json:"local_port"`
FD int `json:"fd,omitempty"`
Socket uint64 `json:"socket,omitempty"`
}
// ProcessSnapshotResponse is the API response for the latest snapshot.
type ProcessSnapshotResponse struct {
Snapshot *ProcessSnapshot `json:"snapshot"`
Processes []Process `json:"processes"`
}
// ProcessDetailResponse is the API response for a single process with related data.
type ProcessDetailResponse struct {
Process Process `json:"process"`
OpenFiles []ProcessRelated `json:"open_files,omitempty"`
OpenSockets []ProcessRelated `json:"open_sockets,omitempty"`
OpenPipes []ProcessRelated `json:"open_pipes,omitempty"`
Environment []ProcessRelated `json:"environment,omitempty"`
MemoryMap []ProcessRelated `json:"memory_map,omitempty"`
Namespaces []ProcessRelated `json:"namespaces,omitempty"`
ListeningPorts []ProcessRelated `json:"listening_ports,omitempty"`
}

View file

@ -0,0 +1,356 @@
import React, { useState } from 'react';
import { X, Activity, Network, FileText, Key, Layers, Box } from 'lucide-react';
import { useProcessDetail } from '@/hooks/useProcesses';
import { cn } from '@/lib/utils';
// Safe JSON parse — returns null on malformed data instead of crashing.
const safeParse = (data: any): any => {
if (typeof data === 'string') {
try { return JSON.parse(data); } catch { return null; }
}
return data;
};
interface ProcessDetailModalProps {
agentId: string;
processId: string;
onClose: () => void;
}
type DetailTab = 'overview' | 'network' | 'files' | 'environment' | 'memory' | 'namespaces';
export const ProcessDetailModal: React.FC<ProcessDetailModalProps> = ({
agentId,
processId,
onClose,
}) => {
const [activeTab, setActiveTab] = useState<DetailTab>('overview');
const { data, isLoading } = useProcessDetail(agentId, processId, true);
if (isLoading) {
return (
<div className="fixed inset-0 z-50 flex items-center justify-center bg-black/50">
<div className="bg-white rounded-lg shadow-xl p-8">
<div className="flex items-center gap-3 text-gray-500">
<Activity className="h-5 w-5 animate-spin" />
Loading process detail...
</div>
</div>
</div>
);
}
if (!data) return null;
const proc = data.process;
const formatBytes = (bytes: number) => {
if (!bytes) return '0 B';
const k = 1024;
const sizes = ['B', 'KB', 'MB', 'GB'];
const i = Math.floor(Math.log(bytes) / Math.log(k));
return `${(bytes / Math.pow(k, i)).toFixed(1)} ${sizes[i]}`;
};
const formatTime = (seconds: number) => {
if (!seconds) return '—';
const d = new Date(seconds * 1000);
return d.toLocaleString();
};
const tabs: { key: DetailTab; label: string; icon: React.ReactNode; count?: number }[] = [
{ key: 'overview', label: 'Overview', icon: <Activity className="h-4 w-4" /> },
{ key: 'network', label: 'Network', icon: <Network className="h-4 w-4" />, count: (data.open_sockets?.length ?? 0) + (data.listening_ports?.length ?? 0) },
{ key: 'files', label: 'Files', icon: <FileText className="h-4 w-4" />, count: data.open_files?.length },
{ key: 'environment', label: 'Env', icon: <Key className="h-4 w-4" />, count: data.environment?.length },
{ key: 'memory', label: 'Memory', icon: <Layers className="h-4 w-4" />, count: data.memory_map?.length },
{ key: 'namespaces', label: 'Namespaces', icon: <Box className="h-4 w-4" />, count: data.namespaces?.length },
];
return (
<div className="fixed inset-0 z-50 flex items-center justify-center bg-black/50" onClick={onClose}>
<div
className="bg-white rounded-lg shadow-xl w-full max-w-4xl max-h-[85vh] flex flex-col"
onClick={(e) => e.stopPropagation()}
>
{/* Header */}
<div className="flex items-center justify-between px-6 py-4 border-b">
<div>
<h3 className="text-lg font-medium text-gray-900">{proc.name}</h3>
<p className="text-xs text-gray-500 font-mono">PID {proc.pid} · {proc.user} · {proc.path || 'no path'}</p>
</div>
<button onClick={onClose} className="p-1 hover:bg-gray-100 rounded">
<X className="h-5 w-5 text-gray-400" />
</button>
</div>
{/* Tabs */}
<div className="flex border-b px-6">
{tabs.map((tab) => (
<button
key={tab.key}
onClick={() => setActiveTab(tab.key)}
className={cn(
'flex items-center gap-1.5 px-3 py-2.5 text-sm border-b-2 -mb-px transition-colors',
activeTab === tab.key
? 'border-red-500 text-red-600'
: 'border-transparent text-gray-500 hover:text-gray-700'
)}
>
{tab.icon}
{tab.label}
{tab.count !== undefined && tab.count > 0 && (
<span className="ml-1 text-xs bg-gray-100 text-gray-600 rounded-full px-1.5">{tab.count}</span>
)}
</button>
))}
</div>
{/* Content */}
<div className="flex-1 overflow-y-auto p-6">
{activeTab === 'overview' && (
<div className="grid grid-cols-2 gap-4">
<Field label="Command" value={proc.cmdline || '—'} mono fullWidth />
<Field label="Path" value={proc.path || '—'} mono fullWidth />
<Field label="Working Directory" value={proc.cwd || '—'} mono />
<Field label="State" value={stateLabel(proc.state)} />
<Field label="UID / GID" value={`${proc.uid} / ${proc.gid}`} />
<Field label="EUID / EGID" value={`${proc.euid} / ${proc.egid}`} />
<Field label="Elevation" value={proc.elevation_status || 'none'} />
<Field label="CPU%" value={`${proc.cpu_percent.toFixed(2)}%`} />
<Field label="Mem%" value={`${proc.mem_percent.toFixed(2)}%`} />
<Field label="RSS" value={formatBytes(proc.rss_bytes)} />
<Field label="VMS" value={formatBytes(proc.vms_bytes)} />
<Field label="Threads" value={String(proc.threads)} />
<Field label="Nice" value={String(proc.nice)} />
<Field label="Parent PID" value={String(proc.parent_pid)} />
<Field label="Process Group" value={String(proc.process_group_id)} />
<Field label="TTY" value={proc.tty_name || String(proc.tty)} />
<Field label="Started" value={formatTime(proc.start_time_seconds)} />
<Field label="Disk Read" value={formatBytes(proc.disk_bytes_read)} />
<Field label="Disk Written" value={formatBytes(proc.disk_bytes_written)} />
<Field label="On Disk" value={proc.on_disk === 1 ? 'yes' : proc.on_disk === 0 ? 'no (deleted)' : 'unknown'} />
</div>
)}
{activeTab === 'network' && (
<div className="space-y-6">
{data.listening_ports && data.listening_ports.length > 0 && (
<div>
<h4 className="text-sm font-medium text-gray-700 mb-2">Listening Ports</h4>
<table className="table w-full text-xs">
<thead>
<tr className="table-header">
<th className="text-left py-1">Protocol</th>
<th className="text-left py-1">Address</th>
<th className="text-right py-1">Port</th>
</tr>
</thead>
<tbody>
{data.listening_ports.map((lp, i) => {
const d = safeParse(lp.data);
return (
<tr key={i} className="table-row">
<td className="table-cell">{d.protocol}</td>
<td className="table-cell font-mono">{d.local_addr || '0.0.0.0'}</td>
<td className="table-cell text-right font-mono">{d.local_port}</td>
</tr>
);
})}
</tbody>
</table>
</div>
)}
{data.open_sockets && data.open_sockets.length > 0 && (
<div>
<h4 className="text-sm font-medium text-gray-700 mb-2">Open Sockets</h4>
<table className="table w-full text-xs">
<thead>
<tr className="table-header">
<th className="text-left py-1">Family</th>
<th className="text-left py-1">Protocol</th>
<th className="text-left py-1">Local</th>
<th className="text-left py-1">Remote</th>
<th className="text-left py-1">State</th>
</tr>
</thead>
<tbody>
{data.open_sockets.map((s, i) => {
const d = safeParse(s.data);
return (
<tr key={i} className="table-row">
<td className="table-cell">{d.family}</td>
<td className="table-cell">{d.protocol}</td>
<td className="table-cell font-mono">{d.local_addr}:{d.local_port}</td>
<td className="table-cell font-mono">{d.remote_addr}:{d.remote_port}</td>
<td className="table-cell">{d.state || '—'}</td>
</tr>
);
})}
</tbody>
</table>
</div>
)}
{(!data.listening_ports || data.listening_ports.length === 0) &&
(!data.open_sockets || data.open_sockets.length === 0) && (
<p className="text-sm text-gray-500 italic">No network activity for this process.</p>
)}
</div>
)}
{activeTab === 'files' && (
<div>
{data.open_files && data.open_files.length > 0 ? (
<table className="table w-full text-xs">
<thead>
<tr className="table-header">
<th className="text-right py-1">FD</th>
<th className="text-left py-1">Type</th>
<th className="text-left py-1">Path</th>
</tr>
</thead>
<tbody>
{data.open_files.map((f, i) => {
const d = safeParse(f.data);
return (
<tr key={i} className="table-row">
<td className="table-cell text-right font-mono">{d.fd}</td>
<td className="table-cell">{d.type}</td>
<td className="table-cell font-mono truncate max-w-[400px]" title={d.path}>{d.path}</td>
</tr>
);
})}
</tbody>
</table>
) : (
<p className="text-sm text-gray-500 italic">No open files recorded.</p>
)}
</div>
)}
{activeTab === 'environment' && (
<div>
{data.environment && data.environment.length > 0 ? (
<div className="space-y-1">
<p className="text-xs text-gray-500 mb-3">Environment variable names only (values omitted for security).</p>
{data.environment.map((e, i) => {
const d = safeParse(e.data);
// d is an array of key names
if (Array.isArray(d)) {
return (
<div key={i} className="flex flex-wrap gap-1">
{d.map((key: string, j: number) => (
<span key={j} className="chip font-mono text-xs">{key}</span>
))}
</div>
);
}
return null;
})}
</div>
) : (
<p className="text-sm text-gray-500 italic">No environment data (may require elevated permissions).</p>
)}
</div>
)}
{activeTab === 'memory' && (
<div>
{data.memory_map && data.memory_map.length > 0 ? (
<div className="overflow-x-auto">
<table className="table w-full text-xs">
<thead>
<tr className="table-header">
<th className="text-left py-1">Address Range</th>
<th className="text-left py-1">Perms</th>
<th className="text-left py-1">Offset</th>
<th className="text-left py-1">Path</th>
</tr>
</thead>
<tbody>
{data.memory_map.slice(0, 200).map((m, i) => {
const d = safeParse(m.data);
return (
<tr key={i} className="table-row">
<td className="table-cell font-mono">
0x{d.start?.toString(16)}-0x{d.end?.toString(16)}
</td>
<td className="table-cell font-mono">{d.permissions}</td>
<td className="table-cell font-mono">0x{d.offset?.toString(16)}</td>
<td className="table-cell font-mono truncate max-w-[300px]" title={d.path}>{d.path || '—'}</td>
</tr>
);
})}
</tbody>
</table>
{data.memory_map.length > 200 && (
<p className="text-xs text-gray-400 mt-2">Showing 200 of {data.memory_map.length} regions.</p>
)}
</div>
) : (
<p className="text-sm text-gray-500 italic">No memory map data.</p>
)}
</div>
)}
{activeTab === 'namespaces' && (
<div>
{data.namespaces && data.namespaces.length > 0 ? (
<table className="table w-full text-xs">
<thead>
<tr className="table-header">
<th className="text-left py-1">Namespace</th>
<th className="text-left py-1">Inode</th>
</tr>
</thead>
<tbody>
{data.namespaces.map((ns, i) => {
const d = safeParse(ns.data);
return (
<tr key={i} className="table-row">
<td className="table-cell">{d.type}</td>
<td className="table-cell font-mono">{d.inode}</td>
</tr>
);
})}
</tbody>
</table>
) : (
<p className="text-sm text-gray-500 italic">No namespace data.</p>
)}
</div>
)}
</div>
</div>
</div>
);
};
// Helper components
const Field: React.FC<{ label: string; value: string; mono?: boolean; fullWidth?: boolean }> = ({
label,
value,
mono,
fullWidth,
}) => (
<div className={fullWidth ? 'col-span-2' : ''}>
<dt className="text-xs text-gray-500">{label}</dt>
<dd className={cn('text-sm text-gray-900 mt-0.5', mono && 'font-mono break-all')}>{value}</dd>
</div>
);
const stateLabel = (state: string) => {
const map: Record<string, string> = {
R: 'Running',
S: 'Sleeping',
D: 'Disk Sleep',
Z: 'Zombie',
T: 'Stopped',
t: 'Tracing Stop',
X: 'Dead',
x: 'Dead',
K: 'Wakekill',
W: 'Waking',
P: 'Parked',
};
return map[state] ?? state;
};

View file

@ -0,0 +1,237 @@
import React, { useState, useMemo } from 'react';
import { Activity, Search, RefreshCw, ArrowUpDown, Clock, Users } from 'lucide-react';
import { useProcessSnapshot, useTriggerProcessScan } from '@/hooks/useProcesses';
import { ProcessDetailModal } from '@/components/ProcessDetailModal';
import type { Process, ProcessFilter } from '@/types/process';
import { cn } from '@/lib/utils';
interface ProcessesTabProps {
agentId: string;
}
export const ProcessesTab: React.FC<ProcessesTabProps> = ({ agentId }) => {
const [selectedProcessId, setSelectedProcessId] = useState<string | null>(null);
const [filter, setFilter] = useState<ProcessFilter>({
sort_by: 'cpu',
sort_dir: 'desc',
limit: 500,
});
const [searchText, setSearchText] = useState('');
const { data, isLoading, isFetching } = useProcessSnapshot(agentId, filter);
const triggerScan = useTriggerProcessScan();
const processes = data?.processes ?? [];
const snapshot = data?.snapshot;
const total = data?.total ?? 0;
// Client-side search filtering (name/cmdline)
const filteredProcesses = useMemo(() => {
if (!searchText) return processes;
const lower = searchText.toLowerCase();
return processes.filter(
(p) =>
p.name.toLowerCase().includes(lower) ||
p.cmdline.toLowerCase().includes(lower)
);
}, [processes, searchText]);
const handleSort = (col: ProcessFilter['sort_by']) => {
setFilter((prev) => ({
...prev,
sort_by: col,
sort_dir: prev.sort_by === col && prev.sort_dir === 'desc' ? 'asc' : 'desc',
}));
};
const handleScan = () => {
triggerScan.mutate(agentId);
};
const formatBytes = (bytes: number) => {
if (bytes === 0) return '0 B';
const k = 1024;
const sizes = ['B', 'KB', 'MB', 'GB'];
const i = Math.floor(Math.log(bytes) / Math.log(k));
return `${(bytes / Math.pow(k, i)).toFixed(1)} ${sizes[i]}`;
};
const formatDuration = (ms: number) => {
if (ms < 1000) return `${ms}ms`;
return `${(ms / 1000).toFixed(1)}s`;
};
const SortHeader: React.FC<{ label; col: ProcessFilter['sort_by']; className?: string }> = ({
label,
col,
className,
}) => (
<th
className={cn('text-left py-2 px-2 font-medium cursor-pointer hover:text-gray-900 select-none', className)}
onClick={() => handleSort(col)}
>
<span className="inline-flex items-center gap-1">
{label}
{filter.sort_by === col && (
<ArrowUpDown className="h-3 w-3 text-gray-400" />
)}
</span>
</th>
);
return (
<div className="space-y-4">
{/* Header */}
<div className="flex items-center justify-between">
<div className="flex items-center gap-3">
<Activity className="h-5 w-5 text-gray-500" />
<div>
<h2 className="text-lg font-medium text-gray-900">Processes</h2>
{snapshot && (
<p className="text-xs text-gray-500">
{total} processes · scanned {formatDuration(snapshot.scan_duration_ms)} ·{' '}
{new Date(snapshot.scanned_at).toLocaleString()}
</p>
)}
</div>
</div>
<button
onClick={handleScan}
disabled={triggerScan.isPending}
className={cn(
'btn btn-secondary flex items-center gap-2',
triggerScan.isPending && 'opacity-50 cursor-not-allowed'
)}
>
<RefreshCw className={cn('h-4 w-4', triggerScan.isPending && 'animate-spin')} />
{triggerScan.isPending ? 'Scanning...' : 'Scan Now'}
</button>
</div>
{/* Search and filters */}
<div className="flex items-center gap-3">
<div className="relative flex-1 max-w-sm">
<Search className="absolute left-3 top-1/2 -translate-y-1/2 h-4 w-4 text-gray-400" />
<input
type="text"
placeholder="Filter by name or command..."
value={searchText}
onChange={(e) => setSearchText(e.target.value)}
className="pl-9 pr-4 py-2 w-full border border-gray-300 rounded-lg text-sm focus:outline-none focus:ring-2 focus:ring-primary-500 focus:border-transparent"
/>
</div>
<select
value={filter.state ?? ''}
onChange={(e) => setFilter((prev) => ({ ...prev, state: e.target.value || undefined }))}
className="px-3 py-2 border border-gray-300 rounded-lg text-sm focus:outline-none focus:ring-2 focus:ring-primary-500"
>
<option value="">All states</option>
<option value="R">Running</option>
<option value="S">Sleeping</option>
<option value="D">Disk Sleep</option>
<option value="Z">Zombie</option>
<option value="T">Stopped</option>
</select>
<select
value={filter.user ?? ''}
onChange={(e) => setFilter((prev) => ({ ...prev, user: e.target.value || undefined }))}
className="px-3 py-2 border border-gray-300 rounded-lg text-sm focus:outline-none focus:ring-2 focus:ring-primary-500"
>
<option value="">All users</option>
{/* Populated from process data */}
{[...new Set(processes.map((p) => p.user))].filter(Boolean).sort().map((u) => (
<option key={u} value={u}>{u}</option>
))}
</select>
</div>
{/* Process table */}
{isLoading || (isFetching && !snapshot) ? (
<div className="flex items-center justify-center py-12 text-gray-500">
<RefreshCw className="h-5 w-5 animate-spin mr-2" />
Loading process data...
</div>
) : !snapshot ? (
<div className="text-center py-12 text-gray-500">
<Activity className="h-8 w-8 mx-auto mb-3 text-gray-300" />
<p className="text-sm">No process data available.</p>
<p className="text-xs text-gray-400 mt-1">Click Scan to collect process information.</p>
</div>
) : (
<div className="overflow-x-auto">
<table className="table w-full">
<thead>
<tr className="table-header">
<SortHeader label="Name" col="name" />
<SortHeader label="PID" col="pid" className="text-right" />
<th className="text-left py-2 px-2 font-medium">User</th>
<th className="text-left py-2 px-2 font-medium">State</th>
<SortHeader label="CPU%" col="cpu" className="text-right" />
<SortHeader label="Mem%" col="mem" className="text-right" />
<SortHeader label="RSS" col="rss" className="text-right" />
<SortHeader label="Threads" col="threads" className="text-right" />
<th className="text-right py-2 px-2 font-medium">Nice</th>
</tr>
</thead>
<tbody>
{filteredProcesses.map((proc) => (
<tr
key={proc.id}
className="table-row cursor-pointer hover:bg-gray-50"
onClick={() => setSelectedProcessId(proc.id)}
>
<td className="table-cell font-medium max-w-[200px] truncate" title={proc.cmdline}>
{proc.name}
</td>
<td className="table-cell text-right text-gray-600 font-mono text-xs">
{proc.pid}
</td>
<td className="table-cell text-gray-600">{proc.user || '—'}</td>
<td className="table-cell">
<span
className={cn(
'badge badge-sm',
proc.state === 'R' && 'badge-success',
proc.state === 'S' && 'badge-info',
proc.state === 'D' && 'badge-warning',
proc.state === 'Z' && 'badge-danger',
proc.state === 'T' && 'badge-warning'
)}
>
{proc.state}
</span>
</td>
<td className="table-cell text-right font-mono text-xs">
{proc.cpu_percent > 0 ? `${proc.cpu_percent.toFixed(1)}%` : '—'}
</td>
<td className="table-cell text-right font-mono text-xs">
{proc.mem_percent > 0 ? `${proc.mem_percent.toFixed(1)}%` : '—'}
</td>
<td className="table-cell text-right font-mono text-xs">
{formatBytes(proc.rss_bytes)}
</td>
<td className="table-cell text-right">{proc.threads}</td>
<td className="table-cell text-right">{proc.nice}</td>
</tr>
))}
</tbody>
</table>
{filteredProcesses.length === 0 && processes.length > 0 && (
<p className="text-center py-6 text-sm text-gray-500">
No processes match your filter.
</p>
)}
</div>
)}
{/* Process detail modal */}
{selectedProcessId && (
<ProcessDetailModal
agentId={agentId}
processId={selectedProcessId}
onClose={() => setSelectedProcessId(null)}
/>
)}
</div>
);
};

View file

@ -0,0 +1,36 @@
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query';
import { agentApi } from '@/lib/api';
import type { ProcessSnapshotResponse, ProcessDetailResponse, ProcessFilter } from '@/types/process';
// Fetch the latest process snapshot for an agent.
export function useProcessSnapshot(agentId: string, filter?: ProcessFilter) {
return useQuery<ProcessSnapshotResponse>({
queryKey: ['process-snapshot', agentId, filter],
queryFn: () => agentApi.getLatestProcessSnapshot(agentId, filter as Record<string, string>),
staleTime: 24 * 60 * 60 * 1000, // 24h — data only changes on explicit scan
refetchInterval: false,
enabled: !!agentId,
});
}
// Trigger an on-demand process scan (mutation).
export function useTriggerProcessScan() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: (agentId: string) => agentApi.triggerProcessScan(agentId),
onSuccess: (_data, agentId) => {
// Invalidate snapshot so it refetches after scan completes
queryClient.invalidateQueries({ queryKey: ['process-snapshot', agentId] });
},
});
}
// Fetch process detail with related data (for drill-down modal).
export function useProcessDetail(agentId: string, processId: string, enabled: boolean) {
return useQuery<ProcessDetailResponse>({
queryKey: ['process-detail', agentId, processId],
queryFn: () => agentApi.getProcessDetail(agentId, processId),
enabled: enabled && !!agentId && !!processId,
staleTime: 24 * 60 * 60 * 1000,
});
}

View file

@ -272,6 +272,20 @@ export const agentApi = {
return response.data;
},
// Process Explorer — on-demand process scanning
getLatestProcessSnapshot: async (agentId: string, params?: Record<string, string>): Promise<any> => {
const response = await api.get(`/agents/${agentId}/processes`, { params });
return response.data;
},
getProcessDetail: async (agentId: string, processId: string): Promise<any> => {
const response = await api.get(`/agents/${agentId}/processes/${processId}`);
return response.data;
},
triggerProcessScan: async (agentId: string): Promise<any> => {
const response = await api.post(`/agents/${agentId}/processes/scan`);
return response.data;
},
// Get agent system metrics
getAgentSystemMetrics: async (agentId: string): Promise<any> => {
const response = await api.get(`/agents/${agentId}/metrics/system`);

View file

@ -46,11 +46,12 @@ import { AgentUpdatesModal } from '@/components/AgentUpdatesModal';
import { BulkAgentUpdate } from '@/components/RelayList';
import ChatTimeline from '@/components/ChatTimeline';
import AgentSoftwareBindings from '@/components/AgentSoftwareBindings';
import { ProcessesTab } from '@/components/ProcessesTab';
import { readIntegrations, resolveState } from '@/types/integrations';
type AgentDetailTab = 'overview' | 'storage' | 'updates' | 'software' | 'scanners' | 'history';
type AgentDetailTab = 'overview' | 'processes' | 'storage' | 'updates' | 'software' | 'scanners' | 'history';
const AGENT_DETAIL_TABS: AgentDetailTab[] = ['overview', 'storage', 'updates', 'software', 'scanners', 'history'];
const AGENT_DETAIL_TABS: AgentDetailTab[] = ['overview', 'processes', 'storage', 'updates', 'software', 'scanners', 'history'];
const parseAgentDetailTab = (tab: string | null): AgentDetailTab => {
return AGENT_DETAIL_TABS.includes(tab as AgentDetailTab) ? tab as AgentDetailTab : 'overview';
@ -487,6 +488,18 @@ const Agents: React.FC = () => {
>
<span>Overview</span>
</button>
<button
onClick={() => selectActiveTab('processes')}
className={cn(
'py-3 px-4 border-b-2 font-medium text-sm transition-colors whitespace-nowrap flex items-center space-x-2',
activeTab === 'processes'
? 'border-primary-500 text-primary-600 bg-primary-50 rounded-t-lg'
: 'border-transparent text-gray-500 hover:text-gray-700 hover:border-gray-300 hover:bg-gray-50'
)}
>
<Activity className="h-4 w-4" />
<span>Processes</span>
</button>
<button
onClick={() => selectActiveTab('storage')}
className={cn(
@ -768,7 +781,7 @@ const Agents: React.FC = () => {
className="col-span-full"
processes={topProcesses}
processCount={meta.processes}
onSeeMore={() => navigate(`/updates?agent=${selectedAgent.id}`)}
onSeeMore={() => selectActiveTab('processes')}
/>
</div>
</div>
@ -778,6 +791,10 @@ const Agents: React.FC = () => {
</div>
)}
{activeTab === 'processes' && (
<ProcessesTab agentId={selectedAgent.id} />
)}
{activeTab === 'storage' && (
<AgentStorage agentId={selectedAgent.id} />
)}

82
web/src/types/process.ts Normal file
View file

@ -0,0 +1,82 @@
// Process Explorer types — mirror the server models.
export interface ProcessSnapshot {
id: string;
agent_id: string;
command_id: string;
process_count: number;
scanned_at: string;
scan_duration_ms: number;
created_at: string;
}
export interface Process {
id: string;
snapshot_id: string;
agent_id: string;
pid: number;
name: string;
path: string;
cmdline: string;
cwd: string;
state: string;
uid: number;
gid: number;
euid: number;
egid: number;
user: string;
group: string;
tty: number;
tty_name: string;
cpu_seconds_user: number;
cpu_seconds_system: number;
cpu_percent: number;
rss_bytes: number;
vms_bytes: number;
mem_percent: number;
threads: number;
nice: number;
start_time_seconds: number;
parent_pid: number;
process_group_id: number;
elevation_status: string;
on_disk: number;
disk_bytes_read: number;
disk_bytes_written: number;
created_at: string;
}
export interface ProcessSnapshotResponse {
snapshot: ProcessSnapshot | null;
processes: Process[];
total: number;
}
export interface ProcessDetailResponse {
process: Process;
open_files?: ProcessRelated[];
open_sockets?: ProcessRelated[];
open_pipes?: ProcessRelated[];
environment?: ProcessRelated[];
memory_map?: ProcessRelated[];
namespaces?: ProcessRelated[];
listening_ports?: ProcessRelated[];
}
export interface ProcessRelated {
id: string;
process_id: string;
relation_type: string;
data: Record<string, any>;
created_at: string;
}
export interface ProcessFilter {
name?: string;
user?: string;
state?: string;
sort_by?: 'cpu' | 'mem' | 'pid' | 'name' | 'threads' | 'rss';
sort_dir?: 'asc' | 'desc';
limit?: number;
offset?: number;
}