oficina / home Public

Download this file

1	package operator
2
3	import (
4	"bytes"
5	"context"
6	"database/sql"
7	"errors"
8	"os"
9	"path/filepath"
10	"sync"
11	"sync/atomic"
12	"testing"
13	"time"
14
15	sharedseal "amadan.net/oficina/home/handoff/seal"
16	"amadan.net/oficina/home/internal/home/directorydata"
17	family "amadan.net/rastrillo/rastrillo/crypto"
18	rdb "amadan.net/rastrillo/rastrillo/db"
19	"amadan.net/rastrillo/rastrillo/migrate"
20	_ "modernc.org/sqlite"
21	)
22
23	type engineFixture struct {
24	engine *Engine
25	db *sql.DB
26	raw []byte
27	now *time.Time
28	path string
29	key *family.Keypair
30	auth Authorization
31	}
32
33	func newEngineFixture(t *testing.T) engineFixture {
34	t.Helper()
35	k, v, a := fixture(t)
36	path := filepath.Join(t.TempDir(), "operator.db")
37	handle, e := rdb.Open(path, nil)
38	if e != nil {
39	t.Fatal(e)
40	}
41	t.Cleanup(func() { handle.Close() })
42	db := handle.Writer()
43	db.SetMaxOpenConns(4)
44	if _, e = db.Exec("PRAGMA journal_mode=WAL"); e != nil {
45	t.Fatal(e)
46	}
47	if _, e = migrate.Apply(context.Background(), handle, migrate.Merge(directorydata.Schema, directorydata.RosterSchema, directorydata.OperatorSchema, DirectorySchema)); e != nil {
48	t.Fatal(e)
49	}
50	if _, e = db.Exec("CREATE TABLE test_items(id INTEGER PRIMARY KEY)"); e != nil {
51	t.Fatal(e)
52	}
53	sealer, e := sharedseal.New(bytes.Repeat([]byte{6}, 32), "oficina/operator")
54	if e != nil {
55	t.Fatal(e)
56	}
57	now := time.Unix(a.Issued, 0)
58	engine, e := NewEngine(db, v, sealer, bytes.Repeat([]byte{7}, 32), func() time.Time { return now })
59	if e != nil {
60	t.Fatal(e)
61	}
62	return engineFixture{engine, db, signed(t, k, a), &now, path, k, a}
63	}
64
65	var privateResult = []byte(`{"bearer":"synthetic-private-bearer-keep-out-of-db","receipt":"synthetic-private-proof"}`)
66
67	func insertOne(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
68	_, e := c.ExecContext(ctx, "INSERT INTO test_items VALUES(1)")
69	if e != nil {
70	return nil, e
71	}
72	return privateResult, nil
73	}
74	func countRows(t *testing.T, db *sql.DB, table string) int {
75	t.Helper()
76	var n int
77	if e := db.QueryRow("SELECT count(*) FROM " + table).Scan(&n); e != nil {
78	t.Fatal(e)
79	}
80	return n
81	}
82	func TestEngineCommitsOnceAndRecoversExpiredResultPrivately(t *testing.T) {
83	f := newEngineFixture(t)
84	out, e := f.engine.Execute(context.Background(), f.raw, insertOne)
85	if e != nil || !bytes.Equal(out, privateResult) {
86	t.Fatal(e)
87	}
88	*f.now = f.now.Add(301 * time.Second)
89	out, e = f.engine.Execute(context.Background(), f.raw, func(context.Context, *sql.Conn, Request) ([]byte, error) {
90	t.Fatal("replay executed mutation")
91	return nil, nil
92	})
93	if e != nil || !bytes.Equal(out, privateResult) {
94	t.Fatal(e)
95	}
96	if countRows(t, f.db, "test_items") != 1 || countRows(t, f.db, "directory_operator_results") != 1 {
97	t.Fatal("non-atomic replay")
98	}
99	for _, suffix := range []string{"", "-wal"} {
100	b, e := os.ReadFile(f.path + suffix)
101	if e != nil && !errors.Is(e, os.ErrNotExist) {
102	t.Fatal(e)
103	}
104	for _, secret := range [][]byte{privateResult, []byte("synthetic-private-bearer"), []byte("synthetic-private-proof")} {
105	if bytes.Contains(b, secret) {
106	t.Fatal("private result at rest")
107	}
108	}
109	}
110	if _, e = f.db.Exec("DELETE FROM directory_operator_state"); e != nil {
111	t.Fatal(e)
112	}
113	if _, e = f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrRefused {
114	t.Fatal("missing authority accepted", e)
115	}
116	}
117	func TestEngineFailureRollsBackMutationReceiptAndSequence(t *testing.T) {
118	for _, mode := range []string{"mutation", "expiry", "seal"} {
119	t.Run(mode, func(t *testing.T) {
120	f := newEngineFixture(t)
121	if mode == "seal" {
122	f.engine.sealer = failingSeal{}
123	}
124	_, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
125	if _, e := insertOne(ctx, c, r); e != nil {
126	return nil, e
127	}
128	switch mode {
129	case "mutation":
130	return nil, errors.New("private@example.test")
131	case "expiry":
132	*f.now = f.now.Add(301 * time.Second)
133	}
134	return privateResult, nil
135	})
136	if e != ErrRefused && e != ErrUnavailable {
137	t.Fatal("unexpected error", e)
138	}
139	for _, table := range []string{"test_items", "directory_operator_results", "directory_operator_state"} {
140	if countRows(t, f.db, table) != 0 {
141	t.Fatal("partial mutation", table)
142	}
143	}
144	})
145	}
146	}
147
148	type failingSeal struct{}
149
150	func (failingSeal) Seal(string, []byte) ([]byte, error) {
151	return nil, errors.New("private seal failure")
152	}
153	func (failingSeal) Open(string, []byte) ([]byte, error) {
154	return nil, errors.New("private seal failure")
155	}
156	func TestEngineConcurrentRetryHasOneMutation(t *testing.T) {
157	f := newEngineFixture(t)
158	var called atomic.Int32
159	var wg sync.WaitGroup
160	errs := make(chan error, 8)
161	for i := 0; i < 8; i++ {
162	wg.Add(1)
163	go func() {
164	defer wg.Done()
165	_, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
166	called.Add(1)
167	return insertOne(ctx, c, r)
168	})
169	errs <- e
170	}()
171	}
172	wg.Wait()
173	close(errs)
174	for e := range errs {
175	if e != nil {
176	t.Fatal(e)
177	}
178	}
179	if called.Load() != 1 {
180	t.Fatal("duplicate side effects", called.Load())
181	}
182	}
183	func TestEngineForeignKeyEnforcementUsesMutationConnection(t *testing.T) {
184	f := newEngineFixture(t)
185	if _, e := f.db.Exec("CREATE TABLE parent(id INTEGER PRIMARY KEY); CREATE TABLE child(parent INTEGER REFERENCES parent(id))"); e != nil {
186	t.Fatal(e)
187	}
188	_, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
189	var fk int
190	if e := c.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fk); e != nil || fk != 1 {
191	t.Fatal("foreign keys not on actual connection")
192	}
193	_, e := c.ExecContext(ctx, "INSERT INTO child VALUES(999)")
194	return privateResult, e
195	})
196	if e != ErrUnavailable || countRows(t, f.db, "child") != 0 || countRows(t, f.db, "directory_operator_results") != 0 {
197	t.Fatal("foreign key violation escaped", e)
198	}
199	}
200	func TestEngineRefusesExpiredNewAndCorruptExistingResult(t *testing.T) {
201	f := newEngineFixture(t)
202	*f.now = f.now.Add(300 * time.Second)
203	if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrRefused {
204	t.Fatal("expired new")
205	}
206	if countRows(t, f.db, "test_items") != 0 {
207	t.Fatal("expired side effect")
208	}
209	*f.now = f.now.Add(-300 * time.Second)
210	if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != nil {
211	t.Fatal(e)
212	}
213	if _, e := f.db.Exec("UPDATE directory_operator_results SET sealed_result=zeroblob(29)"); e != nil {
214	t.Fatal(e)
215	}
216	if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrUnavailable {
217	t.Fatal("corrupt result", e)
218	}
219	}
220
221	func TestEngineCommandSchemaCheckRunsInsideTransactionBeforeReplayOrMutation(t *testing.T) {
222	f := newEngineFixture(t)
223	ctx := t.Context()
224	var checked *sql.Conn
225	f.engine.checkSchema = func(ctx context.Context, conn *sql.Conn) error {
226	checked = conn
227	if _, err := conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err == nil {
228	t.Fatal("schema check outside transaction")
229	}
230	return nil
231	}
232	mutate := func(ctx context.Context, conn *sql.Conn, r Request) ([]byte, error) {
233	if conn != checked {
234	t.Fatal("schema checked a different connection")
235	}
236	return insertOne(ctx, conn, r)
237	}
238	if _, err := f.engine.Execute(ctx, f.raw, mutate); err != nil {
239	t.Fatal(err)
240	}
241	f.engine.checkSchema = func(context.Context, *sql.Conn) error { return ErrRefused }
242	if out, err := f.engine.Execute(ctx, f.raw, mutate); err != ErrRefused || len(out) != 0 {
243	t.Fatal("replay bypassed schema refusal", err)
244	}
245	if countRows(t, f.db, "test_items") != 1 || countRows(t, f.db, "directory_operator_results") != 1 {
246	t.Fatal("schema refusal changed committed data")
247	}
248	fresh := newEngineFixture(t)
249	fresh.engine.checkSchema = func(ctx context.Context, conn *sql.Conn) error {
250	if _, err := conn.ExecContext(ctx, "INSERT INTO test_items VALUES(2)"); err != nil {
251	t.Fatal(err)
252	}
253	return ErrUnavailable
254	}
255	if out, err := fresh.engine.Execute(ctx, fresh.raw, func(context.Context, *sql.Conn, Request) ([]byte, error) {
256	t.Fatal("mutation ran after failed schema check")
257	return nil, nil
258	}); err != ErrUnavailable || len(out) != 0 {
259	t.Fatal("schema failure accepted", err)
260	}
261	if countRows(t, fresh.db, "test_items") != 0 || countRows(t, fresh.db, "directory_operator_state") != 0 || countRows(t, fresh.db, "directory_operator_results") != 0 {
262	t.Fatal("failed preflight transaction survived")
263	}
264	}
265