Tighten CodeDeploy config file permissions

.changes/next-release/bugfix-codedeploy-93361.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "codedeploy",
+  "description": "Tighten file permissions for CodeDeploy configuration file"
+}

awscli/customizations/codedeploy/register.py

@@ -11,6 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
+import os
 import sys
 
 from awscli.customizations.codedeploy.systems import DEFAULT_CONFIG_FILE
@@ -162,13 +163,24 @@ def _create_config(self, params):
             f'Creating the on-premises instance configuration file named {DEFAULT_CONFIG_FILE}'
             '...'
         )
-        with open(DEFAULT_CONFIG_FILE, 'w') as f:
-            f.write(
-                '---\n'
-                f'region: {params.region}\n'
-                f'iam_user_arn: {params.iam_user_arn}\n'
-                f'aws_access_key_id: {params.access_key_id}\n'
-                f'aws_secret_access_key: {params.secret_access_key}\n'
+        try:
+            fd = os.open(
+                DEFAULT_CONFIG_FILE,
+                os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+                0o600,
+            )
+            with os.fdopen(fd, 'w') as f:
+                os.chmod(DEFAULT_CONFIG_FILE, 0o600)
+                f.write(
+                    '---\n'
+                    f'region: {params.region}\n'
+                    f'iam_user_arn: {params.iam_user_arn}\n'
+                    f'aws_access_key_id: {params.access_key_id}\n'
+                    f'aws_secret_access_key: {params.secret_access_key}\n'
+                )
+        except OSError as e:
+            raise RuntimeError(
+                f'Failed to create config file {DEFAULT_CONFIG_FILE}: {e}'
             )
         sys.stdout.write('DONE\n')
 

tests/unit/customizations/codedeploy/test_register.py

@@ -55,12 +55,22 @@ def setUp(self):
         self.globals.endpoint_url = self.endpoint_url
         self.globals.verify_ssl = False
 
-        self.open_patcher = mock.patch(
-            'awscli.customizations.codedeploy.register.open',
+        self.os_open_patcher = mock.patch(
+            'awscli.customizations.codedeploy.register.os.open',
+            return_value=3,
+        )
+        self.os_open = self.os_open_patcher.start()
+
+        self.os_fdopen_patcher = mock.patch(
+            'awscli.customizations.codedeploy.register.os.fdopen',
             mock.mock_open(),
-            create=True,
         )
-        self.open = self.open_patcher.start()
+        self.os_fdopen = self.os_fdopen_patcher.start()
+
+        self.os_chmod_patcher = mock.patch(
+            'awscli.customizations.codedeploy.register.os.chmod'
+        )
+        self.os_chmod = self.os_chmod_patcher.start()
 
         self.codedeploy = mock.MagicMock()
 
@@ -80,7 +90,9 @@ def setUp(self):
         self.register = Register(self.session)
 
     def tearDown(self):
-        self.open_patcher.stop()
+        self.os_open_patcher.stop()
+        self.os_fdopen_patcher.stop()
+        self.os_chmod_patcher.stop()
 
     def test_register_throws_on_invalid_region(self):
         self.globals.region = None
@@ -149,8 +161,13 @@ def test_register_with_no_iam_user_arn(self):
         self.assertEqual(self.policy_name, self.args.policy_name)
         self.assertIn('policy_document', self.args)
         self.assertEqual(self.policy_document, self.args.policy_document)
-        self.open.assert_called_with(self.config_file, 'w')
-        self.open().write.assert_called_with(
+        self.os_open.assert_called_with(
+            self.config_file,
+            mock.ANY,
+            0o600,
+        )
+        self.os_fdopen.assert_called_with(3, 'w')
+        self.os_fdopen().write.assert_called_with(
             '---\n'
             f'region: {self.region}\n'
             f'iam_user_arn: {self.iam_user_arn}\n'
@@ -167,7 +184,7 @@ def test_register_with_iam_user_arn(self):
         self.assertFalse(self.register.iam.create_user.called)
         self.assertFalse(self.register.iam.create_access_key.called)
         self.assertFalse(self.register.iam.put_user_policy.called)
-        self.assertFalse(self.open.called)
+        self.assertFalse(self.os_open.called)
         self.register.codedeploy.register_on_premises_instance.assert_called_with(
             instanceName=self.instance_name, iamUserArn=self.iam_user_arn
         )
@@ -192,6 +209,36 @@ def test_register_with_tags(self):
             tags=self.tags, instanceNames=[self.instance_name]
         )
 
+    def test_create_config_raises_runtime_error_on_open_failure(self):
+        self.args.iam_user_arn = None
+        self.os_open.side_effect = OSError('permission denied')
+        with self.assertRaisesRegex(
+            RuntimeError, 'Failed to create config file'
+        ):
+            self.register._create_config(self.args)
+
+    def test_create_config_raises_runtime_error_on_chmod_failure(self):
+        self.args.iam_user_arn = None
+        self.os_chmod.side_effect = OSError('permission denied')
+        with self.assertRaisesRegex(
+            RuntimeError, 'Failed to create config file'
+        ):
+            self.register._create_config(self.args)
+
+    def test_create_config_uses_restricted_permissions(self):
+        self.args.iam_user_arn = None
+        self.register._run_main(self.args, self.globals)
+        self.os_open.assert_called_with(
+            self.config_file,
+            mock.ANY,
+            0o600,
+        )
+
+    def test_create_config_chmods_existing_file(self):
+        self.args.iam_user_arn = None
+        self.register._run_main(self.args, self.globals)
+        self.os_chmod.assert_called_with(self.config_file, 0o600)
+
 
 if __name__ == "__main__":
     unittest.main()