feat(dynamodb): IAM enforcement, resource-based policies and fine-grained access control

README.md

@@ -54,7 +54,7 @@ Works as a drop-in for LocalStack in CI, with Terraform (`endpoints` block), CDK
 - **Single binary.** ~19 MB, ~10 MiB idle, ~300ms startup. No Docker needed to run fakecloud itself.
 - **Full Bedrock surface.** 216 ops across 4 APIs with real `InvokeModel`/`Converse` streaming, guardrails, agents, and flows. Configurable responses + fault injection for deterministic tests. See [`/bedrock-emulator/`](https://fakecloud.dev/bedrock-emulator/).
 - **First-party test SDKs** for TypeScript, Python, Go, PHP, Java, and Rust. Assert on what your code called without raw HTTP.
-- **Opt-in SigV4 verification and IAM enforcement.** Off by default so tests just work; `--verify-sigv4` for real signature checking and `--iam soft|strict` for policy evaluation across IAM/STS/SQS/SNS/S3. See [security docs](https://fakecloud.dev/docs/reference/security/).
+- **Opt-in SigV4 verification and IAM enforcement.** Off by default so tests just work; `--verify-sigv4` for real signature checking and `--iam soft|strict` for policy evaluation across IAM, STS, SQS, SNS, S3, KMS, Lambda, DynamoDB (and Streams), ELBv2 and Scheduler. See [security docs](https://fakecloud.dev/docs/reference/security/).
 - **Run your app unmodified.** An app that expects an instance/task role resolves the AWS SDK default credential chain against fakecloud with no static keys and no code change: point `AWS_CONTAINER_CREDENTIALS_FULL_URI` at `/_fakecloud/credentials`. See [Run an app unmodified](https://fakecloud.dev/docs/guides/instance-credentials/).
 - **LocalStack and real-AWS URL compatibility.** Both `*.localhost.localstack.cloud` and `*.amazonaws.com` Host headers route correctly, including every S3 virtual-hosted variant. Persisted URLs and dev scripts from either system replay unchanged.
 

crates/fakecloud-core/src/service.rs

@@ -747,6 +747,20 @@ pub trait AwsService: Send + Sync {
         None
     }
 
+    /// Every IAM authorization an incoming request needs.
+    ///
+    /// Most operations act on one resource and need one action, which is
+    /// what the default returns ([`AwsService::iam_action_for`]). Some need
+    /// several: a batch or transaction naming several resources needs the
+    /// action on each, and an operation can require more than one action
+    /// (DynamoDB's `CreateTable` with `Tags` also needs `TagResource`).
+    /// Dispatch evaluates every action returned and denies the request if
+    /// any of them is denied. An empty list means the operation has no
+    /// mapping, which strict enforcement denies.
+    fn iam_actions_for(&self, request: &AwsRequest) -> Vec<crate::auth::IamAction> {
+        self.iam_action_for(request).into_iter().collect()
+    }
+
     /// Derive service-specific IAM condition keys for an incoming request.
     ///
     /// Called right after [`AwsService::iam_action_for`] when IAM

crates/fakecloud-dynamodb/src/lib.rs

@@ -1,11 +1,13 @@
 pub mod export_import;
+pub mod resource_policy;
 pub(crate) mod service;
 pub(crate) mod state;
 pub mod streams;
 pub mod streams_dataplane;
 pub mod ttl;
 
 pub use export_import::{import_aws_export, import_aws_exports_dir, ImportOutcome};
+pub use resource_policy::DynamoDbResourcePolicyProvider;
 pub(crate) use service::helpers::schemas::{
     parse_attribute_definitions, parse_key_schema, parse_provisioned_throughput,
 };

crates/fakecloud-dynamodb/src/resource_policy.rs

@@ -0,0 +1,144 @@
+//! DynamoDB implementation of [`ResourcePolicyProvider`]: the policy IAM
+//! enforcement evaluates alongside the caller's identity policies.
+//!
+//! A table's policy also governs its indexes (AWS has no separate index
+//! policy); a stream has a policy of its own. Backups, exports and imports
+//! cannot carry resource-based policies.
+
+use std::sync::Arc;
+
+use fakecloud_core::auth::ResourcePolicyProvider;
+
+use crate::state::SharedDynamoDbState;
+
+pub struct DynamoDbResourcePolicyProvider {
+    state: SharedDynamoDbState,
+}
+
+impl DynamoDbResourcePolicyProvider {
+    pub fn new(state: SharedDynamoDbState) -> Self {
+        Self { state }
+    }
+
+    /// Convenience constructor for server bootstrap's
+    /// `MultiResourcePolicyProvider`.
+    pub fn shared(state: SharedDynamoDbState) -> Arc<dyn ResourcePolicyProvider> {
+        Arc::new(Self::new(state))
+    }
+}
+
+/// The account, table name and sub-resource path of a DynamoDB table-scoped
+/// ARN: `arn:aws:dynamodb:REGION:ACCOUNT:table/NAME[/KIND/ID]`.
+fn parse(arn: &str) -> Option<(&str, &str, Option<&str>)> {
+    let rest = arn.strip_prefix("arn:aws:dynamodb:")?;
+    let (scope, path) = rest.split_once(":table/")?;
+    let account = scope.split(':').nth(1).filter(|a| !a.is_empty())?;
+    let (name, sub) = match path.split_once('/') {
+        Some((name, sub)) => (name, Some(sub)),
+        None => (path, None),
+    };
+    (!name.is_empty()).then_some((account, name, sub))
+}
+
+fn is_dynamodb(service: &str) -> bool {
+    service.eq_ignore_ascii_case("dynamodb") || service.eq_ignore_ascii_case("dynamodbstreams")
+}
+
+impl ResourcePolicyProvider for DynamoDbResourcePolicyProvider {
+    fn resource_policy(&self, service: &str, resource_arn: &str) -> Option<String> {
+        if !is_dynamodb(service) {
+            return None;
+        }
+        let (account, name, sub) = parse(resource_arn)?;
+        let accounts = self.state.read();
+        let state = accounts.get(account)?;
+        let table = state.tables.get(name)?;
+        match sub {
+            None => table.resource_policy.clone(),
+            Some(sub) if sub.starts_with("index/") => table.resource_policy.clone(),
+            Some(sub) if sub.starts_with("stream/") => {
+                state.stream_policies.get(resource_arn).cloned()
+            }
+            Some(_) => None,
+        }
+    }
+
+    fn resource_owner_account(&self, service: &str, resource_arn: &str) -> Option<String> {
+        if !is_dynamodb(service) {
+            return None;
+        }
+        parse(resource_arn).map(|(account, _, _)| account.to_string())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::state::{DynamoDbState, DynamoTable, KeySchemaElement, ProvisionedThroughput};
+    use fakecloud_core::multi_account::MultiAccountState;
+
+    const ARN: &str = "arn:aws:dynamodb:us-east-1:111122223333:table/Orders";
+
+    fn provider() -> DynamoDbResourcePolicyProvider {
+        let mut accounts = MultiAccountState::<DynamoDbState>::new("111122223333", "us-east-1", "");
+        let state = accounts.get_or_create("111122223333");
+        let mut table = DynamoTable::new(
+            "Orders".to_string(),
+            ARN.to_string(),
+            "id".to_string(),
+            vec![KeySchemaElement {
+                attribute_name: "pk".to_string(),
+                key_type: "HASH".to_string(),
+            }],
+            vec![],
+            ProvisionedThroughput {
+                read_capacity_units: 1,
+                write_capacity_units: 1,
+            },
+            "PAY_PER_REQUEST".to_string(),
+            chrono::Utc::now(),
+        );
+        table.resource_policy = Some("table-policy".to_string());
+        state.tables.insert("Orders".to_string(), table);
+        state
+            .stream_policies
+            .insert(format!("{ARN}/stream/label"), "stream-policy".to_string());
+        DynamoDbResourcePolicyProvider::new(Arc::new(parking_lot::RwLock::new(accounts)))
+    }
+
+    #[test]
+    fn tables_and_indexes_share_the_table_policy_and_streams_have_their_own() {
+        let p = provider();
+        assert_eq!(
+            p.resource_policy("dynamodb", ARN).as_deref(),
+            Some("table-policy")
+        );
+        assert_eq!(
+            p.resource_policy("dynamodb", &format!("{ARN}/index/by-g"))
+                .as_deref(),
+            Some("table-policy")
+        );
+        assert_eq!(
+            p.resource_policy("dynamodbstreams", &format!("{ARN}/stream/label"))
+                .as_deref(),
+            Some("stream-policy")
+        );
+        assert_eq!(
+            p.resource_policy("dynamodb", &format!("{ARN}/stream/other")),
+            None
+        );
+        assert_eq!(
+            p.resource_policy("dynamodb", &format!("{ARN}/backup/b")),
+            None
+        );
+        assert_eq!(p.resource_policy("sqs", ARN), None);
+        assert_eq!(
+            p.resource_owner_account(
+                "dynamodb",
+                "arn:aws:dynamodb:us-east-1:444455556666:table/X"
+            )
+            .as_deref(),
+            Some("444455556666")
+        );
+    }
+}

crates/fakecloud-dynamodb/src/service/batch.rs

@@ -1162,7 +1162,7 @@ impl DynamoDbService {
             let outcome = execute_partiql_in_state(state, statement, &parameters)?;
             // ExecuteStatement honors Limit + NextToken on a SELECT result set
             // (AWS paginates PartiQL SELECTs); the token is an opaque cursor.
-            let response = apply_execute_statement_pagination(
+            let mut response = apply_execute_statement_pagination(
                 outcome.response.clone(),
                 outcome
                     .table_name
@@ -1171,6 +1171,12 @@ impl DynamoDbService {
                 limit,
                 next_token.as_deref(),
             );
+            // Columns are projected after paging: the cursor is the last
+            // row's full primary key.
+            super::helpers::partiql::project_partiql_response(
+                &mut response,
+                outcome.projection.as_ref(),
+            );
 
             let kinesis_info = if let (Some(table_name), Some(event_name)) =
                 (outcome.table_name.as_ref(), outcome.event_name.as_ref())
@@ -1255,9 +1261,14 @@ impl DynamoDbService {
 
                 match execute_partiql_in_state(state, statement, &parameters) {
                     Ok(outcome) => {
+                        let mut projected = outcome.response.clone();
+                        super::helpers::partiql::project_partiql_response(
+                            &mut projected,
+                            outcome.projection.as_ref(),
+                        );
                         responses.push(batch_partiql_response(
                             statement,
-                            outcome.response.clone(),
+                            projected,
                             outcome
                                 .table_name
                                 .as_ref()
@@ -1478,7 +1489,12 @@ impl DynamoDbService {
 
             match execute_partiql_in_state(state, statement, &parameters) {
                 Ok(outcome) => {
-                    applied_responses.push(outcome.response);
+                    let mut projected = outcome.response;
+                    super::helpers::partiql::project_partiql_response(
+                        &mut projected,
+                        outcome.projection.as_ref(),
+                    );
+                    applied_responses.push(projected);
                     let table_name = match outcome.table_name {
                         Some(n) => n,
                         None => continue,
@@ -3026,15 +3042,18 @@ mod tests {
         .unwrap();
 
         // A non-ASCII, non-key attribute in the WHERE predicate must not panic
-        // the RETURNING scan (it simply matches nothing).
-        svc.execute_statement(&req_for(
-            "ExecuteStatement",
-            json!({
-                "Statement": "UPDATE \"Widgets\" SET \"x\" = ? WHERE \"café\" = ?",
-                "Parameters": [{ "S": "1" }, { "S": "z" }]
-            }),
-        ))
-        .unwrap();
+        // the RETURNING scan. It is not the key, so DynamoDB rejects it.
+        let err = svc
+            .execute_statement(&req_for(
+                "ExecuteStatement",
+                json!({
+                    "Statement": "UPDATE \"Widgets\" SET \"x\" = ? WHERE \"café\" = ?",
+                    "Parameters": [{ "S": "1" }, { "S": "z" }]
+                }),
+            ))
+            .err()
+            .expect("a WHERE without the key is rejected");
+        assert_eq!(err.code(), "ValidationException");
 
         // Read back through a batch single-item SELECT with tight `"pk"=?`
         // spacing (no surrounding spaces) to confirm the key-check tolerates it.
@@ -3070,14 +3089,32 @@ mod tests {
         ))
         .unwrap();
 
-        let updated = svc
+        // Without a WHERE the statement names no item, which DynamoDB rejects;
+        // the RETURNING clause is still stripped rather than parsed as part of
+        // the SET expression, so the error is the key check, not a garbled
+        // update expression.
+        let err = svc
             .execute_statement(&req_for(
                 "ExecuteStatement",
                 json!({
                     "Statement": "UPDATE \"Widgets\" SET \"flag\" = ? RETURNING ALL NEW *",
                     "Parameters": [{ "S": "on" }]
                 }),
             ))
+            .err()
+            .expect("an UPDATE without a key WHERE is rejected");
+        assert_eq!(
+            err.to_string(),
+            "ValidationException: Where clause does not contain a mandatory equality on all key attributes"
+        );
+        let updated = svc
+            .execute_statement(&req_for(
+                "ExecuteStatement",
+                json!({
+                    "Statement": "UPDATE \"Widgets\" SET \"flag\" = ? WHERE \"pk\" = 'a' RETURNING ALL NEW *",
+                    "Parameters": [{ "S": "on" }]
+                }),
+            ))
             .unwrap();
         assert_eq!(response_body(&updated)["Item"]["flag"]["S"], "on");
     }

crates/fakecloud-dynamodb/src/service/helpers/mod.rs

@@ -810,6 +810,10 @@ pub(crate) struct PartiqlOutcome {
     pub keys: Option<HashMap<String, AttributeValue>>,
     pub old_image: Option<HashMap<String, AttributeValue>>,
     pub new_image: Option<HashMap<String, AttributeValue>>,
+    /// A SELECT's column list as a projection request, applied to the rows
+    /// the caller returns -- after any pagination, whose cursor needs each
+    /// row's full primary key. `None` for `*` and for writes.
+    pub projection: Option<Value>,
 }
 
 /// AST for a parsed PartiQL WHERE clause. Leaf conditions reuse