Phase 4: SSO scaffolding, RBAC enforcement, tenant-scoped dashboards, audit logging, K8s deployment
RBAC (api/internal/authz) is live on /query and /dashboards, backed by a new enterprise/ module (session issuance, audit logging, RBAC storage, OIDC/SAML protocol wiring) that core never imports -- only calls over HTTP. Found and fixed a real cross-tenant vulnerability in dashboards (no tenant_id filtering at all) while writing the threat model doc. Two things are explicitly NOT done, documented rather than hidden: tenant isolation for log data itself (/query still shares one ClickHouse connection and Tantivy index across every tenant -- RBAC controls who can query, not what a query can see), and human SSO login (protocol wiring exists, no HTTP handler calls it yet). See docs/security/threat-model.md and docs/phase-4-runbook.md. Also adds deploy/ (Go Operator + Helm chart, validated offline only -- no cluster was reachable in this environment).
This commit is contained in:
@@ -0,0 +1,185 @@
|
||||
// Package audit implements the append-only, hash-chained query audit
|
||||
// log described in /docs/phase-4-isolation-design.md's audit-logging
|
||||
// section. Two independent defenses back the "no update/delete path
|
||||
// from the application layer" requirement -- both verified against a
|
||||
// live Postgres, not just written: audit_writer (this package's own
|
||||
// Postgres role, via its own connection pool, never the shared `sentry`
|
||||
// role every other store uses) has only INSERT+SELECT grants, and a
|
||||
// BEFORE UPDATE OR DELETE trigger (metadata/migrations/0015-0016)
|
||||
// rejects the operation for *any* role, including the table owner --
|
||||
// confirmed live: even `sentry` cannot UPDATE a row without first
|
||||
// disabling the trigger, a privileged operation distinct from ordinary
|
||||
// application access.
|
||||
//
|
||||
// The hash chain (prev_hash/row_hash) proves internal consistency --
|
||||
// detects tampering with existing rows -- but does not by itself prove
|
||||
// truth against a privileged attacker who can rewrite the whole table
|
||||
// and regenerate a self-consistent chain from row 1. See checkpoint.go
|
||||
// for the external-anchoring half of that guarantee.
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type Source string
|
||||
|
||||
const (
|
||||
SourceAPI Source = "api"
|
||||
SourceWeb Source = "web"
|
||||
SourceCLI Source = "cli"
|
||||
SourceAlerting Source = "alerting"
|
||||
)
|
||||
|
||||
type EventType string
|
||||
|
||||
const (
|
||||
EventQuery EventType = "query"
|
||||
EventRoleChange EventType = "role_change"
|
||||
EventGrantChange EventType = "grant_change"
|
||||
EventSSOConfigChange EventType = "sso_config_change"
|
||||
EventSecretReveal EventType = "secret_reveal"
|
||||
)
|
||||
|
||||
type Status string
|
||||
|
||||
const (
|
||||
StatusSuccess Status = "success"
|
||||
StatusError Status = "error"
|
||||
)
|
||||
|
||||
// Entry is what a caller supplies. UserID is nil for system/alerting-
|
||||
// sourced entries (see /docs/phase-4-isolation-design.md's alerting
|
||||
// service-identity finding -- alerting evaluations are audited, but
|
||||
// aren't attributable to a human user).
|
||||
type Entry struct {
|
||||
TenantID string
|
||||
UserID *string
|
||||
Source Source
|
||||
EventType EventType
|
||||
QueryText *string
|
||||
RowCount *int
|
||||
DurationMS *int
|
||||
Status Status
|
||||
ErrorMessage *string
|
||||
Detail json.RawMessage
|
||||
}
|
||||
|
||||
// Record is a written entry plus the fields the store assigned.
|
||||
type Record struct {
|
||||
Entry
|
||||
ID int64
|
||||
PrevHash *string
|
||||
RowHash string
|
||||
}
|
||||
|
||||
// Store writes via a connection pool authenticated as the audit_writer
|
||||
// role -- never the shared pool other stores in this repo use. Passing
|
||||
// a pool opened with any other role's credentials silently defeats the
|
||||
// grant-restriction half of this package's guarantee; there's no way
|
||||
// for this package to verify its own pool's role at runtime, so this is
|
||||
// an integration-time discipline documented here, not something this
|
||||
// code can enforce on itself.
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// advisoryLockKey serializes concurrent Append calls so two writers
|
||||
// never read the same prev_hash and each compute a hash chained off it
|
||||
// -- that would fork the chain. Arbitrary fixed value, held only for
|
||||
// the duration of one transaction (pg_advisory_xact_lock releases
|
||||
// automatically at commit/rollback).
|
||||
const advisoryLockKey = 784129035
|
||||
|
||||
func (s *Store) Append(ctx context.Context, e Entry) (*Record, error) {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("audit: beginning transaction: %w", err)
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
if _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock($1)", advisoryLockKey); err != nil {
|
||||
return nil, fmt.Errorf("audit: acquiring serialization lock: %w", err)
|
||||
}
|
||||
|
||||
var prevHash *string
|
||||
row := tx.QueryRow(ctx, "SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1")
|
||||
if err := row.Scan(&prevHash); err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, fmt.Errorf("audit: reading previous row hash: %w", err)
|
||||
}
|
||||
|
||||
if len(e.Detail) == 0 {
|
||||
e.Detail = json.RawMessage(`{}`)
|
||||
}
|
||||
rec := &Record{Entry: e, PrevHash: prevHash, RowHash: computeHash(prevHash, e)}
|
||||
|
||||
err = tx.QueryRow(ctx, `
|
||||
INSERT INTO audit_log (tenant_id, user_id, source, event_type, query_text, row_count,
|
||||
duration_ms, status, error_message, detail, prev_hash, row_hash)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
|
||||
RETURNING id`,
|
||||
e.TenantID, e.UserID, e.Source, e.EventType, e.QueryText, e.RowCount,
|
||||
e.DurationMS, e.Status, e.ErrorMessage, e.Detail, prevHash, rec.RowHash,
|
||||
).Scan(&rec.ID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("audit: inserting row: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, fmt.Errorf("audit: committing: %w", err)
|
||||
}
|
||||
return rec, nil
|
||||
}
|
||||
|
||||
// computeHash is deliberately a fixed, explicit field order (not "hash
|
||||
// the JSON encoding," which is not guaranteed stable across Go versions
|
||||
// or map key ordering) -- \x00 is used as a field separator since it
|
||||
// cannot appear in any of these string fields in practice, and even if
|
||||
// it somehow did, the goal here is deterministic tamper-detection
|
||||
// against accidental/naive modification, not cryptographic
|
||||
// collision-resistance against a chosen-plaintext adversary.
|
||||
func computeHash(prevHash *string, e Entry) string {
|
||||
h := sha256.New()
|
||||
write := func(s string) {
|
||||
h.Write([]byte(s))
|
||||
h.Write([]byte{0})
|
||||
}
|
||||
write(deref(prevHash))
|
||||
write(e.TenantID)
|
||||
write(deref(e.UserID))
|
||||
write(string(e.Source))
|
||||
write(string(e.EventType))
|
||||
write(deref(e.QueryText))
|
||||
write(intToStr(e.RowCount))
|
||||
write(intToStr(e.DurationMS))
|
||||
write(string(e.Status))
|
||||
write(deref(e.ErrorMessage))
|
||||
write(string(e.Detail))
|
||||
return hex.EncodeToString(h.Sum(nil))
|
||||
}
|
||||
|
||||
func deref(s *string) string {
|
||||
if s == nil {
|
||||
return ""
|
||||
}
|
||||
return *s
|
||||
}
|
||||
|
||||
func intToStr(n *int) string {
|
||||
if n == nil {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("%d", *n)
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Checkpoint is a rolling hash over a range of audit_log rows, chained
|
||||
// to the previous checkpoint the same way row_hash chains individual
|
||||
// rows. The chain in audit_log alone only proves internal consistency:
|
||||
// anyone with enough Postgres privilege to wipe the table can
|
||||
// regenerate a perfectly self-consistent new chain from row 1.
|
||||
// Checkpoints answer a different question -- "does what's in Postgres
|
||||
// right now match what it was an hour ago" -- but only if they're
|
||||
// written somewhere the same privileged actor can't also reach. That's
|
||||
// CheckpointSink's job, not this package's: this package computes
|
||||
// checkpoints correctly and hands them to a sink; it does not claim any
|
||||
// particular sink is actually tamper-proof.
|
||||
type Checkpoint struct {
|
||||
FromID int64
|
||||
ToID int64
|
||||
PrevCheckpointHash string
|
||||
Hash string
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// CheckpointSink persists checkpoints somewhere external. FileSink
|
||||
// (below) is a working, testable implementation appropriate for
|
||||
// development -- a real deployment needs a sink that genuinely isn't
|
||||
// reachable by whatever could tamper with Postgres (S3 with Object
|
||||
// Lock, a separate append-only service, etc.), which this package
|
||||
// deliberately does not implement: that's an operational/infrastructure
|
||||
// decision, not something to hardcode a specific cloud vendor's SDK for
|
||||
// without discussing the dependency first.
|
||||
type CheckpointSink interface {
|
||||
// LastCheckpoint returns the most recently written checkpoint, or
|
||||
// nil if none exists yet.
|
||||
LastCheckpoint(ctx context.Context) (*Checkpoint, error)
|
||||
Write(ctx context.Context, cp Checkpoint) error
|
||||
}
|
||||
|
||||
// Checkpointer periodically rolls up new audit_log rows since the last
|
||||
// checkpoint into a new one.
|
||||
type Checkpointer struct {
|
||||
store *Store
|
||||
sink CheckpointSink
|
||||
}
|
||||
|
||||
func NewCheckpointer(store *Store, sink CheckpointSink) *Checkpointer {
|
||||
return &Checkpointer{store: store, sink: sink}
|
||||
}
|
||||
|
||||
// Run computes and writes at most one new checkpoint covering every
|
||||
// audit_log row added since the last one. Returns (nil, nil) if there's
|
||||
// nothing new to checkpoint. Call on a schedule (e.g. hourly) from
|
||||
// cmd/enterprise-auth -- this package doesn't run its own ticker, same
|
||||
// "caller owns scheduling" shape as /alerting's evaluator.
|
||||
func (c *Checkpointer) Run(ctx context.Context) (*Checkpoint, error) {
|
||||
last, err := c.sink.LastCheckpoint(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("audit: reading last checkpoint: %w", err)
|
||||
}
|
||||
|
||||
fromID := int64(1)
|
||||
prevHash := ""
|
||||
if last != nil {
|
||||
fromID = last.ToID + 1
|
||||
prevHash = last.Hash
|
||||
}
|
||||
|
||||
rowHashes, maxID, err := c.store.rowHashesFrom(ctx, fromID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("audit: reading rows for checkpoint: %w", err)
|
||||
}
|
||||
if len(rowHashes) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
h := sha256.New()
|
||||
h.Write([]byte(prevHash))
|
||||
for _, rh := range rowHashes {
|
||||
h.Write([]byte{0})
|
||||
h.Write([]byte(rh))
|
||||
}
|
||||
|
||||
cp := Checkpoint{
|
||||
FromID: fromID, ToID: maxID,
|
||||
PrevCheckpointHash: prevHash,
|
||||
Hash: hex.EncodeToString(h.Sum(nil)),
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}
|
||||
if err := c.sink.Write(ctx, cp); err != nil {
|
||||
return nil, fmt.Errorf("audit: writing checkpoint: %w", err)
|
||||
}
|
||||
return &cp, nil
|
||||
}
|
||||
|
||||
// rowHashesFrom returns row_hash values for id >= fromID, in id order,
|
||||
// plus the highest id seen (so the caller knows where the next
|
||||
// checkpoint should resume).
|
||||
func (s *Store) rowHashesFrom(ctx context.Context, fromID int64) ([]string, int64, error) {
|
||||
rows, err := s.pool.Query(ctx, "SELECT id, row_hash FROM audit_log WHERE id >= $1 ORDER BY id ASC", fromID)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var hashes []string
|
||||
var maxID int64
|
||||
for rows.Next() {
|
||||
var id int64
|
||||
var hash string
|
||||
if err := rows.Scan(&id, &hash); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
hashes = append(hashes, hash)
|
||||
maxID = id
|
||||
}
|
||||
return hashes, maxID, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// FileSink is a working CheckpointSink appropriate for development and
|
||||
// testing -- appends one JSON line per checkpoint to a local file.
|
||||
// **Not a real external-anchoring guarantee**: a local file on the same
|
||||
// host as Postgres is reachable by exactly the kind of privileged actor
|
||||
// checkpointing is meant to defend against. A production deployment
|
||||
// needs a genuinely separate-trust-domain sink (S3 with Object Lock, a
|
||||
// separate append-only service) -- deliberately not implemented here,
|
||||
// per checkpoint.go's doc comment.
|
||||
type FileSink struct {
|
||||
path string
|
||||
}
|
||||
|
||||
func NewFileSink(path string) *FileSink {
|
||||
return &FileSink{path: path}
|
||||
}
|
||||
|
||||
type fileSinkLine struct {
|
||||
FromID int64 `json:"from_id"`
|
||||
ToID int64 `json:"to_id"`
|
||||
PrevCheckpointHash string `json:"prev_checkpoint_hash"`
|
||||
Hash string `json:"hash"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
func (f *FileSink) Write(_ context.Context, cp Checkpoint) error {
|
||||
file, err := os.OpenFile(f.path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
||||
if err != nil {
|
||||
return fmt.Errorf("audit: opening checkpoint file: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
line := fileSinkLine{
|
||||
FromID: cp.FromID, ToID: cp.ToID,
|
||||
PrevCheckpointHash: cp.PrevCheckpointHash, Hash: cp.Hash, CreatedAt: cp.CreatedAt,
|
||||
}
|
||||
encoded, err := json.Marshal(line)
|
||||
if err != nil {
|
||||
return fmt.Errorf("audit: encoding checkpoint: %w", err)
|
||||
}
|
||||
if _, err := fmt.Fprintln(file, string(encoded)); err != nil {
|
||||
return fmt.Errorf("audit: writing checkpoint: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *FileSink) LastCheckpoint(_ context.Context) (*Checkpoint, error) {
|
||||
file, err := os.Open(f.path)
|
||||
if os.IsNotExist(err) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("audit: opening checkpoint file: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
var lastLine string
|
||||
scanner := bufio.NewScanner(file)
|
||||
for scanner.Scan() {
|
||||
if line := strings.TrimSpace(scanner.Text()); line != "" {
|
||||
lastLine = line
|
||||
}
|
||||
}
|
||||
if err := scanner.Err(); err != nil {
|
||||
return nil, fmt.Errorf("audit: reading checkpoint file: %w", err)
|
||||
}
|
||||
if lastLine == "" {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var line fileSinkLine
|
||||
if err := json.Unmarshal([]byte(lastLine), &line); err != nil {
|
||||
return nil, fmt.Errorf("audit: decoding last checkpoint line: %w", err)
|
||||
}
|
||||
return &Checkpoint{
|
||||
FromID: line.FromID, ToID: line.ToID,
|
||||
PrevCheckpointHash: line.PrevCheckpointHash, Hash: line.Hash, CreatedAt: line.CreatedAt,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestFileSinkRoundTrip(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "checkpoints.jsonl")
|
||||
sink := NewFileSink(path)
|
||||
ctx := context.Background()
|
||||
|
||||
none, err := sink.LastCheckpoint(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("LastCheckpoint on a nonexistent file: %v", err)
|
||||
}
|
||||
if none != nil {
|
||||
t.Fatalf("expected nil for a nonexistent checkpoint file, got %+v", none)
|
||||
}
|
||||
|
||||
cp1 := Checkpoint{FromID: 1, ToID: 10, Hash: "hash1", CreatedAt: time.Now().UTC().Truncate(time.Second)}
|
||||
if err := sink.Write(ctx, cp1); err != nil {
|
||||
t.Fatalf("Write: %v", err)
|
||||
}
|
||||
cp2 := Checkpoint{FromID: 11, ToID: 20, PrevCheckpointHash: "hash1", Hash: "hash2", CreatedAt: time.Now().UTC().Truncate(time.Second)}
|
||||
if err := sink.Write(ctx, cp2); err != nil {
|
||||
t.Fatalf("Write: %v", err)
|
||||
}
|
||||
|
||||
last, err := sink.LastCheckpoint(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("LastCheckpoint: %v", err)
|
||||
}
|
||||
if last == nil {
|
||||
t.Fatalf("expected a checkpoint, got nil")
|
||||
}
|
||||
if last.ToID != cp2.ToID || last.Hash != cp2.Hash {
|
||||
t.Fatalf("got %+v, want the most recently written checkpoint %+v", last, cp2)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,275 @@
|
||||
// Integration tests against a real Postgres, authenticated as the real
|
||||
// audit_writer role -- this package's whole point is a set of guarantees
|
||||
// (grants, the trigger, hash-chain correctness under concurrency) that a
|
||||
// mocked pgxpool can't actually exercise. Skipped unless
|
||||
// AUDIT_TEST_POSTGRES_ADDR is set; run via:
|
||||
//
|
||||
// docker run --rm --network sentry_default -v $(pwd)/../../..:/src -w /src/enterprise \
|
||||
// -e AUDIT_TEST_POSTGRES_ADDR=metadata-postgres:5432 \
|
||||
// -e AUDIT_TEST_POSTGRES_PASSWORD=audit-writer-dev-only \
|
||||
// -e AUDIT_TEST_ADMIN_PASSWORD=sentry-dev-only \
|
||||
// golang:1.25-alpine go test ./internal/audit/... -v
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func testPool(t *testing.T, user, password string) *pgxpool.Pool {
|
||||
t.Helper()
|
||||
addr := os.Getenv("AUDIT_TEST_POSTGRES_ADDR")
|
||||
if addr == "" {
|
||||
t.Skip("AUDIT_TEST_POSTGRES_ADDR not set -- skipping live-Postgres integration test")
|
||||
}
|
||||
dsn := fmt.Sprintf("postgres://%s:%s@%s/sentry_metadata", user, password, addr)
|
||||
pool, err := pgxpool.New(context.Background(), dsn)
|
||||
if err != nil {
|
||||
t.Fatalf("opening pool: %v", err)
|
||||
}
|
||||
t.Cleanup(pool.Close)
|
||||
return pool
|
||||
}
|
||||
|
||||
func cleanupAuditLog(t *testing.T, adminPool *pgxpool.Pool) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
// Errors here were previously swallowed (_, _ =) -- that hid the
|
||||
// real cause of a test failure (rows accumulating across test runs)
|
||||
// behind what looked like a row-count/ID-assumption bug instead.
|
||||
// Surface them.
|
||||
if _, err := adminPool.Exec(ctx, "ALTER TABLE audit_log DISABLE TRIGGER audit_log_immutable"); err != nil {
|
||||
t.Fatalf("cleanup: disabling trigger: %v", err)
|
||||
}
|
||||
tag, err := adminPool.Exec(ctx, "DELETE FROM audit_log")
|
||||
if err != nil {
|
||||
t.Fatalf("cleanup: deleting rows: %v", err)
|
||||
}
|
||||
t.Logf("cleanup: deleted %d pre-existing rows", tag.RowsAffected())
|
||||
if _, err := adminPool.Exec(ctx, "ALTER TABLE audit_log ENABLE TRIGGER audit_log_immutable"); err != nil {
|
||||
t.Fatalf("cleanup: re-enabling trigger: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAppendAndVerifyChainRealPostgres(t *testing.T) {
|
||||
writerPool := testPool(t, "audit_writer", os.Getenv("AUDIT_TEST_POSTGRES_PASSWORD"))
|
||||
adminPool := testPool(t, "sentry", os.Getenv("AUDIT_TEST_ADMIN_PASSWORD"))
|
||||
cleanupAuditLog(t, adminPool)
|
||||
defer cleanupAuditLog(t, adminPool)
|
||||
|
||||
store := NewStore(writerPool)
|
||||
ctx := context.Background()
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
q := fmt.Sprintf("service=api | stats count %d", i)
|
||||
rec, err := store.Append(ctx, Entry{
|
||||
TenantID: "default", Source: SourceAPI, EventType: EventQuery,
|
||||
QueryText: &q, Status: StatusSuccess,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Append %d: %v", i, err)
|
||||
}
|
||||
if rec.RowHash == "" {
|
||||
t.Fatalf("expected a non-empty row hash")
|
||||
}
|
||||
}
|
||||
|
||||
result, err := store.VerifyChain(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("VerifyChain: %v", err)
|
||||
}
|
||||
if !result.OK {
|
||||
t.Fatalf("expected an intact chain, got broken at id=%d after %d rows checked", result.FirstBadID, result.RowsChecked)
|
||||
}
|
||||
if result.RowsChecked != 5 {
|
||||
t.Fatalf("RowsChecked = %d, want 5", result.RowsChecked)
|
||||
}
|
||||
}
|
||||
|
||||
// TestVerifyChainDetectsTampering proves the chain actually catches an
|
||||
// in-place row modification -- not just that VerifyChain runs without
|
||||
// erroring on untampered data, which a bug returning OK unconditionally
|
||||
// would also pass.
|
||||
func TestVerifyChainDetectsTampering(t *testing.T) {
|
||||
writerPool := testPool(t, "audit_writer", os.Getenv("AUDIT_TEST_POSTGRES_PASSWORD"))
|
||||
adminPool := testPool(t, "sentry", os.Getenv("AUDIT_TEST_ADMIN_PASSWORD"))
|
||||
cleanupAuditLog(t, adminPool)
|
||||
defer cleanupAuditLog(t, adminPool)
|
||||
|
||||
store := NewStore(writerPool)
|
||||
ctx := context.Background()
|
||||
|
||||
var lastID int64
|
||||
for i := 0; i < 3; i++ {
|
||||
q := "service=api"
|
||||
rec, err := store.Append(ctx, Entry{TenantID: "default", Source: SourceAPI, EventType: EventQuery, QueryText: &q, Status: StatusSuccess})
|
||||
if err != nil {
|
||||
t.Fatalf("Append: %v", err)
|
||||
}
|
||||
lastID = rec.ID
|
||||
}
|
||||
|
||||
before, err := store.VerifyChain(ctx)
|
||||
if err != nil || !before.OK {
|
||||
t.Fatalf("expected chain to verify before tampering: ok=%v err=%v", before.OK, err)
|
||||
}
|
||||
|
||||
// Simulate tampering: a privileged actor disables the trigger (the
|
||||
// same escape hatch confirmed live in the design doc's verification
|
||||
// -- this is the "even the trigger doesn't stop a superuser" case)
|
||||
// and rewrites a row's status without recomputing the hash chain.
|
||||
if _, err := adminPool.Exec(ctx, "ALTER TABLE audit_log DISABLE TRIGGER audit_log_immutable"); err != nil {
|
||||
t.Fatalf("disabling trigger for the tamper simulation: %v", err)
|
||||
}
|
||||
if _, err := adminPool.Exec(ctx, "UPDATE audit_log SET status = 'error' WHERE id = $1", lastID); err != nil {
|
||||
t.Fatalf("simulated tamper UPDATE: %v", err)
|
||||
}
|
||||
if _, err := adminPool.Exec(ctx, "ALTER TABLE audit_log ENABLE TRIGGER audit_log_immutable"); err != nil {
|
||||
t.Fatalf("re-enabling trigger: %v", err)
|
||||
}
|
||||
|
||||
after, err := store.VerifyChain(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("VerifyChain after tampering: %v", err)
|
||||
}
|
||||
if after.OK {
|
||||
t.Fatalf("expected VerifyChain to detect the tampered row, got OK")
|
||||
}
|
||||
if after.FirstBadID != lastID {
|
||||
t.Fatalf("FirstBadID = %d, want %d", after.FirstBadID, lastID)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppendConcurrentWritesProduceAValidChain exercises the advisory
|
||||
// lock: without it, concurrent Append calls could read the same
|
||||
// prev_hash and fork the chain. Real concurrency, real Postgres, not a
|
||||
// unit test of the Go code alone.
|
||||
func TestAppendConcurrentWritesProduceAValidChain(t *testing.T) {
|
||||
writerPool := testPool(t, "audit_writer", os.Getenv("AUDIT_TEST_POSTGRES_PASSWORD"))
|
||||
adminPool := testPool(t, "sentry", os.Getenv("AUDIT_TEST_ADMIN_PASSWORD"))
|
||||
cleanupAuditLog(t, adminPool)
|
||||
defer cleanupAuditLog(t, adminPool)
|
||||
|
||||
store := NewStore(writerPool)
|
||||
ctx := context.Background()
|
||||
|
||||
const n = 20
|
||||
var wg sync.WaitGroup
|
||||
errs := make(chan error, n)
|
||||
for i := 0; i < n; i++ {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
q := fmt.Sprintf("query-%d", i)
|
||||
_, err := store.Append(ctx, Entry{TenantID: "default", Source: SourceAPI, EventType: EventQuery, QueryText: &q, Status: StatusSuccess})
|
||||
errs <- err
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
close(errs)
|
||||
for err := range errs {
|
||||
if err != nil {
|
||||
t.Fatalf("concurrent Append failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
result, err := store.VerifyChain(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("VerifyChain: %v", err)
|
||||
}
|
||||
if !result.OK {
|
||||
t.Fatalf("expected an intact chain after %d concurrent appends, got broken at id=%d", n, result.FirstBadID)
|
||||
}
|
||||
if result.RowsChecked != n {
|
||||
t.Fatalf("RowsChecked = %d, want %d", result.RowsChecked, n)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckpointerRun ties Store + FileSink together against real
|
||||
// audit_log data: writes some rows, checkpoints, writes more, checkpoints
|
||||
// again, and confirms the second checkpoint picks up exactly where the
|
||||
// first left off (FromID = previous ToID + 1) with a hash chained off
|
||||
// the previous checkpoint's hash.
|
||||
func TestCheckpointerRun(t *testing.T) {
|
||||
writerPool := testPool(t, "audit_writer", os.Getenv("AUDIT_TEST_POSTGRES_PASSWORD"))
|
||||
adminPool := testPool(t, "sentry", os.Getenv("AUDIT_TEST_ADMIN_PASSWORD"))
|
||||
cleanupAuditLog(t, adminPool)
|
||||
defer cleanupAuditLog(t, adminPool)
|
||||
|
||||
store := NewStore(writerPool)
|
||||
sink := NewFileSink(filepath.Join(t.TempDir(), "checkpoints.jsonl"))
|
||||
checkpointer := NewCheckpointer(store, sink)
|
||||
ctx := context.Background()
|
||||
|
||||
// DELETE doesn't reset the BIGSERIAL sequence, so IDs are not
|
||||
// guaranteed to start at 1 -- but Checkpoint.FromID is a *cursor
|
||||
// position* (1, or the previous checkpoint's ToID+1), not "the
|
||||
// lowest row ID that happens to still exist." In real usage audit_log
|
||||
// never has gaps (append-only, protected by the immutability
|
||||
// trigger), so those always coincide; here, this test's own
|
||||
// destructive cleanupAuditLog between test functions creates a gap
|
||||
// (rows from earlier tests were deleted, advancing the sequence)
|
||||
// that real usage never produces -- so FromID is asserted against
|
||||
// the cursor's own logic (1, since no prior checkpoint exists for
|
||||
// this fresh FileSink), and ToID against the actual last ID Append
|
||||
// returned.
|
||||
var firstBatchLastID int64
|
||||
for i := 0; i < 3; i++ {
|
||||
q := "first batch"
|
||||
rec, err := store.Append(ctx, Entry{TenantID: "default", Source: SourceAPI, EventType: EventQuery, QueryText: &q, Status: StatusSuccess})
|
||||
if err != nil {
|
||||
t.Fatalf("Append: %v", err)
|
||||
}
|
||||
firstBatchLastID = rec.ID
|
||||
}
|
||||
|
||||
cp1, err := checkpointer.Run(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("first Run: %v", err)
|
||||
}
|
||||
if cp1 == nil {
|
||||
t.Fatalf("expected a checkpoint after 3 rows, got nil")
|
||||
}
|
||||
if cp1.FromID != 1 || cp1.ToID != firstBatchLastID {
|
||||
t.Fatalf("cp1 = %+v, want FromID=1 ToID=%d", cp1, firstBatchLastID)
|
||||
}
|
||||
|
||||
// Nothing new since the last checkpoint -- Run should be a no-op.
|
||||
noop, err := checkpointer.Run(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("no-op Run: %v", err)
|
||||
}
|
||||
if noop != nil {
|
||||
t.Fatalf("expected nil (nothing new to checkpoint), got %+v", noop)
|
||||
}
|
||||
|
||||
var secondBatchLastID int64
|
||||
for i := 0; i < 2; i++ {
|
||||
q := "second batch"
|
||||
rec, err := store.Append(ctx, Entry{TenantID: "default", Source: SourceAPI, EventType: EventQuery, QueryText: &q, Status: StatusSuccess})
|
||||
if err != nil {
|
||||
t.Fatalf("Append: %v", err)
|
||||
}
|
||||
secondBatchLastID = rec.ID
|
||||
}
|
||||
|
||||
cp2, err := checkpointer.Run(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("second Run: %v", err)
|
||||
}
|
||||
if cp2 == nil {
|
||||
t.Fatalf("expected a second checkpoint, got nil")
|
||||
}
|
||||
if cp2.FromID != cp1.ToID+1 || cp2.ToID != secondBatchLastID {
|
||||
t.Fatalf("cp2 = %+v, want FromID=%d ToID=%d", cp2, cp1.ToID+1, secondBatchLastID)
|
||||
}
|
||||
if cp2.PrevCheckpointHash != cp1.Hash {
|
||||
t.Fatalf("cp2.PrevCheckpointHash = %q, want %q (chained to cp1)", cp2.PrevCheckpointHash, cp1.Hash)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// VerifyResult reports whether the chain is intact and, if not, the
|
||||
// first row where it breaks -- everything after that point is
|
||||
// untrustworthy regardless of whether later rows individually
|
||||
// "verify," since a break means the chain was forked or rows were
|
||||
// altered/removed at that point.
|
||||
type VerifyResult struct {
|
||||
OK bool
|
||||
FirstBadID int64 // 0 if OK
|
||||
RowsChecked int64
|
||||
}
|
||||
|
||||
// VerifyChain walks audit_log in id order, recomputing each row's hash
|
||||
// from its own fields plus the previous row's hash, and confirms it
|
||||
// matches the stored row_hash and that prev_hash matches the actual
|
||||
// previous row -- catching both in-place tampering (a row's fields
|
||||
// changed, its stored row_hash no longer matches what recomputing it
|
||||
// produces) and forgery (a row inserted with a prev_hash that doesn't
|
||||
// match what actually preceded it).
|
||||
//
|
||||
// This proves internal consistency only. It cannot detect an attacker
|
||||
// who deletes the whole table and replays a self-consistent chain from
|
||||
// row 1 -- that's what checkpoint.go's external anchoring is for. Run
|
||||
// both in the runbook/threat-model verification, not just this one.
|
||||
func (s *Store) VerifyChain(ctx context.Context) (VerifyResult, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, tenant_id, user_id, source, event_type, query_text, row_count,
|
||||
duration_ms, status, error_message, detail, prev_hash, row_hash
|
||||
FROM audit_log ORDER BY id ASC`)
|
||||
if err != nil {
|
||||
return VerifyResult{}, fmt.Errorf("audit: querying for verification: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var expectedPrevHash *string
|
||||
var checked int64
|
||||
for rows.Next() {
|
||||
var rec Record
|
||||
if err := rows.Scan(&rec.ID, &rec.TenantID, &rec.UserID, &rec.Source, &rec.EventType,
|
||||
&rec.QueryText, &rec.RowCount, &rec.DurationMS, &rec.Status, &rec.ErrorMessage,
|
||||
&rec.Detail, &rec.PrevHash, &rec.RowHash); err != nil {
|
||||
return VerifyResult{}, fmt.Errorf("audit: scanning row for verification: %w", err)
|
||||
}
|
||||
checked++
|
||||
|
||||
if !hashPtrEqual(rec.PrevHash, expectedPrevHash) {
|
||||
return VerifyResult{OK: false, FirstBadID: rec.ID, RowsChecked: checked}, nil
|
||||
}
|
||||
recomputed := computeHash(rec.PrevHash, rec.Entry)
|
||||
if recomputed != rec.RowHash {
|
||||
return VerifyResult{OK: false, FirstBadID: rec.ID, RowsChecked: checked}, nil
|
||||
}
|
||||
hash := rec.RowHash
|
||||
expectedPrevHash = &hash
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return VerifyResult{}, fmt.Errorf("audit: reading verification rows: %w", err)
|
||||
}
|
||||
|
||||
return VerifyResult{OK: true, RowsChecked: checked}, nil
|
||||
}
|
||||
|
||||
func hashPtrEqual(a, b *string) bool {
|
||||
if a == nil || b == nil {
|
||||
return a == b
|
||||
}
|
||||
return *a == *b
|
||||
}
|
||||
|
||||
// ListForTenant reads a tenant's audit trail, most recent first --
|
||||
// what a tenant Admin/Owner sees per /docs/phase-4-rbac-design.md's
|
||||
// permission matrix.
|
||||
func (s *Store) ListForTenant(ctx context.Context, tenantID string, limit int) ([]Record, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, tenant_id, user_id, source, event_type, query_text, row_count,
|
||||
duration_ms, status, error_message, detail, prev_hash, row_hash
|
||||
FROM audit_log WHERE tenant_id = $1 ORDER BY id DESC LIMIT $2`, tenantID, limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("audit: listing for tenant: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Record
|
||||
for rows.Next() {
|
||||
var rec Record
|
||||
if err := rows.Scan(&rec.ID, &rec.TenantID, &rec.UserID, &rec.Source, &rec.EventType,
|
||||
&rec.QueryText, &rec.RowCount, &rec.DurationMS, &rec.Status, &rec.ErrorMessage,
|
||||
&rec.Detail, &rec.PrevHash, &rec.RowHash); err != nil {
|
||||
return nil, fmt.Errorf("audit: scanning row: %w", err)
|
||||
}
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
Reference in New Issue
Block a user