From 77f5e06d6b2aaa53470bf60d2c408397c1ccd972 Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 22 Jun 2026 23:54:24 +0200 Subject: [PATCH] feat(audit): DegradingSink + durable buffer + loki/ntfy + reconcile (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The classification-aware I5 audit path (§4.4): - DegradingSink.Reserve: central up → AuditCentral; central down + confidential → refuse (no buffer); central down + internal/public + buffer writable → AuditBuffered; central down + buffer unwritable → floor refuse. Record executes the reserved outcome and, when buffered, fires an ntfy alert. - FileBuffer: durable JSONL buffer that survives process restart; Confirm rewrites the file without a record, so a buffered record is cleared ONLY after its central write is confirmed. - LokiCentral: /ready probe + /loki/api/v1/push (full audit entry as the structured line). NtfyNotifier: degraded-state alerts; token only in the auth header, never logged (regression-tested). - Reconcile + StartReconcile: replay buffered records to central on recovery, confirm-then-clear per record; a failed push keeps the record buffered (no loss). SlogSink updated to the two-phase shape (always central, never fails) — the default when no loki endpoint is set. Co-Authored-By: Claude Opus 4.8 (1M context) --- ingestion/internal/audit/buffer.go | 167 ++++++++++++++++++ ingestion/internal/audit/degrading.go | 96 +++++++++++ ingestion/internal/audit/degrading_test.go | 191 +++++++++++++++++++++ ingestion/internal/audit/loki.go | 99 +++++++++++ ingestion/internal/audit/loki_ntfy_test.go | 86 ++++++++++ ingestion/internal/audit/ntfy.go | 55 ++++++ ingestion/internal/audit/reconcile.go | 65 +++++++ ingestion/internal/audit/slog.go | 16 +- ingestion/internal/audit/slog_test.go | 4 +- 9 files changed, 773 insertions(+), 6 deletions(-) create mode 100644 ingestion/internal/audit/buffer.go create mode 100644 ingestion/internal/audit/degrading.go create mode 100644 ingestion/internal/audit/degrading_test.go create mode 100644 ingestion/internal/audit/loki.go create mode 100644 ingestion/internal/audit/loki_ntfy_test.go create mode 100644 ingestion/internal/audit/ntfy.go create mode 100644 ingestion/internal/audit/reconcile.go diff --git a/ingestion/internal/audit/buffer.go b/ingestion/internal/audit/buffer.go new file mode 100644 index 0000000..aa3846d --- /dev/null +++ b/ingestion/internal/audit/buffer.go @@ -0,0 +1,167 @@ +package audit + +import ( + "bufio" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" +) + +// FileBuffer is a durable, restart-surviving audit buffer backed by a +// JSONL file: one {id, entry} record per line. It is the internal/public +// tier fallback when loki is unreachable. Confirm rewrites the file +// without the confirmed record, so a record is cleared only after its +// central write is confirmed. +// +// Access is serialised by a mutex; the buffer is low-throughput (only +// written during a loki outage), so a whole-file rewrite on Confirm is +// acceptable and keeps the on-disk format trivially correct. +type FileBuffer struct { + path string + mu sync.Mutex +} + +type bufferLine struct { + ID string `json:"id"` + Entry capture.AuditEntry `json:"entry"` +} + +// NewFileBuffer returns a buffer backed by path. The parent directory is +// created if needed. The file itself is created lazily on first Append. +func NewFileBuffer(path string) (*FileBuffer, error) { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return nil, fmt.Errorf("create buffer dir: %w", err) + } + return &FileBuffer{path: path}, nil +} + +// Writable reports whether the buffer file can be appended to. It probes +// by opening the file for append (creating it if absent) — the same +// operation Append performs — so Reserve's check matches Append's reality. +func (b *FileBuffer) Writable() error { + b.mu.Lock() + defer b.mu.Unlock() + f, err := os.OpenFile(b.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return err + } + return f.Close() +} + +// Append durably writes one audit record. The ID is derived from the +// content + timestamp so it is stable and unique per record. +func (b *FileBuffer) Append(e capture.AuditEntry) error { + b.mu.Lock() + defer b.mu.Unlock() + + line := bufferLine{ID: recordID(e), Entry: e} + data, err := json.Marshal(line) + if err != nil { + return fmt.Errorf("marshal buffer line: %w", err) + } + f, err := os.OpenFile(b.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644) + if err != nil { + return err + } + defer func() { _ = f.Close() }() + if _, err := f.Write(append(data, '\n')); err != nil { + return err + } + return f.Sync() +} + +// Pending reads all buffered records. A missing file means none. +func (b *FileBuffer) Pending() ([]Buffered, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.readAllLocked() +} + +func (b *FileBuffer) readAllLocked() ([]Buffered, error) { + f, err := os.Open(b.path) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + defer func() { _ = f.Close() }() + + var out []Buffered + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for sc.Scan() { + raw := sc.Bytes() + if len(raw) == 0 { + continue + } + var l bufferLine + if err := json.Unmarshal(raw, &l); err != nil { + return nil, fmt.Errorf("parse buffer line: %w", err) + } + out = append(out, Buffered(l)) + } + return out, sc.Err() +} + +// Confirm removes a single record after its central write is confirmed, by +// rewriting the file without it. Unknown IDs are a no-op. +func (b *FileBuffer) Confirm(id string) error { + b.mu.Lock() + defer b.mu.Unlock() + + all, err := b.readAllLocked() + if err != nil { + return err + } + tmp := b.path + ".tmp" + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return err + } + w := bufio.NewWriter(f) + kept := 0 + for _, rec := range all { + if rec.ID == id { + continue + } + data, _ := json.Marshal(bufferLine(rec)) + if _, err := w.Write(append(data, '\n')); err != nil { + _ = f.Close() + return err + } + kept++ + } + if err := w.Flush(); err != nil { + _ = f.Close() + return err + } + if err := f.Sync(); err != nil { + _ = f.Close() + return err + } + if err := f.Close(); err != nil { + return err + } + // Empty buffer → remove the file entirely so Pending sees nothing. + if kept == 0 { + _ = os.Remove(tmp) + return os.Remove(b.path) + } + return os.Rename(tmp, b.path) +} + +// recordID is a stable per-record identifier: sha256 of the principal, +// timestamp, and item list. Distinct captures never collide; the same +// buffered record always hashes the same. +func recordID(e capture.AuditEntry) string { + h := sha256.New() + _, _ = fmt.Fprintf(h, "%s|%s|%v|%s", e.Principal, e.Timestamp.UTC().Format("2006-01-02T15:04:05.000000000Z07:00"), e.Items, e.SessionRef) + return hex.EncodeToString(h.Sum(nil))[:16] +} diff --git a/ingestion/internal/audit/degrading.go b/ingestion/internal/audit/degrading.go new file mode 100644 index 0000000..017d6ab --- /dev/null +++ b/ingestion/internal/audit/degrading.go @@ -0,0 +1,96 @@ +package audit + +import ( + "context" + "fmt" + + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/mathiasbq/hyperguild/ingestion/internal/classification" +) + +// Central is the central audit substrate (loki). Ready is a cheap +// reachability probe used by the pre-write reserve; Push writes a record. +type Central interface { + Ready(ctx context.Context) error + Push(ctx context.Context, e capture.AuditEntry) error +} + +// Buffer is the durable local fallback for internal/public-tier records +// when the central sink is unreachable. It must survive process restart. +type Buffer interface { + // Writable reports whether the buffer can currently be appended to. + Writable() error + Append(e capture.AuditEntry) error + // Pending returns buffered records awaiting reconciliation, each with a + // stable ID used to Confirm (delete) it after a confirmed central write. + Pending() ([]Buffered, error) + Confirm(id string) error +} + +// Buffered is a buffered audit record plus its stable buffer ID. +type Buffered struct { + ID string + Entry capture.AuditEntry +} + +// Notifier raises an out-of-band alert (ntfy) about a degraded state. +type Notifier interface { + Notify(ctx context.Context, msg string) error +} + +// DegradingSink is the classification-aware AuditSink (§4.4): +// +// - central reachable → AuditCentral (all tiers). +// - central down + confidential → refuse (no buffer): confidential must +// be centrally auditable at write time. +// - central down + internal/public + buffer writable → AuditBuffered. +// - central down + (confidential, or buffer not writable) → refuse (floor). +// +// The decision is made in Reserve, before any write; Record then executes it. +type DegradingSink struct { + central Central + buffer Buffer + notifier Notifier +} + +// NewDegradingSink wires the central sink, durable buffer, and notifier. +func NewDegradingSink(central Central, buffer Buffer, notifier Notifier) *DegradingSink { + return &DegradingSink{central: central, buffer: buffer, notifier: notifier} +} + +// Reserve decides, before any write, how the capture will be audited — or +// returns an error to refuse it. +func (d *DegradingSink) Reserve(ctx context.Context, level classification.Level) (capture.AuditOutcome, error) { + if err := d.central.Ready(ctx); err == nil { + return capture.AuditCentral, nil + } + // Central sink is down. + if level == classification.Confidential { + return 0, fmt.Errorf("confidential capture requires the central audit sink, which is unreachable") + } + if err := d.buffer.Writable(); err != nil { + // Floor: neither central nor local buffer can record the audit. + return 0, fmt.Errorf("audit floor: central sink down and local buffer unwritable: %w", err) + } + return capture.AuditBuffered, nil +} + +// Record persists the entry per the reserved outcome. For AuditBuffered it +// also fires the degraded-state alert. +func (d *DegradingSink) Record(ctx context.Context, e capture.AuditEntry, outcome capture.AuditOutcome) error { + switch outcome { + case capture.AuditBuffered: + if err := d.buffer.Append(e); err != nil { + return fmt.Errorf("buffer audit record: %w", err) + } + // Best-effort alert; the record is already durably buffered. + if d.notifier != nil { + _ = d.notifier.Notify(ctx, fmt.Sprintf( + "capture audit BUFFERED LOCALLY (loki unreachable) — principal=%s class=%s items=%d", + e.Principal, e.EffectiveClassification, len(e.Items))) + } + return nil + default: + return d.central.Push(ctx, e) + } +} diff --git a/ingestion/internal/audit/degrading_test.go b/ingestion/internal/audit/degrading_test.go new file mode 100644 index 0000000..09703e6 --- /dev/null +++ b/ingestion/internal/audit/degrading_test.go @@ -0,0 +1,191 @@ +package audit_test + +import ( + "context" + "errors" + "path/filepath" + "testing" + + "github.com/mathiasbq/hyperguild/ingestion/internal/audit" + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/mathiasbq/hyperguild/ingestion/internal/classification" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --- fakes --- + +type fakeCentral struct { + down bool + pushed []capture.AuditEntry + pushErr error +} + +func (f *fakeCentral) Ready(context.Context) error { + if f.down { + return errors.New("loki down") + } + return nil +} + +func (f *fakeCentral) Push(_ context.Context, e capture.AuditEntry) error { + if f.pushErr != nil { + return f.pushErr + } + f.pushed = append(f.pushed, e) + return nil +} + +type fakeNotifier struct{ msgs []string } + +func (f *fakeNotifier) Notify(_ context.Context, msg string) error { + f.msgs = append(f.msgs, msg) + return nil +} + +// unwritableBuffer always reports it cannot be written (floor condition). +type unwritableBuffer struct{} + +func (unwritableBuffer) Writable() error { return errors.New("disk full") } +func (unwritableBuffer) Append(capture.AuditEntry) error { return errors.New("disk full") } +func (unwritableBuffer) Pending() ([]audit.Buffered, error) { return nil, nil } +func (unwritableBuffer) Confirm(string) error { return nil } + +func newFileBuffer(t *testing.T) *audit.FileBuffer { + t.Helper() + b, err := audit.NewFileBuffer(filepath.Join(t.TempDir(), "audit-buffer.jsonl")) + require.NoError(t, err) + return b +} + +func entry(principal string) capture.AuditEntry { + return capture.AuditEntry{Principal: principal, EffectiveClassification: "internal", Items: []string{"insight:x"}} +} + +// --- Reserve: classification-aware decision --- + +func TestReserveCentralUpGrantsCentral(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{}, newFileBuffer(t), &fakeNotifier{}) + for _, lvl := range []classification.Level{classification.Public, classification.Internal, classification.Confidential} { + out, err := d.Reserve(context.Background(), lvl) + require.NoError(t, err) + assert.Equal(t, capture.AuditCentral, out) + } +} + +func TestReserveConfidentialSinkDownRefuses(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{down: true}, newFileBuffer(t), &fakeNotifier{}) + _, err := d.Reserve(context.Background(), classification.Confidential) + require.Error(t, err, "confidential + sink down → refuse, no buffer") +} + +func TestReserveInternalSinkDownBuffers(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{down: true}, newFileBuffer(t), &fakeNotifier{}) + out, err := d.Reserve(context.Background(), classification.Internal) + require.NoError(t, err) + assert.Equal(t, capture.AuditBuffered, out) +} + +func TestReserveFloorRefusesWhenNothingCanRecord(t *testing.T) { + d := audit.NewDegradingSink(&fakeCentral{down: true}, unwritableBuffer{}, &fakeNotifier{}) + _, err := d.Reserve(context.Background(), classification.Internal) + require.Error(t, err, "central down AND buffer unwritable → floor refuse") +} + +// --- Record: executes the reserved outcome --- + +func TestRecordCentralPushes(t *testing.T) { + c := &fakeCentral{} + d := audit.NewDegradingSink(c, newFileBuffer(t), &fakeNotifier{}) + require.NoError(t, d.Record(context.Background(), entry("p"), capture.AuditCentral)) + assert.Len(t, c.pushed, 1) +} + +func TestRecordBufferedAppendsAndNotifies(t *testing.T) { + buf := newFileBuffer(t) + nt := &fakeNotifier{} + d := audit.NewDegradingSink(&fakeCentral{down: true}, buf, nt) + require.NoError(t, d.Record(context.Background(), entry("p"), capture.AuditBuffered)) + + pending, err := buf.Pending() + require.NoError(t, err) + assert.Len(t, pending, 1) + assert.NotEmpty(t, nt.msgs, "degraded state alerts via ntfy") +} + +// --- FileBuffer durability + Confirm --- + +func TestFileBufferSurvivesRestart(t *testing.T) { + path := filepath.Join(t.TempDir(), "buf.jsonl") + b1, err := audit.NewFileBuffer(path) + require.NoError(t, err) + require.NoError(t, b1.Append(entry("p1"))) + require.NoError(t, b1.Append(entry("p2"))) + + // "restart": a fresh FileBuffer over the same file sees the records. + b2, err := audit.NewFileBuffer(path) + require.NoError(t, err) + pending, err := b2.Pending() + require.NoError(t, err) + assert.Len(t, pending, 2) +} + +func TestFileBufferConfirmRemovesOnlyThatRecord(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("keep"))) + require.NoError(t, buf.Append(entry("drop"))) + + pending, _ := buf.Pending() + require.Len(t, pending, 2) + var dropID string + for _, p := range pending { + if p.Entry.Principal == "drop" { + dropID = p.ID + } + } + require.NoError(t, buf.Confirm(dropID)) + + after, _ := buf.Pending() + require.Len(t, after, 1) + assert.Equal(t, "keep", after[0].Entry.Principal) +} + +// --- Reconcile --- + +func TestReconcileReplaysAndClearsOnlyAfterConfirmedWrite(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("a"))) + require.NoError(t, buf.Append(entry("b"))) + c := &fakeCentral{} // up + nt := &fakeNotifier{} + + n, err := audit.Reconcile(context.Background(), c, buf, nt) + require.NoError(t, err) + assert.Equal(t, 2, n) + assert.Len(t, c.pushed, 2, "buffered records replayed to central") + + pending, _ := buf.Pending() + assert.Empty(t, pending, "buffer cleared after confirmed central writes") +} + +func TestReconcileNoopWhenCentralDown(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("a"))) + n, err := audit.Reconcile(context.Background(), &fakeCentral{down: true}, buf, &fakeNotifier{}) + require.NoError(t, err) + assert.Equal(t, 0, n) + pending, _ := buf.Pending() + assert.Len(t, pending, 1, "records stay buffered while central is down") +} + +func TestReconcileKeepsRecordWhenPushFails(t *testing.T) { + buf := newFileBuffer(t) + require.NoError(t, buf.Append(entry("a"))) + // Ready ok but Push fails → record must remain buffered (not lost). + c := &fakeCentral{pushErr: errors.New("push rejected")} + n, err := audit.Reconcile(context.Background(), c, buf, &fakeNotifier{}) + require.NoError(t, err) + assert.Equal(t, 0, n) + pending, _ := buf.Pending() + assert.Len(t, pending, 1) +} diff --git a/ingestion/internal/audit/loki.go b/ingestion/internal/audit/loki.go new file mode 100644 index 0000000..176a369 --- /dev/null +++ b/ingestion/internal/audit/loki.go @@ -0,0 +1,99 @@ +package audit + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" + + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" +) + +// LokiCentral pushes capture audit records to a Grafana Loki instance via +// its push API, and probes readiness via /ready. It is the central audit +// substrate behind DegradingSink. +type LokiCentral struct { + baseURL string + labels map[string]string + http *http.Client +} + +// NewLokiCentral constructs a LokiCentral for the given base URL (e.g. +// http://loki:3100). Returns nil when baseURL is empty so callers can +// treat missing config as "no central sink" with a single nil check. +func NewLokiCentral(baseURL string) *LokiCentral { + if baseURL == "" { + return nil + } + return &LokiCentral{ + baseURL: strings.TrimRight(baseURL, "/"), + labels: map[string]string{"service": "brain-capture", "kind": "audit"}, + http: &http.Client{Timeout: 10 * time.Second}, + } +} + +// Ready probes Loki's readiness endpoint. +func (l *LokiCentral) Ready(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, l.baseURL+"/ready", nil) + if err != nil { + return err + } + resp, err := l.http.Do(req) + if err != nil { + return fmt.Errorf("loki not ready: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("loki not ready: status %d", resp.StatusCode) + } + return nil +} + +// pushPayload is the Loki push API body: one stream, one entry whose line +// is the JSON-encoded audit record. +type pushPayload struct { + Streams []lokiStream `json:"streams"` +} + +type lokiStream struct { + Stream map[string]string `json:"stream"` + Values [][2]string `json:"values"` +} + +// Push writes one audit record to Loki as a structured log line. +func (l *LokiCentral) Push(ctx context.Context, e capture.AuditEntry) error { + line, err := json.Marshal(e) + if err != nil { + return fmt.Errorf("marshal audit entry: %w", err) + } + ts := e.Timestamp + if ts.IsZero() { + ts = time.Now() + } + body, err := json.Marshal(pushPayload{Streams: []lokiStream{{ + Stream: l.labels, + Values: [][2]string{{strconv.FormatInt(ts.UTC().UnixNano(), 10), string(line)}}, + }}}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + l.baseURL+"/loki/api/v1/push", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := l.http.Do(req) + if err != nil { + return fmt.Errorf("loki push: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("loki push: status %d", resp.StatusCode) + } + return nil +} diff --git a/ingestion/internal/audit/loki_ntfy_test.go b/ingestion/internal/audit/loki_ntfy_test.go new file mode 100644 index 0000000..1c8d2f0 --- /dev/null +++ b/ingestion/internal/audit/loki_ntfy_test.go @@ -0,0 +1,86 @@ +package audit_test + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/mathiasbq/hyperguild/ingestion/internal/audit" + "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLokiReadyAndPush(t *testing.T) { + var pushBody string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/ready": + w.WriteHeader(http.StatusOK) + case "/loki/api/v1/push": + b, _ := io.ReadAll(r.Body) + pushBody = string(b) + w.WriteHeader(http.StatusNoContent) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer srv.Close() + + c := audit.NewLokiCentral(srv.URL) + require.NotNil(t, c) + require.NoError(t, c.Ready(context.Background())) + + err := c.Push(context.Background(), capture.AuditEntry{ + Principal: "koala-cli", EffectiveClassification: "internal", Items: []string{"insight:x"}, + }) + require.NoError(t, err) + assert.Contains(t, pushBody, "streams") + assert.Contains(t, pushBody, "koala-cli", "audit entry serialised into the loki line") +} + +func TestLokiReadyFailsWhenDown(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + require.Error(t, audit.NewLokiCentral(srv.URL).Ready(context.Background())) +} + +func TestLokiNilWhenUnconfigured(t *testing.T) { + assert.Nil(t, audit.NewLokiCentral("")) +} + +func TestNtfyNotify(t *testing.T) { + var gotBody, gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + gotBody = string(b) + gotAuth = r.Header.Get("Authorization") + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + n := audit.NewNtfyNotifier(srv.URL, "ntfy-token") + require.NotNil(t, n) + require.NoError(t, n.Notify(context.Background(), "audit buffered locally")) + assert.Contains(t, gotBody, "audit buffered locally") + assert.Equal(t, "Bearer ntfy-token", gotAuth) +} + +func TestNtfyDoesNotLeakTokenOnError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + err := audit.NewNtfyNotifier(srv.URL, "secret-token").Notify(context.Background(), "x") + require.Error(t, err) + assert.False(t, strings.Contains(err.Error(), "secret-token"), "token must not leak into errors") +} + +func TestNtfyNilWhenUnconfigured(t *testing.T) { + assert.Nil(t, audit.NewNtfyNotifier("", "tok")) +} diff --git a/ingestion/internal/audit/ntfy.go b/ingestion/internal/audit/ntfy.go new file mode 100644 index 0000000..7aea375 --- /dev/null +++ b/ingestion/internal/audit/ntfy.go @@ -0,0 +1,55 @@ +package audit + +import ( + "context" + "fmt" + "net/http" + "strings" + "time" +) + +// NtfyNotifier posts alerts to an ntfy topic URL. Used to surface a +// degraded audit state (records buffered locally during a loki outage). +type NtfyNotifier struct { + topicURL string + token string + http *http.Client +} + +// NewNtfyNotifier constructs a notifier for the given ntfy topic URL +// (e.g. https://ntfy.sh/my-topic). token is an optional bearer for +// protected ntfy instances; it is held here and only sent in the +// Authorization header, never logged. Returns nil when topicURL is empty. +func NewNtfyNotifier(topicURL, token string) *NtfyNotifier { + if topicURL == "" { + return nil + } + return &NtfyNotifier{ + topicURL: strings.TrimRight(topicURL, "/"), + token: token, + http: &http.Client{Timeout: 10 * time.Second}, + } +} + +// Notify posts a message to the ntfy topic. +func (n *NtfyNotifier) Notify(ctx context.Context, msg string) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.topicURL, strings.NewReader(msg)) + if err != nil { + return err + } + req.Header.Set("Title", "brain-capture audit degraded") + req.Header.Set("Priority", "high") + req.Header.Set("Tags", "warning,brain") + if n.token != "" { + req.Header.Set("Authorization", "Bearer "+n.token) + } + resp, err := n.http.Do(req) + if err != nil { + return fmt.Errorf("ntfy notify: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("ntfy notify: status %d", resp.StatusCode) + } + return nil +} diff --git a/ingestion/internal/audit/reconcile.go b/ingestion/internal/audit/reconcile.go new file mode 100644 index 0000000..845286e --- /dev/null +++ b/ingestion/internal/audit/reconcile.go @@ -0,0 +1,65 @@ +package audit + +import ( + "context" + "fmt" + "log/slog" + "time" +) + +// Reconcile replays locally-buffered audit records to the central sink +// when it is reachable again. A record is removed from the buffer ONLY +// after its central write is confirmed, so a crash mid-reconcile re-plays +// rather than loses. Returns the number of records reconciled. +// +// A no-op (0, nil) when the central sink is still unreachable or the +// buffer is empty. +func Reconcile(ctx context.Context, central Central, buffer Buffer, notifier Notifier) (int, error) { + if err := central.Ready(ctx); err != nil { + return 0, nil // still down; try again next tick + } + pending, err := buffer.Pending() + if err != nil { + return 0, fmt.Errorf("read buffer: %w", err) + } + reconciled := 0 + for _, rec := range pending { + if err := central.Push(ctx, rec.Entry); err != nil { + // Central went away mid-drain; stop and keep the rest buffered. + break + } + if err := buffer.Confirm(rec.ID); err != nil { + return reconciled, fmt.Errorf("confirm buffered record %s: %w", rec.ID, err) + } + reconciled++ + } + if reconciled > 0 && notifier != nil { + _ = notifier.Notify(ctx, fmt.Sprintf("reconciled %d buffered capture audit record(s) to loki", reconciled)) + } + return reconciled, nil +} + +// StartReconcile runs Reconcile on a ticker until ctx is cancelled. It is +// the recovery half of the degrade-and-buffer path; pair it with a +// DegradingSink sharing the same buffer + central. +func StartReconcile(ctx context.Context, central Central, buffer Buffer, notifier Notifier, interval time.Duration) { + if interval <= 0 { + interval = time.Minute + } + go func() { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if n, err := Reconcile(ctx, central, buffer, notifier); err != nil { + slog.Warn("audit reconcile failed", "err", err) + } else if n > 0 { + slog.Info("audit reconcile", "reconciled", n) + } + } + } + }() +} diff --git a/ingestion/internal/audit/slog.go b/ingestion/internal/audit/slog.go index b7d57d2..dda0f4b 100644 --- a/ingestion/internal/audit/slog.go +++ b/ingestion/internal/audit/slog.go @@ -11,11 +11,14 @@ import ( "log/slog" "github.com/mathiasbq/hyperguild/ingestion/internal/capture" + "github.com/mathiasbq/hyperguild/ingestion/internal/classification" ) -// SlogSink records audit entries to an slog.Logger. It never fails, so it -// does not exercise the I5 floor (refuse-if-unauditable) — that is #54's -// loki+buffer sink. A nil logger falls back to slog.Default(). +// SlogSink records audit entries to an slog.Logger. It never fails and is +// always centrally available, so its Reserve always grants AuditCentral — +// it does not exercise the I5 degradation/floor. That is DegradingSink's +// job (loki + durable buffer). SlogSink is the default for deployments +// without a loki endpoint configured. A nil logger ⇒ slog.Default(). type SlogSink struct { logger *slog.Logger } @@ -28,10 +31,15 @@ func NewSlogSink(logger *slog.Logger) *SlogSink { return &SlogSink{logger: logger} } +// Reserve always grants central recording — slog is always available. +func (s *SlogSink) Reserve(_ context.Context, _ classification.Level) (capture.AuditOutcome, error) { + return capture.AuditCentral, nil +} + // Record emits the audit entry at info level. Security events, when // present, are logged at warn level so they surface independently of the // routine audit stream. -func (s *SlogSink) Record(_ context.Context, e capture.AuditEntry) error { +func (s *SlogSink) Record(_ context.Context, e capture.AuditEntry, _ capture.AuditOutcome) error { s.logger.Info("capture audit", "principal", e.Principal, "actor", e.Actor, diff --git a/ingestion/internal/audit/slog_test.go b/ingestion/internal/audit/slog_test.go index 687528d..b1ab29a 100644 --- a/ingestion/internal/audit/slog_test.go +++ b/ingestion/internal/audit/slog_test.go @@ -22,7 +22,7 @@ func TestSlogSinkRecordsEntryAndSecurityEvents(t *testing.T) { EffectiveClassification: "confidential", Items: []string{"insight:wiki/a/facts/x.md"}, SecurityEvents: []string{"asserted-vs-derived origin mismatch"}, - }) + }, capture.AuditCentral) require.NoError(t, err) out := buf.String() @@ -36,6 +36,6 @@ func TestSlogSinkRecordsEntryAndSecurityEvents(t *testing.T) { func TestSlogSinkNilLoggerDefaults(t *testing.T) { // nil logger must not panic. require.NotPanics(t, func() { - _ = audit.NewSlogSink(nil).Record(context.Background(), capture.AuditEntry{}) + _ = audit.NewSlogSink(nil).Record(context.Background(), capture.AuditEntry{}, capture.AuditCentral) }) }