Bound config-fetch audit suppression without evicting recent agents

Keep at most 4096 remembered org/agent pairs and use an expiry index instead of repeated full-map scans. Audit unremembered overflow deliveries conservatively, preserving daily suppression for established agents, changed-config/token events and every failed fetch. Cover capacity, expiry, concurrency and the real handler contract.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-02 21:18:55 +01:00
parent e5d0843851
commit e225bf2abc
8 changed files with 334 additions and 20 deletions
+10 -1
View File
@@ -36,7 +36,16 @@ Pulse automatically captures the following events:
| `agent_profile_assigned` | Agent profile assignments | Profile `production` assigned to agent |
| `agent_profile_unassigned` | Agent profile removals | Profile removed from agent |
| `user_roles_updated` | RBAC role assignments changed | Updated roles for user jane: [operator] |
| `agent_config_fetch` | Every failed agent configuration fetch. A successful fetch is recorded on the agent's first delivery after Pulse starts, when its token or delivered configuration changes, and otherwise once a day; agents poll every minute, so repeat polls are not recorded | `agent_id=… token_id=… config=sha256:… reason=config_changed` |
| `agent_config_fetch` | Every failed agent configuration fetch. A successful fetch is normally recorded on the agent's first delivery after Pulse starts, when its token or delivered configuration changes, and otherwise once a day; unchanged minute polls are suppressed while the agent is remembered (see below) | `agent_id=… token_id=… config=sha256:… reason=config_changed` |
Config-fetch suppression remembers up to 4,096 organisation/agent pairs per
Pulse process. New agents can replace entries last recorded at least 24 hours
ago; a forgotten agent's next successful fetch is recorded again. If all slots
are recent, additional agents' successful fetches are recorded on every poll
with `reason=capacity` until a slot expires. This keeps memory bounded without
hiding access events or displacing recent agents' suppression. It does not
limit enrolment or configuration delivery, and every failed fetch is still
recorded. No existing audit rows are removed by this tracker.
Each event includes:
- **Timestamp** (UTC)
@@ -5336,8 +5336,11 @@ pending or unknown rather than in-sync.
Routine config polling is not a per-poll audit event:
`/api/agents/agent/{id}/config` audits every failed fetch, and a successful one
only on the agent's first delivery after start, a token or desired-config
change, or once a day (see the API contract).
normally on the agent's first delivery after start, a token or desired-config
change, or once a day (see the API contract). Suppression is bounded to 4,096
organisation/agent pairs. New keys replace only expired suppression entries;
overflow deliveries are audited with `reason=capacity`, without evicting
recent agents or restricting enrolment, config access, or command authority.
That same canonical /api/auto-register path must also complete the live
post-registration contract after persistence: it must trigger discovery refresh
and emit the canonical `node_auto_registered` WebSocket payload instead of
@@ -8776,8 +8776,20 @@ unchanged delivery last recorded 24 hours earlier. Agents poll every minute,
and auditing each poll made this event over 99.9% of audit rows. Recorded
successes carry `config=<desiredConfig hash>` and
`reason=first_since_start|token_changed|config_changed|daily` in their details.
The suppression tracker remembers at most 4,096 organisation/agent pairs.
An expiry index replaces only entries last audited at least 24 hours ago,
with no full-map scan on new-key admission. Recent entries are never evicted
by key churn, so their unchanged minute polls stay suppressed. If every slot
is recent, unremembered agents' successful fetches are recorded on every poll
with `reason=capacity` until an expired slot is available. This conservative
overflow can produce extra audit writes but must never suppress access events,
change config authorisation/delivery, or grow either map or expiry index beyond
the limit. Token, config and daily changes update the entry's expiry position;
concurrent and out-of-order observations preserve that index.
`TestAgentConfigFetchAuditsNewDeliveriesAndEveryFailure` and
`TestAgentConfigFetchAuditTrackerRecordsOnlyNewDeliveries` pin the rule.
`TestAgentConfigFetchAuditTrackerRecordsOnlyNewDeliveries` pin the rule;
`TestAgentConfigFetchAuditCapacityContract` and the tracker capacity, expiry,
churn and concurrency controls pin bounded state and conservative overflow.
Agent profile delete and unassign clients must now also route canonical `204`
success handling through shared allowed-status helpers in
`frontend-modern/src/api/responseUtils.ts` instead of open-coding local
@@ -2687,6 +2687,13 @@ vdev layout is reported` in
## Current State
The agent config-fetch audit suppression cache is bounded to 4,096
organisation/agent pairs, independently of persisted security audit history.
Expiry-index pruning forgets only suppression state, never audit rows or
recovery evidence. Unremembered overflow deliveries remain auditable with
`reason=capacity`; cache expiry, overflow, or restart is not a successful
storage operation or a reason to delete historical events.
### Container diagnostics shares private bootstrap transport (1 October 2026)
Recovery-adjacent diagnostics now reuse the canonical complete installer
+10 -1
View File
@@ -36,7 +36,16 @@ Pulse automatically captures the following events:
| `agent_profile_assigned` | Agent profile assignments | Profile `production` assigned to agent |
| `agent_profile_unassigned` | Agent profile removals | Profile removed from agent |
| `user_roles_updated` | RBAC role assignments changed | Updated roles for user jane: [operator] |
| `agent_config_fetch` | Every failed agent configuration fetch. A successful fetch is recorded on the agent's first delivery after Pulse starts, when its token or delivered configuration changes, and otherwise once a day; agents poll every minute, so repeat polls are not recorded | `agent_id=… token_id=… config=sha256:… reason=config_changed` |
| `agent_config_fetch` | Every failed agent configuration fetch. A successful fetch is normally recorded on the agent's first delivery after Pulse starts, when its token or delivered configuration changes, and otherwise once a day; unchanged minute polls are suppressed while the agent is remembered (see below) | `agent_id=… token_id=… config=sha256:… reason=config_changed` |
Config-fetch suppression remembers up to 4,096 organisation/agent pairs per
Pulse process. New agents can replace entries last recorded at least 24 hours
ago; a forgotten agent's next successful fetch is recorded again. If all slots
are recent, additional agents' successful fetches are recorded on every poll
with `reason=capacity` until a slot expires. This keeps memory bounded without
hiding access events or displacing recent agents' suppression. It does not
limit enrolment or configuration delivery, and every failed fetch is still
recorded. No existing audit rows are removed by this tracker.
Each event includes:
- **Timestamp** (UTC)
+57 -15
View File
@@ -1,6 +1,7 @@
package api
import (
"container/heap"
"sync"
"time"
)
@@ -10,9 +11,9 @@ import (
// continued access.
const agentConfigFetchAuditInterval = 24 * time.Hour
// maxAgentConfigFetchAudits bounds remembered deliveries. Beyond it, entries
// older than agentConfigFetchAuditInterval are dropped; they would be
// re-recorded on their next fetch anyway.
// maxAgentConfigFetchAudits bounds remembered deliveries. At capacity, only an
// expired entry can be replaced. An unremembered delivery is always audited;
// churn must not evict recent agents and turn their minute polls into writes.
const maxAgentConfigFetchAudits = 4096
// agentConfigFetchAuditTracker decides which successful agent config fetches
@@ -21,8 +22,9 @@ const maxAgentConfigFetchAudits = 4096
// agent), burying the logins and failures the log exists to show. Failed
// fetches are always audited and never pass through here.
type agentConfigFetchAuditTracker struct {
mu sync.Mutex
last map[agentConfigFetchAuditKey]agentConfigFetchAuditEntry
mu sync.Mutex
last map[agentConfigFetchAuditKey]*agentConfigFetchAuditEntry
oldest agentConfigFetchAuditHeap
}
type agentConfigFetchAuditKey struct {
@@ -31,26 +33,55 @@ type agentConfigFetchAuditKey struct {
}
type agentConfigFetchAuditEntry struct {
key agentConfigFetchAuditKey
tokenID string
configHash string
auditedAt time.Time
index int
}
// The expiry index avoids a full-map scan for every new key at capacity. It
// contains exactly the remembered entries, including after token/config/daily
// updates. Ordering by audit time also handles out-of-order observations.
type agentConfigFetchAuditHeap []*agentConfigFetchAuditEntry
func (h agentConfigFetchAuditHeap) Len() int { return len(h) }
func (h agentConfigFetchAuditHeap) Less(i, j int) bool {
return h[i].auditedAt.Before(h[j].auditedAt)
}
func (h agentConfigFetchAuditHeap) Swap(i, j int) {
h[i], h[j] = h[j], h[i]
h[i].index, h[j].index = i, j
}
func (h *agentConfigFetchAuditHeap) Push(value any) {
entry := value.(*agentConfigFetchAuditEntry)
entry.index = len(*h)
*h = append(*h, entry)
}
func (h *agentConfigFetchAuditHeap) Pop() any {
last := len(*h) - 1
entry := (*h)[last]
(*h)[last] = nil
*h = (*h)[:last]
entry.index = -1
return entry
}
func newAgentConfigFetchAuditTracker() *agentConfigFetchAuditTracker {
return &agentConfigFetchAuditTracker{last: make(map[agentConfigFetchAuditKey]agentConfigFetchAuditEntry)}
return &agentConfigFetchAuditTracker{last: make(map[agentConfigFetchAuditKey]*agentConfigFetchAuditEntry)}
}
// observe records a successful delivery and reports whether it is new audit
// information, with the reason: the agent's first delivery since startup, a
// different token or delivered config, or an unchanged delivery last audited
// at least agentConfigFetchAuditInterval ago. A nil tracker audits every
// delivery.
// at least agentConfigFetchAuditInterval ago. At capacity, unremembered agents
// are audited with reason "capacity" on every delivery until a slot expires.
// A nil tracker audits every delivery.
func (t *agentConfigFetchAuditTracker) observe(orgID, agentID, tokenID, configHash string, now time.Time) (string, bool) {
if t == nil {
return "", true
}
key := agentConfigFetchAuditKey{orgID: orgID, agentID: agentID}
entry := agentConfigFetchAuditEntry{tokenID: tokenID, configHash: configHash, auditedAt: now}
t.mu.Lock()
defer t.mu.Unlock()
@@ -68,13 +99,24 @@ func (t *agentConfigFetchAuditTracker) observe(orgID, agentID, tokenID, configHa
default:
return "", false
}
if !seen && len(t.last) >= maxAgentConfigFetchAudits {
for staleKey, stale := range t.last {
if now.Sub(stale.auditedAt) >= agentConfigFetchAuditInterval {
delete(t.last, staleKey)
}
}
if seen {
previous.tokenID, previous.configHash, previous.auditedAt = tokenID, configHash, now
heap.Fix(&t.oldest, previous.index)
return reason, true
}
if len(t.last) >= maxAgentConfigFetchAudits {
stale := t.oldest[0]
if now.Sub(stale.auditedAt) < agentConfigFetchAuditInterval {
return "capacity", true
}
heap.Pop(&t.oldest)
delete(t.last, stale.key)
}
if t.last == nil {
t.last = make(map[agentConfigFetchAuditKey]*agentConfigFetchAuditEntry)
}
entry := &agentConfigFetchAuditEntry{key: key, tokenID: tokenID, configHash: configHash, auditedAt: now}
heap.Push(&t.oldest, entry)
t.last[key] = entry
return reason, true
}
@@ -0,0 +1,183 @@
package api
import (
"fmt"
"sync"
"sync/atomic"
"testing"
"time"
)
func assertAgentConfigFetchAuditIndex(t *testing.T, tracker *agentConfigFetchAuditTracker) {
t.Helper()
if len(tracker.last) > maxAgentConfigFetchAudits || len(tracker.oldest) != len(tracker.last) {
t.Fatalf("audit state: map=%d index=%d limit=%d", len(tracker.last), len(tracker.oldest), maxAgentConfigFetchAudits)
}
for i, entry := range tracker.oldest {
if entry.index != i || tracker.last[entry.key] != entry {
t.Fatalf("expiry index lost entry %d", i)
}
if i > 0 && entry.auditedAt.Before(tracker.oldest[(i-1)/2].auditedAt) {
t.Fatalf("expiry index out of order at %d", i)
}
}
}
func fillAgentConfigFetchAuditTracker(t *testing.T, tracker *agentConfigFetchAuditTracker, start time.Time) {
t.Helper()
for i := 0; i < maxAgentConfigFetchAudits; i++ {
if reason, audit := tracker.observe("org", fmt.Sprint(i), "token", "hash", start.Add(time.Duration(i)*time.Second)); !audit || reason != "first_since_start" {
t.Fatalf("first delivery %d: audit=%v reason=%q", i, audit, reason)
}
}
assertAgentConfigFetchAuditIndex(t, tracker)
}
func TestAgentConfigFetchAuditTrackerBoundsRecentKeys(t *testing.T) {
tracker := newAgentConfigFetchAuditTracker()
start := time.Date(2026, 10, 2, 12, 0, 0, 0, time.UTC)
fillAgentConfigFetchAuditTracker(t, tracker, start)
now := start.Add(2 * time.Hour)
for i := 0; i < 2*maxAgentConfigFetchAudits; i++ {
if reason, audit := tracker.observe("overflow-org", fmt.Sprint(i), "token", "hash", now); !audit || reason != "capacity" {
t.Fatalf("overflow delivery %d: audit=%v reason=%q, want capacity audit", i, audit, reason)
}
}
// Overflow is fail-open for audit, not for state growth or cache churn.
// Repeat, token and config changes for an unremembered agent all remain
// visible; none displaces a remembered agent's useful daily suppression.
for _, step := range []struct{ token, hash string }{{"token", "hash"}, {"rotated", "hash"}, {"rotated", "changed"}} {
if reason, audit := tracker.observe("overflow-org", "0", step.token, step.hash, now); !audit || reason != "capacity" {
t.Fatalf("unremembered delivery: audit=%v reason=%q", audit, reason)
}
}
for minute := 0; minute < 1440; minute++ {
if reason, audit := tracker.observe("org", "0", "token", "hash", start.Add(time.Duration(minute)*time.Minute)); audit || reason != "" {
t.Fatalf("remembered minute poll %d was re-audited: %q", minute, reason)
}
}
assertAgentConfigFetchAuditIndex(t, tracker)
if len(tracker.last) != maxAgentConfigFetchAudits {
t.Fatal("recent agents were evicted by new-key churn")
}
}
func TestAgentConfigFetchAuditTrackerReusesOnlyExpiredSlots(t *testing.T) {
tracker := newAgentConfigFetchAuditTracker()
start := time.Date(2026, 10, 2, 12, 0, 0, 0, time.UTC)
fillAgentConfigFetchAuditTracker(t, tracker, start)
before := start.Add(agentConfigFetchAuditInterval - time.Nanosecond)
if reason, audit := tracker.observe("new-org", "0", "token", "hash", before); !audit || reason != "capacity" {
t.Fatal("a recent entry was replaced before its daily interval")
}
atExpiry := start.Add(agentConfigFetchAuditInterval)
if reason, audit := tracker.observe("new-org", "0", "token", "hash", atExpiry); !audit || reason != "first_since_start" {
t.Fatalf("expired slot was not reused: audit=%v reason=%q", audit, reason)
}
if tracker.last[agentConfigFetchAuditKey{"org", "0"}] != nil || tracker.last[agentConfigFetchAuditKey{"org", "1"}] == nil {
t.Fatal("replacement removed a recent entry instead of the oldest expired entry")
}
if _, audit := tracker.observe("new-org", "0", "token", "hash", atExpiry.Add(time.Minute)); audit {
t.Fatal("admitted overflow agent did not gain repeat-poll suppression")
}
// Forgetting a stale slot never suppresses that agent's next fetch.
if reason, audit := tracker.observe("org", "0", "token", "hash", atExpiry); !audit || reason != "capacity" {
t.Fatal("an expired, displaced agent lost its next audit")
}
assertAgentConfigFetchAuditIndex(t, tracker)
}
func TestAgentConfigFetchAuditTrackerReordersChangedEntries(t *testing.T) {
for _, change := range []string{"token", "config", "daily", "backward-clock"} {
t.Run(change, func(t *testing.T) {
tracker := newAgentConfigFetchAuditTracker()
start := time.Date(2026, 10, 2, 12, 0, 0, 0, time.UTC)
fillAgentConfigFetchAuditTracker(t, tracker, start)
agent, token, hash, reason := "0", "token", "hash", "token_changed"
changedAt := start.Add(2 * time.Hour)
oldest := "1"
switch change {
case "token":
token = "rotated"
case "config":
hash, reason = "changed", "config_changed"
case "daily":
changedAt, reason = start.Add(agentConfigFetchAuditInterval), "daily"
case "backward-clock":
agent, token, changedAt, oldest = "10", "rotated", start.Add(-time.Hour), "10"
}
if got, audit := tracker.observe("org", agent, token, hash, changedAt); !audit || got != reason {
t.Fatalf("changed delivery: audit=%v reason=%q, want %q", audit, got, reason)
}
assertAgentConfigFetchAuditIndex(t, tracker)
if tracker.oldest[0].key.agentID != oldest {
t.Fatalf("oldest agent=%s, want %s", tracker.oldest[0].key.agentID, oldest)
}
if _, audit := tracker.observe("other", "new", "token", "hash", start.Add(agentConfigFetchAuditInterval+time.Second)); !audit {
t.Fatal("replacement lost a delivery")
}
if tracker.last[agentConfigFetchAuditKey{"org", oldest}] != nil {
t.Fatal("replacement ignored reordered expiry")
}
assertAgentConfigFetchAuditIndex(t, tracker)
})
}
}
func TestAgentConfigFetchAuditTrackerConcurrentCapacityAndRestart(t *testing.T) {
tracker := newAgentConfigFetchAuditTracker()
start := time.Date(2026, 10, 2, 12, 0, 0, 0, time.UTC)
var audits atomic.Int32
var wg sync.WaitGroup
for worker := 0; worker < 16; worker++ {
wg.Add(1)
go func(worker int) {
defer wg.Done()
for i := 0; i < maxAgentConfigFetchAudits/8; i++ {
if _, audit := tracker.observe(fmt.Sprint(worker), fmt.Sprint(i), "token", "hash", start); audit {
audits.Add(1)
}
}
}(worker)
}
wg.Wait()
if got := audits.Load(); got != 2*maxAgentConfigFetchAudits {
t.Fatalf("concurrent first deliveries audited=%d, want %d", got, 2*maxAgentConfigFetchAudits)
}
assertAgentConfigFetchAuditIndex(t, tracker)
if len(tracker.last) != maxAgentConfigFetchAudits {
t.Fatal("concurrent delivery did not fill the bounded state")
}
// A restart forgets suppression, never the persisted audit trail.
restarted := newAgentConfigFetchAuditTracker()
for _, org := range []string{"org", "other-org"} {
if reason, audit := restarted.observe(org, "0", "token", "hash", start); !audit || reason != "first_since_start" {
t.Fatal("restart or organisation isolation suppressed a first delivery")
}
}
var zero agentConfigFetchAuditTracker
if _, audit := zero.observe("org", "0", "token", "hash", start); !audit {
t.Fatal("zero-value tracker suppressed a first delivery")
}
}
// Run with -benchtime=8192x for a fixed recent-key-churn comparison. This is
// an in-memory tracker measurement, not an installed write/CPU claim.
func BenchmarkAgentConfigFetchAuditTrackerRecentKeyChurn(b *testing.B) {
tracker := newAgentConfigFetchAuditTracker()
start := time.Date(2026, 10, 2, 12, 0, 0, 0, time.UTC)
for i := 0; i < maxAgentConfigFetchAudits; i++ {
tracker.observe("org", fmt.Sprint(i), "token", "hash", start)
}
keys := make([]string, b.N)
for i := range keys {
keys[i] = fmt.Sprint(i)
}
b.ReportAllocs()
b.ResetTimer()
for _, key := range keys {
tracker.observe("overflow-org", key, "token", "hash", start.Add(time.Hour))
}
b.StopTimer()
b.ReportMetric(float64(len(tracker.last)), "remembered-keys")
}
+49
View File
@@ -72,6 +72,55 @@ import (
tmock "github.com/stretchr/testify/mock"
)
func TestAgentConfigFetchAuditCapacityContract(t *testing.T) {
capture := &auditCaptureLogger{}
previousLogger, previousManager := audit.GetLogger(), GetTenantAuditManager()
audit.SetLogger(capture)
SetTenantAuditManager(nil)
t.Cleanup(func() {
audit.SetLogger(previousLogger)
SetTenantAuditManager(previousManager)
})
handler, monitor := newUnifiedAgentHandlers(t, nil)
hostID := seedUnifiedAgentHost(t, monitor)
monitorState(t, monitor).UpsertHost(models.Host{ID: hostID, Hostname: "node-1", TokenID: "runtime-token"})
for i := 0; i < maxAgentConfigFetchAudits; i++ {
handler.configFetchAudits.observe("other-org", fmt.Sprint(i), "token", "hash", time.Now())
}
for _, scopes := range [][]string{
{config.ScopeAgentConfigRead, config.ScopeAgentReport},
{config.ScopeAgentConfigRead, config.ScopeAgentReport},
{config.ScopeMonitoringRead},
} {
req := httptest.NewRequest(http.MethodGet, "/api/agents/agent/"+hostID+"/config", nil)
attachAPITokenRecord(req, &config.APITokenRecord{ID: "runtime-token", Scopes: scopes})
rec := httptest.NewRecorder()
handler.HandleConfig(rec, req)
if (rec.Code == http.StatusOK) != (scopes[0] == config.ScopeAgentConfigRead) {
t.Fatalf("capacity changed config authorisation: status=%d", rec.Code)
}
}
capture.mu.Lock()
defer capture.mu.Unlock()
successes, failures := 0, 0
for _, event := range capture.events {
if event.EventType != "agent_config_fetch" {
continue
}
if event.Success {
successes++
if !strings.Contains(event.Details, "reason=capacity") || !strings.Contains(event.Details, "config=sha256:") {
t.Fatal("unremembered config delivery lost its capacity reason or config hash")
}
} else {
failures++
}
}
if successes != 2 || failures != 1 || len(handler.configFetchAudits.last) != maxAgentConfigFetchAudits {
t.Fatalf("capacity audit contract: successes=%d failures=%d remembered=%d", successes, failures, len(handler.configFetchAudits.last))
}
}
func TestHandleVersionBuildIdentityContract(t *testing.T) {
oldBuildVersion := updates.BuildVersion
t.Cleanup(func() { updates.BuildVersion = oldBuildVersion })