diff --git a/crates/engine/src/batch_get_item.rs b/crates/engine/src/batch_get_item.rs index 955fa4f53..d392c4a1c 100755 --- a/crates/engine/src/batch_get_item.rs +++ b/crates/engine/src/batch_get_item.rs @@ -16,12 +16,10 @@ use crate::OperationContext; use crate::capacity_helpers; use crate::create_table::storage_err_to_dynamo; use crate::expression_helpers::build_expression_maps; +use crate::request_limits; use crate::serialize_output; use crate::{DispatchMetrics, DispatchResult}; -/// Maximum number of keys across all tables in a single `BatchGetItem` request. -const MAX_BATCH_GET_KEYS: usize = 100; - /// Handle a `BatchGetItem` request. /// /// Reads items from one or more tables by primary key. Each table's keys are @@ -38,44 +36,15 @@ pub async fn handle_batch_get_item( let input: BatchGetItemInput = serde_json::from_value(body).map_err(crate::deserialize_error)?; - // Validate: RequestItems must not be empty. The message is the request - // model's length constraint (measured; note "Value at", with no value - // echoed). BatchWriteItem keeps its distinct required-parameter sentence, - // which is likewise measured. - if input.request_items.is_empty() { - return Err(DynamoDbError::ValidationException( - "1 validation error detected: Value at 'RequestItems' failed to satisfy \ - constraint: Member must have length greater than or equal to 1" - .to_owned(), - )); - } - - // Validate: per-table keys <= 100 - for (table_name, ka) in &input.request_items { - if ka.keys.len() > MAX_BATCH_GET_KEYS { - return Err(DynamoDbError::ValidationException(format!( - "1 validation error detected: Value at 'RequestItems.{table_name}.member.Keys' failed to satisfy constraint: \ - Member must have length less than or equal to 100" - ))); - } - } - - // Validate: total keys across all tables <= 100 - let total_keys: usize = input.request_items.values().map(|ka| ka.keys.len()).sum(); - if total_keys > MAX_BATCH_GET_KEYS { - return Err(DynamoDbError::ValidationException( - "Too many items requested for the BatchGetItem call".to_owned(), - )); - } - - // Validate: each table must have at least one key - for (table_name, ka) in &input.request_items { - if ka.keys.is_empty() { - return Err(DynamoDbError::ValidationException(format!( - "1 validation error detected: Value '[]' at 'requestItems.{table_name}.member.keys' failed to satisfy constraint: Member must have length greater than or equal to 1" - ))); - } - } + // Validate the table and key counts: RequestItems not empty, at most 100 + // tables, at most 100 keys per table and in total, at least one key per + // table. The server runs the same checks before authorization. + let counts: Vec<(&str, Option)> = input + .request_items + .iter() + .map(|(table_name, ka)| (table_name.as_str(), Some(ka.keys.len()))) + .collect(); + request_limits::check_batch_counts("BatchGetItem", &counts)?; // Reject mixing expression and non-expression parameters anywhere in the // request. The check is request-wide: ProjectionExpression on one table and @@ -189,8 +158,7 @@ pub async fn handle_batch_get_item( for (table_name, ka) in &input.request_items { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, table_name) + .table_key_info_uncached(table_name) .await .map_err(storage_err_to_dynamo)?; diff --git a/crates/engine/src/batch_write_item.rs b/crates/engine/src/batch_write_item.rs index e285a921a..b0381c7a7 100755 --- a/crates/engine/src/batch_write_item.rs +++ b/crates/engine/src/batch_write_item.rs @@ -21,13 +21,11 @@ use extenddb_core::validation::{ use crate::OperationContext; use crate::capacity_helpers; use crate::create_table::storage_err_to_dynamo; +use crate::request_limits; use crate::serialize_output; use crate::stream_capture; use crate::{DispatchMetrics, DispatchResult}; -/// Maximum number of write requests across all tables in a single `BatchWriteItem`. -const MAX_BATCH_WRITE_ITEMS: usize = 25; - /// Handle a `BatchWriteItem` request. /// /// Writes (put or delete) items across one or more tables. Each operation is @@ -44,45 +42,15 @@ pub async fn handle_batch_write_item( let input: BatchWriteItemInput = serde_json::from_value(body).map_err(crate::deserialize_error)?; - // Validate: RequestItems must not be empty - if input.request_items.is_empty() { - return Err(DynamoDbError::ValidationException( - "The requestItems parameter is required for BatchWriteItem".to_owned(), - )); - } - - // Validate: per-table operations <= 25 - for (table_name, reqs) in &input.request_items { - if reqs.len() > MAX_BATCH_WRITE_ITEMS { - let items_repr = reqs - .iter() - .map(|_| "WriteRequest") - .collect::>() - .join(", "); - return Err(DynamoDbError::ValidationException(format!( - "1 validation error detected: Value '{{{table_name}=[{items_repr}]}}' at 'requestItems' failed to satisfy constraint: \ - Map value must satisfy constraint: [Member must have length less than or equal to 25, \ - Member must have length greater than or equal to 1]" - ))); - } - } - - // Validate: total operations across all tables <= 25 - let total_ops: usize = input.request_items.values().map(std::vec::Vec::len).sum(); - if total_ops > MAX_BATCH_WRITE_ITEMS { - return Err(DynamoDbError::ValidationException( - "Too many items requested for the BatchWriteItem call".to_owned(), - )); - } - - // Validate: each table must have at least one request - for (table_name, reqs) in &input.request_items { - if reqs.is_empty() { - return Err(DynamoDbError::ValidationException(format!( - "1 validation error detected: Value '[]' at 'requestItems.{table_name}.member' failed to satisfy constraint: Member must have length greater than or equal to 1" - ))); - } - } + // Validate the table and request counts: RequestItems not empty, at most + // 25 tables, at most 25 requests per table and in total, at least one + // request per table. The server runs the same checks before authorization. + let counts: Vec<(&str, Option)> = input + .request_items + .iter() + .map(|(table_name, reqs)| (table_name.as_str(), Some(reqs.len()))) + .collect(); + request_limits::check_batch_counts("BatchWriteItem", &counts)?; // Validate: each WriteRequest has exactly one of PutRequest or DeleteRequest for reqs in input.request_items.values() { @@ -113,8 +81,7 @@ pub async fn handle_batch_write_item( for (table_name, reqs) in &input.request_items { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, table_name) + .table_key_info_uncached(table_name) .await .map_err(storage_err_to_dynamo)?; diff --git a/crates/engine/src/create_table.rs b/crates/engine/src/create_table.rs index 257cd2299..6375112c8 100755 --- a/crates/engine/src/create_table.rs +++ b/crates/engine/src/create_table.rs @@ -110,7 +110,16 @@ pub async fn handle_create_table( serialize_output(&output) } -pub(crate) fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> DynamoDbError { +/// Map a storage-layer error to its `DynamoDB` wire error. +/// +/// Exposed so the server's authorization path classifies a key-info lookup +/// fault the same way the engine classifies the identical fault on the data +/// path: a transient or connection error becomes `ServiceUnavailable`, the +/// catalog and backend-internal errors become `InternalServerError`, and a +/// not-active table becomes `ResourceInUseException`. Without a single mapper +/// the same database hiccup returned different wire errors depending on which +/// layer hit it first. +pub fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> DynamoDbError { use extenddb_storage::error::StorageError; match e { StorageError::TableNotFound(name) => DynamoDbError::ResourceNotFoundException( diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index 114e240a6..b31461533 100755 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -28,6 +28,7 @@ mod list_tables; mod put_item; mod query; mod read_helpers; +mod request_limits; mod scan; mod search_vectors; pub mod stream_capture; @@ -44,6 +45,7 @@ mod vector_gate; pub use batch_get_item::handle_batch_get_item; pub use batch_write_item::handle_batch_write_item; pub use create_table::handle_create_table; +pub use create_table::storage_err_to_dynamo; pub use delete_item::handle_delete_item; pub use delete_table::handle_delete_table; pub use describe_endpoints::handle_describe_endpoints; @@ -54,6 +56,7 @@ pub use import_export::{handle_export_table, handle_import_table}; pub use list_tables::handle_list_tables; pub use put_item::handle_put_item; pub use query::handle_query; +pub use request_limits::check_request_item_counts; pub use scan::handle_scan; pub use search_vectors::handle_search_vectors; pub use streams::{ @@ -71,6 +74,7 @@ use std::sync::Arc; use extenddb_core::error::DynamoDbError; use extenddb_core::limits::LimitsConfig; +use extenddb_core::types::TableKeyInfo; /// Check whether an operation name is recognized by the dispatch table. /// @@ -315,17 +319,24 @@ impl DispatchResult { } } +/// The `TableKeyInfo` the authorization layer read for a request, by table +/// name. A `None` value records a table that authorization found missing. +pub type AuthorizedKeyInfo = std::collections::HashMap>; + /// Context passed to every operation handler. /// Fix #9: region and `account_id` use Arc to avoid per-request cloning. /// /// # Catalog Query Discipline (P118) /// -/// For single-table item operations (`GetItem`, `PutItem`, `DeleteItem`, -/// `UpdateItem`, `Query`, `Scan`), the auth layer pre-fetches `TableKeyInfo` -/// and stores it in `pre_fetched_key_info`. Engine handlers MUST use this -/// pre-fetched value instead of calling `storage.table_key_info()` directly. -/// New per-request catalog roundtrips require justification in the discussion -/// file and principal reviewer approval. +/// The auth layer reads `TableKeyInfo` for every table whose partition key it +/// evaluates (`dynamodb:LeadingKeys`) and stores what it read in +/// `authorized_key_info`. Engine handlers MUST resolve those tables from this +/// snapshot instead of reading the catalog again, through +/// [`OperationContext::table_key_info`] or +/// [`OperationContext::table_key_info_uncached`]. Authorization and the engine +/// then act on the same key schema, and on the same answer to whether the +/// table exists. New per-request catalog roundtrips require justification in +/// the discussion file and principal reviewer approval. pub struct OperationContext { pub storage: Arc, pub limits: Arc, @@ -337,34 +348,56 @@ pub struct OperationContext { /// Allowed directories for export file operations. Empty means exports /// are disabled (secure default). pub export_paths: Arc<[Arc]>, - /// Pre-fetched `TableKeyInfo` from the auth layer (P118 optimization #2). - /// Populated for single-table item-level operations; `None` for table-level - /// and batch/transact operations. - pub pre_fetched_key_info: Option, + /// The `TableKeyInfo` the auth layer read for this request (P118 + /// optimization #2). Populated for single-table item operations, `Query`, + /// `Scan`, and the batch and transaction operations; empty for table-level + /// operations. In developer mode authorization is skipped after the + /// single-table read, so only a single-table operation has an entry. + pub authorized_key_info: AuthorizedKeyInfo, /// Auth/authz cache handles. Used by control-plane and tagging operations /// to issue write-through cache invalidations after the underlying state /// changes (e.g. `TagResource` invalidates the resource-tags cache). pub auth_cache: extenddb_auth::AuthCacheRegistry, - /// Optional cached `TableKeyInfo` lookup. When set, batch / transact / - /// multi-table engine handlers route through this instead of calling - /// `storage.table_key_info` directly. When unset (e.g. unit tests), the - /// engine falls back to direct storage lookups. + /// Optional cached `TableKeyInfo` lookup, used by + /// [`OperationContext::table_key_info`] for a table the auth layer did not + /// read. When unset (e.g. unit tests), the engine falls back to direct + /// storage lookups. pub table_key_info_lookup: Option>, } impl OperationContext { - /// Return pre-fetched `TableKeyInfo` if available and matching the requested - /// table, otherwise fetch from storage. This is the single entry point for - /// obtaining `TableKeyInfo` in engine handlers (P118 Catalog Query Discipline). + /// The auth layer's answer for `table_name`, if it read that table: the key + /// info it used, or `TableNotFound` when it found the table missing. A + /// missing table stays missing for the rest of the request even if it has + /// since become ACTIVE, because authorization left `dynamodb:LeadingKeys` + /// absent for it and an absent key satisfies a `ForAllValues` condition. + /// The snapshot is authoritative for every table it names: an entry for + /// another account is an internal error, never a reason to read the table + /// again. + fn authorized_table_key_info( + &self, + table_name: &str, + ) -> Option> { + use extenddb_storage::error::StorageError; + Some(match self.authorized_key_info.get(table_name)? { + Some(ki) if *ki.account_id == *self.account_id => Ok(ki.clone()), + Some(_) => Err(StorageError::Internal(format!( + "authorized key info for table {table_name} belongs to another account" + ))), + None => Err(StorageError::TableNotFound(table_name.to_owned())), + }) + } + + /// `TableKeyInfo` for a single-table operation: the auth layer's snapshot + /// when it read the table, otherwise the cached lookup, otherwise storage. + /// This is the entry point for single-table handlers (P118 Catalog Query + /// Discipline). pub(crate) async fn table_key_info( &self, table_name: &str, - ) -> Result { - if let Some(ref ki) = self.pre_fetched_key_info - && ki.table_name == table_name - && *ki.account_id == *self.account_id - { - return Ok(ki.clone()); + ) -> Result { + if let Some(result) = self.authorized_table_key_info(table_name) { + return result; } if let Some(ref lookup) = self.table_key_info_lookup { return lookup.lookup(&self.account_id, table_name).await; @@ -373,6 +406,23 @@ impl OperationContext { .table_key_info(&self.account_id, table_name) .await } + + /// `TableKeyInfo` for a batch or transaction handler: the auth layer's + /// snapshot when it read the table, otherwise a direct storage read. The + /// fallback skips the cache, as these handlers always have, so a request + /// that authorization did not evaluate (developer mode) sees the catalog as + /// it is now rather than a cached or negative entry. + pub(crate) async fn table_key_info_uncached( + &self, + table_name: &str, + ) -> Result { + if let Some(result) = self.authorized_table_key_info(table_name) { + return result; + } + self.storage + .table_key_info(&self.account_id, table_name) + .await + } } /// Dispatch an operation by name. diff --git a/crates/engine/src/request_limits.rs b/crates/engine/src/request_limits.rs new file mode 100644 index 000000000..385e34e4e --- /dev/null +++ b/crates/engine/src/request_limits.rs @@ -0,0 +1,370 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Count limits of the batch and transaction operations, and the +//! `ValidationException` each limit returns. +//! +//! The handlers check these limits on the parsed request. The server checks +//! them on the raw request body before authorization, through +//! [`check_request_item_counts`], because `DynamoDB` returns these errors before +//! it evaluates any policy: an empty or 101-item transaction, an empty +//! `RequestItems`, a table with no keys or requests, a 26-request +//! `BatchWriteItem` and a 101-key `BatchGetItem` each return +//! `ValidationException` to a principal with no permissions (measured +//! 2026-10-21). Checking first also bounds the work authorization does per +//! request: at most 100 tables and 100 keys for `BatchGetItem`, 25 tables and +//! 25 requests for `BatchWriteItem`, and 100 items for a transaction, instead of +//! a number set by the body size. Authorization walks every sub-operation of a +//! `TransactWriteItems` item before the handler's one-operation rule rejects an +//! item with several, so a transaction reads at most 400 tables (100 items, four +//! sub-operations each) and evaluates at most 400 targets. Both paths run +//! [`check_batch_counts`] and +//! [`check_transact_count`], so they apply the checks in one order and return +//! the same text. + +use extenddb_core::error::DynamoDbError; +use serde_json::Value; + +/// Maximum number of keys across all tables in one `BatchGetItem` request, +/// and the maximum number of tables in its `RequestItems`. +pub(crate) const MAX_BATCH_GET_KEYS: usize = 100; +/// Maximum number of write requests across all tables in one `BatchWriteItem`, +/// and the maximum number of tables in its `RequestItems`. +pub(crate) const MAX_BATCH_WRITE_ITEMS: usize = 25; +/// Maximum number of items in one `TransactGetItems` or `TransactWriteItems`. +pub(crate) const MAX_TRANSACT_ITEMS: usize = 100; + +/// Check the item count of a transaction. `member` is `TransactGetItem` or +/// `TransactWriteItem`. Over the limit, the message lists the member name once +/// per item; the service lists each item's full serialization instead, so the +/// count matches and the item text does not. +/// +/// # Errors +/// +/// `ValidationException` when `count` is 0 or above [`MAX_TRANSACT_ITEMS`]. +pub(crate) fn check_transact_count(count: usize, member: &str) -> Result<(), DynamoDbError> { + if count == 0 { + return Err(DynamoDbError::ValidationException( + "1 validation error detected: Value '[]' at 'transactItems' failed to satisfy constraint: Member must have length greater than or equal to 1".to_owned(), + )); + } + if count > MAX_TRANSACT_ITEMS { + let items = vec![member; count].join(", "); + return Err(DynamoDbError::ValidationException(format!( + "1 validation error detected: Value '[{items}]' at 'transactItems' failed to satisfy constraint: Member must have length less than or equal to 100" + ))); + } + Ok(()) +} + +/// The write-request list of one `BatchWriteItem` table as the service echoes +/// it in a validation message, with each request printed as its type name. +fn write_requests_repr(count: usize) -> String { + vec!["WriteRequest"; count].join(", ") +} + +/// Check the table and item counts of a `BatchGetItem` or `BatchWriteItem`. +/// +/// `tables` lists each entry of `RequestItems` with its key count +/// (`BatchGetItem`) or write-request count (`BatchWriteItem`), or `None` when +/// the entry is not the expected shape; the handler's deserialization rejects +/// such an entry, so it counts towards the number of tables only. The checks +/// run in this order: an empty `RequestItems`, too many tables, too many items +/// in one table, too many items in total, a table with no items. +/// +/// # Errors +/// +/// `ValidationException` for the first check that fails. +pub(crate) fn check_batch_counts( + operation: &str, + tables: &[(&str, Option)], +) -> Result<(), DynamoDbError> { + let get = operation == "BatchGetItem"; + let limit = if get { + MAX_BATCH_GET_KEYS + } else { + MAX_BATCH_WRITE_ITEMS + }; + + if tables.is_empty() { + return Err(DynamoDbError::ValidationException(if get { + "1 validation error detected: Value at 'RequestItems' failed to satisfy constraint: Member must have length greater than or equal to 1".to_owned() + } else { + "The requestItems parameter is required for BatchWriteItem".to_owned() + })); + } + + if tables.len() > limit { + return Err(DynamoDbError::ValidationException(if get { + "1 validation error detected: Value at 'RequestItems' failed to satisfy constraint: Member must have length less than or equal to 100".to_owned() + } else { + let map = tables + .iter() + .map(|(t, n)| format!("{t}=[{}]", write_requests_repr(n.unwrap_or(0)))) + .collect::>() + .join(", "); + format!( + "1 validation error detected: Value '{{{map}}}' at 'requestItems' failed to satisfy constraint: Member must have length less than or equal to 25" + ) + })); + } + + for &(table_name, count) in tables { + if let Some(count) = count + && count > limit + { + return Err(per_table_count_error(get, table_name, count)); + } + } + + let total: usize = tables.iter().filter_map(|(_, n)| *n).sum(); + if total > limit { + return Err(DynamoDbError::ValidationException(format!( + "Too many items requested for the {operation} call" + ))); + } + + for &(table_name, count) in tables { + if count == Some(0) { + return Err(per_table_count_error(get, table_name, 0)); + } + } + Ok(()) +} + +/// One table's key or write-request count is 0 or above the limit. The service +/// returns one message for both bounds of a `BatchWriteItem` table and one +/// message per bound for a `BatchGetItem` table (measured 2026-10-21). +fn per_table_count_error(get: bool, table_name: &str, count: usize) -> DynamoDbError { + if get { + let bound = if count == 0 { + "greater than or equal to 1" + } else { + "less than or equal to 100" + }; + DynamoDbError::ValidationException(format!( + "1 validation error detected: Value at 'RequestItems.{table_name}.member.Keys' failed to satisfy constraint: Member must have length {bound}" + )) + } else { + DynamoDbError::ValidationException(format!( + "1 validation error detected: Value '{{{table_name}=[{}]}}' at 'requestItems' failed to satisfy constraint: \ + Map value must satisfy constraint: [Member must have length less than or equal to 25, \ + Member must have length greater than or equal to 1]", + write_requests_repr(count) + )) + } +} + +/// Refuse a batch or transaction request whose counts fall outside the +/// operation's limits, reading the counts from the raw request body. +/// +/// Runs [`check_batch_counts`] or [`check_transact_count`], the checks the +/// handlers run on the parsed request. A body whose top-level container is +/// missing or not the expected type is left to the handler, which rejects it +/// when it parses the request. Any other operation passes. +/// +/// # Errors +/// +/// Returns the `ValidationException` the operation's handler returns for the +/// same request. +pub fn check_request_item_counts(operation: &str, body: &Value) -> Result<(), DynamoDbError> { + match operation { + "TransactGetItems" | "TransactWriteItems" => { + let member = if operation == "TransactGetItems" { + "TransactGetItem" + } else { + "TransactWriteItem" + }; + if let Some(items) = body.get("TransactItems").and_then(Value::as_array) { + check_transact_count(items.len(), member)?; + } + } + "BatchGetItem" | "BatchWriteItem" => { + if let Some(map) = body.get("RequestItems").and_then(Value::as_object) { + let tables: Vec<(&str, Option)> = map + .iter() + .map(|(table_name, entry)| { + let items = if operation == "BatchGetItem" { + entry.get("Keys") + } else { + Some(entry) + }; + ( + table_name.as_str(), + items.and_then(Value::as_array).map(Vec::len), + ) + }) + .collect(); + check_batch_counts(operation, &tables)?; + } + } + _ => {} + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn message(result: Result<(), DynamoDbError>) -> String { + match result { + Err(DynamoDbError::ValidationException(m)) => m, + other => panic!("expected ValidationException, got {other:?}"), + } + } + + fn transact(op: &str, n: usize) -> Value { + let entry = if op == "TransactGetItems" { + json!({"Get": {"TableName": "t", "Key": {"pk": {"S": "a"}}}}) + } else { + json!({"ConditionCheck": {"TableName": "t", "Key": {"pk": {"S": "a"}}, + "ConditionExpression": "attribute_exists(pk)"}}) + }; + json!({ "TransactItems": vec![entry; n] }) + } + + fn bgi(tables: &[(&str, usize)]) -> Value { + let key = json!({"pk": {"S": "a"}}); + let map: serde_json::Map = tables + .iter() + .map(|(t, n)| ((*t).to_owned(), json!({"Keys": vec![key.clone(); *n]}))) + .collect(); + json!({ "RequestItems": map }) + } + + fn bwi(tables: &[(&str, usize)]) -> Value { + let put = json!({"PutRequest": {"Item": {"pk": {"S": "a"}}}}); + let map: serde_json::Map = tables + .iter() + .map(|(t, n)| ((*t).to_owned(), json!(vec![put.clone(); *n]))) + .collect(); + json!({ "RequestItems": map }) + } + + fn names(n: usize) -> Vec { + (0..n).map(|i| format!("t{i:03}")).collect() + } + + #[test] + fn transactions_outside_the_limits_fail() { + for (op, member) in [ + ("TransactGetItems", "TransactGetItem"), + ("TransactWriteItems", "TransactWriteItem"), + ] { + assert!(check_request_item_counts(op, &transact(op, 1)).is_ok()); + assert!(check_request_item_counts(op, &transact(op, 100)).is_ok()); + let m = message(check_request_item_counts(op, &transact(op, 101))); + assert!(m.starts_with(&format!("1 validation error detected: Value '[{member}, "))); + assert!(m.ends_with("Member must have length less than or equal to 100")); + assert_eq!(m.matches(member).count(), 101); + assert_eq!( + message(check_request_item_counts(op, &transact(op, 0))), + "1 validation error detected: Value '[]' at 'transactItems' failed to satisfy constraint: Member must have length greater than or equal to 1" + ); + } + } + + #[test] + fn batch_get_limits_in_order() { + assert!(check_request_item_counts("BatchGetItem", &bgi(&[("a", 100)])).is_ok()); + assert_eq!( + message(check_request_item_counts("BatchGetItem", &bgi(&[]))), + "1 validation error detected: Value at 'RequestItems' failed to satisfy constraint: Member must have length greater than or equal to 1" + ); + // 101 tables of one key each: the table count is reported, not the total. + let many = names(101); + let tables: Vec<(&str, usize)> = many.iter().map(|t| (t.as_str(), 1)).collect(); + assert_eq!( + message(check_request_item_counts("BatchGetItem", &bgi(&tables))), + "1 validation error detected: Value at 'RequestItems' failed to satisfy constraint: Member must have length less than or equal to 100" + ); + // 101 tables with no keys: still the table count. + let tables: Vec<(&str, usize)> = many.iter().map(|t| (t.as_str(), 0)).collect(); + assert!( + message(check_request_item_counts("BatchGetItem", &bgi(&tables))) + .ends_with("Member must have length less than or equal to 100") + ); + assert_eq!( + message(check_request_item_counts( + "BatchGetItem", + &bgi(&[("a", 101)]) + )), + "1 validation error detected: Value at 'RequestItems.a.member.Keys' failed to satisfy constraint: Member must have length less than or equal to 100" + ); + assert_eq!( + message(check_request_item_counts( + "BatchGetItem", + &bgi(&[("a", 60), ("b", 41)]) + )), + "Too many items requested for the BatchGetItem call" + ); + assert_eq!( + message(check_request_item_counts( + "BatchGetItem", + &bgi(&[("a", 1), ("b", 0)]) + )), + "1 validation error detected: Value at 'RequestItems.b.member.Keys' failed to satisfy constraint: Member must have length greater than or equal to 1" + ); + } + + #[test] + fn batch_write_limits_in_order() { + assert!(check_request_item_counts("BatchWriteItem", &bwi(&[("a", 25)])).is_ok()); + assert_eq!( + message(check_request_item_counts("BatchWriteItem", &bwi(&[]))), + "The requestItems parameter is required for BatchWriteItem" + ); + let many = names(26); + let tables: Vec<(&str, usize)> = many.iter().map(|t| (t.as_str(), 1)).collect(); + let m = message(check_request_item_counts("BatchWriteItem", &bwi(&tables))); + assert!(m.starts_with("1 validation error detected: Value '{t000=[WriteRequest], ")); + assert!(m.ends_with( + "}' at 'requestItems' failed to satisfy constraint: Member must have length less than or equal to 25" + )); + let m = message(check_request_item_counts( + "BatchWriteItem", + &bwi(&[("a", 26)]), + )); + assert!(m.starts_with("1 validation error detected: Value '{a=[WriteRequest, ")); + assert!(m.ends_with("Map value must satisfy constraint: [Member must have length less than or equal to 25, Member must have length greater than or equal to 1]")); + assert_eq!( + message(check_request_item_counts( + "BatchWriteItem", + &bwi(&[("a", 13), ("b", 13)]) + )), + "Too many items requested for the BatchWriteItem call" + ); + assert_eq!( + message(check_request_item_counts( + "BatchWriteItem", + &bwi(&[("a", 1), ("b", 0)]) + )), + "1 validation error detected: Value '{b=[]}' at 'requestItems' failed to satisfy constraint: Map value must satisfy constraint: [Member must have length less than or equal to 25, Member must have length greater than or equal to 1]" + ); + } + + #[test] + fn malformed_entries_count_as_tables_only() { + // A table entry of the wrong shape is left to the handler, but it still + // counts towards the number of tables. + assert!( + check_request_item_counts("BatchWriteItem", &json!({"RequestItems": {"a": 1}})).is_ok() + ); + assert!( + check_request_item_counts("BatchGetItem", &json!({"RequestItems": {"a": {}}})).is_ok() + ); + let map: serde_json::Map = + names(101).into_iter().map(|t| (t, json!({}))).collect(); + assert!( + message(check_request_item_counts( + "BatchGetItem", + &json!({ "RequestItems": map }) + )) + .ends_with("Member must have length less than or equal to 100") + ); + assert!(check_request_item_counts("TransactWriteItems", &json!({})).is_ok()); + assert!(check_request_item_counts("Query", &transact("TransactGetItems", 101)).is_ok()); + } +} diff --git a/crates/engine/src/transact_get_items.rs b/crates/engine/src/transact_get_items.rs index 761213172..ec6825cff 100755 --- a/crates/engine/src/transact_get_items.rs +++ b/crates/engine/src/transact_get_items.rs @@ -21,12 +21,10 @@ use crate::OperationContext; use crate::capacity_helpers; use crate::create_table::storage_err_to_dynamo; use crate::expression_helpers::build_expression_maps; +use crate::request_limits; use crate::serialize_output; use crate::{DispatchMetrics, DispatchResult}; -/// Maximum number of items in a single `TransactGetItems` request. -const MAX_TRANSACT_GET_ITEMS: usize = 100; - /// Handle a `TransactGetItems` request. /// /// Reads up to 100 items atomically in a single consistent snapshot. @@ -42,23 +40,9 @@ pub async fn handle_transact_get_items( let input: TransactGetItemsInput = serde_json::from_value(body).map_err(crate::deserialize_error)?; - if input.transact_items.is_empty() { - return Err(DynamoDbError::ValidationException( - "1 validation error detected: Value '[]' at 'transactItems' failed to satisfy constraint: Member must have length greater than or equal to 1".to_owned(), - )); - } - - if input.transact_items.len() > MAX_TRANSACT_GET_ITEMS { - return Err(DynamoDbError::ValidationException(format!( - "1 validation error detected: Value '[{}]' at 'transactItems' failed to satisfy constraint: Member must have length less than or equal to 100", - input - .transact_items - .iter() - .map(|_| "TransactGetItem") - .collect::>() - .join(", ") - ))); - } + // At least one and at most 100 items. The server runs the same check + // before authorization. + request_limits::check_transact_count(input.transact_items.len(), "TransactGetItem")?; // Resolve table key info for each item let mut seen_keys: HashSet> = HashSet::with_capacity(input.transact_items.len()); @@ -75,8 +59,7 @@ pub async fn handle_transact_get_items( let mut key_infos = Vec::with_capacity(input.transact_items.len()); for tgi in &input.transact_items { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, &tgi.get.table_name) + .table_key_info_uncached(&tgi.get.table_name) .await .map_err(storage_err_to_dynamo)?; // Key type validation is deferred to the storage layer so that diff --git a/crates/engine/src/transact_write_items.rs b/crates/engine/src/transact_write_items.rs index cc4a8951b..43fb78a1c 100755 --- a/crates/engine/src/transact_write_items.rs +++ b/crates/engine/src/transact_write_items.rs @@ -11,6 +11,7 @@ use crate::OperationContext; use crate::capacity_helpers; use crate::create_table::storage_err_to_dynamo; use crate::expression_helpers::{build_expression_maps, parse_optional_condition}; +use crate::request_limits; use crate::serialize_output; use crate::stream_capture; use crate::transact_write_helpers::{ @@ -27,9 +28,6 @@ use extenddb_core::validation::{ validate_item_nesting_depth, validate_item_size, validate_key_not_empty, }; -/// Maximum number of items in a single `TransactWriteItems` request. -const MAX_TRANSACT_WRITE_ITEMS: usize = 100; - /// Handle a `TransactWriteItems` request. /// /// Executes up to 100 write operations atomically. All operations succeed @@ -55,23 +53,9 @@ pub async fn handle_transact_write_items( .map(|t| compute_fingerprint(&body, t)) .unwrap_or_default(); - if input.transact_items.is_empty() { - return Err(DynamoDbError::ValidationException( - "1 validation error detected: Value '[]' at 'transactItems' failed to satisfy constraint: Member must have length greater than or equal to 1".to_owned(), - )); - } - - if input.transact_items.len() > MAX_TRANSACT_WRITE_ITEMS { - let items_repr = input - .transact_items - .iter() - .map(|_| "TransactWriteItem") - .collect::>() - .join(", "); - return Err(DynamoDbError::ValidationException(format!( - "1 validation error detected: Value '[{items_repr}]' at 'transactItems' failed to satisfy constraint: Member must have length less than or equal to 100" - ))); - } + // At least one and at most 100 items. The server runs the same check + // before authorization. + request_limits::check_transact_count(input.transact_items.len(), "TransactWriteItem")?; // Validate each item has exactly one operation for (i, twi) in input.transact_items.iter().enumerate() { @@ -364,8 +348,7 @@ async fn prepare_write_op( ) -> Result { if let Some(put) = &twi.put { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, &put.table_name) + .table_key_info_uncached(&put.table_name) .await .map_err(storage_err_to_dynamo)?; validate_item_nesting_depth(&put.item)?; @@ -418,8 +401,7 @@ async fn prepare_write_op( if let Some(del) = &twi.delete { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, &del.table_name) + .table_key_info_uncached(&del.table_name) .await .map_err(storage_err_to_dynamo)?; // Empty or oversize key values are up-front input validation in @@ -470,8 +452,7 @@ async fn prepare_write_op( if let Some(upd) = &twi.update { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, &upd.table_name) + .table_key_info_uncached(&upd.table_name) .await .map_err(storage_err_to_dynamo)?; // Empty or oversize key values are up-front input validation in @@ -541,8 +522,7 @@ async fn prepare_write_op( if let Some(cc) = &twi.condition_check { let key_info = ctx - .storage - .table_key_info(&ctx.account_id, &cc.table_name) + .table_key_info_uncached(&cc.table_name) .await .map_err(storage_err_to_dynamo)?; // Empty or oversize key values are up-front input validation in diff --git a/crates/server/src/handler.rs b/crates/server/src/handler.rs index 7d609c3e0..ef3765bda 100755 --- a/crates/server/src/handler.rs +++ b/crates/server/src/handler.rs @@ -107,7 +107,7 @@ pub(crate) async fn handle_request( // --- Authz segment --- let authz_start = std::time::Instant::now(); - let pre_fetched_key_info; + let authorized_key_info; { if state.catalog_store.is_none() { tracing::error!("Authorization required but catalog_store is not configured"); @@ -137,7 +137,7 @@ pub(crate) async fn handle_request( ) .await { - Ok(ki) => pre_fetched_key_info = ki, + Ok(ki) => authorized_key_info = ki, Err(e) => return error_response(&e, &request_id), } } @@ -152,7 +152,7 @@ pub(crate) async fn handle_request( account_id, import_paths: state.import_paths.clone(), export_paths: state.export_paths.clone(), - pre_fetched_key_info, + authorized_key_info, auth_cache: state.auth_cache.clone(), table_key_info_lookup: Some( state.table_key_info_cache.clone() as Arc diff --git a/crates/server/src/key_info_cache.rs b/crates/server/src/key_info_cache.rs index b8b53d594..def236150 100644 --- a/crates/server/src/key_info_cache.rs +++ b/crates/server/src/key_info_cache.rs @@ -1,16 +1,18 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 -//! `CachedTableKeyInfoStore` — stale-while-revalidate cache for +//! `CachedTableKeyInfoStore`, a stale-while-revalidate cache for //! `TableKeyInfo` lookups. //! -//! `TableKeyInfo` is fetched per request by the auth path -//! (`request_helpers.rs::authorize_request` for `LeadingKeys` extraction) -//! and by engine handlers that operate on multiple tables (batch / transact -//! operations). Caching it eliminates a catalog roundtrip per request on the -//! steady-state hot path. +//! `TableKeyInfo` is fetched through this cache by the auth path for the +//! single-table operations (`request_helpers.rs::authorize_request`, for +//! `LeadingKeys` extraction) and by single-table engine handlers for a table +//! the auth path did not read. The batch and transaction operations do not use +//! it: authorization reads their tables from storage and the engine resolves +//! them from what authorization read. Caching it eliminates a catalog +//! roundtrip per request on the steady-state hot path. //! -//! Negative results — `Err(TableNotFound)` or table missing — are stored as +//! Negative results (`Err(TableNotFound)` or table missing) are stored as //! `None` and capped at the configured `negative_ttl` (defaults to 5s) so //! newly-created tables become available quickly without restart. //! @@ -109,16 +111,6 @@ impl CachedTableKeyInfoStore { } } - /// Same as `get`, but returns `Ok(None)` for missing tables — matches the - /// `.ok()` semantic used by the auth path's `LeadingKeys` extraction. - pub async fn get_optional(&self, account_id: &str, table_name: &str) -> Option { - self.cache - .get((account_id.to_owned(), table_name.to_owned())) - .await - .ok() - .flatten() - } - /// Drop the cached entry for `(account_id, table_name)`. /// /// Called by engine handlers after `CreateTable`, `UpdateTable`, diff --git a/crates/server/src/leading_keys.rs b/crates/server/src/leading_keys.rs new file mode 100644 index 000000000..8be8e7cb8 --- /dev/null +++ b/crates/server/src/leading_keys.rs @@ -0,0 +1,515 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `dynamodb:LeadingKeys` extraction for request authorization. +//! +//! IAM evaluates `dynamodb:LeadingKeys` against the partition key value(s) a +//! request touches. The server populates the key for every operation that +//! targets one or more specific partitions, so a `ForAllValues`/`ForAnyValue` +//! LeadingKeys condition on such a request is evaluated against the partitions +//! it reads or writes, not vacuously. `Scan` and `SearchVectors` target no +//! single partition: they reach the engine, return data, and carry no +//! LeadingKeys, so a `ForAllValues` condition is satisfied vacuously (the read +//! is allowed) and an operator fences them with `ForAnyValue` or a `Null` +//! guard, which is how the service treats a `Scan`. Operations the engine +//! rejects before any data flows (an unparseable or non-equality key condition, +//! a missing key condition, an unresolved placeholder name, a table that does +//! not exist) are left with the key absent on purpose: the engine returns +//! `ValidationException` or `ResourceNotFoundException`, which is what live +//! `DynamoDB` returns for those shapes even under a LeadingKeys policy (measured +//! 2026-10-20). +//! +//! Behaviour measured against live `DynamoDB`: +//! +//! - Item operations (`GetItem`, `PutItem`, `UpdateItem`, `DeleteItem`): the +//! partition key value from the request's `Key`/`Item`. +//! - `Query` on the base table or a local secondary index: the partition key +//! value from the key condition (an LSI shares the base partition key). +//! - `Query` on a global secondary index: the index's partition key value from +//! the key condition. The service binds LeadingKeys to the queried index's +//! partition key, not the base table's. +//! - `Scan` and `SearchVectors`: no LeadingKeys (neither targets a single +//! partition, so the service sets none). +//! - `BatchGetItem`/`BatchWriteItem`: one authorization per (action, table) +//! with the set of every key that table receives in the request. +//! - `TransactGetItems`/`TransactWriteItems`: one authorization per item, each +//! with that single item's partition key. Items that give the same (action, +//! table, key) are authorized once. +//! +//! An N partition key is written into LeadingKeys in one of two forms, +//! depending on the operation, because that is what the service does +//! (measured 2026-10-20). `GetItem`, `PutItem`, `UpdateItem`, `DeleteItem`, +//! `Query` and `BatchGetItem` use the normalized number (`010`, `10.0` and +//! `1E1` are all `10`). `BatchWriteItem`, `TransactGetItems` and +//! `TransactWriteItems` use the number as Java's `BigDecimal` prints it, which +//! keeps the scale the request was written with (`10.0` stays `10.0`, `1E1` +//! is `1E+1`, `010` is `10`). See [`NumberForm`]. A B partition key is its +//! base64 text and an S key is the string as sent. + +use std::collections::HashMap; + +use extenddb_core::expression::{Expr, PathElement, parse_key_condition, tokenize_with_limit}; +use extenddb_core::types::{KeyType, TableKeyInfo}; +use serde_json::Value; + +/// A single `(action, table)` authorization to perform for a batch or +/// transaction operation, carrying the LeadingKeys the service would evaluate. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct AuthzTarget { + /// The bare operation string the action maps to (no `dynamodb:` prefix). + pub action: String, + /// The table the action is authorized against. + pub table: String, + /// The partition key value(s) for this authorization, or `None` when the + /// key cannot be determined (the engine then rejects the request). + pub leading_keys: Option>, +} + +/// Find the HASH (partition) key attribute name of a key schema. +fn hash_attr(schema: &[extenddb_core::types::KeySchemaElement]) -> Option<&str> { + schema + .iter() + .find(|e| e.key_type == KeyType::Hash) + .map(|e| e.attribute_name.as_str()) +} + +/// How an N partition key is written into `dynamodb:LeadingKeys`. The service +/// writes it differently depending on the operation (measured against live +/// `DynamoDB` on 2026-10-20 over 21 spellings, including `1E-7`, `1.23E+4`, +/// `0E+5`, `1.00E-130` and negative values). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum NumberForm { + /// The normalized number: `010`, `10.0`, `1E1` and `+10` are all `10`. + /// Used by `GetItem`, `PutItem`, `UpdateItem`, `DeleteItem`, `Query` and + /// `BatchGetItem`. + Normalized, + /// The number as Java's `new BigDecimal(s).toString()` prints it, which + /// keeps the scale the request was written with: `10.0` stays `10.0`, `1E1` + /// is `1E+1`, `100E-2` is `1.00`, and `010`, `+10` and `1.0E1` are `10`. + /// Used by `BatchWriteItem` put requests, `TransactGetItems` and every + /// `TransactWriteItems` action. + Decimal, +} + +/// Turn a `DynamoDB` typed scalar (`{"S": "v"}`, `{"N": "10"}`, `{"B": "AQI="}`) +/// into the string the service compares for LeadingKeys. +/// +/// S and B are compared as sent. An N value is written in `form`, the form the +/// service uses for the operation path. The policy value is left exactly as +/// written, which is what the service does, so under a fence of `["10"]` the +/// spelling `10.0` matches on `GetItem` and does not match on +/// `TransactGetItems`. An N value the engine would reject as malformed yields +/// `None`, so the engine returns `ValidationException` before any data flows. +fn extract_typed_value(type_val: &Value, form: NumberForm) -> Option { + let (type_tag, inner) = type_val.as_object()?.iter().next()?; + let raw = inner.as_str().unwrap_or_default(); + if type_tag == "N" { + let normalized = + extenddb_core::validation::number::validate_and_normalize_number(raw).ok()?; + return match form { + NumberForm::Normalized => Some(normalized), + NumberForm::Decimal => java_decimal_string(raw), + }; + } + Some(raw.to_owned()) +} + +/// Java's `new BigDecimal(raw).toString()` for a number string that has +/// already passed `validate_and_normalize_number`. +/// +/// The unscaled value is every digit written, without its leading zeros, and +/// the scale is the number of fraction digits minus the exponent. `toString` +/// prints plain notation when the scale is not negative and the adjusted +/// exponent (the exponent of the first digit) is at least -6, and scientific +/// notation with an explicit exponent sign otherwise. A zero carries no sign. +fn java_decimal_string(raw: &str) -> Option { + let (negative, rest) = match raw.strip_prefix('-') { + Some(r) => (true, r), + None => (false, raw.strip_prefix('+').unwrap_or(raw)), + }; + let (mantissa, exponent) = match rest.split_once(['e', 'E']) { + Some((m, e)) => (m, e.parse::().ok()?), + None => (rest, 0), + }; + let (int_part, frac_part) = mantissa.split_once('.').unwrap_or((mantissa, "")); + let all_digits = format!("{int_part}{frac_part}"); + let coefficient = match all_digits.trim_start_matches('0') { + "" => "0", + c => c, + }; + let scale = i64::try_from(frac_part.len()).ok()? - exponent; + let adjusted = i64::try_from(coefficient.len()).ok()? - 1 - scale; + + let mut out = String::new(); + if negative && coefficient != "0" { + out.push('-'); + } + if scale >= 0 && adjusted >= -6 { + let scale = usize::try_from(scale).ok()?; + if scale == 0 { + out.push_str(coefficient); + } else if coefficient.len() > scale { + let (whole, fraction) = coefficient.split_at(coefficient.len() - scale); + out.push_str(whole); + out.push('.'); + out.push_str(fraction); + } else { + out.push_str("0."); + out.extend(std::iter::repeat_n('0', scale - coefficient.len())); + out.push_str(coefficient); + } + } else { + let (first, tail) = coefficient.split_at(1); + out.push_str(first); + if !tail.is_empty() { + out.push('.'); + out.push_str(tail); + } + out.push('E'); + if adjusted >= 0 { + out.push('+'); + } + out.push_str(&adjusted.to_string()); + } + Some(out) +} + +/// Extract the partition key value from a `Key`/`Item` map given the partition +/// attribute name. +fn pk_from_map(map: &Value, pk_attr: &str, form: NumberForm) -> Option { + extract_typed_value(map.as_object()?.get(pk_attr)?, form) +} + +/// Resolve the top-level attribute name of a parsed key path, applying +/// `ExpressionAttributeNames` for `#name` references. Returns `None` when the +/// reference is undefined (the engine rejects such a request). +fn resolve_attr_name(path: &[PathElement], names: &HashMap) -> Option { + match path.first()? { + PathElement::Attribute(name) => match name.strip_prefix('#') { + Some(reference) => names.get(reference).cloned(), + None => Some(name.clone()), + }, + PathElement::Index(_) => None, + } +} + +/// The LeadingKeys for a single-table operation, or `None` when the key is not +/// determinable (the engine then produces the request's error). +pub(crate) fn single_table_leading_keys( + input: &Value, + operation: &str, + key_info: Option<&TableKeyInfo>, + max_tokens: usize, +) -> Option> { + let key_info = key_info?; + match operation { + "GetItem" | "DeleteItem" | "UpdateItem" => { + let pk_attr = hash_attr(&key_info.key_schema)?; + Some(vec![pk_from_map( + input.get("Key")?, + pk_attr, + NumberForm::Normalized, + )?]) + } + "PutItem" => { + let pk_attr = hash_attr(&key_info.key_schema)?; + Some(vec![pk_from_map( + input.get("Item")?, + pk_attr, + NumberForm::Normalized, + )?]) + } + "Query" => query_leading_keys(input, key_info, max_tokens).map(|v| vec![v]), + // Scan and every other operation set no LeadingKeys. + _ => None, + } +} + +/// The partition key value a `Query` touches, resolved against the queried key +/// schema (base table, or the named index). `None` when it cannot be +/// determined, which is every shape the engine rejects before returning data. +fn query_leading_keys(input: &Value, key_info: &TableKeyInfo, max_tokens: usize) -> Option { + // The partition attribute of whatever the query targets: the named index's + // partition key, or the base table's. A GSI has its own partition key; an + // LSI shares the base table's. An unknown index name is a request the engine + // rejects, so leave the key absent. + let pk_attr: String = match index_name(input) { + Some(index) => { + let info = key_info + .global_secondary_indexes + .iter() + .chain(key_info.local_secondary_indexes.iter()) + .find(|i| i.index_name == index)?; + hash_attr(&info.key_schema)?.to_owned() + } + None => hash_attr(&key_info.key_schema)?.to_owned(), + }; + + let names = names_map(input); + let values = values_map(input); + + if let Some(kce) = input.get("KeyConditionExpression").and_then(Value::as_str) { + let tokens = tokenize_with_limit(kce, max_tokens).ok()?; + let mut kc = parse_key_condition(&tokens).ok()?; + // Assign PK vs SK using the partition attribute, mirroring the engine. + kc.resolve_pk_sk(&pk_attr, &names).ok()?; + // The condition must actually pin the partition key; if it does not, the + // engine rejects the query, so the key is not determinable here. + if resolve_attr_name(&kc.pk_path, &names).as_deref() != Some(pk_attr.as_str()) { + return None; + } + return value_of(&kc.pk_value, &values); + } + + // Legacy KeyConditions: an EQ on the partition attribute. + if let Some(conds) = input.get("KeyConditions").and_then(Value::as_object) { + let cond = conds.get(&pk_attr)?.as_object()?; + if cond.get("ComparisonOperator").and_then(Value::as_str) != Some("EQ") { + return None; + } + let list = cond.get("AttributeValueList")?.as_array()?; + if list.len() != 1 { + return None; + } + return extract_typed_value(&list[0], NumberForm::Normalized); + } + + None +} + +/// Resolve a key-condition value expression to its scalar string. Only a +/// placeholder resolved through `ExpressionAttributeValues` is a valid +/// partition key value in a key condition. +fn value_of(expr: &Expr, values: &HashMap) -> Option { + match expr { + Expr::Placeholder(name) => extract_typed_value(values.get(name)?, NumberForm::Normalized), + _ => None, + } +} + +fn index_name(input: &Value) -> Option { + input + .get("IndexName") + .and_then(Value::as_str) + .map(ToOwned::to_owned) +} + +/// `ExpressionAttributeNames` with the leading `#` stripped from each key, +/// matching how the engine builds its name map. +fn names_map(input: &Value) -> HashMap { + let mut out = HashMap::new(); + if let Some(m) = input + .get("ExpressionAttributeNames") + .and_then(Value::as_object) + { + for (k, v) in m { + if let Some(s) = v.as_str() { + out.insert(k.strip_prefix('#').unwrap_or(k).to_owned(), s.to_owned()); + } + } + } + out +} + +/// `ExpressionAttributeValues` with the leading `:` stripped from each key, +/// matching how a placeholder name is stored by the tokenizer. +fn values_map(input: &Value) -> HashMap { + let mut out = HashMap::new(); + if let Some(m) = input + .get("ExpressionAttributeValues") + .and_then(Value::as_object) + { + for (k, v) in m { + out.insert(k.strip_prefix(':').unwrap_or(k).to_owned(), v.clone()); + } + } + out +} + +/// The per-(action, table) or per-item authorizations for a batch or +/// transaction operation, each carrying the LeadingKeys the service evaluates. +/// +/// Returns `None` for non-batch/transact operations. Batch operations +/// authorize once per table with the set of all keys for that table; +/// transaction operations authorize once per item with that item's key, which +/// is how the service evaluates a `ForAnyValue` condition differently for the +/// two (a batch of one own and one other key passes `ForAnyValue`; a +/// transaction of the same two does not). Transaction items that give the same +/// target are returned once, in the order of their first occurrence: the +/// decision for a target depends only on its action, table and LeadingKeys, so +/// a hundred `ConditionCheck`s on one partition need one evaluation. +/// +/// `key_info_by_table` is the authorization snapshot; a `None` entry is a table +/// that does not exist, whose targets carry no LeadingKeys (the engine returns +/// `ResourceNotFoundException`). +pub(crate) fn batch_transact_targets( + operation: &str, + input: &Value, + key_info_by_table: &HashMap>, +) -> Option> { + let pk_attr_of = |table: &str| -> Option { + key_info_by_table + .get(table) + .and_then(Option::as_ref) + .and_then(|ki| hash_attr(&ki.key_schema).map(ToOwned::to_owned)) + }; + let mut seen = std::collections::HashSet::new(); + let mut push_unique = |out: &mut Vec, target: AuthzTarget| { + if seen.insert(target.clone()) { + out.push(target); + } + }; + + match operation { + "BatchGetItem" | "BatchWriteItem" => { + let items = input.get("RequestItems")?.as_object()?; + let mut out = Vec::new(); + for (table, body) in items { + let keys = batch_table_keys(operation, body, pk_attr_of(table).as_deref()); + out.push(AuthzTarget { + action: operation.to_owned(), + table: table.clone(), + leading_keys: keys, + }); + } + Some(out) + } + "TransactGetItems" => { + let items = input.get("TransactItems")?.as_array()?; + let mut out = Vec::new(); + for entry in items { + if let Some(get) = entry.get("Get") + && let Some(table) = get.get("TableName").and_then(Value::as_str) + { + let lk = pk_attr_of(table) + .and_then(|pk| { + get.get("Key") + .and_then(|k| pk_from_map(k, &pk, NumberForm::Decimal)) + }) + .map(|v| vec![v]); + push_unique( + &mut out, + AuthzTarget { + action: "GetItem".to_owned(), + table: table.to_owned(), + leading_keys: lk, + }, + ); + } + } + Some(out) + } + "TransactWriteItems" => { + let items = input.get("TransactItems")?.as_array()?; + let mut out = Vec::new(); + for entry in items { + for (sub, action, key_field) in [ + ("Put", "PutItem", "Item"), + ("Delete", "DeleteItem", "Key"), + ("Update", "UpdateItem", "Key"), + ("ConditionCheck", "ConditionCheckItem", "Key"), + ] { + if let Some(op) = entry.get(sub) + && let Some(table) = op.get("TableName").and_then(Value::as_str) + { + let lk = pk_attr_of(table) + .and_then(|pk| { + op.get(key_field) + .and_then(|m| pk_from_map(m, &pk, NumberForm::Decimal)) + }) + .map(|v| vec![v]); + push_unique( + &mut out, + AuthzTarget { + action: action.to_owned(), + table: table.to_owned(), + leading_keys: lk, + }, + ); + } + } + } + Some(out) + } + _ => None, + } +} + +/// The distinct table names a batch or transaction operation addresses, or +/// `None` for any other operation (and for a batch or transaction request whose +/// items container is absent). The `Some`/`None` result mirrors +/// `batch_transact_targets`, so the caller gates the batch and transaction +/// branch on this and then builds the per-target authorizations once. Walking +/// the body for table names here avoids building a throwaway set of full +/// targets only to discover which tables to fetch key info for. +pub(crate) fn batch_transact_tables( + operation: &str, + input: &Value, +) -> Option> { + let mut tables = std::collections::BTreeSet::new(); + match operation { + "BatchGetItem" | "BatchWriteItem" => { + for table in input.get("RequestItems")?.as_object()?.keys() { + tables.insert(table.clone()); + } + } + "TransactGetItems" => { + for entry in input.get("TransactItems")?.as_array()? { + if let Some(table) = entry + .get("Get") + .and_then(|g| g.get("TableName")) + .and_then(Value::as_str) + { + tables.insert(table.to_owned()); + } + } + } + "TransactWriteItems" => { + for entry in input.get("TransactItems")?.as_array()? { + for sub in ["Put", "Delete", "Update", "ConditionCheck"] { + if let Some(table) = entry + .get(sub) + .and_then(|o| o.get("TableName")) + .and_then(Value::as_str) + { + tables.insert(table.to_owned()); + } + } + } + } + _ => return None, + } + Some(tables) +} + +/// Collect the partition key values a single table receives in a batch request. +/// Returns `None` when the table's key schema is unknown (the engine then +/// rejects), or when no key yields a value. +fn batch_table_keys(operation: &str, body: &Value, pk_attr: Option<&str>) -> Option> { + let pk_attr = pk_attr?; + let mut out = Vec::new(); + if operation == "BatchGetItem" { + for key in body.get("Keys")?.as_array()? { + if let Some(v) = pk_from_map(key, pk_attr, NumberForm::Normalized) { + out.push(v); + } + } + } else { + // BatchWriteItem: a list of {PutRequest:{Item}} / {DeleteRequest:{Key}}. + for req in body.as_array()? { + let map = req + .get("PutRequest") + .and_then(|p| p.get("Item")) + .or_else(|| req.get("DeleteRequest").and_then(|d| d.get("Key"))); + if let Some(m) = map + && let Some(v) = pk_from_map(m, pk_attr, NumberForm::Decimal) + { + out.push(v); + } + } + } + if out.is_empty() { None } else { Some(out) } +} + +#[cfg(test)] +mod tests; diff --git a/crates/server/src/leading_keys/tests.rs b/crates/server/src/leading_keys/tests.rs new file mode 100644 index 000000000..7cc17626c --- /dev/null +++ b/crates/server/src/leading_keys/tests.rs @@ -0,0 +1,652 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Unit tests for `dynamodb:LeadingKeys` extraction. +//! +//! Expectations mirror behaviour measured against live `DynamoDB` on +//! 2026-10-20. + +use super::*; +use extenddb_core::types::{ + IndexInfo, IndexType, KeySchemaElement, KeyType, Projection, ProjectionType, TableKeyInfo, +}; +use serde_json::json; + +const MAX_TOKENS: usize = 4096; + +fn ks(name: &str, kt: KeyType) -> KeySchemaElement { + KeySchemaElement { + attribute_name: name.to_owned(), + key_type: kt, + } +} + +fn proj() -> Projection { + Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + } +} + +/// A table with base key (pk HASH, sk RANGE), an LSI sharing pk, and a GSI on +/// its own partition key gpk. +fn main_table() -> TableKeyInfo { + TableKeyInfo { + table_name: "t".to_owned(), + key_schema: vec![ks("pk", KeyType::Hash), ks("sk", KeyType::Range)], + base_key_schema: vec![ks("pk", KeyType::Hash), ks("sk", KeyType::Range)], + local_secondary_indexes: vec![IndexInfo { + index_name: "lsi1".to_owned(), + index_id: "lsi1".to_owned(), + index_type: IndexType::Lsi, + key_schema: vec![ks("pk", KeyType::Hash), ks("lsi_sk", KeyType::Range)], + projection: proj(), + }], + global_secondary_indexes: vec![IndexInfo { + index_name: "gsi1".to_owned(), + index_id: "gsi1".to_owned(), + index_type: IndexType::Gsi, + key_schema: vec![ks("gpk", KeyType::Hash), ks("gsk", KeyType::Range)], + projection: proj(), + }], + ..Default::default() + } +} + +fn num_table() -> TableKeyInfo { + TableKeyInfo { + table_name: "n".to_owned(), + key_schema: vec![ks("pk", KeyType::Hash), ks("sk", KeyType::Range)], + base_key_schema: vec![ks("pk", KeyType::Hash), ks("sk", KeyType::Range)], + ..Default::default() + } +} + +// ---- item operations ---------------------------------------------------- + +#[test] +fn get_item_extracts_partition_key() { + let input = json!({"TableName": "t", "Key": {"pk": {"S": "alice"}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "GetItem", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn put_item_extracts_partition_key_from_item() { + let input = json!({"TableName": "t", "Item": {"pk": {"S": "alice"}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "PutItem", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn number_partition_key_is_the_canonical_string() { + // Live DynamoDB compares an N key of 10 as the string "10". + let input = json!({"TableName": "n", "Key": {"pk": {"N": "10"}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "GetItem", Some(&num_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["10".to_owned()])); +} + +#[test] +fn binary_partition_key_is_the_base64_text() { + let input = json!({"TableName": "b", "Key": {"pk": {"B": "AQI="}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "GetItem", Some(&num_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["AQI=".to_owned()])); +} + +#[test] +fn n_leading_key_is_normalized_on_item_operations() { + // GetItem compares the normalized number: under a policy fencing to "10", + // a request pk of 010, 10.0, 1E1 or +10 is allowed (measured 2026-10-20). + for spelling in ["010", "10.0", "1E1", "+10"] { + let input = json!({"TableName": "n", "Key": {"pk": {"N": spelling}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "GetItem", Some(&num_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["10".to_owned()]), "spelling {spelling}"); + } +} + +#[test] +fn n_leading_key_zero_spellings_normalize_to_zero() { + // Under a fence to "0" the service allows -0, 00 and 0.0 on GetItem; each + // normalizes to "0" (measured 2026-10-20). + for spelling in ["-0", "00", "0.0"] { + let input = json!({"TableName": "n", "Key": {"pk": {"N": spelling}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "GetItem", Some(&num_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["0".to_owned()]), "spelling {spelling}"); + } +} + +#[test] +fn query_n_leading_key_is_normalized() { + let input = json!({ + "TableName": "n", + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"N": "1E1"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&num_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["10".to_owned()])); +} + +#[test] +fn batch_get_n_leading_keys_are_normalized() { + let ki = HashMap::from([("n".to_owned(), Some(num_table()))]); + let bg = json!({"RequestItems": {"n": {"Keys": [ + {"pk": {"N": "010"}, "sk": {"S": "x"}}, + {"pk": {"N": "1E1"}, "sk": {"S": "y"}}, + ]}}}); + let t = batch_transact_targets("BatchGetItem", &bg, &ki).unwrap(); + assert_eq!( + t[0].leading_keys, + Some(vec!["10".to_owned(), "10".to_owned()]) + ); +} + +#[test] +fn batch_write_n_leading_keys_keep_the_written_scale() { + // BatchWriteItem put and delete requests both compare the number as Java's + // BigDecimal prints it (measured 2026-10-20, 25 of 25 attempts each). + let ki = HashMap::from([("n".to_owned(), Some(num_table()))]); + let bw = json!({"RequestItems": {"n": [ + {"PutRequest": {"Item": {"pk": {"N": "10.0"}, "sk": {"S": "x"}}}}, + {"DeleteRequest": {"Key": {"pk": {"N": "1E1"}, "sk": {"S": "y"}}}}, + {"PutRequest": {"Item": {"pk": {"N": "010"}, "sk": {"S": "z"}}}}, + ]}}); + let t = batch_transact_targets("BatchWriteItem", &bw, &ki).unwrap(); + assert_eq!( + t[0].leading_keys, + Some(vec!["10.0".to_owned(), "1E+1".to_owned(), "10".to_owned()]) + ); +} + +#[test] +fn transact_n_leading_keys_keep_the_written_scale() { + // TransactGetItems and every TransactWriteItems action compare the number + // as Java's BigDecimal prints it: under a fence of "10", 10.0 and 1E1 are + // denied and 010, +10 and 1.0E1 are allowed (measured 2026-10-20). + let ki = HashMap::from([("n".to_owned(), Some(num_table()))]); + for (spelling, expected) in [ + ("10.0", "10.0"), + ("1E1", "1E+1"), + ("010", "10"), + ("+10", "10"), + ("1.0E1", "10"), + ] { + let tg = json!({"TransactItems": [ + {"Get": {"TableName": "n", "Key": {"pk": {"N": spelling}, "sk": {"S": "x"}}}}, + ]}); + let t = batch_transact_targets("TransactGetItems", &tg, &ki).unwrap(); + assert_eq!( + t[0].leading_keys, + Some(vec![expected.to_owned()]), + "{spelling}" + ); + + let tw = json!({"TransactItems": [ + {"Put": {"TableName": "n", "Item": {"pk": {"N": spelling}, "sk": {"S": "x"}}}}, + {"Update": {"TableName": "n", "Key": {"pk": {"N": spelling}, "sk": {"S": "y"}}}}, + {"Delete": {"TableName": "n", "Key": {"pk": {"N": spelling}, "sk": {"S": "z"}}}}, + {"ConditionCheck": {"TableName": "n", "Key": {"pk": {"N": spelling}, "sk": {"S": "w"}}}}, + ]}); + let t = batch_transact_targets("TransactWriteItems", &tw, &ki).unwrap(); + for target in &t { + assert_eq!( + target.leading_keys, + Some(vec![expected.to_owned()]), + "{spelling} {}", + target.action + ); + } + } +} + +#[test] +fn java_decimal_string_matches_the_measured_service_values() { + // Each pair is a request spelling and the LeadingKeys value the service + // matched on TransactGetItems and BatchWriteItem puts (measured 2026-10-20). + for (spelling, expected) in [ + ("100E-2", "1.00"), + ("0.0000001", "1E-7"), + ("0.000001", "0.000001"), + ("123E2", "1.23E+4"), + ("-5.50", "-5.50"), + ("0.00", "0.00"), + ("0E5", "0E+5"), + (".5", "0.5"), + ("5.", "5"), + ("+1.5E+3", "1.5E+3"), + ("3e1", "3E+1"), + ("12.3400E-1", "1.23400"), + ("-0", "0"), + ("1.00E-130", "1.00E-130"), + ("7E0", "7"), + ("80E-1", "8.0"), + ("10.0", "10.0"), + ("1E1", "1E+1"), + ("1.0E1", "10"), + ("010", "10"), + ("+10", "10"), + ] { + assert_eq!( + java_decimal_string(spelling).as_deref(), + Some(expected), + "spelling {spelling}" + ); + } +} + +#[test] +fn malformed_n_leading_key_is_absent() { + // An N the engine would reject as malformed is not a determinable partition + // value, so the key is left absent and the engine returns ValidationException. + let input = json!({"TableName": "n", "Key": {"pk": {"N": "1.2.3"}, "sk": {"S": "x"}}}); + let r = single_table_leading_keys(&input, "GetItem", Some(&num_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +// ---- query: the defect --------------------------------------------------- + +#[test] +fn query_kce_extracts_partition_key() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn query_kce_resolves_expression_attribute_name() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "#p = :v", + "ExpressionAttributeNames": {"#p": "pk"}, + "ExpressionAttributeValues": {":v": {"S": "bob"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["bob".to_owned()])); +} + +#[test] +fn query_kce_with_sort_key_condition() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "pk = :v AND sk = :s", + "ExpressionAttributeValues": {":v": {"S": "alice"}, ":s": {"S": "x"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn query_kce_sort_key_before_partition_key() { + // resolve_pk_sk must still pick the partition key when the clauses are swapped. + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "sk = :s AND pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}, ":s": {"S": "x"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn query_legacy_key_conditions() { + let input = json!({ + "TableName": "t", + "KeyConditions": {"pk": {"ComparisonOperator": "EQ", + "AttributeValueList": [{"S": "alice"}]}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn query_on_lsi_uses_base_partition_key() { + let input = json!({ + "TableName": "t", + "IndexName": "lsi1", + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn query_on_gsi_uses_index_partition_key() { + // Measured: the service binds LeadingKeys to the queried GSI's partition key. + let input = json!({ + "TableName": "t", + "IndexName": "gsi1", + "KeyConditionExpression": "gpk = :v", + "ExpressionAttributeValues": {":v": {"S": "galice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["galice".to_owned()])); +} + +/// A table carrying a GSI whose partition key has two HASH attributes (g1, g2). +/// ExtendDB allows this shape and so does live `DynamoDB` (a two-HASH GSI is +/// accepted; a two-HASH base table is rejected). Measured 2026-10-20. +fn multi_hash_gsi_table() -> TableKeyInfo { + TableKeyInfo { + table_name: "m".to_owned(), + key_schema: vec![ks("pk", KeyType::Hash)], + base_key_schema: vec![ks("pk", KeyType::Hash)], + global_secondary_indexes: vec![IndexInfo { + index_name: "gsi2".to_owned(), + index_id: "gsi2".to_owned(), + index_type: IndexType::Gsi, + key_schema: vec![ks("g1", KeyType::Hash), ks("g2", KeyType::Hash)], + projection: proj(), + }], + ..Default::default() + } +} + +#[test] +fn query_on_multi_hash_gsi_uses_first_hash_attribute_only() { + // Measured 2026-10-20: for a Query on a GSI whose partition key has two + // HASH attributes, live DynamoDB evaluates dynamodb:LeadingKeys against the + // FIRST HASH attribute's value only. Under ForAllValues:StringEquals + // ["alice"] it allowed a query of g1=alice with any g2 value and denied + // g1=bob, so the second HASH attribute's value is not part of LeadingKeys. + let input = json!({ + "TableName": "m", + "IndexName": "gsi2", + "KeyConditionExpression": "g1 = :a AND g2 = :b", + "ExpressionAttributeValues": {":a": {"S": "alice"}, ":b": {"S": "red"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&multi_hash_gsi_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +#[test] +fn query_on_multi_hash_gsi_is_clause_order_independent() { + // The same query with the two HASH clauses written in the other order still + // yields the first HASH attribute's value. + let input = json!({ + "TableName": "m", + "IndexName": "gsi2", + "KeyConditionExpression": "g2 = :b AND g1 = :a", + "ExpressionAttributeValues": {":a": {"S": "alice"}, ":b": {"S": "blue"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&multi_hash_gsi_table()), MAX_TOKENS); + assert_eq!(r, Some(vec!["alice".to_owned()])); +} + +// ---- query: shapes the engine rejects (key left absent) ----------------- + +#[test] +fn query_missing_key_condition_is_absent() { + let input = json!({"TableName": "t"}); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +#[test] +fn query_non_equality_partition_key_is_absent() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "pk > :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +#[test] +fn query_malformed_key_condition_is_absent() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "pk :: :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +#[test] +fn query_unresolved_name_is_absent() { + // #p with no ExpressionAttributeNames entry: the engine rejects it. + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "#p = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +#[test] +fn query_condition_only_on_sort_key_is_absent() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "sk = :s", + "ExpressionAttributeValues": {":s": {"S": "x"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +#[test] +fn query_unknown_index_is_absent() { + let input = json!({ + "TableName": "t", + "IndexName": "nope", + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + let r = single_table_leading_keys(&input, "Query", Some(&main_table()), MAX_TOKENS); + assert_eq!(r, None); +} + +#[test] +fn query_without_key_info_is_absent() { + let input = json!({ + "TableName": "t", + "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}, + }); + assert_eq!( + single_table_leading_keys(&input, "Query", None, MAX_TOKENS), + None + ); +} + +// ---- scan sets no leading keys ------------------------------------------ + +#[test] +fn scan_sets_no_leading_keys() { + let input = json!({"TableName": "t"}); + assert_eq!( + single_table_leading_keys(&input, "Scan", Some(&main_table()), MAX_TOKENS), + None + ); +} + +// ---- batch: per-table key sets ------------------------------------------ + +fn key_info_map() -> HashMap> { + HashMap::from([("t".to_owned(), Some(main_table()))]) +} + +#[test] +fn batch_get_collects_all_keys_for_a_table() { + let input = json!({"RequestItems": {"t": {"Keys": [ + {"pk": {"S": "alice"}, "sk": {"S": "x"}}, + {"pk": {"S": "bob"}, "sk": {"S": "y"}}, + ]}}}); + let t = batch_transact_targets("BatchGetItem", &input, &key_info_map()).unwrap(); + assert_eq!(t.len(), 1); + assert_eq!(t[0].action, "BatchGetItem"); + assert_eq!(t[0].table, "t"); + assert_eq!( + t[0].leading_keys, + Some(vec!["alice".to_owned(), "bob".to_owned()]) + ); +} + +#[test] +fn batch_write_collects_put_and_delete_keys() { + let input = json!({"RequestItems": {"t": [ + {"PutRequest": {"Item": {"pk": {"S": "alice"}, "sk": {"S": "x"}}}}, + {"DeleteRequest": {"Key": {"pk": {"S": "bob"}, "sk": {"S": "y"}}}}, + ]}}); + let t = batch_transact_targets("BatchWriteItem", &input, &key_info_map()).unwrap(); + assert_eq!(t.len(), 1); + assert_eq!( + t[0].leading_keys, + Some(vec!["alice".to_owned(), "bob".to_owned()]) + ); +} + +// ---- transaction: one authorization per item ---------------------------- + +#[test] +fn transact_get_one_target_per_item() { + let input = json!({"TransactItems": [ + {"Get": {"TableName": "t", "Key": {"pk": {"S": "alice"}, "sk": {"S": "x"}}}}, + {"Get": {"TableName": "t", "Key": {"pk": {"S": "bob"}, "sk": {"S": "y"}}}}, + ]}); + let t = batch_transact_targets("TransactGetItems", &input, &key_info_map()).unwrap(); + assert_eq!(t.len(), 2); + assert_eq!(t[0].action, "GetItem"); + assert_eq!(t[0].leading_keys, Some(vec!["alice".to_owned()])); + assert_eq!(t[1].leading_keys, Some(vec!["bob".to_owned()])); +} + +#[test] +fn transact_write_maps_each_sub_op_to_its_action_and_key() { + let input = json!({"TransactItems": [ + {"Put": {"TableName": "t", "Item": {"pk": {"S": "alice"}, "sk": {"S": "x"}}}}, + {"Delete": {"TableName": "t", "Key": {"pk": {"S": "bob"}, "sk": {"S": "y"}}}}, + {"Update": {"TableName": "t", "Key": {"pk": {"S": "carol"}, "sk": {"S": "z"}}}}, + {"ConditionCheck": {"TableName": "t", "Key": {"pk": {"S": "dave"}, "sk": {"S": "w"}}}}, + ]}); + let t = batch_transact_targets("TransactWriteItems", &input, &key_info_map()).unwrap(); + let got: Vec<(&str, Option>)> = t + .iter() + .map(|a| (a.action.as_str(), a.leading_keys.clone())) + .collect(); + assert_eq!( + got, + vec![ + ("PutItem", Some(vec!["alice".to_owned()])), + ("DeleteItem", Some(vec!["bob".to_owned()])), + ("UpdateItem", Some(vec!["carol".to_owned()])), + ("ConditionCheckItem", Some(vec!["dave".to_owned()])), + ] + ); +} + +#[test] +fn non_batch_operation_has_no_targets() { + let input = json!({"TableName": "t"}); + assert_eq!( + batch_transact_targets("GetItem", &input, &key_info_map()), + None + ); +} + +// ---- transaction targets are de-duplicated ------------------------------- + +#[test] +fn transact_targets_with_the_same_action_table_and_key_are_evaluated_once() { + // A hundred ConditionChecks on one partition need one evaluation; distinct + // keys and distinct actions each keep their own target, in the order of + // their first occurrence. + let mut items: Vec = (0..100) + .map(|i| { + json!({"ConditionCheck": {"TableName": "t", + "Key": {"pk": {"S": "alice"}, "sk": {"S": format!("s{i}")}}, + "ConditionExpression": "attribute_exists(pk)"}}) + }) + .collect(); + items.push(json!({"Put": {"TableName": "t", + "Item": {"pk": {"S": "alice"}, "sk": {"S": "p"}}}})); + items.push(json!({"ConditionCheck": {"TableName": "t", + "Key": {"pk": {"S": "bob"}, "sk": {"S": "x"}}, + "ConditionExpression": "attribute_exists(pk)"}})); + let input = json!({ "TransactItems": items }); + let t = batch_transact_targets("TransactWriteItems", &input, &key_info_map()).unwrap(); + let got: Vec<(&str, Option>)> = t + .iter() + .map(|a| (a.action.as_str(), a.leading_keys.clone())) + .collect(); + assert_eq!( + got, + vec![ + ("ConditionCheckItem", Some(vec!["alice".to_owned()])), + ("PutItem", Some(vec!["alice".to_owned()])), + ("ConditionCheckItem", Some(vec!["bob".to_owned()])), + ] + ); + + let gets: Vec = (0..3) + .map(|i| { + json!({"Get": {"TableName": "t", + "Key": {"pk": {"S": "alice"}, "sk": {"S": format!("s{i}")}}}}) + }) + .collect(); + let t = batch_transact_targets( + "TransactGetItems", + &json!({ "TransactItems": gets }), + &key_info_map(), + ) + .unwrap(); + assert_eq!(t.len(), 1); +} + +#[test] +fn a_table_missing_from_the_snapshot_leaves_the_key_absent() { + // A `None` entry is a table authorization found missing: its targets carry + // no LeadingKeys, and the engine, bound to the same snapshot, returns + // ResourceNotFoundException for it. + let ki = HashMap::from([("t".to_owned(), None)]); + let input = json!({"TransactItems": [ + {"Get": {"TableName": "t", "Key": {"pk": {"S": "bob"}, "sk": {"S": "x"}}}}, + ]}); + let t = batch_transact_targets("TransactGetItems", &input, &ki).unwrap(); + assert_eq!(t.len(), 1); + assert_eq!(t[0].leading_keys, None); +} + +// ---- the key condition token limit --------------------------------------- + +use extenddb_core::expression::{ExpressionKind, tokenize_for}; + +/// The token-count limit is the one place authorization (`tokenize_with_limit`) +/// and the engine's Query path (`tokenize_for` with +/// `ExpressionKind::KeyCondition`, called from `parse_key_condition_expr`) +/// could disagree. Both enforce the same limit through the same `push_token`, +/// so a key condition at the limit is accepted by both and one over it is +/// rejected by both. The engine side here is the engine's tokenizer call. +#[test] +fn token_limit_boundary_matches_between_engine_and_auth() { + let kce = "pk = :v"; + let input = json!({"TableName": "t", + "KeyConditionExpression": kce, + "ExpressionAttributeValues": {":v": {"S": "alice"}}}); + let main = main_table(); + + // "pk = :v" is three tokens. At the limit both accept and agree. + assert!(tokenize_for(kce, 3, ExpressionKind::KeyCondition).is_ok()); + assert_eq!( + single_table_leading_keys(&input, "Query", Some(&main), 3), + Some(vec!["alice".to_owned()]) + ); + + // One token under the limit: both reject (the engine errors, authorization + // leaves the key absent). + assert!( + tokenize_for(kce, 2, ExpressionKind::KeyCondition).is_err(), + "the engine must reject over the token limit" + ); + assert_eq!( + single_table_leading_keys(&input, "Query", Some(&main), 2), + None, + "authorization must leave the key absent over the token limit" + ); +} diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 7a2471de9..a1ff087cc 100755 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -15,6 +15,7 @@ pub mod dev_credentials; mod handler; mod import_export_paths; pub mod key_info_cache; +mod leading_keys; pub mod management; mod metrics_endpoint; pub mod rate_limit; diff --git a/crates/server/src/request_helpers.rs b/crates/server/src/request_helpers.rs index 583ac89f3..ed2599920 100755 --- a/crates/server/src/request_helpers.rs +++ b/crates/server/src/request_helpers.rs @@ -5,6 +5,7 @@ use axum::http::HeaderMap; use extenddb_core::error::DynamoDbError; +use extenddb_storage::error::StorageError; use serde_json::Value; use crate::AppState; @@ -37,8 +38,8 @@ pub(crate) fn extract_operation(headers: &HeaderMap) -> Result Option { input .get("TableName") @@ -46,11 +47,51 @@ pub(crate) fn extract_table_name(input: &Value) -> Option { .map(ToOwned::to_owned) } +/// Decide how a `TableKeyInfo` lookup result feeds request authorization. +/// +/// A present table yields its key info. A missing table yields `None`: the +/// `dynamodb:LeadingKeys` value stays absent and the engine returns +/// `ResourceNotFoundException`, which is what the service returns for a +/// LeadingKeys-restricted principal on a missing table (measured 2026-10-20). +/// +/// Any other lookup failure (connection, transient, backend internal) must not +/// reach the policy evaluator with the key absent. A +/// `ForAllValues:StringEquals dynamodb:LeadingKeys` condition is satisfied +/// vacuously when the key is absent, so a request whose key could not be +/// determined would be authorized and the engine could then read the real +/// table on its own lookup. Such a failure returns an error before dispatch, so +/// no request is ever authorized on a key the server could not determine. +/// +/// The error is classified by the engine's own `storage_err_to_dynamo`, so the +/// server and the engine return the same wire error for the same storage fault: +/// a transient or connection fault is `ServiceUnavailable` (503, SDK-retryable), +/// the catalog and backend-internal faults are `InternalServerError` (500), and +/// a not-active table is `ResourceInUseException`. Using one mapper keeps a +/// database hiccup from returning 503 when the engine hits it and 500 when +/// authorization hits it first. The cache reports a missing table as +/// `StorageError::TableNotFound` and every other fault as a distinct error, so +/// the two are told apart here rather than collapsed to `None`. Every fault +/// class fails closed: none authorizes the request. +fn key_info_for_authz( + result: Result, +) -> Result, DynamoDbError> { + match result { + Ok(info) => Ok(Some(info)), + Err(StorageError::TableNotFound(_)) => Ok(None), + Err(e) => Err(extenddb_engine::storage_err_to_dynamo(e)), + } +} + /// Evaluate IAM policies for an authenticated identity. /// -/// Returns the pre-fetched `TableKeyInfo` for single-table item-level operations -/// (P118 optimization #2). The caller passes this into `OperationContext` to -/// avoid a redundant catalog roundtrip in the engine layer. +/// Returns the `TableKeyInfo` authorization read for each table it evaluated, +/// including a `None` for a table it found missing (P118 optimization #2). The +/// caller passes this into `OperationContext::authorized_key_info`, and the +/// engine resolves those tables from it instead of reading the catalog again. +/// Authorization and the engine therefore act on one snapshot: a table that +/// authorization saw as missing, and so evaluated with `dynamodb:LeadingKeys` +/// absent, is missing to the engine for the rest of the request, even if it has +/// become ACTIVE in between. /// /// All authorization data is fetched via `state.authz_cache`, which sits on /// top of the underlying `AuthorizationStore` and serves cached, pre-parsed @@ -63,29 +104,26 @@ pub(crate) async fn authorize_request( account_id: &str, source_ip: Option<&str>, user_agent: Option<&str>, -) -> Result, DynamoDbError> { +) -> Result { let table_name = extract_table_name(input); + let mut authorized_key_info = extenddb_engine::AuthorizedKeyInfo::new(); // P118: Fetch table_key_info for item-level operations via the SWR cache. // The result is used for LeadingKeys extraction here AND returned to the // caller to avoid a redundant fetch in the engine layer. - let key_info = match operation { - "GetItem" | "PutItem" | "DeleteItem" | "UpdateItem" | "Query" | "Scan" => { - if let Some(ref tn) = table_name { - state - .table_key_info_cache - .get_optional(account_id, tn) - .await - } else { - None - } - } - _ => None, - }; + if let Some(ref tn) = table_name + && matches!( + operation, + "GetItem" | "PutItem" | "DeleteItem" | "UpdateItem" | "Query" | "Scan" + ) + { + let ki = key_info_for_authz(state.table_key_info_cache.get(account_id, tn).await)?; + authorized_key_info.insert(tn.clone(), ki); + } // Developer mode opens authorization for the authenticated caller. SigV4 // verification already ran upstream, so the request is authenticated; only - // the IAM policy decision is skipped. key_info is still returned so the + // the IAM policy decision is skipped. The key info is still returned so the // engine layer can reuse it. // // This check is deliberately positioned ahead of EVERY authorization branch @@ -95,7 +133,7 @@ pub(crate) async fn authorize_request( // item operations passed. Any future operation-specific branch must be added // below this point, or it will reintroduce that defect. if state.dev_mode { - return Ok(key_info); + return Ok(authorized_key_info); } // Batch and transaction operations address multiple tables in nested request @@ -108,13 +146,52 @@ pub(crate) async fn authorize_request( // Deny on one table be bypassed. Evaluate every (action, table) pair and // reject the entire request if any is denied (all-or-nothing), matching // DynamoDB. Verified against the AWS IAM Service Authorization Reference. - if let Some(targets) = batch_transact_authz_targets(operation, input) { - for (action_op, table) in targets { - let resource_arn = build_resource_arn(&state.region, account_id, Some(&table)); + // `batch_transact_tables` returns `Some` for exactly the batch and + // transaction operations. Batch operations authorize once per table with + // the set of keys that table receives; transaction operations authorize + // once per item with that item's key. Each target carries the + // `dynamodb:LeadingKeys` value(s) the service evaluates, extracted per the + // table's key schema. The resource ARN per target is unchanged: each is + // still that table's ARN. (Measured against live DynamoDB 2026-10-20.) + // Targets with the same (action, table, LeadingKeys) are evaluated once, + // since the decision depends only on those three for a given identity. + if let Some(tables) = crate::leading_keys::batch_transact_tables(operation, input) { + // A request outside the operation's count limits is refused before any + // key info is read or any policy is evaluated, with the engine's own + // error. DynamoDB does the same: an empty or 101-item transaction, an + // empty RequestItems, a table with no keys or requests, more than 100 + // tables or keys in a BatchGetItem and more than 25 tables or requests + // in a BatchWriteItem each return ValidationException, including to a + // principal with no permissions (measured 2026-10-21). Checking first + // also bounds the tables read and the policies evaluated below by those + // limits instead of by the body size. A TransactWriteItems item can name + // up to four sub-operations here, each on its own table, before the + // handler rejects it, so a transaction reads at most 400 tables. + extenddb_engine::check_request_item_counts(operation, input)?; + + // Read each distinct table's key info from storage, not the cache, and + // hand what was read to the engine. The engine resolves these tables + // from the snapshot, so it never acts on a table authorization saw as + // missing, and it sees the catalog as fresh as these handlers always + // read it. + for table in tables { + let ki = key_info_for_authz(state.storage.table_key_info(account_id, &table).await)?; + authorized_key_info.insert(table, ki); + } + let targets = + crate::leading_keys::batch_transact_targets(operation, input, &authorized_key_info) + .unwrap_or_default(); + + if targets.is_empty() { + // Malformed body (the count checks above have already refused an + // empty one): still require the operation's action on the table + // wildcard, as the generic path would, then let the engine reject + // the body. This never silently skips authorization. + let resource_arn = build_resource_arn(&state.region, account_id, None); authorization::check_authorization( state.authz_cache.as_ref(), identity, - &action_op, + operation, &resource_arn, false, extenddb_auth::policy::context::RequestParams { @@ -126,18 +203,45 @@ pub(crate) async fn authorize_request( }, ) .await?; + } else { + for target in targets { + let resource_arn = + build_resource_arn(&state.region, account_id, Some(&target.table)); + let params = extenddb_auth::policy::context::RequestParams { + leading_keys: target.leading_keys, + region: Some(state.region.to_string()), + secure_transport: state.tls_enabled, + source_ip: source_ip.map(ToOwned::to_owned), + user_agent: user_agent.map(ToOwned::to_owned), + ..Default::default() + }; + authorization::check_authorization( + state.authz_cache.as_ref(), + identity, + &target.action, + &resource_arn, + false, + params, + ) + .await?; + } } - return Ok(None); + return Ok(authorized_key_info); } let resource_arn = build_resource_arn(&state.region, account_id, table_name.as_deref()); - - let pk_attr = key_info + let key_info = table_name .as_ref() - .map(|ki| ki.key_schema[0].attribute_name.clone()); + .and_then(|tn| authorized_key_info.get(tn)) + .and_then(Option::as_ref); let params = extenddb_auth::policy::context::RequestParams { - leading_keys: extract_leading_keys(input, operation, pk_attr.as_deref()), + leading_keys: crate::leading_keys::single_table_leading_keys( + input, + operation, + key_info, + state.limits.max_expression_tokens, + ), attributes: extract_attributes(input), select: input .get("Select") @@ -167,43 +271,7 @@ pub(crate) async fn authorize_request( ) .await?; - Ok(key_info) -} - -/// Extract partition key values from the request body for `dynamodb:LeadingKeys`. -/// -/// For item-level operations (`GetItem`, `PutItem`, `DeleteItem`, `UpdateItem`), -/// extracts the partition key value from the `Key` or `Item` using the table's -/// key schema. For `Query`, the leading key comes from `KeyConditionExpression` -/// values, but extracting that requires expression parsing — deferred to the -/// engine layer. -/// Returns `None` for table-level and batch/transact operations, or when -/// `pk_attr` is not available. -fn extract_leading_keys( - input: &Value, - operation: &str, - pk_attr: Option<&str>, -) -> Option> { - let pk_attr = pk_attr?; - match operation { - "GetItem" | "DeleteItem" | "UpdateItem" => extract_pk_value(input.get("Key")?, pk_attr), - "PutItem" => extract_pk_value(input.get("Item")?, pk_attr), - _ => None, - } -} - -/// Extract the partition key value from a `DynamoDB` key/item map using the -/// known PK attribute name. -/// -/// `DynamoDB` keys are `{"attrName": {"S": "value"}}`. We extract the typed -/// value of the partition key attribute as a string. -fn extract_pk_value(map: &Value, pk_attr: &str) -> Option> { - let obj = map.as_object()?; - let type_val = obj.get(pk_attr)?; - let type_obj = type_val.as_object()?; - let (_, val) = type_obj.iter().next()?; - let s = val.as_str().unwrap_or_default(); - Some(vec![s.to_owned()]) + Ok(authorized_key_info) } /// Extract attribute names from the request for `dynamodb:Attributes`. @@ -231,69 +299,6 @@ pub(crate) fn extract_attributes(input: &Value) -> Option> { if names.is_empty() { None } else { Some(names) } } -/// IAM `(action, table)` pairs to authorize for a batch/transact operation. -/// -/// Returns `None` for non-batch/transact operations (the caller uses the -/// single-table path). Actions follow the AWS IAM Service Authorization -/// Reference: `BatchGetItem`/`BatchWriteItem` authorize under their own action -/// per table; `TransactGetItems` authorizes as `GetItem` per item's table; -/// `TransactWriteItems` authorizes per sub-op (`Put`→`PutItem`, -/// `Delete`→`DeleteItem`, `Update`→`UpdateItem`, `ConditionCheck`→ -/// `ConditionCheckItem`). Pairs are de-duplicated. Table names are the bare -/// operation-string form (no `dynamodb:` prefix), matching `check_authorization` -/// which prepends `dynamodb:`. -fn batch_transact_authz_targets(operation: &str, input: &Value) -> Option> { - let mut out: Vec<(String, String)> = Vec::new(); - let mut push = |action: &str, table: &str| { - let pair = (action.to_owned(), table.to_owned()); - if !out.contains(&pair) { - out.push(pair); - } - }; - match operation { - "BatchGetItem" | "BatchWriteItem" => { - // RequestItems is a map keyed by table name. - let items = input.get("RequestItems")?.as_object()?; - for table in items.keys() { - push(operation, table); - } - } - "TransactGetItems" => { - let items = input.get("TransactItems")?.as_array()?; - for entry in items { - if let Some(t) = entry - .get("Get") - .and_then(|g| g.get("TableName")) - .and_then(Value::as_str) - { - push("GetItem", t); - } - } - } - "TransactWriteItems" => { - let items = input.get("TransactItems")?.as_array()?; - for entry in items { - for (sub, action) in [ - ("Put", "PutItem"), - ("Delete", "DeleteItem"), - ("Update", "UpdateItem"), - ("ConditionCheck", "ConditionCheckItem"), - ] { - if let Some(t) = entry - .get(sub) - .and_then(|s| s.get("TableName")) - .and_then(Value::as_str) - { - push(action, t); - } - } - } - } - _ => return None, - } - Some(out) -} - /// Build a `DynamoDB` table ARN for authorization. /// /// If no table name is available (e.g. `ListTables`, `DescribeEndpoints`), @@ -308,8 +313,85 @@ fn build_resource_arn(region: &str, account_id: &str, table_name: Option<&str>) #[cfg(test)] mod tests { use super::*; + use extenddb_core::types::TableKeyInfo; use serde_json::json; + fn sample_key_info() -> TableKeyInfo { + TableKeyInfo { + table_name: "t".to_owned(), + account_id: "123".to_owned(), + ..Default::default() + } + } + + #[test] + fn key_info_for_authz_present_table_returns_key_info() { + let out = + key_info_for_authz(Ok(sample_key_info())).expect("a present table is not an error"); + assert!(out.is_some(), "a present table must yield its key info"); + } + + #[test] + fn key_info_for_authz_missing_table_returns_none() { + let out = key_info_for_authz(Err(StorageError::TableNotFound("t".to_owned()))) + .expect("a missing table is not a lookup failure"); + assert!( + out.is_none(), + "a missing table must leave the leading key absent so the engine returns not-found" + ); + } + + #[test] + fn key_info_for_authz_lookup_error_fails_closed() { + // Each non-not-found fault fails closed, classified the same way the + // engine classifies it: transient and connection faults are retryable + // 503s, the catalog and backend-internal faults are 500s, and a + // not-active table is the engine's ResourceInUseException. None yields + // an Ok(Some)/Ok(None) that would authorize the request. + let service_unavailable = [ + StorageError::Transient("db hiccup".to_owned()), + StorageError::Connection("pool timeout".to_owned()), + ]; + for err in service_unavailable { + let out = key_info_for_authz(Err(err.clone())); + assert!( + matches!(out, Err(DynamoDbError::ServiceUnavailable(_))), + "a transient or connection fault must map to ServiceUnavailable, got: {out:?} for {err:?}" + ); + } + + let internal = [ + StorageError::Internal("backend fault".to_owned()), + StorageError::CatalogVersionMismatch { + expected: "3".to_owned(), + found: "2".to_owned(), + }, + StorageError::CatalogNotInitialized, + ]; + for err in internal { + let out = key_info_for_authz(Err(err.clone())); + assert!( + matches!(out, Err(DynamoDbError::InternalServerError(_))), + "a catalog or backend-internal fault must map to InternalServerError, got: {out:?} for {err:?}" + ); + } + + // TableNotActive maps to whatever the engine maps it to, so the server + // and the engine agree on a not-active table. It still fails closed (an + // Err, not an authorizing Ok). + let out = key_info_for_authz(Err(StorageError::TableNotActive("t".to_owned()))); + let engine = + extenddb_engine::storage_err_to_dynamo(StorageError::TableNotActive("t".to_owned())); + assert!( + matches!(out, Err(ref e) if std::mem::discriminant(e) == std::mem::discriminant(&engine)), + "TableNotActive must map to the engine's error class ({engine:?}), got: {out:?}" + ); + assert!( + matches!(engine, DynamoDbError::ResourceInUseException(_)), + "the engine maps TableNotActive to ResourceInUseException, got: {engine:?}" + ); + } + #[test] fn extract_attributes_resolves_expression_attribute_names() { let input = json!({ diff --git a/docs/manuals/07-upgrade-manual.md b/docs/manuals/07-upgrade-manual.md index bd98e8b7a..a57a75304 100755 --- a/docs/manuals/07-upgrade-manual.md +++ b/docs/manuals/07-upgrade-manual.md @@ -218,6 +218,85 @@ Catalog upgrades change the schema; behavior changes alter how the running serve interprets existing data or configuration. Review these before upgrading a live deployment, even when no catalog migration is required. +### v0.1.14 LeadingKeys now enforced on Query, batch and transaction operations + +**What changed.** `dynamodb:LeadingKeys` is now evaluated on `Query`, +`BatchGetItem`, `BatchWriteItem`, `TransactGetItems` and `TransactWriteItems`. +Before this release those five operations carried no LeadingKeys value, so a +`ForAllValues:StringEquals dynamodb:LeadingKeys` condition on them was satisfied +vacuously and a principal fenced to its own partition could read or write any +partition through them. The item operations (`GetItem`, `PutItem`, `UpdateItem`, +`DeleteItem`) already enforced the key; they change only for a numeric (`N`) +partition key, which is now normalized (see below). Before this release the +item operations compared the number exactly as the request spelled it. + +Details of how the key is bound, matching DynamoDB: + +- A `Query` on the base table or a local secondary index binds LeadingKeys to + the partition key value in the key condition (an LSI shares the base table's + partition key). +- A `Query` on a global secondary index binds LeadingKeys to the queried + index's partition key value (its first HASH attribute), not the base table's. +- `BatchGetItem` and `BatchWriteItem` are authorized per table over the set of + keys that table receives; `TransactGetItems` and `TransactWriteItems` are + authorized per item. A `ForAnyValue` condition therefore passes a mixed batch + but not a mixed transaction. +- `Scan` and `SearchVectors` carry no LeadingKeys, because they target no single + partition. A `ForAllValues` condition does not fence them (it is satisfied + vacuously); fence a full-table read with `ForAnyValue:StringEquals` or a + `Null` guard (`"Null": {"dynamodb:LeadingKeys": "false"}`) on the statement. +- A numeric (`N`) partition key is written into LeadingKeys the way the + service writes it, which depends on the operation. `GetItem`, `PutItem`, + `UpdateItem`, `DeleteItem`, `Query` and `BatchGetItem` use the normalized + number, so `10`, `010`, `10.0` and `1E1` all compare as `"10"`. + `BatchWriteItem`, `TransactGetItems` and `TransactWriteItems` keep the scale + the request was written with, as Java's `BigDecimal` prints it: `010` and + `1.0E1` compare as `"10"`, `10.0` as `"10.0"` and `1E1` as `"1E+1"`. The + policy value is matched as written. Under a fence of `["10"]`, a client that + sends `10.0` or `1E1` is allowed on the first group and denied on the second, + as on DynamoDB. +- A request outside the count limits of the batch and transaction operations + returns `ValidationException` before any policy is evaluated, as on DynamoDB: + a `TransactWriteItems` or `TransactGetItems` with no items or more than 100, + a `BatchGetItem` or `BatchWriteItem` with an empty `RequestItems` or a table + with no keys or requests, a `BatchGetItem` with more than 100 tables or keys + and a `BatchWriteItem` with more than 25 tables or requests. Before this + release a principal without permission on the table received + `AccessDeniedException` for each of these requests except an empty + `TransactItems` or `RequestItems`. + +**Who is affected.** Any deployment with a principal whose policy fences +`dynamodb:LeadingKeys` with `ForAllValues:StringEquals` and whose Query, batch or +transaction requests previously crossed its own partition. After upgrade those +cross-partition requests return `AccessDeniedException`. Own-partition access is +unchanged, except for numeric partition keys. + +A principal with a numeric LeadingKeys fence sees a change on the item +operations even if it never uses Query, batch or transactions. Under a fence of +`["10"]`, a `GetItem` on `010`, `10.0` or `1E1` was denied and is now allowed. +Under a fence written in a non-canonical form, such as `["010"]` or `["1E1"]`, +a request spelled the same way was allowed and is now denied: the policy value +is matched as written and no request is normalized to it. + +**Migration.** No catalog migration is required. Before upgrading, audit +LeadingKeys policies with `extenddb manage list-user-policies` / +`list-role-policies` and confirm: + +- A fence intended to cover reads and writes names the partition values the + principal legitimately uses on Query, batch and transaction, not only item + operations. +- Any policy relying on `Scan` or `SearchVectors` being fenced by a + `ForAllValues` LeadingKeys condition is rewritten to `ForAnyValue` or adds a + `Null` guard, since neither operation is fenced by `ForAllValues`. +- A numeric LeadingKeys value is written in the form each operation compares + (see the numeric partition key item above). A fence of `["10"]` covers every + spelling of 10 on the item operations, Query and BatchGetItem, and the + spellings `10`, `010`, `+10` and `1.0E1` on batch writes and transactions. A + fence value written as `010`, `10.0` or `1E1` matches nothing on the item + operations, Query and BatchGetItem; rewrite it as `10`. + +--- + ### v0.1.6 — IAM: bare operators on multivalued condition keys are no-ops (BR-7085) **What changed.** A **bare** condition operator (no `ForAnyValue:` / `ForAllValues:` diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md index 7eb64fe2d..2a8d86bec 100755 --- a/docs/manuals/10-security-model.md +++ b/docs/manuals/10-security-model.md @@ -131,6 +131,8 @@ extenddb supports the same access control patterns as real AWS IAM: **Fine-Grained Access Control (FGAC)**: Use `dynamodb:LeadingKeys` to restrict access to specific partition key values. Use `dynamodb:Attributes` to restrict which attributes can be read or written. Example: allow a user to access only items where the partition key matches their user ID. +`dynamodb:LeadingKeys` is evaluated on `GetItem`, `PutItem`, `UpdateItem`, `DeleteItem`, `Query`, `BatchGetItem`, `BatchWriteItem`, `TransactGetItems` and `TransactWriteItems`. A `Query` on a global secondary index binds the key to the index's partition key value (its first HASH attribute). `Scan` and `SearchVectors` target no single partition and carry no `dynamodb:LeadingKeys` value, so a `ForAllValues` condition does not fence them; fence a full-table read with `ForAnyValue:StringEquals` or a `Null` guard on the statement. A numeric partition key is compared as the normalized number on `GetItem`, `PutItem`, `UpdateItem`, `DeleteItem`, `Query` and `BatchGetItem`, and as Java's `BigDecimal` prints it (`10.0` stays `10.0`, `1E1` is `1E+1`) on `BatchWriteItem`, `TransactGetItems` and `TransactWriteItems`, which is what DynamoDB does. + ### Resource ARNs Resources are identified by ARN: `arn:aws:dynamodb:::table/`. Wildcard matching (`*`, `?`) in policy Resource fields follows AWS IAM conventions. `NotResource` is also supported. diff --git a/tests/test_fgac_leading_keys.py b/tests/test_fgac_leading_keys.py new file mode 100644 index 000000000..c0dd9f708 --- /dev/null +++ b/tests/test_fgac_leading_keys.py @@ -0,0 +1,1161 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 + +"""Fine-grained access control tests for dynamodb:LeadingKeys on Query, batch +and transaction operations. + +dynamodb:LeadingKeys restricts a principal to the partition key(s) it may +touch. It was enforced on the item operations (GetItem, PutItem, UpdateItem, +DeleteItem) but not on Query, BatchGetItem, BatchWriteItem, TransactGetItems or +TransactWriteItems, so a principal fenced to its own partition could read or +write any partition through those operations. These tests pin the enforced +behaviour. + +Every expected outcome here was measured against live DynamoDB and IAM on +2026-10-20 or 2026-10-21; each test that uses a 2026-10-21 measurement says +so. The assertions are identical whatever backend the suite targets; a +difference is a bug (REQ-TEST-004). + +Measured model: + - Query on the base table or an LSI binds LeadingKeys to the partition key + value in the key condition (an LSI shares the base partition key). + - Query on a GSI binds LeadingKeys to the queried index's partition key value. + - Scan sets no LeadingKeys: a ForAllValues allow is vacuously satisfied (scan + allowed), a ForAnyValue allow is not (scan denied). + - BatchGetItem/BatchWriteItem authorize per table over the set of that + table's keys: a ForAnyValue allow passes a mixed own+other batch. + - TransactGetItems/TransactWriteItems authorize per item: a ForAnyValue allow + denies a mixed own+other transaction, because the other item fails on its + own. + +Prerequisites: + - extenddb running with ``auth.provider = "builtin"`` on EXTENDDB_TEST_ENDPOINT + - Admin credentials in EXTENDDB_ADMIN_USER / EXTENDDB_ADMIN_PASSWORD env vars + +Run: + EXTENDDB_TEST_ENDPOINT=https://localhost:18443 \\ + EXTENDDB_ADMIN_USER=admin \\ + EXTENDDB_ADMIN_PASSWORD= \\ + pytest tests/test_fgac_leading_keys.py -v + +REQ-TEST-001, REQ-TEST-004, REQ-AUTH-002 +""" + +from __future__ import annotations + +import os +import time +import uuid +from typing import Any + +import boto3 +import pytest +from botocore.config import Config as BotoConfig +from botocore.exceptions import ClientError + +from conftest import wait_for_active, wait_for_deleted +from management_helpers import ManagementClient + +OWN = "alice" +OTHER = "bob" +G_OWN = "galice" # the alice item's GSI partition key value +G_OTHER = "gbob" # the bob item's GSI partition key value + + +def _require_auth_env() -> tuple[str, str, str]: + endpoint = os.environ.get("EXTENDDB_TEST_ENDPOINT", "").strip() + admin_user = os.environ.get("EXTENDDB_ADMIN_USER", "").strip() + admin_pass = os.environ.get("EXTENDDB_ADMIN_PASSWORD", "").strip() + if not endpoint or not admin_user or not admin_pass: + pytest.fail( + "MISCONFIGURED: LeadingKeys FGAC tests require EXTENDDB_TEST_ENDPOINT, " + "EXTENDDB_ADMIN_USER, and EXTENDDB_ADMIN_PASSWORD. " + "These must be set by devtools/run-tests before test execution." + ) + return endpoint, admin_user, admin_pass + + +def _make_client(endpoint_url: str, access_key: str, secret_key: str, + region: str, *, parameter_validation: bool = True) -> Any: + kwargs: dict = dict( + service_name="dynamodb", + endpoint_url=endpoint_url, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + region_name=region, + config=BotoConfig(retries={"max_attempts": 0}, + parameter_validation=parameter_validation), + ) + if endpoint_url.startswith("https://"): + kwargs["verify"] = False + return boto3.client(**kwargs) + + +def _denied(exc: ClientError) -> bool: + return exc.value.response["Error"]["Code"] == "AccessDeniedException" + + +@pytest.fixture(scope="module") +def auth_env(): + return _require_auth_env() + + +@pytest.fixture(scope="module") +def mgmt(auth_env) -> ManagementClient: + endpoint, admin_user, admin_pass = auth_env + return ManagementClient(endpoint, admin_user, admin_pass) + + +@pytest.fixture(scope="module") +def region() -> str: + return os.environ.get("AWS_DEFAULT_REGION", "us-east-1") + + +@pytest.fixture(scope="module") +def account_id(mgmt) -> str: + acct_id = f"{uuid.uuid4().int % 10**12:012d}" + resp = mgmt.create_account(acct_id, f"lk-test-{acct_id}") + assert resp.status_code == 201, resp.text + yield acct_id + mgmt.delete_account(acct_id) + + +ALLOW_ALL = { + "Version": "2012-10-17", + "Statement": [{"Effect": "Allow", "Action": "dynamodb:*", "Resource": "*"}], +} + + +class TestLeadingKeys: + """LeadingKeys enforcement on Query, batch and transaction operations.""" + + @pytest.fixture(autouse=True) + def setup(self, auth_env, mgmt, account_id, region): + self.endpoint = auth_env[0] + self.mgmt = mgmt + self.account_id = account_id + self.region = region + self._users: list[str] = [] + yield + for u in self._users: + try: + self.mgmt.delete_user(self.account_id, u) + except Exception: + pass + + # -- helpers ----------------------------------------------------------- + + def _user_client(self, policies: list[tuple[str, dict]], *, + parameter_validation: bool = True) -> Any: + """Create a user with the given named policies, return a client.""" + name = f"lk-{uuid.uuid4().hex[:8]}" + resp = self.mgmt.create_user(self.account_id, name, "TestPass123!") + assert resp.status_code == 201, resp.text + self._users.append(name) + resp = self.mgmt.create_access_key(self.account_id, name) + assert resp.status_code == 201, resp.text + creds = resp.json() + for pname, doc in policies: + resp = self.mgmt.put_user_policy(self.account_id, name, pname, doc) + assert resp.status_code == 204, resp.text + return _make_client(self.endpoint, creds["access_key_id"], + creds["secret_access_key"], self.region, + parameter_validation=parameter_validation) + + def _leading_keys_policy(self, qualifier: str, values: list[str], + table: str) -> dict: + arn = f"arn:aws:dynamodb:{self.region}:{self.account_id}:table/{table}" + op = f"{qualifier}:StringEquals" if qualifier else "StringEquals" + return { + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "dynamodb:*", + "Resource": [arn, arn + "/index/*"], + "Condition": {op: {"dynamodb:LeadingKeys": values}}, + }], + } + + def _make_table(self) -> tuple[str, Any]: + """Owner (full access), a table with a sort key, one LSI and one GSI, + seeded with an own ("alice") and an other ("bob") partition.""" + owner = f"owner-{uuid.uuid4().hex[:8]}" + resp = self.mgmt.create_user(self.account_id, owner, "TestPass123!") + assert resp.status_code == 201, resp.text + self._users.append(owner) + resp = self.mgmt.create_access_key(self.account_id, owner) + assert resp.status_code == 201, resp.text + creds = resp.json() + resp = self.mgmt.put_user_policy(self.account_id, owner, "full", ALLOW_ALL) + assert resp.status_code == 204, resp.text + client = _make_client(self.endpoint, creds["access_key_id"], + creds["secret_access_key"], self.region) + + table = f"lk-{uuid.uuid4().hex[:8]}" + client.create_table( + TableName=table, + AttributeDefinitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}, + {"AttributeName": "lsi_sk", "AttributeType": "S"}, + {"AttributeName": "gpk", "AttributeType": "S"}, + {"AttributeName": "gsk", "AttributeType": "S"}, + ], + KeySchema=[ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}, + ], + LocalSecondaryIndexes=[{ + "IndexName": "lsi1", + "KeySchema": [ + {"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "lsi_sk", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + }], + GlobalSecondaryIndexes=[{ + "IndexName": "gsi1", + "KeySchema": [ + {"AttributeName": "gpk", "KeyType": "HASH"}, + {"AttributeName": "gsk", "KeyType": "RANGE"}, + ], + "Projection": {"ProjectionType": "ALL"}, + }], + BillingMode="PAY_PER_REQUEST", + ) + wait_for_active(client, table) + client.put_item(TableName=table, Item={ + "pk": {"S": OWN}, "sk": {"S": "x"}, "lsi_sk": {"S": "l"}, + "gpk": {"S": G_OWN}, "gsk": {"S": "g"}, "data": {"S": "own"}}) + client.put_item(TableName=table, Item={ + "pk": {"S": OTHER}, "sk": {"S": "x"}, "lsi_sk": {"S": "l"}, + "gpk": {"S": G_OTHER}, "gsk": {"S": "g"}, "data": {"S": "other"}}) + return table, client + + @pytest.fixture() + def table(self): + name, owner = self._make_table() + yield name + try: + owner.delete_table(TableName=name) + wait_for_deleted(owner, name) + except Exception: + pass + + @staticmethod + def _key(pk: str) -> dict: + return {"pk": {"S": pk}, "sk": {"S": "x"}} + + # -- item operations: the pre-existing enforcement still holds ---------- + + def test_item_ops_fenced_to_own_partition(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.get_item(TableName=table, Key=self._key(OWN)) + with pytest.raises(ClientError) as e: + c.get_item(TableName=table, Key=self._key(OTHER)) + assert _denied(e) + c.put_item(TableName=table, Item=self._key(OWN)) + with pytest.raises(ClientError) as e: + c.put_item(TableName=table, Item=self._key(OTHER)) + assert _denied(e) + + # -- Query: the core defect -------------------------------------------- + + def test_query_own_partition_allowed(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.query(TableName=table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": OWN}}) + + def test_query_other_partition_denied(self, table): + # The bypass in the report: before the fix this returned the other + # tenant's rows instead of AccessDeniedException. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + with pytest.raises(ClientError) as e: + c.query(TableName=table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": OTHER}}) + assert _denied(e) + + def test_query_other_via_expression_attribute_name_denied(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + with pytest.raises(ClientError) as e: + c.query(TableName=table, KeyConditionExpression="#p = :v", + ExpressionAttributeNames={"#p": "pk"}, + ExpressionAttributeValues={":v": {"S": OTHER}}) + assert _denied(e) + + def test_query_legacy_key_conditions_fenced(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.query(TableName=table, KeyConditions={ + "pk": {"ComparisonOperator": "EQ", "AttributeValueList": [{"S": OWN}]}}) + with pytest.raises(ClientError) as e: + c.query(TableName=table, KeyConditions={ + "pk": {"ComparisonOperator": "EQ", + "AttributeValueList": [{"S": OTHER}]}}) + assert _denied(e) + + def test_query_lsi_uses_base_partition_key(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.query(TableName=table, IndexName="lsi1", + KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": OWN}}) + with pytest.raises(ClientError) as e: + c.query(TableName=table, IndexName="lsi1", + KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": OTHER}}) + assert _denied(e) + + def test_query_gsi_uses_index_partition_key(self, table): + # Under a base-partition policy, the GSI partition value (galice) is not + # the permitted value (alice), so the GSI query is denied. + base = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + with pytest.raises(ClientError) as e: + base.query(TableName=table, IndexName="gsi1", + KeyConditionExpression="gpk = :v", + ExpressionAttributeValues={":v": {"S": G_OWN}}) + assert _denied(e) + # A policy naming the GSI partition value permits that GSI query, and + # still denies a base-table query on the base partition (alice). + idx = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [G_OWN], table))]) + idx.query(TableName=table, IndexName="gsi1", + KeyConditionExpression="gpk = :v", + ExpressionAttributeValues={":v": {"S": G_OWN}}) + with pytest.raises(ClientError) as e: + idx.query(TableName=table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": OWN}}) + assert _denied(e) + + # -- Scan: no LeadingKeys ---------------------------------------------- + + def test_scan_allowed_under_for_all_values(self, table): + # Scan targets no partition, so a ForAllValues allow is vacuously met. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.scan(TableName=table) + + def test_scan_denied_under_for_any_value(self, table): + # A ForAnyValue allow needs at least one matching key; a scan has none. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAnyValue", [OWN], table))]) + with pytest.raises(ClientError) as e: + c.scan(TableName=table) + assert _denied(e) + + # -- BatchGetItem / BatchWriteItem: per-table key set ------------------ + + def test_batch_get_fenced_to_own_partition(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.batch_get_item(RequestItems={table: {"Keys": [self._key(OWN)]}}) + with pytest.raises(ClientError) as e: + c.batch_get_item(RequestItems={table: {"Keys": [self._key(OTHER)]}}) + assert _denied(e) + # Mixed own+other under ForAllValues: every key must match, so denied. + with pytest.raises(ClientError) as e: + c.batch_get_item(RequestItems={ + table: {"Keys": [self._key(OWN), self._key(OTHER)]}}) + assert _denied(e) + + def test_batch_write_fenced_to_own_partition(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.batch_write_item(RequestItems={ + table: [{"PutRequest": {"Item": {"pk": {"S": OWN}, "sk": {"S": "bw"}}}}]}) + with pytest.raises(ClientError) as e: + c.batch_write_item(RequestItems={ + table: [{"PutRequest": {"Item": {"pk": {"S": OTHER}, "sk": {"S": "bw"}}}}]}) + assert _denied(e) + + def test_batch_get_mixed_passes_for_any_value(self, table): + # A batch authorizes per table over the whole key set, so ForAnyValue is + # satisfied when any key matches, even in a mixed own+other batch. This + # is the distinction from a transaction (next test). + c = self._user_client([("lk", self._leading_keys_policy( + "ForAnyValue", [OWN], table))]) + c.batch_get_item(RequestItems={ + table: {"Keys": [self._key(OWN), self._key(OTHER)]}}) + + # -- TransactGetItems / TransactWriteItems: per item ------------------- + + def test_transact_get_fenced_to_own_partition(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": self._key(OWN)}}]) + with pytest.raises(ClientError) as e: + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": self._key(OTHER)}}]) + assert _denied(e) + + def test_transact_write_fenced_to_own_partition(self, table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [OWN], table))]) + c.transact_write_items(TransactItems=[ + {"Put": {"TableName": table, "Item": {"pk": {"S": OWN}, "sk": {"S": "tw"}}}}]) + with pytest.raises(ClientError) as e: + c.transact_write_items(TransactItems=[ + {"Put": {"TableName": table, "Item": {"pk": {"S": OTHER}, "sk": {"S": "tw"}}}}]) + assert _denied(e) + # Each sub-op kind is authorized as its own action; an other-partition + # Update, Delete or ConditionCheck is denied too. + for item in [ + {"Update": {"TableName": table, "Key": self._key(OTHER), + "UpdateExpression": "SET d = :v", + "ExpressionAttributeValues": {":v": {"S": "u"}}}}, + {"Delete": {"TableName": table, "Key": self._key(OTHER)}}, + {"ConditionCheck": {"TableName": table, "Key": self._key(OTHER), + "ConditionExpression": "attribute_exists(pk)"}}, + ]: + with pytest.raises(ClientError) as e: + c.transact_write_items(TransactItems=[item]) + assert _denied(e) + + def test_transact_get_mixed_denied_for_any_value(self, table): + # A transaction authorizes each item on its own, so a ForAnyValue allow + # does NOT pass a mixed own+other transaction: the other item fails. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAnyValue", [OWN], table))]) + with pytest.raises(ClientError) as e: + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": self._key(OWN)}}, + {"Get": {"TableName": table, "Key": self._key(OTHER)}}]) + assert _denied(e) + + # -- GSI with a two-attribute partition key ---------------------------- + # + # Live DynamoDB accepts a GSI whose partition key has two HASH attributes + # (g1, g2); a two-HASH base table is rejected. Measured 2026-10-20: a Query + # on such a GSI evaluates dynamodb:LeadingKeys against the FIRST HASH + # attribute's value only. Under ForAllValues:StringEquals ["alice"] a query + # of g1=alice was allowed for any g2 value and g1=bob was denied; a policy + # naming the second attribute's value ("blue") denied even g1=alice,g2=blue. + + def _make_multi_hash_gsi_table(self) -> tuple[str, Any]: + owner = f"owner-{uuid.uuid4().hex[:8]}" + resp = self.mgmt.create_user(self.account_id, owner, "TestPass123!") + assert resp.status_code == 201, resp.text + self._users.append(owner) + resp = self.mgmt.create_access_key(self.account_id, owner) + assert resp.status_code == 201, resp.text + creds = resp.json() + resp = self.mgmt.put_user_policy(self.account_id, owner, "full", ALLOW_ALL) + assert resp.status_code == 204, resp.text + client = _make_client(self.endpoint, creds["access_key_id"], + creds["secret_access_key"], self.region) + table = f"lk-mh-{uuid.uuid4().hex[:8]}" + client.create_table( + TableName=table, + AttributeDefinitions=[ + {"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "g1", "AttributeType": "S"}, + {"AttributeName": "g2", "AttributeType": "S"}, + ], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + GlobalSecondaryIndexes=[{ + "IndexName": "gsi2", + "KeySchema": [ + {"AttributeName": "g1", "KeyType": "HASH"}, + {"AttributeName": "g2", "KeyType": "HASH"}, + ], + "Projection": {"ProjectionType": "ALL"}, + }], + BillingMode="PAY_PER_REQUEST", + ) + wait_for_active(client, table) + for pk, g1, g2 in [("i1", "alice", "blue"), ("i2", "alice", "red"), + ("i3", "bob", "blue"), ("i4", "bob", "red")]: + client.put_item(TableName=table, Item={ + "pk": {"S": pk}, "g1": {"S": g1}, "g2": {"S": g2}}) + return table, client + + @pytest.fixture() + def mh_table(self): + name, owner = self._make_multi_hash_gsi_table() + yield name + try: + owner.delete_table(TableName=name) + wait_for_deleted(owner, name) + except Exception: + pass + + @staticmethod + def _q_gsi2(client, table, g1, g2): + return client.query( + TableName=table, IndexName="gsi2", + KeyConditionExpression="g1 = :a AND g2 = :b", + ExpressionAttributeValues={":a": {"S": g1}, ":b": {"S": g2}}) + + def test_multi_hash_gsi_binds_first_hash_attribute_only(self, mh_table): + # Policy fences the first HASH attribute (g1) to "alice". + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["alice"], mh_table))]) + # g1=alice is allowed for either g2 value: the second attribute is not + # part of LeadingKeys. + self._q_gsi2(c, mh_table, "alice", "blue") + self._q_gsi2(c, mh_table, "alice", "red") + # g1=bob is denied regardless of g2. + with pytest.raises(ClientError) as e: + self._q_gsi2(c, mh_table, "bob", "blue") + assert _denied(e) + with pytest.raises(ClientError) as e: + self._q_gsi2(c, mh_table, "bob", "red") + assert _denied(e) + + def test_multi_hash_gsi_second_attribute_value_not_in_leading_keys(self, mh_table): + # A policy naming the second HASH attribute's value ("blue") matches no + # leading key, so even g1=alice,g2=blue is denied. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["blue"], mh_table))]) + with pytest.raises(ClientError) as e: + self._q_gsi2(c, mh_table, "alice", "blue") + assert _denied(e) + + def test_multi_hash_gsi_base_partition_key_unchanged(self, mh_table): + # The base table partition key still binds to its own value when the + # multi-attribute GSI exists. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["i1"], mh_table))]) + c.get_item(TableName=mh_table, Key={"pk": {"S": "i1"}}) + c.query(TableName=mh_table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": "i1"}}) + with pytest.raises(ClientError) as e: + c.get_item(TableName=mh_table, Key={"pk": {"S": "i3"}}) + assert _denied(e) + + # -- Numeric (N) partition key rendering -------------------------------- + # + # The service writes an N partition key into dynamodb:LeadingKeys in one of + # two forms, depending on the operation, and matches the policy value as + # written (measured against live DynamoDB on 2026-10-20 over 21 spellings): + # - GetItem, PutItem, UpdateItem, DeleteItem, Query and BatchGetItem use + # the normalized number, so 010, 10.0, 1E1, 1.0E1 and +10 are all "10"; + # - BatchWriteItem (put and delete), TransactGetItems and every + # TransactWriteItems action use the number as Java's BigDecimal prints + # it: 010, +10 and 1.0E1 are "10", 10.0 is "10.0", 1E1 is "1E+1". + # A fence of ["1E1"] matches nothing on either path. + + def _make_num_table(self) -> tuple[str, Any]: + owner = f"owner-{uuid.uuid4().hex[:8]}" + resp = self.mgmt.create_user(self.account_id, owner, "TestPass123!") + assert resp.status_code == 201, resp.text + self._users.append(owner) + resp = self.mgmt.create_access_key(self.account_id, owner) + assert resp.status_code == 201, resp.text + creds = resp.json() + resp = self.mgmt.put_user_policy(self.account_id, owner, "full", ALLOW_ALL) + assert resp.status_code == 204, resp.text + client = _make_client(self.endpoint, creds["access_key_id"], + creds["secret_access_key"], self.region) + table = f"lk-n-{uuid.uuid4().hex[:8]}" + client.create_table( + TableName=table, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "N"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], + BillingMode="PAY_PER_REQUEST", + ) + wait_for_active(client, table) + client.put_item(TableName=table, Item={ + "pk": {"N": "10"}, "secret": {"S": "ten-secret"}}) + client.put_item(TableName=table, Item={ + "pk": {"N": "20"}, "secret": {"S": "twenty-secret"}}) + return table, client + + @pytest.fixture() + def num_table(self): + name, owner = self._make_num_table() + yield name + try: + owner.delete_table(TableName=name) + wait_for_deleted(owner, name) + except Exception: + pass + + # Every non-canonical spelling of the principal's own numeric partition is + # allowed under a canonical fence, and the other partition stays denied. + NONCANON_OWN = ["010", "10.0", "1E1", "1.0E1", "+10"] + + def test_num_item_ops_canonical_fence_allows_noncanonical_own(self, num_table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["10"], num_table))]) + for spelling in self.NONCANON_OWN: + c.get_item(TableName=num_table, Key={"pk": {"N": spelling}}) + # The other partition stays denied whatever the spelling. + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": "20"}}) + assert _denied(e) + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": "2.0E1"}}) + assert _denied(e) + + def test_num_query_canonical_fence_allows_noncanonical_own(self, num_table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["10"], num_table))]) + for spelling in self.NONCANON_OWN: + c.query(TableName=num_table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"N": spelling}}) + with pytest.raises(ClientError) as e: + c.query(TableName=num_table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"N": "20"}}) + assert _denied(e) + + def test_num_batch_get_canonical_fence_allows_noncanonical_own(self, num_table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["10"], num_table))]) + # BatchGetItem compares the normalized number. 1E1 and 10.0 both + # normalize to partition 10, so they cannot share one batch (the + # engine rejects duplicate keys), hence one key per request. + for spelling in self.NONCANON_OWN: + c.batch_get_item(RequestItems={ + num_table: {"Keys": [{"pk": {"N": spelling}}]}}) + with pytest.raises(ClientError) as e: + c.batch_get_item(RequestItems={num_table: {"Keys": [{"pk": {"N": "20"}}]}}) + assert _denied(e) + # A mixed own+other batch under ForAllValues is denied: the other key + # does not normalize to 10. + with pytest.raises(ClientError) as e: + c.batch_get_item(RequestItems={ + num_table: {"Keys": [{"pk": {"N": "010"}}, {"pk": {"N": "20"}}]}}) + assert _denied(e) + + # TransactGetItems, every TransactWriteItems action and BatchWriteItem put + # requests compare the number as Java's BigDecimal prints it, which keeps + # the scale the request was written with: 10.0 stays "10.0" and 1E1 is + # "1E+1", while 010, +10 and 1.0E1 are "10". + DECIMAL_SAME = ["10", "010", "+10", "1.0E1"] + DECIMAL_DIFFERENT = ["10.0", "1E1"] + + def _decimal_path_ops(self, c, table: str, spelling: str) -> dict: + key = {"pk": {"N": spelling}} + return { + "transact_get": lambda: c.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": key}}]), + "transact_put": lambda: c.transact_write_items(TransactItems=[ + {"Put": {"TableName": table, "Item": key}}]), + "transact_update": lambda: c.transact_write_items(TransactItems=[ + {"Update": {"TableName": table, "Key": key, + "UpdateExpression": "SET a = :x", + "ExpressionAttributeValues": {":x": {"S": "y"}}}}]), + "transact_delete": lambda: c.transact_write_items(TransactItems=[ + {"Delete": {"TableName": table, "Key": key}}]), + "transact_condition_check": lambda: c.transact_write_items(TransactItems=[ + {"ConditionCheck": {"TableName": table, "Key": key, + "ConditionExpression": + "attribute_exists(pk) OR attribute_not_exists(pk)"}}]), + "batch_write_put": lambda: c.batch_write_item(RequestItems={ + table: [{"PutRequest": {"Item": key}}]}), + "batch_write_delete": lambda: c.batch_write_item(RequestItems={ + table: [{"DeleteRequest": {"Key": key}}]}), + } + + def test_num_decimal_path_canonical_fence(self, num_table): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["10"], num_table))]) + for spelling in self.DECIMAL_SAME: + for name, op in self._decimal_path_ops(c, num_table, spelling).items(): + op() + for spelling in self.DECIMAL_DIFFERENT: + for name, op in self._decimal_path_ops(c, num_table, spelling).items(): + with pytest.raises(ClientError) as e: + op() + assert _denied(e), (spelling, name) + with pytest.raises(ClientError) as e: + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": num_table, "Key": {"pk": {"N": "20"}}}}]) + assert _denied(e) + + def test_num_decimal_fence_matches_only_the_decimal_path(self, num_table): + # A fence written as "10.0" matches a request spelled 10.0 on the + # decimal path and matches nothing on the normalized path, where every + # spelling of 10 is "10". The same holds for "1E+1" and the spelling 1E1. + for fence, spelling in (("10.0", "10.0"), ("1E+1", "1E1")): + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", [fence], num_table))]) + for name, op in self._decimal_path_ops(c, num_table, spelling).items(): + op() + for other in ("10", "010", "1.0E1"): + with pytest.raises(ClientError) as e: + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": num_table, "Key": {"pk": {"N": other}}}}]) + assert _denied(e), (fence, other) + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": spelling}}) + assert _denied(e), fence + with pytest.raises(ClientError) as e: + c.query(TableName=num_table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"N": spelling}}) + assert _denied(e), fence + with pytest.raises(ClientError) as e: + c.batch_get_item(RequestItems={ + num_table: {"Keys": [{"pk": {"N": spelling}}]}}) + assert _denied(e), fence + + def test_num_deny_statement_follows_the_operation_form(self, num_table): + # A Deny on ["10"] matches every spelling of 10 on the normalized path, + # and on the decimal path only the spellings whose decimal form is "10". + arn = f"arn:aws:dynamodb:{self.region}:{self.account_id}:table/{num_table}" + c = self._user_client([("lk", { + "Version": "2012-10-17", + "Statement": [ + {"Effect": "Allow", "Action": "dynamodb:*", "Resource": arn}, + {"Effect": "Deny", "Action": "dynamodb:*", "Resource": arn, + "Condition": {"ForAnyValue:StringEquals": {"dynamodb:LeadingKeys": ["10"]}}}, + ]})]) + for spelling in ("10", "10.0"): + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": spelling}}) + assert _denied(e), spelling + with pytest.raises(ClientError) as e: + c.delete_item(TableName=num_table, Key={"pk": {"N": spelling}}) + assert _denied(e), spelling + decimal_ops = ("transact_get", "batch_write_put", "batch_write_delete") + ops_10_0 = self._decimal_path_ops(c, num_table, "10.0") + for name in decimal_ops: + ops_10_0[name]() + ops_10 = self._decimal_path_ops(c, num_table, "10") + for name in decimal_ops: + with pytest.raises(ClientError) as e: + ops_10[name]() + assert _denied(e), name + + def test_num_decimal_path_edge_spellings(self, num_table): + # The decimal form of each spelling, as the service matched it on + # TransactGetItems. The normalized form of each differs, so GetItem is + # denied under the same fence. + cases = {"100E-2": "1.00", "0.0000001": "1E-7", "123E2": "1.23E+4", + "-5.50": "-5.50", "+1.5E+3": "1.5E+3", "12.3400E-1": "1.23400", + "1.00E-130": "1.00E-130", "80E-1": "8.0"} + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", sorted(cases.values()), num_table))]) + for spelling in cases: + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": num_table, "Key": {"pk": {"N": spelling}}}}]) + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": spelling}}) + assert _denied(e), spelling + + def test_num_noncanonical_fence_matches_nothing(self, num_table): + # The policy value ["1E1"] is not normalized by the service, so no + # request normalizes to it: every operation is denied, including a + # request spelled 1E1 that reads canonical partition 10. Before the fix + # ExtendDB served partition 10 here. + c = self._user_client([("lk", self._leading_keys_policy( + "ForAllValues", ["1E1"], num_table))]) + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": "1E1"}}) + assert _denied(e) + with pytest.raises(ClientError) as e: + c.get_item(TableName=num_table, Key={"pk": {"N": "10"}}) + assert _denied(e) + with pytest.raises(ClientError) as e: + c.query(TableName=num_table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"N": "1E1"}}) + assert _denied(e) + with pytest.raises(ClientError) as e: + c.batch_get_item(RequestItems={num_table: {"Keys": [{"pk": {"N": "1E1"}}]}}) + assert _denied(e) + with pytest.raises(ClientError) as e: + c.transact_get_items(TransactItems=[ + {"Get": {"TableName": num_table, "Key": {"pk": {"N": "1E1"}}}}]) + assert _denied(e) + + # -- Authorization and the engine agree on the queried partition -------- + # + # Each Query shape below is sent as the table owner, as a principal fenced + # to the partition the shape names and as a principal fenced to a value no + # item has. The owner's result shows which partition the engine executed. + # The fenced principal must get the same rows, and the other principal must + # be denied, so a shape where authorization and the engine read a different + # partition key (or authorization reads none) fails here. The engine side is + # the server's own Query path, not a copy of it. + # + # Measured against live DynamoDB on 2026-10-21 with the same shapes on + # tables with the same key schemas: every accepted shape returned the named + # partition to the owner and to the fenced principal and was denied to the + # other principal. Every rejected shape returned ValidationException to the + # owner and to the fenced principal. To the other principal the service + # returned ValidationException, except where QUERY_REJECTED_OTHER_CODE + # names a different code. For the shapes in that map ExtendDB returns the + # other of the two codes, so the test asserts only that no data is + # returned. + + QUERY_ACCEPTED = [ + ("pk eq placeholder", "main", {"KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}, + "pk", "alice"), + ("hash name ref", "main", {"KeyConditionExpression": "#p = :v", + "ExpressionAttributeNames": {"#p": "pk"}, + "ExpressionAttributeValues": {":v": {"S": "alice"}}}, + "pk", "alice"), + ("single parens", "main", {"KeyConditionExpression": "(pk = :v)", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}, + "pk", "alice"), + ("pk and sk eq", "main", {"KeyConditionExpression": "pk = :v AND sk = :s", + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":s": {"S": "x"}}}, + "pk", "alice"), + ("sk before pk", "main", {"KeyConditionExpression": "sk = :s AND pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":s": {"S": "x"}}}, + "pk", "alice"), + ("no spaces", "main", {"KeyConditionExpression": "pk=:v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}, + "pk", "alice"), + ("extra spaces", "main", {"KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}, + "pk", "alice"), + ("begins_with sk", "main", {"KeyConditionExpression": "pk = :v AND begins_with(sk, :p)", + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":p": {"S": "x"}}}, + "pk", "alice"), + ("between sk", "main", {"KeyConditionExpression": "pk = :v AND sk BETWEEN :lo AND :hi", + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":lo": {"S": "a"}, + ":hi": {"S": "z"}}}, + "pk", "alice"), + ("both name refs", "main", {"KeyConditionExpression": "#p = :v AND #s = :s", + "ExpressionAttributeNames": {"#p": "pk", "#s": "sk"}, + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":s": {"S": "x"}}}, + "pk", "alice"), + ("lowercase and", "main", {"KeyConditionExpression": "pk = :v and sk = :s", + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":s": {"S": "x"}}}, + "pk", "alice"), + ("parenthesized clauses", "main", {"KeyConditionExpression": "(pk = :v) AND (sk = :s)", + "ExpressionAttributeValues": {":v": {"S": "alice"}, + ":s": {"S": "x"}}}, + "pk", "alice"), + ("number key", "num", {"KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"N": "10"}}}, + "pk", "10"), + ("legacy eq", "main", {"KeyConditions": { + "pk": {"ComparisonOperator": "EQ", "AttributeValueList": [{"S": "alice"}]}}}, + "pk", "alice"), + ("legacy eq with sk", "main", {"KeyConditions": { + "pk": {"ComparisonOperator": "EQ", "AttributeValueList": [{"S": "alice"}]}, + "sk": {"ComparisonOperator": "BEGINS_WITH", "AttributeValueList": [{"S": "x"}]}}}, + "pk", "alice"), + ("lsi base pk", "main", {"IndexName": "lsi1", "KeyConditionExpression": "pk = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}, + "pk", "alice"), + ("gsi index pk", "main", {"IndexName": "gsi1", "KeyConditionExpression": "gpk = :v", + "ExpressionAttributeValues": {":v": {"S": G_OWN}}}, + "gpk", G_OWN), + ("multi hash gsi", "mh", {"IndexName": "gsi2", + "KeyConditionExpression": "g1 = :a AND g2 = :b", + "ExpressionAttributeValues": {":a": {"S": "alice"}, + ":b": {"S": "blue"}}}, + "g1", "alice"), + ("multi hash gsi swapped", "mh", {"IndexName": "gsi2", + "KeyConditionExpression": "g2 = :b AND g1 = :a", + "ExpressionAttributeValues": {":a": {"S": "alice"}, + ":b": {"S": "blue"}}}, + "g1", "alice"), + ] + + QUERY_REJECTED = [ + ("non-eq pk", "main", {"KeyConditionExpression": "pk > :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}), + ("malformed", "main", {"KeyConditionExpression": "pk :: :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}), + ("sk only", "main", {"KeyConditionExpression": "sk = :s", + "ExpressionAttributeValues": {":s": {"S": "x"}}}), + ("unresolved name", "main", {"KeyConditionExpression": "#p = :v", + "ExpressionAttributeValues": {":v": {"S": "alice"}}}), + ("legacy non-eq pk", "main", {"KeyConditions": { + "pk": {"ComparisonOperator": "LE", "AttributeValueList": [{"S": "alice"}]}}}), + ("legacy begins_with pk", "main", {"KeyConditions": { + "pk": {"ComparisonOperator": "BEGINS_WITH", + "AttributeValueList": [{"S": "alice"}]}}}), + ("multi hash gsi missing g2", "mh", {"IndexName": "gsi2", + "KeyConditionExpression": "g1 = :a", + "ExpressionAttributeValues": {":a": {"S": "alice"}}}), + ] + + # Rejected shapes where the service and ExtendDB return different error + # codes to the principal fenced away from the value. The value is the code + # the service returned. + QUERY_REJECTED_OTHER_CODE = { + "non-eq pk": "AccessDeniedException", + "legacy non-eq pk": "AccessDeniedException", + "legacy begins_with pk": "AccessDeniedException", + "multi hash gsi missing g2": "ValidationException", + } + + def _fence_policy(self, values: list[str], tables: list[str]) -> dict: + resources = [] + for t in tables: + arn = f"arn:aws:dynamodb:{self.region}:{self.account_id}:table/{t}" + resources += [arn, arn + "/index/*"] + return { + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "dynamodb:*", + "Resource": resources, + "Condition": {"ForAllValues:StringEquals": {"dynamodb:LeadingKeys": values}}, + }], + } + + def test_query_shapes_authorize_the_partition_the_engine_reads( + self, table, num_table, mh_table): + tables = {"main": table, "num": num_table, "mh": mh_table} + owner = self._user_client([("full", ALLOW_ALL)]) + fenced = self._user_client([("lk", self._fence_policy( + [OWN, G_OWN, "10"], list(tables.values())))]) + other = self._user_client([("lk", self._fence_policy( + ["nobody"], list(tables.values())))]) + + def call(client, table_key, kwargs): + try: + return client.query(TableName=tables[table_key], **kwargs), None + except ClientError as exc: + return None, exc.response["Error"]["Code"] + + # Secondary indexes are eventually consistent: wait until the owner + # sees the seeded rows through each index before comparing results. + for label, table_key, kwargs, _attr, _value in self.QUERY_ACCEPTED: + if "IndexName" not in kwargs: + continue + deadline = time.monotonic() + 30 + while True: + got, err = call(owner, table_key, kwargs) + if err is not None or got["Count"] > 0 or time.monotonic() > deadline: + break + time.sleep(0.25) + + failures = [] + for label, table_key, kwargs, attr, value in self.QUERY_ACCEPTED: + got, err = call(owner, table_key, kwargs) + if err is not None: + failures.append(f"{label}: owner got {err}") + continue + parts = {next(iter(it[attr].values())) for it in got["Items"]} + if parts != {value}: + failures.append(f"{label}: engine read partitions {parts}, expected {{{value!r}}}") + got_f, err_f = call(fenced, table_key, kwargs) + if err_f is not None or got_f["Count"] != got["Count"]: + failures.append(f"{label}: fenced principal got {err_f or got_f['Count']}, " + f"owner got {got['Count']} rows") + _, err_o = call(other, table_key, kwargs) + if err_o != "AccessDeniedException": + failures.append(f"{label}: other principal got {err_o or 'rows'}, expected denial") + + for label, table_key, kwargs in self.QUERY_REJECTED: + for who, client in (("owner", owner), ("fenced", fenced)): + _, err = call(client, table_key, kwargs) + if err != "ValidationException": + failures.append(f"{label}: {who} got {err or 'rows'}, expected ValidationException") + _, err_o = call(other, table_key, kwargs) + if label in self.QUERY_REJECTED_OTHER_CODE: + if err_o not in ("AccessDeniedException", "ValidationException"): + failures.append(f"{label}: other principal got {err_o or 'rows'}") + elif err_o != "ValidationException": + failures.append(f"{label}: other principal got {err_o or 'rows'}, " + "expected ValidationException") + + assert not failures, "\n".join(failures) + + # -- A table read while it is CREATING ---------------------------------- + # + # A request sent while a table is CREATING finds no table. Before the fix, + # authorization read the table key info through a cache that keeps a + # missing table for its negative TTL (5 seconds by default), while the + # batch and transaction handlers read storage. One such request during + # CREATING left authorization seeing no table, and so no LeadingKeys, for + # the rest of the TTL after the table became ACTIVE, while the engine saw + # the table and ran the request. + # + # Measured against live DynamoDB on 2026-10-21: after DescribeTable + # reported ACTIVE, the four operations returned ResourceNotFoundException + # for 2 to 4 seconds on both partitions, then were denied on the other + # partition and served on the principal's own. The other partition was + # never served. The test accepts ResourceNotFoundException for up to 15 + # seconds after ACTIVE, and fails if the other partition is served at any + # point. + + def test_batch_and_transaction_fenced_once_a_created_table_is_active(self): + owner = self._user_client([("full", ALLOW_ALL)]) + table = None + fenced = None + for _ in range(3): + name = f"lk-c-{uuid.uuid4().hex[:8]}" + fenced = self._user_client([("lk", self._fence_policy([OWN], [name]))]) + owner.create_table( + TableName=name, + AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}, + {"AttributeName": "sk", "AttributeType": "S"}], + KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}, + {"AttributeName": "sk", "KeyType": "RANGE"}], + BillingMode="PAY_PER_REQUEST", + ) + hit = [] + for send in ( + lambda: fenced.batch_write_item(RequestItems={name: [ + {"PutRequest": {"Item": self._key(OWN)}}]}), + lambda: fenced.get_item(TableName=name, Key=self._key(OWN)), + ): + try: + send() + except ClientError as exc: + hit.append(exc.response["Error"]["Code"] == "ResourceNotFoundException") + wait_for_active(owner, name) + if hit and all(hit) and len(hit) == 2: + table = name + break + owner.delete_table(TableName=name) + if table is None: + pytest.skip("the table was ACTIVE before a request could reach it while CREATING") + + try: + other_key = self._key(OTHER) + denied = { + "BatchWriteItem": lambda: fenced.batch_write_item(RequestItems={table: [ + {"PutRequest": {"Item": other_key}}]}), + "BatchGetItem": lambda: fenced.batch_get_item(RequestItems={table: { + "Keys": [other_key]}}), + "TransactGetItems": lambda: fenced.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": other_key}}]), + "TransactWriteItems": lambda: fenced.transact_write_items(TransactItems=[ + {"Put": {"TableName": table, "Item": other_key}}]), + } + failures = [] + + def settle(send): + # The response once the table is visible on the data plane: + # "ok", or the error code other than ResourceNotFoundException. + deadline = time.monotonic() + 15 + while True: + try: + send() + return "ok" + except ClientError as exc: + code = exc.response["Error"]["Code"] + if code != "ResourceNotFoundException" or time.monotonic() > deadline: + return code + time.sleep(0.25) + + for op, send in denied.items(): + got = settle(send) + if got == "ok": + failures.append(f"{op} on the other partition was served") + elif got != "AccessDeniedException": + failures.append(f"{op} on the other partition got {got}") + # The fenced principal's own partition is served on the batch and + # transaction paths once the table is visible. + own_key = self._key(OWN) + for op, send in { + "BatchWriteItem": lambda: fenced.batch_write_item(RequestItems={table: [ + {"PutRequest": {"Item": own_key}}]}), + "BatchGetItem": lambda: fenced.batch_get_item(RequestItems={table: { + "Keys": [own_key]}}), + "TransactGetItems": lambda: fenced.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": own_key}}]), + }.items(): + got = settle(send) + if got != "ok": + failures.append(f"{op} on the own partition got {got}") + # The single-item operations never return the other partition. The + # owner seeds it through BatchWriteItem, which reads the catalog. + owner.batch_write_item(RequestItems={table: [ + {"PutRequest": {"Item": {**other_key, "data": {"S": "other"}}}}]}) + for op, send in { + "GetItem": lambda: fenced.get_item(TableName=table, Key=other_key), + "Query": lambda: fenced.query( + TableName=table, KeyConditionExpression="pk = :v", + ExpressionAttributeValues={":v": {"S": OTHER}}), + }.items(): + try: + resp = send() + if resp.get("Item") or resp.get("Items"): + failures.append(f"{op} returned the other partition") + except ClientError: + pass + assert not failures, "\n".join(failures) + finally: + try: + owner.delete_table(TableName=table) + wait_for_deleted(owner, table) + except Exception: + pass + + # -- Batch and transaction requests outside the count limits ------------ + # + # Measured against live DynamoDB on 2026-10-21: a TransactWriteItems or + # TransactGetItems with 101 items, a BatchWriteItem with 26 requests on one + # table or across two, and a BatchGetItem with 101 keys on one table or + # across two each returned ValidationException to a principal with no + # policy, to a principal whose LeadingKeys fence excluded every key, to one + # whose fence included them and to the owner. So did an empty TransactItems + # or RequestItems and a table with no keys or requests. A BatchGetItem with + # 101 tables and a BatchWriteItem with 26 tables, one item each, returned + # ValidationException to a principal with no policy and to the owner, and + # the same BatchGetItem with 100 tables returned AccessDeniedException to + # the principal with no policy. The count checks come before authorization. + + def test_oversized_requests_fail_validation_before_authorization(self, table, num_table): + no_policy = self._user_client([], parameter_validation=False) + fenced_out = self._user_client([("lk", self._fence_policy( + ["nobody"], [table, num_table]))], parameter_validation=False) + other_key = self._key(OTHER) + cases = { + "TransactWriteItems": lambda c: c.transact_write_items(TransactItems=[ + {"ConditionCheck": {"TableName": table, + "Key": {"pk": {"S": OTHER}, "sk": {"S": f"o{i}"}}, + "ConditionExpression": "attribute_exists(pk)"}} + for i in range(101)]), + "TransactGetItems": lambda c: c.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, + "Key": {"pk": {"S": OTHER}, "sk": {"S": f"o{i}"}}}} + for i in range(101)]), + "BatchWriteItem one table": lambda c: c.batch_write_item(RequestItems={table: [ + {"PutRequest": {"Item": {"pk": {"S": OTHER}, "sk": {"S": f"o{i}"}}}} + for i in range(26)]}), + "BatchWriteItem two tables": lambda c: c.batch_write_item(RequestItems={ + table: [{"PutRequest": {"Item": {"pk": {"S": OTHER}, "sk": {"S": f"o{i}"}}}} + for i in range(13)], + num_table: [{"PutRequest": {"Item": {"pk": {"N": str(1000 + i)}}}} + for i in range(13)]}), + "BatchGetItem one table": lambda c: c.batch_get_item(RequestItems={table: { + "Keys": [{"pk": {"S": OTHER}, "sk": {"S": f"o{i}"}} for i in range(101)]}}), + "BatchGetItem two tables": lambda c: c.batch_get_item(RequestItems={ + table: {"Keys": [{"pk": {"S": OTHER}, "sk": {"S": f"o{i}"}} + for i in range(60)]}, + num_table: {"Keys": [{"pk": {"N": str(1000 + i)}} for i in range(41)]}}), + "BatchGetItem 101 tables": lambda c: c.batch_get_item(RequestItems={ + f"{table}-{i}": {"Keys": [other_key]} for i in range(101)}), + "BatchWriteItem 26 tables": lambda c: c.batch_write_item(RequestItems={ + f"{table}-{i}": [{"PutRequest": {"Item": other_key}}] for i in range(26)}), + "TransactWriteItems empty": lambda c: c.transact_write_items(TransactItems=[]), + "TransactGetItems empty": lambda c: c.transact_get_items(TransactItems=[]), + "BatchWriteItem empty": lambda c: c.batch_write_item(RequestItems={}), + "BatchGetItem empty": lambda c: c.batch_get_item(RequestItems={}), + "BatchWriteItem table with no requests": lambda c: c.batch_write_item( + RequestItems={table: []}), + "BatchGetItem table with no keys": lambda c: c.batch_get_item( + RequestItems={table: {"Keys": []}}), + } + failures = [] + for label, send in cases.items(): + for who, client in (("no policy", no_policy), ("fenced out", fenced_out)): + try: + send(client) + failures.append(f"{label} as {who} was served") + except ClientError as exc: + if exc.response["Error"]["Code"] != "ValidationException": + failures.append(f"{label} as {who} got " + f"{exc.response['Error']['Code']}") + # Control: a request within the limits is still refused to both. + for who, client in (("no policy", no_policy), ("fenced out", fenced_out)): + with pytest.raises(ClientError) as e: + client.transact_get_items(TransactItems=[ + {"Get": {"TableName": table, "Key": other_key}}]) + assert _denied(e), who + assert not failures, "\n".join(failures)