diff --git a/docs/invoke-query-budget.md b/docs/invoke-query-budget.md new file mode 100644 index 00000000..a8cc8380 --- /dev/null +++ b/docs/invoke-query-budget.md @@ -0,0 +1,149 @@ +# `/invoke` per-request DB statement budget (TC-411) + +This document records the exact SQL statement budgets enforced for a single +`/invoke` request. The budgets are checked in as counting-seam tests +(`tinycloud-core/src/auth_graph.rs`, `tinycloud-core/src/db.rs`); a regression +in any of these numbers is a test failure, not just a benchmark regression. + +## Authorization graph load + +Before TC-411, `validate` re-walked the proof closure once for chain-lock key +derivation and again (per ancestor) for revocation checks and chain-window +validation. TC-411 builds one invocation-scoped `AuthGraphSnapshot` after the +shared chain guards are acquired: + +1. Derive the guarded closure on the guarded connection and require it to + match the pre-guard `lock_keys` **exactly** (`AuthGraphSnapshot::load_guarded`). + A mismatch is treated the same as a database failure: fail closed. +2. Batch-load, one query each, for every node in the bounded closure (not + just the cited proof roots): delegation rows, ability/caveat rows, and + revocation rows. +3. Run all chain, revocation, and caveat-containment checks against that one + in-memory snapshot for the rest of the request. + +Ability/caveat rows are loaded for the whole closure — cited roots *and* +their ancestors — because `constrained_statement_caveat_candidates` walks +each root's ancestor chain looking for a caveat. An ancestor-only caveat (the +descendant delegation carries none of its own) must still be visible, and it +is already part of the bounded, already-loaded closure, so this costs no +extra statement. + +### Structural statement counts + +| Depth | Closure query | Delegation | Ability | Revocation | Total | +|-------|---------------|------------|---------|------------|-------| +| 0 (no proof) | 0 | 0 | 0 | 0 | **0** | +| 1 (delegated) | pre-guard + guarded (2) | 1 | 1 | 1 | **5** | +| 4 (delegated)| pre-guard + guarded (2) | 1 | 1 | 1 | **5** | + +Depth 4 equals depth 1: statement count is independent of chain depth +because every node in the closure is loaded in one `IN (...)` query per +table, not one query per ancestor. Counts are asserted directly in +`auth_graph::tests::snapshot_depths_zero_one_and_four_match_per_node_traversal` +and `auth_graph::tests::snapshot_query_batches_are_bounded_versus_depth_amplified_traversal`. + +The closure itself is capped at `MAX_CHAIN_TRAVERSAL_NODES`; traversal beyond +that limit is rejected (`ChainTraversalError::LimitExceeded`), never +truncated, so these batched queries stay bounded in size regardless of chain +shape. + +## KV operation statement shapes + +These counts cover the request body only (pool acquisition, transaction +begin/body, closure/graph load, replay, and audit remain as before — see +below — and are not re-counted here): + +| Operation | Index/read work | Notes | +|-----------|------------------|-------| +| `kv/get` | 1 statement | Single current-state read; batched via `batch_get_kv_entities` so an N-item batch of `get`/`metadata` capabilities in one invocation still costs 1 statement, not N. | +| `kv/head` (metadata) | 1 statement | Same batched read path as `get`, object body not fetched from block store. | +| `kv/list` | 1 statement | Single bounded index scan regardless of result page size. | +| `kv/put` | graph load + 1 persistence | Object bytes are written to the object store *before* the DB transaction begins (see "Transaction boundaries" below); history (`kv_write`) and projection (`current_kv`) persistence for every put in the invocation is batched into exactly 2 statements independent of item count (`invocation::save`, via `kv_write::Entity::insert_many` + `upsert_current_kv_batch`). | +| `kv/delete` | 0 extra statements when reused | `db.rs` already loads the current `current_kv` row (including its owning `invocation` id) for the precondition/version check; that id is threaded through `VersionedOperation::KvDelete::deleted_invocation_id` and `invocation::resolve_deleted_invocation_id` returns it directly instead of re-querying `kv_write`. Falls back to one lookup only when deleting a key with no live current row (never written, or already deleted). | + +### Batch and multipart statement counts + +| Operation | Statement count | Test | +|-----------|------------------|------| +| Batch get/head, 1 item | 1 statement | `db::test::batch_get_kv_entities_issues_one_statement_regardless_of_item_count` | +| Batch get/head, 100 items | 1 statement | same test | +| Batch put (multipart history/projection), 1 item | 2 statements | `models::invocation::tests::multipart_put_persistence_is_two_statements_regardless_of_item_count` | +| Batch put (multipart history/projection), 100 items | 2 statements | same test | +| `kv/delete` with pre-loaded current state | 0 statements | `models::invocation::tests::delete_reuses_preloaded_invocation_id_without_extra_kv_write_query` | +| `kv/delete` without pre-loaded current state (fallback) | 1 statement | same test | + +Each test wraps a real `sea_orm::DatabaseConnection` with `set_metric_callback` +to count every SQL statement executed, so these are exact counts, not +estimates. + +## Replay and audit + +- **Replay protection** remains exactly one durable, atomic uniqueness + insert attempt performed before any side effect. This is unchanged by + TC-411: timestamps only bound retention and are never the replay decision. +- **Isolated read-audit persistence** is at most four statements per + committed batch. +- `event_spaces` only performs a revocation lookup when the batch actually + contains a `Revocation` event; a batch with none skips the round trip + entirely instead of issuing an empty `IN (...)` query. + +## Transaction boundaries + +No explicit database transaction spans an object-store read/write or +tenant-SQL execution: + +- Immutable blobs are persisted to the object store before publication in + the DB. If the object-store write fails, publication never happens. If the + object store succeeds but the following database write fails, the result + is an unreachable content-addressed blob — never one that is addressable + through committed KV state. +- Tenant SQL execution (the SQL capability path) is likewise kept outside + the KV authorization/mutation transaction boundary. + +## Unaffected stages + +The following per-request stages retain their existing shape and are +explicitly out of scope for this change: pool acquire, transaction +begin/body, guard wait, replay, and audit wait. The periodic pool probe is a +pool-level background operation, not a per-request statement, and is not +counted against any request's budget. + +## Out of scope / excluded from these budgets + +Setup, migrations, retention pruning, telemetry probes, object-store calls, +and cold SQL hydration are excluded from the counts above and are reported +separately (see `tinycloud-core/src/telemetry.rs` stage labels). + +## Security invariants preserved + +- Revocation remains immediate and fail-closed: the shared chain guards + cover the full closure through authorization and mutation commit, and a + guarded-state mismatch (or a database failure while re-deriving it) + rejects the request rather than falling back to the pre-guard read. +- No pre-guard snapshot is ever used to authorize a request; only the + guarded, re-verified snapshot is used for authorization/mutation + decisions. +- Ancestor caveats and caveat containment remain binding: a uniquely + tightest contained caveat wins, and incomparable candidates return 403. +- A caveat that declares itself `constrained-statements` (directly or nested + under a `"constrained-statements"` key) but fails to parse is a malformed + *declared* caveat, not an unrelated one: + `AuthGraphSnapshot::constrained_statement_caveat_candidates` returns an + error for it (`TxError::MalformedSqlCaveat`, mapped to `403 Forbidden`) + instead of silently dropping it as if the grant were unconstrained. See + `auth_graph::tests::constrained_statement_caveat_candidates_fails_closed_on_malformed_direct_caveat`, + `..._malformed_nested_caveat`, and `..._malformed_ancestor_caveat`. +- A cyclic `parent_delegations` closure is rejected outright + (`load_closure_edges` runs a cycle check over the loaded closure before it + is used to derive lock keys or the snapshot) rather than being silently + accepted because a per-node visited-set traversal would otherwise + terminate against the cycle. See + `auth_graph::tests::load_closure_edges_fails_closed_on_cyclic_proof`. +- Closure memory and cycle-detection recursion are bounded by distinct node + count, not edge count: a wide (fan-out) graph can hold far more distinct + nodes than `MAX_CHAIN_TRAVERSAL_NODES` while its edge count stays far + below `edge_cap` (`MAX_CHAIN_TRAVERSAL_NODES^2`), since each node may have + only one edge. `load_closure_edges` rejects on distinct-node count before + `has_cycle`'s recursion ever runs over it. See + `auth_graph::tests::load_closure_edges_fails_closed_on_wide_over_limit_graph`. +- No cross-request authorization cache is introduced by this change. diff --git a/tinycloud-core/src/auth_graph.rs b/tinycloud-core/src/auth_graph.rs index eef84589..99f6906a 100644 --- a/tinycloud-core/src/auth_graph.rs +++ b/tinycloud-core/src/auth_graph.rs @@ -5,9 +5,9 @@ //! revocation lookup per ancestor (repeated for chain locking, revocation //! checks, and chain-window validation), the snapshot batch-loads the whole //! proof closure once: parent edges, then the -//! delegation rows, the cited proofs' ability/caveat rows, and the closure's -//! revocations in one query each. All chain checks then run in memory against -//! the same consistent view. +//! delegation rows, the closure's ability/caveat rows (cited roots and their +//! ancestors alike), and the closure's revocations in one query each. All +//! chain checks then run in memory against the same consistent view. use crate::hash::Hash; use crate::models::{abilities, delegation, revocation}; @@ -22,6 +22,73 @@ use std::collections::{HashMap, HashSet}; pub(crate) use crate::models::revocation::ChainTraversalError; use crate::models::revocation::MAX_CHAIN_TRAVERSAL_NODES; +/// Depth-first cycle detection over a child->parents edge map, using the +/// classic white/gray/black coloring so a node currently on the DFS stack +/// (gray) being revisited proves a cycle. `edges` is bounded by +/// `MAX_CHAIN_TRAVERSAL_NODES` before this runs, so recursion depth is +/// bounded too. +fn has_cycle(edges: &HashMap>) -> bool { + #[derive(Clone, Copy, PartialEq, Eq)] + enum Mark { + InProgress, + Done, + } + fn visit( + node: Hash, + edges: &HashMap>, + marks: &mut HashMap, + ) -> bool { + match marks.get(&node) { + Some(Mark::InProgress) => return true, + Some(Mark::Done) => return false, + None => {} + } + marks.insert(node, Mark::InProgress); + if let Some(parents) = edges.get(&node) { + for parent in parents { + if visit(*parent, edges, marks) { + return true; + } + } + } + marks.insert(node, Mark::Done); + false + } + + let mut marks: HashMap = HashMap::new(); + edges + .keys() + .any(|node| marks.get(node) != Some(&Mark::Done) && visit(*node, edges, &mut marks)) +} + +/// A single caveat value declares itself a `constrained-statements` caveat +/// either directly (top-level `mode: "constrained-statements"`) or nested +/// under a `"constrained-statements"` key. Unrelated caveat values (neither +/// shape present) are `Ok(None)` and silently skipped, matching prior +/// behavior. A value that *does* declare one of these shapes but fails to +/// parse (missing/malformed `statements`, `readOnly`, etc.) is a malformed +/// declared caveat and returns `Err`, so the caller fails closed instead of +/// treating a broken grant as an absent one. +fn declared_constrained_statement_caveat( + v: &serde_json::Value, +) -> Result< + Option, + crate::policy_capability::RejectionCode, +> { + let declares_mode_directly = v + .as_object() + .and_then(|o| o.get("mode")) + .and_then(serde_json::Value::as_str) + == Some("constrained-statements"); + if declares_mode_directly { + return crate::policy_capability::sql_caveat::parse(v).map(Some); + } + if let Some(inner) = v.as_object().and_then(|o| o.get("constrained-statements")) { + return crate::policy_capability::sql_caveat::parse(inner).map(Some); + } + Ok(None) +} + /// Batched ancestor-closure load over `parent_delegations`. A recursive CTE /// fetches all reachable edges in one query rather than walking one node at a /// time. The in-memory pass enforces the same fail-closed node budget as the @@ -110,6 +177,34 @@ pub(crate) async fn load_closure_edges( parents.sort_by(|left, right| left.as_ref().cmp(right.as_ref())); } + // TC-411: `rows.len() <= edge_cap` bounds edge *count*, not distinct node + // *count* -- a sparse, wide graph (e.g. many nodes with few parents each) + // can pass the edge check while still spanning far more than + // `MAX_CHAIN_TRAVERSAL_NODES` distinct nodes. `has_cycle` below recurses + // per distinct node, so that bound must be enforced first, over the + // decoded closure, before any recursive traversal runs. + let mut distinct_nodes: HashSet = HashSet::new(); + for (child, parents) in &edges { + distinct_nodes.insert(*child); + distinct_nodes.extend(parents.iter().copied()); + } + if distinct_nodes.len() > MAX_CHAIN_TRAVERSAL_NODES { + return Err(ChainTraversalError::LimitExceeded); + } + + // TC-411: a cycle in the loaded closure has no well-defined ancestor + // order for caveat/revocation resolution. A per-node traversal's visited + // set would still terminate against a cycle and silently accept the + // chain; fail closed instead, reusing `LimitExceeded` (already the + // established "reject this traversal outright" signal here -- see + // `load_guarded`'s guarded/reloaded mismatch case below) rather than + // widening the shared `ChainTraversalError` enum used across the + // delegate/revoke paths outside this module's scope. `has_cycle`'s + // recursion is now bounded by the distinct-node check above. + if has_cycle(&edges) { + return Err(ChainTraversalError::LimitExceeded); + } + let mut visited = nodes.iter().copied().collect::>(); let mut frontier = nodes.clone(); while let Some(current) = frontier.pop() { @@ -143,6 +238,29 @@ impl AuthGraphSnapshot { roots: &[Hash], ) -> Result { let (nodes, parents) = load_closure_edges(db, roots).await?; + Self::load_from_closure(db, nodes, parents).await + } + + /// Same as [`Self::load`], but for a closure (`nodes`/`parents`) already + /// known from a prior `load_closure_edges` call on this same connection's + /// database. Only safe to call with a closure that is already *proven* + /// complete for the connection being read from -- see [`Self::load_guarded`] + /// for the production caller, which re-derives and verifies the closure + /// under the caller's chain guards instead of trusting a pre-guard read. + /// + /// Ability/caveat rows are loaded for every node in the bounded closure + /// (`nodes`), not just the cited roots: `constrained_statement_caveat_candidates` + /// walks each root's full ancestor chain and reads `abilities()` at every + /// step, so an ancestor-only caveat (the descendant delegation carries no + /// caveat of its own) would otherwise be silently invisible even though it + /// is part of the already-loaded, already-bounded closure. `nodes` is + /// capped at `MAX_CHAIN_TRAVERSAL_NODES` by `load_closure_edges`, so this + /// stays a single bounded-size statement, not an unbounded one. + pub(crate) async fn load_from_closure( + db: &C, + nodes: Vec, + parents: HashMap>, + ) -> Result { if nodes.is_empty() { return Ok(Self { parents, @@ -159,12 +277,9 @@ impl AuthGraphSnapshot { .map(|row| (row.id, row)) .collect(); - let mut root_ids: Vec = roots.to_vec(); - root_ids.sort_by(|left, right| left.as_ref().cmp(right.as_ref())); - root_ids.dedup(); let mut ability_rows: HashMap> = HashMap::new(); for row in abilities::Entity::find() - .filter(abilities::Column::Delegation.is_in(root_ids)) + .filter(abilities::Column::Delegation.is_in(nodes.iter().copied())) .all(db) .await? { @@ -187,11 +302,43 @@ impl AuthGraphSnapshot { }) } + /// TC-411: the production entry point for the invocation path. `roots` + /// is the invocation's cited proofs; `guarded_keys` is the pre-guard + /// closure node set the caller already holds chain guards over (see + /// `SpaceDatabase::acquire_shared_chain_guards_for_keys`). + /// + /// A registration racing the guard acquisition (its exclusive guard + /// released just as this invocation's shared guard is granted) can leave + /// a cited root visible for the first time with ancestor edges that the + /// pre-guard closure read never saw -- `guarded_keys` would then be + /// missing those ancestors, and reusing it blindly would authorize + /// against an incomplete chain. This re-derives the closure on `db` + /// (expected to be the guarded connection/transaction) and fails closed + /// with [`ChainTraversalError::LimitExceeded`] if the freshly observed + /// node set is not *exactly* `guarded_keys`, instead of silently + /// authorizing against the stale pre-guard view. `parent_delegations` + /// rows are insert-only, so equality here proves the guarded view was + /// already complete. + pub(crate) async fn load_guarded( + db: &C, + roots: &[Hash], + guarded_keys: &[Hash], + ) -> Result { + let (nodes, edges) = load_closure_edges(db, roots).await?; + let guarded: HashSet = guarded_keys.iter().copied().collect(); + let reloaded: HashSet = nodes.iter().copied().collect(); + if reloaded != guarded { + return Err(ChainTraversalError::LimitExceeded); + } + Self::load_from_closure(db, nodes, edges).await + } + pub(crate) fn delegation(&self, id: &Hash) -> Option<&delegation::Model> { self.delegations.get(id) } - /// Persisted ability/caveat rows for a cited proof root. + /// Persisted ability/caveat rows for any node in the loaded closure + /// (a cited proof root or one of its ancestors). pub(crate) fn abilities(&self, id: &Hash) -> &[abilities::Model] { self.abilities.get(id).map(Vec::as_slice).unwrap_or(&[]) } @@ -220,6 +367,49 @@ impl AuthGraphSnapshot { ordered } + /// TC-411: every distinct SQL constrained-statement caveat reachable + /// from `roots` (each root plus its full ancestor closure), read purely + /// from this already-loaded snapshot -- zero additional statements. + /// Mirrors the persisted-caveat shapes accepted by the historical + /// per-request database walk (`constrained-statements` value directly, + /// or nested under a `"constrained-statements"` key); resolving + /// ambiguity across the returned candidates (zero/one/many) is the + /// caller's responsibility so SQL-specific fail-closed semantics stay + /// out of the shared authorization graph. + /// + /// A caveat value that *declares* itself as `constrained-statements` + /// (top-level `mode`, or nested under a `"constrained-statements"` key) + /// but fails to parse is a malformed declared caveat, not an unrelated + /// one -- this fails closed with the underlying `RejectionCode` instead + /// of silently dropping the caveat and leaving SQL unconstrained. + pub(crate) fn constrained_statement_caveat_candidates( + &self, + roots: &[Hash], + ) -> Result< + Vec, + crate::policy_capability::RejectionCode, + > { + let mut visited = HashSet::new(); + let mut found = Vec::new(); + for root in roots { + for id in self.chain_ids_from(root) { + if !visited.insert(id) { + continue; + } + for row in self.abilities(&id) { + for v in row.caveats.0.values() { + if let Some(caveat) = declared_constrained_statement_caveat(v)? { + if !found.contains(&caveat) { + found.push(caveat); + } + } + } + } + } + } + Ok(found) + } + /// First revoked strict ancestor of `start`, as a CID string. pub(crate) fn first_revoked_ancestor(&self, start: &Hash) -> Option { self.chain_ids_from(start) @@ -468,4 +658,231 @@ mod tests { assert_eq!(optimized_queries, 5, "depth {depth}"); } } + + /// Cost characteristics of the low-level `load_from_closure` primitive + /// in isolation: given a closure already known to be complete for the + /// connection being read from, it issues only the + /// delegations/abilities/revocations batch (3 queries) instead of + /// `load`'s 4 (which re-walks the recursive edge CTE), and depth 4 costs + /// exactly what depth 1 costs. The production invocation path does NOT + /// blindly reuse a pre-guard closure this way -- see + /// `AuthGraphSnapshot::load_guarded` and + /// `snapshot_load_guarded_fails_closed_on_concurrent_registration` below + /// for why a pre-guard closure cannot be trusted without re-verification. + #[tokio::test] + async fn snapshot_reuses_preguard_closure_without_a_second_edge_query() { + for depth in [0, 1, 4] { + let (db, counter) = counted_database().await; + let ids = insert_chain(&db, &format!("reuse-{depth}"), depth).await; + let leaf = ids[0]; + + let before = counter.load(Ordering::SeqCst); + let (nodes, parents) = load_closure_edges(&db, &[leaf]).await.unwrap(); + let lock_query_count = counter.load(Ordering::SeqCst) - before; + assert_eq!(lock_query_count, 1, "depth {depth}: lock-key closure query"); + + let before = counter.load(Ordering::SeqCst); + let snapshot = AuthGraphSnapshot::load_from_closure(&db, nodes, parents) + .await + .unwrap(); + let reuse_query_count = counter.load(Ordering::SeqCst) - before; + + assert_eq!( + reuse_query_count, 3, + "depth {depth}: reused-closure snapshot query count must be depth-independent" + ); + assert_eq!(snapshot.chain_ids_from(&leaf).len(), depth + 1); + // Depth 1 and depth 4 must cost exactly the same: 1 lock-key + // query + 3 reused-closure snapshot queries, independent of how + // many ancestors are in the chain. + if depth == 1 || depth == 4 { + assert_eq!( + lock_query_count, 1, + "depth {depth} lock-key cost vs depth 1/4 parity" + ); + assert_eq!( + reuse_query_count, 3, + "depth {depth} snapshot cost vs depth 1/4 parity" + ); + } + } + } + + /// TC-411: `load_guarded` re-derives the closure on the guarded + /// connection and must fail closed when the guarded key set (computed + /// pre-guard) does not match what is actually reachable now -- the + /// signature of a delegation whose registration committed while this + /// invocation was waiting to acquire its chain guard, so the pre-guard + /// closure never saw the new ancestor edge. + #[tokio::test] + async fn snapshot_load_guarded_fails_closed_on_concurrent_registration() { + let (db, _) = counted_database().await; + let ids = insert_chain(&db, "race", 1).await; + let leaf = ids[0]; + + // Simulates the pre-guard closure read happening before the + // ancestor edge (leaf -> ids[1]) is visible: the caller believes + // `leaf` has no parents and only guards `[leaf]`. + let stale_guarded_keys = vec![leaf]; + assert!(matches!( + AuthGraphSnapshot::load_guarded(&db, &[leaf], &stale_guarded_keys).await, + Err(ChainTraversalError::LimitExceeded) + )); + + // Once the guarded key set matches what is actually reachable, the + // same call succeeds and exposes the full chain. + let complete_guarded_keys = ids.clone(); + let snapshot = AuthGraphSnapshot::load_guarded(&db, &[leaf], &complete_guarded_keys) + .await + .unwrap(); + assert_eq!(snapshot.chain_ids_from(&leaf).len(), 2); + } + + async fn insert_ability( + db: &DatabaseConnection, + delegation: Hash, + caveat_value: serde_json::Value, + ) { + use crate::types::Caveats; + use std::collections::BTreeMap; + let mut caveats = BTreeMap::new(); + caveats.insert("caveat".to_string(), caveat_value); + abilities::ActiveModel { + resource: Set("tinycloud:did:key:actor:files/kv/doc".parse().unwrap()), + ability: Set("tinycloud.kv/put".to_string().try_into().unwrap()), + delegation: Set(delegation), + caveats: Set(Caveats(caveats)), + } + .insert(db) + .await + .unwrap(); + } + + /// TC-411 regression: a caveat that declares `mode: + /// "constrained-statements"` directly on the cited root but is missing + /// required fields (`readOnly`, `statements`) must fail closed rather + /// than being silently dropped as if the grant were unconstrained. + #[tokio::test] + async fn constrained_statement_caveat_candidates_fails_closed_on_malformed_direct_caveat() { + let (db, _) = counted_database().await; + let ids = insert_chain(&db, "malformed-direct", 0).await; + insert_ability( + &db, + ids[0], + serde_json::json!({"mode": "constrained-statements"}), + ) + .await; + + let snapshot = AuthGraphSnapshot::load(&db, &[ids[0]]).await.unwrap(); + assert!(snapshot + .constrained_statement_caveat_candidates(&[ids[0]]) + .is_err()); + } + + /// Same as above, but the malformed declaration lives under the nested + /// `"constrained-statements"` key rather than as a top-level `mode`. + #[tokio::test] + async fn constrained_statement_caveat_candidates_fails_closed_on_malformed_nested_caveat() { + let (db, _) = counted_database().await; + let ids = insert_chain(&db, "malformed-nested", 0).await; + insert_ability( + &db, + ids[0], + serde_json::json!({"constrained-statements": {"mode": "constrained-statements"}}), + ) + .await; + + let snapshot = AuthGraphSnapshot::load(&db, &[ids[0]]).await.unwrap(); + assert!(snapshot + .constrained_statement_caveat_candidates(&[ids[0]]) + .is_err()); + } + + /// A malformed declared caveat on a strict ancestor (not the cited root + /// itself) must also fail closed -- `chain_ids_from` walks the whole + /// closure, so this is not limited to the directly-cited proof. + #[tokio::test] + async fn constrained_statement_caveat_candidates_fails_closed_on_malformed_ancestor_caveat() { + let (db, _) = counted_database().await; + let ids = insert_chain(&db, "malformed-ancestor", 1).await; + insert_ability( + &db, + ids[1], + serde_json::json!({"mode": "constrained-statements", "readOnly": "not-a-bool"}), + ) + .await; + + let snapshot = AuthGraphSnapshot::load(&db, &[ids[0]]).await.unwrap(); + assert!(snapshot + .constrained_statement_caveat_candidates(&[ids[0]]) + .is_err()); + } + + /// An unrelated caveat (no `mode` field and no nested + /// `"constrained-statements"` key) is not a declared SQL caveat and must + /// be silently skipped rather than rejected. + #[tokio::test] + async fn constrained_statement_caveat_candidates_ignores_unrelated_caveats() { + let (db, _) = counted_database().await; + let ids = insert_chain(&db, "unrelated", 0).await; + insert_ability(&db, ids[0], serde_json::json!({"tables": ["foo"]})).await; + + let snapshot = AuthGraphSnapshot::load(&db, &[ids[0]]).await.unwrap(); + assert_eq!( + snapshot + .constrained_statement_caveat_candidates(&[ids[0]]) + .unwrap(), + Vec::new() + ); + } + + /// TC-411 regression: a cyclic `parent_delegations` closure must fail + /// closed. A per-node traversal's visited set would silently terminate + /// against the cycle and accept the chain instead. + #[tokio::test] + async fn load_closure_edges_fails_closed_on_cyclic_proof() { + let (db, _) = counted_database().await; + let a = hash(b"cycle-a"); + let b = hash(b"cycle-b"); + insert_delegation(&db, a).await; + insert_delegation(&db, b).await; + insert_edge(&db, a, b).await; + insert_edge(&db, b, a).await; + + assert!(matches!( + load_closure_edges(&db, &[a]).await, + Err(ChainTraversalError::LimitExceeded) + )); + assert!(matches!( + AuthGraphSnapshot::load(&db, &[a]).await, + Err(ChainTraversalError::LimitExceeded) + )); + } + + /// TC-411 regression: a wide (fan-out) graph can hold far more distinct + /// nodes than `MAX_CHAIN_TRAVERSAL_NODES` while its edge count stays far + /// below `edge_cap` (`MAX_CHAIN_TRAVERSAL_NODES^2`), since each node + /// here has only one edge. The distinct-node bound must reject this + /// before `has_cycle`'s recursion ever runs over it -- the `rows.len() > + /// edge_cap` check alone would let it through. + #[tokio::test] + async fn load_closure_edges_fails_closed_on_wide_over_limit_graph() { + let (db, _) = counted_database().await; + let leaf = hash(b"wide-leaf"); + insert_delegation(&db, leaf).await; + for index in 0..=MAX_CHAIN_TRAVERSAL_NODES { + let parent = hash(format!("wide-parent-{index}").as_bytes()); + insert_delegation(&db, parent).await; + insert_edge(&db, leaf, parent).await; + } + + assert!(matches!( + load_closure_edges(&db, &[leaf]).await, + Err(ChainTraversalError::LimitExceeded) + )); + assert!(matches!( + AuthGraphSnapshot::load(&db, &[leaf]).await, + Err(ChainTraversalError::LimitExceeded) + )); + } } diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index 250d7ca1..3038f7b7 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -136,6 +136,11 @@ pub struct KvInvokeOptions { pub max_response_bytes: Option, pub list_limit: Option, pub list_cursor: Option, + /// TC-411: when set, populate `TransactResult::sql_constrained_statement_candidates` + /// from the request-scoped authorization snapshot that this invocation + /// already builds -- zero additional statements. Off by default so + /// ordinary KV invocations don't pay even the in-memory scan cost. + pub derive_sql_constrained_statement_caveat: bool, } #[derive(Debug, Clone)] @@ -153,6 +158,13 @@ pub struct TransactResult { /// CIDs of delegations that were processed (saved) regardless of space existence. /// Used to return a CID even when all spaces were skipped. pub delegation_cids: Vec, + /// TC-411: distinct SQL constrained-statement caveats found on the + /// validated request's proof chain, populated only when + /// `KvInvokeOptions::derive_sql_constrained_statement_caveat` was set. + /// Always empty otherwise. Resolving zero/one/many candidates into an + /// effective (or ambiguous, fail-closed) caveat is the caller's job. + pub sql_constrained_statement_candidates: + Vec, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -196,6 +208,11 @@ pub enum TxError { Encryption(#[from] crate::encryption::EncryptionError), #[error("delegation-chain-traversal-limit-exceeded")] ChainTraversalLimitExceeded, + /// TC-411: a caveat on the validated proof chain declares itself a + /// `constrained-statements` SQL caveat but fails to parse. Fails closed + /// rather than treating the grant as unconstrained. + #[error("malformed sql constrained-statement caveat: {0}")] + MalformedSqlCaveat(&'static str), } #[non_exhaustive] @@ -1587,9 +1604,8 @@ where .collect(); let authz_start = Instant::now(); let closure_start = Instant::now(); - let lock_keys = crate::auth_graph::load_closure_edges(&self.conn, &roots) + let closure = crate::auth_graph::load_closure_edges(&self.conn, &roots) .await - .map(|(keys, _)| keys) .map_err(|error| match error { revocation::ChainTraversalError::Db(error) => TxError::Db(error), revocation::ChainTraversalError::LimitExceeded => { @@ -1598,11 +1614,11 @@ where }); crate::telemetry::observe_stage( crate::telemetry::InvocationStage::ChainClosureQuery, - crate::telemetry::StageOutcome::from(lock_keys.is_ok()), + crate::telemetry::StageOutcome::from(closure.is_ok()), closure_start.elapsed(), ); - let lock_keys = match lock_keys { - Ok(keys) => keys, + let (lock_keys, closure_edges) = match closure { + Ok(closure) => closure, Err(error) => { crate::telemetry::observe_stage( crate::telemetry::InvocationStage::AuthorizationGraphLoad, @@ -1618,7 +1634,9 @@ where // serialized against this authorization decision. What is dropped is // invocation-vs-invocation exclusion, which the revocation-ordering // invariant never depended on. - let _chain_guards = self.acquire_shared_chain_guards_for_keys(lock_keys).await; + let _chain_guards = self + .acquire_shared_chain_guards_for_keys(lock_keys.clone()) + .await; let mutation_keys = invocation .0 .capabilities @@ -1636,9 +1654,92 @@ where }) .collect::>(); if mutation_keys.is_empty() { - return self.invoke_read_only::(invocation, options, mode).await; + // TC-411: build the request-scoped snapshot once, after the + // shared chain guards above are held, and pass it straight into + // read authorization so `validate` does not perform a second + // closure/graph load. `load_guarded` re-derives the closure on + // `self.conn` (now guarded) and fails closed if it does not + // exactly match the pre-guard `lock_keys` — see + // `AuthGraphSnapshot::load_guarded` for why the pre-guard + // closure alone cannot be trusted — then re-reads + // delegations/abilities/revocations fresh, so revocation state + // is current as of after guard acquisition. + let _ = closure_edges; + let auth_graph = match crate::auth_graph::AuthGraphSnapshot::load_guarded( + &self.conn, &roots, &lock_keys, + ) + .await + { + Ok(graph) => graph, + Err(error) => { + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::AuthorizationGraphLoad, + crate::telemetry::StageOutcome::Error, + authz_start.elapsed(), + ); + return Err(TxStoreError::Tx(match error { + revocation::ChainTraversalError::Db(error) => TxError::Db(error), + revocation::ChainTraversalError::LimitExceeded => { + TxError::ChainTraversalLimitExceeded + } + })); + } + }; + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::AuthorizationGraphLoad, + crate::telemetry::StageOutcome::Ok, + authz_start.elapsed(), + ); + return self + .invoke_read_only::(invocation, options, mode, Some(&auth_graph)) + .await; } let _kv_object_guards = self.acquire_kv_object_guards(&mutation_keys).await; + // TC-411: authorize the whole invocation-scoped snapshot before any + // object-store persistence below (see the blob-persist comment + // further down). Loading it here, on `self.conn` rather than inside + // the not-yet-open transaction, is safe because `_chain_guards` + // (shared, acquired above) already excludes any concurrent + // revocation for `lock_keys` -- so a malformed/cyclic proof, + // mismatched guarded closure, unauthorized ability, or revoked + // delegation rejects the request before any durable side effect, + // matching the read-only branch above instead of after blobs are + // already persisted. + let _ = closure_edges; + let auth_graph = match crate::auth_graph::AuthGraphSnapshot::load_guarded( + &self.conn, &roots, &lock_keys, + ) + .await + { + Ok(graph) => graph, + Err(error) => { + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::AuthorizationGraphLoad, + crate::telemetry::StageOutcome::Error, + authz_start.elapsed(), + ); + return Err(TxStoreError::Tx(match error { + revocation::ChainTraversalError::Db(error) => TxError::Db(error), + revocation::ChainTraversalError::LimitExceeded => { + TxError::ChainTraversalLimitExceeded + } + })); + } + }; + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::AuthorizationGraphLoad, + crate::telemetry::StageOutcome::Ok, + authz_start.elapsed(), + ); + // SQL routes can carry a KV mutation capability in the same + // invocation. Derive the chain caveat before taking that mutation + // path so the SQL handler receives the same fail-closed candidates + // as a SQL-only (read-only) invocation. This is an in-memory scan of + // the guarded authorization snapshot and adds no database work. + let sql_constrained_statement_candidates = + constrained_statement_candidates(&invocation, &options, &auth_graph).map_err( + |rejection| TxStoreError::Tx(TxError::MalformedSqlCaveat(rejection.as_str())), + )?; let mut stages = HashMap::new(); let mut ops = Vec::new(); let mut write_hashes = HashMap::new(); @@ -1680,12 +1781,72 @@ where space: space.clone(), key: path.clone(), version: None, + deleted_invocation_id: None, }); } _ => {} } } + // Resolve every requested KV read before persisting blobs or opening + // the mutation transaction. A response-size or object-store error is + // an invocation failure, so it must leave the mutation, replay row, + // and history uncommitted. This keeps object-store I/O outside the + // database transaction without changing the pre-TC-411 all-or-nothing + // API behavior for a mixed get + mutation invocation. + let mut preloaded_gets = Vec::new(); + for cap in invocation.0.capabilities.iter() { + let Some(resource) = cap.resource.tinycloud_resource() else { + continue; + }; + if resource.service().as_str() != "kv" + || crate::policy_capability::resolve_alias(cap.ability.as_ref().as_ref()) + != "tinycloud.kv/get" + { + continue; + } + let Some(path) = resource.path().cloned() else { + continue; + }; + let space = resource.space().clone(); + let data = get_kv(&self.conn, &self.storage, &space, &path) + .await + .map_err(|e| match e { + EitherError::A(e) => TxStoreError::Tx(e.into()), + EitherError::B(e) => TxStoreError::StoreRead(e), + })?; + if let (Some(limit), Some((_, _, content))) = + (options.max_response_bytes, data.as_ref()) + { + if content.len() > limit { + return Err(TxStoreError::KvResponseTooLarge { + size: content.len(), + limit, + }); + } + } + preloaded_gets.push((space, path, data)); + } + + // TC-411: persist immutable put blobs to the object store only after + // the guarded authorization snapshot above has already accepted the + // request -- an unauthorized, revoked, or malformed-proof put never + // reaches this line, so it can never create an orphan blob -- and + // before opening the database transaction below, since an explicit + // db tx must never span an object-store call (see + // docs/invoke-query-budget.md). Object-store failure still prevents + // publication outright; a blob persisted here that never ends up + // referenced by a committed `kv_write` row (e.g. a later database + // failure aborts the transaction) is a harmless, unreachable, + // content-addressed orphan that is never surfaced through committed + // KV state. + for (key, stage) in std::mem::take(&mut stages) { + self.storage + .persist(&key.0, stage) + .await + .map_err(TxStoreError::StoreWrite)?; + } + let has_preconditions = !options.preconditions.is_empty(); let isolation_level = if has_preconditions { conditional_kv_isolation_level(&self.conn) @@ -1709,28 +1870,16 @@ where // recorded as a failure; it is disarmed to `ok` right before commit. let tx_body_timer = crate::telemetry::StageTimer::start(crate::telemetry::InvocationStage::DbTxBody); - let auth_graph = match crate::auth_graph::AuthGraphSnapshot::load(&tx, &roots).await { - Ok(graph) => graph, - Err(error) => { - crate::telemetry::observe_stage( - crate::telemetry::InvocationStage::AuthorizationGraphLoad, - crate::telemetry::StageOutcome::Error, - authz_start.elapsed(), - ); - return Err(TxStoreError::Tx(match error { - revocation::ChainTraversalError::Db(error) => TxError::Db(error), - revocation::ChainTraversalError::LimitExceeded => { - TxError::ChainTraversalLimitExceeded - } - })); - } - }; - crate::telemetry::observe_stage( - crate::telemetry::InvocationStage::AuthorizationGraphLoad, - crate::telemetry::StageOutcome::Ok, - authz_start.elapsed(), - ); + // TC-411: the guarded authorization snapshot was already loaded above + // (before blob persistence, on `self.conn`) and is reused here rather + // than re-derived on `&tx`. `_chain_guards` covers this whole span -- + // from before that load through commit below -- so no revocation can + // land between the snapshot load and commit; re-loading inside the + // transaction would only repeat the same five statements for no + // additional safety. let mut deleted_hashes = HashMap::new(); + let mut deleted_versions = HashMap::new(); + let mut deleted_invocation_ids = HashMap::new(); for key @ (space, path) in &mutation_keys { let current = get_kv_entity(&tx, space, path).await?; if let Some(precondition) = options.preconditions.get(key) { @@ -1744,6 +1893,29 @@ where } if let Some(entry) = current { deleted_hashes.insert(key.clone(), entry.value); + deleted_versions.insert(key.clone(), (entry.seq, entry.epoch, entry.epoch_seq)); + deleted_invocation_ids.insert(key.clone(), entry.invocation); + } + } + // TC-411: delete reuses the current-state row already loaded just + // above -- both its version tuple and its owning `invocation` id -- + // instead of letting `invocation::save` re-derive either with a + // second `kv_write` lookup. + for op in ops.iter_mut() { + if let Operation::KvDelete { + space, + key, + version, + deleted_invocation_id, + } = op + { + let map_key = (space.clone(), key.clone()); + if let Some(v) = deleted_versions.get(&map_key) { + *version = Some(*v); + } + if let Some(inv) = deleted_invocation_ids.get(&map_key) { + *deleted_invocation_id = Some(*inv); + } } } let caps = invocation.0.capabilities.clone(); @@ -1769,7 +1941,7 @@ where InvokeMode::Admitted => Event::AdmittedInvocation(Box::new(invocation), ops), InvokeMode::Public => Event::Invocation(Box::new(invocation), ops), }; - let commit = transact( + let mut commit = transact( &tx, &self.storage, &self.secrets, @@ -1785,6 +1957,7 @@ where TxStoreError::Tx(error) } })?; + commit.sql_constrained_statement_candidates = sql_constrained_statement_candidates; let mut results = Vec::new(); // perform and record side effects @@ -1802,23 +1975,13 @@ where }) { match cap { (space, "kv", "tinycloud.kv/get", path) => { - let data = - get_kv(&tx, &self.storage, space, path) - .await - .map_err(|e| match e { - EitherError::A(e) => TxStoreError::Tx(e.into()), - EitherError::B(e) => TxStoreError::StoreRead(e), - })?; - if let (Some(limit), Some((_, _, content))) = - (options.max_response_bytes, data.as_ref()) - { - if content.len() > limit { - return Err(TxStoreError::KvResponseTooLarge { - size: content.len(), - limit, - }); - } - } + let index = preloaded_gets + .iter() + .position(|(loaded_space, loaded_path, _)| { + loaded_space == space && loaded_path == path + }) + .expect("preflight loads every requested kv/get"); + let (_, _, data) = preloaded_gets.swap_remove(index); results.push(InvocationOutcome::KvRead(data)); } (space, "kv", "tinycloud.kv/list", path) => { @@ -1840,15 +2003,10 @@ where )) } (space, "kv", "tinycloud.kv/put", path) => { - if let Some(stage) = stages.remove(&(space.clone(), path.clone())) { - self.storage - .persist(space, stage) - .await - .map_err(TxStoreError::StoreWrite)?; - let hash = write_hashes - .get(&(space.clone(), path.clone())) - .copied() - .expect("staged KV writes have a content hash"); + // TC-411: the blob was already persisted to the object + // store before this transaction was opened (see above); + // this only looks up the already-known content hash. + if let Some(hash) = write_hashes.get(&(space.clone(), path.clone())).copied() { results.push(InvocationOutcome::KvWrite(hash)) } } @@ -1921,6 +2079,7 @@ where invocation: Invocation, options: KvInvokeOptions, mode: InvokeMode, + auth_graph: Option<&crate::auth_graph::AuthGraphSnapshot>, ) -> Result<(TransactResult, Vec>), TxStoreError> where B: ImmutableWriteStore + ImmutableReadStore, @@ -1931,19 +2090,28 @@ where // once at the admission boundary; only re-check authorization and // signed-time validity here rather than verifying the signature a // second time. + // TC-411: the caller already loaded the request-scoped snapshot under + // the shared chain guards, so pass it through instead of letting + // `validate` perform a second closure/graph load. match mode { - InvokeMode::Admitted => { - invocation::authorize_admitted(&self.conn, &invocation.0, OffsetDateTime::now_utc()) - .await - .map_err(TxError::::from)? - } - InvokeMode::Public | InvokeMode::Internal => invocation::verify_and_authorize( + InvokeMode::Admitted => invocation::authorize_admitted( &self.conn, &invocation.0, OffsetDateTime::now_utc(), + auth_graph, ) .await .map_err(TxError::::from)?, + InvokeMode::Public | InvokeMode::Internal => { + invocation::verify_and_authorize_with_graph( + &self.conn, + &invocation.0, + OffsetDateTime::now_utc(), + auth_graph, + ) + .await + .map_err(TxError::::from)? + } }; let requested_spaces = invocation.0.spaces().cloned().collect::>(); @@ -1970,6 +2138,28 @@ where .and_then(|value| serde_json::from_value(value.clone()).ok()) }) }); + // TC-411: batch the `current_kv` index lookup for every `kv/get` + // and `kv/metadata` capability in this invocation into one + // statement, independent of item count, instead of one lookup per + // capability (see `batch_get_kv_entities`). + let index_keys = invocation + .0 + .capabilities + .iter() + .filter_map(|capability| { + let resource = capability.resource.tinycloud_resource()?; + if resource.service().as_str() != "kv" { + return None; + } + let ability = + crate::policy_capability::resolve_alias(capability.ability.as_ref().as_ref()); + if !matches!(ability, "tinycloud.kv/get" | "tinycloud.kv/metadata") { + return None; + } + Some((resource.space().clone(), resource.path()?.clone())) + }) + .collect::>(); + let kv_entities = batch_get_kv_entities(&self.conn, &index_keys).await?; let mut results = Vec::new(); for cap in invocation.0.capabilities.iter().filter_map(|capability| { capability @@ -1988,12 +2178,16 @@ where }) { match cap { (space, "kv", "tinycloud.kv/get", path) => { - let data = get_kv(&self.conn, &self.storage, space, path) - .await - .map_err(|error| match error { - EitherError::A(error) => TxStoreError::Tx(error.into()), - EitherError::B(error) => TxStoreError::StoreRead(error), - })?; + let entry = kv_entities.get(&(space.clone(), path.clone())); + let data = match entry { + Some(entry) => self + .storage + .read(space, &entry.value) + .await + .map_err(TxStoreError::StoreRead)? + .map(|content| (entry.metadata.clone(), entry.value, content)), + None => None, + }; if let (Some(limit), Some((_, _, content))) = (options.max_response_bytes, data.as_ref()) { @@ -2012,9 +2206,10 @@ where results.push(InvocationOutcome::KvList(list, truncated, None)); } (space, "kv", "tinycloud.kv/metadata", path) => { - results.push(InvocationOutcome::KvMetadata( - metadata_with_hash(&self.conn, space, path).await?, - )); + let metadata = kv_entities + .get(&(space.clone(), path.clone())) + .map(|entry| (entry.metadata.clone(), entry.value)); + results.push(InvocationOutcome::KvMetadata(metadata)); } (space, "capabilities", "tinycloud.capabilities/read", path) if path.as_str() == "all" => @@ -2064,17 +2259,54 @@ where read_audit_start.elapsed(), ); record_result?; + // TC-411: derive the SQL constrained-statement caveat candidates + // straight from the already-loaded, already-validated `auth_graph` + // snapshot -- an in-memory scan, zero additional statements -- so a + // caller (the SQL route) never has to re-walk `parent_delegations` + // itself. + let sql_constrained_statement_candidates = match auth_graph { + Some(graph) => constrained_statement_candidates(&invocation, &options, graph).map_err( + |rejection| TxStoreError::Tx(TxError::MalformedSqlCaveat(rejection.as_str())), + )?, + None => Vec::new(), + }; Ok(( TransactResult { commits: HashMap::new(), skipped_spaces: Vec::new(), delegation_cids: Vec::new(), + sql_constrained_statement_candidates, }, results, )) } } +/// Extract constrained SQL caveats from the request's already-guarded +/// authorization snapshot. Both read-only and mutation-bearing invocations +/// use this helper: SQL routing is selected by SQL capabilities, while the +/// transaction path is selected independently by KV mutations. +fn constrained_statement_candidates( + invocation: &Invocation, + options: &KvInvokeOptions, + auth_graph: &crate::auth_graph::AuthGraphSnapshot, +) -> Result< + Vec, + crate::policy_capability::RejectionCode, +> { + if !options.derive_sql_constrained_statement_caveat { + return Ok(Vec::new()); + } + let roots: Vec = invocation + .0 + .parents + .iter() + .copied() + .map(Hash::from) + .collect(); + auth_graph.constrained_statement_caveat_candidates(&roots) +} + fn chain_isolation_level(db: &C) -> Option { match db.get_database_backend() { // SQLite's default transaction mode is serializable; sqlx rejects an @@ -2131,6 +2363,7 @@ fn already_registered_result(retained_hash: Hash) -> TransactResult { commits: HashMap::new(), skipped_spaces: Vec::new(), delegation_cids: vec![retained_hash], + sql_constrained_statement_candidates: Vec::new(), } } @@ -2386,15 +2619,23 @@ async fn event_spaces<'a, C: ConnectionTrait>( ) -> Result>, DbErr> { // get orderings of events listed as revoked by events in the ev list let mut spaces = HashMap::>::new(); - let revoked_events = event_order::Entity::find() - .filter( - event_order::Column::Event.is_in(ev.iter().filter_map(|(_, e)| match e { - Event::Revocation(r) => Some(Hash::from(r.0.revoked)), - _ => None, - })), - ) - .all(db) - .await?; + let revoked_hashes: Vec = ev + .iter() + .filter_map(|(_, e)| match e { + Event::Revocation(r) => Some(Hash::from(r.0.revoked)), + _ => None, + }) + .collect(); + // Skip the round trip entirely when this batch contains no Revocation + // event; an empty IN-list would otherwise still hit the DB every call. + let revoked_events = if revoked_hashes.is_empty() { + Vec::new() + } else { + event_order::Entity::find() + .filter(event_order::Column::Event.is_in(revoked_hashes)) + .all(db) + .await? + }; for e in ev { match &e.1 { Event::Delegation(d) => { @@ -2876,6 +3117,7 @@ pub(crate) async fn transact( .collect(), skipped_spaces, delegation_cids, + sql_constrained_statement_candidates: Vec::new(), }) } else { // All spaces were skipped (delegation-only with no existing spaces). @@ -2919,6 +3161,7 @@ pub(crate) async fn transact( commits: HashMap::new(), skipped_spaces, delegation_cids, + sql_constrained_statement_candidates: Vec::new(), }) } } @@ -3147,6 +3390,56 @@ async fn get_kv_entity( Ok(result) } +/// TC-411: batched form of [`get_kv_entity`] -- one `current_kv` index +/// statement for every requested key, independent of how many keys are +/// requested, instead of one statement per key. `keys` may repeat and may +/// span multiple spaces; the `space`/`key` filters are independent `IN` +/// lists (a superset filter), so callers key the returned map by the exact +/// `(space, key)` pair to drop any cross-product rows. +async fn batch_get_kv_entities( + db: &C, + keys: &[(SpaceId, Path)], +) -> Result, DbErr> { + if keys.is_empty() { + return Ok(HashMap::new()); + } + let start = Instant::now(); + let spaces = keys + .iter() + .map(|(space, _)| SpaceIdWrap(space.clone())) + .collect::>(); + let paths = keys + .iter() + .map(|(_, path)| crate::types::Path(path.clone())) + .collect::>(); + let query_result = current_kv::Entity::find() + .filter(current_kv::Column::Space.is_in(spaces)) + .filter(current_kv::Column::Key.is_in(paths)) + .filter(current_kv::Column::Deleted.eq(false)) + .all(db) + .await; + let rows = match query_result { + Ok(rows) => rows, + Err(error) => { + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::KvIndexLookup, + crate::telemetry::StageOutcome::Error, + start.elapsed(), + ); + return Err(error); + } + }; + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::KvIndexLookup, + crate::telemetry::StageOutcome::Ok, + start.elapsed(), + ); + Ok(rows + .into_iter() + .map(|row| ((row.space.0.clone(), row.key.0.clone()), row)) + .collect()) +} + /// Half-open `[lower, upper)` bounds selecting every `ability.resource` that /// belongs to `space_id`. /// @@ -4040,7 +4333,7 @@ mod test { .insert(&db.conn) .await .unwrap(); - invocation::upsert_current_kv(&db.conn, write) + invocation::upsert_current_kv_batch(&db.conn, vec![write]) .await .unwrap(); } @@ -4219,7 +4512,7 @@ mod test { .await .unwrap() .unwrap(); - invocation::upsert_current_kv(&db.conn, winner) + invocation::upsert_current_kv_batch(&db.conn, vec![winner]) .await .unwrap(); @@ -7105,4 +7398,74 @@ mod test { .ok(); exercise.expect("TC-320 PostgreSQL collation resilience"); } + + /// TC-411: `batch_get_kv_entities` must issue exactly one `current_kv` + /// statement for a batch request, independent of how many keys are in + /// it -- proving the one-item and 100-item budgets in + /// docs/invoke-query-budget.md instead of the pre-TC-411 one-lookup-per- + /// capability loop. + #[tokio::test] + async fn batch_get_kv_entities_issues_one_statement_regardless_of_item_count() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + let mut conn = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + Migrator::up(&conn, None).await.unwrap(); + conn.execute(Statement::from_string( + DbBackend::Sqlite, + "PRAGMA foreign_keys = OFF".to_string(), + )) + .await + .unwrap(); + + let space = test_space_id("batch-get-count"); + let mut all_keys = Vec::new(); + for index in 0..100 { + let key: Path = format!("k/{index}").parse().unwrap(); + let write = crate::models::kv_write::Model { + space: SpaceIdWrap(space.clone()), + key: crate::types::Path(key.clone()), + invocation: crate::hash::hash(format!("inv-{index}").as_bytes()), + seq: 0, + epoch: crate::hash::hash(format!("epoch-{index}").as_bytes()), + epoch_seq: 0, + value: crate::hash::hash(format!("value-{index}").as_bytes()), + metadata: crate::types::Metadata(Default::default()), + }; + crate::models::kv_write::ActiveModel::from(write.clone()) + .insert(&conn) + .await + .unwrap(); + crate::models::invocation::upsert_current_kv_batch(&conn, vec![write]) + .await + .unwrap(); + all_keys.push((space.clone(), key)); + } + + let counter = Arc::new(AtomicUsize::new(0)); + let query_counter = Arc::clone(&counter); + conn.set_metric_callback(move |_info| { + query_counter.fetch_add(1, Ordering::SeqCst); + }); + + let before = counter.load(Ordering::SeqCst); + let one = batch_get_kv_entities(&conn, &all_keys[..1]).await.unwrap(); + assert_eq!( + counter.load(Ordering::SeqCst) - before, + 1, + "one-item batch must issue exactly one statement" + ); + assert_eq!(one.len(), 1); + + let before = counter.load(Ordering::SeqCst); + let hundred = batch_get_kv_entities(&conn, &all_keys).await.unwrap(); + assert_eq!( + counter.load(Ordering::SeqCst) - before, + 1, + "100-item batch must issue exactly one statement, not one per item" + ); + assert_eq!(hundred.len(), 100); + } } diff --git a/tinycloud-core/src/events/mod.rs b/tinycloud-core/src/events/mod.rs index 8b3d99f1..742b1bda 100644 --- a/tinycloud-core/src/events/mod.rs +++ b/tinycloud-core/src/events/mod.rs @@ -62,6 +62,14 @@ pub(crate) enum Operation { space: SpaceId, key: Path, version: Option<(i64, Hash, i64)>, + /// TC-411: the `current_kv.invocation` id already loaded by the + /// caller (`db.rs`) while resolving `version` above, threaded + /// through so `invocation::save` can persist the `kv_delete` audit + /// row without re-querying `kv_write` for it. `None` when no + /// current (non-deleted) row existed at delete time -- e.g. + /// deleting an already-deleted or never-written key -- in which + /// case `invocation::save` still falls back to its own lookup. + deleted_invocation_id: Option, }, } @@ -86,10 +94,12 @@ impl Operation { space, key, version, + deleted_invocation_id, } => VersionedOperation::KvDelete { space, key, version, + deleted_invocation_id, seq, epoch, epoch_seq, @@ -120,6 +130,7 @@ pub(crate) enum VersionedOperation { space: SpaceId, key: Path, version: Option<(i64, Hash, i64)>, + deleted_invocation_id: Option, seq: i64, epoch: Hash, epoch_seq: i64, diff --git a/tinycloud-core/src/models/invocation.rs b/tinycloud-core/src/models/invocation.rs index 9218b48e..119d95d6 100644 --- a/tinycloud-core/src/models/invocation.rs +++ b/tinycloud-core/src/models/invocation.rs @@ -20,7 +20,9 @@ use serde::Serialize; use std::collections::HashMap; use time::{format_description::well_known::Rfc3339, OffsetDateTime}; use tinycloud_auth::{ - authorization::TinyCloudInvocation, identity::did_principal_matches, resource::Path, + authorization::TinyCloudInvocation, + identity::did_principal_matches, + resource::{Path, SpaceId}, ssi::dids::AnyDidMethod, }; @@ -182,13 +184,14 @@ pub(crate) async fn authorize_admitted( db: &C, invocation: &util::InvocationInfo, now: OffsetDateTime, + auth_graph: Option<&crate::auth_graph::AuthGraphSnapshot>, ) -> Result<(), Error> { invocation .invocation .payload() .validate_time(None) .map_err(|_| InvocationError::InvalidTime)?; - validate(db, invocation, Some(now), None).await + validate(db, invocation, Some(now), auth_graph).await } pub async fn verify_invocation(invocation: &TinyCloudInvocation) -> Result<(), Error> { @@ -227,9 +230,24 @@ pub async fn verify_and_authorize( db: &C, invocation: &util::InvocationInfo, now: OffsetDateTime, +) -> Result<(), Error> { + verify_and_authorize_with_graph(db, invocation, now, None).await +} + +/// Same as [`verify_and_authorize`], but lets a caller inside this crate that +/// already holds a request-scoped [`crate::auth_graph::AuthGraphSnapshot`] +/// (loaded under the shared chain guards) pass it straight through instead of +/// letting `validate` perform a second closure/graph load (TC-411). Not +/// `pub`: `AuthGraphSnapshot` is `pub(crate)`, so this signature cannot cross +/// the crate boundary. +pub(crate) async fn verify_and_authorize_with_graph( + db: &C, + invocation: &util::InvocationInfo, + now: OffsetDateTime, + auth_graph: Option<&crate::auth_graph::AuthGraphSnapshot>, ) -> Result<(), Error> { verify_invocation(&invocation.invocation).await?; - validate(db, invocation, Some(now), None).await + validate(db, invocation, Some(now), auth_graph).await } // verify parenthood and authorization @@ -551,8 +569,13 @@ async fn save( .await?; } - for param in ¶meters { - match param { + // TC-411: history (kv_write) and projection (current_kv) persistence for + // every put in this invocation is batched into exactly two statements -- + // one multi-row insert per table -- independent of item count, instead + // of one round trip per put (see `upsert_current_kv_batch`). + let put_writes: Vec = parameters + .iter() + .filter_map(|param| match param { VersionedOperation::KvWrite { key, value, @@ -561,67 +584,55 @@ async fn save( seq, epoch, epoch_seq, - } => { - let write = kv_write::Model { - invocation: hash, - key: key.clone().into(), - value: *value, - space: space.clone().into(), - metadata: metadata.clone(), - seq: *seq, - epoch: *epoch, - epoch_seq: *epoch_seq, - }; - kv_write::Entity::insert(kv_write::ActiveModel::from(write.clone())) - .exec(db) + } => Some(kv_write::Model { + invocation: hash, + key: key.clone().into(), + value: *value, + space: space.clone().into(), + metadata: metadata.clone(), + seq: *seq, + epoch: *epoch, + epoch_seq: *epoch_seq, + }), + VersionedOperation::KvDelete { .. } => None, + }) + .collect(); + if !put_writes.is_empty() { + kv_write::Entity::insert_many(put_writes.iter().cloned().map(kv_write::ActiveModel::from)) + .exec(db) + .await?; + upsert_current_kv_batch(db, put_writes).await?; + } + + for param in ¶meters { + if let VersionedOperation::KvDelete { + key, + version, + deleted_invocation_id, + space, + seq: _, + epoch: _, + epoch_seq: _, + } = param + { + let deleted_invocation_id = + resolve_deleted_invocation_id(db, space, key, *version, *deleted_invocation_id) .await?; - upsert_current_kv(db, write).await?; - } - VersionedOperation::KvDelete { - key, - version, - space, - seq: _, - epoch: _, - epoch_seq: _, - } => { - let deleted_invocation_id = if let Some((s, e, es)) = version { - kv_write::Entity::find().filter( - Condition::all() - .add(kv_write::Column::Key.eq(key.as_str())) - .add(kv_write::Column::Space.eq(SpaceIdWrap(space.clone()))) - .add(kv_write::Column::Seq.eq(*s)) - .add(kv_write::Column::Epoch.eq(*e)) - .add(kv_write::Column::EpochSeq.eq(*es)), - ) - } else { - kv_write::Entity::find() - .filter(kv_write::Column::Key.eq(key.as_str())) - .filter(kv_write::Column::Space.eq(SpaceIdWrap(space.clone()))) - .order_by_desc(kv_write::Column::Seq) - .order_by_desc(kv_write::Column::Epoch) - .order_by_desc(kv_write::Column::EpochSeq) - } - .one(db) - .await? - .ok_or_else(|| InvocationError::MissingKvWrite(key.clone()))? - .invocation; - kv_delete::Entity::insert(kv_delete::ActiveModel::from(kv_delete::Model { - key: key.clone().into(), - invocation_id: hash, - space: space.clone().into(), - deleted_invocation_id, - })) - .exec(db) - .await?; - delete_current_kv_if_invocation( - db, - &SpaceIdWrap(space.clone()), - key.as_str(), - deleted_invocation_id, - ) - .await?; - } + kv_delete::Entity::insert(kv_delete::ActiveModel::from(kv_delete::Model { + key: key.clone().into(), + invocation_id: hash, + space: space.clone().into(), + deleted_invocation_id, + })) + .exec(db) + .await?; + delete_current_kv_if_invocation( + db, + &SpaceIdWrap(space.clone()), + key.as_str(), + deleted_invocation_id, + ) + .await?; } } @@ -630,6 +641,47 @@ async fn save( Ok(hash) } +/// TC-411: `db.rs` already loads the current `current_kv` row (and its +/// owning `invocation` id) under the mutation transaction before staging a +/// delete op; `known_invocation_id` is that already-loaded id, reused here +/// instead of re-querying `kv_write`. Only falls back to a query when the +/// caller had no already-loaded row -- deleting a key with no live current +/// entry (never written, or already deleted) -- which still requires +/// deriving the most recent `kv_write` row for the audit trail. +async fn resolve_deleted_invocation_id( + db: &C, + space: &SpaceId, + key: &Path, + version: Option<(i64, Hash, i64)>, + known_invocation_id: Option, +) -> Result { + if let Some(id) = known_invocation_id { + return Ok(id); + } + let query = if let Some((s, e, es)) = version { + kv_write::Entity::find().filter( + Condition::all() + .add(kv_write::Column::Key.eq(key.as_str())) + .add(kv_write::Column::Space.eq(SpaceIdWrap(space.clone()))) + .add(kv_write::Column::Seq.eq(s)) + .add(kv_write::Column::Epoch.eq(e)) + .add(kv_write::Column::EpochSeq.eq(es)), + ) + } else { + kv_write::Entity::find() + .filter(kv_write::Column::Key.eq(key.as_str())) + .filter(kv_write::Column::Space.eq(SpaceIdWrap(space.clone()))) + .order_by_desc(kv_write::Column::Seq) + .order_by_desc(kv_write::Column::Epoch) + .order_by_desc(kv_write::Column::EpochSeq) + }; + Ok(query + .one(db) + .await? + .ok_or_else(|| InvocationError::MissingKvWrite(key.clone()))? + .invocation) +} + fn incoming_is_newer(existing: current_kv::Entity, incoming: Alias) -> Condition { Condition::any() .add( @@ -664,21 +716,18 @@ fn incoming_is_newer(existing: current_kv::Entity, incoming: Alias) -> Condition ) } -pub(crate) async fn upsert_current_kv( +/// TC-411: batched form of the single-write upsert -- one projection +/// (`current_kv`) statement for every write in `writes`, independent of item +/// count, instead of one upsert per write. Each row's ON CONFLICT action +/// still resolves against that row's own `EXCLUDED`/`VALUES` values, so the +/// per-row newer-wins ordering below is unchanged by batching. +pub(crate) async fn upsert_current_kv_batch( db: &C, - write: kv_write::Model, + writes: Vec, ) -> Result<(), DbErr> { - let current = current_kv::Model { - space: write.space, - key: write.key, - invocation: write.invocation, - seq: write.seq, - epoch: write.epoch, - epoch_seq: write.epoch_seq, - value: write.value, - metadata: write.metadata, - deleted: false, - }; + if writes.is_empty() { + return Ok(()); + } let mut conflict = OnConflict::columns([current_kv::Column::Space, current_kv::Column::Key]); if db.get_database_backend() == DatabaseBackend::MySql { // MySQL ignores ON CONFLICT's action WHERE. Keep the ordering fields @@ -717,7 +766,18 @@ pub(crate) async fn upsert_current_kv( Alias::new("excluded"), )); } - match current_kv::Entity::insert(current_kv::ActiveModel::from(current)) + let models = writes.into_iter().map(|write| current_kv::Model { + space: write.space, + key: write.key, + invocation: write.invocation, + seq: write.seq, + epoch: write.epoch, + epoch_seq: write.epoch_seq, + value: write.value, + metadata: write.metadata, + deleted: false, + }); + match current_kv::Entity::insert_many(models.map(current_kv::ActiveModel::from)) .on_conflict(conflict.to_owned()) .exec(db) .await @@ -922,7 +982,7 @@ mod tests { .insert(&tx) .await .unwrap(); - upsert_current_kv(&tx, write).await.unwrap(); + upsert_current_kv_batch(&tx, vec![write]).await.unwrap(); tx.commit().await.unwrap(); } @@ -983,7 +1043,9 @@ mod tests { .insert(&tx) .await .unwrap(); - upsert_current_kv(&tx, rolled_back).await.unwrap(); + upsert_current_kv_batch(&tx, vec![rolled_back]) + .await + .unwrap(); tx.rollback().await.unwrap(); assert_eq!(kv_write::Entity::find().count(&db).await.unwrap(), 2); let watermark = current_kv::Entity::find_by_id(( @@ -1018,7 +1080,9 @@ mod tests { .insert(&db) .await .unwrap(); - upsert_current_kv(&db, newer.clone()).await.unwrap(); + upsert_current_kv_batch(&db, vec![newer.clone()]) + .await + .unwrap(); kv_delete::ActiveModel { invocation_id: Set(crate::hash::hash(b"delete-newer")), space: Set(SpaceIdWrap(space.clone())), @@ -1041,7 +1105,7 @@ mod tests { .insert(&db) .await .unwrap(); - upsert_current_kv(&db, older).await.unwrap(); + upsert_current_kv_batch(&db, vec![older]).await.unwrap(); let watermark = current_kv::Entity::find_by_id(( SpaceIdWrap(space), @@ -1066,7 +1130,7 @@ mod tests { let seq_newer = test_write(&space, "by-seq", "seq-newer", 2); let seq_older = test_write(&space, "by-seq", "seq-older", 1); for write in [seq_newer.clone(), seq_older] { - upsert_current_kv(&db, write).await.unwrap(); + upsert_current_kv_batch(&db, vec![write]).await.unwrap(); } let mut epoch_a = test_write(&space, "by-epoch", "epoch-a", 3); @@ -1076,8 +1140,12 @@ mod tests { } else { (epoch_b.clone(), epoch_a.clone()) }; - upsert_current_kv(&db, epoch_newer.clone()).await.unwrap(); - upsert_current_kv(&db, epoch_older).await.unwrap(); + upsert_current_kv_batch(&db, vec![epoch_newer.clone()]) + .await + .unwrap(); + upsert_current_kv_batch(&db, vec![epoch_older]) + .await + .unwrap(); epoch_a.key = "by-epoch-seq".parse::().unwrap().into(); epoch_b.key = epoch_a.key.clone(); @@ -1085,8 +1153,10 @@ mod tests { epoch_b.epoch = epoch_a.epoch; epoch_a.epoch_seq = 2; epoch_b.epoch_seq = 1; - upsert_current_kv(&db, epoch_a.clone()).await.unwrap(); - upsert_current_kv(&db, epoch_b).await.unwrap(); + upsert_current_kv_batch(&db, vec![epoch_a.clone()]) + .await + .unwrap(); + upsert_current_kv_batch(&db, vec![epoch_b]).await.unwrap(); for (key, invocation) in [ ("by-seq", seq_newer.invocation), @@ -1287,4 +1357,116 @@ mod tests { caveats_contain_child(&parent, &child_with_constraints) .expect("narrowing on an unconstrained parent must be allowed"); } + + /// TC-411: multipart put persistence (`kv_write` history insert + + /// `current_kv` projection upsert) must be exactly two statements, + /// independent of how many puts are in the batch, instead of `2 * N`. + #[tokio::test] + async fn multipart_put_persistence_is_two_statements_regardless_of_item_count() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + for item_count in [1usize, 100] { + let mut db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); + db.execute(Statement::from_string( + DatabaseBackend::Sqlite, + "PRAGMA foreign_keys = OFF".to_string(), + )) + .await + .unwrap(); + let space = test_space("multipart-put-count"); + let writes: Vec = (0..item_count) + .map(|index| test_write(&space, &format!("k/{index}"), "batched", index as i64)) + .collect(); + + let counter = Arc::new(AtomicUsize::new(0)); + let query_counter = Arc::clone(&counter); + db.set_metric_callback(move |_info| { + query_counter.fetch_add(1, Ordering::SeqCst); + }); + + let before = counter.load(Ordering::SeqCst); + kv_write::Entity::insert_many(writes.iter().cloned().map(kv_write::ActiveModel::from)) + .exec(&db) + .await + .unwrap(); + upsert_current_kv_batch(&db, writes).await.unwrap(); + assert_eq!( + counter.load(Ordering::SeqCst) - before, + 2, + "{item_count}-item put batch must issue exactly two statements" + ); + } + } + + /// TC-411: when `db.rs` already loaded the current `current_kv` row for + /// a delete and threads its `invocation` id through as + /// `known_invocation_id`, `resolve_deleted_invocation_id` must not + /// re-query `kv_write` to re-derive it. Compares the statement count of + /// the reuse path against the fallback (no pre-loaded id) path, which + /// still performs exactly one extra lookup. + #[tokio::test] + async fn delete_reuses_preloaded_invocation_id_without_extra_kv_write_query() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + async fn run(known_invocation_id: Option) -> usize { + let mut db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())) + .await + .unwrap(); + Migrator::up(&db, None).await.unwrap(); + db.execute(Statement::from_string( + DatabaseBackend::Sqlite, + "PRAGMA foreign_keys = OFF".to_string(), + )) + .await + .unwrap(); + let space = test_space("delete-reuse-count"); + let write = test_write(&space, "reused", "orig", 1); + kv_write::ActiveModel::from(write.clone()) + .insert(&db) + .await + .unwrap(); + upsert_current_kv_batch(&db, vec![write.clone()]) + .await + .unwrap(); + + let counter = Arc::new(AtomicUsize::new(0)); + let query_counter = Arc::clone(&counter); + db.set_metric_callback(move |_info| { + query_counter.fetch_add(1, Ordering::SeqCst); + }); + + let before = counter.load(Ordering::SeqCst); + let resolved = resolve_deleted_invocation_id( + &db, + &space, + &"reused".parse::().unwrap(), + Some((write.seq, write.epoch, write.epoch_seq)), + known_invocation_id, + ) + .await + .unwrap(); + assert_eq!( + resolved, write.invocation, + "must resolve to the same invocation id regardless of path taken" + ); + counter.load(Ordering::SeqCst) - before + } + + let reused = run(Some(crate::hash::hash(b"invocation-orig"))).await; + let requeried = run(None).await; + assert_eq!( + requeried, + reused + 1, + "omitting the pre-loaded invocation id must cost exactly one extra kv_write query" + ); + assert_eq!( + reused, 0, + "reusing the pre-loaded id must cost zero statements" + ); + } } diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 809f7ad4..ed2cbbe7 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -517,6 +517,7 @@ pub async fn delegate( commits, skipped_spaces, delegation_cids, + sql_constrained_statement_candidates: _, } = result; let activated: Vec = commits.keys().map(|s| s.to_string()).collect(); let skipped: Vec = skipped_spaces.iter().map(|s| s.to_string()).collect(); @@ -1024,6 +1025,7 @@ fn kv_invoke_options_for_capabilities_with_cursor( max_response_bytes, list_limit, list_cursor, + derive_sql_constrained_statement_caveat: false, }) } @@ -2135,9 +2137,11 @@ async fn handle_sql_invoke( // path is a holdover (and is still consulted as a fallback so the // tinycloud.sql/write path keeps working) but a constrained-statements // caveat on the delegation chain MUST win and fail-closed. - let parent_cids: Vec<_> = admitted.invocation().0.parents.to_vec(); - let chain_constrained = derive_chain_constrained_caveat(tinycloud, &parent_cids).await?; - + // + // TC-411: the caveat candidates are read out of `auth_result` below, + // straight from the request-scoped authorization snapshot that + // `verify_auth_admitted` already builds to authorize the invocation -- + // no separate `parent_delegations` walk, so this adds zero statements. let facts_caveats: Option = admitted .invocation() .0 @@ -2158,7 +2162,16 @@ async fn handle_sql_invoke( // boundary; `verify_auth_admitted` uses the admitted core entry point so this // shared SQL/DuckDB authorization path does not re-run signature // verification a second time. - let auth_result = verify_auth_admitted("server.sql.auth", admitted, tinycloud).await?; + let auth_result = verify_auth_admitted( + "server.sql.auth", + admitted, + tinycloud, + KvInvokeOptions { + derive_sql_constrained_statement_caveat: true, + ..Default::default() + }, + ) + .await?; let body_start = Instant::now(); let body_result = read_json_body(data).await; crate::prometheus::observe_span( @@ -2184,7 +2197,9 @@ async fn handle_sql_invoke( // primitive-only non-fixed binds). The chain caveat — NOT the // invocation envelope's facts — is the source of truth so a holder // cannot widen or drop their grant by editing the invocation. - let constrained = chain_constrained; + let constrained = resolve_constrained_statement_caveat( + auth_result.sql_constrained_statement_candidates.clone(), + )?; let sql_request = if let Some(caveat) = &constrained { enforce_constrained_profile(caveat, sql_request)? } else { @@ -2308,34 +2323,65 @@ fn sql_request_requires_admin(request: &SqlRequest) -> bool { } } -/// W1 (D): walk the validated transitive delegation chain starting from the -/// invocation's directly-cited parents and return the first SQL -/// constrained-statement caveat present on any ancestor's persisted abilities -/// row. The persisted `caveats` JSON (NOT the invocation envelope's facts) is -/// the source of truth so a holder cannot widen or drop their grant by -/// editing the invocation. Walking ancestors closes the audit gap where a -/// child citing a no-caveat descendant would otherwise bypass an ancestor -/// caveat row. -async fn derive_chain_constrained_caveat( - tinycloud: &State, - parent_cids: &[tinycloud_auth::authorization::Cid], +/// W1 (D) / TC-411: resolve a set of distinct SQL constrained-statement +/// caveat candidates found on a validated proof chain into the single +/// effective caveat, or `None` if there were none. This is pure in-memory +/// selection over already-collected candidates -- no database access -- so +/// it is shared by the production path (candidates read straight out of the +/// request-scoped authorization snapshot, zero additional statements) and +/// the database-backed reference walk exercised by tests below. +/// +/// - zero distinct caveats found -> no constraint (unchanged behavior); +/// - exactly one distinct caveat found -> that caveat binds; +/// - more than one distinct caveat found -> the tightest one binds only if +/// it is a client of (contained by) every other candidate found; if no +/// such unique tightest caveat exists, the constraints are incomparable +/// and authorization fails closed rather than guessing (selection can +/// never depend on row order since it does not depend on discovery +/// order at all). +fn resolve_constrained_statement_caveat( + found: Vec, ) -> Result< Option, (Status, String), > { - if parent_cids.is_empty() { - return Ok(None); + use tinycloud_core::policy_capability::sql_caveat; + + match found.len() { + 0 => Ok(None), + 1 => Ok(found.into_iter().next()), + _ => { + // The effective caveat must be contained by every other + // candidate (i.e. it is the tightest of the set). Selection by + // structural containment, not by discovery order. + let tightest: Vec<_> = found + .iter() + .filter(|candidate| { + found + .iter() + .all(|other| sql_caveat::contains(other, candidate).is_ok()) + }) + .collect(); + match tightest.as_slice() { + [single] => Ok(Some((*single).clone())), + _ => Err(( + Status::Forbidden, + "sql_constrained_statements_ambiguous_chain".to_string(), + )), + } + } } - let conn = tinycloud - .readable() - .await - .map_err(|e| (Status::InternalServerError, e.to_string()))?; - derive_chain_constrained_caveat_with_conn(&conn, parent_cids).await } -/// W1 (D): the actual chain-walk against any seaorm `ConnectionTrait`. -/// Split out for direct test access without requiring a Rocket-managed -/// `State`. +/// W1 (D) / TC-411: database-backed reference implementation of the chain +/// walk that the production path replaced with an in-memory scan of the +/// request-scoped authorization snapshot (see +/// `AuthGraphSnapshot::constrained_statement_caveat_candidates` and +/// `resolve_constrained_statement_caveat` above). Test-only: kept to give +/// the selection algorithm direct database-backed coverage (multiple +/// roots/ancestors, ambiguous chains) without needing access to +/// `tinycloud-core`'s crate-private snapshot type from this crate's tests. +#[cfg(test)] async fn derive_chain_constrained_caveat_with_conn( conn: &C, parent_cids: &[tinycloud_auth::authorization::Cid], @@ -2353,10 +2399,14 @@ async fn derive_chain_constrained_caveat_with_conn = parent_cids.iter().copied().map(Hash::from).collect(); let mut visited: HashSet = HashSet::new(); + let mut found: Vec = + Vec::new(); while !frontier.is_empty() { let batch: Vec = frontier.drain(..).filter(|h| visited.insert(*h)).collect(); @@ -2370,12 +2420,14 @@ async fn derive_chain_constrained_caveat_with_conn { database_error_status(error) } + // TC-411: a declared SQL constrained-statement caveat on + // the chain failed to parse -- fail closed with the same + // status as an incomparable/ambiguous caveat selection + // (see `resolve_constrained_statement_caveat`), not the + // generic Unauthorized catch-all. + TxStoreError::Tx(TxError::MalformedSqlCaveat(_)) => Status::Forbidden, _ => Status::Unauthorized, }, e.to_string(), @@ -3184,10 +3249,11 @@ async fn verify_auth_admitted( span: &'static str, invocation: AdmittedInvocation, tinycloud: &State, + options: KvInvokeOptions, ) -> Result { let start = Instant::now(); let result = tinycloud - .invoke_admitted::(invocation, HashMap::new()) + .invoke_with_options_admitted::(invocation, HashMap::new(), options) .await .map_err(|e| { ( @@ -3196,6 +3262,8 @@ async fn verify_auth_admitted( TxStoreError::Tx(TxError::Db(error) | TxError::EpochInsert(error)) => { database_error_status(error) } + // TC-411: see the matching arm in `verify_auth` above. + TxStoreError::Tx(TxError::MalformedSqlCaveat(_)) => Status::Forbidden, _ => Status::Unauthorized, }, e.to_string(), @@ -4187,7 +4255,8 @@ mod tests { } #[tokio::test] - async fn w1_rocket_http_invoke_enforces_chain_constrained_sql_and_revoke() -> Result<()> { + async fn w1_rocket_http_invoke_enforces_chain_constrained_sql_with_kv_mutation_and_revoke( + ) -> Result<()> { use rocket::http::{ContentType, Header, Status}; use rocket::local::asynchronous::Client; use serde_json::json; @@ -4195,12 +4264,14 @@ mod tests { use tinycloud_auth::ssi::{dids::DIDURLBuf, ucan::Payload}; use tinycloud_auth::ucan_capabilities_object::Capabilities; use tinycloud_core::models::{ - abilities, actor, delegation as deleg_model, revocation as revo_model, + abilities, actor, current_kv, delegation as deleg_model, epoch, + invocation as invocation_model, kv_write, revocation as revo_model, space as space_model, }; + use tinycloud_core::relationships::event_order; use tinycloud_core::sea_orm::ActiveModelTrait; use tinycloud_core::sea_orm::ActiveValue::Set; - use tinycloud_core::types::{Caveats, SpaceIdWrap}; + use tinycloud_core::types::{Caveats, Metadata, Path, SpaceIdWrap}; let tempdir = TempDir::new()?; let db = Database::connect(ConnectOptions::new("sqlite::memory:".to_string())).await?; @@ -4296,6 +4367,16 @@ mod tests { None, None, ); + // Keep a KV mutation capability on every invocation below. `/invoke` + // dispatches to SQL when any SQL capability is present, while core + // authorization takes its mutation path when this `kv/del` is + // present. That mixed shape must not drop the SQL chain caveat. + let kv_resource: ResourceId = space.clone().to_resource( + "kv".parse::()?, + Some("discard".parse::()?), + None, + None, + ); let constrained_caveat = json!({ "mode": "constrained-statements", "readOnly": true, @@ -4315,6 +4396,68 @@ mod tests { } .insert(&conn) .await?; + abilities::ActiveModel { + delegation: Set(parent_hash), + resource: Set(Resource::TinyCloud(kv_resource.clone())), + ability: Set(Ability::try_from("tinycloud.kv/del".to_string()).unwrap()), + caveats: Set(Caveats::default()), + } + .insert(&conn) + .await?; + let seed_invocation = tinycloud_core::hash::hash(b"w1-mixed-kv-delete-invocation"); + let seed_epoch = tinycloud_core::hash::hash(b"w1-mixed-kv-delete-epoch"); + let seed_value = tinycloud_core::hash::hash(b"w1-mixed-kv-delete-value"); + invocation_model::ActiveModel { + id: Set(seed_invocation), + invoker: Set(space.did().to_string()), + issued_at: Set(OffsetDateTime::now_utc()), + facts: Set(None), + serialization: Set(b"w1-mixed-kv-delete-invocation".to_vec()), + } + .insert(&conn) + .await?; + epoch::ActiveModel { + seq: Set(0), + id: Set(seed_epoch), + space: Set(SpaceIdWrap(space.clone())), + } + .insert(&conn) + .await?; + event_order::ActiveModel { + seq: Set(0), + epoch: Set(seed_epoch), + epoch_seq: Set(0), + event: Set(seed_invocation), + space: Set(SpaceIdWrap(space.clone())), + } + .insert(&conn) + .await?; + let seed_write = kv_write::Model { + space: SpaceIdWrap(space.clone()), + key: Path::try_from("discard".to_string())?, + invocation: seed_invocation, + seq: 0, + epoch: seed_epoch, + epoch_seq: 0, + value: seed_value, + metadata: Metadata(Default::default()), + }; + kv_write::ActiveModel::from(seed_write.clone()) + .insert(&conn) + .await?; + current_kv::ActiveModel { + space: Set(seed_write.space), + key: Set(seed_write.key), + invocation: Set(seed_invocation), + seq: Set(0), + epoch: Set(seed_epoch), + epoch_seq: Set(0), + value: Set(seed_value), + metadata: Set(Metadata(Default::default())), + deleted: Set(false), + } + .insert(&conn) + .await?; let parent_cid: AuthCid = parent_hash.to_cid(0x55); let mut invocation_nb = std::collections::BTreeMap::new(); for (key, value) in constrained_caveat @@ -4323,13 +4466,20 @@ mod tests { { invocation_nb.insert(key.clone(), value.clone()); } - let make_auth_header = |nonce: &str| -> Result { + let make_auth_header = |nonce: &str, include_kv_delete: bool| -> Result { let mut invocation_caps = Capabilities::new(); invocation_caps.with_action( sql_resource.as_uri(), "tinycloud.sql/read".parse::()?, [invocation_nb.clone()], ); + if include_kv_delete { + invocation_caps.with_action( + kv_resource.as_uri(), + "tinycloud.kv/del".parse::()?, + [std::collections::BTreeMap::::new()], + ); + } let invocation = Payload { issuer: verification_method.parse::()?, audience: verification_method @@ -4358,7 +4508,7 @@ mod tests { .sign(jwk.get_algorithm().unwrap_or_default(), &jwk)?; Ok(invocation.encode()?) }; - let auth_header = make_auth_header("urn:uuid:00000000-0000-4000-8000-000000000001")?; + let auth_header = make_auth_header("urn:uuid:00000000-0000-4000-8000-000000000001", false)?; let rocket = rocket::build() .mount("/", rocket::routes![invoke]) @@ -4393,6 +4543,31 @@ mod tests { assert_eq!(json["rowCount"], 1); assert_eq!(json["rows"][0][0], 111); + // Regression: a mixed SQL + KV mutation invocation follows the core + // mutation branch. The persisted constrained-statements caveat must + // still win, so raw SQL remains forbidden rather than bypassing the + // chain profile. + let raw_auth_header = + make_auth_header("urn:uuid:00000000-0000-4000-8000-000000000003", true)?; + let response = client + .post("/invoke") + .header(Header::new("Authorization", raw_auth_header)) + .header(ContentType::JSON) + .body(serde_json::to_string(&SqlRequest::Execute { + schema: None, + sql: "SELECT val FROM labels WHERE label = 'beta'".to_string(), + params: vec![], + })?) + .dispatch() + .await; + let status = response.status(); + let body = response.into_string().await.unwrap_or_default(); + assert_eq!( + status, + Status::Forbidden, + "mixed SQL + KV invocation must preserve the chain SQL caveat: {body}" + ); + let response = client .post("/invoke") .header(Header::new("Authorization", auth_header.clone())) @@ -4420,7 +4595,7 @@ mod tests { .insert(&conn) .await?; - let auth_header = make_auth_header("urn:uuid:00000000-0000-4000-8000-000000000002")?; + let auth_header = make_auth_header("urn:uuid:00000000-0000-4000-8000-000000000002", false)?; let response = client .post("/invoke") .header(Header::new("Authorization", auth_header.clone()))