fix(s3): decrypt SSE-KMS bodies for internal object readers

crates/fakecloud-cloudformation/src/extras.rs

@@ -592,12 +592,31 @@ impl CloudFormationService {
         let (bucket, key) = parse_s3_url(url)?;
         let mut accounts = self.deps.s3.write();
         let state = accounts.get_or_create(account_id);
-        let body_ref = {
+        let (body_ref, sse_algorithm, kms_key_id) = {
             let b = state.buckets.get(&bucket)?;
-            b.objects.get(&key)?.body.clone()
+            let o = b.objects.get(&key)?;
+            (
+                o.body.clone(),
+                o.sse_algorithm.clone(),
+                o.sse_kms_key_id.clone(),
+            )
         };
         let bytes = state.read_body(&body_ref).ok()?;
-        String::from_utf8(bytes.to_vec()).ok()
+        // An SSE-KMS bucket stores an envelope, not the object -- and
+        // `cdk bootstrap` makes its assets bucket `aws:kms`. The envelope is
+        // base64 ASCII, so it survives `from_utf8` and passes the non-empty
+        // check below, producing a "template" with no resources: the stack then
+        // reports CREATE_COMPLETE having provisioned nothing.
+        let plaintext = fakecloud_s3::sse::decrypt_body(
+            self.kms_hook.as_ref(),
+            account_id,
+            &bucket,
+            sse_algorithm.as_deref(),
+            kms_key_id.as_deref(),
+            bytes.to_vec(),
+        )
+        .ok()?;
+        String::from_utf8(plaintext).ok()
     }
 
     /// Whether a stack resource's physical resource still exists in its

crates/fakecloud-cloudformation/src/resource_provisioner/mod.rs

@@ -3224,14 +3224,51 @@ impl ResourceProvisioner {
                 .objects
                 .get(key)
                 .ok_or_else(|| format!("S3 object s3://{bucket}/{key} does not exist"))?;
-            object.body.clone()
+            (
+                object.body.clone(),
+                object.sse_algorithm.clone(),
+                object.sse_kms_key_id.clone(),
+            )
         };
         // `read_body` consults the body cache (which is owned by the state),
         // so re-borrow `state` after dropping the bucket borrow above.
-        state
-            .read_body(&body_ref)
+        let (body_ref, sse_algorithm, kms_key_id) = body_ref;
+        self.read_object_body(
+            state,
+            bucket,
+            &body_ref,
+            sse_algorithm.as_deref(),
+            kms_key_id.as_deref(),
+        )
+    }
+
+    /// Read a stored object body and unwrap it when it is an SSE-KMS envelope.
+    ///
+    /// Every provisioner read of an S3 object goes through here. An SSE-KMS
+    /// bucket stores an envelope, not the object, and `cdk bootstrap` makes its
+    /// assets bucket `aws:kms`: a reader that skips this hands Lambda a base64
+    /// KMS blob in place of its ZIP, or parses a nested-stack "template" with no
+    /// resources.
+    fn read_object_body(
+        &self,
+        state: &fakecloud_s3::S3State,
+        bucket: &str,
+        body_ref: &fakecloud_persistence::s3::BodyRef,
+        sse_algorithm: Option<&str>,
+        kms_key_id: Option<&str>,
+    ) -> Result<Vec<u8>, String> {
+        let stored = state
+            .read_body(body_ref)
             .map(|b| b.to_vec())
-            .map_err(|e| format!("S3 read failed: {e}"))
+            .map_err(|e| format!("S3 read failed: {e}"))?;
+        fakecloud_s3::sse::decrypt_body(
+            self.kms_hook.as_ref(),
+            &self.account_id,
+            bucket,
+            sse_algorithm,
+            kms_key_id,
+            stored,
+        )
     }
 
     /// Read a specific object version's bytes. Used when a CFN property
@@ -3256,24 +3293,39 @@ impl ResourceProvisioner {
                 .objects
                 .get(key)
                 .filter(|o| o.version_id.as_deref() == Some(version_id))
-                .map(|o| o.body.clone());
+                .map(|o| {
+                    (
+                        o.body.clone(),
+                        o.sse_algorithm.clone(),
+                        o.sse_kms_key_id.clone(),
+                    )
+                });
             from_current
                 .or_else(|| {
                     b.object_versions.get(key).and_then(|versions| {
                         versions
                             .iter()
                             .find(|o| o.version_id.as_deref() == Some(version_id))
-                            .map(|o| o.body.clone())
+                            .map(|o| {
+                                (
+                                    o.body.clone(),
+                                    o.sse_algorithm.clone(),
+                                    o.sse_kms_key_id.clone(),
+                                )
+                            })
                     })
                 })
                 .ok_or_else(|| {
                     format!("S3 object s3://{bucket}/{key} version {version_id} does not exist")
                 })?
         };
-        state
-            .read_body(&body_ref)
-            .map(|b| b.to_vec())
-            .map_err(|e| format!("S3 read failed: {e}"))
+        self.read_object_body(
+            state,
+            bucket,
+            &body_ref.0,
+            body_ref.1.as_deref(),
+            body_ref.2.as_deref(),
+        )
     }
 
     /// Build a canonical Lambda resource-policy statement from a CFN
@@ -3679,10 +3731,14 @@ impl ResourceProvisioner {
             .objects
             .get(key)
             .ok_or_else(|| format!("S3 object not found: {bucket}/{key}"))?;
-        let bytes = s3_state
-            .read_body(&obj.body)
-            .map_err(|e| format!("Failed to read S3 object body: {e}"))?;
-        String::from_utf8(bytes.to_vec()).map_err(|e| format!("S3 object is not valid UTF-8: {e}"))
+        let bytes = self.read_object_body(
+            s3_state,
+            bucket,
+            &obj.body,
+            obj.sse_algorithm.as_deref(),
+            obj.sse_kms_key_id.as_deref(),
+        )?;
+        String::from_utf8(bytes).map_err(|e| format!("S3 object is not valid UTF-8: {e}"))
     }
 }
 
@@ -7952,6 +8008,90 @@ mod tests {
         assert!(prov.delete_resource(&sr).is_ok());
     }
 
+    /// Store `plaintext` the way S3 does in an SSE-KMS bucket: as the KMS
+    /// envelope, with `sse_algorithm = aws:kms`. `version` lands on the
+    /// current object.
+    fn put_sse_kms_object(
+        prov: &ResourceProvisioner,
+        bucket: &str,
+        key: &str,
+        version: &str,
+        plaintext: &[u8],
+    ) {
+        let mut ctx = std::collections::HashMap::new();
+        ctx.insert(
+            "aws:s3:arn".to_string(),
+            fakecloud_aws::arn::Arn::s3(bucket).to_string(),
+        );
+        let envelope = prov
+            .kms_hook
+            .as_ref()
+            .expect("make_provisioner wires KMS")
+            .encrypt(
+                &prov.account_id,
+                &prov.region,
+                "aws/s3",
+                plaintext,
+                "s3.amazonaws.com",
+                ctx,
+            )
+            .expect("encrypt");
+        assert_ne!(
+            envelope.as_bytes(),
+            plaintext,
+            "the stored body must be the envelope"
+        );
+        let mut accounts = prov.s3_state.write();
+        let state = accounts.get_or_create(&prov.account_id);
+        let mut b = fakecloud_s3::S3Bucket::new(bucket, &prov.region, &prov.account_id);
+        b.objects.insert(
+            key.to_string(),
+            fakecloud_s3::S3Object {
+                key: key.to_string(),
+                size: envelope.len() as u64,
+                body: fakecloud_s3::memory_body(bytes::Bytes::from(envelope.into_bytes())),
+                sse_algorithm: Some("aws:kms".to_string()),
+                version_id: Some(version.to_string()),
+                ..Default::default()
+            },
+        );
+        state.buckets.insert(bucket.to_string(), b);
+    }
+
+    #[test]
+    fn current_object_read_decrypts_an_sse_kms_body() {
+        let prov = make_provisioner();
+        put_sse_kms_object(&prov, "assets", "code.zip", "v1", b"plaintext body");
+        assert_eq!(
+            prov.read_s3_object_bytes("assets", "code.zip").unwrap(),
+            b"plaintext body"
+        );
+    }
+
+    #[test]
+    fn pinned_version_read_decrypts_an_sse_kms_body() {
+        // A Step Functions `DefinitionS3Location.Version` takes this path.
+        let prov = make_provisioner();
+        put_sse_kms_object(&prov, "assets", "sm.json", "v1", br#"{"StartAt":"A"}"#);
+        assert_eq!(
+            prov.read_s3_object_version_bytes("assets", "sm.json", "v1")
+                .unwrap(),
+            br#"{"StartAt":"A"}"#
+        );
+    }
+
+    #[test]
+    fn nested_stack_template_read_decrypts_an_sse_kms_body() {
+        // CDK uploads nested-stack templates to its `aws:kms` assets bucket.
+        let prov = make_provisioner();
+        let template = r#"{"Resources":{"Q":{"Type":"AWS::SQS::Queue"}}}"#;
+        put_sse_kms_object(&prov, "assets", "nested.json", "v1", template.as_bytes());
+        assert_eq!(
+            prov.fetch_s3_template("assets", "nested.json").unwrap(),
+            template
+        );
+    }
+
     fn cn_provisioner() -> ResourceProvisioner {
         let mut prov = make_provisioner();
         prov.region = "cn-north-1".to_string();

crates/fakecloud-cloudformation/src/service.rs

@@ -682,8 +682,9 @@ pub struct CloudFormationService {
     /// real store via `with_s3_store` once it has been built.
     s3_store: Arc<dyn S3Store>,
     /// The server's KMS hook (persisting minted keys), used by provisioners
-    /// that report an AWS-managed key for a default-encrypted resource.
-    kms_hook: Option<Arc<dyn fakecloud_core::delivery::KmsHook>>,
+    /// that report an AWS-managed key for a default-encrypted resource, and to
+    /// unwrap SSE-KMS bodies read from S3 (a `TemplateURL`, Lambda code).
+    pub(crate) kms_hook: Option<Arc<dyn fakecloud_core::delivery::KmsHook>>,
     /// Service Quotas, so EC2 resources a stack provisions are held to the
     /// account's applied quotas like direct API calls.
     quota_provider: Option<Arc<dyn fakecloud_core::quota::QuotaProvider>>,