vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-09 16:00:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Dedupe internal/api handler families behind shared flows and generics

Browse source 
Clears the sixteen dupl pair groups in internal/api plus the pkg/pulsecli
pair:

- router.go: privileged settings endpoints share the
  serveSetupTokenOrSettingsWrite gate; patrol findings convert via one
  unifiedFindingFromAI; the five infrastructure-summary chart loops share
  collectGuestChartData / fillChartSeriesFromBatch; the VM/LXC workloads
  summary loops share appendGuestWorkloadSummaries.
- deploy_handlers.go: preflight and job status/SSE handlers share
  handleDeployJobStatus / handleDeployJobEvents.
- recovery_handlers.go: points series/facets share
  parseRecoveryListPointsOptions.
- truenas_handlers.go / vmware_handlers.go / router_routes_registration.go:
  the connection update flow (locate, decode-with-fallback, normalize,
  preserve masked secrets, validate, save, redact) moves to the new
  platform_connection_shared.go (updatePlatformConnection +
  decodeOptionalInstanceRequest + the admin-gated item-route builder);
  per-platform wrappers carry nolint'd declarative wiring only.
- docker_metadata.go / guest_metadata.go: GET/PUT payload semantics move
  to metadata_handlers_shared.go; the zero-record-instead-of-404 contract
  is pinned by TestContract_MetadataGetPayloadsUseZeroRecordsInsteadOf404.
- kubernetes_agents.go / docker_agents.go: lifecycle PUTs share
  per-handler action helpers.
- config_node_handlers.go: PBS/PMG probes share
  testProxmoxPlatformConnection.
- cloud_handoff_handlers.go / purchase_return_redemptions.go: secrets
  sqlite stores open through openHardenedSecretsDB so permission
  hardening stays single-sourced.
- ai_handlers.go / chat_service_adapter.go: the GetMessages adapters are
  deliberate contract mirrors — suppressed with nolint and enforced by
  TestOrchestratorAndChatAdaptersMapTheSameMessageFields.
- pkg/pulsecli/actions.go: action subcommands seed env defaults via
  actionAPIDefaults (tested); the audit/events cobra registration pair is
  nolint'd parallel wiring.
- .golangci.yml: exclude gitignored tmp/ from ./... typechecking.
- subsystem_lookup_test.py: refresh the pinned api-contracts.md line
  numbers shifted by the contract additions.

golangci-lint run ./... is now fully green. Full internal/api and
pkg/pulsecli test suites pass.
This commit is contained in:
rcourtman 2026-06-10 10:52:05 +01:00
parent 26c7ca910c
commit 6340cd36f2
24 changed files with 977 additions and 1132 deletions
Download patch file Download diff file Expand all files Collapse all files

4
.golangci.yml
View file

@ -12,6 +12,10 @@ linters:
issues:
max-same-issues: 0
max-issues-per-linter: 0
# tmp/ is gitignored scratch (dev-mode module/toolchain caches live there);
# ./... must not typecheck it.
exclude-dirs:
- tmp
exclude-rules:
# Exclude dupl findings in test files and mock generators — test boilerplate
# is inherently repetitive and mock data generators share patterns by design.

8
docs/release-control/v6/internal/subsystems/ai-runtime.md
View file

@ -79,7 +79,13 @@ memory history files load through the generic `loadMemoryHistory`
`UnifiedFinding`/`unifiedFindingJSON` are deliberate marshal-mirror twins —
do not merge them; the mirror invariants are enforced by
`TestFindingJSONMirrorStaysInSync` and
`TestUnifiedFindingJSONMirrorStaysInSync`.
`TestUnifiedFindingJSONMirrorStaysInSync`. The same mirror posture applies on
the transport side: `orchestratorChatAdapter.GetMessages`
(`internal/api/ai_handlers.go`) and `chatServiceAdapter.GetMessages`
(`internal/api/chat_service_adapter.go`) convert the same chat-service
messages onto deliberately separate output contracts; do not merge them, and
keep their field mapping aligned —
`TestOrchestratorAndChatAdaptersMapTheSameMessageFields` enforces it.
Assistant frontend presentation changes under
`frontend-modern/src/components/AI/Chat/` must keep live tool activity aligned

26
docs/release-control/v6/internal/subsystems/api-contracts.md
View file

@ -26,6 +26,8 @@ product API routes free of maintainer commercial analytics.
## Canonical Files
1. `internal/api/contract_test.go`
1a. `internal/api/platform_connection_shared.go`
1b. `internal/api/metadata_handlers_shared.go`
2. `internal/api/resources.go`
3. `internal/api/discovery_handlers.go`
3. `internal/api/alerts.go`
@ -814,6 +816,30 @@ the canonical monitored-system blocked payload.
## Extension Points
0. Parallel platform/resource handler families share their flow logic instead
of per-platform copies. `internal/api/platform_connection_shared.go` owns
the platform connection update flow (locate by trimmed ID, decode with
stored-record fallback via the generic `decodeOptionalInstanceRequest`,
normalize, preserve masked secrets, validate, persist, respond redacted —
`updatePlatformConnection` + `platformConnectionUpdateSpec`) and the
admin+settings:write item-route mux builder
(`platformConnectionItemRoute`); a new platform connection surface wires a
spec instead of re-rolling the flow, so masked-secret preservation cannot
be forgotten. `internal/api/metadata_handlers_shared.go` owns the
guest/docker metadata GET/PUT payload semantics (empty object instead of
null, zero record instead of 404 — pinned by
`TestContract_MetadataGetPayloadsUseZeroRecordsInsteadOf404`). Deploy
preflights and jobs share `handleDeployJobStatus` /
`handleDeployJobEvents` in `internal/api/deploy_handlers.go`; recovery
points series/facets share `parseRecoveryListPointsOptions`; docker and
kubernetes agent lifecycle PUTs share per-handler
`handleHostLifecycleAction` / `handleClusterLifecycleAction`; PBS/PMG node
connection probes share `testProxmoxPlatformConnection`; secrets-backed
sqlite stores (handoff JTI, purchase-return redemptions) open through
`openHardenedSecretsDB` so the private-permission hardening policy stays
single-sourced; privileged settings endpoints gate through
`serveSetupTokenOrSettingsWrite` in `internal/api/router.go`; patrol
findings convert to the unified store through one `unifiedFindingFromAI`.
1. Add or change payload fields through handler + contract tests together
2. Update frontend API types in lockstep with backend contract changes.
Websocket-backed API consumers such as `frontend-modern/src/components/Settings/useAPITokenManagerState.ts` and `frontend-modern/src/components/Settings/useInfrastructureOperationsState.tsx` may read runtime context only through `frontend-modern/src/contexts/appRuntime.ts`; they must not import `frontend-modern/src/App.tsx`, because payload ownership remains in the API contract rather than the root shell.

1
internal/api/ai_handlers.go
View file

@ -2029,6 +2029,7 @@ func (a *orchestratorChatAdapter) ExecuteStream(ctx context.Context, req aicontr
})
}
//nolint:dupl // mirrors chatServiceAdapter.GetMessages: same source messages mapped onto a deliberately separate output contract that may diverge
func (a *orchestratorChatAdapter) GetMessages(ctx context.Context, sessionID string) ([]aicontracts.OrchestratorMessage, error) {
chatMessages, err := a.svc.GetMessages(ctx, sessionID)
if err != nil {

46
internal/api/ai_handlers_test.go
View file

@ -3316,3 +3316,49 @@ func TestAISettingsHandler_PatrolPreflight_RejectsInvalidProviderName(t *testing
assert.Equal(t, http.StatusBadRequest, rec.Code)
}
// TestOrchestratorAndChatAdaptersMapTheSameMessageFields keeps the deliberate
// GetMessages mirror between orchestratorChatAdapter (ai_handlers.go) and
// chatServiceAdapter (chat_service_adapter.go) honest: both convert the same
// chat-service messages onto separate output contracts, and a field mapped by
// one must be mapped by the other.
func TestOrchestratorAndChatAdaptersMapTheSameMessageFields(t *testing.T) {
requiredMappings := []string{
"ID:",
"Role:",
"Content:",
"ReasoningContent:",
"Timestamp:",
"ToolUseID:",
"IsError:",
".NormalizeCollections()",
}
for _, tc := range []struct {
file string
fn string
}{
{"ai_handlers.go", "func (a *orchestratorChatAdapter) GetMessages("},
{"chat_service_adapter.go", "func (a *chatServiceAdapter) GetMessages("},
} {
source, err := os.ReadFile(tc.file)
if err != nil {
t.Fatalf("read %s: %v", tc.file, err)
}
text := string(source)
start := strings.Index(text, tc.fn)
if start < 0 {
t.Fatalf("%s must define %q", tc.file, tc.fn)
}
end := strings.Index(text[start:], "\nfunc ")
if end < 0 {
end = len(text) - start
}
body := text[start : start+end]
for _, required := range requiredMappings {
if !strings.Contains(body, required) {
t.Errorf("%s GetMessages must map %q — keep the adapter mirror in sync", tc.file, required)
}
}
}
}

1
internal/api/chat_service_adapter.go
View file

@ -59,6 +59,7 @@ func (a *chatServiceAdapter) ExecutePatrolStream(ctx context.Context, req ai.Pat
}, nil
}
//nolint:dupl // mirrors orchestratorChatAdapter.GetMessages: same source messages mapped onto a deliberately separate output contract that may diverge
func (a *chatServiceAdapter) GetMessages(ctx context.Context, sessionID string) ([]ai.ChatMessage, error) {
messages, err := a.svc.GetMessages(ctx, sessionID)
if err != nil {

96
internal/api/cloud_handoff_handlers.go
View file

@ -56,41 +56,57 @@ const deleteExpiredHandoffJTIQuery = `
DELETE FROM handoff_jti INDEXED BY idx_handoff_jti_expires_at
WHERE expires_at <= ?`
// openHardenedSecretsDB opens (creating if needed) a single-connection WAL
// sqlite database under <configDir>/secrets with private dir/file
// permissions, applying schema. dirLabel and label feed error wrapping
// ("handoff" / "handoff jti", "purchase return" / "purchase return
// redemption"). Single-sourcing keeps the permission-hardening policy shared
// by every secrets-backed store.
func openHardenedSecretsDB(configDir, fileName, dirLabel, label, schema string) (*sql.DB, error) {
dir := filepath.Clean(configDir)
if strings.TrimSpace(dir) == "" {
return nil, fmt.Errorf("configDir is required")
}
secretsDir := filepath.Join(dir, "secrets")
if err := os.MkdirAll(secretsDir, handoffPrivateDirPerm); err != nil {
return nil, fmt.Errorf("create %s secrets dir: %w", dirLabel, err)
}
if err := os.Chmod(secretsDir, handoffPrivateDirPerm); err != nil {
return nil, fmt.Errorf("chmod %s secrets dir: %w", dirLabel, err)
}
dbPath := filepath.Join(secretsDir, fileName)
dsn := dbPath + "?" + url.Values{
"_pragma": []string{
"busy_timeout(30000)",
"journal_mode(WAL)",
"synchronous(NORMAL)",
},
}.Encode()
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, fmt.Errorf("open %s db: %w", label, err)
}
db.SetMaxOpenConns(1)
db.SetMaxIdleConns(1)
db.SetConnMaxLifetime(0)
if _, err := db.Exec(schema); err != nil {
_ = db.Close()
return nil, fmt.Errorf("init %s schema: %w", label, err)
}
for _, path := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
if err := hardenPrivateFile(path, handoffPrivateFilePerm); err != nil {
_ = db.Close()
return nil, fmt.Errorf("harden %s file permissions: %w", label, err)
}
}
return db, nil
}
func (s *jtiReplayStore) init() {
s.once.Do(func() {
dir := filepath.Clean(s.configDir)
if strings.TrimSpace(dir) == "" {
s.initErr = fmt.Errorf("configDir is required")
return
}
secretsDir := filepath.Join(dir, "secrets")
if err := os.MkdirAll(secretsDir, handoffPrivateDirPerm); err != nil {
s.initErr = fmt.Errorf("create handoff secrets dir: %w", err)
return
}
if err := os.Chmod(secretsDir, handoffPrivateDirPerm); err != nil {
s.initErr = fmt.Errorf("chmod handoff secrets dir: %w", err)
return
}
dbPath := filepath.Join(secretsDir, "handoff_jti.db")
dsn := dbPath + "?" + url.Values{
"_pragma": []string{
"busy_timeout(30000)",
"journal_mode(WAL)",
"synchronous(NORMAL)",
},
}.Encode()
db, err := sql.Open("sqlite", dsn)
if err != nil {
s.initErr = fmt.Errorf("open handoff jti db: %w", err)
return
}
db.SetMaxOpenConns(1)
db.SetMaxIdleConns(1)
db.SetConnMaxLifetime(0)
schema := `
CREATE TABLE IF NOT EXISTS handoff_jti (
jti TEXT PRIMARY KEY,
@ -98,19 +114,11 @@ func (s *jtiReplayStore) init() {
);
CREATE INDEX IF NOT EXISTS idx_handoff_jti_expires_at ON handoff_jti(expires_at);
`
if _, err := db.Exec(schema); err != nil {
_ = db.Close()
s.initErr = fmt.Errorf("init handoff jti schema: %w", err)
db, err := openHardenedSecretsDB(s.configDir, "handoff_jti.db", "handoff", "handoff jti", schema)
if err != nil {
s.initErr = err
return
}
for _, path := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
if err := hardenPrivateFile(path, handoffPrivateFilePerm); err != nil {
_ = db.Close()
s.initErr = fmt.Errorf("harden handoff jti file permissions: %w", err)
return
}
}
s.db = db
})
}

95
internal/api/config_node_handlers.go
View file

@ -1948,21 +1948,16 @@ func (h *ConfigHandlers) handleTestNode(w http.ResponseWriter, r *http.Request)
json.NewEncoder(w).Encode(testResult)
}
func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} {
// testProxmoxPlatformConnection runs the shared PBS/PMG node connection
// probe: build a client from the request credentials, exercise one read
// endpoint with a 10s timeout, and report latency. connect returns the
// platform-specific probe call.
func testProxmoxPlatformConnection(req NodeConfigRequest, successMsg string, connect func(verifySSL bool) (func(context.Context) error, error)) map[string]interface{} {
verifySSL := false
if req.VerifySSL != nil {
verifySSL = *req.VerifySSL
}
clientConfig := pbs.ClientConfig{
Host: req.Host,
User: req.User,
Password: req.Password,
TokenName: req.TokenName,
TokenValue: req.TokenValue,
VerifySSL: verifySSL,
Fingerprint: req.Fingerprint,
}
client, err := pbs.NewClient(clientConfig)
probe, err := connect(verifySSL)
if err != nil {
return map[string]interface{}{
"status": "error",
@ -1974,7 +1969,7 @@ func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if _, err := client.GetDatastores(ctx); err != nil {
if err := probe(ctx); err != nil {
return map[string]interface{}{
"status": "error",
"message": sanitizeErrorMessage(err, "connection"),
@ -1983,47 +1978,49 @@ func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} {
latency := time.Since(startTime).Milliseconds()
return map[string]interface{}{
"status": "success",
"message": "Connected to PBS instance",
"message": successMsg,
"latency": latency,
}
}
func testProxmoxBackupConnection(req NodeConfigRequest) map[string]interface{} {
return testProxmoxPlatformConnection(req, "Connected to PBS instance", func(verifySSL bool) (func(context.Context) error, error) {
client, err := pbs.NewClient(pbs.ClientConfig{
Host: req.Host,
User: req.User,
Password: req.Password,
TokenName: req.TokenName,
TokenValue: req.TokenValue,
VerifySSL: verifySSL,
Fingerprint: req.Fingerprint,
})
if err != nil {
return nil, err
}
return func(ctx context.Context) error {
_, err := client.GetDatastores(ctx)
return err
}, nil
})
}
func testProxmoxMailGatewayConnection(req NodeConfigRequest) map[string]interface{} {
verifySSL := false
if req.VerifySSL != nil {
verifySSL = *req.VerifySSL
}
clientConfig := pmg.ClientConfig{
Host: req.Host,
User: req.User,
Password: req.Password,
TokenName: req.TokenName,
TokenValue: req.TokenValue,
VerifySSL: verifySSL,
Fingerprint: req.Fingerprint,
}
client, err := pmg.NewClient(clientConfig)
if err != nil {
return map[string]interface{}{
"status": "error",
"message": sanitizeErrorMessage(err, "create_client"),
return testProxmoxPlatformConnection(req, "Connected to PMG instance", func(verifySSL bool) (func(context.Context) error, error) {
client, err := pmg.NewClient(pmg.ClientConfig{
Host: req.Host,
User: req.User,
Password: req.Password,
TokenName: req.TokenName,
TokenValue: req.TokenValue,
VerifySSL: verifySSL,
Fingerprint: req.Fingerprint,
})
if err != nil {
return nil, err
}
}
startTime := time.Now()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if _, err := client.GetVersion(ctx); err != nil {
return map[string]interface{}{
"status": "error",
"message": sanitizeErrorMessage(err, "connection"),
}
}
latency := time.Since(startTime).Milliseconds()
return map[string]interface{}{
"status": "success",
"message": "Connected to PMG instance",
"latency": latency,
}
return func(ctx context.Context) error {
_, err := client.GetVersion(ctx)
return err
}, nil
})
}

213
internal/api/deploy_handlers.go
View file

@ -393,43 +393,45 @@ func (h *DeployHandlers) HandleCreatePreflight(w http.ResponseWriter, r *http.Re
json.NewEncoder(w).Encode(resp)
}
// HandleGetPreflight returns the current status of a preflight job.
// GET /api/agent-deploy/preflights/{preflightId}
func (h *DeployHandlers) HandleGetPreflight(w http.ResponseWriter, r *http.Request) {
// handleDeployJobStatus serves the shared GET status handler for preflight
// and deploy jobs (both are deploy.Job records in the same store). pathPrefix
// is the route prefix to strip, label the user-facing noun ("Preflight",
// "Job"), logNoun the log vocabulary ("preflight", "deploy").
func (h *DeployHandlers) handleDeployJobStatus(w http.ResponseWriter, r *http.Request, pathPrefix, label, logNoun string) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
preflightID := extractPathSuffix(r.URL.Path, "/api/agent-deploy/preflights/")
if preflightID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Preflight ID is required", nil)
jobID := extractPathSuffix(r.URL.Path, pathPrefix)
if jobID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_id", label+" ID is required", nil)
return
}
// Strip /events suffix if present (shouldn't happen via routing, but be safe).
preflightID = strings.TrimSuffix(preflightID, "/events")
jobID = strings.TrimSuffix(jobID, "/events")
job, err := h.store.GetJob(r.Context(), preflightID)
job, err := h.store.GetJob(r.Context(), jobID)
if err != nil {
log.Error().Err(err).Str("id", preflightID).Msg("Failed to get preflight job")
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get preflight", nil)
log.Error().Err(err).Str("id", jobID).Msgf("Failed to get %s job", logNoun)
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get "+strings.ToLower(label), nil)
return
}
if job == nil {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil)
writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil)
return
}
// Tenant isolation: verify the job belongs to the caller's org.
orgID := resolveTenantOrgID(r)
if job.OrgID != orgID {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil)
writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil)
return
}
targets, err := h.store.GetTargetsForJob(r.Context(), preflightID)
targets, err := h.store.GetTargetsForJob(r.Context(), jobID)
if err != nil {
log.Error().Err(err).Str("id", preflightID).Msg("Failed to get preflight targets")
log.Error().Err(err).Str("id", jobID).Msgf("Failed to get %s targets", strings.ToLower(label))
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get targets", nil)
return
}
@ -446,38 +448,39 @@ func (h *DeployHandlers) HandleGetPreflight(w http.ResponseWriter, r *http.Reque
json.NewEncoder(w).Encode(resp)
}
// HandlePreflightEvents streams SSE events for a preflight job.
// GET /api/agent-deploy/preflights/{preflightId}/events
func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Request) {
// handleDeployJobEvents streams SSE events for a preflight or deploy job:
// replay persisted events, then live events with heartbeats until the client
// disconnects or the job completes.
func (h *DeployHandlers) handleDeployJobEvents(w http.ResponseWriter, r *http.Request, pathPrefix, label, logNoun string) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
// Extract preflight ID: /api/agent-deploy/preflights/{id}/events
path := strings.TrimPrefix(r.URL.Path, "/api/agent-deploy/preflights/")
preflightID := strings.TrimSuffix(path, "/events")
if preflightID == "" || preflightID == path {
writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Preflight ID is required", nil)
// Extract the job ID: <pathPrefix>{id}/events
path := strings.TrimPrefix(r.URL.Path, pathPrefix)
jobID := strings.TrimSuffix(path, "/events")
if jobID == "" || jobID == path {
writeErrorResponse(w, http.StatusBadRequest, "missing_id", label+" ID is required", nil)
return
}
// Verify the preflight exists.
job, err := h.store.GetJob(r.Context(), preflightID)
// Verify the job exists.
job, err := h.store.GetJob(r.Context(), jobID)
if err != nil {
log.Error().Err(err).Str("id", preflightID).Msg("Failed to get preflight job for SSE")
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get preflight", nil)
log.Error().Err(err).Str("id", jobID).Msgf("Failed to get %s job for SSE", logNoun)
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get "+strings.ToLower(label), nil)
return
}
if job == nil {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil)
writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil)
return
}
// Tenant isolation: verify the job belongs to the caller's org.
orgID := resolveTenantOrgID(r)
if job.OrgID != orgID {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Preflight not found", nil)
writeErrorResponse(w, http.StatusNotFound, "not_found", label+" not found", nil)
return
}
@ -493,13 +496,13 @@ func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Re
// Register SSE client.
clientID := generateID("sse")
eventCh := h.addSSEClient(preflightID, clientID)
defer h.removeSSEClient(preflightID, clientID)
eventCh := h.addSSEClient(jobID, clientID)
defer h.removeSSEClient(jobID, clientID)
// Send existing events first (replay).
events, replayErr := h.store.GetEventsForJob(r.Context(), preflightID)
events, replayErr := h.store.GetEventsForJob(r.Context(), jobID)
if replayErr != nil {
log.Error().Err(replayErr).Str("id", preflightID).Msg("Failed to load events for SSE replay")
log.Error().Err(replayErr).Str("id", jobID).Msg("Failed to load events for SSE replay")
// Send an error event so the client knows replay is incomplete.
fmt.Fprintf(w, "event: error\ndata: {\"message\":\"failed to load event history\"}\n\n")
}
@ -541,6 +544,18 @@ func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Re
}
}
// HandleGetPreflight returns the current status of a preflight job.
// GET /api/agent-deploy/preflights/{preflightId}
func (h *DeployHandlers) HandleGetPreflight(w http.ResponseWriter, r *http.Request) {
h.handleDeployJobStatus(w, r, "/api/agent-deploy/preflights/", "Preflight", "preflight")
}
// HandlePreflightEvents streams SSE events for a preflight job.
// GET /api/agent-deploy/preflights/{preflightId}/events
func (h *DeployHandlers) HandlePreflightEvents(w http.ResponseWriter, r *http.Request) {
h.handleDeployJobEvents(w, r, "/api/agent-deploy/preflights/", "Preflight", "preflight")
}
// --- Progress processing ---
// processPreflightProgress reads deploy progress events from the agent and
@ -1255,143 +1270,13 @@ func (h *DeployHandlers) HandleCreateJob(w http.ResponseWriter, r *http.Request)
// HandleGetJob returns the current status of a deploy job.
// GET /api/agent-deploy/jobs/{jobId}
func (h *DeployHandlers) HandleGetJob(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
jobID := extractPathSuffix(r.URL.Path, "/api/agent-deploy/jobs/")
if jobID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Job ID is required", nil)
return
}
jobID = strings.TrimSuffix(jobID, "/events")
job, err := h.store.GetJob(r.Context(), jobID)
if err != nil {
log.Error().Err(err).Str("id", jobID).Msg("Failed to get deploy job")
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get job", nil)
return
}
if job == nil {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil)
return
}
orgID := resolveTenantOrgID(r)
if job.OrgID != orgID {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil)
return
}
targets, err := h.store.GetTargetsForJob(r.Context(), jobID)
if err != nil {
log.Error().Err(err).Str("id", jobID).Msg("Failed to get job targets")
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get targets", nil)
return
}
resp := struct {
*deploy.Job
Targets []deploy.Target `json:"targets"`
}{
Job: job,
Targets: targets,
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resp)
h.handleDeployJobStatus(w, r, "/api/agent-deploy/jobs/", "Job", "deploy")
}
// HandleJobEvents streams SSE events for a deploy job.
// GET /api/agent-deploy/jobs/{jobId}/events
func (h *DeployHandlers) HandleJobEvents(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
// Extract job ID: /api/agent-deploy/jobs/{id}/events
path := strings.TrimPrefix(r.URL.Path, "/api/agent-deploy/jobs/")
jobID := strings.TrimSuffix(path, "/events")
if jobID == "" || jobID == path {
writeErrorResponse(w, http.StatusBadRequest, "missing_id", "Job ID is required", nil)
return
}
job, err := h.store.GetJob(r.Context(), jobID)
if err != nil {
log.Error().Err(err).Str("id", jobID).Msg("Failed to get deploy job for SSE")
writeErrorResponse(w, http.StatusInternalServerError, "store_error", "Failed to get job", nil)
return
}
if job == nil {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil)
return
}
orgID := resolveTenantOrgID(r)
if job.OrgID != orgID {
writeErrorResponse(w, http.StatusNotFound, "not_found", "Job not found", nil)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
writeErrorResponse(w, http.StatusInternalServerError, "streaming_unsupported", "Streaming not supported", nil)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
clientID := generateID("sse")
eventCh := h.addSSEClient(jobID, clientID)
defer h.removeSSEClient(jobID, clientID)
// Replay existing events.
events, replayErr := h.store.GetEventsForJob(r.Context(), jobID)
if replayErr != nil {
log.Error().Err(replayErr).Str("id", jobID).Msg("Failed to load events for SSE replay")
fmt.Fprintf(w, "event: error\ndata: {\"message\":\"failed to load event history\"}\n\n")
}
for _, evt := range events {
data, _ := json.Marshal(evt)
fmt.Fprintf(w, "data: %s\n\n", data)
}
flusher.Flush()
// If job is terminal, send final and close.
if isDeployJobTerminal(job.Status) {
data, _ := json.Marshal(map[string]string{
"type": "job_complete",
"status": string(job.Status),
})
fmt.Fprintf(w, "data: %s\n\n", data)
flusher.Flush()
return
}
// Stream new events.
heartbeat := time.NewTicker(15 * time.Second)
defer heartbeat.Stop()
for {
select {
case <-r.Context().Done():
return
case eventData, ok := <-eventCh:
if !ok {
return
}
fmt.Fprintf(w, "data: %s\n\n", eventData)
flusher.Flush()
case <-heartbeat.C:
fmt.Fprint(w, ": heartbeat\n\n")
flusher.Flush()
}
}
h.handleDeployJobEvents(w, r, "/api/agent-deploy/jobs/", "Job", "deploy")
}
// HandleCancelJob cancels a running deploy job.

70
internal/api/docker_agents.go
View file

@ -1,6 +1,7 @@
package api
import (
"context"
"encoding/json"
"errors"
"net/http"
@ -407,20 +408,29 @@ func (h *DockerAgentHandlers) HandleAllowReenroll(w http.ResponseWriter, r *http
}
}
// HandleUnhideHost unhides a previously hidden Docker / Podman host.
func (h *DockerAgentHandlers) HandleUnhideHost(w http.ResponseWriter, r *http.Request) {
// handleHostLifecycleAction runs the shared PUT docker host lifecycle flow:
// resolve the agent ID from the suffixed route, apply the monitor action,
// broadcast state, and respond. action returns the canonical host ID.
func (h *DockerAgentHandlers) handleHostLifecycleAction(
w http.ResponseWriter,
r *http.Request,
pathSuffix string,
successMsg string,
logMsg string,
action func(ctx context.Context, agentID string) (string, error),
) {
if r.Method != http.MethodPut {
writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil)
return
}
agentID := dockerRuntimeAgentIDFromPath(r.URL.Path, "/unhide")
agentID := dockerRuntimeAgentIDFromPath(r.URL.Path, pathSuffix)
if agentID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_agent_id", "Agent ID is required", nil)
return
}
host, err := h.getMonitor(r.Context()).UnhideDockerHost(agentID)
hostID, err := action(r.Context(), agentID)
if err != nil {
writeErrorResponse(w, http.StatusNotFound, "docker_agent_not_found", err.Error(), nil)
return
@ -430,41 +440,35 @@ func (h *DockerAgentHandlers) HandleUnhideHost(w http.ResponseWriter, r *http.Re
if err := utils.WriteJSONResponse(w, map[string]any{
"success": true,
"agentId": host.ID,
"message": "Docker / Podman module unhidden",
"agentId": hostID,
"message": successMsg,
}); err != nil {
log.Error().Err(err).Msg("Failed to serialize docker host unhide response")
log.Error().Err(err).Msg(logMsg)
}
}
// HandleUnhideHost unhides a previously hidden Docker / Podman host.
func (h *DockerAgentHandlers) HandleUnhideHost(w http.ResponseWriter, r *http.Request) {
h.handleHostLifecycleAction(w, r, "/unhide", "Docker / Podman module unhidden", "Failed to serialize docker host unhide response",
func(ctx context.Context, agentID string) (string, error) {
host, err := h.getMonitor(ctx).UnhideDockerHost(agentID)
if err != nil {
return "", err
}
return host.ID, nil
})
}
// HandleMarkPendingUninstall marks a Docker / Podman host as pending uninstall.
func (h *DockerAgentHandlers) HandleMarkPendingUninstall(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut {
writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil)
return
}
agentID := dockerRuntimeAgentIDFromPath(r.URL.Path, "/pending-uninstall")
if agentID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_agent_id", "Agent ID is required", nil)
return
}
host, err := h.getMonitor(r.Context()).MarkDockerHostPendingUninstall(agentID)
if err != nil {
writeErrorResponse(w, http.StatusNotFound, "docker_agent_not_found", err.Error(), nil)
return
}
h.broadcastState(r.Context())
if err := utils.WriteJSONResponse(w, map[string]any{
"success": true,
"agentId": host.ID,
"message": "Docker / Podman module marked as pending uninstall",
}); err != nil {
log.Error().Err(err).Msg("Failed to serialize docker host pending uninstall response")
}
h.handleHostLifecycleAction(w, r, "/pending-uninstall", "Docker / Podman module marked as pending uninstall", "Failed to serialize docker host pending uninstall response",
func(ctx context.Context, agentID string) (string, error) {
host, err := h.getMonitor(ctx).MarkDockerHostPendingUninstall(agentID)
if err != nil {
return "", err
}
return host.ID, nil
})
}
// HandleSetCustomDisplayName updates the custom display name for a Docker / Podman host.

93
internal/api/docker_metadata.go
View file

@ -45,88 +45,25 @@ func (h *DockerMetadataHandler) Store() *config.DockerMetadataStore {
// HandleGetMetadata retrieves metadata for a specific Docker resource or all resources
func (h *DockerMetadataHandler) HandleGetMetadata(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
// Check if requesting specific resource
path := r.URL.Path
// Handle both /api/docker/metadata and /api/docker/metadata/
if path == "/api/docker/metadata" || path == "/api/docker/metadata/" {
// Get all metadata
w.Header().Set("Content-Type", "application/json")
store := h.getStore(r.Context())
allMeta := store.GetAll()
if allMeta == nil {
// Return empty object instead of null
json.NewEncoder(w).Encode(make(map[string]*config.DockerMetadata))
} else {
json.NewEncoder(w).Encode(allMeta)
}
return
}
// Get specific resource ID from path
resourceID := strings.TrimPrefix(path, "/api/docker/metadata/")
w.Header().Set("Content-Type", "application/json")
if resourceID != "" {
// Get specific Docker resource metadata
store := h.getStore(r.Context())
meta := store.Get(resourceID)
if meta == nil {
// Return empty metadata instead of 404
json.NewEncoder(w).Encode(&config.DockerMetadata{ID: resourceID})
} else {
json.NewEncoder(w).Encode(meta)
}
} else {
// This shouldn't happen with current routing, but handle it anyway
http.Error(w, "Invalid request path", http.StatusBadRequest)
}
handleMetadataGetRequest(w, r, "/api/docker/metadata",
func(ctx context.Context) map[string]*config.DockerMetadata { return h.getStore(ctx).GetAll() },
func(ctx context.Context, id string) *config.DockerMetadata { return h.getStore(ctx).Get(id) },
func(id string) *config.DockerMetadata { return &config.DockerMetadata{ID: id} },
)
}
// HandleUpdateMetadata updates metadata for a Docker resource
func (h *DockerMetadataHandler) HandleUpdateMetadata(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut && r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
resourceID := strings.TrimPrefix(r.URL.Path, "/api/docker/metadata/")
if resourceID == "" || resourceID == "metadata" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
// Limit request body to 16KB to prevent memory exhaustion
r.Body = http.MaxBytesReader(w, r.Body, 16*1024)
var meta config.DockerMetadata
if err := json.NewDecoder(r.Body).Decode(&meta); err != nil {
http.Error(w, "Invalid request body", http.StatusBadRequest)
return
}
// Validate URL if provided
if errMsg := validateCustomURL(meta.CustomURL); errMsg != "" {
http.Error(w, errMsg, http.StatusBadRequest)
return
}
store := h.getStore(r.Context())
if err := store.Set(resourceID, &meta); err != nil {
log.Error().Err(err).Str("resourceID", resourceID).Msg("Failed to save Docker metadata")
http.Error(w, metadataSaveErrorMessage(err), http.StatusInternalServerError)
return
}
log.Info().Str("resourceID", resourceID).Str("url", meta.CustomURL).Msg("Updated Docker metadata")
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(&meta)
handleMetadataUpdateRequest(w, r, "/api/docker/metadata",
"Resource ID required",
"resourceID",
"Failed to save Docker metadata",
"Updated Docker metadata",
func(meta *config.DockerMetadata) string { return meta.CustomURL },
func(ctx context.Context, id string, meta *config.DockerMetadata) error {
return h.getStore(ctx).Set(id, meta)
},
)
}
// HandleDeleteMetadata removes metadata for a Docker resource

94
internal/api/guest_metadata.go
View file

@ -2,7 +2,6 @@ package api
import (
"context"
"encoding/json"
"net/http"
"strings"
@ -52,88 +51,25 @@ func (h *GuestMetadataHandler) Store() *config.GuestMetadataStore {
// HandleGetMetadata retrieves metadata for a specific guest or all guests
func (h *GuestMetadataHandler) HandleGetMetadata(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
// Check if requesting specific guest
path := r.URL.Path
// Handle both /api/guests/metadata and /api/guests/metadata/
if path == "/api/guests/metadata" || path == "/api/guests/metadata/" {
// Get all metadata
w.Header().Set("Content-Type", "application/json")
store := h.getStore(r.Context())
allMeta := store.GetAll()
if allMeta == nil {
// Return empty object instead of null
json.NewEncoder(w).Encode(make(map[string]*config.GuestMetadata))
} else {
json.NewEncoder(w).Encode(allMeta)
}
return
}
// Get specific guest ID from path
guestID := strings.TrimPrefix(path, "/api/guests/metadata/")
w.Header().Set("Content-Type", "application/json")
if guestID != "" {
// Get specific guest metadata
store := h.getStore(r.Context())
meta := store.Get(guestID)
if meta == nil {
// Return empty metadata instead of 404
json.NewEncoder(w).Encode(&config.GuestMetadata{ID: guestID})
} else {
json.NewEncoder(w).Encode(meta)
}
} else {
// This shouldn't happen with current routing, but handle it anyway
http.Error(w, "Invalid request path", http.StatusBadRequest)
}
handleMetadataGetRequest(w, r, "/api/guests/metadata",
func(ctx context.Context) map[string]*config.GuestMetadata { return h.getStore(ctx).GetAll() },
func(ctx context.Context, id string) *config.GuestMetadata { return h.getStore(ctx).Get(id) },
func(id string) *config.GuestMetadata { return &config.GuestMetadata{ID: id} },
)
}
// HandleUpdateMetadata updates metadata for a guest
func (h *GuestMetadataHandler) HandleUpdateMetadata(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut && r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
guestID := strings.TrimPrefix(r.URL.Path, "/api/guests/metadata/")
if guestID == "" || guestID == "metadata" {
http.Error(w, "Guest ID required", http.StatusBadRequest)
return
}
// Limit request body to 16KB to prevent memory exhaustion
r.Body = http.MaxBytesReader(w, r.Body, 16*1024)
var meta config.GuestMetadata
if err := json.NewDecoder(r.Body).Decode(&meta); err != nil {
http.Error(w, "Invalid request body", http.StatusBadRequest)
return
}
// Validate URL if provided
if errMsg := validateCustomURL(meta.CustomURL); errMsg != "" {
http.Error(w, errMsg, http.StatusBadRequest)
return
}
store := h.getStore(r.Context())
if err := store.Set(guestID, &meta); err != nil {
log.Error().Err(err).Str("guestID", guestID).Msg("Failed to save guest metadata")
http.Error(w, metadataSaveErrorMessage(err), http.StatusInternalServerError)
return
}
log.Info().Str("guestID", guestID).Str("url", meta.CustomURL).Msg("Updated guest metadata")
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(&meta)
handleMetadataUpdateRequest(w, r, "/api/guests/metadata",
"Guest ID required",
"guestID",
"Failed to save guest metadata",
"Updated guest metadata",
func(meta *config.GuestMetadata) string { return meta.CustomURL },
func(ctx context.Context, id string, meta *config.GuestMetadata) error {
return h.getStore(ctx).Set(id, meta)
},
)
}
// HandleDeleteMetadata removes metadata for a guest

71
internal/api/kubernetes_agents.go
View file

@ -194,22 +194,31 @@ func (h *KubernetesAgentHandlers) HandleAllowReenroll(w http.ResponseWriter, r *
}
}
// HandleUnhideCluster unhides a previously hidden kubernetes cluster.
func (h *KubernetesAgentHandlers) HandleUnhideCluster(w http.ResponseWriter, r *http.Request) {
// handleClusterLifecycleAction runs the shared PUT cluster lifecycle flow:
// resolve the cluster ID from the suffixed route, apply the monitor action,
// broadcast state, and respond. action returns the canonical cluster ID.
func (h *KubernetesAgentHandlers) handleClusterLifecycleAction(
w http.ResponseWriter,
r *http.Request,
pathSuffix string,
successMsg string,
logMsg string,
action func(ctx context.Context, clusterID string) (string, error),
) {
if r.Method != http.MethodPut {
writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil)
return
}
trimmedPath := strings.TrimPrefix(r.URL.Path, "/api/agents/kubernetes/clusters/")
trimmedPath = strings.TrimSuffix(trimmedPath, "/unhide")
trimmedPath = strings.TrimSuffix(trimmedPath, pathSuffix)
clusterID := strings.TrimSpace(trimmedPath)
if clusterID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_cluster_id", "Kubernetes cluster ID is required", nil)
return
}
cluster, err := h.getMonitor(r.Context()).UnhideKubernetesCluster(clusterID)
canonicalID, err := action(r.Context(), clusterID)
if err != nil {
writeErrorResponse(w, http.StatusNotFound, "k8s_cluster_not_found", err.Error(), nil)
return
@ -219,43 +228,35 @@ func (h *KubernetesAgentHandlers) HandleUnhideCluster(w http.ResponseWriter, r *
if err := utils.WriteJSONResponse(w, map[string]any{
"success": true,
"clusterId": cluster.ID,
"message": "Kubernetes cluster unhidden",
"clusterId": canonicalID,
"message": successMsg,
}); err != nil {
log.Error().Err(err).Msg("Failed to serialize kubernetes cluster unhide response")
log.Error().Err(err).Msg(logMsg)
}
}
// HandleUnhideCluster unhides a previously hidden kubernetes cluster.
func (h *KubernetesAgentHandlers) HandleUnhideCluster(w http.ResponseWriter, r *http.Request) {
h.handleClusterLifecycleAction(w, r, "/unhide", "Kubernetes cluster unhidden", "Failed to serialize kubernetes cluster unhide response",
func(ctx context.Context, clusterID string) (string, error) {
cluster, err := h.getMonitor(ctx).UnhideKubernetesCluster(clusterID)
if err != nil {
return "", err
}
return cluster.ID, nil
})
}
// HandleMarkPendingUninstall marks a cluster as pending uninstall.
func (h *KubernetesAgentHandlers) HandleMarkPendingUninstall(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPut {
writeErrorResponse(w, http.StatusMethodNotAllowed, "method_not_allowed", "Only PUT is allowed", nil)
return
}
trimmedPath := strings.TrimPrefix(r.URL.Path, "/api/agents/kubernetes/clusters/")
trimmedPath = strings.TrimSuffix(trimmedPath, "/pending-uninstall")
clusterID := strings.TrimSpace(trimmedPath)
if clusterID == "" {
writeErrorResponse(w, http.StatusBadRequest, "missing_cluster_id", "Kubernetes cluster ID is required", nil)
return
}
cluster, err := h.getMonitor(r.Context()).MarkKubernetesClusterPendingUninstall(clusterID)
if err != nil {
writeErrorResponse(w, http.StatusNotFound, "k8s_cluster_not_found", err.Error(), nil)
return
}
h.broadcastState(r.Context())
if err := utils.WriteJSONResponse(w, map[string]any{
"success": true,
"clusterId": cluster.ID,
"message": "Kubernetes cluster marked as pending uninstall",
}); err != nil {
log.Error().Err(err).Msg("Failed to serialize kubernetes cluster pending uninstall response")
}
h.handleClusterLifecycleAction(w, r, "/pending-uninstall", "Kubernetes cluster marked as pending uninstall", "Failed to serialize kubernetes cluster pending uninstall response",
func(ctx context.Context, clusterID string) (string, error) {
cluster, err := h.getMonitor(ctx).MarkKubernetesClusterPendingUninstall(clusterID)
if err != nil {
return "", err
}
return cluster.ID, nil
})
}
// HandleSetCustomDisplayName updates the custom display name for a kubernetes cluster.

113
internal/api/metadata_handlers_shared.go Normal file
View file

@ -0,0 +1,113 @@
package api
import (
"context"
"encoding/json"
"net/http"
"strings"
"github.com/rs/zerolog/log"
)
// handleMetadataGetRequest serves the GET flow shared by the guest and docker
// metadata handlers: the bare route returns the full metadata map (an empty
// object instead of null), a suffixed route returns that resource's metadata
// (a zero record instead of a 404).
func handleMetadataGetRequest[M any](
w http.ResponseWriter,
r *http.Request,
routePrefix string,
getAll func(ctx context.Context) map[string]*M,
get func(ctx context.Context, id string) *M,
emptyRecord func(id string) *M,
) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
path := r.URL.Path
// Handle both the bare route and its trailing-slash form.
if path == routePrefix || path == routePrefix+"/" {
// Get all metadata
w.Header().Set("Content-Type", "application/json")
allMeta := getAll(r.Context())
if allMeta == nil {
// Return empty object instead of null
json.NewEncoder(w).Encode(make(map[string]*M))
} else {
json.NewEncoder(w).Encode(allMeta)
}
return
}
// Get specific resource ID from path
resourceID := strings.TrimPrefix(path, routePrefix+"/")
w.Header().Set("Content-Type", "application/json")
if resourceID != "" {
meta := get(r.Context(), resourceID)
if meta == nil {
// Return empty metadata instead of 404
json.NewEncoder(w).Encode(emptyRecord(resourceID))
} else {
json.NewEncoder(w).Encode(meta)
}
} else {
// This shouldn't happen with current routing, but handle it anyway
http.Error(w, "Invalid request path", http.StatusBadRequest)
}
}
// handleMetadataUpdateRequest serves the PUT/POST flow shared by the guest
// and docker metadata handlers: decode a bounded body, validate the custom
// URL, persist, and echo the stored record.
func handleMetadataUpdateRequest[M any](
w http.ResponseWriter,
r *http.Request,
routePrefix string,
idRequiredMsg string,
idLogKey string,
saveErrLogMsg string,
updatedLogMsg string,
customURL func(*M) string,
set func(ctx context.Context, id string, meta *M) error,
) {
if r.Method != http.MethodPut && r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
resourceID := strings.TrimPrefix(r.URL.Path, routePrefix+"/")
if resourceID == "" || resourceID == "metadata" {
http.Error(w, idRequiredMsg, http.StatusBadRequest)
return
}
// Limit request body to 16KB to prevent memory exhaustion
r.Body = http.MaxBytesReader(w, r.Body, 16*1024)
var meta M
if err := json.NewDecoder(r.Body).Decode(&meta); err != nil {
http.Error(w, "Invalid request body", http.StatusBadRequest)
return
}
// Validate URL if provided
if errMsg := validateCustomURL(customURL(&meta)); errMsg != "" {
http.Error(w, errMsg, http.StatusBadRequest)
return
}
if err := set(r.Context(), resourceID, &meta); err != nil {
log.Error().Err(err).Str(idLogKey, resourceID).Msg(saveErrLogMsg)
http.Error(w, metadataSaveErrorMessage(err), http.StatusInternalServerError)
return
}
log.Info().Str(idLogKey, resourceID).Str("url", customURL(&meta)).Msg(updatedLogMsg)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(&meta)
}

128
internal/api/platform_connection_shared.go Normal file
View file

@ -0,0 +1,128 @@
package api
import (
"bytes"
"encoding/json"
"io"
"net/http"
"strings"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
)
// decodeOptionalInstanceRequest decodes an optional JSON body onto a copy of
// base. The results are (instance, decoded, ok): an empty body returns base
// with decoded=false, a malformed body writes a 400 response and returns
// ok=false. Shared by the TrueNAS and VMware connection handlers.
func decodeOptionalInstanceRequest[T any](w http.ResponseWriter, r *http.Request, base T) (T, bool, bool) {
var zero T
r.Body = http.MaxBytesReader(w, r.Body, 32*1024)
defer r.Body.Close()
body, err := io.ReadAll(r.Body)
if err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()})
return zero, false, false
}
if len(bytes.TrimSpace(body)) == 0 {
return base, false, true
}
instance := base
if err := json.Unmarshal(body, &instance); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()})
return zero, false, false
}
return instance, true, true
}
// platformConnectionUpdateSpec wires the per-platform pieces of the shared
// connection-update flow. The flow itself — locate by trimmed ID, decode with
// the stored record as fallback, normalize, preserve masked secrets,
// validate, persist, respond redacted — is platform-invariant policy;
// single-sourcing it means a new platform cannot forget masked-secret
// preservation.
type platformConnectionUpdateSpec[T any] struct {
notFoundCode string
saveFailCode string
saveFailMsg string
id func(T) string
decode func(http.ResponseWriter, *http.Request, T) (T, bool)
setID func(*T, string)
normalize func(*T)
preserve func(updated *T, stored T)
validate func(T) error
redacted func(T) any
save func([]T) error
}
// updatePlatformConnection runs the shared platform connection update flow
// over the loaded instances for the resolved connection ID.
func updatePlatformConnection[T any](w http.ResponseWriter, r *http.Request, connectionID string, instances []T, spec platformConnectionUpdateSpec[T]) {
index := -1
for i := range instances {
if strings.TrimSpace(spec.id(instances[i])) == connectionID {
index = i
break
}
}
if index < 0 {
writeErrorResponse(w, http.StatusNotFound, spec.notFoundCode, "Connection not found", nil)
return
}
instance, ok := spec.decode(w, r, instances[index])
if !ok {
return
}
spec.setID(&instance, connectionID)
spec.normalize(&instance)
spec.preserve(&instance, instances[index])
if err := spec.validate(instance); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil)
return
}
instances[index] = instance
if err := spec.save(instances); err != nil {
writeErrorResponse(w, http.StatusInternalServerError, spec.saveFailCode, spec.saveFailMsg, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, spec.redacted(instance))
}
// platformConnectionItemHandlers carries the per-platform handlers for the
// "/api/<platform>/connections/" item routes.
type platformConnectionItemHandlers struct {
test http.HandlerFunc
preview http.HandlerFunc
delete http.HandlerFunc
update http.HandlerFunc
}
// platformConnectionItemRoute builds the shared mux handler for platform
// connection item routes: POST .../test, POST .../preview, DELETE, PUT — all
// admin + settings:write gated. handlers is resolved per request so late
// handler wiring and nil-service checks keep working.
func (r *Router) platformConnectionItemRoute(unavailableCode, unavailableMsg string, handlers func() *platformConnectionItemHandlers) http.HandlerFunc {
return func(w http.ResponseWriter, req *http.Request) {
h := handlers()
if h == nil {
writeErrorResponse(w, http.StatusServiceUnavailable, unavailableCode, unavailableMsg, nil)
return
}
if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/test") {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.test))(w, req)
} else if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/preview") {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.preview))(w, req)
} else if req.Method == http.MethodDelete {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.delete))(w, req)
} else if req.Method == http.MethodPut {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, h.update))(w, req)
} else {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
}
}
}

50
internal/api/purchase_return_redemptions.go
View file

@ -3,9 +3,6 @@ package api
import (
"database/sql"
"fmt"
"net/url"
"os"
"path/filepath"
"strings"
"sync"
"time"
@ -59,39 +56,6 @@ type purchaseReturnRedemptionRecord struct {
func (s *purchaseReturnRedemptionStore) init() {
s.once.Do(func() {
dir := filepath.Clean(s.configDir)
if strings.TrimSpace(dir) == "" {
s.initErr = fmt.Errorf("configDir is required")
return
}
secretsDir := filepath.Join(dir, "secrets")
if err := os.MkdirAll(secretsDir, handoffPrivateDirPerm); err != nil {
s.initErr = fmt.Errorf("create purchase return secrets dir: %w", err)
return
}
if err := os.Chmod(secretsDir, handoffPrivateDirPerm); err != nil {
s.initErr = fmt.Errorf("chmod purchase return secrets dir: %w", err)
return
}
dbPath := filepath.Join(secretsDir, "purchase_return_redemptions.db")
dsn := dbPath + "?" + url.Values{
"_pragma": []string{
"busy_timeout(30000)",
"journal_mode(WAL)",
"synchronous(NORMAL)",
},
}.Encode()
db, err := sql.Open("sqlite", dsn)
if err != nil {
s.initErr = fmt.Errorf("open purchase return redemption db: %w", err)
return
}
db.SetMaxOpenConns(1)
db.SetMaxIdleConns(1)
db.SetConnMaxLifetime(0)
schema := `
CREATE TABLE IF NOT EXISTS purchase_return_redemptions (
portal_handoff_id TEXT NOT NULL,
@ -115,19 +79,11 @@ func (s *purchaseReturnRedemptionStore) init() {
CREATE INDEX IF NOT EXISTS idx_purchase_return_redemptions_status
ON purchase_return_redemptions(status);
`
if _, err := db.Exec(schema); err != nil {
_ = db.Close()
s.initErr = fmt.Errorf("init purchase return redemption schema: %w", err)
db, err := openHardenedSecretsDB(s.configDir, "purchase_return_redemptions.db", "purchase return", "purchase return redemption", schema)
if err != nil {
s.initErr = err
return
}
for _, path := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
if err := hardenPrivateFile(path, handoffPrivateFilePerm); err != nil {
_ = db.Close()
s.initErr = fmt.Errorf("harden purchase return redemption file permissions: %w", err)
return
}
}
s.db = db
})
}

66
internal/api/recovery_handlers.go
View file

@ -292,31 +292,25 @@ func parseBoolQuery(qs map[string][]string, key string) bool {
}
}
func (h *RecoveryHandlers) HandleListSeries(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
qs := r.URL.Query()
// parseRecoveryListPointsOptions parses the recovery points filter options
// shared by the points-series and points-facets handlers. It writes a 400
// response and returns false when a time bound is not RFC3339.
func parseRecoveryListPointsOptions(w http.ResponseWriter, qs url.Values) (recovery.ListPointsOptions, bool) {
var from, to *time.Time
if t, err := parseRFC3339QueryTime(qs.Get("from")); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_from", "Invalid from time; must be RFC3339", map[string]string{"error": err.Error()})
return
return recovery.ListPointsOptions{}, false
} else {
from = t
}
if t, err := parseRFC3339QueryTime(qs.Get("to")); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_to", "Invalid to time; must be RFC3339", map[string]string{"error": err.Error()})
return
return recovery.ListPointsOptions{}, false
} else {
to = t
}
tzOffsetMin := parseIntQuery(qs, "tzOffsetMinutes", 0)
opts := recovery.ListPointsOptions{
return recovery.ListPointsOptions{
Provider: parseRecoveryPlatformQuery(qs),
Kind: recovery.Kind(strings.TrimSpace(qs.Get("kind"))),
Mode: recovery.Mode(strings.TrimSpace(qs.Get("mode"))),
@ -332,8 +326,23 @@ func (h *RecoveryHandlers) HandleListSeries(w http.ResponseWriter, r *http.Reque
NamespaceLabel: strings.TrimSpace(firstNonEmpty(qs.Get("namespace"), qs.Get("namespaceLabel"))),
WorkloadOnly: strings.TrimSpace(qs.Get("scope")) == "workload" || parseBoolQuery(qs, "workloadOnly"),
Verification: strings.TrimSpace(firstNonEmpty(qs.Get("verification"), qs.Get("verified"))),
}, true
}
func (h *RecoveryHandlers) HandleListSeries(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
qs := r.URL.Query()
opts, ok := parseRecoveryListPointsOptions(w, qs)
if !ok {
return
}
tzOffsetMin := parseIntQuery(qs, "tzOffsetMinutes", 0)
var series []recovery.PointsSeriesBucket
if mock.IsMockEnabled() {
all := mock.CurrentFixtureGraph().RecoveryPoints()
@ -368,36 +377,9 @@ func (h *RecoveryHandlers) HandleListFacets(w http.ResponseWriter, r *http.Reque
qs := r.URL.Query()
var from, to *time.Time
if t, err := parseRFC3339QueryTime(qs.Get("from")); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_from", "Invalid from time; must be RFC3339", map[string]string{"error": err.Error()})
opts, ok := parseRecoveryListPointsOptions(w, qs)
if !ok {
return
} else {
from = t
}
if t, err := parseRFC3339QueryTime(qs.Get("to")); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_to", "Invalid to time; must be RFC3339", map[string]string{"error": err.Error()})
return
} else {
to = t
}
opts := recovery.ListPointsOptions{
Provider: parseRecoveryPlatformQuery(qs),
Kind: recovery.Kind(strings.TrimSpace(qs.Get("kind"))),
Mode: recovery.Mode(strings.TrimSpace(qs.Get("mode"))),
Outcome: recovery.Outcome(strings.TrimSpace(qs.Get("outcome"))),
ItemType: recovery.NormalizeRecoveryItemType(firstNonEmpty(qs.Get("itemType"), qs.Get("type"))),
SubjectResourceID: parseRecoveryItemResourceIDQuery(qs),
RollupID: strings.TrimSpace(qs.Get("rollupId")),
From: from,
To: to,
Query: strings.TrimSpace(firstNonEmpty(qs.Get("q"), qs.Get("query"))),
ClusterLabel: strings.TrimSpace(firstNonEmpty(qs.Get("cluster"), qs.Get("clusterLabel"))),
NodeHostLabel: strings.TrimSpace(firstNonEmpty(qs.Get("node"), qs.Get("nodeHost"), qs.Get("nodeHostLabel"))),
NamespaceLabel: strings.TrimSpace(firstNonEmpty(qs.Get("namespace"), qs.Get("namespaceLabel"))),
WorkloadOnly: strings.TrimSpace(qs.Get("scope")) == "workload" || parseBoolQuery(qs, "workloadOnly"),
Verification: strings.TrimSpace(firstNonEmpty(qs.Get("verification"), qs.Get("verified"))),
}
var facets recovery.PointsFacets

658
internal/api/router.go
View file

@ -1075,15 +1075,14 @@ func (r *Router) handleAgentWebSocket(w http.ResponseWriter, req *http.Request)
r.agentExecServer.HandleWebSocket(w, req)
}
func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Request) {
if r.configHandlers == nil {
http.Error(w, "Service unavailable", http.StatusServiceUnavailable)
return
}
// serveSetupTokenOrSettingsWrite gates a privileged settings endpoint behind
// either a valid setup token (setup scripts) or full authentication plus
// proxy-auth admin and settings:write scope. logLabel tags the unauthorized
// log line; nonAdminMsg is the proxy-auth denial log message.
func (r *Router) serveSetupTokenOrSettingsWrite(w http.ResponseWriter, req *http.Request, logLabel, nonAdminMsg string, delegate http.HandlerFunc) {
// Check setup token first (for setup scripts)
if r.isValidSetupTokenForRequest(req) {
r.configHandlers.HandleVerifyTemperatureSSH(w, req)
delegate(w, req)
return
}
@ -1094,7 +1093,7 @@ func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Req
Str("ip", req.RemoteAddr).
Str("path", req.URL.Path).
Str("method", req.Method).
Msg("Unauthorized access attempt (verify-temperature-ssh)")
Msg("Unauthorized access attempt (" + logLabel + ")")
if !authWriter.wrote {
writeAuthenticationRequired(w, req)
@ -1108,18 +1107,26 @@ func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Req
log.Warn().
Str("ip", GetClientIP(req)).
Str("username", username).
Msg("Non-admin user attempted verify-temperature-ssh")
Msg(nonAdminMsg)
http.Error(w, "Admin privileges required", http.StatusForbidden)
return
}
}
// Require admin session identity or settings:write token for privileged SSH probes.
// Require admin session identity or settings:write token for privileged writes.
if !ensureSettingsWriteScope(r.config, w, req) {
return
}
r.configHandlers.HandleVerifyTemperatureSSH(w, req)
delegate(w, req)
}
func (r *Router) handleVerifyTemperatureSSH(w http.ResponseWriter, req *http.Request) {
if r.configHandlers == nil {
http.Error(w, "Service unavailable", http.StatusServiceUnavailable)
return
}
r.serveSetupTokenOrSettingsWrite(w, req, "verify-temperature-ssh", "Non-admin user attempted verify-temperature-ssh", r.configHandlers.HandleVerifyTemperatureSSH)
}
// handleSSHConfig handles SSH config writes with setup token or API auth
@ -1128,46 +1135,7 @@ func (r *Router) handleSSHConfig(w http.ResponseWriter, req *http.Request) {
http.Error(w, "Service unavailable", http.StatusServiceUnavailable)
return
}
// Check setup token first (for setup scripts)
if r.isValidSetupTokenForRequest(req) {
r.systemSettingsHandler.HandleSSHConfig(w, req)
return
}
// Require authentication
authWriter := &responseCapture{ResponseWriter: w}
if !checkAuth(r.config, authWriter, req, false) {
log.Warn().
Str("ip", req.RemoteAddr).
Str("path", req.URL.Path).
Str("method", req.Method).
Msg("Unauthorized access attempt (ssh-config)")
if !authWriter.wrote {
writeAuthenticationRequired(w, req)
}
return
}
// Check admin privileges for proxy auth users
if r.config.ProxyAuthSecret != "" {
if valid, username, isAdmin := CheckProxyAuth(r.config, req); valid && !isAdmin {
log.Warn().
Str("ip", GetClientIP(req)).
Str("username", username).
Msg("Non-admin user attempted ssh-config update")
http.Error(w, "Admin privileges required", http.StatusForbidden)
return
}
}
// Require admin session identity or settings:write token for privileged SSH config writes.
if !ensureSettingsWriteScope(r.config, w, req) {
return
}
r.systemSettingsHandler.HandleSSHConfig(w, req)
r.serveSetupTokenOrSettingsWrite(w, req, "ssh-config", "Non-admin user attempted ssh-config update", r.systemSettingsHandler.HandleSSHConfig)
}
func extractSetupToken(req *http.Request) string {
@ -1693,6 +1661,69 @@ func (r *Router) StartPatrolForOrg(ctx context.Context, orgID string) {
}
}
// unifiedLifecycleFromAI converts patrol finding lifecycle events into their
// unified-store representation.
func unifiedLifecycleFromAI(events []ai.FindingLifecycleEvent) []unified.UnifiedFindingLifecycleEvent {
if len(events) == 0 {
return nil
}
out := make([]unified.UnifiedFindingLifecycleEvent, 0, len(events))
for _, e := range events {
out = append(out, unified.UnifiedFindingLifecycleEvent{
At: e.At,
Type: e.Type,
Message: e.Message,
From: e.From,
To: e.To,
Metadata: e.Metadata,
})
}
return out
}
// unifiedFindingFromAI converts a patrol ai.Finding into its unified-store
// representation. Shared by the live patrol callback and the startup sync of
// persisted findings.
func unifiedFindingFromAI(f *ai.Finding) *unified.UnifiedFinding {
return &unified.UnifiedFinding{
ID: f.ID,
Source: unified.SourceAIPatrol,
Severity: unified.UnifiedSeverity(f.Severity),
Category: unified.UnifiedCategory(f.Category),
ResourceID: f.ResourceID,
ResourceName: f.ResourceName,
ResourceType: f.ResourceType,
Node: f.Node,
Title: f.Title,
Description: f.Description,
Impact: f.Impact,
PreviousResolvedFixSummary: f.PreviousResolvedFixSummary,
Recommendation: f.Recommendation,
Evidence: f.Evidence,
DetectedAt: f.DetectedAt,
LastSeenAt: f.LastSeenAt,
ResolvedAt: f.ResolvedAt,
AutoResolved: f.AutoResolved,
InvestigationSessionID: f.InvestigationSessionID,
InvestigationStatus: f.InvestigationStatus,
InvestigationOutcome: f.InvestigationOutcome,
LastInvestigatedAt: f.LastInvestigatedAt,
InvestigationAttempts: f.InvestigationAttempts,
InvestigationRecord: f.InvestigationRecord,
LoopState: f.LoopState,
Lifecycle: unifiedLifecycleFromAI(f.Lifecycle),
RegressionCount: f.RegressionCount,
LastRegressionAt: f.LastRegressionAt,
AcknowledgedAt: f.AcknowledgedAt,
SnoozedUntil: f.SnoozedUntil,
DismissedReason: f.DismissedReason,
UserNote: f.UserNote,
Suppressed: f.Suppressed,
TimesRaised: f.TimesRaised,
RemindAt: f.RemindAt,
}
}
func (r *Router) startPatrolForContext(ctx context.Context, orgID string) bool {
if r == nil || r.aiSettingsHandler == nil {
return false
@ -1865,62 +1896,8 @@ func (r *Router) startPatrolForContext(ctx context.Context, orgID string) bool {
patrol := aiService.GetPatrolService()
if patrol != nil {
if unifiedStore := r.aiSettingsHandler.GetUnifiedStoreForOrg(orgID); unifiedStore != nil {
toUnifiedLifecycle := func(events []ai.FindingLifecycleEvent) []unified.UnifiedFindingLifecycleEvent {
if len(events) == 0 {
return nil
}
out := make([]unified.UnifiedFindingLifecycleEvent, 0, len(events))
for _, e := range events {
out = append(out, unified.UnifiedFindingLifecycleEvent{
At: e.At,
Type: e.Type,
Message: e.Message,
From: e.From,
To: e.To,
Metadata: e.Metadata,
})
}
return out
}
patrol.SetUnifiedFindingCallback(func(f *ai.Finding) bool {
// Convert ai.Finding to unified.UnifiedFinding
uf := &unified.UnifiedFinding{
ID: f.ID,
Source: unified.SourceAIPatrol,
Severity: unified.UnifiedSeverity(f.Severity),
Category: unified.UnifiedCategory(f.Category),
ResourceID: f.ResourceID,
ResourceName: f.ResourceName,
ResourceType: f.ResourceType,
Node: f.Node,
Title: f.Title,
Description: f.Description,
Impact: f.Impact,
PreviousResolvedFixSummary: f.PreviousResolvedFixSummary,
Recommendation: f.Recommendation,
Evidence: f.Evidence,
DetectedAt: f.DetectedAt,
LastSeenAt: f.LastSeenAt,
ResolvedAt: f.ResolvedAt,
AutoResolved: f.AutoResolved,
InvestigationSessionID: f.InvestigationSessionID,
InvestigationStatus: f.InvestigationStatus,
InvestigationOutcome: f.InvestigationOutcome,
LastInvestigatedAt: f.LastInvestigatedAt,
InvestigationAttempts: f.InvestigationAttempts,
InvestigationRecord: f.InvestigationRecord,
LoopState: f.LoopState,
Lifecycle: toUnifiedLifecycle(f.Lifecycle),
RegressionCount: f.RegressionCount,
LastRegressionAt: f.LastRegressionAt,
AcknowledgedAt: f.AcknowledgedAt,
SnoozedUntil: f.SnoozedUntil,
DismissedReason: f.DismissedReason,
UserNote: f.UserNote,
Suppressed: f.Suppressed,
TimesRaised: f.TimesRaised,
RemindAt: f.RemindAt,
}
uf := unifiedFindingFromAI(f)
_, isNew := unifiedStore.AddFromAI(uf)
// Publish a finding.created event to the agent SSE
// stream when the finding is new and not auto-dismissed
@ -2051,43 +2028,7 @@ func (r *Router) startPatrolForContext(ctx context.Context, orgID string) bool {
if f == nil {
continue
}
uf := &unified.UnifiedFinding{
ID: f.ID,
Source: unified.SourceAIPatrol,
Severity: unified.UnifiedSeverity(f.Severity),
Category: unified.UnifiedCategory(f.Category),
ResourceID: f.ResourceID,
ResourceName: f.ResourceName,
ResourceType: f.ResourceType,
Node: f.Node,
Title: f.Title,
Description: f.Description,
Impact: f.Impact,
PreviousResolvedFixSummary: f.PreviousResolvedFixSummary,
Recommendation: f.Recommendation,
Evidence: f.Evidence,
DetectedAt: f.DetectedAt,
LastSeenAt: f.LastSeenAt,
ResolvedAt: f.ResolvedAt,
AutoResolved: f.AutoResolved,
InvestigationSessionID: f.InvestigationSessionID,
InvestigationStatus: f.InvestigationStatus,
InvestigationOutcome: f.InvestigationOutcome,
LastInvestigatedAt: f.LastInvestigatedAt,
InvestigationAttempts: f.InvestigationAttempts,
InvestigationRecord: f.InvestigationRecord,
LoopState: f.LoopState,
Lifecycle: toUnifiedLifecycle(f.Lifecycle),
RegressionCount: f.RegressionCount,
LastRegressionAt: f.LastRegressionAt,
AcknowledgedAt: f.AcknowledgedAt,
SnoozedUntil: f.SnoozedUntil,
DismissedReason: f.DismissedReason,
UserNote: f.UserNote,
Suppressed: f.Suppressed,
TimesRaised: f.TimesRaised,
RemindAt: f.RemindAt,
}
uf := unifiedFindingFromAI(f)
// Copy resolution timestamp if resolved
if f.ResolvedAt != nil || f.AutoResolved {
now := time.Now()
@ -5471,101 +5412,10 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) {
currentTime := time.Now().UnixMilli() // JavaScript timestamp format
oldestTimestamp := currentTime
// Process VMs - batch-load historical data (1-2 SQL calls instead of N).
vmList := readState.VMs()
vmRequests := make([]monitoring.GuestChartRequest, 0, len(vmList))
for _, vm := range vmList {
if vm == nil {
continue
}
if vid := vm.SourceID(); vid != "" {
vmRequests = append(vmRequests, monitoring.GuestChartRequest{InMemoryKey: vid, SQLResourceID: vid})
}
}
vmBatchMetrics := monitor.GetGuestMetricsForChartBatch("vm", vmRequests, duration, infrastructureSummaryMetricOrder...)
for _, vm := range vmList {
if vm == nil {
continue
}
vid := vm.SourceID()
if vid == "" {
continue
}
chartData[vid] = make(VMChartData)
if batchMetrics, ok := vmBatchMetrics[vid]; ok {
for metricType, points := range batchMetrics {
if !sparklineMetrics[metricType] {
continue
}
chartData[vid][metricType] = make([]MetricPoint, len(points))
for i, point := range points {
ts := point.Timestamp.UnixMilli()
if ts < oldestTimestamp {
oldestTimestamp = ts
}
chartData[vid][metricType][i] = MetricPoint{
Timestamp: ts,
Value: point.Value,
}
}
}
}
if len(chartData[vid]["cpu"]) == 0 {
chartData[vid]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: vm.CPUPercent()}}
chartData[vid]["memory"] = []MetricPoint{{Timestamp: currentTime, Value: vm.MemoryPercent()}}
chartData[vid]["disk"] = []MetricPoint{{Timestamp: currentTime, Value: vm.DiskPercent()}}
chartData[vid]["netin"] = []MetricPoint{{Timestamp: currentTime, Value: vm.NetIn()}}
chartData[vid]["netout"] = []MetricPoint{{Timestamp: currentTime, Value: vm.NetOut()}}
}
}
// Process Containers - batch-load historical data (1-2 SQL calls instead of N).
ctList := readState.Containers()
ctRequests := make([]monitoring.GuestChartRequest, 0, len(ctList))
for _, ct := range ctList {
if ct == nil {
continue
}
if cid := ct.SourceID(); cid != "" {
ctRequests = append(ctRequests, monitoring.GuestChartRequest{InMemoryKey: cid, SQLResourceID: cid})
}
}
ctBatchMetrics := monitor.GetGuestMetricsForChartBatch("container", ctRequests, duration, infrastructureSummaryMetricOrder...)
for _, ct := range ctList {
if ct == nil {
continue
}
cid := ct.SourceID()
if cid == "" {
continue
}
chartData[cid] = make(VMChartData)
if batchMetrics, ok := ctBatchMetrics[cid]; ok {
for metricType, points := range batchMetrics {
if !sparklineMetrics[metricType] {
continue
}
chartData[cid][metricType] = make([]MetricPoint, len(points))
for i, point := range points {
ts := point.Timestamp.UnixMilli()
if ts < oldestTimestamp {
oldestTimestamp = ts
}
chartData[cid][metricType][i] = MetricPoint{
Timestamp: ts,
Value: point.Value,
}
}
}
}
if len(chartData[cid]["cpu"]) == 0 {
chartData[cid]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: ct.CPUPercent()}}
chartData[cid]["memory"] = []MetricPoint{{Timestamp: currentTime, Value: ct.MemoryPercent()}}
chartData[cid]["disk"] = []MetricPoint{{Timestamp: currentTime, Value: ct.DiskPercent()}}
chartData[cid]["netin"] = []MetricPoint{{Timestamp: currentTime, Value: ct.NetIn()}}
chartData[cid]["netout"] = []MetricPoint{{Timestamp: currentTime, Value: ct.NetOut()}}
}
}
// Process VMs and Containers - batch-load historical data (1-2 SQL calls
// per family instead of N).
oldestTimestamp = collectGuestChartData(monitor, "vm", readState.VMs(), duration, chartData, currentTime, oldestTimestamp)
oldestTimestamp = collectGuestChartData(monitor, "container", readState.Containers(), duration, chartData, currentTime, oldestTimestamp)
// Process Storage - batch-load historical data (1-2 SQL calls instead of N).
storageData := make(map[string]StorageChartData)
@ -5721,22 +5571,7 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) {
}
dockerData[responseKey] = make(VMChartData)
if batchMetrics, ok := dcBatchMetrics[request.SQLResourceID]; ok {
for metricType, points := range batchMetrics {
if !sparklineMetrics[metricType] {
continue
}
dockerData[responseKey][metricType] = make([]MetricPoint, len(points))
for i, point := range points {
ts := point.Timestamp.UnixMilli()
if ts < oldestTimestamp {
oldestTimestamp = ts
}
dockerData[responseKey][metricType][i] = MetricPoint{
Timestamp: ts,
Value: point.Value,
}
}
}
oldestTimestamp = fillChartSeriesFromBatch(dockerData[responseKey], batchMetrics, oldestTimestamp)
}
if len(dockerData[responseKey]["cpu"]) == 0 {
dockerData[responseKey]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: dc.CPUPercent()}}
@ -5771,22 +5606,7 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) {
}
dockerHostData[dhID] = make(VMChartData)
if batchMetrics, ok := dhBatchMetrics[dhID]; ok {
for metricType, points := range batchMetrics {
if !sparklineMetrics[metricType] {
continue
}
dockerHostData[dhID][metricType] = make([]MetricPoint, len(points))
for i, point := range points {
ts := point.Timestamp.UnixMilli()
if ts < oldestTimestamp {
oldestTimestamp = ts
}
dockerHostData[dhID][metricType][i] = MetricPoint{
Timestamp: ts,
Value: point.Value,
}
}
}
oldestTimestamp = fillChartSeriesFromBatch(dockerHostData[dhID], batchMetrics, oldestTimestamp)
}
if len(dockerHostData[dhID]["cpu"]) == 0 {
dockerHostData[dhID]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: dh.CPUPercent()}}
@ -5818,22 +5638,7 @@ func (r *Router) handleCharts(w http.ResponseWriter, req *http.Request) {
}
agentData[hID] = make(VMChartData)
if batchMetrics, ok := agentBatchMetrics[request.SQLResourceID]; ok {
for metricType, points := range batchMetrics {
if !sparklineMetrics[metricType] {
continue
}
agentData[hID][metricType] = make([]MetricPoint, len(points))
for i, point := range points {
ts := point.Timestamp.UnixMilli()
if ts < oldestTimestamp {
oldestTimestamp = ts
}
agentData[hID][metricType][i] = MetricPoint{
Timestamp: ts,
Value: point.Value,
}
}
}
oldestTimestamp = fillChartSeriesFromBatch(agentData[hID], batchMetrics, oldestTimestamp)
}
if len(agentData[hID]["cpu"]) == 0 {
agentData[hID]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: h.CPUPercent()}}
@ -6449,6 +6254,87 @@ func normalizeInfrastructureSummaryChartSeries(
// sparklineMetrics lists the metric types consumed by summary sparklines
// and density maps. Metrics not in this set are omitted to keep payloads small.
// guestChartSourceView is the guest view subset the infrastructure summary
// chart builder consumes from VMs and LXC containers.
type guestChartSourceView interface {
comparable
SourceID() string
CPUPercent() float64
MemoryPercent() float64
DiskPercent() float64
NetIn() float64
NetOut() float64
}
// collectGuestChartData batch-loads sparkline history for one proxmox guest
// family into chartData (1-2 SQL calls instead of N) and returns the updated
// oldest chart timestamp. Guests without history fall back to a single
// current-value point per metric.
func collectGuestChartData[V guestChartSourceView](
monitor *monitoring.Monitor,
storeType string,
guests []V,
duration time.Duration,
chartData map[string]VMChartData,
currentTime, oldestTimestamp int64,
) int64 {
var zero V
requests := make([]monitoring.GuestChartRequest, 0, len(guests))
for _, g := range guests {
if g == zero {
continue
}
if id := g.SourceID(); id != "" {
requests = append(requests, monitoring.GuestChartRequest{InMemoryKey: id, SQLResourceID: id})
}
}
batch := monitor.GetGuestMetricsForChartBatch(storeType, requests, duration, infrastructureSummaryMetricOrder...)
for _, g := range guests {
if g == zero {
continue
}
id := g.SourceID()
if id == "" {
continue
}
chartData[id] = make(VMChartData)
if batchMetrics, ok := batch[id]; ok {
oldestTimestamp = fillChartSeriesFromBatch(chartData[id], batchMetrics, oldestTimestamp)
}
if len(chartData[id]["cpu"]) == 0 {
chartData[id]["cpu"] = []MetricPoint{{Timestamp: currentTime, Value: g.CPUPercent()}}
chartData[id]["memory"] = []MetricPoint{{Timestamp: currentTime, Value: g.MemoryPercent()}}
chartData[id]["disk"] = []MetricPoint{{Timestamp: currentTime, Value: g.DiskPercent()}}
chartData[id]["netin"] = []MetricPoint{{Timestamp: currentTime, Value: g.NetIn()}}
chartData[id]["netout"] = []MetricPoint{{Timestamp: currentTime, Value: g.NetOut()}}
}
}
return oldestTimestamp
}
// fillChartSeriesFromBatch copies sparkline-eligible batch metric points
// into dst and returns the updated oldest chart timestamp. Shared by the
// per-family infrastructure summary chart loops.
func fillChartSeriesFromBatch(dst VMChartData, batchMetrics map[string][]monitoring.MetricPoint, oldestTimestamp int64) int64 {
for metricType, points := range batchMetrics {
if !sparklineMetrics[metricType] {
continue
}
dst[metricType] = make([]MetricPoint, len(points))
for i, point := range points {
ts := point.Timestamp.UnixMilli()
if ts < oldestTimestamp {
oldestTimestamp = ts
}
dst[metricType][i] = MetricPoint{
Timestamp: ts,
Value: point.Value,
}
}
}
return oldestTimestamp
}
var sparklineMetrics = map[string]bool{
"cpu": true,
"memory": true,
@ -8007,117 +7893,12 @@ func (r *Router) handleWorkloadsSummaryCharts(w http.ResponseWriter, req *http.R
}()
workloadsSummaryBatchWG.Wait()
for idx, vm := range vmList {
responseKey := vmResponseKeys[idx]
metricID := vmRequests[idx].SQLResourceID
guestCounts.Total++
if workloadSummaryStatusIsRunning("", vm.Status()) {
guestCounts.Running++
} else {
guestCounts.Stopped++
}
var guestSummaryPoints int
snapshots, guestSummaryPoints = appendGuestWorkloadSummaries(vmList, vmResponseKeys, vmRequests, vmBatchMetrics, currentTimeTime, &guestCounts, buckets, snapshots, &oldestTimestamp)
guestPointCount += guestSummaryPoints
snapshot := workloadsSummarySnapshot{
id: responseKey,
name: strings.TrimSpace(vm.Name()),
cpu: clampWorkloadPercent(vm.CPUPercent()),
memory: clampWorkloadPercent(vm.MemoryPercent()),
disk: clampWorkloadPercent(vm.DiskPercent()),
network: clampNonNegativeWorkloadValue(vm.NetIn() + vm.NetOut()),
}
if snapshot.name == "" {
snapshot.name = responseKey
}
metrics := vmBatchMetrics[metricID]
cpuPoints := metrics["cpu"]
if len(cpuPoints) == 0 {
cpuPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.CPUPercent()}}
}
memoryPoints := metrics["memory"]
if len(memoryPoints) == 0 {
memoryPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.MemoryPercent()}}
}
diskPoints := metrics["disk"]
if len(diskPoints) == 0 {
diskPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.DiskPercent()}}
}
netInPoints := metrics["netin"]
netOutPoints := metrics["netout"]
if len(netInPoints) == 0 && len(netOutPoints) == 0 {
netInPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.NetIn()}}
netOutPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: vm.NetOut()}}
}
networkPoints := mergeWorkloadNetworkPoints(netInPoints, netOutPoints)
snapshot.cpu = latestSummaryMetricValue(cpuPoints, snapshot.cpu, clampWorkloadPercent)
snapshot.memory = latestSummaryMetricValue(memoryPoints, snapshot.memory, clampWorkloadPercent)
snapshot.disk = latestSummaryMetricValue(diskPoints, snapshot.disk, clampWorkloadPercent)
snapshot.network = latestSummaryMetricValue(networkPoints, snapshot.network, clampNonNegativeWorkloadValue)
guestPointCount += appendWorkloadMetricPoints(buckets, cpuPoints, "cpu", &oldestTimestamp)
guestPointCount += appendWorkloadMetricPoints(buckets, memoryPoints, "memory", &oldestTimestamp)
guestPointCount += appendWorkloadMetricPoints(buckets, diskPoints, "disk", &oldestTimestamp)
guestPointCount += appendWorkloadMetricPoints(buckets, networkPoints, "network", &oldestTimestamp)
snapshots = append(snapshots, snapshot)
}
for idx, ct := range containerList {
responseKey := containerResponseKeys[idx]
metricID := containerRequests[idx].SQLResourceID
guestCounts.Total++
if workloadSummaryStatusIsRunning("", ct.Status()) {
guestCounts.Running++
} else {
guestCounts.Stopped++
}
snapshot := workloadsSummarySnapshot{
id: responseKey,
name: strings.TrimSpace(ct.Name()),
cpu: clampWorkloadPercent(ct.CPUPercent()),
memory: clampWorkloadPercent(ct.MemoryPercent()),
disk: clampWorkloadPercent(ct.DiskPercent()),
network: clampNonNegativeWorkloadValue(ct.NetIn() + ct.NetOut()),
}
if snapshot.name == "" {
snapshot.name = responseKey
}
metrics := containerBatchMetrics[metricID]
cpuPoints := metrics["cpu"]
if len(cpuPoints) == 0 {
cpuPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.CPUPercent()}}
}
memoryPoints := metrics["memory"]
if len(memoryPoints) == 0 {
memoryPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.MemoryPercent()}}
}
diskPoints := metrics["disk"]
if len(diskPoints) == 0 {
diskPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.DiskPercent()}}
}
netInPoints := metrics["netin"]
netOutPoints := metrics["netout"]
if len(netInPoints) == 0 && len(netOutPoints) == 0 {
netInPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.NetIn()}}
netOutPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: ct.NetOut()}}
}
networkPoints := mergeWorkloadNetworkPoints(netInPoints, netOutPoints)
snapshot.cpu = latestSummaryMetricValue(cpuPoints, snapshot.cpu, clampWorkloadPercent)
snapshot.memory = latestSummaryMetricValue(memoryPoints, snapshot.memory, clampWorkloadPercent)
snapshot.disk = latestSummaryMetricValue(diskPoints, snapshot.disk, clampWorkloadPercent)
snapshot.network = latestSummaryMetricValue(networkPoints, snapshot.network, clampNonNegativeWorkloadValue)
guestPointCount += appendWorkloadMetricPoints(buckets, cpuPoints, "cpu", &oldestTimestamp)
guestPointCount += appendWorkloadMetricPoints(buckets, memoryPoints, "memory", &oldestTimestamp)
guestPointCount += appendWorkloadMetricPoints(buckets, diskPoints, "disk", &oldestTimestamp)
guestPointCount += appendWorkloadMetricPoints(buckets, networkPoints, "network", &oldestTimestamp)
snapshots = append(snapshots, snapshot)
}
snapshots, guestSummaryPoints = appendGuestWorkloadSummaries(containerList, containerResponseKeys, containerRequests, containerBatchMetrics, currentTimeTime, &guestCounts, buckets, snapshots, &oldestTimestamp)
guestPointCount += guestSummaryPoints
for _, pod := range podList {
metricKey := kubernetesPodMetricIDFromView(pod)
@ -8350,6 +8131,91 @@ func (r *Router) handleWorkloadsSummaryCharts(w http.ResponseWriter, req *http.R
}
}
// guestWorkloadSummaryView is the guest view subset the workloads summary
// loop consumes from VMs and LXC containers.
type guestWorkloadSummaryView interface {
Status() unifiedresources.ResourceStatus
Name() string
CPUPercent() float64
MemoryPercent() float64
DiskPercent() float64
NetIn() float64
NetOut() float64
}
// appendGuestWorkloadSummaries accumulates workload-summary snapshots and
// chart points for one proxmox guest family (VMs or LXC containers),
// returning the extended snapshot slice and the number of points added.
func appendGuestWorkloadSummaries[V guestWorkloadSummaryView](
guests []V,
responseKeys []string,
requests []monitoring.GuestChartRequest,
batchMetrics map[string]map[string][]monitoring.MetricPoint,
currentTimeTime time.Time,
guestCounts *WorkloadsGuestCounts,
buckets map[int64]*workloadSummaryBuckets,
snapshots []workloadsSummarySnapshot,
oldestTimestamp *int64,
) ([]workloadsSummarySnapshot, int) {
added := 0
for idx, g := range guests {
responseKey := responseKeys[idx]
metricID := requests[idx].SQLResourceID
guestCounts.Total++
if workloadSummaryStatusIsRunning("", g.Status()) {
guestCounts.Running++
} else {
guestCounts.Stopped++
}
snapshot := workloadsSummarySnapshot{
id: responseKey,
name: strings.TrimSpace(g.Name()),
cpu: clampWorkloadPercent(g.CPUPercent()),
memory: clampWorkloadPercent(g.MemoryPercent()),
disk: clampWorkloadPercent(g.DiskPercent()),
network: clampNonNegativeWorkloadValue(g.NetIn() + g.NetOut()),
}
if snapshot.name == "" {
snapshot.name = responseKey
}
metrics := batchMetrics[metricID]
cpuPoints := metrics["cpu"]
if len(cpuPoints) == 0 {
cpuPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.CPUPercent()}}
}
memoryPoints := metrics["memory"]
if len(memoryPoints) == 0 {
memoryPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.MemoryPercent()}}
}
diskPoints := metrics["disk"]
if len(diskPoints) == 0 {
diskPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.DiskPercent()}}
}
netInPoints := metrics["netin"]
netOutPoints := metrics["netout"]
if len(netInPoints) == 0 && len(netOutPoints) == 0 {
netInPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.NetIn()}}
netOutPoints = []monitoring.MetricPoint{{Timestamp: currentTimeTime, Value: g.NetOut()}}
}
networkPoints := mergeWorkloadNetworkPoints(netInPoints, netOutPoints)
snapshot.cpu = latestSummaryMetricValue(cpuPoints, snapshot.cpu, clampWorkloadPercent)
snapshot.memory = latestSummaryMetricValue(memoryPoints, snapshot.memory, clampWorkloadPercent)
snapshot.disk = latestSummaryMetricValue(diskPoints, snapshot.disk, clampWorkloadPercent)
snapshot.network = latestSummaryMetricValue(networkPoints, snapshot.network, clampNonNegativeWorkloadValue)
added += appendWorkloadMetricPoints(buckets, cpuPoints, "cpu", oldestTimestamp)
added += appendWorkloadMetricPoints(buckets, memoryPoints, "memory", oldestTimestamp)
added += appendWorkloadMetricPoints(buckets, diskPoints, "disk", oldestTimestamp)
added += appendWorkloadMetricPoints(buckets, networkPoints, "network", oldestTimestamp)
snapshots = append(snapshots, snapshot)
}
return snapshots, added
}
func workloadSummaryStatusIsRunning(runtimeState string, status unifiedresources.ResourceStatus) bool {
switch strings.ToLower(strings.TrimSpace(runtimeState)) {
case "running", "online", "ok":

46
internal/api/router_routes_registration.go
View file

@ -289,24 +289,17 @@ func (r *Router) registerConfigSystemRoutes(updateHandlers *UpdateHandlers) {
}
})
r.mux.HandleFunc("/api/truenas/connections/", func(w http.ResponseWriter, req *http.Request) {
r.mux.HandleFunc("/api/truenas/connections/", r.platformConnectionItemRoute("truenas_unavailable", "TrueNAS service unavailable", func() *platformConnectionItemHandlers {
if r.trueNASHandlers == nil {
writeErrorResponse(w, http.StatusServiceUnavailable, "truenas_unavailable", "TrueNAS service unavailable", nil)
return
return nil
}
if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/test") {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandleTestSavedConnection))(w, req)
} else if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/preview") {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandlePreviewSavedConnection))(w, req)
} else if req.Method == http.MethodDelete {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandleDelete))(w, req)
} else if req.Method == http.MethodPut {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.trueNASHandlers.HandleUpdate))(w, req)
} else {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return &platformConnectionItemHandlers{
test: r.trueNASHandlers.HandleTestSavedConnection,
preview: r.trueNASHandlers.HandlePreviewSavedConnection,
delete: r.trueNASHandlers.HandleDelete,
update: r.trueNASHandlers.HandleUpdate,
}
})
}))
// VMware vCenter connection management
r.mux.HandleFunc("/api/vmware/connections", func(w http.ResponseWriter, req *http.Request) {
@ -351,24 +344,17 @@ func (r *Router) registerConfigSystemRoutes(updateHandlers *UpdateHandlers) {
}
})
r.mux.HandleFunc("/api/vmware/connections/", func(w http.ResponseWriter, req *http.Request) {
r.mux.HandleFunc("/api/vmware/connections/", r.platformConnectionItemRoute("vmware_unavailable", "VMware service unavailable", func() *platformConnectionItemHandlers {
if r.vmwareHandlers == nil {
writeErrorResponse(w, http.StatusServiceUnavailable, "vmware_unavailable", "VMware service unavailable", nil)
return
return nil
}
if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/test") {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandleTestSavedConnection))(w, req)
} else if req.Method == http.MethodPost && strings.HasSuffix(strings.Trim(req.URL.Path, "/"), "/preview") {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandlePreviewSavedConnection))(w, req)
} else if req.Method == http.MethodDelete {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandleDelete))(w, req)
} else if req.Method == http.MethodPut {
RequireAdmin(r.config, RequireScope(config.ScopeSettingsWrite, r.vmwareHandlers.HandleUpdate))(w, req)
} else {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return &platformConnectionItemHandlers{
test: r.vmwareHandlers.HandleTestSavedConnection,
preview: r.vmwareHandlers.HandlePreviewSavedConnection,
delete: r.vmwareHandlers.HandleDelete,
update: r.vmwareHandlers.HandleUpdate,
}
})
}))
// Config Profile Routes - Protected by Admin Auth, Settings Scope, and Pro License
// SECURITY: Require settings:write scope to prevent low-privilege tokens from modifying agent profiles

69
internal/api/truenas_handlers.go
View file

@ -1,10 +1,8 @@
package api
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"net/url"
"strings"
@ -188,6 +186,8 @@ func (h *TrueNASHandlers) HandleDelete(w http.ResponseWriter, r *http.Request) {
// HandleUpdate replaces a configured TrueNAS connection by ID while preserving
// unchanged masked secrets from the stored record.
//
//nolint:dupl // parallel platform wiring of the shared updatePlatformConnection flow; only platform symbols differ
func (h *TrueNASHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) {
if !h.featureEnabled(w) {
return
@ -214,36 +214,21 @@ func (h *TrueNASHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) {
return
}
index := -1
for i := range instances {
if strings.TrimSpace(instances[i].ID) == connectionID {
index = i
break
}
}
if index < 0 {
writeErrorResponse(w, http.StatusNotFound, "truenas_not_found", "Connection not found", nil)
return
}
instance, ok := decodeTrueNASInstanceRequest(w, r, instances[index])
if !ok {
return
}
instance.ID = connectionID
normalizeTrueNASInstance(&instance)
instance.PreserveMaskedSecrets(instances[index])
if err := instance.Validate(); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil)
return
}
instances[index] = instance
if err := persistence.SaveTrueNASConfig(instances); err != nil {
writeErrorResponse(w, http.StatusInternalServerError, "truenas_save_failed", "Failed to save TrueNAS configuration", map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, instance.Redacted())
updatePlatformConnection(w, r, connectionID, instances, platformConnectionUpdateSpec[config.TrueNASInstance]{
notFoundCode: "truenas_not_found",
saveFailCode: "truenas_save_failed",
saveFailMsg: "Failed to save TrueNAS configuration",
id: func(inst config.TrueNASInstance) string { return inst.ID },
decode: decodeTrueNASInstanceRequest,
setID: func(inst *config.TrueNASInstance, id string) { inst.ID = id },
normalize: normalizeTrueNASInstance,
preserve: func(updated *config.TrueNASInstance, stored config.TrueNASInstance) {
updated.PreserveMaskedSecrets(stored)
},
validate: func(inst config.TrueNASInstance) error { return inst.Validate() },
redacted: func(inst config.TrueNASInstance) any { return inst.Redacted() },
save: persistence.SaveTrueNASConfig,
})
}
// HandlePreviewConnection projects the monitored-system impact of a proposed
@ -535,25 +520,7 @@ func decodeOptionalTrueNASInstanceRequest(
r *http.Request,
base config.TrueNASInstance,
) (config.TrueNASInstance, bool, bool) {
r.Body = http.MaxBytesReader(w, r.Body, 32*1024)
defer r.Body.Close()
body, err := io.ReadAll(r.Body)
if err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()})
return config.TrueNASInstance{}, false, false
}
if len(bytes.TrimSpace(body)) == 0 {
return base, false, true
}
instance := base
if err := json.Unmarshal(body, &instance); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()})
return config.TrueNASInstance{}, false, false
}
return instance, true, true
return decodeOptionalInstanceRequest(w, r, base)
}
func (h *TrueNASHandlers) featureEnabled(w http.ResponseWriter) bool {

69
internal/api/vmware_handlers.go
View file

@ -1,10 +1,8 @@
package api
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"net/url"
"strings"
@ -202,6 +200,8 @@ func (h *VMwareHandlers) HandleDelete(w http.ResponseWriter, r *http.Request) {
// HandleUpdate replaces a configured VMware vCenter connection by ID while
// preserving unchanged masked secrets from the stored record.
//
//nolint:dupl // parallel platform wiring of the shared updatePlatformConnection flow; only platform symbols differ
func (h *VMwareHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) {
if !h.featureEnabled(w) {
return
@ -228,36 +228,21 @@ func (h *VMwareHandlers) HandleUpdate(w http.ResponseWriter, r *http.Request) {
return
}
index := -1
for i := range instances {
if strings.TrimSpace(instances[i].ID) == connectionID {
index = i
break
}
}
if index < 0 {
writeErrorResponse(w, http.StatusNotFound, "vmware_not_found", "Connection not found", nil)
return
}
instance, ok := decodeVMwareInstanceRequest(w, r, instances[index])
if !ok {
return
}
instance.ID = connectionID
normalizeVMwareInstance(&instance)
instance.PreserveMaskedSecrets(instances[index])
if err := instance.Validate(); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil)
return
}
instances[index] = instance
if err := persistence.SaveVMwareConfig(instances); err != nil {
writeErrorResponse(w, http.StatusInternalServerError, "vmware_save_failed", "Failed to save VMware configuration", map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, instance.Redacted())
updatePlatformConnection(w, r, connectionID, instances, platformConnectionUpdateSpec[config.VMwareVCenterInstance]{
notFoundCode: "vmware_not_found",
saveFailCode: "vmware_save_failed",
saveFailMsg: "Failed to save VMware configuration",
id: func(inst config.VMwareVCenterInstance) string { return inst.ID },
decode: decodeVMwareInstanceRequest,
setID: func(inst *config.VMwareVCenterInstance, id string) { inst.ID = id },
normalize: normalizeVMwareInstance,
preserve: func(updated *config.VMwareVCenterInstance, stored config.VMwareVCenterInstance) {
updated.PreserveMaskedSecrets(stored)
},
validate: func(inst config.VMwareVCenterInstance) error { return inst.Validate() },
redacted: func(inst config.VMwareVCenterInstance) any { return inst.Redacted() },
save: persistence.SaveVMwareConfig,
})
}
// HandlePreviewConnection projects the monitored-system impact of a proposed
@ -606,25 +591,7 @@ func decodeOptionalVMwareInstanceRequest(
r *http.Request,
base config.VMwareVCenterInstance,
) (config.VMwareVCenterInstance, bool, bool) {
r.Body = http.MaxBytesReader(w, r.Body, 32*1024)
defer r.Body.Close()
body, err := io.ReadAll(r.Body)
if err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()})
return config.VMwareVCenterInstance{}, false, false
}
if len(bytes.TrimSpace(body)) == 0 {
return base, false, true
}
instance := base
if err := json.Unmarshal(body, &instance); err != nil {
writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid request body", map[string]string{"error": err.Error()})
return config.VMwareVCenterInstance{}, false, false
}
return instance, true, true
return decodeOptionalInstanceRequest(w, r, base)
}
func (h *VMwareHandlers) featureEnabled(w http.ResponseWriter) bool {

60
pkg/pulsecli/actions.go
View file

@ -109,18 +109,26 @@ type actionExecutionResponse struct {
Audit unified.ActionAuditRecord `json:"audit"`
}
// actionAPIDefaults seeds the shared Pulse API URL and token defaults for
// action subcommands from the environment.
func actionAPIDefaults(deps *ActionsDeps) (apiURL, token string) {
apiURL = strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL"))
if apiURL == "" {
apiURL = defaultPulseAPIURL
}
return apiURL, strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN"))
}
func newActionsCmd(deps *ActionsDeps) *cobra.Command {
actionsCmd := &cobra.Command{
Use: "actions",
Short: "Inspect and plan governed Pulse actions",
}
apiURL, token := actionAPIDefaults(deps)
opts := actionPlanOptions{
APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")),
Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")),
}
if opts.APIURL == "" {
opts.APIURL = defaultPulseAPIURL
APIURL: apiURL,
Token: token,
}
planCmd := &cobra.Command{
@ -152,12 +160,10 @@ func newActionsCmd(deps *ActionsDeps) *cobra.Command {
}
func newActionCapabilitiesCmd(deps *ActionsDeps) *cobra.Command {
apiURL, token := actionAPIDefaults(deps)
opts := actionCapabilitiesOptions{
APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")),
Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")),
}
if opts.APIURL == "" {
opts.APIURL = defaultPulseAPIURL
APIURL: apiURL,
Token: token,
}
cmd := &cobra.Command{
@ -174,12 +180,10 @@ func newActionCapabilitiesCmd(deps *ActionsDeps) *cobra.Command {
}
func newActionDecisionCmd(deps *ActionsDeps) *cobra.Command {
apiURL, token := actionAPIDefaults(deps)
opts := actionDecisionOptions{
APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")),
Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")),
}
if opts.APIURL == "" {
opts.APIURL = defaultPulseAPIURL
APIURL: apiURL,
Token: token,
}
cmd := &cobra.Command{
@ -198,12 +202,10 @@ func newActionDecisionCmd(deps *ActionsDeps) *cobra.Command {
}
func newActionExecutionCmd(deps *ActionsDeps) *cobra.Command {
apiURL, token := actionAPIDefaults(deps)
opts := actionExecutionOptions{
APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")),
Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")),
}
if opts.APIURL == "" {
opts.APIURL = defaultPulseAPIURL
APIURL: apiURL,
Token: token,
}
cmd := &cobra.Command{
@ -220,15 +222,14 @@ func newActionExecutionCmd(deps *ActionsDeps) *cobra.Command {
return cmd
}
//nolint:dupl // parallel cobra subcommand registration; flags/help diverge per command
func newActionAuditCmd(deps *ActionsDeps) *cobra.Command {
apiURL, token := actionAPIDefaults(deps)
opts := actionAuditOptions{
APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")),
Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")),
APIURL: apiURL,
Token: token,
Limit: 100,
}
if opts.APIURL == "" {
opts.APIURL = defaultPulseAPIURL
}
cmd := &cobra.Command{
Use: "audit",
@ -245,15 +246,14 @@ func newActionAuditCmd(deps *ActionsDeps) *cobra.Command {
return cmd
}
//nolint:dupl // parallel cobra subcommand registration; flags/help diverge per command
func newActionEventsCmd(deps *ActionsDeps) *cobra.Command {
apiURL, token := actionAPIDefaults(deps)
opts := actionEventsOptions{
APIURL: strings.TrimSpace(actionGetenv(deps, "PULSE_API_URL")),
Token: strings.TrimSpace(actionGetenv(deps, "PULSE_API_TOKEN")),
APIURL: apiURL,
Token: token,
Limit: 100,
}
if opts.APIURL == "" {
opts.APIURL = defaultPulseAPIURL
}
cmd := &cobra.Command{
Use: "events",

28
pkg/pulsecli/actions_test.go
View file

@ -651,3 +651,31 @@ func newTestActionsRootCommand(env map[string]string) *cobra.Command {
},
)
}
func TestActionAPIDefaultsSeedsFromEnvironment(t *testing.T) {
deps := &ActionsDeps{Getenv: func(key string) string {
switch key {
case "PULSE_API_URL":
return " https://pulse.example.test "
case "PULSE_API_TOKEN":
return " token-123 "
}
return ""
}}
apiURL, token := actionAPIDefaults(deps)
if apiURL != "https://pulse.example.test" {
t.Fatalf("apiURL = %q, want trimmed env value", apiURL)
}
if token != "token-123" {
t.Fatalf("token = %q, want trimmed env value", token)
}
empty := &ActionsDeps{Getenv: func(string) string { return "" }}
apiURL, token = actionAPIDefaults(empty)
if apiURL != defaultPulseAPIURL {
t.Fatalf("apiURL = %q, want default %q when env is unset", apiURL, defaultPulseAPIURL)
}
if token != "" {
t.Fatalf("token = %q, want empty when env is unset", token)
}
}

4
scripts/release_control/subsystem_lookup_test.py
View file

@ -2860,8 +2860,8 @@ class SubsystemLookupTest(unittest.TestCase):
{
"heading": "## Shared Boundaries",
"path": "internal/api/access_control_handlers.go",
"line": 450,
"heading_line": 117,
"line": 452,
"heading_line": 119,
}
],
)