amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package hostbox
4
5import (
6 "context"
7 "crypto/sha256"
8 "database/sql"
9 "encoding/hex"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "io"
14 "net/url"
15 "os"
16 "path/filepath"
17 "sort"
18 "strings"
19 "time"
20
21 "github.com/carlosframework/platform/internal/hostguard"
22 "golang.org/x/sys/unix"
23)
24
25// ErrLegacyJournalMigration refuses root startup over legacy claim journals
26// that were never imported. Only the offline operator step imports them.
27var ErrLegacyJournalMigration = errors.New("legacy claim journals need the offline migration: run carlos system journal-migrate")
28
29// CheckLegacyMigration is root startup's only contact with the legacy
30// journals, which live in a directory the agent owns. It never opens them as
31// SQLite and never quiesces anything. With no legacy files and no marker (a
32// fresh box), it records the marker. Unimported legacy files refuse with
33// ErrLegacyJournalMigration. Once the marker exists, nothing the agent puts
34// there can stop root: legacy files that differ from the marked content are
35// ignored, and the returned alert says so.
36func (s *JournalService) CheckLegacyMigration(ctx context.Context, legacy string) (alert error, err error) {
37 if s == nil || s.Launches == nil || !filepath.IsAbs(legacy) {
38 return nil, ErrTypedJournal
39 }
40 unlock, err := lockJournalMigration(ctx, filepath.Join(filepath.Dir(s.Launches.path), "migration.lock"))
41 if err != nil {
42 return nil, err
43 }
44 defer unlock()
45 source, markedFiles, _, marked, err := s.journalMigrationMarker(ctx)
46 if err != nil {
47 return nil, err
48 }
49 files, present, err := legacyJournalFiles(legacy)
50 if marked {
51 if source != legacy {
52 return nil, ErrChanged
53 }
54 if err == nil && files == "absent" {
55 err = os.ErrNotExist
56 }
57 if err != nil {
58 return fmt.Errorf("legacy claim journal %s ignored after migration: %w", legacy, err), nil
59 }
60 if present == 0 || legacyContent(markedFiles) == files {
61 return nil, nil
62 }
63 return fmt.Errorf("legacy claim journal %s changed after migration and is ignored: %w", legacy, ErrChanged), nil
64 }
65 if err != nil {
66 return nil, err
67 }
68 if present != 0 {
69 return nil, ErrLegacyJournalMigration
70 }
71 tx, err := s.Launches.db.BeginTx(ctx, nil)
72 if err != nil {
73 return nil, err
74 }
75 defer tx.Rollback()
76 var count int
77 if err = tx.QueryRowContext(ctx, "SELECT (SELECT count(*) FROM launches)+(SELECT count(*) FROM stop_inventory)").Scan(&count); err != nil || count != 0 {
78 return nil, errors.Join(err, ErrTypedJournal)
79 }
80 if _, err = tx.ExecContext(ctx, "INSERT INTO journal_migration(version,source,files,rows) VALUES(1,?,?,?)", legacy, files, legacyRowsDigest(nil, nil)); err != nil {
81 return nil, err
82 }
83 return nil, tx.Commit()
84}
85
86// journalMigrationMarker creates or upgrades the marker table and reads it.
87func (s *JournalService) journalMigrationMarker(ctx context.Context) (source, files, rows string, marked bool, err error) {
88 if _, err = s.Launches.db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS journal_migration(version INTEGER PRIMARY KEY CHECK(version=1), source TEXT NOT NULL, files TEXT NOT NULL DEFAULT '', rows TEXT NOT NULL DEFAULT '')"); err != nil {
89 return
90 }
91 for _, column := range []string{"files", "rows"} {
92 var n int
93 if err = s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM pragma_table_info('journal_migration') WHERE name=?", column).Scan(&n); err != nil {
94 return
95 }
96 if n == 0 {
97 if _, err = s.Launches.db.ExecContext(ctx, "ALTER TABLE journal_migration ADD COLUMN "+column+" TEXT NOT NULL DEFAULT ''"); err != nil {
98 return
99 }
100 }
101 }
102 err = s.Launches.db.QueryRowContext(ctx, "SELECT source,files,rows FROM journal_migration WHERE version=1").Scan(&source, &files, &rows)
103 if errors.Is(err, sql.ErrNoRows) {
104 return "", "", "", false, nil
105 }
106 return source, files, rows, err == nil, err
107}
108
109// MigrateLegacy is the offline operator step (carlos system journal-migrate);
110// root startup only runs CheckLegacyMigration. Quiesce must prove the old
111// writers are stopped and held stopped for this call. Root never opens SQLite
112// on the old paths: bounded no-follow copies, including WAL, are made in the
113// protected directory, hashed as they are copied, and opened with an untrusted
114// schema, read-only. Every imported field is validated and re-marshalled; no
115// raw body is stored. The transaction imports both inventories and the marker
116// together. A crash before commit imports nothing; a crash afterward never
117// imports a stale legacy database again. The marker binds the digests of the
118// copied legacy bytes and of the rows it imported: if a rolled-back binary
119// later wrote rows there, this step refuses rather than silently ignoring
120// them. Legacy originals remain untouched.
121func (s *JournalService) MigrateLegacy(ctx context.Context, legacy, box, enrollment string, quiesce func(context.Context) (func(), error), verify func(context.Context, LaunchInventory, StopInventory, StopKernelJournal) error, onSuccess ...func(context.Context) error) (resultErr error) {
122 if s == nil || s.Launches == nil || !filepath.IsAbs(legacy) || box == "" || enrollment == "" {
123 return ErrTypedJournal
124 }
125 unlock, err := lockJournalMigration(ctx, filepath.Join(filepath.Dir(s.Launches.path), "migration.lock"))
126 if err != nil {
127 return err
128 }
129 defer unlock()
130 defer func() {
131 if resultErr == nil {
132 for _, recover := range onSuccess {
133 if recover != nil {
134 resultErr = errors.Join(resultErr, recover(ctx))
135 }
136 }
137 }
138 }()
139 // These private snapshots are never authoritative before the SQL marker.
140 // A restart can discard a pre-commit copy and take a fresh frozen snapshot.
141 entries, err := os.ReadDir(filepath.Dir(s.Launches.path))
142 if err != nil {
143 return err
144 }
145 for _, entry := range entries {
146 if entry.IsDir() && strings.HasPrefix(entry.Name(), "migration-") {
147 if err := os.RemoveAll(filepath.Join(filepath.Dir(s.Launches.path), entry.Name())); err != nil {
148 return err
149 }
150 }
151 }
152 source, markedFiles, markedRows, marked, err := s.journalMigrationMarker(ctx)
153 if err != nil {
154 return err
155 }
156 files, present, err := legacyJournalFiles(legacy)
157 if err != nil {
158 return err
159 }
160 if present != 0 && present != 2 {
161 return ErrTypedJournal
162 }
163 if marked {
164 if source != legacy {
165 return ErrChanged
166 }
167 // Unchanged legacy content, or legacy files removed after import:
168 // nothing can have been written there since the marker.
169 if present == 0 || legacyContent(markedFiles) == files {
170 return nil
171 }
172 }
173 if present == 2 {
174 if quiesce == nil {
175 return ErrTypedJournal
176 }
177 release, err := quiesce(ctx)
178 if err != nil {
179 return err
180 }
181 if release == nil {
182 return ErrTypedJournal
183 }
184 defer release()
185 // Stopping the old writers may itself checkpoint; the snapshot is of
186 // the quiesced files.
187 if files, present, err = legacyJournalFiles(legacy); err != nil {
188 return err
189 }
190 if present != 2 {
191 return ErrChanged
192 }
193 }
194 // Prepare and verify bounded immutable records while no root SQLite
195 // connection or write transaction is held. The offline flock and edge
196 // exclusion remain held until the later atomic import finishes.
197 launches, inventories, copied, err := readLegacyJournals(ctx, legacy, filepath.Dir(s.Launches.path), box, enrollment)
198 if err != nil {
199 return err
200 }
201 // A box that never ran the legacy claim writer has no legacy directory
202 // at all (leo, 2026-10-19). The copy names each missing file instead of
203 // "absent"; both say there is nothing to import.
204 if files == "absent" && copied == legacyNoJournals {
205 copied = files
206 }
207 // The rows came from exactly the bytes the identity names: the digests
208 // are of the copies themselves, not of a later re-read.
209 if copied != files {
210 return fmt.Errorf("legacy journal changed while copied: %w", ErrChanged)
211 }
212 rows := legacyRowsDigest(launches, inventories)
213 if marked {
214 // The content changed after import. Only identical rows (a checkpoint)
215 // keep the marker; new history from a rolled-back binary needs the
216 // operator, never a silent skip or an unproven import. A marker from
217 // before rows were bound (4f342ec7) is bound now, but only when root
218 // already holds every legacy row.
219 if markedRows == "" {
220 if err = s.holdsLegacyRows(ctx, launches, inventories); err != nil {
221 return fmt.Errorf("legacy claim journal %s changed after migration: %w", legacy, err)
222 }
223 } else if rows != markedRows {
224 return fmt.Errorf("legacy claim journal %s changed after migration: %w", legacy, ErrChanged)
225 }
226 _, err = s.Launches.db.ExecContext(ctx, "UPDATE journal_migration SET files=?, rows=? WHERE version=1 AND rows=?", files, rows, markedRows)
227 return err
228 }
229 var count int
230 if err = s.Launches.db.QueryRowContext(ctx, "SELECT (SELECT count(*) FROM launches)+(SELECT count(*) FROM stop_inventory)").Scan(&count); err != nil || count != 0 {
231 return errors.Join(err, ErrTypedJournal)
232 }
233 // Copied rows are untrusted until root's signed stop evidence binds both
234 // representations. Unpublished, active or edited histories fail closed.
235 // Budget one roster read plus a namespace list, receipt and signature per
236 // prepared row. Shared namespaces are cached by the verifier. Each remote
237 // read is separately bounded to 30 seconds; caller cancellation always wins.
238 // The 10,000-row cap makes the worst batch budget 900,030 seconds.
239 proofCtx, cancelProof := context.WithTimeout(ctx, time.Duration(1+3*len(launches))*30*time.Second)
240 defer cancelProof()
241 covered := map[string]bool{}
242 for _, in := range launches {
243 if verify == nil {
244 return ErrLegacyJournalProof
245 }
246 inventory, ok := inventories[fmt.Sprintf("%s/%s/%d", in.Target.Host, in.Target.Incarnation, in.Target.Stamp)]
247 if !ok {
248 return ErrLegacyJournalProof
249 }
250 journal, err := launchStopKernelJournal(in, in.Target)
251 if err != nil {
252 return err
253 }
254 if err = ctx.Err(); err != nil {
255 return err
256 }
257 err = verify(proofCtx, in, inventory, journal)
258 if err == nil {
259 err = proofCtx.Err()
260 }
261 if err != nil {
262 return errors.Join(err, ErrLegacyJournalProof)
263 }
264 covered[fmt.Sprintf("%s/%s/%d", in.Target.Host, in.Target.Incarnation, in.Target.Stamp)] = true
265 }
266 if len(inventories) != len(covered) {
267 return ErrLegacyJournalProof
268 }
269 if err = proofCtx.Err(); err != nil {
270 return err
271 }
272 cancelProof()
273 tx, err := s.Launches.db.BeginTx(ctx, nil)
274 if err != nil {
275 return err
276 }
277 defer tx.Rollback()
278 if err = tx.QueryRowContext(ctx, "SELECT (SELECT count(*) FROM launches)+(SELECT count(*) FROM stop_inventory)").Scan(&count); err != nil || count != 0 {
279 return errors.Join(err, ErrTypedJournal)
280 }
281 for _, in := range launches {
282 body, err := json.Marshal(in)
283 if err != nil {
284 return err
285 }
286 if _, err = tx.ExecContext(ctx, "INSERT INTO launches(identity,body) VALUES(?,?)", launchKey(in.Target), body); err != nil {
287 return err
288 }
289 }
290 txctx := context.WithValue(ctx, journalTransactionKey{}, tx)
291 for _, in := range inventories {
292 if err = s.Inventory.WriteStopInventory(txctx, in); err != nil {
293 return err
294 }
295 }
296 if _, err = tx.ExecContext(ctx, "INSERT INTO journal_migration(version,source,files,rows) VALUES(1,?,?,?)", legacy, files, rows); err != nil {
297 return err
298 }
299 return tx.Commit()
300}
301
302// Legacy import bounds. Root's instance-control runs under MemoryMax=256M, so
303// no legacy file, row or cell may be read into memory before it is known to
304// fit. Files are capped at the 16 MiB import budget plus slack for SQLite
305// pages, WAL frames and free space. Cells are sized by SQL before any Scan.
306const (
307 legacyJournalFileMax = 64 << 20
308 legacyImportMax = 16 << 20
309 legacyRowMax = 1 << 20
310 legacyKeyMax = 4096
311 legacyRowsMax = 10000
312)
313
314// holdsLegacyRows proves every legacy row's key is already in root's journal.
315func (s *JournalService) holdsLegacyRows(ctx context.Context, launches []LaunchInventory, inventories map[string]StopInventory) error {
316 for _, in := range launches {
317 var n int
318 if err := s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM launches WHERE identity=?", launchKey(in.Target)).Scan(&n); err != nil || n != 1 {
319 return errors.Join(err, ErrChanged)
320 }
321 }
322 for _, in := range inventories {
323 var n int
324 if err := s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM stop_inventory WHERE host=? AND incarnation=? AND stamp=?", in.Host, in.Incarnation, in.InstallationStamp).Scan(&n); err != nil || n != 1 {
325 return errors.Join(err, ErrChanged)
326 }
327 }
328 return nil
329}
330
331// legacyContent reduces a marker's file identity to names and content
332// digests. Markers from 618f3a8a also carry device, inode, size and times;
333// the digest is their last field. A metadata-only change is not a change.
334func legacyContent(files string) string {
335 if files == "" || files == "absent" {
336 return files
337 }
338 parts := strings.Split(files, "/")
339 for i, part := range parts {
340 fields := strings.Split(part, ":")
341 parts[i] = fields[0] + ":" + fields[len(fields)-1]
342 }
343 return strings.Join(parts, "/")
344}
345
346// legacyJournalFiles names the exact legacy files by the SHA-256 of the bytes
347// of each database and WAL ("-" when absent). Only no-follow, single-link
348// regular files within the copy bound are read.
349func legacyJournalFiles(legacy string) (string, int, error) {
350 dir, err := unix.Open(legacy, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
351 if errors.Is(err, unix.ENOENT) {
352 return "absent", 0, nil
353 }
354 if err != nil {
355 return "", 0, err
356 }
357 defer unix.Close(dir)
358 var parts []string
359 present := 0
360 for _, name := range []string{"launches.db", "launches.db-wal", "inventory.db", "inventory.db-wal"} {
361 fd, err := unix.Openat(dir, name, unix.O_RDONLY|unix.O_NOFOLLOW|unix.O_NONBLOCK|unix.O_CLOEXEC, 0)
362 if errors.Is(err, unix.ENOENT) {
363 parts = append(parts, name+":-")
364 continue
365 }
366 if err != nil {
367 return "", 0, err
368 }
369 f := os.NewFile(uintptr(fd), name)
370 part, err := func() (string, error) {
371 defer f.Close()
372 var st unix.Stat_t
373 if err := unix.Fstat(fd, &st); err != nil {
374 return "", err
375 }
376 if st.Mode&unix.S_IFMT != unix.S_IFREG || st.Nlink != 1 || st.Size > legacyJournalFileMax {
377 return "", ErrTypedJournal
378 }
379 h := sha256.New()
380 if n, err := io.Copy(h, io.LimitReader(f, legacyJournalFileMax+1)); err != nil || n != st.Size {
381 return "", errors.Join(err, ErrChanged)
382 }
383 return fmt.Sprintf("%s:%x", name, h.Sum(nil)), nil
384 }()
385 if err != nil {
386 return "", 0, err
387 }
388 if !strings.HasSuffix(name, "-wal") {
389 present++
390 }
391 parts = append(parts, part)
392 }
393 return strings.Join(parts, "/"), present, nil
394}
395
396// legacyNoJournals is readLegacyJournals' identity when neither legacy
397// journal nor its write-ahead log exists.
398const legacyNoJournals = "launches.db:-/launches.db-wal:-/inventory.db:-/inventory.db-wal:-"
399
400// legacyRowsDigest is the order-independent digest of the canonical rows.
401func legacyRowsDigest(launches []LaunchInventory, inventories map[string]StopInventory) string {
402 var rows []string
403 for _, in := range launches {
404 body, _ := json.Marshal(in)
405 rows = append(rows, "launch\x00"+launchKey(in.Target)+"\x00"+string(body))
406 }
407 for key, in := range inventories {
408 body, _ := json.Marshal(in)
409 rows = append(rows, "inventory\x00"+key+"\x00"+string(body))
410 }
411 sort.Strings(rows)
412 sum := sha256.Sum256([]byte(strings.Join(rows, "\x01")))
413 return hex.EncodeToString(sum[:])
414}
415
416// readLegacyJournals also returns the content identity of the bytes it copied,
417// in legacyJournalFiles' form.
418func readLegacyJournals(ctx context.Context, legacy, dest, box, enrollment string) ([]LaunchInventory, map[string]StopInventory, string, error) {
419 var launches []LaunchInventory
420 keys := map[string]bool{}
421 inventories := map[string]StopInventory{}
422 preparedBytes := 0
423 var identity []string
424 for _, name := range []string{"launches.db", "inventory.db"} {
425 copied, digests, cleanup, err := copyLegacyJournal(legacy, name, dest)
426 if errors.Is(err, os.ErrNotExist) {
427 identity = append(identity, name+":-", name+"-wal:-")
428 continue
429 }
430 if err != nil {
431 return nil, nil, "", err
432 }
433 identity = append(identity, digests...)
434 err = func() error {
435 defer cleanup()
436 // The copy is agent-supplied bytes: no schema-defined SQL runs, no
437 // statement writes, and the whole read is time bounded. The schema
438 // must be exactly the legacy code's before integrity_check reads a
439 // row, and integrity_check never evaluates a CHECK expression.
440 readCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
441 defer cancel()
442 db, err := sql.Open("sqlite", "file:"+url.PathEscape(copied)+"?_pragma=trusted_schema(0)&_pragma=query_only(1)&_pragma=ignore_check_constraints(1)&_pragma=busy_timeout(5000)")
443 if err != nil {
444 return err
445 }
446 defer db.Close()
447 db.SetMaxOpenConns(1)
448 tables := map[string]bool{"launches": true}
449 if name == "inventory.db" {
450 tables = map[string]bool{"stop_inventory": true, "journal_inventory_versions": true}
451 }
452 if err = checkLegacySchema(readCtx, db, tables); err != nil {
453 return err
454 }
455 var result string
456 if err = db.QueryRowContext(readCtx, "PRAGMA integrity_check").Scan(&result); err != nil || result != "ok" {
457 return errors.Join(err, ErrTypedJournal)
458 }
459 // Size every scanned cell before Scan allocates it. octet_length
460 // reads the stored length without loading the value.
461 bounds := "SELECT count(*), coalesce(sum(octet_length(body)),0), coalesce(max(octet_length(body)),0), coalesce(max(octet_length(identity)),0), 0 FROM launches"
462 if name == "inventory.db" {
463 bounds = "SELECT count(*), coalesce(sum(octet_length(body)),0), coalesce(max(octet_length(body)),0), coalesce(max(max(octet_length(host)),max(octet_length(incarnation))),0), coalesce(sum(typeof(stamp)<>'integer'),0) FROM stop_inventory"
464 }
465 var count, total, row, key, untyped int64
466 if err = db.QueryRowContext(readCtx, bounds).Scan(&count, &total, &row, &key, &untyped); err != nil {
467 return errors.Join(err, ErrTypedJournal)
468 }
469 if count > legacyRowsMax || row > legacyRowMax || key > legacyKeyMax || untyped != 0 || int64(preparedBytes)+total > legacyImportMax {
470 return ErrTypedJournal
471 }
472 if name == "launches.db" {
473 rows, err := db.QueryContext(readCtx, "SELECT identity,body FROM launches")
474 if err != nil {
475 return err
476 }
477 defer rows.Close()
478 for rows.Next() {
479 var key string
480 var raw []byte
481 if err = rows.Scan(&key, &raw); err != nil {
482 return err
483 }
484 preparedBytes += len(raw)
485 if preparedBytes > 16<<20 || len(launches) >= 10000 {
486 return ErrTypedJournal
487 }
488 var in LaunchInventory
489 if decodeStrict(raw, &in) != nil || validateImportedLaunch(in) != nil || key != launchKey(in.Target) || in.Target.Box != box || in.Target.Enrollment != enrollment || keys[key] {
490 return ErrTypedJournal
491 }
492 keys[key] = true
493 launches = append(launches, in)
494 }
495 return rows.Err()
496 }
497 rows, err := db.QueryContext(readCtx, "SELECT host,incarnation,stamp,body FROM stop_inventory")
498 if err != nil {
499 return err
500 }
501 defer rows.Close()
502 for rows.Next() {
503 var host, inc string
504 var stamp uint64
505 var raw []byte
506 if err = rows.Scan(&host, &inc, &stamp, &raw); err != nil {
507 return err
508 }
509 preparedBytes += len(raw)
510 if preparedBytes > 16<<20 || len(inventories) >= 10000 {
511 return ErrTypedJournal
512 }
513 var in StopInventory
514 if decodeStrict(raw, &in) != nil || validateImportedInventory(in) != nil || in.Host != host || in.Incarnation != inc || in.InstallationStamp != stamp || in.Box != box || in.Enrollment != enrollment {
515 return ErrTypedJournal
516 }
517 key := fmt.Sprintf("%s/%s/%d", host, inc, stamp)
518 if _, exists := inventories[key]; exists {
519 return ErrTypedJournal
520 }
521 inventories[key] = in
522 }
523 return rows.Err()
524 }()
525 if err != nil {
526 return nil, nil, "", err
527 }
528 }
529 return launches, inventories, strings.Join(identity, "/"), nil
530}
531
532// legacyTableSQL is, byte for byte, the CREATE text each legacy table was
533// written with (SQLite stores it without IF NOT EXISTS).
534var legacyTableSQL = map[string]string{
535 "launches": "CREATE TABLE launches (identity TEXT PRIMARY KEY, body BLOB NOT NULL)",
536 "stop_inventory": "CREATE TABLE stop_inventory(host TEXT NOT NULL,incarnation TEXT NOT NULL,stamp INTEGER NOT NULL,body BLOB NOT NULL,PRIMARY KEY(host,incarnation,stamp))",
537 "journal_inventory_versions": "CREATE TABLE journal_inventory_versions(host TEXT NOT NULL, incarnation TEXT NOT NULL, stamp INTEGER NOT NULL, revision INTEGER NOT NULL CHECK(revision>0 AND typeof(revision)='integer'), request TEXT NOT NULL, PRIMARY KEY(host,incarnation,stamp))",
538}
539
540// checkLegacySchema admits only the named tables, each with exactly its legacy
541// CREATE text and a real root page, and their implicit primary key indexes,
542// which have no SQL text. Triggers, views, virtual tables (root page zero,
543// whatever the text says), expression indexes and any other schema object
544// refuse, so no schema-defined expression is left for a read to run.
545func checkLegacySchema(ctx context.Context, db *sql.DB, tables map[string]bool) error {
546 rows, err := db.QueryContext(ctx, "SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema")
547 if err != nil {
548 return errors.Join(err, ErrTypedJournal)
549 }
550 defer rows.Close()
551 var present []string
552 for rows.Next() {
553 var kind, name, table string
554 var root sql.NullInt64
555 var text sql.NullString
556 if err = rows.Scan(&kind, &name, &table, &root, &text); err != nil {
557 return errors.Join(err, ErrTypedJournal)
558 }
559 switch {
560 case !root.Valid || root.Int64 <= 0:
561 return ErrTypedJournal
562 case kind == "table" && tables[name] && name == table && text.Valid && text.String == legacyTableSQL[name]:
563 present = append(present, name)
564 case kind == "index" && tables[table] && strings.HasPrefix(name, "sqlite_autoindex_"+table+"_") && !text.Valid:
565 default:
566 return ErrTypedJournal
567 }
568 }
569 if err = rows.Err(); err != nil {
570 return err
571 }
572 // One connection: the schema rows are closed before each column query.
573 rows.Close()
574 for _, table := range present {
575 var computed int
576 if err = db.QueryRowContext(ctx, "SELECT count(*) FROM pragma_table_xinfo(?) WHERE hidden<>0", table).Scan(&computed); err != nil || computed != 0 {
577 return errors.Join(err, ErrTypedJournal)
578 }
579 }
580 return nil
581}
582
583func legacyToken(s string) bool {
584 if s == "" || len(s) > 256 {
585 return false
586 }
587 for _, r := range s {
588 if r < 0x21 || r > 0x7e {
589 return false
590 }
591 }
592 return true
593}
594func legacyDigits(s string) bool {
595 if s == "" || len(s) > 32 {
596 return false
597 }
598 for _, r := range s {
599 if r < '0' || r > '9' {
600 return false
601 }
602 }
603 return true
604}
605func legacyTarget(t hostguard.FenceTarget) bool {
606 for _, s := range []string{t.Region, t.ObligationID, t.FenceID, t.Op, t.Phase, t.ClaimETag, t.DirectoryRevision} {
607 if s != "" && !legacyToken(s) {
608 return false
609 }
610 }
611 return validLaunchOwner(t) && component(t.Host) && legacyToken(t.Incarnation) && legacyToken(t.Box) && legacyToken(t.Enrollment) && legacyToken(t.BootID) && legacyToken(t.ControllerID)
612}
613func legacyGroup(g LaunchGroup, identity bool) bool {
614 if !managedName(g.RelativePath) || (g.InvocationID != "" && !legacyToken(g.InvocationID)) {
615 return false
616 }
617 return identity == (g.Device != 0 && g.Inode != 0) && (identity || (g.Device == 0 && g.Inode == 0))
618}
619
620// validateImportedLaunch accepts only a launch record root's own journal
621// methods could have produced. Every field is checked.
622func validateImportedLaunch(in LaunchInventory) error {
623 if !legacyTarget(in.Target) || in.ControllerPID <= 0 || !legacyDigits(in.ProcessStart) || !in.Completed.Valid() || len(in.Groups) == 0 || len(in.Groups) > 4096 || len(in.Files) > 4096 || len(in.ProducerIDs) > 4096 || len(in.Reservations) > 4096 {
624 return ErrTypedJournal
625 }
626 if in.NoState && len(in.Files) != 0 {
627 return ErrTypedJournal
628 }
629 groups := map[string]bool{}
630 for _, g := range in.Groups {
631 if !legacyGroup(g, true) || groups[g.RelativePath] {
632 return ErrTypedJournal
633 }
634 groups[g.RelativePath] = true
635 }
636 files := map[LaunchFile]bool{}
637 for _, f := range in.Files {
638 if f.Device == 0 || f.Inode == 0 || files[f] {
639 return ErrTypedJournal
640 }
641 files[f] = true
642 }
643 producers := map[string]bool{}
644 for _, p := range in.ProducerIDs {
645 if !component(p) || !legacyToken(p) || producers[p] {
646 return ErrTypedJournal
647 }
648 producers[p] = true
649 }
650 ids := map[string]bool{}
651 for _, r := range in.Reservations {
652 if !legacyToken(r.ID) || !legacyToken(r.Role) || !legacyToken(r.InvocationID) || !legacyToken(r.Inputs) || ids[r.ID] || !SameLaunchOwner(r.Target, in.Target) || !legacyTarget(r.Target) || r.Executable.Device == 0 || r.Executable.Inode == 0 || r.Group.InvocationID != r.InvocationID {
653 return ErrTypedJournal
654 }
655 ids[r.ID] = true
656 if r.Parent != (LaunchGroup{}) && !legacyGroup(r.Parent, true) {
657 return ErrTypedJournal
658 }
659 switch r.State {
660 case "intended":
661 if !legacyGroup(r.Group, false) || r.PID != 0 || r.ProcessStart != "" {
662 return ErrTypedJournal
663 }
664 case "reserved":
665 if !legacyGroup(r.Group, true) || r.PID != 0 || r.ProcessStart != "" {
666 return ErrTypedJournal
667 }
668 case "started", "settled":
669 if !legacyGroup(r.Group, true) || r.PID < 0 || (r.PID == 0) != (r.ProcessStart == "") || (r.ProcessStart != "" && !legacyDigits(r.ProcessStart)) || (r.State == "started" && r.PID == 0) {
670 return ErrTypedJournal
671 }
672 default:
673 return ErrTypedJournal
674 }
675 }
676 return nil
677}
678
679func validateImportedInventory(in StopInventory) error {
680 if !component(in.Host) || !legacyToken(in.Incarnation) || in.InstallationStamp == 0 || !legacyToken(in.Box) || !legacyToken(in.Enrollment) || !legacyToken(in.DirectoryRevision) || (len(in.Entries) == 0 && !in.Complete) || len(in.Entries) > 4096 {
681 return ErrTypedJournal
682 }
683 for _, e := range in.Entries {
684 if !component(e.Kind) || !component(e.ArtifactID) || !component(e.ManagedRoot) || !managedName(e.RelativePath) || !legacyToken(e.Ownership) || (e.RowIdentity != "" && !legacyToken(e.RowIdentity)) || (e.Materialization != "intended" && e.Materialization != "materialized") {
685 return ErrTypedJournal
686 }
687 }
688 return nil
689}
690
691// legacyCopyHook lets tests act on a legacy file just before and just after
692// it is copied.
693var legacyCopyHook func(name string, copied bool)
694
695// copyLegacyJournal copies the database and its WAL, returning each file's
696// identity part ("name:sha256", or "name:-" for an absent WAL) computed from
697// the bytes written to the copy.
698func copyLegacyJournal(dir, name, dest string) (string, []string, func(), error) {
699 // A directory fd anchors the source components. The quiescence boundary,
700 // not a pair of pathname checks, makes the copied snapshot coherent.
701 fd, err := unix.Open(dir, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
702 if err != nil {
703 return "", nil, nil, err
704 }
705 defer unix.Close(fd)
706 temp, err := os.MkdirTemp(dest, "migration-")
707 if err != nil {
708 return "", nil, nil, err
709 }
710 clean := func() { _ = os.RemoveAll(temp) }
711 path := filepath.Join(temp, name)
712 var digests []string
713 for _, suffix := range []string{"", "-wal"} {
714 if legacyCopyHook != nil {
715 legacyCopyHook(name+suffix, false)
716 }
717 f, e := unix.Openat(fd, name+suffix, unix.O_RDONLY|unix.O_NOFOLLOW|unix.O_NONBLOCK|unix.O_CLOEXEC, 0)
718 if errors.Is(e, unix.ENOENT) && suffix != "" {
719 digests = append(digests, name+suffix+":-")
720 continue
721 }
722 if e != nil {
723 clean()
724 return "", nil, nil, e
725 }
726 source := os.NewFile(uintptr(f), name+suffix)
727 var st unix.Stat_t
728 e = unix.Fstat(f, &st)
729 if e != nil || st.Mode&unix.S_IFMT != unix.S_IFREG || st.Nlink != 1 || st.Size > legacyJournalFileMax {
730 source.Close()
731 clean()
732 return "", nil, nil, ErrTypedJournal
733 }
734 target, e := os.OpenFile(path+suffix, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
735 if e != nil {
736 source.Close()
737 clean()
738 return "", nil, nil, e
739 }
740 h := sha256.New()
741 n, e := io.Copy(io.MultiWriter(target, h), io.LimitReader(source, legacyJournalFileMax+1))
742 syncErr := target.Sync()
743 closeErr := target.Close()
744 source.Close()
745 if e != nil || syncErr != nil || closeErr != nil || n != st.Size {
746 clean()
747 return "", nil, nil, fmt.Errorf("journal snapshot: %w", errors.Join(e, syncErr, closeErr, ErrChanged))
748 }
749 digests = append(digests, fmt.Sprintf("%s:%x", name+suffix, h.Sum(nil)))
750 if legacyCopyHook != nil {
751 legacyCopyHook(name+suffix, true)
752 }
753 }
754 return path, digests, clean, nil
755}
756
757func lockJournalMigration(ctx context.Context, path string) (func(), error) {
758 fd, err := unix.Open(path, unix.O_RDWR|unix.O_CREAT|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0600)
759 if err != nil {
760 return nil, err
761 }
762 var st unix.Stat_t
763 if err = unix.Fstat(fd, &st); err != nil || st.Mode&unix.S_IFMT != unix.S_IFREG || st.Nlink != 1 || int(st.Uid) != os.Geteuid() {
764 unix.Close(fd)
765 return nil, errors.Join(err, ErrTypedJournal)
766 }
767 for {
768 err = unix.Flock(fd, unix.LOCK_EX|unix.LOCK_NB)
769 if err == nil {
770 return func() { unix.Flock(fd, unix.LOCK_UN); unix.Close(fd) }, nil
771 }
772 if !errors.Is(err, unix.EWOULDBLOCK) {
773 unix.Close(fd)
774 return nil, err
775 }
776 select {
777 case <-ctx.Done():
778 unix.Close(fd)
779 return nil, ctx.Err()
780 case <-time.After(20 * time.Millisecond):
781 }
782 }
783}
784