diff --git a/CHANGELOG.md b/CHANGELOG.md index c367f3ae..4decf1d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,9 +3,10 @@ ## [Unreleased] ### Changes - Encryption at rest for KV blocks, S3 store (TC-887, off by default). The S3 block store seals new KV blocks under `v1/{space_tag}/…` names like the FileSystem store, with the same mode rules; a sealed range read fetches the header, then only the covering segments, pinned to the same object version (`If-Match`), and answers `503` if the object changes twice during the read. Orphan collection deletes both copies of a block and waits while keys are locked. Both block stores now declare the blocks seal path. Mode `off` is unchanged. -- Encryption at rest for KV blocks, FileSystem store (TC-887, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks on FileSystem block storage are sealed (TCE1) under `{blocks}/v1/{space_tag}/…` and served back as plaintext, ranges included; legacy blocks stay readable in `encrypt`, are refused in `strict`, and locked keys answer `503` for sealed data. A writing invocation prepares its spaces' data keys after its preupload authorization and before any upload or write lock, so a refused request creates no key. Orphan collection deletes both copies of a block and waits while keys are locked. The FileSystem store declares the blocks seal path; `encrypt`/`strict` still boot only once SQL artifacts declare theirs. Mode `off` is unchanged. See `docs/at-rest/INTERFACES.md` §7. +- Encryption at rest for KV blocks, FileSystem store (TC-887, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks on FileSystem block storage are sealed (TCE1) under `{blocks}/v1/{space_tag}/…` and served back as plaintext, ranges included; legacy blocks stay readable in `encrypt`, are refused in `strict`, and locked keys answer `503` for sealed data. A writing invocation prepares its spaces' data keys after its preupload authorization and before any upload or write lock, so a request refused by its authorization creates no key (a later precondition or replay refusal can leave the key the space's first write would create). Orphan collection deletes both copies of a block and waits while keys are locked. The FileSystem store declares the blocks seal path; `encrypt`/`strict` still boot only once SQL artifacts declare theirs. Mode `off` is unchanged. See `docs/at-rest/INTERFACES.md` §7. - Encryption at rest groundwork for KV blocks (TC-887, nothing is sealed yet): the S3 and FileSystem block stores count sealed objects (`v1/{space_tag}/…`) by plaintext size at boot, per space tag, and a space's quota joins them once the at-rest keys can name the tag. The S3 store makes one more `ListObjectsV2` listing at boot, under `v1/`. At-rest storage failures answer `503` (keys locked or key store unavailable) or `500` (integrity, unsupported format, legacy refused in `strict`) on `/invoke` and the KV GET routes. Sealed names carry the same `@N` generation suffix as legacy names (`docs/at-rest/INTERFACES.md` §5). - Deploys: the workflow supports an explicit, reason-recorded rollback to an earlier released dstack image on the unchanged N0 compose. It checks image labels, release tag, ancestry, and identical migration sources while preserving the normal forward-release guard. +- Encryption at rest for SQL and DuckDB artifacts: with `at_rest.mode` `encrypt` or `strict`, each `database_artifact` checkpoint and WAL delta is stored as TCE1 ciphertext bound to its service, space, name and content (and, for a delta, its revision). Each sealed save also stores a tag over the row's manifest (revision, which parts it holds, their plaintext hashes, sizes and encodings) under the space's key, and every read checks it first: a dropped or substituted part, or a changed revision, fails as an integrity error instead of reading as a smaller database. A sealed part must be exactly its sealed length. Migration `m20261107_010000_artifact_payload_encoding` adds the nullable `payload_encoding`, `delta_encoding` and `manifest` columns (`NULL` = plaintext and no tag, so existing rows read as before). Sizes and content hashes stay those of the plaintext, so storage usage and quotas are unchanged. Mode `off` (the default) stores exactly what it did before and clears the tag. A write that cannot be sealed, or a sealed read while the keys are locked, answers 503; a part or manifest that fails its integrity check, uses an unsupported format, or is plaintext under `strict` answers 500. Not covered: replacing a row with an older, self-consistent copy of itself, and local working files (SQL and DuckDB working and cache files, upload staging, exports), which rely on the CVM's encrypted storage; operators without a TEE must protect local paths, swap and file-level backups (`docs/at-rest/INTERFACES.md` §8). `tinycloud-sql-identity offline-fingerprint` refuses sealed artifacts. With KV blocks and SQL artifacts both sealing, the node accepts `encrypt` and `strict` once the at-rest gate passes. Details: TC-887 ## [1.25.17] - 2026-11-03 diff --git a/docs/at-rest/INTERFACES.md b/docs/at-rest/INTERFACES.md index 2ad86a09..c3ee46c0 100644 --- a/docs/at-rest/INTERFACES.md +++ b/docs/at-rest/INTERFACES.md @@ -1,9 +1,9 @@ # Encryption at rest: interfaces (TC-887) > **Status (TC-887):** the envelope, key types, `DataKeyProvider`, the keyring -> (`at_rest::keyring`), the node wiring (`docs/at-rest-escrow.md`) and the storage view -> (`at_rest::{AtRestMode, AtRestView, AtRestSource}`) are in. Sealed KV storage (§7) and sealed -> SQL artifacts follow; until both seal, the node refuses `encrypt` and `strict`. +> (`at_rest::keyring`), the node wiring (`docs/at-rest-escrow.md`), the storage view +> (`at_rest::{AtRestMode, AtRestView, AtRestSource}`), sealed KV storage (§7) and sealed SQL +> artifacts (§8) are in. Both declare their seal path, so `encrypt` and `strict` can be configured. Normative interface and format for `tinycloud_core::at_rest` and the node's sealed KV block storage. Plan: TC-887 (encryption at rest inside the TEE). Key continuity and escrow: @@ -109,7 +109,12 @@ pub enum BlobIdentity<'a> { // what the ciphertext is bound checkpoint_content_hash: &'a str, checkpoint_size_bytes: u64 }, SqlWal { service: &'a str, space: &'a str, name: &'a str, revision: i64, checkpoint_content_hash: &'a str, delta_content_hash: &'a str }, + SqlManifest { service: &'a str, space: &'a str, name: &'a str, revision: i64, + content_hash: &'a str, size_bytes: i64, + checkpoint: SqlManifestPart<'a>, delta: Option> }, } +pub struct SqlManifestPart<'a> { pub content_hash: &'a str, pub size_bytes: i64, + pub encoding: Option } // §8 /// Parsed, validated 64-byte header (unknown version/alg/flags/padding refused). pub struct Header { /* raw bytes, dek_id, plaintext_len */ } @@ -228,10 +233,13 @@ Header, 64 bytes, all integers big-endian: Context (identity) bytes, every variable field `u32 BE len ‖ bytes`: ``` ctx = lp("tinycloud/at-rest/v1") ‖ lp(purpose) ‖ lp(field_1) ‖ … ‖ lp(field_n) -purpose ∈ {"kv-block","sql-checkpoint","sql-wal"} +purpose ∈ {"kv-block","sql-checkpoint","sql-wal","sql-manifest"} kv-block: space, content_hash (multihash bytes) sql-checkpoint: service, space, name, checkpoint_content_hash, u64 BE checkpoint_size_bytes sql-wal: service, space, name, i64 BE revision, checkpoint_content_hash, delta_content_hash +sql-manifest: service, space, name, i64 BE revision, content_hash, i64 BE size_bytes, + part(checkpoint), then [0x00] (no delta) or [0x01], part(delta) +part(p): p.content_hash, i64 BE p.size_bytes, p.encoding (empty = NULL, else i16 BE) ``` Keys and nonces: ``` @@ -330,14 +338,22 @@ redacted. preupload transaction (TC-735) authorized it and checked its spaces are hosted, and before any upload, sequence or writer lock, or transaction: written spaces get their current data key (created if new), spaces a `kv/get` reads in the transaction get any data key the database lost - restored from its authenticated mirror (never created). A refused request therefore creates no - data key or mirror object; no separate pre-authorization is needed (the preupload transaction - already authorizes before any key work). Work under the writer lock (a `kv/get` in a writing - invocation, the upload of a protected legacy meeting path) runs in `at_rest::write_locked`, - where the provider only looks keys up and a key that would need creating or restoring is - `NotPrepared` (503), never a key-store write that would wait on the lock the request holds. - Read-only invocations and the KV GET routes hold no writer lock, so a key the database lost is - restored on first use there. + restored from its authenticated mirror (never created). A request refused by its authorization + or hosting checks therefore creates no data key or mirror object; no separate pre-authorization + is needed (the preupload transaction already authorizes before any key work). Work under the + writer lock (a `kv/get` in a writing invocation, the upload of a protected legacy meeting path) + runs in `at_rest::write_locked`, where the provider only looks keys up and a key that would + need creating or restoring is `NotPrepared` (503), never a key-store write that would wait on + the lock the request holds. Read-only invocations and the KV GET routes hold no writer lock, so + a key the database lost is restored on first use there. +- **Accepted residual (late refusals prepare keys):** a refusal that comes after the spaces were + prepared, in the write's own transaction (a failed `If-Match`/`If-None-Match` precondition, 412, + or a replayed invocation, 409), can leave a new data key and its mirror object for a space it + wrote nothing to. It is the key that space's first write would create; nothing is sealed under it. +- **Accepted residual (generation not in the AEAD identity):** a block's AAD binds its space and + content hash but not its `@N` generation (§5), so a sealed object of one generation opens in + place of another generation of the same block. Every generation holds the same plaintext (the + content hash), so only same-content replay is possible. - **Accepted residual (linkability during the transition):** in `encrypt`, a read of a legacy-only block asks for the sealed name first, then the legacy name (`strict` probes the legacy name to answer "legacy refused"). A storage provider's access logs can therefore link a @@ -354,3 +370,84 @@ redacted. stored bytes at boot like a stage in a space directory. - **Reserved names:** The key-continuity key mirror lives under `_tc/` in the block bucket (`datadir/at-rest` on FS). The block boot scan skips it. + +## 8. SQL artifacts +External authoritative payloads are sealed: the durable copy of each SQL and DuckDB database is a +`database_artifact` row, `payload` (the checkpoint) and `delta_payload` (the WAL delta). +`SeaOrmDatabaseArtifactRepository` seals and opens them; it reads a fresh `AtRestView` per +operation (`with_at_rest`), and the node's repository declares `SealPath::SqlArtifacts`. +- **Local working files are not sealed (known limitation, a later phase).** Local working files + (SQL and DuckDB working and cache files, upload staging, exports) rely on the CVM's encrypted + storage; operators without a TEE must protect the node's local paths, swap and file-level + backups themselves (for example, `node.sh`'s `data.tar.gz` is not encrypted). They are: KV + ingress staging (`TempFileSystemStage`, a `NamedTempFile` in the system temp directory, + `tinycloud-node-server/src/storage/file_system.rs`); the SQL and DuckDB working databases and + WAL files under `{datadir}/sql` and `{datadir}/duckdb`, and the hydration staging files a cold + read writes beside them (`CacheFileSink` in `database_artifacts.rs`, `sql/service.rs`, + `sql/database.rs`, `duckdb/database.rs`). +- **Columns:** `payload_encoding` and `delta_encoding` (SMALLINT, `NULL` = legacy plaintext; §5) + and `manifest` (BLOB/BYTEA, `NULL` = no tag), migration + `m20261107_010000_artifact_payload_encoding`. Each encoding is written with its part: a full + save sets `payload_encoding` and clears the delta with its `delta_encoding`; a delta save sets + `delta_encoding` only, so a legacy checkpoint can carry a sealed delta. Every save rewrites + `manifest`. +- **Plaintext metadata:** `size_bytes`, `checkpoint_size_bytes`, `delta_size_bytes` and the three + content hashes stay those of the plaintext database file. They are part of the identity (§4), + and quota (`SqlSizes`) and the SQL growth check read them unchanged. Accepted residual: as + before, anyone who can read the table sees each database's size and content hash. +- **Identity:** checkpoint `SqlCheckpoint { service, space, name, checkpoint_content_hash, + checkpoint_size_bytes }`; delta `SqlWal { service, space, name, revision, checkpoint_content_hash, + delta_content_hash }`, where `revision` is the row revision the delta was written at (always the + row's current revision). A part moved to another row, space, name, revision or checkpoint is + an integrity failure. +- **Manifest (what the row holds, authenticated):** the part identities alone don't say which + parts a row must have. A checkpoint opens whatever the revision and whether or not a delta + belongs with it, so someone who can write the table could drop a sealed delta (clear the delta + columns, reset the unkeyed `content_hash` and `size_bytes`) and a cold read would serve the + checkpoint alone, silently losing committed rows. So every sealed save also stores a tag in + `manifest`: a TCE1 seal of an empty plaintext under the space's current data key with + `SqlManifest { service, space, name, revision, content_hash, size_bytes, checkpoint: + (checkpoint_content_hash, checkpoint_size_bytes, payload_encoding), delta: None | Some( + (delta_content_hash, delta_size_bytes, delta_encoding)) }` (§4). That is a keyed MAC over the + row's manifest, 80 bytes, naming its data key in its header like any sealed part, so it survives + key rotation. A full save writes it with `delta: None`, a delta save with the delta; both + rewrite it in the same `UPDATE` as the parts. Every read (`read_chunks`, and `open_row` for the + offline tool) checks it before any part reaches the sink: a dropped or added delta, a part or + metadata from another revision, a changed revision, or a changed encoding fails as integrity + (`AtRestIntegrity { part: "manifest" }`, 500), never a partial read. Unknown or invalid + encoding columns keep their own errors. Mode `off` writes no tag and clears it (its writes stay + plaintext and need no key), so a row without a tag is accepted only if it has no sealed part, or + if its only sealed part is a checkpoint under a delta mode `off` wrote; a row with any other + sealed part and no tag fails as integrity. + **Not protected:** a whole-row rollback, i.e. replacing the row with an older, valid, + self-consistent snapshot of itself (parts, metadata and tag all from the same earlier + revision), opens. So do rows with no tag: legacy parts are unauthenticated, as in §5, including + a delta written in mode `off` over a sealed checkpoint (`strict` refuses that legacy delta). +- **Stored length and reads:** a TCE1 part stores exactly `ciphertext_len(size)` bytes. The + streamed read slices the stored bytes (`substr`, 4 MiB) and opens them one 64 KiB segment at a + time (`SegmentOpener`); the last slice of a sealed part asks for one byte past that length, so + bytes appended to the part, or a part cut short, fail as integrity. The content hashes are + checked over the plaintext, and sinks receive plaintext, so the cache sink still checks the + SQLite/DuckDB file header. At most one segment is buffered beyond what the sink keeps. +- **Writes:** mode `off` stores plaintext exactly as before. Any other mode seals with + `provider.current_key(space)` or fails (`AtRestUnavailable`), never plaintext. `current_key` + runs before the SQLite writer lock, since it may create the space's key; sealing runs in + `spawn_blocking`. +- **Errors:** keys locked, missing or the encrypt gate closed → 503 (`SqlError::Unavailable`, + `DuckDbError::Unavailable`). `UnknownKey`, a malformed header, a header length that disagrees + with the row, a stored length that disagrees with it, a manifest tag that doesn't match the row + (or is missing where one is required), and any AEAD failure → 500 integrity; `Unsupported` + (header or column `2..=9`) → 500 "upgrade the node"; a legacy part in `strict` → 500. A sealed + part whose column says `NULL` fails the manifest check; it is never served. +- **Accepted residual (legacy injection in `encrypt`):** someone who can write the table can + replace a row with legacy plaintext parts and no tag (any content, with matching hashes), and + `encrypt` serves it, as it serves any legacy row. `strict` (after the backfill) refuses it. +- **Offline tool:** `tinycloud-sql-identity offline-fingerprint` has no keys and refuses sealed + rows (`database_migration::offline_inventory` takes the provider). `open_row` checks the + manifest tag first, like a streamed read. +- **Rollback:** a build without this change ignores the columns. It cannot hydrate a sealed part + (the file-header check fails), and a save it makes over a sealed row without reading it + (DuckDB `import_db`) leaves `payload_encoding = 1` on plaintext and a `manifest` tag for the + previous contents, which this build then refuses as integrity. Roll back below this change only + with no sealed rows; a row an older build overwrote that way needs its encoding columns and + `manifest` set back to `NULL`. diff --git a/tinycloud-core/src/at_rest/envelope.rs b/tinycloud-core/src/at_rest/envelope.rs index 24b041ca..0667432b 100644 --- a/tinycloud-core/src/at_rest/envelope.rs +++ b/tinycloud-core/src/at_rest/envelope.rs @@ -89,6 +89,41 @@ pub enum BlobIdentity<'a> { checkpoint_content_hash: &'a str, delta_content_hash: &'a str, }, + /// A SQL artifact row's manifest: what the row says it holds. Sealed over + /// an empty plaintext, so the blob is a key-bound tag over these fields + /// (`docs/at-rest/INTERFACES.md` §8). + SqlManifest { + service: &'a str, + space: &'a str, + name: &'a str, + revision: i64, + content_hash: &'a str, + size_bytes: i64, + checkpoint: SqlManifestPart<'a>, + delta: Option>, + }, +} + +/// One stored part of a SQL artifact as its manifest records it. +#[derive(Clone, Copy, Debug)] +pub struct SqlManifestPart<'a> { + /// Of the plaintext. + pub content_hash: &'a str, + /// Of the plaintext. + pub size_bytes: i64, + /// The part's encoding column, `None` = legacy plaintext. + pub encoding: Option, +} + +impl SqlManifestPart<'_> { + fn push(&self, out: &mut Vec) { + push_lp(out, self.content_hash.as_bytes()); + push_lp(out, &self.size_bytes.to_be_bytes()); + match self.encoding { + None => push_lp(out, &[]), + Some(encoding) => push_lp(out, &encoding.to_be_bytes()), + } + } } pub(crate) fn push_lp(out: &mut Vec, field: &[u8]) { @@ -140,6 +175,32 @@ impl BlobIdentity<'_> { push_lp(&mut c, checkpoint_content_hash.as_bytes()); push_lp(&mut c, delta_content_hash.as_bytes()); } + Self::SqlManifest { + service, + space, + name, + revision, + content_hash, + size_bytes, + checkpoint, + delta, + } => { + push_lp(&mut c, b"sql-manifest"); + push_lp(&mut c, service.as_bytes()); + push_lp(&mut c, space.as_bytes()); + push_lp(&mut c, name.as_bytes()); + push_lp(&mut c, &revision.to_be_bytes()); + push_lp(&mut c, content_hash.as_bytes()); + push_lp(&mut c, &size_bytes.to_be_bytes()); + checkpoint.push(&mut c); + match delta { + None => push_lp(&mut c, &[0]), + Some(delta) => { + push_lp(&mut c, &[1]); + delta.push(&mut c); + } + } + } } c } diff --git a/tinycloud-core/src/at_rest/keyring/tests.rs b/tinycloud-core/src/at_rest/keyring/tests.rs index 7e495d02..8e208247 100644 --- a/tinycloud-core/src/at_rest/keyring/tests.rs +++ b/tinycloud-core/src/at_rest/keyring/tests.rs @@ -1669,22 +1669,29 @@ async fn a_pinned_escrow_is_never_re_split() { ); } +/// Rolls back from the newest applied migration down to and including the +/// one that creates the key tables, whatever was added after it. +async fn down_through_key_tables(db: &DatabaseConnection) -> Result<(), sea_orm::DbErr> { + let applied = crate::migrations::Migrator::get_applied_migrations(db).await?; + let at = applied + .iter() + .position(|m| m.name() == "m20261107_000000_at_rest_keys") + .expect("the key tables' migration is applied"); + crate::migrations::Migrator::down(db, Some((applied.len() - at) as u32)).await +} + #[tokio::test] async fn migration_down_refuses_to_drop_keys() { let (db, mirror, h) = (db().await, Arc::new(MemoryMirror::default()), holders()); open(&db, &mirror, &ROOT, config(&h, None)).await.unwrap(); - assert!(crate::migrations::Migrator::down(&db, Some(1)) - .await - .is_err()); + assert!(down_through_key_tables(&db).await.is_err()); assert_eq!(at_rest_key::Entity::find().all(&db).await.unwrap().len(), 1); at_rest_escrow::Entity::delete_many() .exec(&db) .await .unwrap(); at_rest_key::Entity::delete_many().exec(&db).await.unwrap(); - crate::migrations::Migrator::down(&db, Some(1)) - .await - .unwrap(); + down_through_key_tables(&db).await.unwrap(); assert!(at_rest_key::Entity::find().all(&db).await.is_err()); } diff --git a/tinycloud-core/src/at_rest/mod.rs b/tinycloud-core/src/at_rest/mod.rs index 4487642b..6647bcdd 100644 --- a/tinycloud-core/src/at_rest/mod.rs +++ b/tinycloud-core/src/at_rest/mod.rs @@ -17,14 +17,14 @@ mod view; pub use envelope::{ ciphertext_len, format_version, open, plaintext_len, seal, BlobIdentity, EnvelopeError, Header, - HEADER_LEN, SEGMENT_LEN, TAG_LEN, + SqlManifestPart, HEADER_LEN, SEGMENT_LEN, TAG_LEN, }; pub use key::{DataKey, DekId, ObjectNameKey}; pub use names::{sealed_object_name, space_tag, PayloadEncoding, SEALED_NAME_PREFIX}; pub use provider::{in_write_lock, write_locked, DataKeyError, DataKeyProvider}; -pub use stream::{DecryptReader, EncryptReader, SegmentRange}; +pub use stream::{DecryptReader, EncryptReader, SegmentOpener, SegmentRange}; #[cfg(any(test, feature = "test-util"))] -pub use testing::FixedKeyProvider; +pub use testing::{FixedKeyProvider, TestAtRestSource}; pub use keyring::{AtRestKeyring, KeyringConfig, KeyringState}; pub use store::{AtRestStore, KeyMirror}; diff --git a/tinycloud-core/src/at_rest/stream.rs b/tinycloud-core/src/at_rest/stream.rs index 4bf120eb..336ada3d 100644 --- a/tinycloud-core/src/at_rest/stream.rs +++ b/tinycloud-core/src/at_rest/stream.rs @@ -376,3 +376,86 @@ impl AsyncRead for DecryptReader { r } } + +/// Push-style counterpart of [`DecryptReader`], for callers that receive the +/// ciphertext in slices they do not choose (SQL artifacts arrive in +/// `substr` slices). Holds at most one segment. Like the reader, it only +/// hands out a segment after it authenticates, and fails for good on the +/// first error. +pub struct SegmentOpener { + cipher: SegmentCipher, + plaintext_len: u64, + next: u64, + segments: u64, + seg: Vec, + /// `seg` holds a segment already handed out; the next push starts anew. + emitted: bool, + failed: bool, +} + +impl SegmentOpener { + /// The bytes pushed next start just after `header`. + pub fn new( + key: &DataKey, + id: &BlobIdentity<'_>, + header: &Header, + ) -> Result { + if header.dek_id() != key.id() { + return Err(EnvelopeError::KeyMismatch); + } + Ok(Self { + cipher: SegmentCipher::new(key, id, header), + plaintext_len: header.plaintext_len(), + next: 0, + segments: header.segment_count(), + seg: Vec::with_capacity(SEGMENT_LEN + TAG_LEN), + emitted: false, + failed: false, + }) + } + + /// Takes bytes from the front of `input` until a segment is complete and + /// returns its authenticated plaintext, or `None` once `input` is used up + /// with the segment still incomplete. Call again while it returns + /// `Some`. Bytes after the last segment are an integrity failure. + pub fn push(&mut self, input: &mut &[u8]) -> Result, EnvelopeError> { + if self.failed { + return Err(EnvelopeError::Integrity); + } + if self.emitted { + self.seg.clear(); + self.emitted = false; + } + if self.next == self.segments { + if input.is_empty() { + return Ok(None); + } + self.failed = true; + return Err(EnvelopeError::Integrity); + } + let len = segment_plain_len(self.plaintext_len, self.next); + let take = (len + TAG_LEN - self.seg.len()).min(input.len()); + self.seg.extend_from_slice(&input[..take]); + *input = &input[take..]; + if self.seg.len() < len + TAG_LEN { + return Ok(None); + } + let (body, tag) = self.seg.split_at_mut(len); + if let Err(error) = self.cipher.open_segment(self.next, body, tag) { + self.failed = true; + return Err(error); + } + self.next += 1; + self.emitted = true; + Ok(Some(&self.seg[..len])) + } + + /// `Ok` only once every segment has authenticated: a blob that stopped + /// early (truncated) is an integrity failure. + pub fn finish(&self) -> Result<(), EnvelopeError> { + if self.failed || self.next != self.segments { + return Err(EnvelopeError::Integrity); + } + Ok(()) + } +} diff --git a/tinycloud-core/src/at_rest/testdata/gen_vectors.py b/tinycloud-core/src/at_rest/testdata/gen_vectors.py index 1e74a68f..280a3d3a 100755 --- a/tinycloud-core/src/at_rest/testdata/gen_vectors.py +++ b/tinycloud-core/src/at_rest/testdata/gen_vectors.py @@ -37,11 +37,28 @@ def context(ident: dict) -> bytes: c += lp(struct.pack(">q", ident["revision"])) for f in ("checkpoint_content_hash", "delta_content_hash"): c += lp(ident[f].encode()) + elif p == "sql-manifest": + for f in ("service", "space", "name"): + c += lp(ident[f].encode()) + c += lp(struct.pack(">q", ident["revision"])) + c += lp(ident["content_hash"].encode()) + c += lp(struct.pack(">q", ident["size_bytes"])) + c += manifest_part(ident["checkpoint"]) + if ident["delta"] is None: + c += lp(b"\x00") + else: + c += lp(b"\x01") + manifest_part(ident["delta"]) else: raise ValueError(p) return c +def manifest_part(part: dict) -> bytes: + enc = b"" if part["encoding"] is None else struct.pack(">h", part["encoding"]) + return (lp(part["content_hash"].encode()) + lp(struct.pack(">q", part["size_bytes"])) + + lp(enc)) + + def header(dek_id: int, plen: int, salt: bytes) -> bytes: h = b"TCE1" + bytes([1, 1, 0, 0]) + struct.pack(">IQ", dek_id, plen) + salt + bytes([16]) h += bytes(64 - len(h)) @@ -85,6 +102,17 @@ def blake3_multihash(digest_hex: str) -> str: WAL = {"purpose": "sql-wal", "service": "sqlite", "space": SPACE, "name": "notes", "revision": 42, "checkpoint_content_hash": "bafkreiexamplecheckpoint", "delta_content_hash": "bafkreiexampledelta"} +MANIFEST = {"purpose": "sql-manifest", "service": "sqlite", "space": SPACE, "name": "notes", + "revision": 42, "content_hash": "bafkreiexamplelogical", "size_bytes": 200015, + "checkpoint": {"content_hash": "bafkreiexamplecheckpoint", "size_bytes": 200000, + "encoding": 1}, + "delta": {"content_hash": "bafkreiexampledelta", "size_bytes": 15, "encoding": 1}} +MANIFEST_NO_DELTA = {"purpose": "sql-manifest", "service": "sqlite", "space": SPACE, + "name": "notes", "revision": 41, + "content_hash": "bafkreiexamplecheckpoint", "size_bytes": 200000, + "checkpoint": {"content_hash": "bafkreiexamplecheckpoint", + "size_bytes": 200000, "encoding": None}, + "delta": None} CASES = [ ("kv-empty", 1, KV, {"hex": ""}), @@ -95,6 +123,8 @@ def blake3_multihash(digest_hex: str) -> str: ("kv-segment-plus-one", 1, KV, {"pattern_len": SEG + 1}), ("sql-checkpoint-multi-segment", 3, CKPT, {"pattern_len": 200000}), ("sql-wal-short", 0xFFFFFFFF, WAL, {"hex": b"wal frame bytes".hex()}), + ("sql-manifest-with-delta", 3, MANIFEST, {"hex": ""}), + ("sql-manifest-no-delta", 3, MANIFEST_NO_DELTA, {"hex": ""}), ] MAX_PLAIN = (1 << 32) * SEG # 2^48: the largest plaintext TCE1 can hold U64_MAX = (1 << 64) - 1 diff --git a/tinycloud-core/src/at_rest/testdata/vectors.json b/tinycloud-core/src/at_rest/testdata/vectors.json index c5a22211..20cc5817 100644 --- a/tinycloud-core/src/at_rest/testdata/vectors.json +++ b/tinycloud-core/src/at_rest/testdata/vectors.json @@ -144,6 +144,64 @@ "ciphertext_len": 95, "ciphertext_sha256": "af0a95ae37ccd587377d67752de78e73c6828a33e16eb007ae5786d5005bc572", "ciphertext": "5443453101010000ffffffff000000000000000f404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f100000000000000000000000db56826432a30e8d1a00143f51d2ae1665783c299e727a22eeb3d49167eb78" + }, + { + "name": "sql-manifest-with-delta", + "dek": "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f", + "dek_id": 3, + "salt": "404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f", + "identity": { + "purpose": "sql-manifest", + "service": "sqlite", + "space": "tinycloud:pkh:eip155:1:0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266:default", + "name": "notes", + "revision": 42, + "content_hash": "bafkreiexamplelogical", + "size_bytes": 200015, + "checkpoint": { + "content_hash": "bafkreiexamplecheckpoint", + "size_bytes": 200000, + "encoding": 1 + }, + "delta": { + "content_hash": "bafkreiexampledelta", + "size_bytes": 15, + "encoding": 1 + } + }, + "plaintext": { + "hex": "" + }, + "ciphertext_len": 80, + "ciphertext_sha256": "d77adb435f71f7b87704741fb48f983cbe28b917f3804d306ed5de7f5d65b79d", + "ciphertext": "5443453101010000000000030000000000000000404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f1000000000000000000000002ae1a3d4c56ca5101f2eff282f84acde" + }, + { + "name": "sql-manifest-no-delta", + "dek": "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f", + "dek_id": 3, + "salt": "404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f", + "identity": { + "purpose": "sql-manifest", + "service": "sqlite", + "space": "tinycloud:pkh:eip155:1:0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266:default", + "name": "notes", + "revision": 41, + "content_hash": "bafkreiexamplecheckpoint", + "size_bytes": 200000, + "checkpoint": { + "content_hash": "bafkreiexamplecheckpoint", + "size_bytes": 200000, + "encoding": null + }, + "delta": null + }, + "plaintext": { + "hex": "" + }, + "ciphertext_len": 80, + "ciphertext_sha256": "784d9c44867981b528e5d0756bebc51d60382f31ee17775e28d1e9fec6b05b50", + "ciphertext": "5443453101010000000000030000000000000000404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f100000000000000000000000bff95f7c1fe0275697e77f181a65b3d4" } ], "object_names": [ diff --git a/tinycloud-core/src/at_rest/testing.rs b/tinycloud-core/src/at_rest/testing.rs index f023ad24..5b20d8cc 100644 --- a/tinycloud-core/src/at_rest/testing.rs +++ b/tinycloud-core/src/at_rest/testing.rs @@ -1,11 +1,49 @@ -//! Test-only key provider. Compiled only for this crate's tests or with the +//! Test-only key provider and at-rest source. Compiled only for this crate's tests or with the //! `test-util` feature, which downstream crates enable on their //! `tinycloud-core` dev-dependency only. use super::key::{DataKey, DekId, ObjectNameKey}; use super::provider::{DataKeyError, DataKeyProvider}; +use super::view::{AtRestMode, AtRestSource, AtRestView}; use async_trait::async_trait; use sha2::{Digest, Sha256}; +use std::sync::{Arc, Mutex}; + +/// Test-only [`AtRestSource`]: a view the test sets, and may change between +/// operations the way a node's mode, gate and keys do. +pub struct TestAtRestSource(Mutex); + +impl TestAtRestSource { + pub fn new( + mode: AtRestMode, + encrypting: bool, + provider: Option>, + ) -> Arc { + Arc::new(Self(Mutex::new(AtRestView { + mode, + encrypting, + provider, + }))) + } + + /// Switch the mode and gate; the keys stay. + pub fn set(&self, mode: AtRestMode, encrypting: bool) { + let mut view = self.0.lock().expect("not poisoned"); + view.mode = mode; + view.encrypting = encrypting; + } + + /// Swap the keys, e.g. for a locked provider. + pub fn set_provider(&self, provider: Option>) { + self.0.lock().expect("not poisoned").provider = provider; + } +} + +impl AtRestSource for TestAtRestSource { + fn view(&self) -> AtRestView { + self.0.lock().expect("not poisoned").clone() + } +} /// Deterministic in-memory provider for tests. Not for production: every key /// is derived from a public seed. diff --git a/tinycloud-core/src/at_rest/tests.rs b/tinycloud-core/src/at_rest/tests.rs index c63079d3..18689221 100644 --- a/tinycloud-core/src/at_rest/tests.rs +++ b/tinycloud-core/src/at_rest/tests.rs @@ -203,6 +203,25 @@ fn with_identity(i: &serde_json::Value, f: impl FnOnce(&BlobIdentity<'_>) -> checkpoint_content_hash: s(i, "checkpoint_content_hash"), delta_content_hash: s(i, "delta_content_hash"), }, + "sql-manifest" => { + fn part(p: &serde_json::Value) -> SqlManifestPart<'_> { + SqlManifestPart { + content_hash: p["content_hash"].as_str().unwrap(), + size_bytes: p["size_bytes"].as_i64().unwrap(), + encoding: p["encoding"].as_i64().map(|e| i16::try_from(e).unwrap()), + } + } + BlobIdentity::SqlManifest { + service: s(i, "service"), + space: s(i, "space"), + name: s(i, "name"), + revision: i["revision"].as_i64().unwrap(), + content_hash: s(i, "content_hash"), + size_bytes: i["size_bytes"].as_i64().unwrap(), + checkpoint: part(&i["checkpoint"]), + delta: (!i["delta"].is_null()).then(|| part(&i["delta"])), + } + } p => panic!("unknown purpose {p}"), }; f(&id) @@ -257,7 +276,7 @@ fn s<'a>(v: &'a serde_json::Value, k: &str) -> &'a str { #[test] fn known_answer_vectors_match_independent_reference() { let v = vectors(); - assert_eq!(v.vectors.len(), 8); + assert_eq!(v.vectors.len(), 10); for t in &v.vectors { let k = DataKey::new(DekId::new(t.dek_id), arr32(&t.dek)); let pt = vector_plaintext(t); @@ -455,6 +474,25 @@ proptest! { prop_assert_eq!(read_all(dec, &bufs).unwrap(), pt); } + #[test] + fn pushed_slices_open_like_the_whole_buffer( + len in edge_len(), + seed: u64, + sizes in prop::collection::vec(1usize..70_000, 1..6), + ) { + let k = key(2, 5); + let h = hash(b"push"); + let pt = bytes(len, seed); + let ct = seal(&k, &kv(&h), pt.clone()).unwrap(); + prop_assert_eq!(push_open(&k, &kv(&h), &ct, &sizes).unwrap(), pt); + // Cut short anywhere, or one byte too long, it never opens. + let cut = (seed as usize) % ct.len(); + prop_assert!(push_open(&k, &kv(&h), &ct[..cut], &sizes).is_err()); + let mut long = ct.clone(); + long.push(0); + prop_assert!(push_open(&k, &kv(&h), &long, &sizes).is_err()); + } + #[test] fn length_inverse(p in 0u64..(1u64 << 44), c in 0u64..(1u64 << 44)) { prop_assert_eq!(plaintext_len(ciphertext_len(p).unwrap()), Some(p)); @@ -507,6 +545,32 @@ proptest! { } } +/// Opens `ct` through [`SegmentOpener`], pushing slices of the cycled +/// `sizes`. +fn push_open( + k: &DataKey, + id: &BlobIdentity<'_>, + ct: &[u8], + sizes: &[usize], +) -> Result, EnvelopeError> { + let header = Header::parse(ct)?; + let mut opener = SegmentOpener::new(k, id, &header)?; + let mut out = Vec::new(); + let mut rest = &ct[HEADER_LEN..]; + for size in sizes.iter().cycle() { + if rest.is_empty() { + break; + } + let (mut slice, tail) = rest.split_at((*size).min(rest.len())); + rest = tail; + while let Some(plain) = opener.push(&mut slice)? { + out.extend_from_slice(plain); + } + } + opener.finish()?; + Ok(out) +} + #[test] fn salt_is_fresh_per_seal() { let k = key(1, 1); @@ -687,6 +751,89 @@ fn ciphertext_is_bound_to_its_identity() { } } +/// A SQL manifest tag opens only under the exact manifest it was sealed for: +/// every field, the presence of the delta, and each part's encoding count. +#[test] +fn a_sql_manifest_is_bound_to_every_field() { + let k = key(1, 6); + let part = |hash, size, encoding| SqlManifestPart { + content_hash: hash, + size_bytes: size, + encoding, + }; + let manifest = |revision, size_bytes, checkpoint, delta| BlobIdentity::SqlManifest { + service: "sql", + space: SPACE, + name: "db", + revision, + content_hash: "logical", + size_bytes, + checkpoint, + delta, + }; + let ckpt = part("c1", 10, Some(1)); + let delta = part("d1", 5, Some(1)); + let good = manifest(2, 15, ckpt, Some(delta)); + let bad = [ + manifest(3, 15, ckpt, Some(delta)), + manifest(2, 16, ckpt, Some(delta)), + manifest(2, 15, ckpt, None), + manifest(2, 15, part("c2", 10, Some(1)), Some(delta)), + manifest(2, 15, part("c1", 11, Some(1)), Some(delta)), + manifest(2, 15, part("c1", 10, None), Some(delta)), + manifest(2, 15, ckpt, Some(part("d2", 5, Some(1)))), + manifest(2, 15, ckpt, Some(part("d1", 6, Some(1)))), + manifest(2, 15, ckpt, Some(part("d1", 5, None))), + BlobIdentity::SqlManifest { + service: "sql", + space: SPACE, + name: "db2", + revision: 2, + content_hash: "logical", + size_bytes: 15, + checkpoint: ckpt, + delta: Some(delta), + }, + BlobIdentity::SqlManifest { + service: "duckdb", + space: SPACE, + name: "db", + revision: 2, + content_hash: "logical", + size_bytes: 15, + checkpoint: ckpt, + delta: Some(delta), + }, + BlobIdentity::SqlManifest { + service: "sql", + space: SPACE, + name: "db", + revision: 2, + content_hash: "other", + size_bytes: 15, + checkpoint: ckpt, + delta: Some(delta), + }, + BlobIdentity::SqlWal { + service: "sql", + space: SPACE, + name: "db", + revision: 2, + checkpoint_content_hash: "c1", + delta_content_hash: "d1", + }, + ]; + let tag = seal(&k, &good, Vec::new()).unwrap(); + assert_eq!(tag.len() as u64, ciphertext_len(0).unwrap()); + assert_eq!(open(&k, &good, tag.clone()).unwrap(), Vec::::new()); + for b in bad { + assert!( + matches!(open(&k, &b, tag.clone()), Err(EnvelopeError::Integrity)), + "{good:?} opened as {b:?}" + ); + } +} + #[test] fn wrong_key_is_rejected() { let h = hash(b"k"); diff --git a/tinycloud-core/src/database_artifacts.rs b/tinycloud-core/src/database_artifacts.rs index a35df20d..4721f011 100644 --- a/tinycloud-core/src/database_artifacts.rs +++ b/tinycloud-core/src/database_artifacts.rs @@ -1,3 +1,5 @@ +use std::sync::Arc; + use async_trait::async_trait; use sea_orm::{ sea_query::{Expr, Func}, @@ -8,6 +10,11 @@ use sea_orm::{ use time::{format_description::well_known::Rfc3339, OffsetDateTime}; use crate::{ + at_rest::{ + self, AtRestMode, AtRestSource, AtRestView, BlobIdentity, DataKey, DataKeyError, + DataKeyProvider, EnvelopeError, Header, PayloadEncoding, SegmentOpener, SqlManifestPart, + HEADER_LEN, + }, db_pool::BulkScope, hash::{hash, Blake3Hasher}, models::database_artifact, @@ -129,6 +136,75 @@ pub enum DatabaseArtifactError { /// the bytes; the message is the sink's own. #[error("{0}")] Sink(String), + /// The at-rest keys a sealed write or read needs are not available now: + /// locked, the encrypt gate closed, or the key store failed. Nothing was + /// written and nothing was read as plaintext. Callers answer 503. + #[error("database artifact at-rest keys are unavailable")] + AtRestUnavailable, + /// A sealed part did not open: tampered, truncated, moved from another + /// artifact, revision or space, or naming a key its space never had. + /// Never read as legacy plaintext. Callers answer 500. + #[error("database artifact {part} failed its at-rest integrity check")] + AtRestIntegrity { part: &'static str }, + /// A part stored in a later at-rest format than this build reads. The + /// marker is unauthenticated, so this fails closed exactly like an + /// integrity failure. Callers answer 500. + #[error("database artifact {part} uses unsupported at-rest format version {version}; upgrade the node")] + AtRestUnsupported { part: &'static str, version: u8 }, + /// A legacy plaintext part read in `strict` mode. Callers answer 500. + #[error("database artifact {part} is legacy plaintext, which strict at-rest mode refuses")] + LegacyPlaintextRefused { part: &'static str }, +} + +/// A key-store failure while reading a sealed part. A key the space never +/// had is an integrity failure of the part; anything else means the keys +/// are unavailable now. +fn read_key_error(part: ArtifactPart, error: DataKeyError) -> DatabaseArtifactError { + key_error_at(part.as_str(), error) +} + +/// [`read_key_error`] for a sealed item named `role`. +fn key_error_at(role: &'static str, error: DataKeyError) -> DatabaseArtifactError { + match error { + DataKeyError::UnknownKey { .. } => { + tracing::error!( + role, + "Sealed database artifact names a data key its space does not have" + ); + DatabaseArtifactError::AtRestIntegrity { part: role } + } + error => { + tracing::warn!(role, %error, "At-rest keys unavailable for a sealed database artifact"); + DatabaseArtifactError::AtRestUnavailable + } + } +} + +/// An envelope failure on a sealed part: `Unsupported` keeps its operator +/// message, everything else is an integrity failure. Both fail closed. +fn envelope_error(part: ArtifactPart, error: EnvelopeError) -> DatabaseArtifactError { + envelope_error_at(part.as_str(), error) +} + +/// [`envelope_error`] for a sealed item named `role`. +fn envelope_error_at(role: &'static str, error: EnvelopeError) -> DatabaseArtifactError { + match error { + EnvelopeError::Unsupported { version } => { + tracing::error!( + role, + version, + "Database artifact uses an unsupported at-rest format; upgrade the node" + ); + DatabaseArtifactError::AtRestUnsupported { + part: role, + version, + } + } + error => { + tracing::error!(role, %error, "Database artifact failed its at-rest integrity check"); + DatabaseArtifactError::AtRestIntegrity { part: role } + } + } } /// The stored metadata of one artifact row: everything but the payload bytes. @@ -138,11 +214,12 @@ pub enum DatabaseArtifactError { /// or a size comparison needs, and what [`DatabaseArtifactRepository::read_chunks`] /// pins a streamed read to. /// -/// Shape note for at-rest encoding (TC-887): sizes and hashes here describe the -/// STORED bytes, which today are the plain database file. An encoding column -/// (`payload_encoding`) would be added to this struct, and chunking would keep -/// applying to the stored (encoded) bytes, so decoding belongs in the sink. -#[derive(Debug, Clone, PartialEq, Eq)] +/// At rest (TC-887): sizes and hashes here describe the PLAINTEXT database +/// file whatever the encoding, so quota and growth checks read them as +/// before; they are also part of what a sealed part is bound to. A TCE1 part +/// stores `at_rest::ciphertext_len(size)` bytes, which the streamed read +/// slices and decodes before hashing. +#[derive(Debug, Clone, PartialEq, Eq, sea_orm::FromQueryResult)] pub struct ArtifactMeta { pub revision: i64, pub size_bytes: i64, @@ -154,6 +231,13 @@ pub struct ArtifactMeta { pub updated_at: String, pub storage_mode: String, pub backend: String, + /// How `payload` is stored (`at_rest::PayloadEncoding::from_column`). + pub payload_encoding: Option, + /// How `delta_payload` is stored. + pub delta_encoding: Option, + /// The row's manifest tag (80 bytes when present): checked before any + /// part is read (`docs/at-rest/INTERFACES.md` §8). + pub manifest: Option>, } impl ArtifactMeta { @@ -616,9 +700,10 @@ pub trait DatabaseArtifactRepository: Send + Sync { /// the caller must discard what the sink received. Prefer [`read_artifact`], /// which also handles the restart. /// - /// TC-887: chunking applies to the STORED bytes; an encoded payload would - /// be decoded by the sink (or a decoding wrapper around it), and the - /// hashes verified here would be the stored-bytes hashes. + /// TC-887: chunking applies to the STORED bytes. A TCE1 part is decoded + /// as it streams, one 64 KiB segment at a time, so the sink receives and + /// the hashes cover the plaintext database file. Whether a part is + /// sealed comes from its encoding column, never from its bytes. async fn read_chunks( &self, service: &str, @@ -678,6 +763,9 @@ struct CurrentArtifact { pub struct SeaOrmDatabaseArtifactRepository { conn: crate::db_pool::RecoveringConnection, sqlite_writer_lock: Option>>, + /// The node's at-rest state, read per operation; `None` is mode `off` + /// with no keys. + at_rest: Option>, /// Test-only rendezvous seam (see [`wait_at_race_barrier`]) that lets two /// writers read the same base revision before either commits, so the /// full-checkpoint CAS conflict path can be exercised deterministically. @@ -690,11 +778,52 @@ impl SeaOrmDatabaseArtifactRepository { Self { conn: conn.into(), sqlite_writer_lock: None, + at_rest: None, #[cfg(test)] race_barrier: None, } } + /// Seal and open parts by the node's at-rest state (TC-887). Mode `off` + /// stores plaintext exactly as without it; any other mode seals every + /// part it writes or fails the write, never falling back to plaintext. + pub fn with_at_rest(mut self, source: Arc) -> Self { + self.at_rest = Some(source); + self + } + + fn at_rest_view(&self) -> AtRestView { + self.at_rest + .as_ref() + .map_or_else(AtRestView::off, |source| source.view()) + } + + /// The key a write in `space` seals with, or `None` in mode `off`. + /// + /// Runs before the SQLite writer lock: `current_key` may create the + /// space's data key, which writes the key store, and the key store + /// takes that lock on SQLite. + async fn write_key(&self, space: &str) -> Result, DatabaseArtifactError> { + let view = self.at_rest_view(); + if view.mode == AtRestMode::Off { + return Ok(None); + } + let provider = match view.provider { + Some(provider) if view.encrypting => provider, + _ => { + tracing::warn!("Database artifact write refused: at-rest encryption is not on"); + return Err(DatabaseArtifactError::AtRestUnavailable); + } + }; + match provider.current_key(space).await { + Ok(key) => Ok(Some(key)), + Err(error) => { + tracing::warn!(%error, "Database artifact write refused: no at-rest data key"); + Err(DatabaseArtifactError::AtRestUnavailable) + } + } + } + /// Share the node's SQLite gate: artifact writes use the capability DB and /// must not advance its WAL beneath a delegation transaction's snapshot. pub fn with_sqlite_writer_lock( @@ -726,44 +855,529 @@ impl SeaOrmDatabaseArtifactRepository { } } +#[derive(Clone, Copy)] struct ArtifactKey<'a> { service: &'a str, space: &'a str, name: &'a str, } +/// What one stored part's TCE1 ciphertext is bound to +/// (`docs/at-rest/INTERFACES.md` §4): the row's key and the part's plaintext +/// hash and size, and for a WAL the revision that wrote it and the checkpoint +/// it applies to. A part moved to another row, revision or checkpoint does +/// not open. Owned, so sealing can run off the async runtime. +enum PartBinding { + Checkpoint { + service: String, + space: String, + name: String, + content_hash: String, + size_bytes: u64, + }, + Wal { + service: String, + space: String, + name: String, + revision: i64, + checkpoint_content_hash: String, + delta_content_hash: String, + }, +} + +impl PartBinding { + fn checkpoint(key: &ArtifactKey<'_>, content_hash: &str, size_bytes: i64) -> Self { + Self::Checkpoint { + service: key.service.to_owned(), + space: key.space.to_owned(), + name: key.name.to_owned(), + content_hash: content_hash.to_owned(), + size_bytes: u64::try_from(size_bytes).unwrap_or(0), + } + } + + fn wal( + key: &ArtifactKey<'_>, + revision: i64, + checkpoint_content_hash: &str, + delta_content_hash: &str, + ) -> Self { + Self::Wal { + service: key.service.to_owned(), + space: key.space.to_owned(), + name: key.name.to_owned(), + revision, + checkpoint_content_hash: checkpoint_content_hash.to_owned(), + delta_content_hash: delta_content_hash.to_owned(), + } + } + + /// The binding of `part` of the row `meta` describes. + fn of(key: &ArtifactKey<'_>, meta: &ArtifactMeta, part: ArtifactPart) -> Self { + match part { + ArtifactPart::Checkpoint => Self::checkpoint( + key, + &meta.checkpoint_content_hash, + meta.checkpoint_size_bytes, + ), + ArtifactPart::Wal => Self::wal( + key, + meta.revision, + &meta.checkpoint_content_hash, + meta.delta_content_hash.as_deref().unwrap_or_default(), + ), + } + } + + fn identity(&self) -> BlobIdentity<'_> { + match self { + Self::Checkpoint { + service, + space, + name, + content_hash, + size_bytes, + } => BlobIdentity::SqlCheckpoint { + service, + space, + name, + checkpoint_content_hash: content_hash, + checkpoint_size_bytes: *size_bytes, + }, + Self::Wal { + service, + space, + name, + revision, + checkpoint_content_hash, + delta_content_hash, + } => BlobIdentity::SqlWal { + service, + space, + name, + revision: *revision, + checkpoint_content_hash, + delta_content_hash, + }, + } + } +} + +/// A row's manifest (`docs/at-rest/INTERFACES.md` §8): its revision, logical +/// hash and size, and for each part it holds the plaintext hash, size and +/// encoding. A sealed write stores a tag for it in `manifest` (a TCE1 seal +/// of nothing under this identity, so a key-bound MAC); a read checks the +/// tag before any part. A dropped delta, a part or metadata from another +/// revision, or a changed encoding then fails as integrity instead of +/// reading as some other database. +struct RowManifest<'a> { + key: ArtifactKey<'a>, + revision: i64, + content_hash: &'a str, + size_bytes: i64, + checkpoint: SqlManifestPart<'a>, + delta: Option>, +} + +impl<'a> RowManifest<'a> { + fn of_meta(key: ArtifactKey<'a>, meta: &'a ArtifactMeta) -> Self { + Self { + key, + revision: meta.revision, + content_hash: &meta.content_hash, + size_bytes: meta.size_bytes, + checkpoint: SqlManifestPart { + content_hash: &meta.checkpoint_content_hash, + size_bytes: meta.checkpoint_size_bytes, + encoding: meta.payload_encoding, + }, + delta: meta + .delta_content_hash + .as_deref() + .map(|content_hash| SqlManifestPart { + content_hash, + size_bytes: meta.delta_size_bytes, + encoding: meta.delta_encoding, + }), + } + } + + fn of_row(row: &'a database_artifact::Model) -> Self { + Self { + key: ArtifactKey { + service: &row.service, + space: &row.space, + name: &row.name, + }, + revision: row.revision, + content_hash: &row.content_hash, + size_bytes: row.size_bytes, + checkpoint: SqlManifestPart { + content_hash: &row.checkpoint_content_hash, + size_bytes: row.checkpoint_size_bytes, + encoding: row.payload_encoding, + }, + delta: row + .delta_content_hash + .as_deref() + .map(|content_hash| SqlManifestPart { + content_hash, + size_bytes: row.delta_size_bytes, + encoding: row.delta_encoding, + }), + } + } + + fn identity(&self) -> BlobIdentity<'_> { + BlobIdentity::SqlManifest { + service: self.key.service, + space: self.key.space, + name: self.key.name, + revision: self.revision, + content_hash: self.content_hash, + size_bytes: self.size_bytes, + checkpoint: self.checkpoint, + delta: self.delta, + } + } + + /// The tag a sealed write stores with the row. + fn seal(&self, key: &DataKey) -> Result, DatabaseArtifactError> { + // `seal` refuses only a plaintext past the TCE1 limit; this one is empty. + at_rest::seal(key, &self.identity(), Vec::new()) + .map_err(|_| DatabaseArtifactError::PayloadTooLarge(0)) + } + + /// Checks the row's stored `tag` against this manifest, before any part + /// is read. Only mode `off` writes a row without a tag: one with no + /// sealed part, or a delta written in `off` over a sealed checkpoint (that + /// legacy delta is as unauthenticated as any legacy part, and `strict` + /// refuses it). Any other row without a tag fails as integrity. + async fn verify( + &self, + tag: Option<&[u8]>, + keys: Option<&dyn DataKeyProvider>, + ) -> Result<(), DatabaseArtifactError> { + const ROLE: &str = "manifest"; + // An unknown or invalid encoding keeps its own error. + let checkpoint = PayloadEncoding::from_column(self.checkpoint.encoding) + .map_err(|error| envelope_error(ArtifactPart::Checkpoint, error))?; + let delta = self + .delta + .map(|delta| { + PayloadEncoding::from_column(delta.encoding) + .map_err(|error| envelope_error(ArtifactPart::Wal, error)) + }) + .transpose()?; + let Some(tag) = tag else { + return match (checkpoint, delta) { + (PayloadEncoding::LegacyPlaintext, None) + | (_, Some(PayloadEncoding::LegacyPlaintext)) => Ok(()), + _ => { + tracing::error!( + role = ROLE, + "Database artifact has sealed parts but no manifest tag" + ); + Err(DatabaseArtifactError::AtRestIntegrity { part: ROLE }) + } + }; + }; + let Some(keys) = keys else { + tracing::warn!( + role = ROLE, + "Sealed database artifact read with no at-rest keys configured" + ); + return Err(DatabaseArtifactError::AtRestUnavailable); + }; + let header = Header::parse(tag).map_err(|error| envelope_error_at(ROLE, error))?; + let data_key = keys + .key(self.key.space, header.dek_id()) + .await + .map_err(|error| key_error_at(ROLE, error))?; + let opened = at_rest::open(&data_key, &self.identity(), tag.to_vec()) + .map_err(|error| envelope_error_at(ROLE, error))?; + if !opened.is_empty() { + return Err(envelope_error_at(ROLE, EnvelopeError::Integrity)); + } + Ok(()) + } +} + +/// Room for sealing a part of `len` plaintext bytes in place. +fn sealed_capacity(len: usize) -> Result { + at_rest::ciphertext_len(len as u64) + .ok() + .and_then(|stored| usize::try_from(stored).ok()) + .ok_or(DatabaseArtifactError::PayloadTooLarge(len as u64)) +} + +/// Seal one part off the async runtime (a checkpoint can be tens of MiB). +/// `plaintext` should already have [`sealed_capacity`], so sealing in place +/// neither reallocates nor copies. +async fn seal_part( + key: DataKey, + binding: PartBinding, + plaintext: Vec, +) -> Result, DatabaseArtifactError> { + let len = plaintext.len() as u64; + tokio::task::spawn_blocking(move || at_rest::seal(&key, &binding.identity(), plaintext)) + .await + .map_err(|error| DatabaseArtifactError::Backend(error.to_string()))? + // `seal` refuses only a plaintext past the TCE1 size limit. + .map_err(|_| DatabaseArtifactError::PayloadTooLarge(len)) +} + +/// Turns one part's stored bytes into its plaintext as they stream. Legacy +/// plaintext passes through; a TCE1 part is opened one authenticated 64 KiB +/// segment at a time, so at most one segment is buffered. What it hands on +/// is what the sink receives and the content hash covers. +enum PartDecoder { + Plain, + /// Collecting the 64-byte header, which names the data key. + Header { + header: Vec, + provider: Arc, + binding: PartBinding, + plaintext_len: u64, + }, + Sealed(Box), +} + +impl PartDecoder { + async fn feed( + &mut self, + part: ArtifactPart, + space: &str, + mut stored: &[u8], + hasher: &mut Blake3Hasher, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + if let Self::Header { + header, + provider, + binding, + plaintext_len, + } = self + { + let take = (HEADER_LEN - header.len()).min(stored.len()); + header.extend_from_slice(&stored[..take]); + stored = &stored[take..]; + if header.len() < HEADER_LEN { + return Ok(()); + } + let parsed = Header::parse(header).map_err(|error| envelope_error(part, error))?; + // The row's plaintext size is part of the stored length already + // read; a header that disagrees is refused before any key lookup. + // The segments authenticate the header either way. + if parsed.plaintext_len() != *plaintext_len { + return Err(envelope_error(part, EnvelopeError::Integrity)); + } + let key = provider + .key(space, parsed.dek_id()) + .await + .map_err(|error| read_key_error(part, error))?; + let opener = SegmentOpener::new(&key, &binding.identity(), &parsed) + .map_err(|error| envelope_error(part, error))?; + *self = Self::Sealed(Box::new(opener)); + } + match self { + Self::Plain => { + hasher.update(stored); + sink.write(stored).await + } + Self::Sealed(opener) => { + while let Some(plain) = opener + .push(&mut stored) + .map_err(|error| envelope_error(part, error))? + { + // An empty part is one empty segment; like a legacy + // empty part, the sink sees no write. + if !plain.is_empty() { + hasher.update(plain); + sink.write(plain).await?; + } + } + Ok(()) + } + Self::Header { .. } => Ok(()), + } + } + + /// Every stored byte was fed: a sealed part must have opened completely. + fn finish(&self, part: ArtifactPart) -> Result<(), DatabaseArtifactError> { + match self { + Self::Plain => Ok(()), + Self::Header { .. } => Err(envelope_error(part, EnvelopeError::Integrity)), + Self::Sealed(opener) => opener.finish().map_err(|error| envelope_error(part, error)), + } + } +} + +/// The plaintext of one whole stored part, for [`open_row`]. +async fn open_whole( + key: &ArtifactKey<'_>, + part: ArtifactPart, + encoding: Option, + stored: Vec, + binding: PartBinding, + keys: Option<&dyn DataKeyProvider>, +) -> Result, DatabaseArtifactError> { + match PayloadEncoding::from_column(encoding).map_err(|error| envelope_error(part, error))? { + PayloadEncoding::LegacyPlaintext => Ok(stored), + PayloadEncoding::Tce1 => { + let Some(keys) = keys else { + return Err(DatabaseArtifactError::AtRestUnavailable); + }; + let header = Header::parse(&stored).map_err(|error| envelope_error(part, error))?; + let data_key = keys + .key(key.space, header.dek_id()) + .await + .map_err(|error| read_key_error(part, error))?; + at_rest::open(&data_key, &binding.identity(), stored) + .map_err(|error| envelope_error(part, error)) + } + } +} + +/// The plaintext checkpoint and WAL of a whole stored row, for the offline +/// tool that inspects designated local snapshots (the node itself only +/// streams). Legacy parts come back as stored. The row's manifest tag is +/// checked first, as on a streamed read; sealed parts need `keys` and are +/// opened and authenticated, or the call fails: ciphertext is never +/// returned as a database file. +pub async fn open_row( + row: database_artifact::Model, + keys: Option<&dyn DataKeyProvider>, +) -> Result<(Vec, Option>), DatabaseArtifactError> { + RowManifest::of_row(&row) + .verify(row.manifest.as_deref(), keys) + .await?; + let key = ArtifactKey { + service: &row.service, + space: &row.space, + name: &row.name, + }; + let checkpoint = open_whole( + &key, + ArtifactPart::Checkpoint, + row.payload_encoding, + row.payload, + PartBinding::checkpoint( + &key, + &row.checkpoint_content_hash, + row.checkpoint_size_bytes, + ), + keys, + ) + .await?; + let wal = match row.delta_payload { + Some(delta) => Some( + open_whole( + &key, + ArtifactPart::Wal, + row.delta_encoding, + delta, + PartBinding::wal( + &key, + row.revision, + &row.checkpoint_content_hash, + row.delta_content_hash.as_deref().unwrap_or_default(), + ), + keys, + ) + .await?, + ), + None => None, + }; + Ok((checkpoint, wal)) +} + impl SeaOrmDatabaseArtifactRepository { - /// Stream one payload column into `sink`, hash it, and return its CID. + /// Stream one payload column into `sink`, decode and hash it, and return + /// the plaintext's CID. async fn read_part( &self, key: &ArtifactKey<'_>, meta: &ArtifactMeta, part: ArtifactPart, - size_bytes: i64, - expected_hash: &str, + view: &AtRestView, sink: &mut dyn ArtifactSink, ) -> Result { - let column = match part { - ArtifactPart::Checkpoint => database_artifact::Column::Payload, - ArtifactPart::Wal => database_artifact::Column::DeltaPayload, + let (column, size_bytes, expected_hash, encoding) = match part { + ArtifactPart::Checkpoint => ( + database_artifact::Column::Payload, + meta.checkpoint_size_bytes, + meta.checkpoint_content_hash.as_str(), + meta.payload_encoding, + ), + ArtifactPart::Wal => ( + database_artifact::Column::DeltaPayload, + meta.delta_size_bytes, + meta.delta_content_hash.as_deref().unwrap_or_default(), + meta.delta_encoding, + ), + }; + let plaintext_len = u64::try_from(size_bytes).unwrap_or(0); + // Legacy or sealed is the column's answer, decided before any byte. + let mut decoder = match PayloadEncoding::from_column(encoding) + .map_err(|error| envelope_error(part, error))? + { + PayloadEncoding::LegacyPlaintext if !view.legacy_readable() => { + tracing::error!( + role = part.as_str(), + "Legacy plaintext database artifact refused in strict at-rest mode" + ); + return Err(DatabaseArtifactError::LegacyPlaintextRefused { + part: part.as_str(), + }); + } + PayloadEncoding::LegacyPlaintext => PartDecoder::Plain, + PayloadEncoding::Tce1 => { + let Some(provider) = view.provider.clone() else { + tracing::warn!( + role = part.as_str(), + "Sealed database artifact read with no at-rest keys configured" + ); + return Err(DatabaseArtifactError::AtRestUnavailable); + }; + PartDecoder::Header { + header: Vec::with_capacity(HEADER_LEN), + provider, + binding: PartBinding::of(key, meta, part), + plaintext_len, + } + } + }; + // A sealed part stores exactly `ciphertext_len(size)` bytes. Its last + // slice asks for one byte more: bytes appended past that length, or + // a part cut short, fail as integrity. + let sealed = !matches!(decoder, PartDecoder::Plain); + let expected_len = if sealed { + at_rest::ciphertext_len(plaintext_len).map_err(|error| envelope_error(part, error))? + } else { + plaintext_len }; - let expected_len = u64::try_from(size_bytes).unwrap_or(0); // An artifact that fits the base-timeout budget is one slice per part: // no extra round trips for the common small database. - // Gated on the part's own size as well, so inconsistent metadata - // can never ask for a slice past the budget. - let budget = BASE_TIMEOUT_PAYLOAD_BYTES as i64; - let slice_bytes = if meta.size_bytes <= budget && size_bytes <= budget { + // Gated on the part's own stored size as well, so inconsistent + // metadata can never ask for a slice past the budget. + let budget = BASE_TIMEOUT_PAYLOAD_BYTES as u64; + let slice_bytes = if meta.size_bytes <= budget as i64 && expected_len <= budget { expected_len.max(1) } else { ARTIFACT_CHUNK_BYTES as u64 }; let mut hasher = Blake3Hasher::new(); sink.begin_part(part).await?; - sink.size_hint(part, expected_len); + sink.size_hint(part, plaintext_len); let mut offset = 0_u64; while offset < expected_len { let len = (expected_len - offset).min(slice_bytes); + let ask = if sealed && offset + len == expected_len { + len + 1 + } else { + len + }; // `substr` is 1-based and takes int4 on PostgreSQL. let start = i32::try_from(offset + 1) .map_err(|_| DatabaseArtifactError::PayloadTooLarge(expected_len))?; @@ -773,7 +1387,7 @@ impl SeaOrmDatabaseArtifactRepository { Func::cust("substr") .arg(Expr::col(column)) .arg(start) - .arg(len as i32), + .arg(ask as i32), ) .filter(database_artifact::Column::Service.eq(key.service)) .filter(database_artifact::Column::Space.eq(key.space)) @@ -792,16 +1406,28 @@ impl SeaOrmDatabaseArtifactRepository { }; let slice = slice.unwrap_or_default(); if slice.len() as u64 != len { + if sealed { + tracing::error!( + role = part.as_str(), + expected = expected_len, + "Sealed database artifact part is not its sealed length" + ); + return Err(DatabaseArtifactError::AtRestIntegrity { + part: part.as_str(), + }); + } return Err(DatabaseArtifactError::SizeMismatch { part: part.as_str(), expected: expected_len, actual: offset + slice.len() as u64, }); } - hasher.update(&slice); - sink.write(&slice).await?; + decoder + .feed(part, key.space, &slice, &mut hasher, sink) + .await?; offset += len; } + decoder.finish(part)?; let actual = hasher.finalize().to_cid(0x55).to_string(); if actual != expected_hash { return Err(DatabaseArtifactError::HashMismatch { @@ -822,7 +1448,7 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { name: &str, ) -> Result, DatabaseArtifactError> { use database_artifact::Column; - let row = database_artifact::Entity::find_by_id(( + Ok(database_artifact::Entity::find_by_id(( service.to_string(), space.to_string(), name.to_string(), @@ -838,45 +1464,12 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { .column(Column::UpdatedAt) .column(Column::StorageMode) .column(Column::Backend) - .into_tuple::<( - i64, - i64, - String, - String, - i64, - Option, - i64, - String, - String, - String, - )>() + .column(Column::PayloadEncoding) + .column(Column::DeltaEncoding) + .column(Column::Manifest) + .into_model::() .one(&self.conn) - .await?; - Ok(row.map( - |( - revision, - size_bytes, - content_hash, - checkpoint_content_hash, - checkpoint_size_bytes, - delta_content_hash, - delta_size_bytes, - updated_at, - storage_mode, - backend, - )| ArtifactMeta { - revision, - size_bytes, - content_hash, - checkpoint_content_hash, - checkpoint_size_bytes, - delta_content_hash, - delta_size_bytes, - updated_at, - storage_mode, - backend, - }, - )) + .await?) } async fn read_chunks( @@ -892,27 +1485,19 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { space, name, }; + // One view for both parts: the mode and keys a read sees. The + // manifest tag is checked before any byte reaches the sink. + let view = self.at_rest_view(); + RowManifest::of_meta(key, meta) + .verify(meta.manifest.as_deref(), view.provider.as_deref()) + .await?; let checkpoint = self - .read_part( - &key, - meta, - ArtifactPart::Checkpoint, - meta.checkpoint_size_bytes, - &meta.checkpoint_content_hash, - sink, - ) + .read_part(&key, meta, ArtifactPart::Checkpoint, &view, sink) .await?; let logical = match &meta.delta_content_hash { Some(delta_hash) => { - self.read_part( - &key, - meta, - ArtifactPart::Wal, - meta.delta_size_bytes, - delta_hash, - sink, - ) - .await?; + self.read_part(&key, meta, ArtifactPart::Wal, &view, sink) + .await?; hash(format!("{checkpoint}:{delta_hash}").as_bytes()) .to_cid(0x55) .to_string() @@ -933,13 +1518,37 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { payload: Vec, expected: ArtifactExpectation, ) -> Result { + // Before the writer lock: preparing the space's key may write the + // key store, which takes that lock on SQLite. + let data_key = self.write_key(space).await?; + let size_bytes = i64::try_from(payload.len()) + .map_err(|_| DatabaseArtifactError::PayloadTooLarge(payload.len() as u64))?; + // Hash and size stay those of the plaintext, sealed or not. + let content_hash = hash(&payload).to_cid(0x55).to_string(); + let (stored, encoding) = match &data_key { + None => (payload.clone(), PayloadEncoding::LegacyPlaintext), + Some(data_key) => { + let mut plaintext = Vec::with_capacity(sealed_capacity(payload.len())?); + plaintext.extend_from_slice(&payload); + let binding = PartBinding::checkpoint( + &ArtifactKey { + service, + space, + name, + }, + &content_hash, + size_bytes, + ); + ( + seal_part(data_key.clone(), binding, plaintext).await?, + PayloadEncoding::Tce1, + ) + } + }; let _writer = match &self.sqlite_writer_lock { Some(lock) => Some(lock.lock().await), None => None, }; - let size_bytes = i64::try_from(payload.len()) - .map_err(|_| DatabaseArtifactError::PayloadTooLarge(payload.len() as u64))?; - let content_hash = hash(&payload).to_cid(0x55).to_string(); let now = OffsetDateTime::now_utc() .format(&Rfc3339) .expect("current timestamps should format as RFC3339"); @@ -948,7 +1557,7 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { // the pool's base statement timeout allows, so its write runs under // the bulk one (TC-1054). Smaller ones, like every delta, fit the // base timeout and pay no extra round trip. - let scope = if payload.len() > BASE_TIMEOUT_PAYLOAD_BYTES { + let scope = if stored.len() > BASE_TIMEOUT_PAYLOAD_BYTES { self.conn.begin_bulk().await? } else { BulkScope::Pool(self.conn.clone()) @@ -1007,6 +1616,29 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { .as_ref() .map(|model| model.revision + 1) .unwrap_or(1); + // The new row's manifest tag when sealing; mode `off` clears it. + let manifest = data_key + .as_ref() + .map(|data_key| { + RowManifest { + key: ArtifactKey { + service, + space, + name, + }, + revision, + content_hash: &content_hash, + size_bytes, + checkpoint: SqlManifestPart { + content_hash: &content_hash, + size_bytes, + encoding: encoding.to_column(), + }, + delta: None, + } + .seal(data_key) + }) + .transpose()?; // Barrier seam (tests only): lets a second writer observe the same base // revision before either commits, exercising the CAS conflict path. @@ -1030,9 +1662,10 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { database_artifact::Column::ContentHash, Expr::value(content_hash.clone()), ) + .col_expr(database_artifact::Column::Payload, Expr::value(stored)) .col_expr( - database_artifact::Column::Payload, - Expr::value(payload.clone()), + database_artifact::Column::PayloadEncoding, + Expr::value(encoding.to_column()), ) .col_expr( database_artifact::Column::SizeBytes, @@ -1070,6 +1703,11 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { database_artifact::Column::DeltaSizeBytes, Expr::value(0_i64), ) + .col_expr( + database_artifact::Column::DeltaEncoding, + Expr::value(Option::::None), + ) + .col_expr(database_artifact::Column::Manifest, Expr::value(manifest)) .filter(database_artifact::Column::Service.eq(service)) .filter(database_artifact::Column::Space.eq(space)) .filter(database_artifact::Column::Name.eq(name)) @@ -1093,7 +1731,7 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { name: Set(name.to_string()), revision: Set(revision), content_hash: Set(content_hash.clone()), - payload: Set(payload.clone()), + payload: Set(stored), size_bytes: Set(size_bytes), backend: Set("storage.database".to_string()), storage_mode: Set("database-blob".to_string()), @@ -1104,6 +1742,9 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(encoding.to_column()), + delta_encoding: Set(None), + manifest: Set(manifest), } .insert(&scope) .await @@ -1147,6 +1788,9 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { payload: Vec, expected: ArtifactExpectation, ) -> Result { + // Before the writer lock: preparing the space's key may write the + // key store, which takes that lock on SQLite. + let data_key = self.write_key(space).await?; let _writer = match &self.sqlite_writer_lock { Some(lock) => Some(lock.lock().await), None => None, @@ -1160,7 +1804,8 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { .column(database_artifact::Column::Revision) .column(database_artifact::Column::CheckpointSizeBytes) .column(database_artifact::Column::CheckpointContentHash) - .into_tuple::<(i64, i64, String)>() + .column(database_artifact::Column::PayloadEncoding) + .into_tuple::<(i64, i64, String, Option)>() .one(&self.conn) .await? .ok_or(DatabaseArtifactError::MissingCheckpoint)?; @@ -1183,6 +1828,60 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { .checked_add(delta_size_bytes) .ok_or(DatabaseArtifactError::PayloadTooLarge(u64::MAX))?; let revision = existing.0 + 1; + // Hash and size above stay those of the plaintext; the sealed WAL is + // bound to the revision it is written at and the checkpoint it + // applies to. + let (stored, encoding) = match &data_key { + None => (payload, PayloadEncoding::LegacyPlaintext), + Some(data_key) => { + let mut plaintext = payload; + plaintext.reserve_exact(sealed_capacity(plaintext.len())? - plaintext.len()); + let binding = PartBinding::wal( + &ArtifactKey { + service, + space, + name, + }, + revision, + &existing.2, + &delta_content_hash, + ); + ( + seal_part(data_key.clone(), binding, plaintext).await?, + PayloadEncoding::Tce1, + ) + } + }; + // The row's new manifest tag when sealing; mode `off` clears it. Its + // checkpoint fields are the columns read above (the update below is + // pinned to that revision and checkpoint hash); a checkpoint whose + // stored size or encoding disagrees still fails its own read checks. + let manifest = data_key + .as_ref() + .map(|data_key| { + RowManifest { + key: ArtifactKey { + service, + space, + name, + }, + revision, + content_hash: &logical_content_hash, + size_bytes, + checkpoint: SqlManifestPart { + content_hash: &existing.2, + size_bytes: existing.1, + encoding: existing.3, + }, + delta: Some(SqlManifestPart { + content_hash: &delta_content_hash, + size_bytes: delta_size_bytes, + encoding: encoding.to_column(), + }), + } + .seal(data_key) + }) + .transpose()?; let now = OffsetDateTime::now_utc() .format(&Rfc3339) .expect("current timestamps should format as RFC3339"); @@ -1191,15 +1890,19 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { .col_expr(database_artifact::Column::Revision, Expr::value(revision)) .col_expr( database_artifact::Column::ContentHash, - Expr::value(logical_content_hash), + Expr::value(logical_content_hash.clone()), ) .col_expr( database_artifact::Column::DeltaPayload, - Expr::value(Some(payload)), + Expr::value(Some(stored)), + ) + .col_expr( + database_artifact::Column::DeltaEncoding, + Expr::value(encoding.to_column()), ) .col_expr( database_artifact::Column::DeltaContentHash, - Expr::value(Some(delta_content_hash)), + Expr::value(Some(delta_content_hash.clone())), ) .col_expr( database_artifact::Column::DeltaSizeBytes, @@ -1214,6 +1917,7 @@ impl DatabaseArtifactRepository for SeaOrmDatabaseArtifactRepository { Expr::value("checkpoint+wal"), ) .col_expr(database_artifact::Column::UpdatedAt, Expr::value(now)) + .col_expr(database_artifact::Column::Manifest, Expr::value(manifest)) .filter(database_artifact::Column::Service.eq(service)) .filter(database_artifact::Column::Space.eq(space)) .filter(database_artifact::Column::Name.eq(name)) @@ -1515,6 +2219,9 @@ mod tests { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(None), + delta_encoding: Set(None), + manifest: Set(None), } .insert(&tx) .await @@ -1601,6 +2308,9 @@ mod tests { updated_at: winner.updated_at, storage_mode: winner.storage_mode, backend: winner.backend, + payload_encoding: None, + delta_encoding: None, + manifest: None, } ); @@ -1860,8 +2570,10 @@ mod tests { let seen = statements.lock().unwrap().clone(); assert_eq!(seen.len(), 2, "{seen:?}"); + // The encoding columns are metadata; the payload columns are not. assert!( - seen.iter().all(|sql| !sql.contains("payload")), + seen.iter() + .all(|sql| !sql.contains("\"payload\"") && !sql.contains("\"delta_payload\"")), "a metadata read must not touch payload columns: {seen:?}" ); @@ -2467,4 +3179,936 @@ mod tests { ); } } + + // ---- sealed at rest (TC-887) ---- + + use crate::at_rest::{ + ciphertext_len, AtRestMode, DataKeyProvider, FixedKeyProvider, TestAtRestSource, + }; + + const SQLITE_HEADER: &[u8] = b"SQLite format 3\0"; + const WAL_HEADER: &[u8] = &[0x37, 0x7f, 0x06, 0x82, 0, 0x2d, 0xe2, 0x18]; + const MARKER: &[u8] = b"tc887-sql-artifact-marker"; + + fn keys() -> Arc { + Arc::new(FixedKeyProvider::new([7; 32])) + } + + /// A database-shaped part: the engine's file header, then filler with a + /// marker in it. + fn database_bytes(header: &[u8], len: usize, seed: u8) -> Vec { + let mut out = bytes(len, seed); + out[..header.len()].copy_from_slice(header); + out[100..100 + MARKER.len()].copy_from_slice(MARKER); + out + } + + fn contains(haystack: &[u8], needle: &[u8]) -> bool { + haystack + .windows(needle.len()) + .any(|window| window == needle) + } + + fn cid(bytes: &[u8]) -> String { + hash(bytes).to_cid(0x55).to_string() + } + + async fn sqlite_conn() -> DatabaseConnection { + let mut options = ConnectOptions::new("sqlite::memory:".to_string()); + options.max_connections(1); + let conn = Database::connect(options).await.unwrap(); + Migrator::up(&conn, None).await.unwrap(); + conn + } + + fn at_rest_repo( + conn: &DatabaseConnection, + mode: AtRestMode, + ) -> (SeaOrmDatabaseArtifactRepository, Arc) { + let source = TestAtRestSource::new(mode, mode != AtRestMode::Off, Some(keys())); + let repo = SeaOrmDatabaseArtifactRepository::new(conn.clone()).with_at_rest(source.clone()); + (repo, source) + } + + /// The stored row, payload bytes included. + async fn stored_row( + conn: &DatabaseConnection, + space: &str, + name: &str, + ) -> Option { + database_artifact::Entity::find_by_id(("sql".to_owned(), space.to_owned(), name.to_owned())) + .one(conn) + .await + .unwrap() + } + + /// Overwrite the stored parts of `space/name` with `row`'s, the way + /// someone with write access to the database could. + async fn put_parts( + conn: &DatabaseConnection, + space: &str, + name: &str, + row: &database_artifact::Model, + ) { + use database_artifact::Column; + database_artifact::Entity::update_many() + .col_expr(Column::Payload, Expr::value(row.payload.clone())) + .col_expr(Column::PayloadEncoding, Expr::value(row.payload_encoding)) + .col_expr(Column::DeltaPayload, Expr::value(row.delta_payload.clone())) + .col_expr(Column::DeltaEncoding, Expr::value(row.delta_encoding)) + .filter(Column::Service.eq("sql")) + .filter(Column::Space.eq(space)) + .filter(Column::Name.eq(name)) + .exec(conn) + .await + .unwrap(); + } + + /// Reads `space/name` into a cache sink that checks the real file headers, + /// as hydration does, and publishes it. + async fn hydrate( + repo: &SeaOrmDatabaseArtifactRepository, + space: &str, + name: &str, + ) -> (Vec, Option>) { + let dir = tempfile::tempdir().unwrap(); + let mut sink = CacheFileSink::new( + "sql", + format!("{space}/{name}"), + dir.path().join("main.db"), + dir.path().join("main.db-wal"), + |head| head.starts_with(SQLITE_HEADER), + |head| head.starts_with(WAL_HEADER), + ); + read_artifact(repo, "sql", space, name, &mut sink) + .await + .unwrap() + .expect("the artifact exists"); + sink.commit().await.unwrap(); + ( + std::fs::read(dir.path().join("main.db")).unwrap(), + std::fs::read(dir.path().join("main.db-wal")).ok(), + ) + } + + /// A read that must fail; whatever the sink staged is gone afterwards. + async fn read_error( + repo: &SeaOrmDatabaseArtifactRepository, + space: &str, + name: &str, + ) -> DatabaseArtifactError { + let dir = tempfile::tempdir().unwrap(); + let mut sink = cache_sink(dir.path()); + let error = read_artifact(repo, "sql", space, name, &mut sink) + .await + .expect_err("must not read"); + drop(sink); + assert!( + dir_names(dir.path()).is_empty(), + "{error:?}: staged bytes left" + ); + error + } + + /// Saves a multi-MiB checkpoint and a WAL delta sealed, and checks what + /// reached the database and what a hydration gets back. The read is + /// sliced at 4 MiB, which never falls on a 64 KiB segment boundary of + /// the stored (ciphertext) bytes. + async fn assert_sealed_round_trip( + repo: &SeaOrmDatabaseArtifactRepository, + conn: &DatabaseConnection, + ) { + let checkpoint = database_bytes(SQLITE_HEADER, 2 * CHUNK + 70_001, 1); + let wal = database_bytes(WAL_HEADER, 200_003, 2); + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + let saved = repo + .save_delta( + "sql", + "space", + "main", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + assert_eq!(saved.delta_size_bytes, wal.len() as i64); + + let row = stored_row(conn, "space", "main").await.unwrap(); + assert_eq!( + (row.payload_encoding, row.delta_encoding), + (Some(1), Some(1)) + ); + let delta = row.delta_payload.as_deref().unwrap(); + for (stored, plain) in [(&row.payload[..], &checkpoint), (delta, &wal)] { + assert!(stored.starts_with(b"TCE1")); + assert!(!stored.starts_with(SQLITE_HEADER) && !stored.starts_with(WAL_HEADER)); + assert!(!contains(stored, MARKER), "plaintext reached the database"); + assert_eq!( + stored.len() as u64, + ciphertext_len(plain.len() as u64).unwrap() + ); + } + // Sizes and hashes stay those of the plaintext: quota, growth checks + // and lineage read them unchanged. + assert_eq!(row.checkpoint_size_bytes, checkpoint.len() as i64); + assert_eq!(row.delta_size_bytes, wal.len() as i64); + assert_eq!(row.size_bytes, (checkpoint.len() + wal.len()) as i64); + assert_eq!(row.checkpoint_content_hash, cid(&checkpoint)); + assert_eq!(row.delta_content_hash, Some(cid(&wal))); + let meta = repo + .load_meta("sql", "space", "main") + .await + .unwrap() + .unwrap(); + assert_eq!(meta.size_bytes, (checkpoint.len() + wal.len()) as i64); + assert_eq!( + (meta.payload_encoding, meta.delta_encoding), + (Some(1), Some(1)) + ); + + assert_eq!( + hydrate(repo, "space", "main").await, + (checkpoint, Some(wal)) + ); + } + + #[tokio::test] + async fn sealed_parts_hold_no_plaintext_and_hydrate_exactly() { + let conn = sqlite_conn().await; + let (repo, source) = at_rest_repo(&conn, AtRestMode::Encrypt); + assert_sealed_round_trip(&repo, &conn).await; + + // Sealed parts are read whenever the keys exist, in every mode. + let expected = hydrate(&repo, "space", "main").await; + for mode in [AtRestMode::Off, AtRestMode::Strict] { + source.set(mode, false); + assert_eq!(hydrate(&repo, "space", "main").await, expected, "{mode:?}"); + } + + // An empty database seals and reads back as empty. + source.set(AtRestMode::Encrypt, true); + repo.save("sql", "space", "empty", vec![], ArtifactExpectation::Any) + .await + .unwrap(); + let row = stored_row(&conn, "space", "empty").await.unwrap(); + assert_eq!(row.payload_encoding, Some(1)); + assert_eq!(row.payload.len() as u64, ciphertext_len(0).unwrap()); + let mut sink = MemorySink::default(); + read_artifact(&repo, "sql", "space", "empty", &mut sink) + .await + .unwrap() + .unwrap(); + assert!(sink.checkpoint.is_empty() && sink.wal.is_none()); + } + + #[tokio::test] + async fn postgres_sealed_parts_round_trip_and_refuse_a_moved_part() { + let Some(fixture) = crate::test_support::PostgresTestSchema::migrated_through( + "postgres_sealed_parts_round_trip_and_refuse_a_moved_part", + None, + ) + .await + else { + return; + }; + let (repo, _source) = at_rest_repo(&fixture.conn, AtRestMode::Encrypt); + assert_sealed_round_trip(&repo, &fixture.conn).await; + + // The same database saved in another space: equal hash and size, but + // its ciphertext does not open as this space's. + let checkpoint = database_bytes(SQLITE_HEADER, 2 * CHUNK + 70_001, 1); + repo.save("sql", "other", "main", checkpoint, ArtifactExpectation::Any) + .await + .unwrap(); + let mut moved = stored_row(&fixture.conn, "space", "main").await.unwrap(); + moved.payload = stored_row(&fixture.conn, "other", "main") + .await + .unwrap() + .payload; + put_parts(&fixture.conn, "space", "main", &moved).await; + assert!(matches!( + read_error(&repo, "space", "main").await, + DatabaseArtifactError::AtRestIntegrity { part: "checkpoint" } + )); + } + + #[tokio::test] + async fn off_writes_plaintext_and_legacy_parts_read_unless_strict() { + let conn = sqlite_conn().await; + let (repo, source) = at_rest_repo(&conn, AtRestMode::Off); + let checkpoint = database_bytes(SQLITE_HEADER, 70_000, 3); + let wal = database_bytes(WAL_HEADER, 5_000, 4); + + // Mode `off` stores the bytes as given, keys or not. + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let row = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!(row.payload, checkpoint); + assert_eq!((row.payload_encoding, row.delta_encoding), (None, None)); + + // In `encrypt` a legacy checkpoint still reads, and a new WAL delta + // on top of it is sealed on its own. + source.set(AtRestMode::Encrypt, true); + assert_eq!( + hydrate(&repo, "space", "main").await, + (checkpoint.clone(), None) + ); + repo.save_delta( + "sql", + "space", + "main", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let row = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!(row.payload, checkpoint); + assert_eq!((row.payload_encoding, row.delta_encoding), (None, Some(1))); + assert_eq!( + hydrate(&repo, "space", "main").await, + (checkpoint.clone(), Some(wal.clone())) + ); + + // `strict` refuses the legacy checkpoint instead of serving it. + source.set(AtRestMode::Strict, true); + assert!(matches!( + read_error(&repo, "space", "main").await, + DatabaseArtifactError::LegacyPlaintextRefused { part: "checkpoint" } + )); + // A full save seals it, and the delta it replaces goes with it. + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let row = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!((row.payload_encoding, row.delta_encoding), (Some(1), None)); + assert_eq!(hydrate(&repo, "space", "main").await, (checkpoint, None)); + + // Back in `off`, a full save over a sealed row is plaintext again. + source.set(AtRestMode::Off, false); + repo.save( + "sql", + "space", + "main", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let row = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!((row.payload.clone(), row.payload_encoding), (wal, None)); + } + + #[tokio::test] + async fn without_unlocked_keys_writes_fail_and_change_nothing() { + let conn = sqlite_conn().await; + let (repo, source) = at_rest_repo(&conn, AtRestMode::Off); + let checkpoint = database_bytes(SQLITE_HEADER, 9_000, 5); + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let before = stored_row(&conn, "space", "main").await.unwrap(); + + let locked: Arc = Arc::new(FixedKeyProvider::new([7; 32]).locked()); + for (mode, encrypting, provider) in [ + // The gate is closed: the mode asks for sealing, nothing can seal. + (AtRestMode::Encrypt, false, Some(keys())), + (AtRestMode::Strict, false, Some(keys())), + // No keyring at all. + (AtRestMode::Encrypt, true, None), + // The keys are locked. + (AtRestMode::Encrypt, true, Some(locked.clone())), + ] { + source.set(mode, encrypting); + source.set_provider(provider); + let case = format!("{mode:?} encrypting={encrypting}"); + assert!( + matches!( + repo.save( + "sql", + "space", + "new", + checkpoint.clone(), + ArtifactExpectation::Any + ) + .await, + Err(DatabaseArtifactError::AtRestUnavailable) + ), + "{case}" + ); + assert!(stored_row(&conn, "space", "new").await.is_none(), "{case}"); + assert!( + matches!( + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Any + ) + .await, + Err(DatabaseArtifactError::AtRestUnavailable) + ), + "{case}" + ); + assert!( + matches!( + repo.save_delta( + "sql", + "space", + "main", + vec![1; 64], + ArtifactExpectation::Any + ) + .await, + Err(DatabaseArtifactError::AtRestUnavailable) + ), + "{case}" + ); + assert_eq!( + stored_row(&conn, "space", "main").await.unwrap(), + before, + "{case}: the row is untouched" + ); + } + + // A sealed row is not served while its keys are locked or absent. + source.set(AtRestMode::Encrypt, true); + source.set_provider(Some(keys())); + repo.save( + "sql", + "space", + "sealed", + checkpoint, + ArtifactExpectation::Any, + ) + .await + .unwrap(); + for provider in [Some(locked), None] { + source.set_provider(provider); + assert!(matches!( + read_error(&repo, "space", "sealed").await, + DatabaseArtifactError::AtRestUnavailable + )); + } + } + + #[tokio::test] + async fn tampered_or_moved_sealed_parts_are_integrity_errors() { + let conn = sqlite_conn().await; + let (repo, _source) = at_rest_repo(&conn, AtRestMode::Encrypt); + let checkpoint = database_bytes(SQLITE_HEADER, 3 * 65_536 + 17, 6); + let wal = database_bytes(WAL_HEADER, 40_000, 7); + for (space, name) in [("space", "main"), ("other", "main"), ("space", "copy")] { + repo.save( + "sql", + space, + name, + checkpoint.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + } + repo.save_delta( + "sql", + "space", + "main", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let good = stored_row(&conn, "space", "main").await.unwrap(); + let other_space = stored_row(&conn, "other", "main").await.unwrap().payload; + let other_name = stored_row(&conn, "space", "copy").await.unwrap().payload; + + let tampered = |edit: &dyn Fn(&mut database_artifact::Model)| { + let mut row = good.clone(); + edit(&mut row); + row + }; + let flip = |bytes: &mut Vec, at: usize| bytes[at] ^= 1; + type Expect = fn(&DatabaseArtifactError) -> bool; + let cases: Vec<(&str, database_artifact::Model, Expect)> = vec![ + ( + "checkpoint byte flipped in its second segment", + tampered(&|row| flip(&mut row.payload, 64 + 65_536 + 16 + 5)), + |e| { + matches!( + e, + DatabaseArtifactError::AtRestIntegrity { part: "checkpoint" } + ) + }, + ), + ( + "checkpoint header flipped", + tampered(&|row| flip(&mut row.payload, 30)), + |e| { + matches!( + e, + DatabaseArtifactError::AtRestIntegrity { part: "checkpoint" } + ) + }, + ), + ( + "wal byte flipped", + tampered(&|row| flip(row.delta_payload.as_mut().unwrap(), 1_000)), + |e| matches!(e, DatabaseArtifactError::AtRestIntegrity { part: "wal" }), + ), + ( + "checkpoint of the same database in another space", + tampered(&|row| row.payload = other_space.clone()), + |e| { + matches!( + e, + DatabaseArtifactError::AtRestIntegrity { part: "checkpoint" } + ) + }, + ), + ( + "checkpoint of the same database under another name", + tampered(&|row| row.payload = other_name.clone()), + |e| { + matches!( + e, + DatabaseArtifactError::AtRestIntegrity { part: "checkpoint" } + ) + }, + ), + ( + "a later format generation", + tampered(&|row| row.payload_encoding = Some(2)), + |e| { + matches!( + e, + DatabaseArtifactError::AtRestUnsupported { + part: "checkpoint", + version: 2 + } + ) + }, + ), + ( + "an invalid encoding", + tampered(&|row| row.delta_encoding = Some(0)), + |e| matches!(e, DatabaseArtifactError::AtRestIntegrity { part: "wal" }), + ), + ( + // Claiming legacy never turns ciphertext into data: the + // manifest binds each part's encoding. + "sealed bytes marked legacy", + tampered(&|row| row.payload_encoding = None), + |e| { + matches!( + e, + DatabaseArtifactError::AtRestIntegrity { part: "manifest" } + ) + }, + ), + ]; + for (what, row, expected) in cases { + put_parts(&conn, "space", "main", &row).await; + let error = read_error(&repo, "space", "main").await; + assert!(expected(&error), "{what}: {error:?}"); + } + put_parts(&conn, "space", "main", &good).await; + assert_eq!( + hydrate(&repo, "space", "main").await, + (checkpoint, Some(wal.clone())) + ); + + // The same WAL saved again at the next revision: the older revision's + // ciphertext does not open in its place. + repo.save_delta("sql", "space", "main", wal, ArtifactExpectation::Any) + .await + .unwrap(); + let mut replayed = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!(replayed.revision, good.revision + 1); + replayed.delta_payload = good.delta_payload.clone(); + put_parts(&conn, "space", "main", &replayed).await; + let error = read_error(&repo, "space", "main").await; + assert!( + matches!( + error, + DatabaseArtifactError::AtRestIntegrity { part: "wal" } + ), + "{error:?}" + ); + } + + /// Sets `edit`'s columns on `space/name` directly, the way someone with + /// write access to the database could. + async fn rewrite_row( + conn: &DatabaseConnection, + space: &str, + name: &str, + edit: impl FnOnce( + sea_orm::UpdateMany, + ) -> sea_orm::UpdateMany, + ) { + use database_artifact::Column; + let update = edit(database_artifact::Entity::update_many()) + .filter(Column::Service.eq("sql")) + .filter(Column::Space.eq(space)) + .filter(Column::Name.eq(name)) + .exec(conn) + .await + .unwrap(); + assert_eq!(update.rows_affected, 1); + } + + /// The WAL delta of a sealed row is dropped in the database: delta + /// columns cleared, the unkeyed logical hash and size reset to the + /// checkpoint's, revision kept. A cold read must fail as integrity, never + /// hand back the checkpoint alone (which would silently lose the rows the + /// delta committed). + #[tokio::test] + async fn a_dropped_wal_delta_fails_integrity_instead_of_a_partial_read() { + for mode in [AtRestMode::Encrypt, AtRestMode::Strict] { + let conn = sqlite_conn().await; + let (repo, _source) = at_rest_repo(&conn, mode); + let checkpoint = database_bytes(SQLITE_HEADER, 70_000, 21); + let wal = database_bytes(WAL_HEADER, 9_000, 22); + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + repo.save_delta("sql", "space", "main", wal, ArtifactExpectation::Any) + .await + .unwrap(); + let row = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!(row.revision, 2); + rewrite_row(&conn, "space", "main", |update| { + use database_artifact::Column; + update + .col_expr(Column::DeltaPayload, Expr::value(Option::>::None)) + .col_expr(Column::DeltaEncoding, Expr::value(Option::::None)) + .col_expr( + Column::DeltaContentHash, + Expr::value(Option::::None), + ) + .col_expr(Column::DeltaSizeBytes, Expr::value(0_i64)) + .col_expr(Column::ContentHash, Expr::value(cid(&checkpoint))) + .col_expr(Column::SizeBytes, Expr::value(checkpoint.len() as i64)) + .col_expr(Column::StorageMode, Expr::value("database-blob")) + }) + .await; + // A cold read: a repository that never saw the row. + let (cold, _source) = at_rest_repo(&conn, mode); + let error = read_error(&cold, "space", "main").await; + assert!( + matches!( + error, + DatabaseArtifactError::AtRestIntegrity { part: "manifest" } + ), + "{mode:?}: {error:?}" + ); + // Clearing the manifest tag as well makes it a row with a sealed + // part and no tag, which only a tampered row is. + rewrite_row(&conn, "space", "main", |update| { + update.col_expr( + database_artifact::Column::Manifest, + Expr::value(Option::>::None), + ) + }) + .await; + let error = read_error(&cold, "space", "main").await; + assert!( + matches!( + error, + DatabaseArtifactError::AtRestIntegrity { part: "manifest" } + ), + "{mode:?}, no tag: {error:?}" + ); + // The offline whole-row read refuses it too. + let row = stored_row(&conn, "space", "main").await.unwrap(); + assert!(matches!( + open_row(row, Some(keys().as_ref())).await, + Err(DatabaseArtifactError::AtRestIntegrity { part: "manifest" }) + )); + } + } + + /// Parts and revisions from another point of the row's history do not + /// read in place of the current ones. + #[tokio::test] + async fn a_part_or_revision_from_another_revision_fails_integrity() { + let conn = sqlite_conn().await; + let (repo, _source) = at_rest_repo(&conn, AtRestMode::Encrypt); + let first = database_bytes(SQLITE_HEADER, 70_000, 31); + let second = database_bytes(SQLITE_HEADER, 80_000, 32); + repo.save( + "sql", + "space", + "main", + first.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + let old = stored_row(&conn, "space", "main").await.unwrap(); + repo.save_delta( + "sql", + "space", + "main", + database_bytes(WAL_HEADER, 4_000, 33), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + repo.save( + "sql", + "space", + "main", + second.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let current = stored_row(&conn, "space", "main").await.unwrap(); + assert_eq!((old.revision, current.revision), (1, 3)); + + // Revision 1's checkpoint and its metadata, at revision 3. + rewrite_row(&conn, "space", "main", |update| { + use database_artifact::Column; + update + .col_expr(Column::Payload, Expr::value(old.payload.clone())) + .col_expr(Column::ContentHash, Expr::value(old.content_hash.clone())) + .col_expr( + Column::CheckpointContentHash, + Expr::value(old.checkpoint_content_hash.clone()), + ) + .col_expr(Column::SizeBytes, Expr::value(old.size_bytes)) + .col_expr( + Column::CheckpointSizeBytes, + Expr::value(old.checkpoint_size_bytes), + ) + }) + .await; + let error = read_error(&repo, "space", "main").await; + assert!( + matches!( + error, + DatabaseArtifactError::AtRestIntegrity { part: "manifest" } + ), + "older checkpoint: {error:?}" + ); + + // The current parts under another revision number. + put_parts(&conn, "space", "main", ¤t).await; + rewrite_row(&conn, "space", "main", |update| { + use database_artifact::Column; + update + .col_expr( + Column::ContentHash, + Expr::value(current.content_hash.clone()), + ) + .col_expr( + Column::CheckpointContentHash, + Expr::value(current.checkpoint_content_hash.clone()), + ) + .col_expr(Column::SizeBytes, Expr::value(current.size_bytes)) + .col_expr( + Column::CheckpointSizeBytes, + Expr::value(current.checkpoint_size_bytes), + ) + .col_expr(Column::Revision, Expr::value(7_i64)) + }) + .await; + let error = read_error(&repo, "space", "main").await; + assert!( + matches!( + error, + DatabaseArtifactError::AtRestIntegrity { part: "manifest" } + ), + "another revision: {error:?}" + ); + rewrite_row(&conn, "space", "main", |update| { + update.col_expr(database_artifact::Column::Revision, Expr::value(3_i64)) + }) + .await; + assert_eq!(hydrate(&repo, "space", "main").await, (second, None)); + } + + /// Rows written in mode `off`, and legacy rows from before encryption, + /// carry nothing sealed and still read: legacy parts in `off` and + /// `encrypt`, a legacy checkpoint under a sealed delta, and a sealed + /// checkpoint under a delta written in `off`. + #[tokio::test] + async fn legacy_and_off_rows_still_read() { + let conn = sqlite_conn().await; + let (repo, source) = at_rest_repo(&conn, AtRestMode::Off); + let checkpoint = database_bytes(SQLITE_HEADER, 70_000, 41); + let wal = database_bytes(WAL_HEADER, 6_000, 42); + repo.save( + "sql", + "space", + "legacy", + checkpoint.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + repo.save_delta( + "sql", + "space", + "legacy", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let both = (checkpoint.clone(), Some(wal.clone())); + for (mode, encrypting) in [(AtRestMode::Off, false), (AtRestMode::Encrypt, true)] { + source.set(mode, encrypting); + assert_eq!(hydrate(&repo, "space", "legacy").await, both, "{mode:?}"); + } + + // A legacy checkpoint, then a delta sealed in `encrypt`. + source.set(AtRestMode::Off, false); + repo.save( + "sql", + "space", + "mixed", + checkpoint.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + source.set(AtRestMode::Encrypt, true); + repo.save_delta( + "sql", + "space", + "mixed", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + assert_eq!(hydrate(&repo, "space", "mixed").await, both); + source.set(AtRestMode::Off, false); + assert_eq!(hydrate(&repo, "space", "mixed").await, both); + + // A sealed checkpoint, then a delta written in `off`. + source.set(AtRestMode::Encrypt, true); + repo.save( + "sql", + "space", + "back", + checkpoint.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + source.set(AtRestMode::Off, false); + repo.save_delta( + "sql", + "space", + "back", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let row = stored_row(&conn, "space", "back").await.unwrap(); + assert_eq!((row.payload_encoding, row.delta_encoding), (Some(1), None)); + for (mode, encrypting) in [(AtRestMode::Off, false), (AtRestMode::Encrypt, true)] { + source.set(mode, encrypting); + assert_eq!(hydrate(&repo, "space", "back").await, both, "{mode:?}"); + } + } + + /// A sealed part stores exactly `ciphertext_len(size)` bytes: bytes + /// appended to it, or a part cut short, are integrity failures. + #[tokio::test] + async fn a_sealed_part_of_the_wrong_stored_length_fails_integrity() { + let conn = sqlite_conn().await; + let (repo, _source) = at_rest_repo(&conn, AtRestMode::Encrypt); + let checkpoint = database_bytes(SQLITE_HEADER, 70_000, 51); + let wal = database_bytes(WAL_HEADER, 5_000, 52); + repo.save( + "sql", + "space", + "main", + checkpoint.clone(), + ArtifactExpectation::Absent, + ) + .await + .unwrap(); + repo.save_delta( + "sql", + "space", + "main", + wal.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + let good = stored_row(&conn, "space", "main").await.unwrap(); + type Edit = fn(&mut database_artifact::Model); + let edits: [(&str, &str, Edit); 3] = [ + ("bytes appended to the checkpoint", "checkpoint", |row| { + row.payload.extend_from_slice(&[0x5a; 37]) + }), + ("bytes appended to the wal", "wal", |row| { + row.delta_payload.as_mut().unwrap().push(0) + }), + ("checkpoint cut short", "checkpoint", |row| { + row.payload.pop(); + }), + ]; + for (what, part, edit) in edits { + let mut row = good.clone(); + edit(&mut row); + put_parts(&conn, "space", "main", &row).await; + let error = read_error(&repo, "space", "main").await; + assert!( + matches!(&error, DatabaseArtifactError::AtRestIntegrity { part: p } if *p == part), + "{what}: {error:?}" + ); + let error = open_row(row, Some(keys().as_ref())).await.unwrap_err(); + assert!( + matches!(&error, DatabaseArtifactError::AtRestIntegrity { part: p } if *p == part), + "{what}, whole row: {error:?}" + ); + } + put_parts(&conn, "space", "main", &good).await; + assert_eq!( + hydrate(&repo, "space", "main").await, + (checkpoint, Some(wal)) + ); + } } diff --git a/tinycloud-core/src/database_migration.rs b/tinycloud-core/src/database_migration.rs index 0484761d..f6799334 100644 --- a/tinycloud-core/src/database_migration.rs +++ b/tinycloud-core/src/database_migration.rs @@ -11,14 +11,16 @@ use sea_orm::{ sea_query::{Expr, Query}, ActiveModelTrait, ActiveValue::Set, - ColumnTrait, ConnectionTrait, DatabaseConnection, DbErr, EntityTrait, QueryFilter, QuerySelect, - TransactionTrait, + ColumnTrait, ConnectionTrait, DatabaseConnection, DbErr, EntityTrait, IdenStatic, Iterable, + QueryFilter, QuerySelect, TransactionTrait, }; use sea_orm_migration::SchemaManager; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use crate::{ + at_rest::DataKeyProvider, + database_artifacts::{open_row, DatabaseArtifactError}, database_identity::{legacy_duckdb_name, legacy_sql_name, logical_name}, models::{ abilities, database_alias, database_artifact, database_identity_fence, @@ -50,6 +52,8 @@ pub enum MigrationError { Io(#[from] std::io::Error), #[error("checkpoint failed: {0}")] Checkpoint(String), + #[error("artifact is sealed at rest and could not be opened: {0}")] + SealedArtifact(#[source] DatabaseArtifactError), } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -279,26 +283,59 @@ async fn fingerprint( /// Reads artifact bytes only for an explicitly designated local snapshot. /// Production-facing inventory, apply, report, and verify never call this. +/// A part sealed at rest is opened with `keys`; without them it is refused, +/// never fingerprinted as ciphertext. `encoded` says whether the snapshot +/// has the encoding columns; one taken before them holds only plaintext. async fn offline_fingerprint( conn: &C, datadir: &Path, service: &str, space: &str, name: &str, + encoded: bool, + keys: Option<&dyn DataKeyProvider>, ) -> Result, MigrationError> { - let row = database_artifact::Entity::find_by_id(( + use database_artifact::Column; + let mut select = database_artifact::Entity::find_by_id(( service.to_owned(), space.to_owned(), name.to_owned(), )) - .one(conn) - .await?; + .select_only() + .columns(Column::iter().filter(|column| { + !matches!( + column, + Column::PayloadEncoding | Column::DeltaEncoding | Column::Manifest + ) + })); + for column in [Column::PayloadEncoding, Column::DeltaEncoding] { + select = if encoded { + select.column(column) + } else { + select.expr_as(Expr::value(Option::::None), column.as_str()) + }; + } + select = if encoded { + select.column(Column::Manifest) + } else { + select.expr_as( + Expr::value(Option::>::None), + Column::Manifest.as_str(), + ) + }; + let row = select + .into_model::() + .one(conn) + .await?; let temp = tempfile::tempdir()?; let suffix = if service == "sql" { ".db" } else { ".duckdb" }; let path = temp.path().join(format!("artifact{suffix}")); if let Some(row) = row { - std::fs::write(&path, row.payload)?; - if let Some(delta) = row.delta_payload { + let (payload, delta) = open_row(row, keys) + .await + .map_err(MigrationError::SealedArtifact)?; + std::fs::write(&path, payload)?; + if let Some(delta) = delta { let wal = if service == "sql" { format!("{}-wal", path.display()) } else { @@ -341,11 +378,17 @@ async fn offline_fingerprint( /// local snapshot. The caller must enforce that boundary before invoking it. /// DuckDB snapshots undergo the same locked native policy and stored-code audit /// as node connections, while retaining the offline tool's native memory default. +/// `keys` opens artifacts sealed at rest; with `None` a sealed artifact is an +/// error. pub async fn offline_inventory( conn: &DatabaseConnection, datadir: &Path, + keys: Option<&dyn DataKeyProvider>, ) -> Result, MigrationError> { let mut items = inventory(conn, datadir).await?; + let encoded = SchemaManager::new(conn) + .has_column("database_artifact", "payload_encoding") + .await?; for item in &mut items { item.offline_fingerprint = offline_fingerprint( conn, @@ -353,6 +396,8 @@ pub async fn offline_inventory( &item.service, &item.space, &item.physical_name, + encoded, + keys, ) .await?; } @@ -1345,6 +1390,83 @@ mod tests { use sea_orm::QueryTrait; + /// TC-887: the offline tool fingerprints the database a sealed row holds + /// when it has keys, and refuses the row without them instead of reading + /// ciphertext as a database. + #[tokio::test] + async fn offline_fingerprint_opens_sealed_rows_only_with_keys() { + use crate::at_rest::{AtRestMode, DataKeyProvider, FixedKeyProvider, TestAtRestSource}; + use crate::database_artifacts::{ + ArtifactExpectation, DatabaseArtifactRepository, SeaOrmDatabaseArtifactRepository, + }; + use crate::migrations::Migrator; + use sea_orm_migration::MigratorTrait; + use std::sync::Arc; + + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("source.db"); + rusqlite::Connection::open(&file) + .unwrap() + .execute_batch( + "CREATE TABLE items (id INTEGER PRIMARY KEY, body TEXT); + INSERT INTO items (body) VALUES ('a'), ('b');", + ) + .unwrap(); + let payload = std::fs::read(&file).unwrap(); + let expected = fingerprint_sqlite(&file).unwrap(); + + let conn = sea_orm::Database::connect("sqlite::memory:").await.unwrap(); + Migrator::up(&conn, None).await.unwrap(); + let keys: Arc = Arc::new(FixedKeyProvider::new([3; 32])); + SeaOrmDatabaseArtifactRepository::new(conn.clone()) + .with_at_rest(TestAtRestSource::new( + AtRestMode::Encrypt, + true, + Some(keys.clone()), + )) + .save( + "sql", + "space", + "sealed", + payload.clone(), + ArtifactExpectation::Any, + ) + .await + .unwrap(); + SeaOrmDatabaseArtifactRepository::new(conn.clone()) + .save("sql", "space", "legacy", payload, ArtifactExpectation::Any) + .await + .unwrap(); + + let datadir = dir.path().join("data"); + assert_eq!( + offline_fingerprint( + &conn, + &datadir, + "sql", + "space", + "sealed", + true, + Some(keys.as_ref()) + ) + .await + .unwrap(), + Some(expected.clone()) + ); + assert!(matches!( + offline_fingerprint(&conn, &datadir, "sql", "space", "sealed", true, None).await, + Err(MigrationError::SealedArtifact( + DatabaseArtifactError::AtRestUnavailable + )) + )); + assert_eq!( + offline_fingerprint(&conn, &datadir, "sql", "space", "legacy", true, None) + .await + .unwrap(), + Some(expected) + ); + } + #[test] fn production_inventory_queries_never_select_artifact_payloads() { for backend in [ @@ -1398,6 +1520,9 @@ mod tests { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(None), + delta_encoding: Set(None), + manifest: Set(None), }; let legacy = |space: &str, name: &str| database_legacy_artifact::ActiveModel { service: Set("sql".into()), @@ -1617,6 +1742,9 @@ mod tests { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(None), + delta_encoding: Set(None), + manifest: Set(None), } })) .exec(conn) @@ -1877,6 +2005,9 @@ mod tests { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(None), + delta_encoding: Set(None), + manifest: Set(None), } .insert(&conn) .await @@ -2009,6 +2140,9 @@ mod tests { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(None), + delta_encoding: Set(None), + manifest: Set(None), } } diff --git a/tinycloud-core/src/duckdb/service.rs b/tinycloud-core/src/duckdb/service.rs index ac88d42d..1b2e7f71 100644 --- a/tinycloud-core/src/duckdb/service.rs +++ b/tinycloud-core/src/duckdb/service.rs @@ -659,10 +659,13 @@ fn duckdb_wal_path(path: &Path) -> PathBuf { PathBuf::from(format!("{}.wal", path.display())) } -/// `DuckDbError` has no retryable-unavailable variant (adding one would reach -/// every route mapping), so an exhausted `ArtifactChanged` stays `Internal`. +/// An exhausted `ArtifactChanged` stays `Internal`, as before. Unavailable +/// at-rest keys (TC-887) are `Unavailable`: nothing was written or read. fn artifact_error_to_duckdb(err: DatabaseArtifactError) -> DuckDbError { - DuckDbError::Internal(err.to_string()) + match err { + DatabaseArtifactError::AtRestUnavailable => DuckDbError::Unavailable, + err => DuckDbError::Internal(err.to_string()), + } } #[cfg(test)] @@ -1088,6 +1091,101 @@ mod tests { ); } + /// TC-887: a sealed DuckDB artifact holds no DuckDB bytes or row data, + /// a cold cache hydrates and exports the plaintext database, and locked + /// keys answer `Unavailable` (503) rather than serving anything. + #[tokio::test] + async fn sealed_duckdb_artifact_hydrates_a_cold_cache() { + use crate::at_rest::{AtRestMode, DataKeyProvider, FixedKeyProvider, TestAtRestSource}; + use crate::models::database_artifact; + use sea_orm::EntityTrait; + + let db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); + let keys: Arc = Arc::new(FixedKeyProvider::new([5; 32])); + let source = TestAtRestSource::new(AtRestMode::Encrypt, true, Some(keys)); + let repo = Arc::new( + SeaOrmDatabaseArtifactRepository::new(db.clone()).with_at_rest(source.clone()), + ); + let space = test_space_id("duckdb-sealed"); + let marker = "tc887-sealed-duckdb-marker"; + let query = || DuckDbRequest::Query { + sql: "SELECT name FROM events ORDER BY id".to_string(), + params: Vec::new(), + }; + + let cache_one = TempDir::new().unwrap(); + service(&cache_one, repo.clone()) + .execute( + &space, + "analytics", + DuckDbRequest::Execute { + schema: Some(vec![ + "CREATE TABLE events (id INTEGER, name VARCHAR)".to_string() + ]), + sql: format!("INSERT INTO events VALUES (1, '{marker}')"), + params: Vec::new(), + }, + None, + "tinycloud.duckdb/write".to_string(), + false, + ) + .await + .unwrap(); + + let row = database_artifact::Entity::find() + .one(&db) + .await + .unwrap() + .unwrap(); + assert_eq!(row.payload_encoding, Some(1)); + assert!(row.payload.starts_with(b"TCE1")); + assert_ne!(row.payload.get(8..12), Some(&b"DUCK"[..])); + assert!(!row + .payload + .windows(marker.len()) + .any(|window| window == marker.as_bytes())); + + let cache_two = TempDir::new().unwrap(); + let recreated = service(&cache_two, repo.clone()); + let result = recreated + .execute( + &space, + "analytics", + query(), + None, + "tinycloud.duckdb/read".to_string(), + false, + ) + .await + .unwrap(); + match result.response { + DuckDbResponse::Query(rows) => { + assert_eq!(rows.rows, vec![vec![DuckDbValue::Text(marker.to_string())]]); + } + other => panic!("expected query response, got {other:?}"), + } + let exported = recreated.export(&space, "analytics").await.unwrap(); + assert_eq!(exported.get(8..12), Some(&b"DUCK"[..])); + drop(recreated); + + source.set_provider(Some(Arc::new(FixedKeyProvider::new([5; 32]).locked()))); + let cache_three = TempDir::new().unwrap(); + let read = service(&cache_three, repo) + .execute( + &space, + "analytics", + query(), + None, + "tinycloud.duckdb/read".to_string(), + false, + ) + .await; + assert!(matches!(read, Err(DuckDbError::Unavailable)), "{read:?}"); + } + #[tokio::test] async fn duckdb_file_backed_small_writes_persist_wal_not_full_database() { let repo = artifact_repository().await; diff --git a/tinycloud-core/src/duckdb/types.rs b/tinycloud-core/src/duckdb/types.rs index 4ea020b6..af7c0ca5 100644 --- a/tinycloud-core/src/duckdb/types.rs +++ b/tinycloud-core/src/duckdb/types.rs @@ -477,6 +477,10 @@ pub enum DuckDbError { ImportError(String), #[error("Internal error: {0}")] Internal(String), + /// At-rest keys the durable artifact needs are unavailable now (TC-887); + /// nothing was written. Answered 503. + #[error("Database temporarily unavailable")] + Unavailable, } #[cfg(test)] diff --git a/tinycloud-core/src/migrations/m20261107_010000_artifact_payload_encoding.rs b/tinycloud-core/src/migrations/m20261107_010000_artifact_payload_encoding.rs new file mode 100644 index 00000000..862771bf --- /dev/null +++ b/tinycloud-core/src/migrations/m20261107_010000_artifact_payload_encoding.rs @@ -0,0 +1,60 @@ +//! TC-887: how each stored part of a `database_artifact` row is encoded, and +//! the row's manifest tag. Encoding `NULL` is legacy plaintext, so every +//! existing row reads as before; `1` is TCE1 (`docs/at-rest/INTERFACES.md` +//! §5). `manifest` is `NULL` on every existing row and on rows mode `off` +//! writes (§8). + +use sea_orm_migration::prelude::*; + +use crate::models::database_artifact; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + for column in [ + ColumnDef::new(database_artifact::Column::PayloadEncoding) + .small_integer() + .null() + .to_owned(), + ColumnDef::new(database_artifact::Column::DeltaEncoding) + .small_integer() + .null() + .to_owned(), + ColumnDef::new(database_artifact::Column::Manifest) + .blob() + .null() + .to_owned(), + ] { + manager + .alter_table( + Table::alter() + .table(database_artifact::Entity) + .add_column(column) + .to_owned(), + ) + .await?; + } + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + for column in [ + database_artifact::Column::Manifest, + database_artifact::Column::DeltaEncoding, + database_artifact::Column::PayloadEncoding, + ] { + manager + .alter_table( + Table::alter() + .table(database_artifact::Entity) + .drop_column(column) + .to_owned(), + ) + .await?; + } + Ok(()) + } +} diff --git a/tinycloud-core/src/migrations/mod.rs b/tinycloud-core/src/migrations/mod.rs index 5a4e29f9..1156ae24 100644 --- a/tinycloud-core/src/migrations/mod.rs +++ b/tinycloud-core/src/migrations/mod.rs @@ -34,6 +34,7 @@ pub mod m20261022_000000_encryption_network_key; pub mod m20261023_010000_kv_orphan_candidates; pub mod m20261026_000000_kv_write_space_value_index; pub mod m20261107_000000_at_rest_keys; +pub mod m20261107_010000_artifact_payload_encoding; pub struct Migrator; @@ -76,6 +77,7 @@ impl MigratorTrait for Migrator { Box::new(m20261023_010000_kv_orphan_candidates::Migration), Box::new(m20261026_000000_kv_write_space_value_index::Migration), Box::new(m20261107_000000_at_rest_keys::Migration), + Box::new(m20261107_010000_artifact_payload_encoding::Migration), ] } } diff --git a/tinycloud-core/src/models/database_artifact.rs b/tinycloud-core/src/models/database_artifact.rs index b0e27ba5..d9216173 100644 --- a/tinycloud-core/src/models/database_artifact.rs +++ b/tinycloud-core/src/models/database_artifact.rs @@ -22,6 +22,15 @@ pub struct Model { pub delta_payload: Option>, pub delta_content_hash: Option, pub delta_size_bytes: i64, + /// How `payload` is stored: `NULL` legacy plaintext, `1` TCE1 + /// (`at_rest::PayloadEncoding`). + pub payload_encoding: Option, + /// How `delta_payload` is stored, like `payload_encoding`. + pub delta_encoding: Option, + /// The row's manifest tag: a TCE1 seal of nothing, bound to the row's + /// revision, hashes, sizes, encodings and whether it has a delta + /// (`docs/at-rest/INTERFACES.md` §8). `NULL` on rows mode `off` wrote. + pub manifest: Option>, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/tinycloud-core/src/sql/service.rs b/tinycloud-core/src/sql/service.rs index 6cfc3860..10052ca3 100644 --- a/tinycloud-core/src/sql/service.rs +++ b/tinycloud-core/src/sql/service.rs @@ -1059,6 +1059,8 @@ fn unavailable_kind(err: &DatabaseArtifactError) -> Option<&'static str> { match err { DatabaseArtifactError::Db(error) if is_transient_error(error) => Some(failure_kind(error)), DatabaseArtifactError::ArtifactChanged => Some("artifact_changed"), + // No class of its own: the diagnostic classes are a fixed set. + DatabaseArtifactError::AtRestUnavailable => Some("other"), _ => None, } } @@ -1121,6 +1123,10 @@ fn artifact_error_to_sql(err: DatabaseArtifactError) -> SqlError { ); SqlError::Unavailable } + // At-rest keys locked or the encrypt gate closed (TC-887): nothing + // was written, so a later request may succeed. Integrity, unsupported + // format and strict-mode refusals stay `Internal` below. + DatabaseArtifactError::AtRestUnavailable => SqlError::Unavailable, err => SqlError::Internal(err.to_string()), } } @@ -2698,6 +2704,154 @@ mod tests { } } + /// TC-887: in `encrypt` mode the durable copy of a SQL database (its + /// checkpoint and its WAL delta) holds no SQLite bytes and no row data, + /// and a cold cache hydrates the same rows from it. Without usable keys + /// a read or a write answers `Unavailable` (503) and stores nothing. + #[tokio::test] + async fn sealed_sql_database_hydrates_a_cold_cache() { + use crate::at_rest::{AtRestMode, DataKeyProvider, FixedKeyProvider, TestAtRestSource}; + use crate::models::database_artifact; + use sea_orm::EntityTrait; + + let db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); + let keys: Arc = Arc::new(FixedKeyProvider::new([9; 32])); + let source = TestAtRestSource::new(AtRestMode::Encrypt, true, Some(keys)); + let repo = Arc::new( + SeaOrmDatabaseArtifactRepository::new(db.clone()).with_at_rest(source.clone()), + ); + let space = test_space_id("sql-sealed"); + let marker = "tc887-sealed-row-marker"; + let query = || SqlRequest::Query { + sql: "SELECT name FROM items ORDER BY id".to_string(), + params: Vec::new(), + max_rows: None, + max_bytes: None, + }; + + let cache_one = TempDir::new().unwrap(); + let service = SqlService::new( + cache_one.path().to_string_lossy().to_string(), + 0, + repo.clone(), + ); + for (sql, ability) in [ + ( + "CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT NOT NULL)".to_string(), + "tinycloud.sql/schema", + ), + ( + format!("INSERT INTO items (name) VALUES ('{marker}')"), + "tinycloud.sql/admin", + ), + ] { + service + .execute( + &space, + "main", + SqlRequest::Execute { + schema: None, + sql, + params: Vec::new(), + }, + None, + ability.to_string(), + ) + .await + .unwrap(); + } + drop(service); + + let row = database_artifact::Entity::find() + .one(&db) + .await + .unwrap() + .unwrap(); + assert_eq!(row.storage_mode, "checkpoint+wal"); + assert_eq!( + (row.payload_encoding, row.delta_encoding), + (Some(1), Some(1)) + ); + for stored in [&row.payload, row.delta_payload.as_ref().unwrap()] { + assert!(!stored.starts_with(SQLITE_MAGIC)); + assert!(!stored + .windows(marker.len()) + .any(|window| window == marker.as_bytes())); + } + + let cache_two = TempDir::new().unwrap(); + let recreated = SqlService::new( + cache_two.path().to_string_lossy().to_string(), + 0, + repo.clone(), + ); + let result = recreated + .execute( + &space, + "main", + query(), + None, + "tinycloud.sql/read".to_string(), + ) + .await + .unwrap(); + match result.response { + SqlResponse::Query(rows) => { + assert_eq!(rows.rows, vec![vec![SqlValue::Text(marker.to_string())]]); + } + other => panic!("expected query response, got {other:?}"), + } + drop(recreated); + + // Locked keys: the sealed database is not served. + source.set_provider(Some(Arc::new(FixedKeyProvider::new([9; 32]).locked()))); + let cache_three = TempDir::new().unwrap(); + let locked = SqlService::new( + cache_three.path().to_string_lossy().to_string(), + 0, + repo.clone(), + ); + let read = locked + .execute( + &space, + "main", + query(), + None, + "tinycloud.sql/read".to_string(), + ) + .await; + assert!(matches!(read, Err(SqlError::Unavailable)), "{read:?}"); + // The gate closed: a new database's first write is refused, not + // stored as plaintext. + source.set(AtRestMode::Encrypt, false); + let write = locked + .execute( + &space, + "fresh", + SqlRequest::Execute { + schema: None, + sql: "CREATE TABLE t (id INTEGER)".to_string(), + params: Vec::new(), + }, + None, + "tinycloud.sql/schema".to_string(), + ) + .await; + assert!(matches!(write, Err(SqlError::Unavailable)), "{write:?}"); + assert_eq!( + database_artifact::Entity::find() + .all(&db) + .await + .unwrap() + .len(), + 1, + "nothing stored for the refused write" + ); + } + #[test] fn hydration_temp_paths_never_collide_between_checkpoint_and_wal() { let database = PathBuf::from("/cache/space/main.db"); diff --git a/tinycloud-node-server/src/at_rest.rs b/tinycloud-node-server/src/at_rest.rs index cfe22104..b056e521 100644 --- a/tinycloud-node-server/src/at_rest.rs +++ b/tinycloud-node-server/src/at_rest.rs @@ -6,9 +6,10 @@ //! [`setup`] after migrations installs the keyring. //! //! Storage declares its seal paths ([`AtRestRuntime::declare_seal_path`]): -//! block storage when it is attached ([`crate::storage::attach_at_rest`]). -//! Until every seal path is declared, `setup` refuses every mode but `off` -//! and `encrypting()` is always false. +//! block storage when it is attached ([`crate::storage::attach_at_rest`]), +//! and SQL and DuckDB artifacts in `node_artifact_repository`. Until every +//! seal path is declared, `setup` refuses every mode but `off` and +//! `encrypting()` is always false. //! //! With the default config (`mode = "off"`, `root_source = "escrow"`, no //! recipients, no `escrow_none`) nothing is set up: no key reads, no rows, no @@ -63,7 +64,7 @@ pub const PRODUCTION_OS_IMAGE_HASHES: &[&str] = /// both KV block storage and SQL artifacts ([`SealPath`]). `--validate-config` /// decides the seal-path refusal from this; startup checks that the seal /// paths storage actually declared agree with it. -pub const STORAGE_SEALS_EVERY_WRITE: bool = false; +pub const STORAGE_SEALS_EVERY_WRITE: bool = true; /// The root variables as read from the process environment, held only in /// zeroizing buffers. Debug is redacted. @@ -363,8 +364,8 @@ pub enum SealPath { SqlArtifacts, } -/// At-rest state shared by the routes, `/info` and block storage. Rocket -/// manages it as `Arc`. +/// At-rest state shared by the routes, `/info`, block storage and the SQL +/// artifact repository. Rocket manages it as `Arc`. /// /// The contract for the storage seal paths (TC-887): mode `off` behaves /// exactly as without at-rest. Otherwise writes are sealed only while diff --git a/tinycloud-node-server/src/bin/tinycloud-sql-identity.rs b/tinycloud-node-server/src/bin/tinycloud-sql-identity.rs index 04966f21..9f9a68ae 100644 --- a/tinycloud-node-server/src/bin/tinycloud-sql-identity.rs +++ b/tinycloud-node-server/src/bin/tinycloud-sql-identity.rs @@ -32,6 +32,7 @@ enum Command { /// Preview classification and collisions on a copied data directory. DryRun, /// Inspect schema and row counts from a local snapshot, including artifact bytes. + /// The tool holds no at-rest keys, so an artifact sealed at rest is refused. OfflineFingerprint { /// Explicitly attest that this database is a local snapshot or copy. #[arg(long)] @@ -179,7 +180,7 @@ async fn main() -> anyhow::Result<()> { migration::validate_inventory(&items)?; } Command::OfflineFingerprint { .. } => { - let items = migration::offline_inventory(&conn, &args.datadir).await?; + let items = migration::offline_inventory(&conn, &args.datadir, None).await?; println!("{}", serde_json::to_string_pretty(&items)?); } Command::Fence { state } => { diff --git a/tinycloud-node-server/src/invocation_replay/write_snapshot_tests.rs b/tinycloud-node-server/src/invocation_replay/write_snapshot_tests.rs index efbdf033..3bd368f4 100644 --- a/tinycloud-node-server/src/invocation_replay/write_snapshot_tests.rs +++ b/tinycloud-node-server/src/invocation_replay/write_snapshot_tests.rs @@ -104,7 +104,10 @@ async fn race(competitor: Competitor, begin: Begin) -> (Result<(), DbErr>, bool) match competitor { Competitor::GatedReplayInsert | Competitor::GatedArtifactSave => ( crate::node_replay_cache(&node), - Box::new(crate::node_artifact_repository(&node)), + Box::new(crate::node_artifact_repository( + &node, + &crate::at_rest::AtRestRuntime::new(&crate::config::AtRestConfig::default()), + )), ), _ => ( InvocationReplayCache::new(node.connection().clone()), diff --git a/tinycloud-node-server/src/lib.rs b/tinycloud-node-server/src/lib.rs index bb1a582a..df09a2ba 100644 --- a/tinycloud-node-server/src/lib.rs +++ b/tinycloud-node-server/src/lib.rs @@ -553,9 +553,16 @@ fn node_replay_cache(tinycloud: &TinyCloud) -> InvocationReplayCache { .with_sqlite_writer_lock(tinycloud.sqlite_writer_lock()) } -fn node_artifact_repository(tinycloud: &TinyCloud) -> SeaOrmDatabaseArtifactRepository { +/// The node's artifact repository. It seals SQL and DuckDB artifacts by the +/// node's at-rest state, so building it declares that seal path (TC-887). +fn node_artifact_repository( + tinycloud: &TinyCloud, + at_rest: &Arc, +) -> SeaOrmDatabaseArtifactRepository { + at_rest.declare_seal_path(at_rest::SealPath::SqlArtifacts); SeaOrmDatabaseArtifactRepository::new(tinycloud.connection().clone()) .with_sqlite_writer_lock(tinycloud.sqlite_writer_lock()) + .with_at_rest(at_rest.clone()) } pub async fn app(config: &Figment) -> Result> { @@ -973,7 +980,7 @@ pub async fn app_with_control( ]); let database_artifact_repository: Arc = Arc::new(SizeTrackingArtifactRepository::new( - Arc::new(node_artifact_repository(&tinycloud)), + Arc::new(node_artifact_repository(&tinycloud, &at_rest_runtime)), sql_sizes.clone(), )); @@ -1885,7 +1892,10 @@ mod sqlite_tuning_tests { .await .unwrap(); // Exercise the production repository wiring as well as its write path. - let repository = node_artifact_repository(&node); + let repository = node_artifact_repository( + &node, + &at_rest::AtRestRuntime::new(&config::AtRestConfig::default()), + ); let actor_id = "did:key:artifact-snapshot-fixture"; actor::ActiveModel { id: Set(actor_id.to_owned()), diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 086f9b21..e87e2421 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -4104,6 +4104,7 @@ fn duckdb_error_to_status(err: &DuckDbError) -> Status { DuckDbError::ExportError(_) => Status::InternalServerError, DuckDbError::ImportError(_) => Status::InternalServerError, DuckDbError::Internal(_) => Status::InternalServerError, + DuckDbError::Unavailable => Status::ServiceUnavailable, } } diff --git a/tinycloud-node-server/src/share_email.rs b/tinycloud-node-server/src/share_email.rs index 04e11d9e..e229f9da 100644 --- a/tinycloud-node-server/src/share_email.rs +++ b/tinycloud-node-server/src/share_email.rs @@ -2931,6 +2931,9 @@ mod tests { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), + payload_encoding: Set(None), + delta_encoding: Set(None), + manifest: Set(None), } .insert(&conn) .await diff --git a/tinycloud-node-server/tests/at_rest_ceremony.rs b/tinycloud-node-server/tests/at_rest_ceremony.rs index 0e9de195..4f32bad5 100644 --- a/tinycloud-node-server/tests/at_rest_ceremony.rs +++ b/tinycloud-node-server/tests/at_rest_ceremony.rs @@ -191,17 +191,21 @@ async fn ceremony_gate_and_recovery_through_the_app() { ); drop(client); - // No seal path in this build: encrypt and strict are refused at boot, - // even with the escrow pinned and verified, so nothing ever claims to - // encrypt while storage writes plaintext. + // Pinned and verified: encrypt and strict pass the gate and encrypt. for mode in ["encrypt", "strict"] { - let err = refused( + let client = app( dir.path(), 1, json!({"mode": mode, "escrow": escrow(Some(&efp))}), ) - .await; - assert!(err.contains("no at-rest seal path"), "{err}"); + .await + .unwrap(); + let (_, st) = get_json(&client, "/at-rest/status").await; + assert_eq!( + (st["mode"].clone(), st["encrypting"].clone()), + (json!(mode), json!(true)) + ); + drop(client); } // Pinned and verified, mode off: the gate passes, nothing is encrypted. @@ -358,15 +362,10 @@ async fn misconfigurations_are_refused_at_boot() { ) .await; assert!(err.contains("not both"), "{err}"); - // No seal path: encrypt is refused even on a self-hosted Static root - // that acknowledges having no escrow. - let err = refused( - dir.path(), - 1, - json!({"mode": "encrypt", "escrow_none": true}), - ) - .await; - assert!(err.contains("no at-rest seal path"), "{err}"); + // Encrypt with neither escrow nor the self-host acknowledgement: the + // gate refuses the boot. + let err = refused(dir.path(), 1, json!({"mode": "encrypt"})).await; + assert!(err.contains("no escrow recipients"), "{err}"); // Unknown keys in the section are refused, not ignored. refused(dir.path(), 1, json!({"escrow_nonee": true})).await; } diff --git a/tinycloud-node-server/tests/at_rest_fs_node.rs b/tinycloud-node-server/tests/at_rest_fs_node.rs index 2ea2920d..a5323521 100644 --- a/tinycloud-node-server/tests/at_rest_fs_node.rs +++ b/tinycloud-node-server/tests/at_rest_fs_node.rs @@ -4,9 +4,8 @@ //! An `off` node on the same code stores them as before. //! //! The node boots `encrypt` only once every store that writes data declares -//! a seal path (blocks and SQL artifacts), so the `encrypt` tests here are -//! ignored until the SQL artifact seal path lands; `at_rest_fs_store.rs` -//! covers the same behaviour below the app meanwhile. +//! a seal path (blocks and SQL artifacts); `at_rest_fs_store.rs` covers the +//! same behaviour below the app. use std::path::{Path as FsPath, PathBuf}; @@ -236,7 +235,6 @@ fn values() -> Vec<(String, String)> { /// space creates the space's data key, which must happen before the /// invocation's write lock and transaction (`prepare_write`), not inside them. #[tokio::test] -#[ignore = "needs the SQL artifact seal path too (TC-887)"] async fn an_encrypting_filesystem_node_seals_kv_values_and_serves_plaintext() -> Result<()> { tokio::time::timeout(std::time::Duration::from_secs(120), encrypting_node()) .await @@ -321,7 +319,6 @@ async fn an_off_node_still_stores_plaintext_under_legacy_names() -> Result<()> { /// An invocation the node refuses (spaces it does not host) creates no data /// key and no mirror object. #[tokio::test] -#[ignore = "needs the SQL artifact seal path too (TC-887)"] async fn a_refused_put_creates_no_data_key_or_mirror_object() -> Result<()> { let dir = tempfile::tempdir()?; let client = node(dir.path(), json!({"mode": "encrypt", "escrow_none": true})).await?; @@ -360,7 +357,6 @@ async fn a_refused_put_creates_no_data_key_or_mirror_object() -> Result<()> { /// there, is refused by authorization before any key work: no data key and /// no mirror object for the space. #[tokio::test] -#[ignore = "needs the SQL artifact seal path too (TC-887)"] async fn an_unauthorized_put_to_a_hosted_space_creates_no_data_key() -> Result<()> { let dir = tempfile::tempdir()?; let client = node(dir.path(), json!({"mode": "encrypt", "escrow_none": true})).await?; @@ -391,7 +387,6 @@ async fn an_unauthorized_put_to_a_hosted_space_creates_no_data_key() -> Result<( /// before the lock, so the request completes instead of waiting on the lock /// it holds. #[tokio::test] -#[ignore = "needs the SQL artifact seal path too (TC-887)"] async fn a_mixed_read_and_delete_after_losing_the_data_key_row_completes() -> Result<()> { let dir = tempfile::tempdir()?; let at_rest = json!({"mode": "encrypt", "escrow_none": true}); diff --git a/tinycloud-node-server/tests/duckdb_offline_policy.rs b/tinycloud-node-server/tests/duckdb_offline_policy.rs index 1e8125ce..c555de7a 100644 --- a/tinycloud-node-server/tests/duckdb_offline_policy.rs +++ b/tinycloud-node-server/tests/duckdb_offline_policy.rs @@ -285,7 +285,7 @@ async fn safe_controls(root: &Path) { .unwrap(); drop(fixture); - let items = migration::offline_inventory(&metadata, &datadir) + let items = migration::offline_inventory(&metadata, &datadir, None) .await .unwrap(); assert_eq!(items.len(), 1); @@ -322,7 +322,7 @@ async fn safe_controls(root: &Path) { drop(reopened); persist_snapshot(&metadata, &path).await; std::fs::remove_file(&path).unwrap(); - let items = migration::offline_inventory(&metadata, &datadir) + let items = migration::offline_inventory(&metadata, &datadir, None) .await .unwrap(); assert_eq!(items.len(), 1); @@ -378,7 +378,7 @@ async fn unsafe_controls(root: &Path) { .unwrap(); assert_eq!(copied, 0, "source audit must finish before copying any DDL"); drop(fixture); - let error = migration::offline_inventory(&metadata, &datadir) + let error = migration::offline_inventory(&metadata, &datadir, None) .await .unwrap_err(); assert!(!error.to_string().contains(definition)); @@ -399,7 +399,7 @@ async fn unsafe_controls(root: &Path) { } persist_snapshot(&metadata, &path).await; std::fs::remove_file(&path).unwrap(); - migration::offline_inventory(&metadata, &datadir) + migration::offline_inventory(&metadata, &datadir, None) .await .expect_err("unsafe durable snapshot must be rejected"); let output = cli( diff --git a/tinycloud-node-server/tests/sql_identity_cli.rs b/tinycloud-node-server/tests/sql_identity_cli.rs index a838586d..e43be305 100644 --- a/tinycloud-node-server/tests/sql_identity_cli.rs +++ b/tinycloud-node-server/tests/sql_identity_cli.rs @@ -4,7 +4,10 @@ use tinycloud_auth::{resolver::DID_METHODS, resource::SpaceId, ssi::jwk::JWK}; use tinycloud_core::{ migrations::Migrator, models::database_artifact, - sea_orm::{ActiveModelTrait, ActiveValue::Set, Database}, + sea_orm::{ + ActiveValue::{NotSet, Set}, + Database, EntityTrait, + }, sea_orm_migration::MigratorTrait, }; @@ -45,7 +48,9 @@ async fn cli_fence_dry_run_apply_set_verify_clear() { .execute_batch("CREATE TABLE t (v TEXT); INSERT INTO t VALUES ('existing-web-row');") .unwrap(); drop(sqlite); - database_artifact::ActiveModel { + // Insert only the columns a snapshot of that age has: the newest + // migrations are not applied yet. + database_artifact::Entity::insert(database_artifact::ActiveModel { service: Set("sql".into()), space: Set(space.clone()), name: Set("threads".into()), @@ -62,8 +67,11 @@ async fn cli_fence_dry_run_apply_set_verify_clear() { delta_payload: Set(None), delta_content_hash: Set(None), delta_size_bytes: Set(0), - } - .insert(&conn) + payload_encoding: NotSet, + delta_encoding: NotSet, + manifest: NotSet, + }) + .exec_without_returning(&conn) .await .unwrap(); drop(conn);