aws_ssm - apply bucket SSE settings when uploading files to the managed node

changelogs/fragments/3018-aws_ssm-put-file-sse.yml

@@ -0,0 +1,8 @@
+---
+bugfixes:
+  - aws_ssm - apply the transfer bucket's server-side-encryption settings when uploading
+    files from the controller to the managed node (``put_file``), matching the existing
+    ``fetch_file`` behaviour. Previously the upload was made without encryption settings,
+    so buckets whose policy denies unencrypted ``PutObject`` requests rejected every file
+    transfer with ``AccessDenied``
+    (https://github.com/ansible-collections/amazon.aws/issues/3018).

plugins/plugin_utils/ssm/s3clientmanager.py

@@ -111,6 +111,10 @@ def generate_host_commands(
                     f"'{url}'"
                 )  # fmt: skip
         elif method == "put":
+            # The controller uploads the file to S3 with upload_fileobj(ExtraArgs=put_args),
+            # so the bucket's server-side-encryption settings must be applied here too,
+            # otherwise buckets enforcing encryption on PutObject reject the transfer.
+            put_args, _put_headers = generate_encryption_settings(bucket_sse_mode, bucket_sse_kms_key_id)
             url = self.get_url("get_object", bucket_name, s3_path, "GET")
             if is_windows:
                 # Use .NET File.WriteAllBytes instead of -OutFile to properly handle unicode paths

tests/integration/targets/setup_connection_aws_ssm/tasks/encryption.yml

@@ -42,3 +42,28 @@
     bucket_key_enabled: "{{ s3_bucket_encryption | default(false) }}"
   when:
     - encrypted_bucket | default(false)
+
+# When the connection plugin is expected to send encryption headers
+# (bucket_sse_mode is set in the inventory), reject any upload that
+# lacks them.  Default bucket encryption alone can't catch a missing
+# header because S3 then encrypts the object anyway (see issue 3018).
+- name: Ensure bucket denies uploads without encryption headers
+  amazon.aws.s3_bucket:
+    name: "{{ s3_bucket_name }}"
+    region: "{{ s3_bucket_region | default(omit) }}"
+    policy: "{{ deny_unencrypted_uploads_policy | to_json }}"
+  vars:
+    deny_unencrypted_uploads_policy:
+      Version: "2012-10-17"
+      Statement:
+        - Sid: DenyIncorrectEncryptionHeader
+          Effect: Deny
+          Principal: "*"
+          Action: "s3:PutObject"
+          Resource: "arn:aws:s3:::{{ s3_bucket_name }}/*"
+          Condition:
+            StringNotEquals:
+              s3:x-amz-server-side-encryption: "aws:kms"
+  when:
+    - encrypted_bucket | default(false)
+    - not (s3_bucket_encryption | default(false))

tests/unit/plugins/connection/aws_ssm/test_aws_ssm.py

@@ -224,7 +224,7 @@ def test_generate_host_commands(self, m_generate_encryption_settings, method, is
                     "put_object", bucket_name, s3_path, "PUT", extra_args=encryption_args
                 )
             elif method == "put":
-                m_generate_encryption_settings.assert_not_called()
+                m_generate_encryption_settings.assert_called_once_with(bucket_sse_mode, bucket_sse_kms_key_id)
                 if is_windows:
                     assert (
                         "$ErrorActionPreference = 'Stop' ; "
@@ -234,7 +234,7 @@ def test_generate_host_commands(self, m_generate_encryption_settings, method, is
                     ) == test_command_generation
                 else:
                     assert "curl -o 'test/out/path' 'https://test-url';touch 'test/out/path'" == test_command_generation
-                assert put_args is None
+                assert put_args == encryption_args
                 s3_client_manager.get_url.assert_called_once_with("get_object", bucket_name, s3_path, "GET")
 
     def test_get_url_no_extra_args(self):