Commit `5a1b72b671`

5a1b72b671eb9e56155e645404774b73d3eecef7

parent: 5c2215b845

Verified · cmc

cmc <hello@cleberg.net> · 2026-09-28 09:12 UTC

store: hash-chained audit rows, journal copy, chain verification

Ref #275

Layout: unified · split

internal/store/audit.go +128 −5

		@@ -1,21 +1,144 @@
1	1	package store
2	2
3		import "encoding/json"
	3	import (
	4	"crypto/sha256"
	5	"database/sql"
	6	"encoding/hex"
	7	"encoding/json"
	8	"errors"
	9	"time"
	10	)
4	11
5	12	// Audit appends to the security feed. Events are the product feed; this
6	13	// records who did what, from where, for an operator. actorID 0 means the
7	14	// host admin (gitbayd admin commands) or an unauthenticated source.
8	15	func (s *Store) Audit(actorID int64, action string, data map[string]any) {
	16	raw, err := json.Marshal(data)
	17	if err != nil {
	18	raw = []byte("{}")
	19	}
	20	id, createdAt, hash, err := s.appendAudit(actorID, action, string(raw))
	21	if s.AuditJournal == nil {
	22	return
	23	}
	24	if err != nil {
	25	s.AuditJournal.Error("audit: append", "action", action, "err", err)
	26	return
	27	}
	28	s.AuditJournal.Info("audit", "id", id, "actor", actorID, "action", action,
	29	"data", string(raw), "created_at", createdAt, "hash", hash)
	30	}
	31
	32	// appendAudit writes one row and its chain hash in one transaction. The
	33	// store begins every transaction IMMEDIATE, so two writers — the daemon
	34	// and a gitbayd admin command, say — cannot both read the same last
	35	// hash. The timestamp is taken once the write lock is held, so created_at
	36	// rises with id unless the clock steps back.
	37	func (s *Store) appendAudit(actorID int64, action, data string) (int64, string, string, error) {
	38	tx, err := s.DB.Begin()
	39	if err != nil {
	40	return 0, "", "", err
	41	}
	42	defer tx.Rollback()
	43	var prev string
	44	err = tx.QueryRow("SELECT hash FROM audit_log ORDER BY id DESC LIMIT 1").Scan(&prev)
	45	if err != nil && !errors.Is(err, sql.ErrNoRows) {
	46	return 0, "", "", err
	47	}
9	48	var actor any
10	49	if actorID != 0 {
11	50	actor = actorID
12	51	}
13		raw, err := json.Marshal(data)
	52	createdAt := fmtTime(time.Now())
	53	res, err := tx.Exec(
	54	"INSERT INTO audit_log (actor_id, actor_ref, action, data_json, created_at, prev_hash) VALUES (?, ?, ?, ?, ?, ?)",
	55	actor, actorID, action, data, createdAt, prev)
14	56	if err != nil {
15		raw = []byte("{}")
	57	return 0, "", "", err
	58	}
	59	id, err := res.LastInsertId()
	60	if err != nil {
	61	return 0, "", "", err
	62	}
	63	hash := auditHash(prev, id, actorID, action, createdAt, data)
	64	if _, err := tx.Exec("UPDATE audit_log SET hash = ? WHERE id = ?", hash, id); err != nil {
	65	return 0, "", "", err
	66	}
	67	return id, createdAt, hash, tx.Commit()
	68	}
	69
	70	// auditHash covers every column an operator reads, plus the previous
	71	// row's hash. A JSON array keeps field boundaries unambiguous.
	72	func auditHash(prev string, id, actor int64, action, createdAt, data string) string {
	73	b, _ := json.Marshal([]any{prev, id, actor, action, createdAt, data})
	74	sum := sha256.Sum256(b)
	75	return hex.EncodeToString(sum[:])
	76	}
	77
	78	// AuditChain is what VerifyAuditChain found.
	79	type AuditChain struct {
	80	Rows int // rows read
	81	Unchained int // rows from before migration 0064, which carry no hash
	82	First int64 // first chained row; with no unchained rows before it, its prev_hash is taken as given, since retention may have removed the row it names
	83	Last int64
	84	LastHash string
	85	BrokenAt int64 // 0 when the chain is intact
	86	Reason string
	87	}
	88
	89	// VerifyAuditChain recomputes every row's hash in id order and stops at
	90	// the first row that does not match. Rows removed from the end of the
	91	// table cannot be detected from the database, nor can new rows written
	92	// after that under the reused ids (id is not AUTOINCREMENT); the journal
	93	// copy is the record that shows either.
	94	func (s *Store) VerifyAuditChain() (AuditChain, error) {
	95	rows, err := s.DB.Query(`SELECT id, actor_id, actor_ref, action, data_json, created_at, prev_hash, hash
	96	FROM audit_log ORDER BY id`)
	97	if err != nil {
	98	return AuditChain{}, err
	99	}
	100	defer rows.Close()
	101	var res AuditChain
	102	for rows.Next() {
	103	var (
	104	id, actor int64
	105	actorID sql.NullInt64
	106	action, data, createdAt, prev, hash string
	107	)
	108	if err := rows.Scan(&id, &actorID, &actor, &action, &data, &createdAt, &prev, &hash); err != nil {
	109	return res, err
	110	}
	111	res.Rows++
	112	switch {
	113	case hash == "" && res.First == 0:
	114	res.Unchained++
	115	continue
	116	case hash == "":
	117	res.BrokenAt, res.Reason = id, "row has no hash after the chain began"
	118	// The first row appended after migration 0064 names the last
	119	// unchained row's empty hash. Retention removes the oldest rows
	120	// first, so unchained rows before a chained one with a non-empty
	121	// prev_hash had their hashes blanked.
	122	case res.First == 0 && res.Unchained > 0 && prev != "":
	123	res.BrokenAt, res.Reason = id, "chained rows before it lost their hashes"
	124	case res.First != 0 && prev != res.LastHash:
	125	res.BrokenAt, res.Reason = id, "previous hash does not match: a row before it was removed or changed"
	126	case auditHash(prev, id, actor, action, createdAt, data) != hash:
	127	res.BrokenAt, res.Reason = id, "row contents do not match its hash"
	128	// actor_id is not hashed; it may only be the actor_ref written
	129	// with the row, or NULL once that account is deleted.
	130	case actorID.Valid && actorID.Int64 != actor:
	131	res.BrokenAt, res.Reason = id, "actor_id does not match the actor the row was written with"
	132	}
	133	if res.BrokenAt != 0 {
	134	return res, nil
	135	}
	136	if res.First == 0 {
	137	res.First = id
	138	}
	139	res.Last, res.LastHash = id, hash
16	140	}
17		s.DB.Exec("INSERT INTO audit_log (actor_id, action, data_json) VALUES (?, ?, ?)",
18		actor, action, string(raw))
	141	return res, rows.Err()
19	142	}
20	143
21	144	type AuditEntry struct {

internal/store/auditchain_test.go added +253

		@@ -0,0 +1,253 @@
	1	package store
	2
	3	import (
	4	"bytes"
	5	"fmt"
	6	"log/slog"
	7	"strings"
	8	"sync"
	9	"testing"
	10	"time"
	11	)
	12
	13	func chainStore(t testing.T) Store {
	14	t.Helper()
	15	s := open(t)
	16	if err := s.MigrateUp(); err != nil {
	17	t.Fatal(err)
	18	}
	19	return s
	20	}
	21
	22	func TestAuditChainIntact(t *testing.T) {
	23	s := chainStore(t)
	24	s.Audit(0, "a", map[string]any{"n": 1})
	25	s.Audit(0, "b", nil)
	26	s.Audit(0, "c", map[string]any{"n": 3})
	27	res, err := s.VerifyAuditChain()
	28	if err != nil {
	29	t.Fatal(err)
	30	}
	31	if res.Rows != 3 \|\| res.BrokenAt != 0 \|\| res.First != 1 \|\| res.Last != 3 \|\| len(res.LastHash) != 64 {
	32	t.Fatalf("%+v", res)
	33	}
	34	}
	35
	36	// Concurrent writers each read the last hash and insert in one
	37	// transaction; none may chain to a predecessor another already took.
	38	func TestAuditChainConcurrentWriters(t *testing.T) {
	39	s := chainStore(t)
	40	const writers, each = 8, 25
	41	var wg sync.WaitGroup
	42	for w := range writers {
	43	wg.Add(1)
	44	go func() {
	45	defer wg.Done()
	46	for i := range each {
	47	s.Audit(0, fmt.Sprintf("w%d-%d", w, i), nil)
	48	}
	49	}()
	50	}
	51	wg.Wait()
	52	res, err := s.VerifyAuditChain()
	53	if err != nil {
	54	t.Fatal(err)
	55	}
	56	if res.Rows != writers*each \|\| res.Unchained != 0 \|\| res.BrokenAt != 0 {
	57	t.Fatalf("%+v", res)
	58	}
	59	}
	60
	61	func TestAuditChainDetectsAnEditedRow(t *testing.T) {
	62	for _, set := range []string{
	63	"action = 'x'",
	64	"created_at = '2020-01-01T00:00:00.000Z'",
	65	`data_json = '{"n":2}'`,
	66	"actor_ref = 7",
	67	} {
	68	t.Run(set, func(t *testing.T) {
	69	s := chainStore(t)
	70	for _, a := range []string{"a", "b", "c"} {
	71	s.Audit(0, a, nil)
	72	}
	73	if _, err := s.DB.Exec("UPDATE audit_log SET " + set + " WHERE id = 2"); err != nil {
	74	t.Fatal(err)
	75	}
	76	res, err := s.VerifyAuditChain()
	77	if err != nil {
	78	t.Fatal(err)
	79	}
	80	if res.BrokenAt != 2 \|\| !strings.Contains(res.Reason, "contents") {
	81	t.Fatalf("%+v", res)
	82	}
	83	})
	84	}
	85	}
	86
	87	// Blanking the hash of the oldest chained rows would make them read as
	88	// rows from before the migration.
	89	func TestAuditChainDetectsBlankedHashes(t *testing.T) {
	90	s := chainStore(t)
	91	for _, a := range []string{"a", "b", "c"} {
	92	s.Audit(0, a, nil)
	93	}
	94	if _, err := s.DB.Exec("UPDATE audit_log SET hash = '', action = 'x' WHERE id = 1"); err != nil {
	95	t.Fatal(err)
	96	}
	97	res, err := s.VerifyAuditChain()
	98	if err != nil {
	99	t.Fatal(err)
	100	}
	101	if res.BrokenAt != 2 {
	102	t.Fatalf("%+v", res)
	103	}
	104	}
	105
	106	// actor_id is not hashed; it must agree with actor_ref or be NULL.
	107	func TestAuditChainDetectsAChangedActorID(t *testing.T) {
	108	s := chainStore(t)
	109	uid, err := s.CreateUser("alice", false)
	110	if err != nil {
	111	t.Fatal(err)
	112	}
	113	s.Audit(0, "a", nil)
	114	s.Audit(0, "b", nil)
	115	if _, err := s.DB.Exec("UPDATE audit_log SET actor_id = ? WHERE id = 2", uid); err != nil {
	116	t.Fatal(err)
	117	}
	118	res, err := s.VerifyAuditChain()
	119	if err != nil {
	120	t.Fatal(err)
	121	}
	122	if res.BrokenAt != 2 \|\| !strings.Contains(res.Reason, "actor_id") {
	123	t.Fatalf("%+v", res)
	124	}
	125	}
	126
	127	// A clock stepped back leaves created_at out of id order; retention
	128	// still removes a prefix of the table, so the chain stays intact.
	129	func TestAuditChainSurvivesRetentionWithClockStep(t *testing.T) {
	130	s := chainStore(t)
	131	for _, a := range []string{"a", "b", "c", "d"} {
	132	s.Audit(0, a, nil)
	133	}
	134	// Rewrite created_at and the hashes as the rows would have been
	135	// written: row 2 stamped after row 3, both older than the cutoff.
	136	stamps := map[int64]string{
	137	1: "2020-01-01T00:00:00.000Z",
	138	2: "2020-01-03T00:00:00.000Z",
	139	3: "2020-01-02T00:00:00.000Z",
	140	4: "2099-01-01T00:00:00.000Z",
	141	}
	142	prev := ""
	143	for id := int64(1); id <= 4; id++ {
	144	var action, data string
	145	if err := s.DB.QueryRow("SELECT action, data_json FROM audit_log WHERE id = ?", id).Scan(&action, &data); err != nil {
	146	t.Fatal(err)
	147	}
	148	h := auditHash(prev, id, 0, action, stamps[id], data)
	149	if _, err := s.DB.Exec("UPDATE audit_log SET created_at = ?, prev_hash = ?, hash = ? WHERE id = ?",
	150	stamps[id], prev, h, id); err != nil {
	151	t.Fatal(err)
	152	}
	153	prev = h
	154	}
	155	if _, err := s.Sweep(Retention{Audit: time.Hour}, mustTime(t, "2020-01-02T12:00:00.000Z")); err != nil {
	156	t.Fatal(err)
	157	}
	158	res, err := s.VerifyAuditChain()
	159	if err != nil {
	160	t.Fatal(err)
	161	}
	162	if res.BrokenAt != 0 \|\| res.First != 4 \|\| res.Rows != 1 {
	163	t.Fatalf("%+v", res)
	164	}
	165	}
	166
	167	func mustTime(t *testing.T, v string) time.Time {
	168	t.Helper()
	169	tm, err := time.Parse(time.RFC3339, v)
	170	if err != nil {
	171	t.Fatal(err)
	172	}
	173	return tm
	174	}
	175
	176	func TestAuditChainDetectsARemovedRow(t *testing.T) {
	177	s := chainStore(t)
	178	for _, a := range []string{"a", "b", "c"} {
	179	s.Audit(0, a, nil)
	180	}
	181	if _, err := s.DB.Exec("DELETE FROM audit_log WHERE id = 2"); err != nil {
	182	t.Fatal(err)
	183	}
	184	res, err := s.VerifyAuditChain()
	185	if err != nil {
	186	t.Fatal(err)
	187	}
	188	if res.BrokenAt != 3 \|\| !strings.Contains(res.Reason, "previous hash") {
	189	t.Fatalf("%+v", res)
	190	}
	191	}
	192
	193	// Retention removes the oldest rows, and deleting an account nulls
	194	// actor_id; neither is tampering.
	195	func TestAuditChainSurvivesRetentionAndAccountDeletion(t *testing.T) {
	196	s := chainStore(t)
	197	uid, err := s.CreateUser("alice", false)
	198	if err != nil {
	199	t.Fatal(err)
	200	}
	201	s.Audit(0, "a", nil)
	202	s.Audit(uid, "b", nil)
	203	s.Audit(0, "c", nil)
	204	if _, err := s.DB.Exec("DELETE FROM audit_log WHERE id = 1"); err != nil {
	205	t.Fatal(err)
	206	}
	207	if _, err := s.DB.Exec("DELETE FROM users WHERE id = ?", uid); err != nil {
	208	t.Fatal(err)
	209	}
	210	res, err := s.VerifyAuditChain()
	211	if err != nil {
	212	t.Fatal(err)
	213	}
	214	if res.BrokenAt != 0 \|\| res.First != 2 \|\| res.Last != 3 {
	215	t.Fatalf("%+v", res)
	216	}
	217	}
	218
	219	// Rows written before migration 0064 carry no hash; the chain starts
	220	// after them, and a hashless row after that start is a break.
	221	func TestAuditChainLegacyRows(t *testing.T) {
	222	s := chainStore(t)
	223	if _, err := s.DB.Exec("INSERT INTO audit_log (action) VALUES ('legacy')"); err != nil {
	224	t.Fatal(err)
	225	}
	226	s.Audit(0, "a", nil)
	227	res, err := s.VerifyAuditChain()
	228	if err != nil {
	229	t.Fatal(err)
	230	}
	231	if res.Unchained != 1 \|\| res.BrokenAt != 0 \|\| res.First != 2 {
	232	t.Fatalf("%+v", res)
	233	}
	234	if _, err := s.DB.Exec("INSERT INTO audit_log (action) VALUES ('injected')"); err != nil {
	235	t.Fatal(err)
	236	}
	237	if res, _ = s.VerifyAuditChain(); res.BrokenAt != 3 {
	238	t.Fatalf("hashless row after the chain: %+v", res)
	239	}
	240	}
	241
	242	func TestAuditJournal(t *testing.T) {
	243	s := chainStore(t)
	244	var buf bytes.Buffer
	245	s.AuditJournal = slog.New(slog.NewTextHandler(&buf, nil))
	246	s.Audit(0, "cmd repo create", map[string]any{"argv": []string{"a/b"}})
	247	line := buf.String()
	248	for _, want := range []string{"msg=audit", "action=\"cmd repo create\"", "id=1", "hash="} {
	249	if !strings.Contains(line, want) {
	250	t.Fatalf("journal line %q lacks %q", line, want)
	251	}
	252	}
	253	}

internal/store/migrations/0064_audit_chain.down.sql added +3

		@@ -0,0 +1,3 @@
	1	ALTER TABLE audit_log DROP COLUMN hash;
	2	ALTER TABLE audit_log DROP COLUMN prev_hash;
	3	ALTER TABLE audit_log DROP COLUMN actor_ref;

internal/store/migrations/0064_audit_chain.up.sql added +8

		@@ -0,0 +1,8 @@
	1	-- Each row carries the hash of the row before it. actor_ref is the actor
	2	-- id as written: actor_id is set to NULL when the account is deleted,
	3	-- and the hash must not change with it. Rows written before this
	4	-- migration keep an empty hash; the chain starts after them.
	5	ALTER TABLE audit_log ADD COLUMN actor_ref INTEGER NOT NULL DEFAULT 0;
	6	ALTER TABLE audit_log ADD COLUMN prev_hash TEXT NOT NULL DEFAULT '';
	7	ALTER TABLE audit_log ADD COLUMN hash TEXT NOT NULL DEFAULT '';
	8	UPDATE audit_log SET actor_ref = COALESCE(actor_id, 0);

internal/store/retention.go +4 −1

		@@ -73,7 +73,10 @@ func (s *Store) Sweep(r Retention, now time.Time) (Swept, error) {
73	73	where string
74	74	keep time.Duration
75	75	}{
76		{"audit_log", "created_at < ?", r.Audit},
	76	// By id, so a backwards clock step cannot leave a newer row
	77	// removed and an older one kept: the audit chain would read
	78	// that as tampering.
	79	{"audit_log", "id <= (SELECT MAX(id) FROM audit_log WHERE created_at < ?)", r.Audit},
77	80	// Only deliveries that have finished: one still being retried is
78	81	// live state, however old its first attempt.
79	82	{"webhook_deliveries", "created_at < ? AND (delivered_at IS NOT NULL OR failed_at IS NOT NULL)", r.WebhookDeliveries},

internal/store/store.go +6

		@@ -8,6 +8,7 @@ import (
8	8	"errors"
9	9	"fmt"
10	10	"io/fs"
	11	"log/slog"
11	12	"os"
12	13	"sort"
13	14	"strconv"
		@@ -31,6 +32,11 @@ type Store struct {
31	32	// onRevoke runs after each key revocation this process commits.
32	33	revokeMu sync.Mutex
33	34	onRevoke []func(Revoked)
	35
	36	// AuditJournal, when set, receives a copy of every audit row. The
	37	// daemon sets it to its own logger, whose output the service
	38	// journal keeps outside the database.
	39	AuditJournal *slog.Logger
34	40	}
35	41
36	42	// Open opens (creating if needed) the database at path with WAL mode and

Commit 5a1b72b671

Commit `5a1b72b671`