From a6cba36fe9129fd5995b6769435cd60493c88bb0 Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Wed, 4 Nov 2026 21:37:48 +0000 Subject: [PATCH 1/4] feat(core): checks for the table quota and provisioned-capacity maxima Pure checks for max_tables_per_account, per_table_max_rcu/wcu and per_account_max_rcu/wcu, with the LimitExceededException sentences the service returns for a provisioned value over a maximum. --- crates/core/src/validation/mod.rs | 1 + .../core/src/validation/provisioned_limits.rs | 349 ++++++++++++++++ .../validation/provisioned_limits_tests.rs | 381 ++++++++++++++++++ 3 files changed, 731 insertions(+) create mode 100644 crates/core/src/validation/provisioned_limits.rs create mode 100644 crates/core/src/validation/provisioned_limits_tests.rs diff --git a/crates/core/src/validation/mod.rs b/crates/core/src/validation/mod.rs index a849f2c95..12f4f4dbe 100755 --- a/crates/core/src/validation/mod.rs +++ b/crates/core/src/validation/mod.rs @@ -1,6 +1,7 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 pub mod number; +pub mod provisioned_limits; pub mod vector_item; pub use vector_item::{ diff --git a/crates/core/src/validation/provisioned_limits.rs b/crates/core/src/validation/provisioned_limits.rs new file mode 100644 index 000000000..d625107bd --- /dev/null +++ b/crates/core/src/validation/provisioned_limits.rs @@ -0,0 +1,349 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The account limits `CreateTable` and `UpdateTable` enforce: the table quota +//! (`max_tables_per_account`) and the provisioned-capacity maxima +//! (`per_table_max_rcu`, `per_table_max_wcu`, `per_account_max_rcu`, +//! `per_account_max_wcu`). +//! +//! The checks are pure: the engine reads the account's tables and passes in +//! what it found. Every refusal is `LimitExceededException`. The capacity +//! sentences are the ones Amazon `DynamoDB` returns, measured on 2026-10-31; +//! the table-quota sentence was not measured. + +use crate::error::DynamoDbError; +use crate::limits::LimitsConfig; +use crate::types::{ + BillingMode, CreateTableInput, ProvisionedThroughput, TableDescription, UpdateTableInput, +}; + +/// The closing sentence of every provisioned-capacity refusal. +const LIMITS_GUIDE: &str = "Refer to the Amazon DynamoDB Developer Guide for current limits \ + and how to request higher limits."; + +/// Read and write capacity units held by a table and its global secondary +/// indexes, or by every table in an account. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct CapacityUnits { + pub read: i64, + pub write: i64, +} + +impl CapacityUnits { + /// `self` plus `other`, saturating at `i64::MAX`. + #[must_use] + pub fn saturating_add(self, other: Self) -> Self { + Self { + read: self.read.saturating_add(other.read), + write: self.write.saturating_add(other.write), + } + } + + fn of(throughput: &ProvisionedThroughput) -> Self { + Self { + read: throughput.read_capacity_units, + write: throughput.write_capacity_units, + } + } +} + +/// The two units a refusal can name, in the order the service checks them. +#[derive(Debug, Clone, Copy)] +enum Unit { + Read, + Write, +} + +impl Unit { + const BOTH: [Self; 2] = [Self::Read, Self::Write]; + + fn name(self) -> &'static str { + match self { + Self::Read => "ReadCapacityUnits", + Self::Write => "WriteCapacityUnits", + } + } + + fn of(self, units: CapacityUnits) -> i64 { + match self { + Self::Read => units.read, + Self::Write => units.write, + } + } + + fn table_max(self, limits: &LimitsConfig) -> i64 { + let max = match self { + Self::Read => limits.per_table_max_rcu, + Self::Write => limits.per_table_max_wcu, + }; + i64::try_from(max).unwrap_or(i64::MAX) + } + + fn account_max(self, limits: &LimitsConfig) -> i64 { + let max = match self { + Self::Read => limits.per_account_max_rcu, + Self::Write => limits.per_account_max_wcu, + }; + i64::try_from(max).unwrap_or(i64::MAX) + } +} + +/// What a `CreateTable` request provisions: the table's units plus those of +/// every global secondary index. `None` for an on-demand table. +#[must_use] +pub fn create_table_capacity(input: &CreateTableInput) -> Option { + if input.billing_mode.unwrap_or(BillingMode::Provisioned) == BillingMode::PayPerRequest { + return None; + } + let table = CapacityUnits::of(input.provisioned_throughput.as_ref()?); + Some( + input + .global_secondary_indexes + .iter() + .flatten() + .filter_map(|gsi| gsi.provisioned_throughput.as_ref()) + .map(CapacityUnits::of) + .fold(table, CapacityUnits::saturating_add), + ) +} + +/// What a table and its global secondary indexes hold, as `DescribeTable` +/// reports them. An on-demand table holds none. +#[must_use] +pub fn table_capacity(desc: &TableDescription) -> CapacityUnits { + if is_on_demand(desc) { + return CapacityUnits::default(); + } + let table = CapacityUnits { + read: desc.provisioned_throughput.read_capacity_units, + write: desc.provisioned_throughput.write_capacity_units, + }; + desc.global_secondary_indexes + .iter() + .flatten() + .filter_map(|gsi| gsi.provisioned_throughput.as_ref()) + .map(|t| CapacityUnits { + read: t.read_capacity_units, + write: t.write_capacity_units, + }) + .fold(table, CapacityUnits::saturating_add) +} + +fn is_on_demand(desc: &TableDescription) -> bool { + desc.billing_mode_summary + .as_ref() + .is_some_and(|b| b.billing_mode == BillingMode::PayPerRequest) +} + +/// Whether `input` sets a provisioned throughput, on the table or on a global +/// secondary index it creates. Only such a request can raise the capacity a +/// table holds. +#[must_use] +pub fn changes_provisioned_capacity(input: &UpdateTableInput) -> bool { + input.provisioned_throughput.is_some() + || input + .global_secondary_index_updates + .iter() + .flatten() + .filter_map(|u| u.create.as_ref()) + .any(|c| c.provisioned_throughput.is_some()) +} + +/// What the table `existing` and its global secondary indexes would hold once +/// `input` applies: its throughput or the one `input` sets, the indexes it +/// keeps and the indexes `input` creates. `None` when `input` sets no +/// provisioned throughput, or when the table is on-demand afterwards. +#[must_use] +pub fn update_table_capacity( + existing: &TableDescription, + input: &UpdateTableInput, +) -> Option { + if !changes_provisioned_capacity(input) { + return None; + } + let on_demand_after = match input.billing_mode { + Some(mode) => mode == BillingMode::PayPerRequest, + None => is_on_demand(existing), + }; + if on_demand_after { + return None; + } + let updates = input + .global_secondary_index_updates + .as_deref() + .unwrap_or(&[]); + let deleted = |name: &str| { + updates + .iter() + .filter_map(|u| u.delete.as_ref()) + .any(|d| d.index_name == name) + }; + let table = input.provisioned_throughput.as_ref().map_or( + CapacityUnits { + read: existing.provisioned_throughput.read_capacity_units, + write: existing.provisioned_throughput.write_capacity_units, + }, + CapacityUnits::of, + ); + let kept = existing + .global_secondary_indexes + .iter() + .flatten() + .filter(|gsi| !deleted(&gsi.index_name)) + .filter_map(|gsi| gsi.provisioned_throughput.as_ref()) + .map(|t| CapacityUnits { + read: t.read_capacity_units, + write: t.write_capacity_units, + }); + let created = updates + .iter() + .filter_map(|u| u.create.as_ref()) + .filter_map(|c| c.provisioned_throughput.as_ref()) + .map(CapacityUnits::of); + Some( + kept.chain(created) + .fold(table, CapacityUnits::saturating_add), + ) +} + +/// Refuse a `CreateTable` that would provision more than the account or one +/// table may hold. +/// +/// `requested` is what the request provisions ([`create_table_capacity`]) and +/// `reserved` what the account's tables hold already. The account maximum is +/// checked first, read units before write units, as measured. The per-table +/// maximum bounds the table and its global secondary indexes together; its +/// sentence on this path was not measured, so it is the sentence `UpdateTable` +/// returns. +/// +/// # Errors +/// +/// `LimitExceededException` naming the first unit over a maximum. +pub fn check_create_table_capacity( + requested: CapacityUnits, + reserved: CapacityUnits, + limits: &LimitsConfig, + region: &str, +) -> Result<(), DynamoDbError> { + for unit in Unit::BOTH { + let limit = unit.account_max(limits); + let (reserved, requested) = (unit.of(reserved), unit.of(requested)); + if reserved.saturating_add(requested) > limit { + return Err(account_refusal( + unit, region, reserved, limit, None, requested, + )); + } + } + for unit in Unit::BOTH { + let max = unit.table_max(limits); + if unit.of(requested) > max { + return Err(per_table_refusal(unit, unit.of(requested), max, region)); + } + } + Ok(()) +} + +/// Refuse an `UpdateTable` that would raise a table or the account above its +/// maximum. +/// +/// `table_request` is the table's own `ProvisionedThroughput` from the request, +/// checked against the per-table maximum first, as measured. Then the account: +/// `reserved` is what the account holds now, including `current`, what this +/// table and its indexes hold now, which `after` replaces. A unit the request +/// does not raise is not refused, so a table can still be lowered after an +/// operator lowers a maximum below what it holds. +/// +/// # Errors +/// +/// `LimitExceededException` naming the first unit over a maximum. +pub fn check_update_table_capacity( + table_request: Option<&ProvisionedThroughput>, + existing: &TableDescription, + reserved: CapacityUnits, + after: CapacityUnits, + limits: &LimitsConfig, + region: &str, +) -> Result<(), DynamoDbError> { + if let Some(request) = table_request { + let requested = CapacityUnits::of(request); + let held = CapacityUnits { + read: existing.provisioned_throughput.read_capacity_units, + write: existing.provisioned_throughput.write_capacity_units, + }; + for unit in Unit::BOTH { + let max = unit.table_max(limits); + if unit.of(requested) > max && unit.of(requested) > unit.of(held) { + return Err(per_table_refusal(unit, unit.of(requested), max, region)); + } + } + } + let current = table_capacity(existing); + for unit in Unit::BOTH { + let limit = unit.account_max(limits); + let (reserved, current, after) = (unit.of(reserved), unit.of(current), unit.of(after)); + if after > current && reserved.saturating_sub(current).saturating_add(after) > limit { + return Err(account_refusal( + unit, + region, + reserved, + limit, + Some(current), + after, + )); + } + } + Ok(()) +} + +/// Refuse a `CreateTable` in an account that holds `max_tables_per_account` +/// tables already. +/// +/// The service's sentence for this refusal was not measured. This one states +/// the quota the way the `LimitExceededException` reference documents it. +/// +/// # Errors +/// +/// `LimitExceededException` when `existing` is at or above the quota. +pub fn check_table_count(existing: usize, limits: &LimitsConfig) -> Result<(), DynamoDbError> { + if existing >= limits.max_tables_per_account { + return Err(DynamoDbError::LimitExceededException(format!( + "There is a soft account quota of {} tables.", + limits.max_tables_per_account + ))); + } + Ok(()) +} + +fn per_table_refusal(unit: Unit, requested: i64, max: i64, region: &str) -> DynamoDbError { + let unit = unit.name(); + DynamoDbError::LimitExceededException(format!( + "The requested {unit}, {requested}, is above the per table maximum for the account in \ + {region}. Per table maximum: {max}. {LIMITS_GUIDE}" + )) +} + +/// The account sentence. `table_now`, set on `UpdateTable`, is what the table +/// and its indexes hold before the request, and `requested` what they would +/// hold after it; on `CreateTable` `requested` is the new table's units. +fn account_refusal( + unit: Unit, + region: &str, + reserved: i64, + limit: i64, + table_now: Option, + requested: i64, +) -> DynamoDbError { + let unit = unit.name(); + let table = table_now.map_or_else(String::new, |now| { + format!(" Current {unit} provisioned for the table and indexes: {now}.") + }); + DynamoDbError::LimitExceededException(format!( + "This request would have caused the {unit} limit to be exceeded for the account in \ + {region}. Current {unit} reserved by the account: {reserved}. Limit: {limit}.{table} \ + Requested: {requested}. {LIMITS_GUIDE}" + )) +} + +#[cfg(test)] +#[path = "provisioned_limits_tests.rs"] +mod tests; diff --git a/crates/core/src/validation/provisioned_limits_tests.rs b/crates/core/src/validation/provisioned_limits_tests.rs new file mode 100644 index 000000000..be6dad44d --- /dev/null +++ b/crates/core/src/validation/provisioned_limits_tests.rs @@ -0,0 +1,381 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The measured sentences are the ones Amazon `DynamoDB` returned on +//! 2026-10-31 in us-east-1, on an account whose per-table and account maxima +//! were both 150,000 read and 200,000 write units, holding 61 and 46. + +use super::*; +use crate::types::{ + BillingModeSummary, GsiDescription, Projection, ProjectionType, + ProvisionedThroughputDescription, +}; +use serde_json::json; + +const REGION: &str = "us-east-1"; + +/// The limits of the measured account. +fn measured_limits() -> LimitsConfig { + LimitsConfig { + per_table_max_rcu: 150_000, + per_table_max_wcu: 200_000, + per_account_max_rcu: 150_000, + per_account_max_wcu: 200_000, + ..LimitsConfig::default() + } +} + +/// What the measured account held. +const MEASURED_RESERVED: CapacityUnits = CapacityUnits { + read: 61, + write: 46, +}; + +fn units(read: i64, write: i64) -> CapacityUnits { + CapacityUnits { read, write } +} + +fn throughput(read: i64, write: i64) -> ProvisionedThroughput { + ProvisionedThroughput { + read_capacity_units: read, + write_capacity_units: write, + } +} + +fn gsi(name: &str, read: i64, write: i64) -> GsiDescription { + GsiDescription { + index_name: name.to_owned(), + key_schema: Vec::new(), + projection: Projection { + projection_type: ProjectionType::All, + non_key_attributes: None, + }, + index_status: "ACTIVE".to_owned(), + provisioned_throughput: Some(ProvisionedThroughputDescription { + read_capacity_units: read, + write_capacity_units: write, + ..ProvisionedThroughputDescription::default() + }), + index_size_bytes: 0, + item_count: 0, + index_arn: String::new(), + } +} + +/// A provisioned table holding `read` and `write` units, with `gsis`. +fn table(read: i64, write: i64, gsis: Vec) -> TableDescription { + TableDescription { + provisioned_throughput: ProvisionedThroughputDescription { + read_capacity_units: read, + write_capacity_units: write, + ..ProvisionedThroughputDescription::default() + }, + global_secondary_indexes: (!gsis.is_empty()).then_some(gsis), + ..TableDescription::default() + } +} + +fn update(body: &serde_json::Value) -> UpdateTableInput { + serde_json::from_value(body.clone()).expect("UpdateTable input") +} + +fn create(body: &serde_json::Value) -> CreateTableInput { + serde_json::from_value(body.clone()).expect("CreateTable input") +} + +fn refusal(result: Result<(), DynamoDbError>) -> String { + match result { + Err(DynamoDbError::LimitExceededException(msg)) => msg, + other => panic!("expected LimitExceededException, got {other:?}"), + } +} + +#[test] +fn update_table_over_the_per_table_maximum_is_the_measured_sentence() { + let existing = table(5, 5, Vec::new()); + let check = |read, write| { + let request = throughput(read, write); + let after = units(read, write); + check_update_table_capacity( + Some(&request), + &existing, + MEASURED_RESERVED, + after, + &measured_limits(), + REGION, + ) + }; + assert_eq!( + refusal(check(150_001, 5)), + "The requested ReadCapacityUnits, 150001, is above the per table maximum for the \ + account in us-east-1. Per table maximum: 150000. Refer to the Amazon DynamoDB \ + Developer Guide for current limits and how to request higher limits." + ); + assert_eq!( + refusal(check(5, 200_001)), + "The requested WriteCapacityUnits, 200001, is above the per table maximum for the \ + account in us-east-1. Per table maximum: 200000. Refer to the Amazon DynamoDB \ + Developer Guide for current limits and how to request higher limits." + ); +} + +#[test] +fn create_table_over_the_maximum_is_the_measured_account_sentence() { + let check = |read, write| { + check_create_table_capacity( + units(read, write), + MEASURED_RESERVED, + &measured_limits(), + REGION, + ) + }; + let read_sentence = "This request would have caused the ReadCapacityUnits limit to be \ + exceeded for the account in us-east-1. Current ReadCapacityUnits \ + reserved by the account: 61. Limit: 150000. Requested: 150001. Refer \ + to the Amazon DynamoDB Developer Guide for current limits and how to \ + request higher limits."; + assert_eq!(refusal(check(150_001, 1)), read_sentence); + assert_eq!( + refusal(check(1, 200_001)), + "This request would have caused the WriteCapacityUnits limit to be exceeded for the \ + account in us-east-1. Current WriteCapacityUnits reserved by the account: 46. Limit: \ + 200000. Requested: 200001. Refer to the Amazon DynamoDB Developer Guide for current \ + limits and how to request higher limits." + ); + // Both over: the read sentence. + assert_eq!(refusal(check(150_001, 200_001)), read_sentence); +} + +#[test] +fn a_gsi_create_over_the_maximum_is_the_measured_account_sentence() { + let existing = table(1, 1, Vec::new()); + let check = |read, write| { + let input = update(&json!({ + "TableName": "t", + "AttributeDefinitions": [{ "AttributeName": "g", "AttributeType": "S" }], + "GlobalSecondaryIndexUpdates": [{ "Create": { + "IndexName": "gix", + "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], + "Projection": { "ProjectionType": "ALL" }, + "ProvisionedThroughput": { "ReadCapacityUnits": read, "WriteCapacityUnits": write }, + }}], + })); + let after = update_table_capacity(&existing, &input).expect("a provisioned GSI"); + check_update_table_capacity( + input.provisioned_throughput.as_ref(), + &existing, + MEASURED_RESERVED, + after, + &measured_limits(), + REGION, + ) + }; + assert_eq!( + refusal(check(150_001, 1)), + "This request would have caused the ReadCapacityUnits limit to be exceeded for the \ + account in us-east-1. Current ReadCapacityUnits reserved by the account: 61. Limit: \ + 150000. Current ReadCapacityUnits provisioned for the table and indexes: 1. \ + Requested: 150002. Refer to the Amazon DynamoDB Developer Guide for current limits \ + and how to request higher limits." + ); + assert_eq!( + refusal(check(1, 200_001)), + "This request would have caused the WriteCapacityUnits limit to be exceeded for the \ + account in us-east-1. Current WriteCapacityUnits reserved by the account: 46. Limit: \ + 200000. Current WriteCapacityUnits provisioned for the table and indexes: 1. \ + Requested: 200002. Refer to the Amazon DynamoDB Developer Guide for current limits \ + and how to request higher limits." + ); +} + +#[test] +fn create_table_at_each_maximum_is_accepted_and_one_above_is_refused() { + let limits = LimitsConfig::default(); + // At the per-table maximum, with the account at its maximum too. + check_create_table_capacity( + units(40_000, 40_000), + units(40_000, 40_000), + &limits, + REGION, + ) + .expect("at both maxima"); + // One unit over the per-table maximum, the account well under its own. + assert_eq!( + refusal(check_create_table_capacity( + units(40_001, 1), + CapacityUnits::default(), + &limits, + REGION + )), + "The requested ReadCapacityUnits, 40001, is above the per table maximum for the \ + account in us-east-1. Per table maximum: 40000. Refer to the Amazon DynamoDB \ + Developer Guide for current limits and how to request higher limits." + ); + assert!( + refusal(check_create_table_capacity( + units(1, 40_001), + CapacityUnits::default(), + &limits, + REGION + )) + .starts_with("The requested WriteCapacityUnits, 40001, is above the per table maximum") + ); + // One unit over the account maximum, the table under its own. + assert!( + refusal(check_create_table_capacity( + units(40_000, 1), + units(40_001, 0), + &limits, + REGION + )) + .starts_with( + "This request would have caused the ReadCapacityUnits limit to be exceeded for the \ + account in us-east-1. Current ReadCapacityUnits reserved by the account: 40001. \ + Limit: 80000. Requested: 40000." + ) + ); +} + +#[test] +fn update_table_at_each_maximum_is_accepted_and_one_above_is_refused() { + let limits = LimitsConfig::default(); + let existing = table(10, 10, Vec::new()); + let check = |read, write, reserved| { + let request = throughput(read, write); + check_update_table_capacity( + Some(&request), + &existing, + reserved, + units(read, write), + &limits, + REGION, + ) + }; + check(40_000, 40_000, units(10, 10)).expect("at the per-table maximum"); + assert!( + refusal(check(40_001, 10, units(10, 10))) + .starts_with("The requested ReadCapacityUnits, 40001, is above the per table maximum") + ); + // The account holds 70,000 including this table's 10: raising it to 10,010 + // fills the account to 80,000, and 10,011 takes it one over. + check(10_010, 10, units(70_000, 10)).expect("at the account maximum"); + assert!(refusal(check(10_011, 10, units(70_000, 10))).starts_with( + "This request would have caused the ReadCapacityUnits limit to be exceeded for the \ + account in us-east-1. Current ReadCapacityUnits reserved by the account: 70000. \ + Limit: 80000. Current ReadCapacityUnits provisioned for the table and indexes: 10. \ + Requested: 10011." + )); +} + +#[test] +fn a_table_above_a_lowered_maximum_can_still_be_lowered() { + let limits = LimitsConfig { + per_table_max_rcu: 100, + per_table_max_wcu: 100, + per_account_max_rcu: 200, + per_account_max_wcu: 200, + ..LimitsConfig::default() + }; + let existing = table(500, 500, Vec::new()); + let request = throughput(300, 300); + check_update_table_capacity( + Some(&request), + &existing, + units(1_000, 1_000), + units(300, 300), + &limits, + REGION, + ) + .expect("a decrease is not refused"); +} + +#[test] +fn the_table_quota_admits_the_last_table_and_refuses_the_next() { + let limits = LimitsConfig { + max_tables_per_account: 3, + ..LimitsConfig::default() + }; + check_table_count(2, &limits).expect("the third table fills the quota"); + assert_eq!( + refusal(check_table_count(3, &limits)), + "There is a soft account quota of 3 tables." + ); +} + +#[test] +fn create_table_capacity_sums_the_table_and_its_gsis() { + let gsi = |name: &str, pt: serde_json::Value| { + let mut index = json!({ + "IndexName": name, + "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], + "Projection": { "ProjectionType": "ALL" }, + }); + if !pt.is_null() { + index["ProvisionedThroughput"] = pt; + } + index + }; + let provisioned = create(&json!({ + "TableName": "t", + "KeySchema": [{ "AttributeName": "pk", "KeyType": "HASH" }], + "AttributeDefinitions": [{ "AttributeName": "pk", "AttributeType": "S" }], + "ProvisionedThroughput": { "ReadCapacityUnits": 5, "WriteCapacityUnits": 7 }, + "GlobalSecondaryIndexes": [ + gsi("a", json!({ "ReadCapacityUnits": 10, "WriteCapacityUnits": 20 })), + gsi("b", serde_json::Value::Null), + ], + })); + assert_eq!(create_table_capacity(&provisioned), Some(units(15, 27))); + let on_demand = create(&json!({ + "TableName": "t", + "KeySchema": [{ "AttributeName": "pk", "KeyType": "HASH" }], + "AttributeDefinitions": [{ "AttributeName": "pk", "AttributeType": "S" }], + "BillingMode": "PAY_PER_REQUEST", + })); + assert_eq!(create_table_capacity(&on_demand), None); +} + +#[test] +fn table_capacity_counts_gsis_and_nothing_for_on_demand() { + let provisioned = table(5, 6, vec![gsi("a", 10, 20), gsi("b", 1, 2)]); + assert_eq!(table_capacity(&provisioned), units(16, 28)); + let on_demand = TableDescription { + billing_mode_summary: Some(BillingModeSummary { + billing_mode: BillingMode::PayPerRequest, + last_update_to_pay_per_request_date_time: None, + }), + ..table(5, 6, vec![gsi("a", 10, 20)]) + }; + assert_eq!(table_capacity(&on_demand), CapacityUnits::default()); +} + +#[test] +fn update_table_capacity_replaces_the_table_and_follows_index_changes() { + let existing = table(5, 6, vec![gsi("kept", 10, 20), gsi("dropped", 100, 200)]); + let created = json!({ "Create": { + "IndexName": "new", + "KeySchema": [{ "AttributeName": "g", "KeyType": "HASH" }], + "Projection": { "ProjectionType": "ALL" }, + "ProvisionedThroughput": { "ReadCapacityUnits": 3, "WriteCapacityUnits": 4 }, + }}); + let input = update(&json!({ + "TableName": "t", + "ProvisionedThroughput": { "ReadCapacityUnits": 50, "WriteCapacityUnits": 60 }, + "GlobalSecondaryIndexUpdates": [created, { "Delete": { "IndexName": "dropped" } }], + })); + assert_eq!( + update_table_capacity(&existing, &input), + Some(units(50 + 10 + 3, 60 + 20 + 4)) + ); + // Nothing that sets a throughput: no capacity to check. + let tags_only = update(&json!({ "TableName": "t", "DeletionProtectionEnabled": true })); + assert!(!changes_provisioned_capacity(&tags_only)); + assert_eq!(update_table_capacity(&existing, &tags_only), None); + // Switching to on-demand: nothing provisioned afterwards. + let to_on_demand = update(&json!({ + "TableName": "t", + "BillingMode": "PAY_PER_REQUEST", + "ProvisionedThroughput": { "ReadCapacityUnits": 50, "WriteCapacityUnits": 60 }, + })); + assert_eq!(update_table_capacity(&existing, &to_on_demand), None); +} From cd5b1ebc98dc58f9605ab7e73bef7268b3a59a8b Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Wed, 4 Nov 2026 21:41:43 +0000 Subject: [PATCH 2/4] fix(engine): enforce the table quota and provisioned-capacity maxima CreateTable counts the account's tables and refuses one over max_tables_per_account, and CreateTable and UpdateTable refuse a provisioned throughput over the per-table or account maximum with LimitExceededException. A name the account already holds is still refused as taken first. --- crates/engine/src/create_table.rs | 4 + crates/engine/src/lib.rs | 1 + crates/engine/src/table_limits.rs | 126 ++++++++++++++++++++++++++++++ crates/engine/src/update_table.rs | 15 ++++ 4 files changed, 146 insertions(+) create mode 100644 crates/engine/src/table_limits.rs diff --git a/crates/engine/src/create_table.rs b/crates/engine/src/create_table.rs index 7b827cd3a..5a11d0dcf 100755 --- a/crates/engine/src/create_table.rs +++ b/crates/engine/src/create_table.rs @@ -65,6 +65,10 @@ pub async fn handle_create_table( ctx.storage.as_vector_search(), )?; + // After the definition checks, as the service orders it: the table quota + // and the provisioned-capacity maxima. + crate::table_limits::enforce_create_table(ctx, &input).await?; + let table_name = input.table_name.clone(); // Kept for the post-condition check below, since `input` is moved into the // backend call. diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index b8e372742..40465d281 100755 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -34,6 +34,7 @@ mod search_vectors; mod start_key; pub mod stream_capture; mod streams; +mod table_limits; mod tagging; mod transact_get_items; mod transact_write_helpers; diff --git a/crates/engine/src/table_limits.rs b/crates/engine/src/table_limits.rs new file mode 100644 index 000000000..8bbe15c18 --- /dev/null +++ b/crates/engine/src/table_limits.rs @@ -0,0 +1,126 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The account limits `CreateTable` and `UpdateTable` enforce before the +//! backend is called: the table quota and the provisioned-capacity maxima. +//! The checks and their sentences are in +//! [`extenddb_core::validation::provisioned_limits`]; this module reads what +//! the account holds. + +use extenddb_core::error::DynamoDbError; +use extenddb_core::types::{ + CreateTableInput, DescribeTableInput, ListTablesInput, TableDescription, UpdateTableInput, +}; +use extenddb_core::validation::provisioned_limits::{self as limits, CapacityUnits}; +use extenddb_storage::error::StorageError; + +use crate::OperationContext; +use crate::create_table::storage_err_to_dynamo; + +/// The page size the account's tables are listed in. +const LIST_PAGE: i32 = 100; + +/// Refuse a `CreateTable` the account's limits do not admit. +/// +/// The service reports a taken name ahead of any limit (measured), so a name +/// the account already holds is left for the backend to refuse. After that, +/// the capacity a provisioned table asks for, then the table quota. Two +/// concurrent requests can both pass the quota with one slot left, so an +/// account can end one table over it. +/// +/// # Errors +/// +/// `LimitExceededException` when a limit refuses the table, or the storage +/// error that stopped the account's tables from being read. +pub(crate) async fn enforce_create_table( + ctx: &OperationContext, + input: &CreateTableInput, +) -> Result<(), DynamoDbError> { + let names = table_names(ctx).await?; + if names.contains(&input.table_name) { + return Ok(()); + } + if let Some(requested) = limits::create_table_capacity(input) { + let reserved = reserved_capacity(ctx, &names).await?; + limits::check_create_table_capacity(requested, reserved, &ctx.limits, &ctx.region)?; + } + limits::check_table_count(names.len(), &ctx.limits) +} + +/// Refuse an `UpdateTable` that would raise `existing`, or the account, above +/// a provisioned-capacity maximum. +/// +/// # Errors +/// +/// `LimitExceededException` when a maximum refuses the change, or the storage +/// error that stopped the account's tables from being read. +pub(crate) async fn enforce_update_table( + ctx: &OperationContext, + input: &UpdateTableInput, + existing: &TableDescription, +) -> Result<(), DynamoDbError> { + let Some(after) = limits::update_table_capacity(existing, input) else { + return Ok(()); + }; + let names = table_names(ctx).await?; + let reserved = reserved_capacity(ctx, &names).await?; + limits::check_update_table_capacity( + input.provisioned_throughput.as_ref(), + existing, + reserved, + after, + &ctx.limits, + &ctx.region, + ) +} + +/// Every table name in the caller's account. +async fn table_names(ctx: &OperationContext) -> Result, DynamoDbError> { + let mut names = Vec::new(); + let mut start = None; + loop { + let page = ctx + .storage + .list_tables( + &ctx.account_id, + ListTablesInput { + limit: Some(LIST_PAGE), + exclusive_start_table_name: start, + }, + ) + .await + .map_err(storage_err_to_dynamo)?; + let empty = page.table_names.is_empty(); + names.extend(page.table_names); + match page.last_evaluated_table_name { + Some(last) if !empty => start = Some(last), + _ => return Ok(names), + } + } +} + +/// What the tables `names` and their global secondary indexes hold. +async fn reserved_capacity( + ctx: &OperationContext, + names: &[String], +) -> Result { + let mut reserved = CapacityUnits::default(); + for name in names { + let described = ctx + .storage + .describe_table( + &ctx.account_id, + DescribeTableInput { + table_name: name.clone(), + }, + ) + .await; + match described { + Ok(desc) => reserved = reserved.saturating_add(limits::table_capacity(&desc)), + // Deleted since it was listed, so it holds nothing. + Err(StorageError::TableNotFound(_)) => {} + Err(e) => return Err(storage_err_to_dynamo(e)), + } + } + Ok(reserved) +} diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 6db763f82..1f0cbfd2a 100755 --- a/crates/engine/src/update_table.rs +++ b/crates/engine/src/update_table.rs @@ -170,6 +170,21 @@ pub async fn handle_update_table( )?; } + // The provisioned-capacity maxima, checked against what the table holds now. + if extenddb_core::validation::provisioned_limits::changes_provisioned_capacity(&input) { + let existing = ctx + .storage + .describe_table( + &ctx.account_id, + extenddb_core::types::DescribeTableInput { + table_name: input.table_name.clone(), + }, + ) + .await + .map_err(|e| update_table_err_to_dynamo(e, &input.table_name))?; + crate::table_limits::enforce_update_table(ctx, &input, &existing).await?; + } + let table_name = input.table_name.clone(); // Kept for the post-condition check below, since `input` is moved into the // backend call. From 718d11e5bba0ae45a238c9e17ac687a3352a25af Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Wed, 4 Nov 2026 23:17:06 +0000 Subject: [PATCH 3/4] fix(auth): parse policies with the configured max_policy_document_bytes The size limit of a policy document was the literal 6,144 in PolicyDocument::from_json, so the max_policy_document_bytes key changed nothing. The key is now the limit on both paths. - Write: from_json_for_write now takes the limit as an argument. The management API's policy and permissions boundary handlers and the console's policy page pass the key to it. - Read: the authorization cache parses a stored identity policy, permissions boundary or session policy with the key. A stored policy over it does not parse, and authorization answers AccessDenied. from_json keeps the default of 6,144 bytes, now the named constant DEFAULT_MAX_POLICY_DOCUMENT_BYTES. The security model manual names the key. --- crates/auth/src/policy/document.rs | 59 +++++++--- crates/core/src/limits/mod.rs | 8 +- crates/server/src/authz_cache.rs | 108 ++++++++++++++---- crates/server/src/console/mod.rs | 3 + .../server/src/console/pages/policy_pages.rs | 9 +- crates/server/src/lib.rs | 3 + crates/server/src/management/iam_policy.rs | 7 +- crates/server/src/management/mod.rs | 3 + .../src/management/permissions_boundary.rs | 7 +- crates/server/src/serve.rs | 1 + docs/manuals/10-security-model.md | 4 +- 11 files changed, 159 insertions(+), 53 deletions(-) diff --git a/crates/auth/src/policy/document.rs b/crates/auth/src/policy/document.rs index dedcd5e2a..881635b9d 100755 --- a/crates/auth/src/policy/document.rs +++ b/crates/auth/src/policy/document.rs @@ -7,6 +7,7 @@ //! `ActionMatch` and `ResourceMatch` enums prevent invalid states where both //! `Action` and `NotAction` (or `Resource` and `NotResource`) are present. +use extenddb_core::limits::DEFAULT_MAX_POLICY_DOCUMENT_BYTES; use serde_json::Value; use std::fmt; @@ -142,23 +143,27 @@ impl fmt::Display for PolicyParseError { impl std::error::Error for PolicyParseError {} impl PolicyDocument { - /// Parse a policy document from a JSON string. + /// Parse a policy document from a JSON string, with the default size + /// limit ([`DEFAULT_MAX_POLICY_DOCUMENT_BYTES`]). /// - /// Enforces the default 6,144-byte size limit (matching AWS IAM inline - /// policy limit). This applies to both write-path validation and read-path - /// parsing of stored policies — a stored policy exceeding this limit will - /// fail to parse, triggering fail-closed denial (defense in depth). + /// The server parses with the configured `max_policy_document_bytes` + /// instead, through [`Self::from_json_with_size_limit`] on the read path + /// and [`Self::from_json_for_write`] on the write path. /// /// # Errors /// /// Returns `PolicyParseError` if the JSON is malformed or contains /// invalid policy constructs (e.g., both Action and `NotAction`). pub fn from_json(json: &str) -> Result { - Self::from_json_with_size_limit(json, 6_144) + Self::from_json_with_size_limit(json, DEFAULT_MAX_POLICY_DOCUMENT_BYTES) } /// Parse a policy document with an explicit size limit. /// + /// The read path parses a stored policy this way, with the same limit the + /// write path applies: a stored policy over the limit does not parse, and + /// its principal is denied (fail closed). + /// /// # Errors /// /// Returns `PolicyParseError` if the document exceeds `max_bytes` or is malformed. @@ -185,17 +190,18 @@ impl PolicyDocument { /// a managed or boundary policy) and apply the checks the service applies /// only when a policy is written. /// - /// Everything `from_json` checks, plus the `Sid` charset. The `Sid` check - /// runs at write only: a `Sid` is a statement label that does not affect - /// evaluation, so a policy stored before this check existed must keep - /// parsing on the request path rather than fail the principal closed. + /// Everything [`Self::from_json_with_size_limit`] checks with `max_bytes`, + /// the configured `max_policy_document_bytes`, plus the `Sid` charset. The + /// `Sid` check runs at write only: a `Sid` is a statement label that does + /// not affect evaluation, so a policy stored before this check existed must + /// keep parsing on the request path rather than fail the principal closed. /// /// # Errors /// - /// Returns `PolicyParseError` if `from_json` rejects the document or a - /// statement `Sid` is outside the service's charset. - pub fn from_json_for_write(json: &str) -> Result { - let doc = Self::from_json(json)?; + /// Returns `PolicyParseError` if the document exceeds `max_bytes`, does + /// not parse, or has a statement `Sid` outside the service's charset. + pub fn from_json_for_write(json: &str, max_bytes: usize) -> Result { + let doc = Self::from_json_with_size_limit(json, max_bytes)?; validate_statement_sids(&doc.statements)?; Ok(doc) } @@ -681,7 +687,8 @@ mod tests { r#"{{"Version":"2012-10-17","Statement":[{{"Sid":"{sid}","Effect":"Allow","Action":"dynamodb:GetItem","Resource":"*"}}]}}"# ); assert!( - PolicyDocument::from_json_for_write(&json).is_err(), + PolicyDocument::from_json_for_write(&json, DEFAULT_MAX_POLICY_DOCUMENT_BYTES) + .is_err(), "should reject Sid {sid:?} at write" ); // A policy stored before the check existed still parses on the @@ -700,12 +707,32 @@ mod tests { r#"{{"Version":"2012-10-17","Statement":[{{"Sid":"{sid}","Effect":"Allow","Action":"dynamodb:GetItem","Resource":"*"}}]}}"# ); assert!( - PolicyDocument::from_json_for_write(&json).is_ok(), + PolicyDocument::from_json_for_write(&json, DEFAULT_MAX_POLICY_DOCUMENT_BYTES) + .is_ok(), "should accept Sid {sid:?}" ); } } + /// Both paths take the limit from the caller: a document of exactly + /// `max_bytes` parses, and one byte over does not. + #[test] + fn the_size_limit_is_the_one_the_caller_passes() { + let json = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"dynamodb:GetItem","Resource":"*"}]}"#; + let at = json.len(); + assert!(PolicyDocument::from_json_with_size_limit(json, at).is_ok()); + assert!(PolicyDocument::from_json_for_write(json, at).is_ok()); + let over = PolicyDocument::from_json_for_write(json, at - 1).expect_err("one byte over"); + assert_eq!( + over.to_string(), + format!( + "policy parse error: policy document exceeds maximum size ({} bytes)", + at - 1 + ) + ); + assert!(PolicyDocument::from_json_with_size_limit(json, at - 1).is_err()); + } + #[test] fn reject_miscased_operator_names() { // The live service treats operator names, set qualifiers and the diff --git a/crates/core/src/limits/mod.rs b/crates/core/src/limits/mod.rs index a894f1b83..a898da9ed 100755 --- a/crates/core/src/limits/mod.rs +++ b/crates/core/src/limits/mod.rs @@ -148,9 +148,13 @@ fn default_max_expression_length_bytes() -> usize { fn default_max_expression_depth() -> usize { 150 } + +/// The default `max_policy_document_bytes`: the AWS IAM limit of 6,144 bytes +/// for an inline policy. +pub const DEFAULT_MAX_POLICY_DOCUMENT_BYTES: usize = 6_144; + fn default_max_policy_document_bytes() -> usize { - // AWS IAM limit: 6,144 bytes for inline policies. - 6_144 + DEFAULT_MAX_POLICY_DOCUMENT_BYTES } fn default_max_import_file_bytes() -> u64 { // 10 GB default. diff --git a/crates/server/src/authz_cache.rs b/crates/server/src/authz_cache.rs index 246b98199..81764a5a9 100644 --- a/crates/server/src/authz_cache.rs +++ b/crates/server/src/authz_cache.rs @@ -54,6 +54,10 @@ pub struct AuthzCacheConfig { pub principal_tags: SwrCacheConfig, pub resource_tags: SwrCacheConfig, pub session_data: SwrCacheConfig, + /// The size limit a stored policy is parsed with: the configured + /// `max_policy_document_bytes`, which the write path applies too. A + /// stored policy over it does not parse, and its principal is denied. + pub max_policy_document_bytes: usize, } impl Default for AuthzCacheConfig { @@ -84,6 +88,7 @@ impl Default for AuthzCacheConfig { name: "session_data", ..base }, + max_policy_document_bytes: extenddb_core::limits::DEFAULT_MAX_POLICY_DOCUMENT_BYTES, } } } @@ -164,19 +169,20 @@ impl CachedAuthzStore { // The principal creation-time caches reuse the per-principal tag cache // sizing; capture it before the field initializers consume it. let created_at_cfg = cfg.principal_tags.clone(); + let max_bytes = cfg.max_policy_document_bytes; Self { user_policies: mk_cache( - make_user_policies_loader(inner.clone()), + make_user_policies_loader(inner.clone(), max_bytes), cfg.identity_policies.clone(), pass_through, ), user_group_policies: mk_cache( - make_user_group_policies_loader(inner.clone()), + make_user_group_policies_loader(inner.clone(), max_bytes), cfg.group_policies.clone(), pass_through, ), user_boundary: mk_cache( - make_user_boundary_loader(inner.clone()), + make_user_boundary_loader(inner.clone(), max_bytes), cfg.boundary.clone(), pass_through, ), @@ -186,12 +192,12 @@ impl CachedAuthzStore { pass_through, ), role_policies: mk_cache( - make_role_policies_loader(inner.clone()), + make_role_policies_loader(inner.clone(), max_bytes), cfg.identity_policies.clone(), pass_through, ), role_boundary: mk_cache( - make_role_boundary_loader(inner.clone()), + make_role_boundary_loader(inner.clone(), max_bytes), cfg.boundary, pass_through, ), @@ -201,7 +207,7 @@ impl CachedAuthzStore { pass_through, ), session_data: mk_cache( - make_session_data_loader(inner.clone()), + make_session_data_loader(inner.clone(), max_bytes), cfg.session_data, pass_through, ), @@ -746,37 +752,46 @@ type CreatedAtLoader = Loader<(String, String), Arc, OpError>; // not exist" through this API, AND both should expire on the short // `negative_ttl`, not the long `ttl`. Otherwise probes against // nonexistent principals fill the LRU with empty positive entries. -fn make_user_policies_loader(inner: Arc) -> PolDocsLoader { +fn make_user_policies_loader( + inner: Arc, + max_bytes: usize, +) -> PolDocsLoader { Arc::new(move |(account_id, user_name): (String, String)| -> BoxFuture<'static, OpResult>> { let inner = inner.clone(); async move { let raw = inner.fetch_user_policies(&account_id, &user_name).await?; - policy_list_if_nonempty(&raw) + policy_list_if_nonempty(&raw, max_bytes) } .boxed() }) } -fn make_user_group_policies_loader(inner: Arc) -> PolDocsLoader { +fn make_user_group_policies_loader( + inner: Arc, + max_bytes: usize, +) -> PolDocsLoader { Arc::new(move |(account_id, user_name): (String, String)| -> BoxFuture<'static, OpResult>> { let inner = inner.clone(); async move { let raw = inner.fetch_user_group_policies(&account_id, &user_name).await?; - policy_list_if_nonempty(&raw) + policy_list_if_nonempty(&raw, max_bytes) } .boxed() }) } -fn make_role_policies_loader(inner: Arc) -> PolDocsLoader { +fn make_role_policies_loader( + inner: Arc, + max_bytes: usize, +) -> PolDocsLoader { Arc::new(move |(account_id, role_name): (String, String)| -> BoxFuture<'static, OpResult>> { let inner = inner.clone(); async move { let raw = inner.fetch_role_policies(&account_id, &role_name).await?; - policy_list_if_nonempty(&raw) + policy_list_if_nonempty(&raw, max_bytes) } .boxed() }) @@ -786,20 +801,23 @@ fn make_role_policies_loader(inner: Arc) -> PolDocsLoade /// (so the SWR cache stores it as a negative entry with the short /// `negative_ttl`); otherwise wrap the parsed documents in `Some`. /// Parse errors propagate. -fn policy_list_if_nonempty(raw: &[String]) -> OpResult> { +fn policy_list_if_nonempty(raw: &[String], max_bytes: usize) -> OpResult> { if raw.is_empty() { return Ok(None); } - parse_documents(raw).map(|v| Some(Arc::new(v))) + parse_documents(raw, max_bytes).map(|v| Some(Arc::new(v))) } -fn make_user_boundary_loader(inner: Arc) -> BoundaryLoader { +fn make_user_boundary_loader( + inner: Arc, + max_bytes: usize, +) -> BoundaryLoader { Arc::new(move |(account_id, user_name): (String, String)| -> BoxFuture<'static, OpResult>>> { let inner = inner.clone(); async move { match inner.fetch_user_boundary(&account_id, &user_name).await? { - Some(json_str) => Ok(Some(parse_document(&json_str)?)), + Some(json_str) => Ok(Some(parse_document(&json_str, max_bytes)?)), None => Ok(None), } } @@ -807,13 +825,16 @@ fn make_user_boundary_loader(inner: Arc) -> BoundaryLoad }) } -fn make_role_boundary_loader(inner: Arc) -> BoundaryLoader { +fn make_role_boundary_loader( + inner: Arc, + max_bytes: usize, +) -> BoundaryLoader { Arc::new(move |(account_id, role_name): (String, String)| -> BoxFuture<'static, OpResult>>> { let inner = inner.clone(); async move { match inner.fetch_role_boundary(&account_id, &role_name).await? { - Some(json_str) => Ok(Some(parse_document(&json_str)?)), + Some(json_str) => Ok(Some(parse_document(&json_str, max_bytes)?)), None => Ok(None), } } @@ -907,7 +928,7 @@ fn make_role_created_at_loader(inner: Arc) -> CreatedAtL }) } -fn make_session_data_loader(inner: Arc) -> SessionLoader { +fn make_session_data_loader(inner: Arc, max_bytes: usize) -> SessionLoader { Arc::new(move |(account_id, role_name, session_name): (String, String, String)| -> BoxFuture<'static, OpResult>>> { let inner = inner.clone(); @@ -918,7 +939,7 @@ fn make_session_data_loader(inner: Arc) -> SessionLoader let Some(data) = raw else { return Ok(None) }; let session_policy = match data.session_policy { - Some(json_str) => Some(parse_document(&json_str)?), + Some(json_str) => Some(parse_document(&json_str, max_bytes)?), None => None, }; let session_tags: HashMap = data.session_tags.into_iter().collect(); @@ -932,16 +953,16 @@ fn make_session_data_loader(inner: Arc) -> SessionLoader }) } -fn parse_documents(jsons: &[String]) -> OpResult>> { +fn parse_documents(jsons: &[String], max_bytes: usize) -> OpResult>> { let mut docs = Vec::with_capacity(jsons.len()); for s in jsons { - docs.push(parse_document(s)?); + docs.push(parse_document(s, max_bytes)?); } Ok(docs) } -fn parse_document(s: &str) -> OpResult> { - PolicyDocument::from_json(s) +fn parse_document(s: &str, max_bytes: usize) -> OpResult> { + PolicyDocument::from_json_with_size_limit(s, max_bytes) .map(Arc::new) .map_err(|e| OpError::Internal(format!("policy parse failed: {e}"))) } @@ -1189,6 +1210,7 @@ pub(crate) mod tests { ..base.clone() }, session_data: SwrCacheConfig { name: "sd", ..base }, + max_policy_document_bytes: extenddb_core::limits::DEFAULT_MAX_POLICY_DOCUMENT_BYTES, } } @@ -1219,6 +1241,44 @@ pub(crate) mod tests { ); } + /// A stored policy is parsed with the configured + /// `max_policy_document_bytes`. One byte over it, the load fails as a + /// parse failure, which authorization answers with `AccessDenied`. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_stored_policy_over_the_configured_size_does_not_load() { + let policy = allow_policy("dynamodb:GetItem"); + let inner = Arc::new(CountingAuthzStore::default()); + inner + .user_policies + .lock() + .unwrap() + .insert(("a".to_owned(), "u".to_owned()), vec![policy.clone()]); + + let at = CachedAuthzStore::new( + inner.clone(), + AuthzCacheConfig { + max_policy_document_bytes: policy.len(), + ..fast_authz_cfg() + }, + ); + assert_eq!(at.fetch_user_policies("a", "u").await.unwrap().len(), 1); + + let under = CachedAuthzStore::new( + inner, + AuthzCacheConfig { + max_policy_document_bytes: policy.len() - 1, + ..fast_authz_cfg() + }, + ); + match under.fetch_user_policies("a", "u").await { + Err(OpError::Internal(msg)) => assert!( + msg.starts_with("policy parse failed"), + "unexpected message: {msg}" + ), + other => panic!("expected a parse failure, got {other:?}"), + } + } + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn invalidate_user_policies_forces_refetch() { let inner = Arc::new(CountingAuthzStore::default()); diff --git a/crates/server/src/console/mod.rs b/crates/server/src/console/mod.rs index 76cc752c7..c5d543361 100755 --- a/crates/server/src/console/mod.rs +++ b/crates/server/src/console/mod.rs @@ -55,6 +55,9 @@ pub struct ConsoleState { /// `/console/metrics-data` response with latency segments (same data the /// top-level `/metrics` handler adds). pub metrics: Arc, + /// The configured `max_policy_document_bytes`: a policy over it is + /// refused on write. + pub max_policy_document_bytes: usize, } /// Build the console router. diff --git a/crates/server/src/console/pages/policy_pages.rs b/crates/server/src/console/pages/policy_pages.rs index 9c7634cf4..712d27cde 100755 --- a/crates/server/src/console/pages/policy_pages.rs +++ b/crates/server/src/console/pages/policy_pages.rs @@ -310,9 +310,12 @@ async fn put_policy( let invalid = if doc.get("Version").is_none() || doc.get("Statement").is_none() { Some("Policy document must contain Version and Statement".to_owned()) } else { - extenddb_auth::policy::document::PolicyDocument::from_json_for_write(&doc.to_string()) - .err() - .map(|e| format!("Invalid policy document: {e}")) + extenddb_auth::policy::document::PolicyDocument::from_json_for_write( + &doc.to_string(), + state.max_policy_document_bytes, + ) + .err() + .map(|e| format!("Invalid policy document: {e}")) }; if let Some(msg) = invalid { let nav = html::nav_bar(&identity_label(&session.identity)); diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index f608c74d7..65ebd6754 100755 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -141,6 +141,7 @@ pub async fn start_server( let authz_cache_for_mgmt = state.authz_cache.clone(); let table_key_info_cache_for_mgmt = state.table_key_info_cache.clone(); let mgmt_region = state.region.clone(); + let max_policy_document_bytes = state.limits.max_policy_document_bytes; let version_info = state.version_info.clone(); let docs_store = state.docs_store.clone(); let listen_url = { @@ -191,6 +192,7 @@ pub async fn start_server( table_key_info_cache: table_key_info_cache_for_mgmt, region: mgmt_region, secure_transport: tls_enabled, + max_policy_document_bytes, }); let mgmt_router = management::router().with_state(mgmt_state); app = app.nest("/management", mgmt_router); @@ -204,6 +206,7 @@ pub async fn start_server( docs_store: docs_store.clone(), auth_cache: auth_cache_for_console, metrics: metrics_for_console, + max_policy_document_bytes, }); let console_router = console::router().with_state(console_state); app = app diff --git a/crates/server/src/management/iam_policy.rs b/crates/server/src/management/iam_policy.rs index 8e5de48c6..5e08e9289 100755 --- a/crates/server/src/management/iam_policy.rs +++ b/crates/server/src/management/iam_policy.rs @@ -200,9 +200,10 @@ async fn put_policy( // Strict parse-on-write: reject documents the policy evaluator cannot parse. { let json_str = document.to_string(); - if let Err(e) = - extenddb_auth::policy::document::PolicyDocument::from_json_for_write(&json_str) - { + if let Err(e) = extenddb_auth::policy::document::PolicyDocument::from_json_for_write( + &json_str, + state.max_policy_document_bytes, + ) { return ( StatusCode::BAD_REQUEST, format!("Invalid policy document: {e}"), diff --git a/crates/server/src/management/mod.rs b/crates/server/src/management/mod.rs index ec3968766..c9f4acf57 100755 --- a/crates/server/src/management/mod.rs +++ b/crates/server/src/management/mod.rs @@ -58,6 +58,9 @@ pub struct ManagementState { /// Whether the listener serves TLS, resolved into `aws:SecureTransport` /// during trust policy evaluation. pub secure_transport: bool, + /// The configured `max_policy_document_bytes`: a policy or permissions + /// boundary over it is refused on write. + pub max_policy_document_bytes: usize, } /// Build the management API router. diff --git a/crates/server/src/management/permissions_boundary.rs b/crates/server/src/management/permissions_boundary.rs index dd34bcad6..dd63de76e 100755 --- a/crates/server/src/management/permissions_boundary.rs +++ b/crates/server/src/management/permissions_boundary.rs @@ -126,9 +126,10 @@ async fn set_boundary( // a malformed CIDR with MalformedPolicyDocument. { let json_str = document.to_string(); - if let Err(e) = - extenddb_auth::policy::document::PolicyDocument::from_json_for_write(&json_str) - { + if let Err(e) = extenddb_auth::policy::document::PolicyDocument::from_json_for_write( + &json_str, + state.max_policy_document_bytes, + ) { return ( StatusCode::BAD_REQUEST, format!("Invalid permissions boundary document: {e}"), diff --git a/crates/server/src/serve.rs b/crates/server/src/serve.rs index 57bfab1bd..f9cf59624 100644 --- a/crates/server/src/serve.rs +++ b/crates/server/src/serve.rs @@ -477,6 +477,7 @@ async fn serve_inner(params: ServeParams, port: u16) -> anyhow::Result<()> { principal_tags: make_cache_cfg("principal_tags"), resource_tags: make_cache_cfg("resource_tags"), session_data: make_cache_cfg("session_data"), + max_policy_document_bytes: app_config.limits.max_policy_document_bytes, }; Arc::new(if cache_enabled { crate::CachedAuthzStore::new(store, cfg) diff --git a/docs/manuals/10-security-model.md b/docs/manuals/10-security-model.md index fb2617cdf..f01bdb562 100755 --- a/docs/manuals/10-security-model.md +++ b/docs/manuals/10-security-model.md @@ -139,7 +139,7 @@ Policy sources collected for evaluation: - **Auth before JSON parse**: SigV4 signature verification runs before the request body is parsed. Invalid signatures are rejected with constant-time comparison before any business logic executes. - **Concurrent policy fetching**: Identity policies, group policies, and permissions boundaries are fetched concurrently from the catalog when they are not in the server's cache (see Credential and Policy Caching). All must succeed for evaluation to proceed. - **Constant-time rejection for inactive keys**: Inactive or expired access keys are rejected without timing differences that could reveal key existence. -- **Policy document validation on write**: Policy documents are validated for JSON structure and size-capped (6,144 bytes) when attached via the management API. Invalid documents are rejected before storage. +- **Policy document validation on write**: Policy documents are validated for JSON structure and size-capped at `max_policy_document_bytes` (6,144 bytes by default) when attached via the management API or the console. Invalid documents are rejected before storage. A stored policy is parsed with the same cap, so a policy stored before the cap was lowered no longer parses, and its principal is denied. - **Expression depth and token limits**: Expression parsing enforces configurable depth (default 150) and token limits (default 4,096) to prevent resource exhaustion. ### Supported Condition Operators @@ -277,7 +277,7 @@ Input validation is layered: |-------|---------|-------------| | Max expression tokens | 4,096 | Maximum tokens in a parsed expression | | Max expression depth | 150 | Maximum nesting depth in expressions | -| Max policy document size | 6,144 bytes | Maximum size of an IAM policy document | +| Max policy document size | 6,144 bytes | Maximum size of an IAM policy document (`max_policy_document_bytes`) | ### Path Traversal Protection From 29dbae970369d00c7c7710ecb29e8bbbfc92560f Mon Sep 17 00:00:00 2001 From: Lee Hannigan <29015211+LeeroyHannigan@users.noreply.github.com> Date: Wed, 4 Nov 2026 23:46:16 +0000 Subject: [PATCH 4/4] fix(fuzz): pass the size limit to both policy parse paths from_json_for_write now takes the limit as a second argument, and the fuzz target called it with one. The target now drives from_json_with_size_limit and from_json_for_write with DEFAULT_MAX_POLICY_DOCUMENT_BYTES, the two calls the server makes, and its header names them. --- fuzz/fuzz_targets/policy_document.rs | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/fuzz/fuzz_targets/policy_document.rs b/fuzz/fuzz_targets/policy_document.rs index c2949bc4c..aad8f143a 100644 --- a/fuzz/fuzz_targets/policy_document.rs +++ b/fuzz/fuzz_targets/policy_document.rs @@ -4,23 +4,25 @@ // Mirrors the parse paths for identity policies and permissions boundaries. On // a write, crates/server/src/management/iam_policy.rs and // permissions_boundary.rs call PolicyDocument::from_json_for_write; on the -// request path, crates/server/src/authz_cache.rs calls PolicyDocument::from_json -// on a stored identity policy, permissions boundary or session policy. -// from_json caps the document at 6,144 bytes itself. Both are driven here with -// the same arbitrary JSON text. A trust policy is checked on write by -// validate_trust_policy in crates/server/src/management/assume_role.rs, which -// this target does not call. AssumeRole evaluates its conditions with -// parse_conditions, which from_json also calls. +// request path, crates/server/src/authz_cache.rs calls +// PolicyDocument::from_json_with_size_limit on a stored identity policy, +// permissions boundary or session policy. Both take the configured +// max_policy_document_bytes, here its default of 6,144 bytes, and both are +// driven with the same arbitrary JSON text. A trust policy is checked on write +// by validate_trust_policy in crates/server/src/management/assume_role.rs, +// which this target does not call. AssumeRole evaluates its conditions with +// parse_conditions, which the request path also calls. #![no_main] use libfuzzer_sys::fuzz_target; use extenddb_auth::policy::document::PolicyDocument; +use extenddb_core::limits::DEFAULT_MAX_POLICY_DOCUMENT_BYTES; fuzz_target!(|data: &[u8]| { if let Ok(json) = std::str::from_utf8(data) { - let _ = PolicyDocument::from_json(json); - let _ = PolicyDocument::from_json_for_write(json); + let _ = PolicyDocument::from_json_with_size_limit(json, DEFAULT_MAX_POLICY_DOCUMENT_BYTES); + let _ = PolicyDocument::from_json_for_write(json, DEFAULT_MAX_POLICY_DOCUMENT_BYTES); } });