From 38a2e91002144d88ee84904622f7d5106a13443a Mon Sep 17 00:00:00 2001 From: Mathias Date: Mon, 22 Jun 2026 23:54:24 +0200 Subject: [PATCH] feat(capturehttp): 503 on audit-unavailable; wire degrading sink (#54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Map capture.ErrAuditUnavailable → HTTP 503 (audit substrate down / confidential unauditable / floor). - main: buildAuditSink selects the DegradingSink (loki central + durable file buffer under brainDir + optional ntfy) when BRAIN_LOKI_URL is set and starts the reconcile loop; else the plain slog sink. Notifier kept as a nil interface (not typed-nil) when unconfigured so the sink and reconcile skip it cleanly. Env: BRAIN_LOKI_URL, BRAIN_NTFY_URL, BRAIN_NTFY_TOKEN, BRAIN_AUDIT_RECONCILE_INTERVAL (default 60s). Buffer at /.audit-buffer/capture.jsonl. Co-Authored-By: Claude Opus 4.8 (1M context) --- ingestion/cmd/server/main.go | 42 ++++++++++++++++--- ingestion/internal/capturehttp/handler.go | 4 ++ .../internal/capturehttp/handler_test.go | 20 +++++++++ 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/ingestion/cmd/server/main.go b/ingestion/cmd/server/main.go index f513463..9e1de43 100644 --- a/ingestion/cmd/server/main.go +++ b/ingestion/cmd/server/main.go @@ -8,6 +8,7 @@ import ( "net/http" "net/url" "os" + "path/filepath" "strconv" "strings" "time" @@ -15,11 +16,11 @@ import ( chassisauth "gitea.d-ma.be/mathias/mcp-chassis/auth" "github.com/mathiasbq/hyperguild/ingestion/internal/api" - "github.com/mathiasbq/hyperguild/ingestion/internal/claudewatcher" "github.com/mathiasbq/hyperguild/ingestion/internal/audit" "github.com/mathiasbq/hyperguild/ingestion/internal/capture" "github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp" "github.com/mathiasbq/hyperguild/ingestion/internal/classification" + "github.com/mathiasbq/hyperguild/ingestion/internal/claudewatcher" "github.com/mathiasbq/hyperguild/ingestion/internal/embed" "github.com/mathiasbq/hyperguild/ingestion/internal/gitea" "github.com/mathiasbq/hyperguild/ingestion/internal/graphstore" @@ -123,6 +124,35 @@ func envInt(key string, fallback int) int { return fallback } +// buildAuditSink selects the capture audit sink. When BRAIN_LOKI_URL is +// set it builds the classification-aware DegradingSink (loki central + +// durable file buffer + optional ntfy) and starts the reconcile loop; +// otherwise it falls back to a plain slog sink. The buffer lives under the +// brain dir so it survives process restarts. +func buildAuditSink(ctx context.Context, brainDir string, logger *slog.Logger) capture.AuditSink { + lokiURL := os.Getenv("BRAIN_LOKI_URL") + central := audit.NewLokiCentral(lokiURL) + if central == nil { + logger.Info("capture audit: slog sink (BRAIN_LOKI_URL unset)") + return audit.NewSlogSink(logger) + } + buffer, err := audit.NewFileBuffer(filepath.Join(brainDir, ".audit-buffer", "capture.jsonl")) + if err != nil { + logger.Error("capture audit buffer init", "err", err) + os.Exit(1) + } + // Keep notifier as a nil interface (not a typed-nil) when unconfigured + // so DegradingSink/Reconcile skip it cleanly. + var notifier audit.Notifier + if n := audit.NewNtfyNotifier(os.Getenv("BRAIN_NTFY_URL"), os.Getenv("BRAIN_NTFY_TOKEN")); n != nil { + notifier = n + } + reconcileInterval := time.Duration(envInt("BRAIN_AUDIT_RECONCILE_INTERVAL", 60)) * time.Second + audit.StartReconcile(ctx, central, buffer, notifier, reconcileInterval) + logger.Info("capture audit: loki+buffer sink", "loki", lokiURL, "reconcile_s", int(reconcileInterval.Seconds())) + return audit.NewDegradingSink(central, buffer, notifier) +} + // splitList parses a comma-separated env value into a trimmed, // empty-free slice. Used for the capture sovereign-principal allowlist. func splitList(v string) []string { @@ -365,11 +395,12 @@ func main() { mux.Handle("/mcp", chassisauth.BearerMiddleware(mcpToken, jwtValidator, "brain", resourceMetadataURL, mcpSrv)) - // POST /capture (#53): the uniform capture REST door. Needs a ticket + // POST /capture (#53/#54): the uniform capture REST door. Needs a ticket // tracker to file action items, so it only mounts when Gitea is // configured. It reuses the MCP server's graph-wired brain store (one - // implementation), the classification tags for the I1 gate, and a slog - // audit sink (the loki+buffer sink lands in #54). The handler does its + // implementation), the classification tags for the I1 gate, and a + // classification-aware audit sink (loki + durable buffer + ntfy when + // BRAIN_LOKI_URL is set, else a plain slog sink). The handler does its // own auth (static + JWT) because it needs the principal to derive the // trust-zone origin — the chassis middleware hides it. if tracker := mcpSrv.IssueTracker(); tracker != nil { @@ -378,8 +409,9 @@ func main() { logger.Error("load classification config", "err", cerr) os.Exit(1) } + auditSink := buildAuditSink(ctx, brainDir, logger) captureSvc := capture.NewService( - mcpSrv.BrainStore(), tracker, nil, classCfg, audit.NewSlogSink(logger)) + mcpSrv.BrainStore(), tracker, nil, classCfg, auditSink) sovereign := splitList(os.Getenv("BRAIN_CAPTURE_SOVEREIGN_PRINCIPALS")) captureH := capturehttp.New(captureSvc, jwtValidator, mcpToken, "local-cli", capturehttp.NewOriginResolver(sovereign)) diff --git a/ingestion/internal/capturehttp/handler.go b/ingestion/internal/capturehttp/handler.go index 8bc73f0..2fd1c73 100644 --- a/ingestion/internal/capturehttp/handler.go +++ b/ingestion/internal/capturehttp/handler.go @@ -113,6 +113,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { case errors.Is(err, capture.ErrSovereigntyRefused): writeJSON(w, http.StatusForbidden, map[string]string{"error": err.Error()}) return + case errors.Is(err, capture.ErrAuditUnavailable): + // I5 refusal: confidential + audit sink down, or the all-tiers floor. + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": err.Error()}) + return case err != nil: // Pre-write validation failure (fail-closed). writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()}) diff --git a/ingestion/internal/capturehttp/handler_test.go b/ingestion/internal/capturehttp/handler_test.go index a2247cc..e33342c 100644 --- a/ingestion/internal/capturehttp/handler_test.go +++ b/ingestion/internal/capturehttp/handler_test.go @@ -176,3 +176,23 @@ func TestCallerCannotForgeOrigin(t *testing.T) { }) assert.Equal(t, http.StatusForbidden, rr.Code) } + +// refusingAudit refuses at Reserve (e.g. confidential + loki down, or floor). +type refusingAudit struct{} + +func (refusingAudit) Reserve(context.Context, classification.Level) (capture.AuditOutcome, error) { + return 0, errors.New("central audit sink unreachable") +} +func (refusingAudit) Record(context.Context, capture.AuditEntry, capture.AuditOutcome) error { + return nil +} + +func TestAuditUnavailableIs503(t *testing.T) { + cfg, err := classification.Load(t.TempDir()) + require.NoError(t, err) + svc := capture.NewService(brainstore.New(t.TempDir()), fakeTracker{}, nil, cfg, refusingAudit{}) + h := capturehttp.New(svc, nil, staticTok, "local-cli", capturehttp.NewOriginResolver(nil)) + + rr := do(t, h, "Bearer "+staticTok, internalReq()) + assert.Equal(t, http.StatusServiceUnavailable, rr.Code) +}