amadan

oficina / home Public

Add atomic operator execution and sealed idempotent result storage

Paul Campbell pushed by paul@keymail.dev cd598f6c3fb6f1657581921847e2a28e49bbac83 parent f80ea16
4 files changed, +453 −0
  • internal/operator/engine.go +200 −0
  • internal/operator/engine_test.go +215 −0
  • internal/operator/migrations/credential/migrations/0001_operator.sql +19 −0
  • internal/operator/migrations/directory/migrations/0001_operator.sql +19 −0

CI failed — run details

Intent

full ledger: feat-home-operator-core

diff --git a/internal/operator/engine.go b/internal/operator/engine.go
new file mode 100644
index 0000000..a88296f
--- /dev/null
+++ b/internal/operator/engine.go
@@ -0,0 +1,200 @@
+package operator
+
+import (
+ "context"
+ "crypto/hmac"
+ "crypto/sha256"
+ "database/sql"
+ "database/sql/driver"
+ "embed"
+ "errors"
+ "io/fs"
+ "math"
+ "time"
+
+ "github.com/carlosframework/rastrillo/migrate"
+)
+
+//go:embed migrations
+var migrations embed.FS
+
+var CredentialSchema = roleSchema("credential")
+var DirectorySchema = roleSchema("directory")
+
+func roleSchema(role string) *migrate.Set {
+ f, e := fs.Sub(migrations, "migrations/"+role)
+ if e != nil {
+ panic(e)
+ }
+ return migrate.MustFromFS(f, "oficina_"+role+"_operator")
+}
+
+type Sealer interface {
+ Seal(string, []byte) ([]byte, error)
+ Open(string, []byte) ([]byte, error)
+}
+
+// Mutation executes on the engine's already locked connection. It must not
+// commit, open another writer, dial the network or emit its result itself.
+type Mutation func(context.Context, *sql.Conn, Request) ([]byte, error)
+
+type Engine struct {
+ db *sql.DB
+ verifier *Verifier
+ sealer Sealer
+ macKey []byte
+ prefix string
+ now func() time.Time
+}
+
+func NewEngine(db *sql.DB, v *Verifier, sealer Sealer, macKey []byte, now func() time.Time) (*Engine, error) {
+ if db == nil || v == nil || sealer == nil || len(macKey) != 32 || now == nil {
+ return nil, ErrRefused
+ }
+ return &Engine{db, v, sealer, append([]byte(nil), macKey...), v.pins.Role + "_operator", now}, nil
+}
+func (e *Engine) purpose(a Authorization) string {
+ return "operator.result/v1/" + a.Origin + "/" + a.OperationID + "/" + a.Action
+}
+func (e *Engine) mac(r Request) []byte {
+ m := hmac.New(sha256.New, e.macKey)
+ m.Write(r.input)
+ return m.Sum(nil)
+}
+
+// Execute validates against this engine's pinned verifier, then commits the
+// mutation, result and accepted stream together. Expired authentic input is
+// useful only when its exact previously committed result already exists.
+func (e *Engine) Execute(ctx context.Context, raw []byte, mutate Mutation) ([]byte, error) {
+ if e == nil || mutate == nil {
+ return nil, ErrRefused
+ }
+ req, err := e.verifier.Verify(raw)
+ if err != nil {
+ return nil, ErrRefused
+ }
+ a := req.authorization
+ ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+ defer cancel()
+ conn, err := e.db.Conn(ctx)
+ if err != nil {
+ return nil, ErrUnavailable
+ }
+ defer conn.Close()
+ for _, statement := range []string{"PRAGMA foreign_keys=ON", "PRAGMA secure_delete=ON", "PRAGMA busy_timeout=1000"} {
+ if _, err = conn.ExecContext(ctx, statement); err != nil {
+ return nil, ErrUnavailable
+ }
+ }
+ var fk, secure int
+ var journal string
+ if conn.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fk) != nil || fk != 1 || conn.QueryRowContext(ctx, "PRAGMA secure_delete").Scan(&secure) != nil || secure != 1 || conn.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&journal) != nil || journal != "wal" {
+ return nil, ErrUnavailable
+ }
+ violations, err := conn.QueryContext(ctx, "PRAGMA foreign_key_check")
+ if err != nil {
+ return nil, ErrUnavailable
+ }
+ bad := violations.Next()
+ readErr := violations.Err()
+ closeErr := violations.Close()
+ if bad || readErr != nil || closeErr != nil {
+ return nil, ErrUnavailable
+ }
+ if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil {
+ return nil, ErrUnavailable
+ }
+ finished := false
+ defer func() {
+ if !finished {
+ cleanup, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if _, err := conn.ExecContext(cleanup, "ROLLBACK"); err != nil {
+ _ = conn.Raw(func(any) error { return driver.ErrBadConn })
+ }
+ }
+ }()
+ var storedOrigin, storedInstallation, storedRegistry, storedDevice, storedHead string
+ var storedSequence, storedCompleted int64
+ stateErr := conn.QueryRowContext(ctx, "SELECT origin,installation,registry,device,sequence,head,completed_at FROM "+e.prefix+"_state WHERE singleton=1").Scan(&storedOrigin, &storedInstallation, &storedRegistry, &storedDevice, &storedSequence, &storedHead, &storedCompleted)
+ if stateErr != nil && !errors.Is(stateErr, sql.ErrNoRows) {
+ return nil, ErrUnavailable
+ }
+ if stateErr == nil && (storedOrigin != a.Origin || storedInstallation != a.Installation || storedRegistry != a.Registry || storedDevice != a.Device || storedSequence <= 0 || !CanonicalToken(storedHead) || storedCompleted <= 0 || e.now().Unix() < storedCompleted) {
+ return nil, ErrRefused
+ }
+ var oldAction, oldHead string
+ var oldSequence int64
+ var oldMAC, sealed []byte
+ err = conn.QueryRowContext(ctx, "SELECT action,sequence,authorization_head,request_mac,sealed_result FROM "+e.prefix+"_results WHERE operation_id=?", a.OperationID).Scan(&oldAction, &oldSequence, &oldHead, &oldMAC, &sealed)
+ if err == nil {
+ if stateErr != nil || storedSequence < a.Sequence || oldAction != a.Action || oldSequence != a.Sequence || oldHead != req.head || !hmac.Equal(oldMAC, e.mac(req)) {
+ return nil, ErrRefused
+ }
+ result, openErr := e.sealer.Open(e.purpose(a), sealed)
+ if openErr != nil || len(result) == 0 || len(result) > InputLimit {
+ return nil, ErrUnavailable
+ }
+ if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil {
+ return nil, ErrUnavailable
+ }
+ finished = true
+ return result, nil
+ }
+ if !errors.Is(err, sql.ErrNoRows) {
+ return nil, ErrUnavailable
+ }
+ now := e.now()
+ if !req.Fresh(now) {
+ return nil, ErrRefused
+ }
+ var origin, installation, registry, device, head string
+ var sequence, last int64
+ err = conn.QueryRowContext(ctx, "SELECT origin,installation,registry,device,sequence,head,completed_at FROM "+e.prefix+"_state WHERE singleton=1").Scan(&origin, &installation, &registry, &device, &sequence, &head, &last)
+ if errors.Is(err, sql.ErrNoRows) {
+ if !a.Initial || a.Sequence != 1 || a.Previous != "" {
+ return nil, ErrRefused
+ }
+ } else if err != nil {
+ return nil, ErrUnavailable
+ } else {
+ if a.Initial || sequence == math.MaxInt64 || a.Sequence != sequence+1 || a.Previous != head || a.Origin != origin || a.Installation != installation || a.Registry != registry || a.Device != device || now.Unix() < last {
+ return nil, ErrRefused
+ }
+ }
+ var count int
+ if conn.QueryRowContext(ctx, "SELECT count(*) FROM "+e.prefix+"_results").Scan(&count) != nil {
+ return nil, ErrUnavailable
+ }
+ if count >= 10000 {
+ return nil, ErrRefused
+ }
+ result, err := mutate(ctx, conn, req)
+ if err != nil {
+ if errors.Is(err, ErrRefused) {
+ return nil, ErrRefused
+ }
+ return nil, ErrUnavailable
+ }
+ completed := e.now()
+ if !req.Fresh(completed) || completed.Before(now) || len(result) == 0 || len(result) > InputLimit {
+ return nil, ErrRefused
+ }
+ sealed, err = e.sealer.Seal(e.purpose(a), result)
+ if err != nil {
+ return nil, ErrUnavailable
+ }
+ _, err = conn.ExecContext(ctx, "INSERT INTO "+e.prefix+"_results(operation_id,action,sequence,authorization_head,request_mac,sealed_result,completed_at) VALUES(?,?,?,?,?,?,?)", a.OperationID, a.Action, a.Sequence, req.head, e.mac(req), sealed, completed.Unix())
+ if err != nil {
+ return nil, ErrUnavailable
+ }
+ _, err = conn.ExecContext(ctx, "INSERT INTO "+e.prefix+"_state(singleton,origin,installation,registry,device,sequence,head,completed_at) VALUES(1,?,?,?,?,?,?,?) ON CONFLICT(singleton) DO UPDATE SET sequence=excluded.sequence,head=excluded.head,completed_at=excluded.completed_at", a.Origin, a.Installation, a.Registry, a.Device, a.Sequence, req.head, completed.Unix())
+ if err != nil {
+ return nil, ErrUnavailable
+ }
+ if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil {
+ return nil, ErrUnavailable
+ }
+ finished = true
+ return append([]byte(nil), result...), nil
+}
diff --git a/internal/operator/engine_test.go b/internal/operator/engine_test.go
new file mode 100644
index 0000000..a59a899
--- /dev/null
+++ b/internal/operator/engine_test.go
@@ -0,0 +1,215 @@
+package operator
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "errors"
+ "os"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ sharedseal "amadan.net/oficina/home/handoff/seal"
+ rdb "github.com/carlosframework/rastrillo/db"
+ "github.com/carlosframework/rastrillo/migrate"
+ _ "modernc.org/sqlite"
+)
+
+type engineFixture struct {
+ engine *Engine
+ db *sql.DB
+ raw []byte
+ now *time.Time
+ path string
+}
+
+func newEngineFixture(t *testing.T) engineFixture {
+ t.Helper()
+ k, v, a := fixture(t)
+ path := filepath.Join(t.TempDir(), "operator.db")
+ handle, e := rdb.Open(path, nil)
+ if e != nil {
+ t.Fatal(e)
+ }
+ t.Cleanup(func() { handle.Close() })
+ db := handle.Writer()
+ db.SetMaxOpenConns(4)
+ if _, e = db.Exec("PRAGMA journal_mode=WAL"); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = migrate.Apply(context.Background(), handle, DirectorySchema); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = db.Exec("CREATE TABLE test_items(id INTEGER PRIMARY KEY)"); e != nil {
+ t.Fatal(e)
+ }
+ sealer, e := sharedseal.New(bytes.Repeat([]byte{6}, 32), "oficina/operator")
+ if e != nil {
+ t.Fatal(e)
+ }
+ now := time.Unix(a.Issued, 0)
+ engine, e := NewEngine(db, v, sealer, bytes.Repeat([]byte{7}, 32), func() time.Time { return now })
+ if e != nil {
+ t.Fatal(e)
+ }
+ return engineFixture{engine, db, signed(t, k, a), &now, path}
+}
+
+var privateResult = []byte(`{"bearer":"synthetic-private-bearer-keep-out-of-db","receipt":"synthetic-private-proof"}`)
+
+func insertOne(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
+ _, e := c.ExecContext(ctx, "INSERT INTO test_items VALUES(1)")
+ if e != nil {
+ return nil, e
+ }
+ return privateResult, nil
+}
+func countRows(t *testing.T, db *sql.DB, table string) int {
+ t.Helper()
+ var n int
+ if e := db.QueryRow("SELECT count(*) FROM " + table).Scan(&n); e != nil {
+ t.Fatal(e)
+ }
+ return n
+}
+func TestEngineCommitsOnceAndRecoversExpiredResultPrivately(t *testing.T) {
+ f := newEngineFixture(t)
+ out, e := f.engine.Execute(context.Background(), f.raw, insertOne)
+ if e != nil || !bytes.Equal(out, privateResult) {
+ t.Fatal(e)
+ }
+ *f.now = f.now.Add(301 * time.Second)
+ out, e = f.engine.Execute(context.Background(), f.raw, func(context.Context, *sql.Conn, Request) ([]byte, error) {
+ t.Fatal("replay executed mutation")
+ return nil, nil
+ })
+ if e != nil || !bytes.Equal(out, privateResult) {
+ t.Fatal(e)
+ }
+ if countRows(t, f.db, "test_items") != 1 || countRows(t, f.db, "directory_operator_results") != 1 {
+ t.Fatal("non-atomic replay")
+ }
+ for _, suffix := range []string{"", "-wal"} {
+ b, e := os.ReadFile(f.path + suffix)
+ if e != nil && !errors.Is(e, os.ErrNotExist) {
+ t.Fatal(e)
+ }
+ for _, secret := range [][]byte{privateResult, []byte("synthetic-private-bearer"), []byte("synthetic-private-proof")} {
+ if bytes.Contains(b, secret) {
+ t.Fatal("private result at rest")
+ }
+ }
+ }
+ if _, e = f.db.Exec("DELETE FROM directory_operator_state"); e != nil {
+ t.Fatal(e)
+ }
+ if _, e = f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrRefused {
+ t.Fatal("missing authority accepted", e)
+ }
+}
+func TestEngineFailureRollsBackMutationReceiptAndSequence(t *testing.T) {
+ for _, mode := range []string{"mutation", "expiry", "seal"} {
+ t.Run(mode, func(t *testing.T) {
+ f := newEngineFixture(t)
+ if mode == "seal" {
+ f.engine.sealer = failingSeal{}
+ }
+ _, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
+ if _, e := insertOne(ctx, c, r); e != nil {
+ return nil, e
+ }
+ switch mode {
+ case "mutation":
+ return nil, errors.New("private@example.test")
+ case "expiry":
+ *f.now = f.now.Add(301 * time.Second)
+ }
+ return privateResult, nil
+ })
+ if e != ErrRefused && e != ErrUnavailable {
+ t.Fatal("unexpected error", e)
+ }
+ for _, table := range []string{"test_items", "directory_operator_results", "directory_operator_state"} {
+ if countRows(t, f.db, table) != 0 {
+ t.Fatal("partial mutation", table)
+ }
+ }
+ })
+ }
+}
+
+type failingSeal struct{}
+
+func (failingSeal) Seal(string, []byte) ([]byte, error) {
+ return nil, errors.New("private seal failure")
+}
+func (failingSeal) Open(string, []byte) ([]byte, error) {
+ return nil, errors.New("private seal failure")
+}
+func TestEngineConcurrentRetryHasOneMutation(t *testing.T) {
+ f := newEngineFixture(t)
+ var called atomic.Int32
+ var wg sync.WaitGroup
+ errs := make(chan error, 8)
+ for i := 0; i < 8; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ _, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
+ called.Add(1)
+ return insertOne(ctx, c, r)
+ })
+ errs <- e
+ }()
+ }
+ wg.Wait()
+ close(errs)
+ for e := range errs {
+ if e != nil {
+ t.Fatal(e)
+ }
+ }
+ if called.Load() != 1 {
+ t.Fatal("duplicate side effects", called.Load())
+ }
+}
+func TestEngineForeignKeyEnforcementUsesMutationConnection(t *testing.T) {
+ f := newEngineFixture(t)
+ if _, e := f.db.Exec("CREATE TABLE parent(id INTEGER PRIMARY KEY); CREATE TABLE child(parent INTEGER REFERENCES parent(id))"); e != nil {
+ t.Fatal(e)
+ }
+ _, e := f.engine.Execute(context.Background(), f.raw, func(ctx context.Context, c *sql.Conn, r Request) ([]byte, error) {
+ var fk int
+ if e := c.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fk); e != nil || fk != 1 {
+ t.Fatal("foreign keys not on actual connection")
+ }
+ _, e := c.ExecContext(ctx, "INSERT INTO child VALUES(999)")
+ return privateResult, e
+ })
+ if e != ErrUnavailable || countRows(t, f.db, "child") != 0 || countRows(t, f.db, "directory_operator_results") != 0 {
+ t.Fatal("foreign key violation escaped", e)
+ }
+}
+func TestEngineRefusesExpiredNewAndCorruptExistingResult(t *testing.T) {
+ f := newEngineFixture(t)
+ *f.now = f.now.Add(300 * time.Second)
+ if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrRefused {
+ t.Fatal("expired new")
+ }
+ if countRows(t, f.db, "test_items") != 0 {
+ t.Fatal("expired side effect")
+ }
+ *f.now = f.now.Add(-300 * time.Second)
+ if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != nil {
+ t.Fatal(e)
+ }
+ if _, e := f.db.Exec("UPDATE directory_operator_results SET sealed_result=zeroblob(29)"); e != nil {
+ t.Fatal(e)
+ }
+ if _, e := f.engine.Execute(context.Background(), f.raw, insertOne); e != ErrUnavailable {
+ t.Fatal("corrupt result", e)
+ }
+}
diff --git a/internal/operator/migrations/credential/migrations/0001_operator.sql b/internal/operator/migrations/credential/migrations/0001_operator.sql
new file mode 100644
index 0000000..bb52cf6
--- /dev/null
+++ b/internal/operator/migrations/credential/migrations/0001_operator.sql
@@ -0,0 +1,19 @@
+CREATE TABLE credential_operator_state (
+ singleton INTEGER PRIMARY KEY CHECK(singleton=1),
+ origin TEXT NOT NULL,
+ installation TEXT NOT NULL,
+ registry TEXT NOT NULL,
+ device TEXT NOT NULL,
+ sequence INTEGER NOT NULL CHECK(typeof(sequence)='integer' AND sequence>0),
+ head TEXT NOT NULL CHECK(length(head)=43),
+ completed_at INTEGER NOT NULL CHECK(typeof(completed_at)='integer' AND completed_at>0)
+);
+CREATE TABLE credential_operator_results (
+ operation_id TEXT PRIMARY KEY CHECK(length(operation_id)=43),
+ action TEXT NOT NULL,
+ sequence INTEGER NOT NULL UNIQUE CHECK(typeof(sequence)='integer' AND sequence>0),
+ authorization_head TEXT NOT NULL CHECK(length(authorization_head)=43),
+ request_mac BLOB NOT NULL CHECK(typeof(request_mac)='blob' AND length(request_mac)=32),
+ sealed_result BLOB NOT NULL CHECK(typeof(sealed_result)='blob' AND length(sealed_result) BETWEEN 29 AND 49180),
+ completed_at INTEGER NOT NULL CHECK(typeof(completed_at)='integer' AND completed_at>0)
+);
diff --git a/internal/operator/migrations/directory/migrations/0001_operator.sql b/internal/operator/migrations/directory/migrations/0001_operator.sql
new file mode 100644
index 0000000..c6df4b5
--- /dev/null
+++ b/internal/operator/migrations/directory/migrations/0001_operator.sql
@@ -0,0 +1,19 @@
+CREATE TABLE directory_operator_state (
+ singleton INTEGER PRIMARY KEY CHECK(singleton=1),
+ origin TEXT NOT NULL,
+ installation TEXT NOT NULL,
+ registry TEXT NOT NULL,
+ device TEXT NOT NULL,
+ sequence INTEGER NOT NULL CHECK(typeof(sequence)='integer' AND sequence>0),
+ head TEXT NOT NULL CHECK(length(head)=43),
+ completed_at INTEGER NOT NULL CHECK(typeof(completed_at)='integer' AND completed_at>0)
+);
+CREATE TABLE directory_operator_results (
+ operation_id TEXT PRIMARY KEY CHECK(length(operation_id)=43),
+ action TEXT NOT NULL,
+ sequence INTEGER NOT NULL UNIQUE CHECK(typeof(sequence)='integer' AND sequence>0),
+ authorization_head TEXT NOT NULL CHECK(length(authorization_head)=43),
+ request_mac BLOB NOT NULL CHECK(typeof(request_mac)='blob' AND length(request_mac)=32),
+ sealed_result BLOB NOT NULL CHECK(typeof(sealed_result)='blob' AND length(sealed_result) BETWEEN 29 AND 49180),
+ completed_at INTEGER NOT NULL CHECK(typeof(completed_at)='integer' AND completed_at>0)
+);