amadan

oficina / home Public

Clone 
git clone https://amadan.net/oficina/home

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package provisioner
4
5import (
6 "bytes"
7 "context"
8 "database/sql"
9 "encoding/base64"
10 "errors"
11 "net/url"
12 "os"
13 "path/filepath"
14 "sync"
15 "time"
16
17 family "amadan.net/rastrillo/rastrillo/crypto"
18 _ "modernc.org/sqlite"
19)
20
21const recordContext = "oficina/provisioner/record/v1"
22const payloadContext = "oficina/provisioner/private-record/v1"
23
24type journalRecord struct {
25 Version int `json:"version"`
26 Identity floorIdentity `json:"identity"`
27 Installation string `json:"installation"`
28 Sequence int64 `json:"sequence"`
29 Previous string `json:"previous"`
30 Operation string `json:"operation"`
31 Action string `json:"action"`
32 Expected string `json:"expected_state"`
33 Created int64 `json:"created_at"`
34 CiphertextDigest string `json:"ciphertext_digest"`
35}
36type signedRecord struct {
37 Record journalRecord `json:"record"`
38 Signature string `json:"signature"`
39}
40type journal struct {
41 mu sync.Mutex
42 locks []*os.File
43 db *sql.DB
44 dir *privateDirectory
45 floor *nativeFloor
46 key *family.Keypair
47 now func() time.Time
48 boundary func(string) error
49}
50
51func openJournal(dir *privateDirectory, floor *nativeFloor, key *family.Keypair, create bool) (*journal, error) {
52 return openJournalWithLocks(dir, floor, key, create, nil, false)
53}
54
55func openJournalWithLocks(dir *privateDirectory, floor *nativeFloor, key *family.Keypair, create bool, held []*os.File, resumeCreate bool) (*journal, error) {
56 if dir == nil || floor == nil || key == nil || key.SignPriv == nil || key.BoxPriv == nil || !bytes.Equal(key.SignPub(), floor.public) {
57 return nil, ErrRefused
58 }
59 stateInfo, e := dir.file.Stat()
60 floorInfo, floorErr := floor.dir.file.Stat()
61 if e != nil || floorErr != nil || os.SameFile(stateInfo, floorInfo) {
62 return nil, ErrRefused
63 }
64 var stateLock, floorLock *os.File
65 if held == nil {
66 stateLock, e = dir.lock()
67 if e != nil {
68 return nil, e
69 }
70 floorLock, e = floor.dir.lock()
71 if e != nil {
72 stateLock.Close()
73 return nil, e
74 }
75 } else {
76 if len(held) != 2 {
77 return nil, ErrRefused
78 }
79 stateLock = held[0]
80 floorLock = held[1]
81 }
82 keepLocks := false
83 defer func() {
84 if !keepLocks {
85 floorLock.Close()
86 stateLock.Close()
87 }
88 }()
89 if create {
90 if e := dir.create("registry.db", nil); e != nil && !resumeCreate {
91 return nil, e
92 }
93 }
94 if _, e := dir.read("registry.db", 128<<10); e != nil {
95 // Registry size is bounded by record count and private payload limits, not
96 // the small-file reader; inspect its type without reading all ciphertext.
97 if e = checkRegistryFile(dir); e != nil {
98 return nil, e
99 }
100 }
101 u := url.URL{Scheme: "file", Path: filepath.Join(dir.file.Name(), "registry.db")}
102 q := u.Query()
103 q.Set("mode", "rw")
104 u.RawQuery = q.Encode()
105 db, e := sql.Open("sqlite", u.String())
106 if e != nil {
107 return nil, ErrUnavailable
108 }
109 db.SetMaxOpenConns(1)
110 j := &journal{db: db, dir: dir, floor: floor, key: key, now: time.Now, locks: []*os.File{stateLock, floorLock}}
111 fail := func() (*journal, error) { db.Close(); return nil, ErrUnavailable }
112 for _, statement := range []string{"PRAGMA foreign_keys=ON", "PRAGMA secure_delete=ON", "PRAGMA synchronous=FULL", "PRAGMA busy_timeout=1000", "PRAGMA journal_mode=WAL"} {
113 if _, e = db.Exec(statement); e != nil {
114 return fail()
115 }
116 }
117 var fk, secure, synchronous int
118 if db.QueryRow("PRAGMA foreign_keys").Scan(&fk) != nil || db.QueryRow("PRAGMA secure_delete").Scan(&secure) != nil || db.QueryRow("PRAGMA synchronous").Scan(&synchronous) != nil || fk != 1 || secure != 1 || synchronous != 2 {
119 return fail()
120 }
121 if create && resumeCreate {
122 var tables int
123 if e = db.QueryRow("SELECT count(*) FROM sqlite_master WHERE type='table'").Scan(&tables); e != nil {
124 return fail()
125 }
126 if tables != 0 {
127 create = false
128 }
129 }
130 if create {
131 point, _, e := floor.current()
132 if e != nil || point != (floorPoint{}) {
133 db.Close()
134 return nil, ErrRefused
135 }
136 _, e = db.Exec(`BEGIN IMMEDIATE;
137CREATE TABLE identity(singleton INTEGER PRIMARY KEY CHECK(singleton=1), registry TEXT NOT NULL, device TEXT NOT NULL, public_key TEXT NOT NULL, sequence INTEGER NOT NULL, head TEXT NOT NULL);
138CREATE TABLE records(sequence INTEGER PRIMARY KEY CHECK(sequence>0), operation TEXT NOT NULL UNIQUE, signed_record BLOB NOT NULL, sealed_payload BLOB NOT NULL, state TEXT NOT NULL CHECK(state IN ('prepared','finalized')));
139INSERT INTO identity VALUES(1,?,?,?,0,''); COMMIT;`, floor.identity.Registry, floor.identity.Device, floor.identity.Public)
140 if e != nil {
141 return fail()
142 }
143 if e = dir.file.Sync(); e != nil {
144 return fail()
145 }
146 }
147 if e = j.recover(); e != nil {
148 db.Close()
149 return nil, e
150 }
151 keepLocks = true
152 return j, nil
153}
154func (j *journal) close() error {
155 j.mu.Lock()
156 defer j.mu.Unlock()
157 err := j.db.Close()
158 for _, lock := range j.locks {
159 lock.Close()
160 }
161 j.locks = nil
162 return err
163}
164func (j *journal) stop(stage string) error {
165 if j.boundary != nil {
166 return j.boundary(stage)
167 }
168 return nil
169}
170func (j *journal) validate(raw, ciphertext []byte) (journalRecord, error) {
171 var signed signedRecord
172 if len(raw) > 8192 || len(ciphertext) > 96<<10 || closedJSON(raw, &signed) != nil {
173 return journalRecord{}, ErrRefused
174 }
175 r := signed.Record
176 _, installation := tokenBytes(r.Installation, 32)
177 _, operation := tokenBytes(r.Operation, 32)
178 if r.Version != 1 || r.Identity != j.floor.identity || !installation || !operation || r.Sequence <= 0 || r.Sequence > maxRecords || r.Created <= 0 || !recordAction(r.Action) || !recordExpectation(r.Expected) || r.CiphertextDigest != hash(ciphertext) {
179 return journalRecord{}, ErrRefused
180 }
181 body, e := canonicalJSON(r)
182 sig, ok := tokenBytes(signed.Signature, 64)
183 if e != nil || !ok || !family.Verify(j.floor.public, recordContext, body, sig) {
184 return journalRecord{}, ErrRefused
185 }
186 return r, nil
187}
188func recordAction(action string) bool {
189 switch action {
190 case "provisioner.enroll", "provisioner.in-flight", "provisioner.result", "provisioner.outcome", "credential.owner-seed", "directory.initialize", "directory.seed-team", "directory.provision-consumer", "directory.verify-consumer", "directory.activate-consumer":
191 return true
192 }
193 return false
194}
195func recordExpectation(expected string) bool {
196 switch expected {
197 case "absent", "unseeded", "empty", "pending", "live", "prepared", "in-flight", "outcome":
198 return true
199 }
200 return false
201}
202
203func (j *journal) recover() error {
204 var id floorIdentity
205 var accepted floorPoint
206 if e := j.db.QueryRow("SELECT registry,device,public_key,sequence,head FROM identity WHERE singleton=1").Scan(&id.Registry, &id.Device, &id.Public, &accepted.Sequence, &accepted.Head); e != nil {
207 return ErrUnavailable
208 }
209 if id != j.floor.identity {
210 return ErrRefused
211 }
212 rows, e := j.db.Query(`SELECT sequence,
213 CASE WHEN length(operation)=43 THEN operation ELSE '' END,
214 CASE WHEN typeof(signed_record)='blob' AND length(signed_record)<=8192 THEN signed_record ELSE NULL END,
215 CASE WHEN typeof(sealed_payload)='blob' AND length(sealed_payload)<=98304 THEN sealed_payload ELSE NULL END,
216 CASE WHEN length(state)<=9 THEN state ELSE '' END FROM records ORDER BY sequence`)
217 if e != nil {
218 return ErrUnavailable
219 }
220 var latest floorPoint
221 var pending *floorPoint
222 for rows.Next() {
223 var seq int64
224 var op, state string
225 var raw, sealed []byte
226 if rows.Scan(&seq, &op, &raw, &sealed, &state) != nil {
227 rows.Close()
228 return ErrUnavailable
229 }
230 r, e := j.validate(raw, sealed)
231 if e != nil || pending != nil || seq != latest.Sequence+1 || r.Sequence != seq || r.Operation != op || r.Previous != latest.Head {
232 rows.Close()
233 return ErrRefused
234 }
235 point := floorPoint{seq, hash(raw)}
236 if state == "prepared" {
237 if latest != accepted {
238 rows.Close()
239 return ErrRefused
240 }
241 pending = &point
242 } else if state != "finalized" {
243 rows.Close()
244 return ErrRefused
245 }
246 latest = point
247 }
248 e = rows.Err()
249 rows.Close()
250 if e != nil {
251 return ErrUnavailable
252 }
253 if pending == nil && latest != accepted {
254 return ErrRefused
255 }
256 current, _, e := j.floor.current()
257 if e != nil {
258 return e
259 }
260 if pending == nil {
261 if current != accepted {
262 return ErrRefused
263 }
264 return nil
265 }
266 if current == accepted {
267 if e = j.floor.advance(j.key, accepted, *pending); e != nil {
268 return e
269 }
270 } else if current != *pending {
271 return ErrRefused
272 }
273 return j.finalize(accepted, *pending)
274}
275func (j *journal) finalize(old, next floorPoint) error {
276 tx, e := j.db.Begin()
277 if e != nil {
278 return ErrUnavailable
279 }
280 defer tx.Rollback()
281 result, e := tx.Exec("UPDATE identity SET sequence=?,head=? WHERE singleton=1 AND sequence=? AND head=?", next.Sequence, next.Head, old.Sequence, old.Head)
282 if e != nil {
283 return ErrUnavailable
284 }
285 n, e := result.RowsAffected()
286 if e != nil || n != 1 {
287 return ErrRefused
288 }
289 result, e = tx.Exec("UPDATE records SET state='finalized' WHERE sequence=? AND state='prepared'", next.Sequence)
290 if e != nil {
291 return ErrUnavailable
292 }
293 n, e = result.RowsAffected()
294 if e != nil || n != 1 {
295 return ErrRefused
296 }
297 if tx.Commit() != nil {
298 return ErrUnavailable
299 }
300 if _, e = j.db.Exec("PRAGMA wal_checkpoint(FULL)"); e != nil || j.dir.file.Sync() != nil {
301 return ErrUnavailable
302 }
303 return nil
304}
305func (j *journal) append(ctx context.Context, installation, operation, action, expected string, payload []byte) ([]byte, error) {
306 if j == nil || len(payload) == 0 || len(payload) > 64<<10 {
307 return nil, ErrRefused
308 }
309 j.mu.Lock()
310 defer j.mu.Unlock()
311 return j.appendLocked(ctx, installation, operation, action, expected, payload)
312}
313func (j *journal) appendLocked(ctx context.Context, installation, operation, action, expected string, payload []byte) ([]byte, error) {
314 return j.appendAtLocked(ctx, installation, operation, action, expected, payload, j.now().Unix())
315}
316func (j *journal) appendAtLocked(ctx context.Context, installation, operation, action, expected string, payload []byte, created int64) ([]byte, error) {
317 if j == nil || len(payload) == 0 || len(payload) > 64<<10 {
318 return nil, ErrRefused
319 }
320 if len(j.locks) != 2 {
321 return nil, ErrRefused
322 }
323 if e := j.recover(); e != nil {
324 return nil, e
325 }
326 var oldRaw, oldSealed []byte
327 e := j.db.QueryRowContext(ctx, "SELECT CASE WHEN typeof(signed_record)='blob' AND length(signed_record)<=8192 THEN signed_record ELSE NULL END, CASE WHEN typeof(sealed_payload)='blob' AND length(sealed_payload)<=98304 THEN sealed_payload ELSE NULL END FROM records WHERE operation=?", operation).Scan(&oldRaw, &oldSealed)
328 if e == nil {
329 r, e := j.validate(oldRaw, oldSealed)
330 if e != nil || r.Installation != installation || r.Action != action || r.Expected != expected {
331 return nil, ErrRefused
332 }
333 plain, e := family.Open(j.key, payloadContext+"/"+operation, oldSealed)
334 if e != nil || !bytes.Equal(plain, payload) {
335 return nil, ErrRefused
336 }
337 return oldRaw, nil
338 }
339 if !errors.Is(e, sql.ErrNoRows) {
340 return nil, ErrUnavailable
341 }
342 old, _, e := j.floor.current()
343 if e != nil {
344 return nil, e
345 }
346 if old.Sequence >= maxRecords {
347 return nil, ErrRefused
348 }
349 sealed, e := family.Seal(j.key.BoxPub(), payloadContext+"/"+operation, payload)
350 if e != nil {
351 return nil, ErrUnavailable
352 }
353 r := journalRecord{1, j.floor.identity, installation, old.Sequence + 1, old.Head, operation, action, expected, created, hash(sealed)}
354 body, e := canonicalJSON(r)
355 if e != nil {
356 return nil, e
357 }
358 sig, e := family.Sign(j.key, recordContext, body)
359 if e != nil {
360 return nil, ErrUnavailable
361 }
362 raw, e := canonicalJSON(signedRecord{r, base64.RawURLEncoding.EncodeToString(sig)})
363 if e != nil {
364 return nil, e
365 }
366 if _, e = j.validate(raw, sealed); e != nil {
367 return nil, e
368 }
369 if _, e = j.db.ExecContext(ctx, "INSERT INTO records VALUES(?,?,?,?,'prepared')", r.Sequence, operation, raw, sealed); e != nil {
370 return nil, ErrRefused
371 }
372 if e = j.stop("prepared"); e != nil {
373 return nil, e
374 }
375 next := floorPoint{r.Sequence, hash(raw)}
376 if e = j.floor.advance(j.key, old, next); e != nil {
377 return nil, e
378 }
379 if e = j.stop("floor"); e != nil {
380 return nil, e
381 }
382 if e = j.finalize(old, next); e != nil {
383 return nil, e
384 }
385 if e = j.stop("finalized"); e != nil {
386 return nil, e
387 }
388 return raw, nil
389}
390