Files
Privatae 01889e2a48
ci / personal-data (push) Successful in 4s
ci / test (push) Successful in 4m5s
ci / fuzz (push) Successful in 4m10s
bastion 0.58.0
2026-10-29 15:54:51 +00:00

1290 lines
54 KiB
Go

package main
// machinestore.go — the machine's own record, on the machine, for the host to
// read.
//
// WHAT WAS MISSING. The detection plane has computed findings every pass since
// 0.30.0, and they existed nowhere afterwards: `observe:delta` recomputes them
// on demand and the beat carries counts (v146's `detect` column). So the host
// could answer "how many" and could never answer "which ones", and an operator
// reading a machine an hour later saw a fresh pass, not what happened at 03:00.
// Nothing kept the evidence.
//
// This is the keeping. Findings and the alert summaries the box pulled, written
// as files under one root that
// the anchor host reads through `dcn` like any other machine store — a governed
// range-read against a store the user can see and revoke, never a push into the
// host and never a new lane.
//
// THE ACCOUNTING HALF OF AAA IS NOT COMPLETE HERE. `audit_fetch`
// records what the AI ASKED this box to do; what it SAW is op stdout, which is
// still returned down the lane and dropped. That shelf needs the op lane wired
// to a writer and is a later change — said plainly because a store that holds
// two thirds of the record must not be described as holding all of it.
//
// ── A JOURNAL OF TRANSITIONS, NOT A LOG OF PASSES ───────────────────────────
//
// The obvious build appends this pass's findings every pass. That is wrong in a
// way that only shows up in production: a finding is PINNED until a human
// resolves it (deltarun.go:270-273 states the pin for the binary event; the same rule governs findings generally), and the pass
// runs every 60 seconds. One unresolved finding is then 1,440 identical lines a
// day, and the file whose job is to say what happened is mostly a clock.
//
// So a line is written when a finding's state CHANGES:
//
// opened this box is reporting this fingerprint for the first time
// changed same fingerprint, different material content (level, tier, reason,
// endpoints) — a notice that became a critical is not the same finding
// closed it was open and this pass did not report it
//
// Replaying the journal in order reconstructs the open set exactly, which is
// what makes it safe to write only the edges. `closed` is NOT the same claim as
// "resolved" and the host must not read it as one: the overwhelmingly common
// reason a finding stops being reported is a retake of the accounting, and
// oversight.py:39-44 makes the same distinction on its side.
//
// ── WHAT IS DELIBERATELY NOT HERE ───────────────────────────────────────────
//
// No content from the user's files, ever. Findings name executables, units and
// argv shapes — the same material the beat's `agents` scan already carries, and
// gated the same way. This store adds no new class of disclosure; it adds
// persistence to a class that already travelled.
//
// The op-id registry deltanoise.go:88-97 wants ("which op started this") is
// still not here. It needs the op lane to record its launches, which is a
// change in mgmt_audit.go, and it is not made by this file — naming it beats
// implying the `bastion-child` filter can say more than "ours".
import (
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
)
// storeRoot is the DCN machine store's root: a directory INSIDE the bastion's
// own state dir, not beside it. The host is granted this path and nothing
// above it, so a read that escapes the root escapes into the bastion's own
// state — pairings, grants, the TOTP secret. dcn.go's path handling is what
// stops that and it is not re-implemented here; the nesting is the second wall,
// so a mistake in the first one lands somewhere harmless rather than on the
// key material.
const storeDirName = "store"
// The three shelves the machine store keeps.
const (
storeFindings = "findings"
storeAlerts = "alerts"
// storeObserve holds what each observe op RETURNED, one shelf per op:
// observe/<op>/<date>.jsonl. Reinstated with a call site this time — the
// round-1 review removed an unreachable writer with an unbounded
// directory behind it, and this one rides storeAppend's cap and seal like
// the other two shelves.
storeObserve = "observe"
)
// storeRetention is how long a day-file is kept. Thirty days is the plan's
// number and it is the one an operator investigating an incident needs: the
// question "was this here last month" is common and "was this here last year"
// is not, and the difference is a bounded directory on somebody's laptop.
const storeRetention = 30 * 24 * time.Hour
// storeDayCap bounds ONE day-file. Derived, not picked, and the derivation is
// stated with the measurement that backs it: the test fixture line is 339 bytes
// and carries NO Evidence, Was, PIDs or Endpoints, so it is the SHORT end, not
// the long one — an earlier version of this comment had that backwards and
// quoted a capacity it could not support. A real finding with evidence runs to
// roughly 1-2 KB, so 4 MiB is on the order of 2,000-4,000 transitions in a day.
// A machine producing more than that is not producing findings, it is producing
// a storm, and the storm is itself the signal — which is why hitting the cap
// writes a line saying so rather than falling silent. Worst case on disk is
// storeRetention x this x the number of shelves; in steady state it is zero,
// because a quiet machine writes no lines at all.
const storeDayCap = 4 << 20
// storeCapped is the last line a capped day-file gets. It is a real journal
// line, so a host replaying the day sees the truncation in band instead of
// inferring it from a gap.
const storeCapped = "day_capped"
// storeDayOpen is the one line a new day-file opens with when findings were
// already open: the open set as of that moment, so the file is self-contained
// for a reader whose window starts inside it.
const storeDayOpen = "day_open"
// storeSnapshotMaxFPs bounds that line. A fingerprint is 32 hex characters
// (baselineFP: hex of 16 bytes), so 256 of them plus the note measure 9,351
// bytes — about 1/450 of the day's ceiling — and the snapshot can never be
// what fills a file. The previous per-finding version could and did, which is
// the whole reason this is a bound and not a loop. (An earlier comment said 16
// characters and 4.5 KB; the conclusion held with a 450x margin, the numbers
// did not.)
const storeSnapshotMaxFPs = 256
// storeStateFile is the writer's own memory of what is open, kept in the state
// dir and NOT in the store — the host has no business reading the bookkeeping
// that decides what it gets told, and a file the host could write is a file
// that could suppress a finding.
const storeStateName = "store-state.json"
// storeMu serializes the writers. refreshDelta already guarantees one delta
// pass at a time (its in-flight flag, deltarun.go:467-478), but the alerts
// refresher runs ONE GOROUTINE PER PAIRING (run.go starts it inside runPairing),
// so on a box serving several hosts those writers genuinely do collide over
// store-alerts.json.
var storeMu sync.Mutex
// ── the journal line ────────────────────────────────────────────────────────
// storeLine is one line of findings/<date>.jsonl.
//
// It carries the WHOLE event, not a reference to one. A journal whose lines
// point at state held elsewhere stops being replayable the moment that state
// moves on, and the state here is explicitly allowed to move on — the
// accounting gets retaken, delta-state.json gets deleted (deltastate.go:25-30).
// The line has to stand on its own or it is not a record.
type storeLine struct {
// At is when this transition was WRITTEN. Distinct from Event.FirstSeen,
// which is when the box first noticed the finding: a box that was off for a
// weekend writes `opened` on Monday for something it first saw on Friday,
// and collapsing the two would lose that.
At string `json:"at"`
// Transition is opened | changed | closed | day_open | day_capped.
Transition string `json:"transition"`
// Machine and Version are stamped per line, not per file. A day-file can be
// read as a range by a host that never saw its header, and a line that
// cannot say which machine it came from is not evidence.
Machine string `json:"machine"`
Version string `json:"version"`
// Baseline is the digest of the accounting this finding stands against.
// Without it, two lines from either side of a retake look comparable and
// are not.
Baseline string `json:"baseline,omitempty"`
Event *deltaEvent `json:"event,omitempty"`
// OpenCount and OpenFPs belong to a day_open line and to nothing else: how
// many findings were open when the day-file was started, and their
// fingerprints. The COUNT is separate from the list because the list is
// bounded — a reader must be able to tell "3 open, here they are" from
// "3,000 open, here are the first 256".
OpenCount int `json:"open_count,omitempty"`
OpenFPs []string `json:"open_fps,omitempty"`
// Note carries the reason for a line that is not about one event.
Note string `json:"note,omitempty"`
}
// storeState is what the writer remembers between passes: the open set, keyed
// by fingerprint, valued by the digest of the event's material content. The
// value is what makes `changed` detectable without keeping the events.
type storeState struct {
Version string `json:"version"`
// Baseline is the accounting the open set was built against. A retake
// invalidates it for the reason deltastate.go:36-38 gives; storeRecordPass
// closes every open finding explicitly on the pass that notices, rather
// than the loader dropping them where no line is ever written.
Baseline string `json:"baseline"`
Open map[string]string `json:"open,omitempty"`
// OpenKind: open fingerprint -> the entry kind of its finding, so a pass
// that did not compare a class closes nothing of that class. An open
// fingerprint with no kind was opened before the listener and identity
// classes existed, so it is a process or a scheduler, which are never
// left out one class at a time.
OpenKind map[string]string `json:"open_kind,omitempty"`
// Day is the UTC day of the file the open set was last written into, so a
// rollover can carry the open set into the new one. Empty on a fresh state,
// which correctly carries nothing.
Day string `json:"day,omitempty"`
}
// Bumped from v1 when storeState gained Day. A v1 file is MIGRATED, not
// discarded: rejecting it wholesale re-opened every current finding and — the
// part that matters — never closed a finding that had gone away in between,
// because the loader forgot it was ever open. `zzz` open under v1, absent after
// the upgrade, got no `closed` line anywhere, ever. v1 lacks only `day`, which
// reads as "" and costs exactly one skipped snapshot; everything else carries.
const storeStateVersion = "privatae-store-state-v2"
// storeStateVersionsKnown are the versions the loader will carry forward.
var storeStateVersionsKnown = map[string]bool{"privatae-store-state-v1": true, storeStateVersion: true}
// storeAlertStateVersion is the alerts bookkeeping's OWN constant. It shared
// storeStateVersion, so the findings-side bump re-journalled every firing
// alert once on upgrade for a schema that had not changed.
const storeAlertStateVersion = "privatae-store-alerts-v1"
// storeRootPath is storeDir WITHOUT the MkdirAll — the path this box reports on
// its beat as `machine_record_root`. The beat fires every 60 seconds and
// reporting a path must not create a directory as a side effect; configDirPath
// exists for the same reason one level up (config.go:121-128).
func storeRootPath() (string, error) {
d, err := configDirPath()
if err != nil {
return "", err
}
return filepath.Join(d, storeDirName), nil
}
func storeDir() (string, error) {
d, err := configDir()
if err != nil {
return "", err
}
p := filepath.Join(d, storeDirName)
return p, os.MkdirAll(p, 0o700)
}
// storeShelf is one shelf directory under the root, created on first write.
func storeShelf(name string) (string, error) {
root, err := storeDir()
if err != nil {
return "", err
}
p := filepath.Join(root, name)
return p, os.MkdirAll(p, 0o700)
}
func storeStatePath() (string, error) {
d, err := configDir()
if err != nil {
return "", err
}
return filepath.Join(d, storeStateName), nil
}
// storeStateLoadRaw reads the state AS IT STANDS, without judging it against
// the current accounting.
//
// THE JUDGING MOVED TO THE CALLER ON PURPOSE. This used to take the current
// baseline digest and return a fresh empty state when it disagreed — which
// meant a retake silently erased the open set before anything could write a
// line about it, and every finding measured against the old accounting left the
// journal without a trace. storeRecordPass now compares the digests itself and
// closes what it is about to stop tracking.
//
// An unreadable or corrupt file is still not an error, for deltaStateLoad's
// reason (deltastate.go:97-106). The consequence here is the safe direction: a
// lost open set re-opens every current finding, so the journal repeats itself.
// It never loses one.
func storeStateLoadRaw() *storeState {
fresh := &storeState{Version: storeStateVersion, Open: map[string]string{}}
p, err := storeStatePath()
if err != nil {
return fresh
}
b, err := os.ReadFile(p)
if err != nil {
return fresh
}
var st storeState
if json.Unmarshal(b, &st) != nil {
return fresh
}
if !storeStateVersionsKnown[st.Version] {
return fresh
}
st.Version = storeStateVersion
if st.Open == nil {
st.Open = map[string]string{}
}
return &st
}
func (s *storeState) save() error {
p, err := storeStatePath()
if err != nil {
return err
}
b, err := json.MarshalIndent(s, "", " ")
if err != nil {
return err
}
if err := os.WriteFile(p, b, 0o600); err != nil {
return err
}
// Re-chmod for the reason baselineSave and deltaState.save do: WriteFile
// applies its mode only on create.
_ = os.Chmod(p, 0o600)
return nil
}
// storeEventDigest is the material content of a finding — what has to differ
// for `changed` to be the truth rather than churn.
//
// WHAT IS EXCLUDED IS THE POINT. PIDs are not here: a service restarting gets
// new pids every time and that is not a change in the finding, it is the same
// finding on a new process. FirstSeen is not here either — it is fixed at the
// first sighting by definition, so including it would be inert, and reading it
// as material would make a re-stamp (a deleted delta-state.json) look like the
// finding itself changed.
func storeEventDigest(ev *deltaEvent) string {
added := append([]string(nil), ev.Added...)
sort.Strings(added)
return strings.Join([]string{
ev.Kind, ev.Reason, ev.Level, ev.Tier,
// ArgvShape and not a "command" field: a scheduler's command IS its argv
// shape here (delta.go:305 compares exactly this pair to raise
// deltaReasonCommand), and on a process it is part of the slot
// (delta.go:158). One field covers both because the code that decides a
// `changed` already treats it as one.
ev.Entry.ExeDigest, ev.Entry.ArgvShape,
// Was.FP: which baseline entry this displaced. departedAtSlot
// (delta.go:229-233) can name a DIFFERENT displaced entry across passes
// while every field above stays equal, so without this "this replaced X"
// silently becomes "this replaced Y". Narrow, but it is the direction
// where a real change is swallowed.
storeWasFP(ev),
strings.Join(added, ","),
}, "\x00")
}
// storeWasFP is the fingerprint of the baseline entry a `changed` displaced,
// or "" when the event displaced nothing.
func storeWasFP(ev *deltaEvent) string {
if ev.Was == nil {
return ""
}
return ev.Was.FP
}
// ── writing ─────────────────────────────────────────────────────────────────
// storeDayPath is the file for one shelf and one day, UTC.
//
// UTC AND NOT LOCAL TIME, because the reader is not on this machine. A fleet
// spanning timezones whose day-files roll at local midnight cannot be asked
// "what happened on the 14th" without knowing where each box stands, and the
// one thing a host reading a store does not have is the box's clock offset.
func storeDayPath(shelf string, now time.Time) (string, error) {
d, err := storeShelf(shelf)
if err != nil {
return "", err
}
return filepath.Join(d, now.UTC().Format("2006-01-02")+".jsonl"), nil
}
// storeNoticeReserve is room held back from storeDayCap so the capped-notice
// always fits INSIDE the ceiling.
//
// Without it the notice is written past the cap — measured at 221 bytes over on
// the first run of TestACappedDayFileSaysSoOnceAndThenStops — and then
// storeDayCap is not a ceiling, it is a ceiling plus whatever the last line
// happened to be. Either the notice is exempt and the bound is soft, or room is
// reserved for it and the bound is exact; this takes the second, because a
// bound quoted in a comment as "cannot cross" has to be one. 512 bytes is the
// notice plus headroom for a long hostname. MEASURED AT 319 BYTES after the
// notice text grew; an earlier version of this comment still said 221, which is
// the sort of stale number that makes a bound look safer than it is. Linux caps
// os.Hostname at 64, so 512 holds.
const storeNoticeReserve = 512
// storeSealSuffix marks a day-file that has hit its ceiling. A SIBLING MARKER
// rather than a size test, because a size test cannot seal a file: the line
// that trips the cap may be large and the notice that replaces it small, so the
// file stays under the ceiling and the NEXT small line fits. Measured in review
// — four `day_capped` notices in one file with ordinary lines between them, and
// the notice's own text says the file "stops here". The marker makes sealing a
// fact about the day rather than an inference from its length.
const storeSealSuffix = ".capped"
// storeAppend adds one line to a shelf's day-file, honouring the cap.
//
// RETURNS WHETHER THE LINE WAS WRITTEN, and every caller must act on it. The
// first version returned only an error, so a capped file made a dropped line
// indistinguishable from a written one — and storeRecordPass then recorded the
// finding as open, which meant it was never emitted again on any later pass or
// any later day. A silently swallowed finding is the exact failure this whole
// file exists to prevent.
func storeAppend(shelf string, line any, now time.Time) (bool, error) {
p, err := storeDayPath(shelf, now)
if err != nil {
return false, err
}
b, err := json.Marshal(line)
if err != nil {
return false, err
}
b = append(b, '\n')
// Sealed: this day said what it had to say and is closed to further lines.
if _, sErr := os.Stat(p + storeSealSuffix); sErr == nil {
return false, nil
}
fi, err := os.Stat(p)
if err != nil && !errors.Is(err, os.ErrNotExist) {
return false, err
}
size := int64(0)
if err == nil {
size = fi.Size()
}
// Checked even when the file does not exist yet, so a single oversized line
// cannot create a file already past the ceiling.
if size+int64(len(b)) > storeDayCap-storeNoticeReserve {
notice, mErr := json.Marshal(storeLine{
At: stamp(now), Transition: storeCapped, Machine: storeMachineName(),
Version: clientVersion,
Note: fmt.Sprintf("this day-file reached its %d-byte ceiling and stops here; "+
"nothing further is recorded for this UTC day, and a machine producing "+
"this many transitions in a day is itself the finding",
storeDayCap),
})
if mErr != nil {
return false, mErr
}
if wErr := storeWriteLine(p, append(notice, '\n')); wErr != nil {
return false, wErr
}
// Seal AFTER the notice, so the notice itself is never refused by the
// seal it creates.
if wErr := storeWriteLine(p+storeSealSuffix, nil); wErr != nil {
return false, wErr
}
return false, nil
}
if wErr := storeWriteLine(p, b); wErr != nil {
return false, wErr
}
return true, nil
}
func storeWriteLine(p string, b []byte) error {
f, err := os.OpenFile(p, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600)
if err != nil {
return err
}
defer f.Close()
// Re-chmod, and NOT for the umask reason the other writers in this tree
// give. A umask only ever clears bits, so an O_CREATE of 0600 cannot come
// out looser than 0600 however permissive the umask is — the mutation that
// deletes this line survives a umask-based test, which is how the wrong
// rationale got caught here.
//
// What it actually guards is an EXISTING file whose mode is already loose:
// O_APPEND|O_CREATE leaves the mode of a file it did not create untouched,
// so a day-file restored from a backup, copied in with `cp -p`, or written
// by some earlier build goes on being world-readable for the rest of the
// day while this process appends the machine's findings to it.
_ = os.Chmod(p, 0o600)
_, err = f.Write(b)
return err
}
func storeMachineName() string {
hn, _ := os.Hostname()
return hn
}
// ── the findings journal ────────────────────────────────────────────────────
// storeRecordPass folds one delta pass into the journal and returns how many
// lines it wrote.
//
// IT IS CALLED WITH THE DAEMON'S PASS ONLY. An operator running `observe:delta`
// gets a report and writes nothing, for the same reason deltaRoleOperator does
// not advance the remembered binary pair (delta_test.go:1051): a person looking
// at the machine must not be able to move the record by looking at it.
//
// A pass that could not compare — no authority, no accounting, an accounting
// under other rules — writes NOTHING and closes nothing. This is the one that
// is easy to get wrong: treating "I could not look" as "there is nothing there"
// would close every open finding on the box the moment somebody deleted the
// accounting, and the journal would show a clean resolution of everything.
func storeRecordPass(r deltaReport, now time.Time) (int, error) {
if r.Baseline != deltaBaselineOK && r.Baseline != deltaBaselineEdit {
return 0, nil
}
storeMu.Lock()
defer storeMu.Unlock()
machine, wrote := storeMachineName(), 0
day := now.UTC().Format("2006-01-02")
// A line is only remembered as open once it is actually ON DISK. storeAppend
// refuses a line on a sealed day, and treating a refusal as a write is how
// the first version lost findings for good.
// put writes an OPENING-side line (opened / changed) and reports whether it
// landed. A REFUSAL IS A NO-OP ON THE OPEN SET: an `opened` that did not
// land was never in the set, so there is nothing to undo; a `changed` that
// did not land stays recorded under its OLD digest, so the next unsealed day
// writes `changed` — the truth — rather than a second `opened` with no close
// between. (An earlier version deleted the entry on any refusal, which
// produced exactly that double open; the review's mutation removing the
// delete was the better behaviour.)
//
// IT IS NOT USED FOR CLOSES. A refused `closed` must also stay in the open
// set, and the close loop handles its own refusals the same way — so both
// sides now share one rule: the set changes only when a line lands.
put := func(st *storeState, transition string, ev *deltaEvent, note string) (bool, error) {
ok, err := storeAppend(storeFindings, storeLine{
At: stamp(now), Transition: transition, Machine: machine,
Version: clientVersion, Baseline: r.BaselineDigest, Event: ev, Note: note,
}, now)
if err != nil || !ok {
return false, err
}
wrote++
st.Open[ev.FP] = storeEventDigest(ev)
if ev.Entry.Kind != "" {
if st.OpenKind == nil {
st.OpenKind = map[string]string{}
}
st.OpenKind[ev.FP] = ev.Entry.Kind
}
return true, nil
}
st := storeStateLoadRaw()
// ── A NEW DAY-FILE OPENS WITH ONE SNAPSHOT LINE ─────────────────────────
//
// Retention deletes day-files after 30 days, so a finding pinned longer than
// that had its `opened` line swept while it was still open, and its eventual
// `closed` referred to a fingerprint the retained window never opened. The
// day-file has to say what was already open when it started.
//
// ONE LINE, NOT ONE PER FINDING, AND THIS IS THE SECOND ATTEMPT. The first
// wrote a `carried` line per open finding, ahead of the day's opens and
// closes — and measured in review that is a PERMANENT LOCKUP, not a bad day:
// a carried line is ~324 bytes, so the carry alone fills a day at ~12,900
// open findings. Once it does, `closed` (which runs last) is starved first,
// so the open set can never drain, so the carry never shrinks. Seven
// simulated days ended with the shelf writing nothing but its own memory of
// the day it locked, and a new `critical` that was never journalled once.
// That defeats the cap's own purpose — the storm is meant to BE the signal —
// and undid the fix whose point was that the next unsealed day takes the
// refused finding.
//
// It was also the "clock" shape this file's header forbids, just amortised:
// an edge journal turned into a daily O(open set) snapshot.
//
// AND IT RUNS BEFORE THE RETAKE-CLOSE. The retake empties the open set; a
// snapshot taken after it lists nothing, and a day-file that begins with
// two `closed` lines for fingerprints it never introduced is exactly the
// un-self-contained file this line exists to prevent (measured in review:
// `[closed aaa, closed bbb, opened aaa]` with no snapshot). Snapshot first,
// then the closes, then the opens: `[day_open{aaa,bbb}, closed, closed,
// opened]` replays correctly from that file alone.
//
// So: one bounded line carrying the fingerprints, truncated with a count
// when there are more than fit. A reader learns exactly what was open and,
// when the list is cut, that it was cut and by how much — which is strictly
// more than the per-finding version delivered once it had eaten the day.
if st.Day != "" && st.Day != day && len(st.Open) > 0 {
openFPs := make([]string, 0, len(st.Open))
for fp := range st.Open {
openFPs = append(openFPs, fp)
}
sort.Strings(openFPs)
total := len(openFPs)
note := fmt.Sprintf("%d finding(s) were already open when this day-file was "+
"started; their opening lines are in earlier files, which retention may "+
"already have swept", total)
if total > storeSnapshotMaxFPs {
openFPs = openFPs[:storeSnapshotMaxFPs]
note += fmt.Sprintf(" — this line names the first %d of them, in fingerprint "+
"order; the rest are counted, not listed", storeSnapshotMaxFPs)
}
ok, err := storeAppend(storeFindings, storeLine{
At: stamp(now), Transition: storeDayOpen, Machine: machine,
Version: clientVersion, Baseline: r.BaselineDigest,
OpenCount: total, OpenFPs: openFPs, Note: note,
}, now)
if err != nil {
return wrote, err
}
if ok {
wrote++
}
}
st.Day = day
// ── A RETAKE CLOSES WHAT IT MADE INCOMPARABLE ───────────────────────────
//
// A retake redefines normal, so findings measured against the old
// accounting cannot be compared to the new one. The first version silently
// dropped them — storeStateLoad returned a fresh set on a digest mismatch,
// so the close loop had nothing to iterate and the old findings vanished
// from the journal without a line. That also falsified this file's own
// claim that replaying reconstructs the open set exactly. They are closed
// explicitly now, and the note says which of the two reasons it was.
if st.Baseline != r.BaselineDigest {
old := st.Open
carried := map[string]string{}
var prior []string
for fp := range old {
prior = append(prior, fp)
}
sort.Strings(prior)
allClosed := true
for _, fp := range prior {
ok, err := storeAppend(storeFindings, storeLine{
At: stamp(now), Transition: "closed", Machine: machine,
Version: clientVersion, Baseline: r.BaselineDigest,
Event: &deltaEvent{FP: fp},
Note: "the accounting was retaken, so this finding is no longer comparable " +
"and is not being tracked against the new one — that is not the same " +
"as resolved" + map[bool]string{
false: ", and anything still present will re-open on this pass",
true: "; this pass could not walk the machine, so anything still present re-opens on the next one that can",
}[r.WalkFailed != ""],
}, now)
if err != nil {
return wrote, err
}
if ok {
wrote++
continue
}
// REFUSED, SO THE CLOSE DID NOT HAPPEN. Keep the entry and, below,
// do NOT advance the remembered baseline — otherwise the retake
// branch is never re-entered, nothing retries, and that finding's
// only line anywhere in the store is its original `opened`.
// Measured in review as a permanent drop.
carried[fp] = old[fp]
allClosed = false
}
if allClosed {
st = &storeState{Version: storeStateVersion, Baseline: r.BaselineDigest,
Open: map[string]string{}, Day: st.Day}
} else {
// Baseline deliberately left at the OLD digest: the next pass, on an
// unsealed day, re-enters this branch and finishes the job.
st.Open = carried
}
}
nowOpen := make(map[string]string, len(r.Findings))
for i := range r.Findings {
nowOpen[r.Findings[i].FP] = storeEventDigest(&r.Findings[i])
}
// opened / changed, in the order the pass reported them.
for i := range r.Findings {
ev := &r.Findings[i]
was, had := st.Open[ev.FP]
transition := ""
switch {
case !had:
transition = "opened"
case was != nowOpen[ev.FP]:
transition = "changed"
default:
continue
}
if _, err := put(st, transition, ev, ""); err != nil {
return wrote, err
}
}
// ── closed ──────────────────────────────────────────────────────────────
//
// A NARROWER SCAN THAN THE ACCOUNTING CLOSES NOTHING — and the test is the
// MISMATCH, not incompleteness on its own. That distinction is the fix for a
// real trap: baseline_all_other.go returns false unconditionally on darwin
// and windows, so `Complete` is permanently false there. Gating on it alone
// meant no finding ever closed on a Mac or a Windows box, the open set grew
// monotonically for the life of the machine, and the day_open snapshot
// walked it upward for ever.
//
// Those platforms are not blind, they are consistently filtered: the
// accounting was taken through the same recognised-only walk the pass uses
// (baselineCoverage names each half), so the two sides agree about scope and
// an absence really is an absence. What must never close is the case where
// the accounting saw MORE than this pass could — an all-processes baseline
// compared against a partial walk — because there the missing entry may
// simply not have been looked at.
if storeScopeAgrees(r) {
// A CLASS THIS PASS DID NOT COMPARE CLOSES NOTHING. Its findings are
// absent from r.Findings because nothing looked for them, not because
// they went away: an `ss` that fails every other pass would otherwise
// close and reopen a live finding on every beat, and the journal would
// say a backdoor on 0.0.0.0:4444 was resolved twelve times in half an
// hour. The same rule the walk-failure guard applies, one class at a time.
skipped := map[string]bool{}
for _, k := range r.Skipped {
skipped[k] = true
}
var gone []string
for fp := range st.Open {
if _, still := nowOpen[fp]; !still && !skipped[st.OpenKind[fp]] {
gone = append(gone, fp)
}
}
// Stable order — map iteration is random and a journal whose line order
// changes between identical runs cannot be diffed in a test or by a person.
sort.Strings(gone)
for _, fp := range gone {
ok, err := storeAppend(storeFindings, storeLine{
At: stamp(now), Transition: "closed", Machine: machine,
Version: clientVersion, Baseline: r.BaselineDigest,
Event: &deltaEvent{FP: fp},
Note: "this finding was not reported by the pass that closed it — which " +
"is not the same as resolved",
}, now)
if err != nil {
return wrote, err
}
// ONLY ON A WRITE. A refused close leaves the entry OPEN, because
// the close did not happen — the next unsealed day writes it. The
// retake path deleted unconditionally and lost closes permanently
// (NB2); this is the same rule stated the same way in both places.
if ok {
wrote++
delete(st.Open, fp)
}
}
}
for fp := range st.OpenKind {
if _, open := st.Open[fp]; !open {
delete(st.OpenKind, fp)
}
}
if err := st.save(); err != nil {
return wrote, err
}
return wrote, nil
}
// ── the alerts journal ──────────────────────────────────────────────────────
// storeAlertLine is one line of alerts/<date>.jsonl.
//
// IT CARRIES A FINGERPRINT AND NOT AN ALERT. An Alertmanager fingerprint is a
// hash that resolves to nothing without the Alertmanager — agents.go:678-681
// settled that this is why the counts may ride the beat ungated, and the same
// reasoning is what lets them be written here: this shelf holds no alert names,
// no labels and no hostnames. The rows that do carry labels stay behind
// `observe:alerts`, which is grant-gated.
type storeAlertLine struct {
At string `json:"at"`
Transition string `json:"transition"`
Machine string `json:"machine"`
Version string `json:"version"`
FP string `json:"fp,omitempty"`
// Counts are the summary AS OF this line, so a reader who joins the journal
// mid-day has the totals without replaying from the start of it.
Firing int `json:"firing"`
Critical int `json:"critical"`
Warning int `json:"warning"`
Note string `json:"note,omitempty"`
}
// storeAlertState is the open set for alerts, kept beside the findings one.
type storeAlertState struct {
Version string `json:"version"`
Open []string `json:"open,omitempty"`
// Truncated is the "<firing>/<listed>" shape of the last truncation line
// written, so the notice is written when the truncation CHANGES rather than
// on every refresh. Without it the shelf became a clock.
Truncated string `json:"truncated,omitempty"`
}
const storeAlertStateName = "store-alerts.json"
func storeAlertStatePath() (string, error) {
d, err := configDir()
if err != nil {
return "", err
}
return filepath.Join(d, storeAlertStateName), nil
}
// storeRecordAlerts journals the alert set's transitions.
//
// A NIL SUMMARY IS NOT AN EMPTY ONE, and this is the same wall the beat puts up
// (observe_pull.go:878-883): nil means this box has no Alertmanager configured
// or has not refreshed yet, so it is not watching anything and must not be read
// as watching nothing. It closes nothing and writes nothing — only a real
// summary that no longer lists a fingerprint closes it.
func storeRecordAlerts(s *alertsSummary, now time.Time) (int, error) {
if s == nil {
return 0, nil
}
storeMu.Lock()
defer storeMu.Unlock()
open := map[string]bool{}
prevTruncated := ""
if p, err := storeAlertStatePath(); err == nil {
if b, rErr := os.ReadFile(p); rErr == nil {
var st storeAlertState
if json.Unmarshal(b, &st) == nil && st.Version == storeAlertStateVersion {
for _, fp := range st.Open {
open[fp] = true
}
prevTruncated = st.Truncated
}
}
}
firingNow := map[string]bool{}
for _, fp := range s.Fingerprints {
firingNow[fp] = true
}
machine, wrote := storeMachineName(), 0
write := func(transition, fp, note string) error {
ok, err := storeAppend(storeAlerts, storeAlertLine{
At: stamp(now), Transition: transition, Machine: machine,
Version: clientVersion, FP: fp, Firing: s.Firing,
Critical: s.Critical, Warning: s.Warning, Note: note,
}, now)
if err != nil {
return err
}
if ok {
wrote++
}
return nil
}
// THE OPEN SET FOLLOWS WHAT WAS WRITTEN, not what was firing. A sealed
// day-file refuses lines, and recording an alert as open whose `firing` line
// never landed would mean it is never journalled again — the same silent
// loss the findings shelf had. `write` reports whether the line landed.
nextOpen := map[string]bool{}
for fp := range open {
nextOpen[fp] = true
}
fresh := append([]string(nil), s.Fingerprints...)
sort.Strings(fresh)
for _, fp := range fresh {
if open[fp] {
continue
}
before := wrote
if err := write("firing", fp, ""); err != nil {
return wrote, err
}
if wrote > before {
nextOpen[fp] = true
}
}
var gone []string
for fp := range open {
if !firingNow[fp] {
gone = append(gone, fp)
}
}
sort.Strings(gone)
for _, fp := range gone {
before := wrote
if err := write("cleared", fp, ""); err != nil {
return wrote, err
}
if wrote > before {
delete(nextOpen, fp)
}
}
// ONLY WHEN THE TRUNCATION ITSELF CHANGES. Written unconditionally this was
// a clock: refreshAlerts runs every 60s per pairing, so a fleet sitting on
// 57 firing alerts wrote 1,440 identical lines a day per host served — the
// exact shape this file's header forbids, measured in review at 5 refreshes
// producing 21, 1, 1, 1, 1 lines.
//
// THE SUMMARY IS CAPPED AT TWENTY FINGERPRINTS (observe_pull.go's narrowing,
// mirrored host-side in narrow_alerts). Past that the set this journal sees
// is a sample, so "cleared" would be a lie about the ones that fell off the
// end — say so in band rather than letting a truncated list read as the
// whole one.
truncated := ""
if s.Firing > len(s.Fingerprints) {
truncated = fmt.Sprintf("%d/%d", s.Firing, len(s.Fingerprints))
}
// THE STATE ONLY MOVES IF THE LINE LANDED. Saving `truncated` after a
// refused write meant a sealed day recorded the notice as delivered and the
// next unsealed day, with the same summary, never wrote it — the value was
// gone for good. Same rule as the open set: bookkeeping follows the disk.
truncatedWritten := prevTruncated
if truncated != "" && truncated != prevTruncated {
before := wrote
if err := write("truncated", "", fmt.Sprintf(
"%d alerts are firing and this summary names %d of them; the rest are "+
"counted, not listed, so this journal cannot say when they cleared",
s.Firing, len(s.Fingerprints))); err != nil {
return wrote, err
}
if wrote > before {
truncatedWritten = truncated
}
} else if truncated == "" {
// No longer truncated: forget the old value so a later truncation is
// reported again rather than compared against a stale one.
truncatedWritten = ""
}
if wrote == 0 && truncatedWritten == prevTruncated {
return 0, nil
}
stillOpen := make([]string, 0, len(nextOpen))
for fp := range nextOpen {
stillOpen = append(stillOpen, fp)
}
sort.Strings(stillOpen)
st := storeAlertState{Version: storeAlertStateVersion, Open: stillOpen, Truncated: truncatedWritten}
b, err := json.MarshalIndent(st, "", " ")
if err != nil {
return wrote, err
}
p, err := storeAlertStatePath()
if err != nil {
return wrote, err
}
if err := os.WriteFile(p, b, 0o600); err != nil {
return wrote, err
}
_ = os.Chmod(p, 0o600)
return wrote, nil
}
// storeScopeAgrees reports whether this pass looked at as much as the
// accounting it is being compared against.
//
// A complete walk always agrees. An incomplete one agrees only when the
// accounting was itself taken through the recognised-only walk — the
// consistently-filtered platform — which baselineCoverage records as
// `processes=recognised-only` (baseline.go:517-526). An accounting that claims
// `processes=all` compared against a partial walk is the genuine mismatch, and
// the only case where an absence is not evidence.
//
// A pass whose walk FAILED agrees with nothing: it compared no entries, so an
// absence in it is not evidence on any platform — least of all the
// recognised-only ones, where Complete=false is the everyday state and the
// coverage label would otherwise wave the pass through.
func storeScopeAgrees(r deltaReport) bool {
if r.WalkFailed != "" {
return false
}
if r.Complete {
return true
}
return strings.Contains(r.Coverage, "processes=recognised-only")
}
// ── the observe shelf ───────────────────────────────────────────────────────
//
// THE ACCOUNTING HALF OF AAA. audit_fetch records what the host
// ASKED this box to do. Until now nothing recorded what it SAW: an observe
// op's result went back down the lane and was dropped, so "what did the AI
// learn from this machine" — the question an incident turns on — had no
// answer on either side. Each result is now a line in observe/<op>/<date>.jsonl,
// with WHO asked (the pairing) and WHAT they passed.
//
// WHO CAN READ IT. The store is granted to the anchor host (or the only host on
// a one-host box). On an anchored box a member host with `observe` can run
// these ops, and its results land here where the ANCHOR can read them. That is
// intended and it is not a leak: the anchor administers the machine, can run
// every one of these ops itself, and the record is the machine's — which is
// why each line names the asking host rather than hiding it.
// storeObserveLineMax bounds ONE LINE — the whole line on disk, not just the
// result inside it. The first version bounded `len(result)` alone and recorded
// the host's args verbatim, so a member host with `observe` could send a 5 MiB
// argument to an op that ignores arguments, get a normal answer, and seal the
// shelf for the rest of the UTC day — erasing the record of what it saw. The
// accounting half of AAA, erasable by the party being accounted for. Measured
// in review: one 2,097,461-byte line; a seal on the first frame.
//
// Three walls now: args are kept ONLY for the ops that consume them, each arg
// is cut at storeObserveArgMax, and the marshalled line is measured and the
// result's head shrunk until the whole line fits. A max-size line on disk
// measures storeObserveLineMax exactly, so at the append ceiling of
// storeDayCap-storeNoticeReserve (4,193,792 bytes) at least 127 full-size
// results fit in a day before the cap — 127, not the 128 an earlier comment
// claimed, because the envelope around the result is not free.
//
// WHY 32 KiB. observe:units and observe:sockets on a busy box run to tens of
// KB; at 288 host polls a day a 20 KB result is 5.7 MB, past the day ceiling
// by mid-afternoon. A larger result keeps its first storeObserveHeadKeep bytes
// plus how much was cut — the shape of what was seen, never nothing.
const (
storeObserveLineMax = 32 << 10
storeObserveHeadKeep = 4 << 10
storeObserveArgMax = 256
)
// observeArgsRecorded is which ops CONSUME caller arguments, and how many.
// Everything else on the family ignores its args, and an arg nobody reads is
// not part of what was asked — it is bytes the host chose to send, and
// recording them lets the host choose the size of the record. The table is
// pinned against the handlers by TestObserveArgsRecordedMatchesTheHandlers.
var observeArgsRecorded = map[string]int{
"observe:net": 1, // host[:port], charset-checked before use
"observe:metrics": 1, // a name from a fixed query table
"observe:logs": 2, // a name, and an optional minutes window (a whole number)
"observe:folders": 2, // an absolute path, and an optional depth
"observe:journal": 2, // a unit name, and an optional minutes window
"observe:logins": 1, // an optional window in hours
"observe:cert": 2, // a local port, and an optional server name
}
// storeObserveLine is one line of observe/<op>/<date>.jsonl.
//
// WHAT COMES TO REST HERE. Most of the ops return fixed schemas the beat
// already summarises (posture ints, unit names, sockets, a probe's TLS
// subject, a served certificate's names). observe:alerts and observe:metrics
// carry label maps that can name namespaces and tenants. observe:folders
// carries folder paths, which can carry a person's account or project name.
// observe:logins carries account names and the addresses they came from, the
// failed ones as the other side sent them (bounded by loginName).
// observe:logs and observe:journal are the outliers: their result is raw
// production log text, up to the line bound, persisted for 30 days — the
// payloads whose content this repo does not control, and so the places a
// token or a user identifier can come to rest. Files are 0600 under a 0700
// tree; the reader is the machine's own OS account and the anchor's approved
// store. Named here so the next person choosing what to log through these ops
// knows it stays.
type storeObserveLine struct {
At string `json:"at"`
Op string `json:"op"`
Machine string `json:"machine"`
Version string `json:"version"`
AskedBy string `json:"asked_by"`
Args []string `json:"args,omitempty"`
// Result is the op's reply as it was sent, or its first bytes as Head
// when Truncated; Bytes is the reply's full size either way.
Result json.RawMessage `json:"result,omitempty"`
Head string `json:"head,omitempty"`
Truncated bool `json:"truncated,omitempty"`
Bytes int `json:"bytes,omitempty"`
}
// storeObserveDropped counts results that could not be written, PER SHELF.
// It rides the beat as machine_record_dropped so a record with holes is never
// mistaken for a quiet machine — and it names the shelf, because "there are
// holes somewhere" is not something the host can act on. Process-lifetime and
// monotonic, like a counter should be: a per-window count would make a hole
// disappear when the window closed.
//
// It is one counter for the MACHINE, reported on every pairing's beat. On an
// anchored box a member's beat therefore carries drops another pairing's ops
// caused. That is the record's nature — it is the machine's, not the
// pairing's — and the anchor, which reads the record, is the one that acts.
var (
storeObserveDroppedMu sync.Mutex
storeObserveDropped = map[string]int64{}
)
func storeObserveDroppedAdd(shelf string) {
storeObserveDroppedMu.Lock()
storeObserveDropped[shelf]++
storeObserveDroppedMu.Unlock()
}
// storeObserveDroppedSnapshot is a copy for the beat; nil when nothing dropped.
func storeObserveDroppedSnapshot() map[string]int64 {
storeObserveDroppedMu.Lock()
defer storeObserveDroppedMu.Unlock()
if len(storeObserveDropped) == 0 {
return nil
}
out := make(map[string]int64, len(storeObserveDropped))
for k, v := range storeObserveDropped {
out[k] = v
}
return out
}
func storeObserveDroppedReset() {
storeObserveDroppedMu.Lock()
storeObserveDropped = map[string]int64{}
storeObserveDroppedMu.Unlock()
}
// storeSafeOpName reduces an op name to one path-safe component. `observe:net`
// becomes `observe-net`; anything outside the allowed set is dropped, and a
// name that reduces to nothing is refused by the caller.
//
// COLLISION IS POSSIBLE IN PRINCIPLE, from two sources: separator folding
// (`observe:k8s`, `observe-k8s` and `observe_k8s` all reduce to
// `observe-k8s`) and, since the lowercasing below, case folding
// (`observe:K8s` and `observe:k8s`). Neither can bite today — the op reaches
// this writer only from the observe names registered in handlerOps, all
// lowercase, which reduce to as many distinct directories
// (TestEveryRegisteredOpSanitisesToTheHostsShelfRule walks them) — but a
// future pair like `observe:k8s-nodes` / `observe:k8s_nodes` would interleave
// into one day-file. Whoever adds such a pair renames one of them.
func storeSafeOpName(op string) string {
var b strings.Builder
for _, r := range op {
switch {
case r >= 'A' && r <= 'Z':
// Lowercased, so the host's shelf-name rule ([a-z0-9-]) is a fact
// about what this emits rather than a hope. No registered op has
// an uppercase letter; this is what makes that a property.
b.WriteRune(r + ('a' - 'A'))
case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
b.WriteRune(r)
case r == ':', r == '_', r == '-':
b.WriteRune('-')
}
}
return strings.Trim(b.String(), "-")
}
// storeRecordObserve journals one observe result. Never returns an error to
// the op path — the reply to the host must go regardless — but counts what it
// could not write, per shelf.
func storeRecordObserve(op, askedBy string, args []string, result []byte, now time.Time) bool {
clean := storeSafeOpName(op)
if clean == "" || !strings.HasPrefix(op, "observe:") {
return false
}
line := storeObserveLine{
At: stamp(now), Op: op, Machine: storeMachineName(), Version: clientVersion,
AskedBy: askedBy, Bytes: len(result),
}
// Args: only for the ops that read them, only as many as they read, each
// cut at storeObserveArgMax. An op that ignores its args gets none
// recorded — those bytes were never part of what was asked.
if n := observeArgsRecorded[op]; n > 0 && len(args) > 0 {
if len(args) > n {
args = args[:n]
}
kept := make([]string, 0, len(args))
for _, a := range args {
if len(a) > storeObserveArgMax {
a = a[:storeObserveArgMax] + "…(cut)"
}
kept = append(kept, a)
}
line.Args = kept
}
// A SHORTCUT, not the wall: the loop below is what bounds the line, and a
// result at the bound cannot fit whole once any envelope surrounds it, so
// `>` here versus `>=` changes nothing (the review's mutation of it is
// equivalent by construction). This only spares marshalling a huge result
// once before the loop truncates it anyway.
if len(result) > storeObserveLineMax || !json.Valid(result) {
line.Truncated = true
line.Head = string(result[:min(len(result), storeObserveHeadKeep)])
} else {
line.Result = json.RawMessage(result)
}
// THE LINE IS BOUNDED BY CONSTRUCTION. Measure the marshalled line; if it
// is still over, the head is what gives — halved until it fits. A result
// under the bound can still produce an over-bound line once the envelope
// (machine name, args) is around it, which is exactly the gap the first
// version had.
for {
b, err := json.Marshal(line)
if err != nil {
storeObserveDroppedAdd(clean)
return false
}
if len(b)+1 <= storeObserveLineMax {
break
}
if !line.Truncated {
line.Truncated = true
line.Result = nil
line.Head = string(result[:min(len(result), storeObserveHeadKeep)])
continue
}
if len(line.Head) == 0 {
// Nothing left to give: the envelope alone is over the bound,
// which no real op can produce. Refuse rather than seal the day.
storeObserveDroppedAdd(clean)
return false
}
line.Head = line.Head[:len(line.Head)/2]
}
storeMu.Lock()
defer storeMu.Unlock()
ok, err := storeAppend(filepath.Join(storeObserve, clean), line, now)
if err != nil || !ok {
storeObserveDroppedAdd(clean)
return false
}
return true
}
// ── retention ───────────────────────────────────────────────────────────────
// storeSweep deletes day-files older than storeRetention across every shelf.
//
// IT READS THE NAME, NOT THE MTIME. A file's mtime moves when it is touched,
// copied or restored from a backup, so a sweep keyed on it deletes by when the
// filesystem was last written rather than by what the file is about — and the
// restore case is exactly when somebody wants the history. The name is the day,
// it was written by this code, and a file whose name does not parse is left
// alone rather than guessed at.
// storeSweepLast is the UTC date the sweep last ran, so it runs once a day
// rather than once a minute. Not persisted: a restart costing one extra walk of
// a directory with at most a few dozen files in it is not worth a file, and the
// walk is cheap precisely because retention keeps the directory small.
var (
storeSweepMu sync.Mutex
storeSweepLast string
)
// storeSweepIfDue runs the sweep at most once per UTC day.
//
// KEYED ON THE DATE AND NOT ON AN INTERVAL. A 24-hour interval on a machine
// that is suspended most of the time measures UPTIME, so a laptop awake two
// hours a day sweeps every twelve days and its store grows past the retention
// it advertises. Comparing dates means a box that wakes after a fortnight
// sweeps on its first pass.
func storeSweepIfDue(now time.Time) {
day := now.UTC().Format("2006-01-02")
storeSweepMu.Lock()
if storeSweepLast == day {
storeSweepMu.Unlock()
return
}
storeSweepLast = day
storeSweepMu.Unlock()
// The error is dropped: a sweep that cannot read its own directory is not a
// reason to stop journalling, and the next pass tries again tomorrow. What
// it would cost is disk, which the cap already bounds.
_, _ = storeSweep(now)
}
func storeSweep(now time.Time) (removed int, err error) {
root, rErr := storeDir()
if rErr != nil {
return 0, rErr
}
cutoff := now.UTC().Add(-storeRetention)
walkErr := filepath.Walk(root, func(p string, fi os.FileInfo, e error) error {
if e != nil || fi == nil || fi.IsDir() {
return nil
}
// Both shapes this store writes: `<date>.jsonl` and its `<date>.jsonl.capped`
// seal marker. A name that does not reduce to a date is left alone rather
// than guessed at — the sweep deletes things, so an unrecognised name is
// the one case where doing nothing is always right.
// REQUIRE A SUFFIX THIS STORE WRITES. Trimming optionally meant a bare
// `2026-07-01` with no extension also parsed and was deleted — measured
// in review. Nothing else writes into this 0700 directory today, so
// nothing was at risk; a sweep that deletes a shape its own comment does
// not claim is a sweep whose blast radius is larger than it reads.
name := fi.Name()
switch {
case strings.HasSuffix(name, ".jsonl"+storeSealSuffix):
name = strings.TrimSuffix(name, ".jsonl"+storeSealSuffix)
case strings.HasSuffix(name, ".jsonl"):
name = strings.TrimSuffix(name, ".jsonl")
default:
return nil
}
day, pErr := time.Parse("2006-01-02", name)
if pErr != nil {
return nil
}
if day.Before(cutoff) {
if os.Remove(p) == nil {
removed++
}
}
return nil
})
return removed, walkErr
}