mathiasandClaude Opus 4.8
77f5e06d6b
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) <noreply@anthropic.com>
97 lines
3.4 KiB
Go
97 lines
3.4 KiB
Go
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)
|
|
}
|
|
}
|