hub v0.46.0: observability pass — per-box log pulls, bundle custody, 72h TTL + secret gate

log_bundle_requests + log_bundles store (gzip, newest-3, 72h TTL purged on the 60s sweep); SaveLogBundle secret gate fail-closed (blocked flag row, no payload; REDACTED/checksums pass). Report ACK gains controller_log_requested + ingests controller_log_tail; heartbeat envelope gains log_tail_requested + ingests log_tail (consume-once on arrival; pre-0.83 agents stay visibly pending). Host detail Diagnostics section: request buttons (controller/agent), state rows with honest latency hints, View/Download endpoint. Red-proofs: gate disabled and clear-on-arrival removed both FAIL their tests. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PSK5g6qYLknKj8u3QAFEr6
2026-07-11 16:57:47 +02:00
parent 1ee5559772
commit e35b1ae0e6
13 changed files with 898 additions and 4 deletions
@@ -0,0 +1,271 @@
+package store
+
+import (
+	"bytes"
+	"compress/gzip"
+	"database/sql"
+	"fmt"
+	"io"
+	"regexp"
+	"strings"
+	"time"
+)
+
+// Component log bundles (v0.46.0 observability) — the box-component (controller /
+// agent) sibling of the per-app log tails. Same sovereignty posture: the operator's
+// request is a PENDING flag the box consumes on its own cycle (report ACK for the
+// controller, heartbeat envelope for the agent); the hub never connects in. The
+// received tail is gzip-stored with a 72 h TTL; the bundler runs a token-pattern
+// gate BEFORE storing — a hit stores NOTHING and flags the bundle `blocked` (fail-
+// closed belt over the "keys never values" logging convention).
+
+const (
+	// LogBundleComponentController scopes by CUSTOMER (the report channel is per-customer).
+	LogBundleComponentController = "controller"
+	// LogBundleComponentAgent scopes by HOST (the heartbeat channel is per-host).
+	LogBundleComponentAgent = "agent"
+
+	// logBundleTTL is how long a received bundle is served before the purge sweep drops it.
+	logBundleTTL = 72 * time.Hour
+
+	// logBundleMaxRawBytes is the belt over the wire caps (both channels cap at 128 KiB):
+	// oversize input keeps the NEWEST lines within this budget.
+	logBundleMaxRawBytes = 1 << 20
+
+	// logBundleKeep bounds stored bundles per (scope, component) — newest kept.
+	logBundleKeep = 3
+)
+
+// secretPatterns is the bundle gate: content shaped like a credential. The logging
+// conventions forbid secret VALUES in any log line, so a hit here is a violation —
+// fail closed (store nothing, flag the bundle). Deliberately NOT a generic hex
+// matcher: fingerprints/checksums are logged by design and are public.
+var secretPatterns = []*regexp.Regexp{
+	regexp.MustCompile(`-----BEGIN [A-Z ]*PRIVATE KEY-----`),
+	regexp.MustCompile(`\bre_[A-Za-z0-9]{10,}`),   // Resend API key shape
+	regexp.MustCompile(`PVEAPIToken=[^\s"']+:\S+`), // Proxmox token WITH secret part
+	regexp.MustCompile(`(?i)\b(password|passwd|api_key|apikey|token_secret|client_secret)=[^\s"']{6,}`),
+	regexp.MustCompile(`Authorization:\s*Bearer\s+\S{16,}`),
+}
+
+// findSecretPattern returns the name of the first matching pattern ("" = clean).
+// A match containing "[REDACTED]" is skipped — box-side redaction deliberately
+// leaves `password=[REDACTED]` shapes, which are proof of redaction, not a leak. Pure.
+func findSecretPattern(lines []string) string {
+	for _, l := range lines {
+		for _, re := range secretPatterns {
+			if m := re.FindString(l); m != "" && !strings.Contains(m, "[REDACTED]") {
+				return re.String()
+			}
+		}
+	}
+	return ""
+}
+
+// LogBundleRequest is one pending operator pull intent.
+type LogBundleRequest struct {
+	ScopeID     string
+	Component   string
+	RequestedAt time.Time
+}
+
+// LogBundleMeta is a stored bundle's row without the payload.
+type LogBundleMeta struct {
+	ID            int
+	ScopeID       string
+	Component     string
+	CollectedAt   time.Time
+	ReceivedAt    time.Time
+	SizeBytes     int64
+	Blocked       bool
+	BlockedReason string
+}
+
+// RequestLogBundle records (or refreshes) the operator's pending pull for one
+// component. One active request per (scope, component) — a re-click refreshes.
+func (s *Store) RequestLogBundle(scopeID, component string) error {
+	if component != LogBundleComponentController && component != LogBundleComponentAgent {
+		return fmt.Errorf("unknown log bundle component %q", component)
+	}
+	_, err := s.db.Exec(`
+		INSERT INTO log_bundle_requests (scope_id, component, requested_at)
+		VALUES (?, ?, ?)
+		ON CONFLICT(scope_id, component) DO UPDATE SET requested_at = excluded.requested_at`,
+		scopeID, component, time.Now().UTC())
+	return err
+}
+
+// PendingLogBundleRequest reports whether a pull is pending for (scope, component)
+// — the value the report ACK / heartbeat envelope advertises.
+func (s *Store) PendingLogBundleRequest(scopeID, component string) (bool, error) {
+	var n int
+	err := s.db.QueryRow(`SELECT COUNT(1) FROM log_bundle_requests WHERE scope_id = ? AND component = ?`,
+		scopeID, component).Scan(&n)
+	return n > 0, err
+}
+
+// GetLogBundleRequests returns the pending requests for a scope (for the UI states).
+func (s *Store) GetLogBundleRequests(scopeID string) ([]LogBundleRequest, error) {
+	rows, err := s.db.Query(`SELECT scope_id, component, requested_at FROM log_bundle_requests WHERE scope_id = ?`, scopeID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []LogBundleRequest
+	for rows.Next() {
+		var r LogBundleRequest
+		if err := rows.Scan(&r.ScopeID, &r.Component, &r.RequestedAt); err == nil {
+			out = append(out, r)
+		}
+	}
+	return out, rows.Err()
+}
+
+// SaveLogBundle stores a received component tail: secret-gate → (gzip+store |
+// blocked row), prune to the newest logBundleKeep, and CLEAR the pending request
+// (consume-once — same shape as SaveAppLogTail). Returns blocked=true when the
+// gate fired (nothing stored beyond the flag row).
+func (s *Store) SaveLogBundle(scopeID, component string, collectedAt time.Time, lines []string) (blocked bool, err error) {
+	lines = capBundleLines(lines, logBundleMaxRawBytes)
+	raw := strings.Join(lines, "\n")
+
+	var gz []byte
+	reason := ""
+	if hit := findSecretPattern(lines); hit != "" {
+		// Fail-closed: a secret-shaped token means the content is NOT stored at all.
+		blocked = true
+		reason = "possible secret (pattern " + hit + ")"
+	} else {
+		var buf bytes.Buffer
+		zw := gzip.NewWriter(&buf)
+		if _, err := zw.Write([]byte(raw)); err != nil {
+			return false, err
+		}
+		if err := zw.Close(); err != nil {
+			return false, err
+		}
+		gz = buf.Bytes()
+	}
+
+	tx, err := s.db.Begin()
+	if err != nil {
+		return blocked, err
+	}
+	defer tx.Rollback()
+	if _, err := tx.Exec(`
+		INSERT INTO log_bundles (scope_id, component, collected_at, received_at, size_bytes, gz, blocked, blocked_reason)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+		scopeID, component, collectedAt, time.Now().UTC(), int64(len(raw)), gz, boolInt(blocked), reason); err != nil {
+		return blocked, err
+	}
+	if _, err := tx.Exec(`
+		DELETE FROM log_bundles
+		WHERE scope_id = ? AND component = ? AND id NOT IN (
+			SELECT id FROM log_bundles WHERE scope_id = ? AND component = ?
+			ORDER BY id DESC LIMIT ?
+		)`, scopeID, component, scopeID, component, logBundleKeep); err != nil {
+		return blocked, err
+	}
+	// Consume-once: the fulfilled request is cleared (a blocked bundle also fulfills —
+	// re-requesting the same secret-carrying ring would just block again).
+	if _, err := tx.Exec(`DELETE FROM log_bundle_requests WHERE scope_id = ? AND component = ?`,
+		scopeID, component); err != nil {
+		return blocked, err
+	}
+	return blocked, tx.Commit()
+}
+
+// GetLogBundles returns a scope's stored bundle rows (meta only), newest first.
+func (s *Store) GetLogBundles(scopeID string) ([]LogBundleMeta, error) {
+	rows, err := s.db.Query(`
+		SELECT id, scope_id, component, collected_at, received_at, size_bytes, blocked, blocked_reason
+		FROM log_bundles WHERE scope_id = ? ORDER BY id DESC`, scopeID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []LogBundleMeta
+	for rows.Next() {
+		m, serr := scanBundleMeta(rows)
+		if serr != nil {
+			continue
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// GetLogBundleContent returns one bundle's meta + gunzipped lines, scoped (no
+// cross-scope reads). A blocked bundle returns meta with nil lines.
+func (s *Store) GetLogBundleContent(id int, scopeID string) (*LogBundleMeta, []string, error) {
+	row := s.db.QueryRow(`
+		SELECT id, scope_id, component, collected_at, received_at, size_bytes, blocked, blocked_reason, gz
+		FROM log_bundles WHERE id = ? AND scope_id = ?`, id, scopeID)
+	var m LogBundleMeta
+	var blockedInt int
+	var gz []byte
+	if err := row.Scan(&m.ID, &m.ScopeID, &m.Component, &m.CollectedAt, &m.ReceivedAt,
+		&m.SizeBytes, &blockedInt, &m.BlockedReason, &gz); err != nil {
+		if err == sql.ErrNoRows {
+			return nil, nil, nil
+		}
+		return nil, nil, err
+	}
+	m.Blocked = blockedInt != 0
+	if m.Blocked || len(gz) == 0 {
+		return &m, nil, nil
+	}
+	zr, err := gzip.NewReader(bytes.NewReader(gz))
+	if err != nil {
+		return &m, nil, err
+	}
+	raw, err := io.ReadAll(io.LimitReader(zr, logBundleMaxRawBytes+1))
+	zr.Close()
+	if err != nil {
+		return &m, nil, err
+	}
+	return &m, strings.Split(string(raw), "\n"), nil
+}
+
+// PurgeExpiredLogBundles drops bundles older than the 72 h TTL (the honest custody
+// bound — pulled logs are transient diagnostics, not an archive). now is injectable
+// for the S5 test.
+func (s *Store) PurgeExpiredLogBundles(now time.Time) (int, error) {
+	res, err := s.db.Exec(`DELETE FROM log_bundles WHERE received_at < ?`, now.Add(-logBundleTTL).UTC())
+	if err != nil {
+		return 0, err
+	}
+	n, _ := res.RowsAffected()
+	return int(n), nil
+}
+
+func scanBundleMeta(rows *sql.Rows) (LogBundleMeta, error) {
+	var m LogBundleMeta
+	var blockedInt int
+	err := rows.Scan(&m.ID, &m.ScopeID, &m.Component, &m.CollectedAt, &m.ReceivedAt,
+		&m.SizeBytes, &blockedInt, &m.BlockedReason)
+	m.Blocked = blockedInt != 0
+	return m, err
+}
+
+// capBundleLines keeps the NEWEST lines within the raw byte budget (belt over the
+// wire caps; mirrors the box-side newest-kept semantics).
+func capBundleLines(lines []string, maxBytes int) []string {
+	total := 0
+	start := len(lines)
+	for i := len(lines) - 1; i >= 0; i-- {
+		total += len(lines[i]) + 1
+		if total > maxBytes {
+			break
+		}
+		start = i
+	}
+	return lines[start:]
+}
+
+func boolInt(b bool) int {
+	if b {
+		return 1
+	}
+	return 0
+}
@@ -0,0 +1,147 @@
+package store
+
+import (
+	"io"
+	"log"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func newBundleStore(t *testing.T) *Store {
+	t.Helper()
+	st, err := New(filepath.Join(t.TempDir(), "test.db"), log.New(io.Discard, "", 0))
+	if err != nil {
+		t.Fatalf("store.New: %v", err)
+	}
+	t.Cleanup(func() { st.Close() })
+	return st
+}
+
+// Round-trip: request → pending → save → stored + request CLEARED (consume-once).
+func TestLogBundle_RoundTripConsumeOnce(t *testing.T) {
+	st := newBundleStore(t)
+	if err := st.RequestLogBundle("host-1", LogBundleComponentAgent); err != nil {
+		t.Fatal(err)
+	}
+	if pending, _ := st.PendingLogBundleRequest("host-1", LogBundleComponentAgent); !pending {
+		t.Fatal("request not pending after RequestLogBundle")
+	}
+	blocked, err := st.SaveLogBundle("host-1", LogBundleComponentAgent, time.Now().UTC(),
+		[]string{"2026-07-11T10:00:00Z [INFO] netmount: ensured network mount name=nas"})
+	if err != nil || blocked {
+		t.Fatalf("save: blocked=%v err=%v", blocked, err)
+	}
+	if pending, _ := st.PendingLogBundleRequest("host-1", LogBundleComponentAgent); pending {
+		t.Error("request survived fulfillment — consume-once broken")
+	}
+	bundles, err := st.GetLogBundles("host-1")
+	if err != nil || len(bundles) != 1 || bundles[0].Blocked {
+		t.Fatalf("bundles = %+v err=%v, want 1 clean bundle", bundles, err)
+	}
+	meta, lines, err := st.GetLogBundleContent(bundles[0].ID, "host-1")
+	if err != nil || meta == nil || len(lines) != 1 || !strings.Contains(lines[0], "ensured network mount") {
+		t.Fatalf("content = %+v / %v / %v", meta, lines, err)
+	}
+	// Scoping: another scope cannot read it.
+	if m, _, _ := st.GetLogBundleContent(bundles[0].ID, "other-host"); m != nil {
+		t.Error("cross-scope bundle read must return nothing")
+	}
+}
+
+// S4 secret gate: a planted Resend-shaped token → the bundle is BLOCKED, its content
+// is NOT stored (nil lines, no gz), and the request is still cleared. Companion
+// red-proof: empty the secretPatterns table → blocked=false + content stored → FAIL.
+func TestLogBundle_SecretGateFailClosed(t *testing.T) {
+	st := newBundleStore(t)
+	if err := st.RequestLogBundle("cust-a", LogBundleComponentController); err != nil {
+		t.Fatal(err)
+	}
+	blocked, err := st.SaveLogBundle("cust-a", LogBundleComponentController, time.Now().UTC(), []string{
+		"[INFO] all fine",
+		"[DEBUG] oops leaked key re_XZq81hd7wJq2M9Yv in a log line",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !blocked {
+		t.Fatal("secret-shaped token not blocked — the gate is the last belt, it must fire")
+	}
+	bundles, _ := st.GetLogBundles("cust-a")
+	if len(bundles) != 1 || !bundles[0].Blocked || bundles[0].BlockedReason == "" {
+		t.Fatalf("blocked flag row missing/incomplete: %+v", bundles)
+	}
+	meta, lines, err := st.GetLogBundleContent(bundles[0].ID, "cust-a")
+	if err != nil || meta == nil || !meta.Blocked {
+		t.Fatalf("blocked meta = %+v err=%v", meta, err)
+	}
+	if lines != nil {
+		t.Fatalf("blocked bundle served content — nothing may be stored: %v", lines)
+	}
+	// The request is still consumed (re-pulling the same ring would block again).
+	if pending, _ := st.PendingLogBundleRequest("cust-a", LogBundleComponentController); pending {
+		t.Error("blocked save left the request pending")
+	}
+}
+
+// The gate must NOT fire on deliberately-redacted values or public checksums.
+func TestLogBundle_GateAllowsRedactedAndHashes(t *testing.T) {
+	st := newBundleStore(t)
+	blocked, err := st.SaveLogBundle("cust-a", LogBundleComponentController, time.Now().UTC(), []string{
+		"[INFO] env applied password=[REDACTED]",
+		"[INFO] artifact sha256=9828c5f7aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaf50b",
+		"[DEBUG] leaf fingerprint_sha256=ab:cd:ef",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if blocked {
+		t.Fatal("gate false-positive on redacted/checksum content — legitimate diagnostics would never arrive")
+	}
+}
+
+// S5 TTL: past 72 h → purged; before → served. Injectable clock.
+func TestLogBundle_TTLPurge(t *testing.T) {
+	st := newBundleStore(t)
+	if _, err := st.SaveLogBundle("host-1", LogBundleComponentAgent, time.Now().UTC(), []string{"line"}); err != nil {
+		t.Fatal(err)
+	}
+	// Before the TTL: kept.
+	if n, err := st.PurgeExpiredLogBundles(time.Now().Add(71 * time.Hour)); err != nil || n != 0 {
+		t.Fatalf("purge at 71h dropped %d (err=%v), want 0", n, err)
+	}
+	if b, _ := st.GetLogBundles("host-1"); len(b) != 1 {
+		t.Fatal("bundle gone before the TTL")
+	}
+	// Past the TTL: dropped.
+	if n, err := st.PurgeExpiredLogBundles(time.Now().Add(73 * time.Hour)); err != nil || n != 1 {
+		t.Fatalf("purge at 73h dropped %d (err=%v), want 1", n, err)
+	}
+	if b, _ := st.GetLogBundles("host-1"); len(b) != 0 {
+		t.Fatal("expired bundle still served")
+	}
+}
+
+// Retention: only the newest 3 bundles per (scope, component) are kept.
+func TestLogBundle_KeepNewest(t *testing.T) {
+	st := newBundleStore(t)
+	for i := 0; i < 5; i++ {
+		if _, err := st.SaveLogBundle("host-1", LogBundleComponentAgent, time.Now().UTC(), []string{"line"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	b, _ := st.GetLogBundles("host-1")
+	if len(b) != 3 {
+		t.Fatalf("kept %d bundles, want 3 (newest)", len(b))
+	}
+}
+
+// Oversize input keeps the NEWEST lines within the raw budget (belt over wire caps).
+func TestCapBundleLines_KeepsNewest(t *testing.T) {
+	lines := []string{"old " + strings.Repeat("x", 100), "new " + strings.Repeat("y", 100)}
+	capped := capBundleLines(lines, 110)
+	if len(capped) != 1 || !strings.HasPrefix(capped[0], "new") {
+		t.Fatalf("capped = %v, want only the newest line", capped)
+	}
+}
@@ -494,6 +494,34 @@ func (s *Store) migrate() error {
 		return err
 	}

+	// Component log bundles (v0.46.0 observability, see logbundle.go): the operator's
+	// pending pull intents per (scope, component) — scope = customer_id for the
+	// controller (report ACK channel) / host_id for the agent (heartbeat envelope) —
+	// plus the received gzip bundles (72 h TTL, purge on the 60 s sweep; a secret-gate
+	// hit stores a BLOCKED flag row with no payload).
+	_, err = s.db.Exec(`
+		CREATE TABLE IF NOT EXISTS log_bundle_requests (
+			scope_id     TEXT NOT NULL,
+			component    TEXT NOT NULL,
+			requested_at DATETIME NOT NULL,
+			PRIMARY KEY (scope_id, component)
+		);
+		CREATE TABLE IF NOT EXISTS log_bundles (
+			id             INTEGER PRIMARY KEY AUTOINCREMENT,
+			scope_id       TEXT NOT NULL,
+			component      TEXT NOT NULL,
+			collected_at   DATETIME NOT NULL,
+			received_at    DATETIME NOT NULL,
+			size_bytes     INTEGER NOT NULL,
+			gz             BLOB,
+			blocked        INTEGER NOT NULL DEFAULT 0,
+			blocked_reason TEXT NOT NULL DEFAULT ''
+		);
+	`)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }