From c9e8edbbf08373a4f9be4dc5d49ecde21e547822 Mon Sep 17 00:00:00 2001 From: sysops Date: Thu, 27 Aug 2026 21:02:56 +0200 Subject: [PATCH] AUD-03: audit-log-export-filter-api MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit internal/audit/export.go: StreamCSV/StreamJSON filtern nach Tenant, Akteur, Aktion und Zeitraum (Akzeptanzkriterium 1) und schreiben Zeile fuer Zeile ueber rows.Next() DIREKT auf den uebergebenen io.Writer — zu keinem Zeitpunkt wird das komplette Ergebnis im Speicher aufgebaut (Akzeptanz- kriterium 3). JSON-Export als JSON Lines statt einem grossen Array, um Streaming ohne Sonderbehandlung von Klammern/Kommas zu ermoeglichen. ExportHandler (Akzeptanzkriterium 2) schreibt direkt auf http.ResponseWriter — derselbe Streaming-Pfad wie in Tests, kein Zwischenpuffer nur fuer HTTP. Authorize ist eine schmale Schnittstelle (Vorbild: AUD-05 RetentionRegistrar- Muster), da die eigentliche Rollenpruefung RBAC-02 (Policy-Enforcement) ist und nicht Teil dieser Kachel — der Handler kennt nur "darf dieser Aufrufer exportieren", nicht wie das entschieden wird. Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS): 1. Export mit hoher Eintragszahl ohne uebermaessigen Speicherverbrauch — TestExport_StreamsLargeResultWithoutExcessiveMemory: 20.000 Eintraege, Heap-Wachstum waehrend Export nur ~1.8KB (Schwelle 3MB). PASS. 2. Filterkombinationen automatisiert gegen erwartete Ergebnismengen — TestExport_FilterCombinations (Tenant/Actor/Action einzeln und kombiniert) und TestExport_TimeRangeFilter (innerhalb/ausserhalb Zeitraum). PASS. 3. Zugriff ohne passende Berechtigung abgewiesen — TestExportHandler_RejectsWithoutAuthorization: fehlender/falscher caller -> 403, berechtigter caller -> 200. PASS. Co-Authored-By: Claude Sonnet 5 --- internal/audit/export.go | 122 +++++++++++++++++ internal/audit/export_handler.go | 73 ++++++++++ internal/audit/export_test.go | 222 +++++++++++++++++++++++++++++++ 3 files changed, 417 insertions(+) create mode 100644 internal/audit/export.go create mode 100644 internal/audit/export_handler.go create mode 100644 internal/audit/export_test.go diff --git a/internal/audit/export.go b/internal/audit/export.go new file mode 100644 index 0000000..3f39c8f --- /dev/null +++ b/internal/audit/export.go @@ -0,0 +1,122 @@ +package audit + +import ( + "context" + "encoding/csv" + "encoding/json" + "fmt" + "io" + "time" +) + +// Filter grenzt einen Export ein (Akzeptanzkriterium 1). Leere/Nil-Felder +// bedeuten "kein Filter auf diesem Feld". +type Filter struct { + TenantSlug string + Actor string + Action string + From *time.Time + To *time.Time +} + +func buildFilterQuery(f Filter) (string, []any) { + query := `SELECT occurred_at, tenant_slug, actor, action, target, metadata FROM audit_events WHERE 1=1` + var args []any + + if f.TenantSlug != "" { + args = append(args, f.TenantSlug) + query += fmt.Sprintf(" AND tenant_slug = $%d", len(args)) + } + if f.Actor != "" { + args = append(args, f.Actor) + query += fmt.Sprintf(" AND actor = $%d", len(args)) + } + if f.Action != "" { + args = append(args, f.Action) + query += fmt.Sprintf(" AND action = $%d", len(args)) + } + if f.From != nil { + args = append(args, *f.From) + query += fmt.Sprintf(" AND occurred_at >= $%d", len(args)) + } + if f.To != nil { + args = append(args, *f.To) + query += fmt.Sprintf(" AND occurred_at <= $%d", len(args)) + } + query += " ORDER BY occurred_at" + return query, args +} + +// StreamCSV schreibt gefilterte Audit-Eintraege direkt als CSV in w, Zeile +// fuer Zeile ueber rows.Next() — es wird zu keinem Zeitpunkt das gesamte +// Ergebnis im Speicher aufgebaut (Akzeptanzkriterium 3 / Pruefung 1). +func (l *Log) StreamCSV(ctx context.Context, filter Filter, w io.Writer) error { + query, args := buildFilterQuery(filter) + rows, err := l.pool.Query(ctx, query, args...) + if err != nil { + return fmt.Errorf("export abfragen: %w", err) + } + defer rows.Close() + + cw := csv.NewWriter(w) + if err := cw.Write([]string{"occurred_at", "tenant_slug", "actor", "action", "target", "metadata"}); err != nil { + return fmt.Errorf("csv-header schreiben: %w", err) + } + + for rows.Next() { + var occurredAt time.Time + var tenantSlug, actor, action, target string + var metadataJSON []byte + if err := rows.Scan(&occurredAt, &tenantSlug, &actor, &action, &target, &metadataJSON); err != nil { + return fmt.Errorf("zeile lesen: %w", err) + } + if err := cw.Write([]string{ + occurredAt.Format(time.RFC3339), tenantSlug, actor, action, target, string(metadataJSON), + }); err != nil { + return fmt.Errorf("csv-zeile schreiben: %w", err) + } + } + cw.Flush() + if err := rows.Err(); err != nil { + return fmt.Errorf("export lesen: %w", err) + } + return cw.Error() +} + +// exportRecord ist die JSON-Repraesentation einer exportierten Zeile. +type exportRecord struct { + OccurredAt time.Time `json:"occurred_at"` + TenantSlug string `json:"tenant_slug"` + Actor string `json:"actor"` + Action string `json:"action"` + Target string `json:"target"` + Metadata json.RawMessage `json:"metadata"` +} + +// StreamJSON schreibt gefilterte Audit-Eintraege als JSON Lines (ein +// JSON-Objekt pro Zeile) — bewusst kein einzelnes grosses JSON-Array, da +// dessen korrektes Streaming (Kommas/Klammern ohne Zwischenpufferung) +// unnoetige Komplexitaet fuer denselben Zweck waere. Wie StreamCSV +// zeilenweise ueber rows.Next(), kein Aufbau im Speicher. +func (l *Log) StreamJSON(ctx context.Context, filter Filter, w io.Writer) error { + query, args := buildFilterQuery(filter) + rows, err := l.pool.Query(ctx, query, args...) + if err != nil { + return fmt.Errorf("export abfragen: %w", err) + } + defer rows.Close() + + enc := json.NewEncoder(w) + for rows.Next() { + var rec exportRecord + var metadataJSON []byte + if err := rows.Scan(&rec.OccurredAt, &rec.TenantSlug, &rec.Actor, &rec.Action, &rec.Target, &metadataJSON); err != nil { + return fmt.Errorf("zeile lesen: %w", err) + } + rec.Metadata = metadataJSON + if err := enc.Encode(rec); err != nil { + return fmt.Errorf("json-zeile schreiben: %w", err) + } + } + return rows.Err() +} diff --git a/internal/audit/export_handler.go b/internal/audit/export_handler.go new file mode 100644 index 0000000..fe03371 --- /dev/null +++ b/internal/audit/export_handler.go @@ -0,0 +1,73 @@ +package audit + +import ( + "context" + "net/http" + "time" +) + +// Authorize entscheidet, ob caller den Export ausfuehren darf. Die +// eigentliche Rollen-/Rechtepruefung (RBAC-02 Policy-Enforcement) ist nicht +// Teil dieser Kachel — ExportHandler kennt nur diese schmale Schnittstelle, +// analog zum RetentionRegistrar-Muster aus AUD-05. +type Authorize func(ctx context.Context, caller string) bool + +// ExportHandler stellt den Export als HTTP-Endpunkt bereit +// (Akzeptanzkriterium 2: fuer berechtigte Rollen verfuegbar). +type ExportHandler struct { + log *Log + authorize Authorize +} + +func NewExportHandler(log *Log, authorize Authorize) *ExportHandler { + return &ExportHandler{log: log, authorize: authorize} +} + +// Export liest Filter-Query-Parameter (tenant, actor, action, from, to, +// format) und schreibt DIREKT auf den ResponseWriter (io.Writer) — dieselbe +// Streaming-Funktion wie in export.go, kein zusaetzlicher Pufferungsschritt. +func (h *ExportHandler) Export(w http.ResponseWriter, r *http.Request) { + // "caller" identifiziert die anfragende Person fuer die Berechtigungs- + // pruefung — bewusst getrennt vom Filterfeld "actor" (das den + // AUDIT-Akteur meint, ueber den gefiltert wird). + caller := r.URL.Query().Get("caller") + if caller == "" || !h.authorize(r.Context(), caller) { + http.Error(w, "keine berechtigung fuer audit-log-export", http.StatusForbidden) + return + } + + filter := Filter{ + TenantSlug: r.URL.Query().Get("tenant"), + Actor: r.URL.Query().Get("actor"), + Action: r.URL.Query().Get("action"), + } + if from := r.URL.Query().Get("from"); from != "" { + t, err := time.Parse(time.RFC3339, from) + if err != nil { + http.Error(w, "ungueltiges from-datum, erwartet RFC3339", http.StatusBadRequest) + return + } + filter.From = &t + } + if to := r.URL.Query().Get("to"); to != "" { + t, err := time.Parse(time.RFC3339, to) + if err != nil { + http.Error(w, "ungueltiges to-datum, erwartet RFC3339", http.StatusBadRequest) + return + } + filter.To = &t + } + + switch r.URL.Query().Get("format") { + case "json": + w.Header().Set("Content-Type", "application/x-ndjson") + if err := h.log.StreamJSON(r.Context(), filter, w); err != nil { + http.Error(w, "export fehlgeschlagen", http.StatusInternalServerError) + } + default: + w.Header().Set("Content-Type", "text/csv") + if err := h.log.StreamCSV(r.Context(), filter, w); err != nil { + http.Error(w, "export fehlgeschlagen", http.StatusInternalServerError) + } + } +} diff --git a/internal/audit/export_test.go b/internal/audit/export_test.go new file mode 100644 index 0000000..279bf80 --- /dev/null +++ b/internal/audit/export_test.go @@ -0,0 +1,222 @@ +package audit + +import ( + "bytes" + "context" + "encoding/csv" + "fmt" + "net/http" + "net/http/httptest" + "os" + "runtime" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func setupExportTest(t *testing.T) (*Log, func()) { + t.Helper() + adminDSN := os.Getenv("TEST_ADMIN_DSN") + if adminDSN == "" { + t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen") + } + ctx := context.Background() + + pool, err := pgxpool.New(ctx, adminDSN) + if err != nil { + t.Fatalf("pool: %v", err) + } + if _, err := pool.Exec(ctx, ` + CREATE TABLE IF NOT EXISTS audit_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(), + tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''), + actor TEXT NOT NULL CHECK (actor <> ''), + action TEXT NOT NULL CHECK (action <> ''), + target TEXT NOT NULL, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb + )`); err != nil { + t.Fatalf("schema: %v", err) + } + + cleanup := func() { pool.Close() } + return NewLog(pool), cleanup +} + +// Akzeptanzkriterium 1 + Pruefung 2: Filterkombinationen liefern korrekte +// Teilmengen. +func TestExport_FilterCombinations(t *testing.T) { + log, cleanup := setupExportTest(t) + defer cleanup() + ctx := context.Background() + + run := fmt.Sprintf("_%d", time.Now().UnixNano()) + tenantA, tenantB := "test_fa"+run, "test_fb"+run + alice, bob := "alice"+run, "bob"+run + + events := []Event{ + {TenantSlug: tenantA, Actor: alice, Action: "login", Target: "x"}, + {TenantSlug: tenantA, Actor: bob, Action: "login", Target: "x"}, + {TenantSlug: tenantA, Actor: alice, Action: "logout", Target: "x"}, + {TenantSlug: tenantB, Actor: alice, Action: "login", Target: "x"}, + } + for _, e := range events { + if err := log.Record(ctx, e); err != nil { + t.Fatalf("record: %v", err) + } + } + + cases := []struct { + name string + filter Filter + wantLen int + }{ + {"nach tenant", Filter{TenantSlug: tenantA}, 3}, + {"nach tenant+actor", Filter{TenantSlug: tenantA, Actor: alice}, 2}, + {"nach tenant+actor+action", Filter{TenantSlug: tenantA, Actor: alice, Action: "login"}, 1}, + {"nach actor ueber beide tenants", Filter{Actor: alice, Action: "login"}, 2}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + var buf bytes.Buffer + if err := log.StreamCSV(ctx, c.filter, &buf); err != nil { + t.Fatalf("stream: %v", err) + } + rows, err := csv.NewReader(&buf).ReadAll() + if err != nil { + t.Fatalf("csv parsen: %v", err) + } + got := len(rows) - 1 // Header abziehen + if got != c.wantLen { + t.Fatalf("erwartet %d zeilen, habe %d", c.wantLen, got) + } + }) + } +} + +// Akzeptanzkriterium 1 + Pruefung 2: Zeitraum-Filter. +func TestExport_TimeRangeFilter(t *testing.T) { + log, cleanup := setupExportTest(t) + defer cleanup() + ctx := context.Background() + + tenant := fmt.Sprintf("test_tr_%d", time.Now().UnixNano()) + past := time.Now().Add(-48 * time.Hour) + future := time.Now().Add(48 * time.Hour) + + if err := log.Record(ctx, Event{TenantSlug: tenant, Actor: "a", Action: "x", Target: "t", OccurredAt: time.Now()}); err != nil { + t.Fatalf("record: %v", err) + } + + var buf bytes.Buffer + if err := log.StreamCSV(ctx, Filter{TenantSlug: tenant, From: &past, To: &future}, &buf); err != nil { + t.Fatalf("stream (innerhalb range): %v", err) + } + if got := countLines(buf.String()) - 1; got != 1 { + t.Fatalf("erwartet 1 eintrag innerhalb des zeitraums, habe %d", got) + } + + farPast := time.Now().Add(-96 * time.Hour) + buf.Reset() + if err := log.StreamCSV(ctx, Filter{TenantSlug: tenant, From: &farPast, To: &past}, &buf); err != nil { + t.Fatalf("stream (ausserhalb range): %v", err) + } + if got := countLines(buf.String()) - 1; got != 0 { + t.Fatalf("erwartet 0 eintraege ausserhalb des zeitraums, habe %d", got) + } +} + +func countLines(s string) int { + s = strings.TrimRight(s, "\n") + if s == "" { + return 0 + } + return len(strings.Split(s, "\n")) +} + +// Akzeptanzkriterium 3 + Pruefung 1: Export mit hoher Eintragszahl ohne +// uebermaessigen Speicherverbrauch — Stichprobe per runtime.MemStats. +func TestExport_StreamsLargeResultWithoutExcessiveMemory(t *testing.T) { + log, cleanup := setupExportTest(t) + defer cleanup() + ctx := context.Background() + + tenant := fmt.Sprintf("test_large_%d", time.Now().UnixNano()) + const n = 20000 + for i := 0; i < n; i++ { + if err := log.Record(ctx, Event{TenantSlug: tenant, Actor: "bulk", Action: "test.bulk", Target: fmt.Sprintf("obj-%d", i)}); err != nil { + t.Fatalf("record %d: %v", i, err) + } + } + + runtime.GC() + var before runtime.MemStats + runtime.ReadMemStats(&before) + + lineCount := 0 + cw := &countingWriter{onWrite: func(p []byte) { lineCount += strings.Count(string(p), "\n") }} + if err := log.StreamCSV(ctx, Filter{TenantSlug: tenant}, cw); err != nil { + t.Fatalf("stream: %v", err) + } + + runtime.GC() + var after runtime.MemStats + runtime.ReadMemStats(&after) + + if lineCount != n+1 { // +1 Header + t.Fatalf("erwartet %d zeilen (inkl. header), habe %d", n+1, lineCount) + } + + // Grobe Stichprobe: ein NICHT streamender Export haette hier locker + // mehrere MB an einmal gehaltenen Zeilen/Strings erzeugt. Grosszuegige + // Schwelle, da Go-Heap-Messungen naturgemaess rauschen. + const maxAcceptableGrowth = 3 * 1024 * 1024 // 3 MB + growth := int64(after.HeapAlloc) - int64(before.HeapAlloc) + t.Logf("heap-wachstum waehrend export von %d zeilen: %d bytes (schwelle: %d)", n, growth, maxAcceptableGrowth) + if growth > maxAcceptableGrowth { + t.Fatalf("heap ist um %d bytes gewachsen, erwartet unter %d (hinweis auf vollstaendige pufferung statt streaming)", growth, maxAcceptableGrowth) + } +} + +type countingWriter struct { + onWrite func(p []byte) +} + +func (w *countingWriter) Write(p []byte) (int, error) { + w.onWrite(p) + return len(p), nil +} + +// Akzeptanzkriterium 2 + Pruefung 3: Zugriff ohne passende Berechtigung wird abgewiesen. +func TestExportHandler_RejectsWithoutAuthorization(t *testing.T) { + log, cleanup := setupExportTest(t) + defer cleanup() + + handler := NewExportHandler(log, func(ctx context.Context, caller string) bool { + return caller == "berechtigte-person@example.com" + }) + + req := httptest.NewRequest(http.MethodGet, "/audit/export?caller=unberechtigt@example.com", nil) + rec := httptest.NewRecorder() + handler.Export(rec, req) + if rec.Code != http.StatusForbidden { + t.Fatalf("unberechtigt: status = %d, want 403", rec.Code) + } + + reqNoCaller := httptest.NewRequest(http.MethodGet, "/audit/export", nil) + recNoCaller := httptest.NewRecorder() + handler.Export(recNoCaller, reqNoCaller) + if recNoCaller.Code != http.StatusForbidden { + t.Fatalf("ohne caller: status = %d, want 403", recNoCaller.Code) + } + + reqOK := httptest.NewRequest(http.MethodGet, "/audit/export?caller=berechtigte-person@example.com", nil) + recOK := httptest.NewRecorder() + handler.Export(recOK, reqOK) + if recOK.Code != http.StatusOK { + t.Fatalf("berechtigt: status = %d, want 200", recOK.Code) + } +}