supply chain gate: a vuln is a full stop now — admin signs for it or it doesn't ship
delivery plumbing that got us there: - acks clear on result-recorded, not command lifecycle status (no more 34-deep recycling) - timeouts, cancels, dropped acks/receipts, failed actions all land in history instead of dying on stdout - one shared closure-cleared predicate so auto-confirm and manual approve can't drift override waives the vuln call only and gets journaled; signing and hash verification stay non-negotiable.
This commit is contained in:
Fimeg
parent
1b6892ad2d
commit
dc3bfdf493
9 changed files with 363 additions and 73 deletions
|
|
@ -130,31 +130,40 @@ func (t *Tracker) IncrementRetry(commandID string) {
|
|||
}
|
||||
}
|
||||
|
||||
// Cleanup removes old or over-retried pending results
|
||||
func (t *Tracker) Cleanup() int {
|
||||
// DroppedResult describes a pending result-ack that Cleanup abandoned. The agent
|
||||
// only ever redelivers the command ID (not the result payload), so a result the
|
||||
// server never recorded can sit here unrecoverable until it ages out — and a drop
|
||||
// is the silent loss of an auditable event. Cleanup returns these so the caller
|
||||
// journals each one inward (ETHOS #1) rather than discarding it to /dev/null.
|
||||
type DroppedResult struct {
|
||||
CommandID string
|
||||
Reason string // "max_age" | "max_retries"
|
||||
RetryCount int
|
||||
AgeSeconds int
|
||||
}
|
||||
|
||||
// Cleanup removes old or over-retried pending results and returns what it dropped
|
||||
// so the loss can be recorded as history. Returns an empty slice when nothing aged out.
|
||||
func (t *Tracker) Cleanup() []DroppedResult {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
now := time.Now().UTC()
|
||||
removed := 0
|
||||
var dropped []DroppedResult
|
||||
|
||||
for id, result := range t.pending {
|
||||
// Remove if too old
|
||||
if now.Sub(result.SentAt) > t.maxAge {
|
||||
age := now.Sub(result.SentAt)
|
||||
switch {
|
||||
case age > t.maxAge:
|
||||
dropped = append(dropped, DroppedResult{CommandID: id, Reason: "max_age", RetryCount: result.RetryCount, AgeSeconds: int(age.Seconds())})
|
||||
delete(t.pending, id)
|
||||
removed++
|
||||
continue
|
||||
}
|
||||
|
||||
// Remove if retried too many times
|
||||
if result.RetryCount >= t.maxRetries {
|
||||
case result.RetryCount >= t.maxRetries:
|
||||
dropped = append(dropped, DroppedResult{CommandID: id, Reason: "max_retries", RetryCount: result.RetryCount, AgeSeconds: int(age.Seconds())})
|
||||
delete(t.pending, id)
|
||||
removed++
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
return removed
|
||||
return dropped
|
||||
}
|
||||
|
||||
// Stats returns statistics about pending acknowledgments
|
||||
|
|
|
|||
|
|
@ -363,6 +363,48 @@ func RunPollingLoop(loopCtx *LoopContext) error {
|
|||
}
|
||||
}
|
||||
|
||||
// Bound the delivery trackers and journal whatever they abandon. A dropped
|
||||
// result-ack or receipt is the silent loss of an auditable event — the agent
|
||||
// only ever redelivers the command ID, never the payload, so a result the
|
||||
// server never recorded is unrecoverable once it ages out. ETHOS #1: it
|
||||
// becomes history, not a stdout line. BufferEvent persists to disk and flushes
|
||||
// later, so the record survives the same network loss that caused the drop.
|
||||
if dropped := ctx.AckTracker.Cleanup(); len(dropped) > 0 {
|
||||
for _, d := range dropped {
|
||||
log.Printf("[WARNING] [agent] [acknowledgment] result_ack_dropped command_id=%s reason=%s retries=%d age_s=%d",
|
||||
d.CommandID, d.Reason, d.RetryCount, d.AgeSeconds)
|
||||
ctx.APIClient.BufferEvent(models.EventTypeError, models.SubtypeFailed, models.SeverityWarning,
|
||||
models.ComponentAgent,
|
||||
fmt.Sprintf("Abandoned delivery of command result %s (%s) — server never confirmed receipt", d.CommandID, d.Reason),
|
||||
map[string]interface{}{
|
||||
"kind": "result_ack",
|
||||
"command_id": d.CommandID,
|
||||
"reason": d.Reason,
|
||||
"retry_count": d.RetryCount,
|
||||
"age_seconds": d.AgeSeconds,
|
||||
})
|
||||
}
|
||||
if err := ctx.AckTracker.Save(); err != nil {
|
||||
log.Printf("[ERROR] [agent] [acknowledgment] save_failed error=%v", err)
|
||||
}
|
||||
}
|
||||
if dropped := ctx.ReceiptTracker.Cleanup(); len(dropped) > 0 {
|
||||
for _, d := range dropped {
|
||||
log.Printf("[WARNING] [agent] [receipt] receipt_dropped command_id=%s age_s=%d", d.CommandID, d.AgeSeconds)
|
||||
ctx.APIClient.BufferEvent(models.EventTypeError, models.SubtypeFailed, models.SeverityWarning,
|
||||
models.ComponentAgent,
|
||||
fmt.Sprintf("Abandoned receipt confirmation for command %s — server never acknowledged receipt before max-age", d.CommandID),
|
||||
map[string]interface{}{
|
||||
"kind": "receipt",
|
||||
"command_id": d.CommandID,
|
||||
"age_seconds": d.AgeSeconds,
|
||||
})
|
||||
}
|
||||
if err := ctx.ReceiptTracker.Save(); err != nil {
|
||||
log.Printf("[ERROR] [agent] [receipt] save_failed error=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Drop confirmed completions the server acknowledged via ReportLog.
|
||||
// This is the key fix for duplicate command rejections: if the server
|
||||
// has confirmed a command as completed, the agent should NOT reject it
|
||||
|
|
|
|||
|
|
@ -126,22 +126,32 @@ func (t *Tracker) Confirm(commandIDs []string) {
|
|||
}
|
||||
}
|
||||
|
||||
// Cleanup discards receipts older than maxAge. Returns the count removed. Should be
|
||||
// run periodically to bound disk and memory growth in the pathological case where
|
||||
// the server forgets a command id permanently.
|
||||
func (t *Tracker) Cleanup() int {
|
||||
// DroppedReceipt describes a receipt confirmation abandoned by Cleanup: the agent
|
||||
// received a command but the server never confirmed receipt, and the entry aged
|
||||
// out. A drop means the command's lifecycle silently diverged between the two
|
||||
// sides — the caller journals it inward (ETHOS #1) rather than discarding it.
|
||||
type DroppedReceipt struct {
|
||||
CommandID string
|
||||
AgeSeconds int
|
||||
}
|
||||
|
||||
// Cleanup discards receipts older than maxAge and returns what it dropped so the
|
||||
// loss can be recorded as history. Bounds disk/memory growth in the pathological
|
||||
// case where the server forgets a command id permanently. Returns an empty slice
|
||||
// when nothing aged out.
|
||||
func (t *Tracker) Cleanup() []DroppedReceipt {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
now := time.Now().UTC()
|
||||
removed := 0
|
||||
var dropped []DroppedReceipt
|
||||
for id, p := range t.pending {
|
||||
if now.Sub(p.ReceivedAt) > t.maxAge {
|
||||
if age := now.Sub(p.ReceivedAt); age > t.maxAge {
|
||||
dropped = append(dropped, DroppedReceipt{CommandID: id, AgeSeconds: int(age.Seconds())})
|
||||
delete(t.pending, id)
|
||||
removed++
|
||||
}
|
||||
}
|
||||
return removed
|
||||
return dropped
|
||||
}
|
||||
|
||||
// Len returns the current size of the pending set.
|
||||
|
|
|
|||
|
|
@ -951,8 +951,9 @@ func (h *AgentHandler) GetCommands(c *gin.Context) {
|
|||
}
|
||||
|
||||
log.Printf("DEBUG: Processing %d pending acknowledgments for agent %s: %v", len(metrics.PendingAcknowledgments), agentID, metrics.PendingAcknowledgments)
|
||||
// Verify which commands from agent's pending list have been recorded
|
||||
verified, err := h.commandQueries.VerifyCommandsCompleted(metrics.PendingAcknowledgments)
|
||||
// Ack every command whose result the server has durably recorded, regardless
|
||||
// of its lifecycle status — that receipt is all the agent's pending ack waits on.
|
||||
verified, err := h.commandQueries.VerifyResultsRecorded(metrics.PendingAcknowledgments)
|
||||
if err != nil {
|
||||
log.Printf("Warning: Failed to verify command acknowledgments for agent %s: %v", agentID, err)
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -497,6 +497,77 @@ func (h *UpdateHandler) GetPackageVersions(c *gin.Context) {
|
|||
})
|
||||
}
|
||||
|
||||
// supplyChainHold is the result of the manual-approval supply-chain gate: whether
|
||||
// approval must stop, whether the cause was an unverifiable closure (vs. a known
|
||||
// vuln), a human-readable reason, and the top-level vulnerabilities to echo back.
|
||||
type supplyChainHold struct {
|
||||
blocked bool
|
||||
unverified bool // closure OSV could not run — "trust the void", not a known vuln
|
||||
reason string
|
||||
vulns []services.VulnerabilityInfo
|
||||
}
|
||||
|
||||
// evaluateSupplyChainHold decides whether a manual approval must full-stop before
|
||||
// minting. A known vulnerability anywhere we have looked — the top-level package
|
||||
// (this call or persisted) or any artifact in the resolved closure — blocks. For
|
||||
// closure-gated ecosystems (dnf/apt), a closure OSV never managed to check also
|
||||
// blocks: minting over it would trust artifacts nothing vetted. Mirrors the
|
||||
// auto-confirm gate (orchestrator.closureCleared) via the shared models predicates.
|
||||
func (h *UpdateHandler) evaluateSupplyChainHold(update *models.UpdateState, freshVulns []services.VulnerabilityInfo) supplyChainHold {
|
||||
if len(freshVulns) > 0 || models.MetadataHasVulns(*update, "supply_chain_vulns") {
|
||||
return supplyChainHold{blocked: true, reason: "known vulnerabilities in the package", vulns: freshVulns}
|
||||
}
|
||||
if models.MetadataHasVulns(*update, "closure_vulns") {
|
||||
return supplyChainHold{blocked: true, reason: "known vulnerabilities in a dependency in the resolved closure", vulns: freshVulns}
|
||||
}
|
||||
if services.NeedsCapabilityGate(update.PackageType) && !models.ClosureChecked(*update) {
|
||||
return supplyChainHold{
|
||||
blocked: true,
|
||||
unverified: true,
|
||||
reason: "dependency closure could not be checked against OSV (service unreachable) — installing trusts unverified artifacts",
|
||||
vulns: freshVulns,
|
||||
}
|
||||
}
|
||||
return supplyChainHold{}
|
||||
}
|
||||
|
||||
// recordSupplyChainOverride journals an operator's deliberate decision to install
|
||||
// past the supply-chain gate. ETHOS #1: the void is trusted only by a named human,
|
||||
// on the record. The signed token minted afterward still binds the real artifact
|
||||
// hashes — this records that the vulnerability/verification judgment was waived,
|
||||
// not that authenticity was.
|
||||
func (h *UpdateHandler) recordSupplyChainOverride(update *models.UpdateState, hold supplyChainHold, operatorReason string) {
|
||||
subtype := "vuln_overridden"
|
||||
if hold.unverified {
|
||||
subtype = "unverified_overridden"
|
||||
}
|
||||
log.Printf("[SECURITY] [server] [supply_chain] gate_overridden id=%s pkg=%s version=%s cause=%q operator_reason=%q",
|
||||
update.ID, update.PackageName, update.AvailableVersion, hold.reason, operatorReason)
|
||||
event := &models.SystemEvent{
|
||||
ID: uuid.New(),
|
||||
AgentID: &update.AgentID,
|
||||
EventType: "supply_chain_override",
|
||||
EventSubtype: subtype,
|
||||
Severity: models.SeverityWarning,
|
||||
Component: models.ComponentSecurity,
|
||||
Message: fmt.Sprintf("Operator overrode the supply-chain gate for %s %s and approved install anyway: %s",
|
||||
update.PackageName, update.AvailableVersion, hold.reason),
|
||||
Metadata: map[string]interface{}{
|
||||
"package_name": update.PackageName,
|
||||
"package_type": update.PackageType,
|
||||
"version": update.AvailableVersion,
|
||||
"cause": hold.reason,
|
||||
"unverified": hold.unverified,
|
||||
"operator_reason": operatorReason,
|
||||
"approved_by": "admin",
|
||||
},
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}
|
||||
if err := h.agentQueries.CreateSystemEvent(event); err != nil {
|
||||
log.Printf("[WARNING] [server] [supply_chain] override_event_write_failed id=%s error=%v", update.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// ApproveUpdate marks an update as approved, with OSV.dev supply chain check
|
||||
func (h *UpdateHandler) ApproveUpdate(c *gin.Context) {
|
||||
idStr := c.Param("id")
|
||||
|
|
@ -506,6 +577,13 @@ func (h *UpdateHandler) ApproveUpdate(c *gin.Context) {
|
|||
return
|
||||
}
|
||||
|
||||
// Optional override body. Absent body = no override, the default and safe path.
|
||||
var req struct {
|
||||
OverrideSupplyChain bool `json:"override_supply_chain"`
|
||||
OverrideReason string `json:"override_reason"`
|
||||
}
|
||||
_ = c.ShouldBindJSON(&req)
|
||||
|
||||
// Look up update details for supply chain check
|
||||
update, err := h.updateQueries.GetUpdateByID(id)
|
||||
if err != nil {
|
||||
|
|
@ -570,6 +648,40 @@ func (h *UpdateHandler) ApproveUpdate(c *gin.Context) {
|
|||
}
|
||||
}
|
||||
|
||||
// Supply-chain full stop. A known vulnerability — top-level or anywhere in the
|
||||
// resolved dependency closure — or a closure OSV could not verify, must not
|
||||
// silently approve and mint a capability token. This is the hole the old
|
||||
// "sovereignty principle" left open: it let vulns through with only a warning.
|
||||
// Now it is a hard stop; only an explicit admin override ("I need this version
|
||||
// anyway") proceeds, and that decision is journaled to system_events. The override
|
||||
// waives the VULNERABILITY judgment only — it never touches the token's signature
|
||||
// or artifact-hash verification, which remain absolute (ETHOS: no skip-verify path).
|
||||
hold := h.evaluateSupplyChainHold(update, vulns)
|
||||
if hold.blocked {
|
||||
if !req.OverrideSupplyChain {
|
||||
log.Printf("[SECURITY] [server] [supply_chain] approval_blocked id=%s pkg=%s reason=%q unverified=%t",
|
||||
id, update.PackageName, hold.reason, hold.unverified)
|
||||
c.JSON(http.StatusConflict, gin.H{
|
||||
"error": "approval blocked by supply chain gate",
|
||||
"reason": hold.reason,
|
||||
"package": update.PackageName,
|
||||
"version": update.AvailableVersion,
|
||||
"unverified": hold.unverified,
|
||||
"vulnerabilities": vulns,
|
||||
"override_hint": "resubmit with override_supply_chain=true and a non-empty override_reason to install anyway",
|
||||
})
|
||||
return
|
||||
}
|
||||
if strings.TrimSpace(req.OverrideReason) == "" {
|
||||
c.JSON(http.StatusBadRequest, gin.H{
|
||||
"error": "supply chain override requires a reason",
|
||||
"reason": hold.reason,
|
||||
})
|
||||
return
|
||||
}
|
||||
h.recordSupplyChainOverride(update, hold, req.OverrideReason)
|
||||
}
|
||||
|
||||
// Proceed with approval (sovereignty principle: vulns + warn-mode age findings
|
||||
// don't block; only "block" enforcement of the age gate does, handled above).
|
||||
if len(vulns) > 0 || (!ageDecision.Unknown && ageDecision.WarnMessage != "") {
|
||||
|
|
@ -746,14 +858,20 @@ func (h *UpdateHandler) ReportLog(c *gin.Context) {
|
|||
return
|
||||
}
|
||||
|
||||
// Idempotency check: if command_id is provided and command is already
|
||||
// completed/failed, reject duplicate submissions (ETHOS #4).
|
||||
// Idempotency check: reject a resubmission only once the command is
|
||||
// FINALIZED (ETHOS #4). This deliberately keys off terminal status, not
|
||||
// "result already recorded" — those are different questions and must not be
|
||||
// conflated. A command can carry a result while still open: update_agent
|
||||
// reports "started" (which records a result) and then reports its terminal
|
||||
// success/failed under the same command_id. Rejecting on result-present would
|
||||
// 409 that finalizing report and the command would never close. (The pending
|
||||
// result-ack, by contrast, correctly clears on result-recorded — it only
|
||||
// needs to know the server received the report, not that it finalized.)
|
||||
if req.CommandID != "" {
|
||||
commandID, err := uuid.Parse(req.CommandID)
|
||||
if err == nil {
|
||||
command, err := h.commandQueries.GetCommandByID(commandID)
|
||||
if err == nil && command != nil {
|
||||
// Command already has a terminal status — reject duplicate
|
||||
if command.Status == models.CommandStatusCompleted || command.Status == models.CommandStatusFailed || command.Status == models.CommandStatusTimedOut {
|
||||
log.Printf("[INFO] [server] [updates] duplicate_log_rejected agent_id=%s command_id=%s status=%s",
|
||||
agentID, commandID, command.Status)
|
||||
|
|
@ -904,10 +1022,12 @@ func (h *UpdateHandler) ReportLog(c *gin.Context) {
|
|||
}
|
||||
}
|
||||
|
||||
// Emit a system_event for failed agent actions so the failure surfaces in the
|
||||
// events API and dashboard. Best-effort — log and continue on failure, don't
|
||||
// block the response.
|
||||
if validResult == "failed" && (req.Action == "update_agent" || req.Action == "verify_command") {
|
||||
// ETHOS #1: every failed agent action is an auditable error — journal it inward
|
||||
// so it surfaces in the events API / history, not only in update_logs. This was
|
||||
// previously gated to update_agent/verify_command, which left scan, dry-run, and
|
||||
// confirm_dependencies failures invisible to the operator. Best-effort — log and
|
||||
// continue on failure, don't block the response.
|
||||
if validResult == "failed" {
|
||||
failureReason := req.Stderr
|
||||
if failureReason == "" && req.ExitCode != 0 {
|
||||
failureReason = fmt.Sprintf("exit code %d", req.ExitCode)
|
||||
|
|
@ -916,11 +1036,18 @@ func (h *UpdateHandler) ReportLog(c *gin.Context) {
|
|||
failureReason = "agent reported failure"
|
||||
}
|
||||
|
||||
// Preserve the existing event type for update flows; everything else is a
|
||||
// generic command failure.
|
||||
eventType := models.EventTypeCommandFailed
|
||||
if req.Action == "update_agent" {
|
||||
eventType = "agent_update"
|
||||
}
|
||||
|
||||
message := services.RenderUpdateLog(req.Action, validResult, req.Stderr)
|
||||
event := &models.SystemEvent{
|
||||
ID: uuid.New(),
|
||||
AgentID: &agentID,
|
||||
EventType: "agent_update",
|
||||
EventType: eventType,
|
||||
EventSubtype: "failed",
|
||||
Severity: "error",
|
||||
Component: "agent",
|
||||
|
|
@ -1124,6 +1251,21 @@ func (h *UpdateHandler) ApproveUpdates(c *gin.Context) {
|
|||
}
|
||||
}
|
||||
|
||||
// Supply-chain full stop, same gate as single-approve. Bulk approval never
|
||||
// carries a blanket override — a known vuln or an unverifiable closure is
|
||||
// returned as blocked, and the operator must approve that one individually
|
||||
// with an explicit reason. Keeps "override everything at once" impossible.
|
||||
if hold := h.evaluateSupplyChainHold(update, vulns); hold.blocked {
|
||||
log.Printf("[SECURITY] [server] [supply_chain] bulk_approval_blocked id=%s pkg=%s reason=%q unverified=%t",
|
||||
id, update.PackageName, hold.reason, hold.unverified)
|
||||
blockedList = append(blockedList, blocked{
|
||||
UpdateID: idStr,
|
||||
PackageName: update.PackageName,
|
||||
Reason: hold.reason,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
needsMeta := len(vulns) > 0 || (!ageDecision.Unknown && ageDecision.WarnMessage != "")
|
||||
if needsMeta {
|
||||
if err := h.updateQueries.ApproveUpdateWithVulns(id, "admin", update.Metadata); err != nil {
|
||||
|
|
@ -1905,12 +2047,37 @@ func (h *UpdateHandler) CancelCommand(c *gin.Context) {
|
|||
return
|
||||
}
|
||||
|
||||
// Cancel the command
|
||||
// Capture the command before cancelling so the audit event carries agent + type.
|
||||
command, _ := h.commandQueries.GetCommandByID(id)
|
||||
|
||||
if err := h.commandQueries.CancelCommand(id); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("failed to cancel command: %v", err)})
|
||||
return
|
||||
}
|
||||
|
||||
// ETHOS #1: cancellation is an operator-driven state change against an agent's
|
||||
// command — journal it inward so it surfaces in the events API / history.
|
||||
if command != nil {
|
||||
event := &models.SystemEvent{
|
||||
ID: uuid.New(),
|
||||
AgentID: &command.AgentID,
|
||||
EventType: models.EventTypeCommandFailed,
|
||||
EventSubtype: "cancelled",
|
||||
Severity: models.SeverityWarning,
|
||||
Component: models.ComponentServer,
|
||||
Message: fmt.Sprintf("Command %s (%s) cancelled by operator", command.CommandType, command.ID),
|
||||
Metadata: map[string]interface{}{
|
||||
"command_id": command.ID.String(),
|
||||
"command_type": command.CommandType,
|
||||
"prior_status": command.Status,
|
||||
},
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}
|
||||
if err := h.agentQueries.CreateSystemEvent(event); err != nil {
|
||||
log.Printf("[WARNING] [server] [updates] system_event_write_failed command_id=%s error=%v", command.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"message": "command cancelled"})
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -341,12 +341,16 @@ func (q *CommandQueries) GetLatestRetryableCommandByUpdateID(updateID uuid.UUID)
|
|||
return &command, nil
|
||||
}
|
||||
|
||||
// CancelCommand marks a command as cancelled
|
||||
// CancelCommand marks a command as cancelled. It also records a result: a
|
||||
// cancelled command is resolved server-side, so the agent's pending result-ack
|
||||
// (which clears on result-recorded) should stop retrying rather than stranding
|
||||
// until its 24h max-age. COALESCE preserves any result the agent already reported.
|
||||
func (q *CommandQueries) CancelCommand(id uuid.UUID) error {
|
||||
now := time.Now().UTC()
|
||||
query := `
|
||||
UPDATE agent_commands
|
||||
SET status = 'cancelled', completed_at = $1
|
||||
SET status = 'cancelled', completed_at = $1,
|
||||
result = COALESCE(result, '{"cancelled": true, "reason": "operator cancelled command"}'::jsonb)
|
||||
WHERE id = $2 AND status IN ('pending', 'sent')
|
||||
`
|
||||
_, err := q.db.Exec(query, now, id)
|
||||
|
|
@ -660,9 +664,18 @@ func (q *CommandQueries) GetStuckCommands(agentID uuid.UUID, olderThan time.Dura
|
|||
return commands, err
|
||||
}
|
||||
|
||||
// VerifyCommandsCompleted checks which command IDs from the provided list have been completed or failed
|
||||
// Returns the list of command IDs that have been successfully recorded (completed or failed status)
|
||||
func (q *CommandQueries) VerifyCommandsCompleted(commandIDs []string) ([]string, error) {
|
||||
// VerifyResultsRecorded returns the subset of the given command IDs whose result
|
||||
// the server has durably stored. The pending ack exists for one thing: at-least-once
|
||||
// delivery of the agent's result. Every ReportLog path writes that result onto the
|
||||
// command row, so a non-null result is the receipt — once it's set, the agent can
|
||||
// stop resending. A command with no result yet stays unacked and the agent keeps
|
||||
// retrying, which is exactly what at-least-once requires.
|
||||
//
|
||||
// This deliberately does not key off command lifecycle status. Doing so stranded
|
||||
// acks whenever a command settled into a status the agent's result didn't drive
|
||||
// (progress reports, cancelled/archived rows) — they recycled every poll until the
|
||||
// agent's 24h max-age finally dropped them.
|
||||
func (q *CommandQueries) VerifyResultsRecorded(commandIDs []string) ([]string, error) {
|
||||
if len(commandIDs) == 0 {
|
||||
return []string{}, nil
|
||||
}
|
||||
|
|
@ -701,7 +714,7 @@ func (q *CommandQueries) VerifyCommandsCompleted(commandIDs []string) ([]string,
|
|||
SELECT id
|
||||
FROM agent_commands
|
||||
WHERE id::text = ANY(%s)
|
||||
AND status IN ('completed', 'failed', 'timed_out')
|
||||
AND result IS NOT NULL
|
||||
`, fmt.Sprintf("ARRAY[%s]", strings.Join(placeholders, ",")))
|
||||
|
||||
var completedUUIDs []uuid.UUID
|
||||
|
|
|
|||
53
server/internal/models/supply_chain_gate.go
Normal file
53
server/internal/models/supply_chain_gate.go
Normal file
|
|
@ -0,0 +1,53 @@
|
|||
package models
|
||||
|
||||
import "strings"
|
||||
|
||||
// Supply-chain gate predicates. These are the single source of truth for "is this
|
||||
// package's supply-chain posture clean enough to mint a capability token over it?"
|
||||
// Both the automatic confirmation sweep (orchestrator) and the manual operator
|
||||
// approval path (handlers) call them, so the two enforcement points can never drift.
|
||||
|
||||
// MetadataHasVulns reports whether metadata[key] records a non-empty vulnerability
|
||||
// set. An unexpected shape is treated as vulnerable, conservatively.
|
||||
func MetadataHasVulns(pkg UpdateState, key string) bool {
|
||||
if pkg.Metadata == nil {
|
||||
return false
|
||||
}
|
||||
v, ok := pkg.Metadata[key]
|
||||
if !ok || v == nil {
|
||||
return false
|
||||
}
|
||||
switch t := v.(type) {
|
||||
case []interface{}:
|
||||
return len(t) > 0
|
||||
case string:
|
||||
s := strings.TrimSpace(t)
|
||||
return s != "" && s != "[]" && s != "null"
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// ClosureChecked reports whether the resolved dependency closure was OSV-checked at
|
||||
// all (closure_checked_at present and non-empty). False means OSV could not vet the
|
||||
// closure — the "trust the void" case the operator must consciously accept.
|
||||
func ClosureChecked(pkg UpdateState) bool {
|
||||
if pkg.Metadata == nil {
|
||||
return false
|
||||
}
|
||||
v, ok := pkg.Metadata["closure_checked_at"]
|
||||
if !ok || v == nil {
|
||||
return false
|
||||
}
|
||||
if s, isStr := v.(string); isStr && strings.TrimSpace(s) == "" {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// ClosureCleared reports whether the resolved dependency closure was OSV-checked
|
||||
// AND came back clean. A token is auto-minted over the closure only when this holds;
|
||||
// the manual path requires an explicit operator override to proceed without it.
|
||||
func ClosureCleared(pkg UpdateState) bool {
|
||||
return ClosureChecked(pkg) && !MetadataHasVulns(pkg, "closure_vulns")
|
||||
}
|
||||
|
|
@ -67,30 +67,10 @@ func shouldAutoApprove(pkg models.UpdateState, maxSeverity string) bool {
|
|||
// hasBlockingVulns reports whether the package metadata records supply-chain
|
||||
// vulnerabilities on the top-level package. Auto-approval defers to the operator
|
||||
// whenever vulns are present, regardless of the severity ceiling — the vuln
|
||||
// review is a human call.
|
||||
// review is a human call. Delegates to the shared gate predicate so auto-confirm
|
||||
// and the manual approval path enforce the same rule.
|
||||
func hasBlockingVulns(pkg models.UpdateState) bool {
|
||||
return hasVulns(pkg, "supply_chain_vulns")
|
||||
}
|
||||
|
||||
// hasVulns reports whether metadata[key] records a non-empty vulnerability set.
|
||||
// An unexpected shape is treated as vulnerable, conservatively.
|
||||
func hasVulns(pkg models.UpdateState, key string) bool {
|
||||
if pkg.Metadata == nil {
|
||||
return false
|
||||
}
|
||||
v, ok := pkg.Metadata[key]
|
||||
if !ok || v == nil {
|
||||
return false
|
||||
}
|
||||
switch t := v.(type) {
|
||||
case []interface{}:
|
||||
return len(t) > 0
|
||||
case string:
|
||||
s := strings.TrimSpace(t)
|
||||
return s != "" && s != "[]" && s != "null"
|
||||
default:
|
||||
return true
|
||||
}
|
||||
return models.MetadataHasVulns(pkg, "supply_chain_vulns")
|
||||
}
|
||||
|
||||
// closureCleared reports whether the resolved dependency closure has been
|
||||
|
|
@ -100,15 +80,5 @@ func hasVulns(pkg models.UpdateState, key string) bool {
|
|||
// token is never minted over transitive artifacts OSV did not vet. A closure
|
||||
// carrying vulns is left for an operator regardless of the severity ceiling.
|
||||
func closureCleared(pkg models.UpdateState) bool {
|
||||
if pkg.Metadata == nil {
|
||||
return false
|
||||
}
|
||||
v, ok := pkg.Metadata["closure_checked_at"]
|
||||
if !ok || v == nil {
|
||||
return false
|
||||
}
|
||||
if s, isStr := v.(string); isStr && strings.TrimSpace(s) == "" {
|
||||
return false
|
||||
}
|
||||
return !hasVulns(pkg, "closure_vulns")
|
||||
return models.ClosureCleared(pkg)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -210,6 +210,31 @@ func (ts *TimeoutService) timeoutCommand(command *models.AgentCommand) error {
|
|||
// Don't return error here as the main timeout operation succeeded
|
||||
}
|
||||
|
||||
// ETHOS #1: a timed-out command is an auditable delivery failure. update_logs
|
||||
// alone does not surface in the events API / history pane (reconcileAgentUpdates
|
||||
// in this same service writes system_events; this path must too). Best-effort.
|
||||
if ts.agentQueries != nil {
|
||||
event := &models.SystemEvent{
|
||||
ID: uuid.New(),
|
||||
AgentID: &command.AgentID,
|
||||
EventType: models.EventTypeCommandFailed,
|
||||
EventSubtype: "timed_out",
|
||||
Severity: models.SeverityWarning,
|
||||
Component: models.ComponentServer,
|
||||
Message: fmt.Sprintf("Command %s (%s) timed out after %v with no agent result", command.CommandType, command.ID, appliedTimeout),
|
||||
Metadata: map[string]interface{}{
|
||||
"command_id": command.ID.String(),
|
||||
"command_type": command.CommandType,
|
||||
"prior_status": command.Status,
|
||||
"timeout": appliedTimeout.String(),
|
||||
},
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}
|
||||
if err := ts.agentQueries.CreateSystemEvent(event); err != nil {
|
||||
log.Printf("[WARNING] [server] [timeout] system_event_write_failed command_id=%s error=%v", command.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("Successfully timed out command %s", command.ID)
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue